mirror of
https://github.com/Twigpine/zero.git
synced 2026-11-08 01:09:46 -10:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
57ef38ebcb | ||
|
|
31e57915c9 | ||
|
|
a0548d707b | ||
|
|
83a371d669 | ||
|
|
84823d7be1 |
No files matched your search
@@ -211,11 +211,11 @@ func (provider *Provider) stream(ctx context.Context, body []byte, events chan<-
|
||||
err = providerio.ScanSSEDataWithContext(streamCtx, cancelStream, response.Body, provider.streamIdleTimeout, func(data string) bool {
|
||||
return provider.emitPayload(ctx, data, state, events)
|
||||
})
|
||||
if errors.Is(err, providerio.ErrStreamIdle) {
|
||||
if errors.Is(err, providerio.ErrStreamIdle) || errors.Is(err, providerio.ErrStreamStalled) {
|
||||
state.closeOpen(ctx, events)
|
||||
providerio.SendEvent(ctx, events, zeroruntime.StreamEvent{
|
||||
Type: zeroruntime.StreamEventError,
|
||||
Error: provider.redact(fmt.Sprintf("provider stream error: idle timeout after %s (upstream stopped sending data)", provider.streamIdleTimeout)),
|
||||
Error: provider.redact("provider stream error: " + providerio.StreamTimeoutMessage(err, provider.streamIdleTimeout)),
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
@@ -264,6 +264,13 @@ func TestNewRejectsUnsupportedProviderKind(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestNewRoutesChatGPTCatalogToCodexProvider(t *testing.T) {
|
||||
// Isolate the OAuth token store to an empty temp path so the factory reads no
|
||||
// stored login — otherwise this test picks up the developer's real chatgpt
|
||||
// OAuth token and the "want empty chatgpt-account-id" assertion fails locally
|
||||
// (it still passes in CI, where no login is stored). Mirrors the isolation in
|
||||
// TestNewRoutesChatGPTCatalogWithStoredAccountID.
|
||||
t.Setenv("ZERO_OAUTH_TOKENS_PATH", t.TempDir()+"/tokens.json")
|
||||
|
||||
transport := &captureTransport{
|
||||
responseBody: "data: [DONE]\n\n",
|
||||
}
|
||||
|
||||
@@ -158,10 +158,10 @@ func (provider *Provider) stream(ctx context.Context, body []byte, events chan<-
|
||||
err = providerio.ScanSSEDataWithContext(streamCtx, cancelStream, response.Body, provider.streamIdleTimeout, func(data string) bool {
|
||||
return provider.emitPayload(ctx, data, &state, events)
|
||||
})
|
||||
if errors.Is(err, providerio.ErrStreamIdle) {
|
||||
if errors.Is(err, providerio.ErrStreamIdle) || errors.Is(err, providerio.ErrStreamStalled) {
|
||||
providerio.SendEvent(ctx, events, zeroruntime.StreamEvent{
|
||||
Type: zeroruntime.StreamEventError,
|
||||
Error: provider.redact(fmt.Sprintf("provider stream error: idle timeout after %s (upstream stopped sending data)", provider.streamIdleTimeout)),
|
||||
Error: provider.redact("provider stream error: " + providerio.StreamTimeoutMessage(err, provider.streamIdleTimeout)),
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
@@ -59,6 +59,7 @@ const (
|
||||
responsesEventOutputItemAdded = "response.output_item.added"
|
||||
responsesEventContentPartAdded = "response.content_part.added"
|
||||
responsesEventOutputTextDelta = "response.output_text.delta"
|
||||
responsesEventReasoningDelta = "response.reasoning_summary_text.delta"
|
||||
responsesEventOutputTextDone = "response.output_text.done"
|
||||
responsesEventFunctionArgsDelta = "response.function_call_arguments.delta"
|
||||
responsesEventContentPartDone = "response.content_part.done"
|
||||
@@ -86,6 +87,10 @@ type responsesRequest struct {
|
||||
// here; omitted entirely when the caller requests no (or an unsupported) effort.
|
||||
type responsesReasoning struct {
|
||||
Effort string `json:"effort,omitempty"`
|
||||
// Summary requests a streamed reasoning summary ("auto" lets the API pick a
|
||||
// level). Without it the backend emits no reasoning events, so a long thinking
|
||||
// phase produces zero visible output and reads as a hang in the UI.
|
||||
Summary string `json:"summary,omitempty"`
|
||||
}
|
||||
|
||||
// inputItem is one element of the Responses `input` array. The Type field
|
||||
@@ -256,7 +261,9 @@ func (p *CodexProvider) buildResponsesRequest(request zeroruntime.CompletionRequ
|
||||
// and an empty or unsupported effort simply omits the field — without this the
|
||||
// caller's chosen effort was silently dropped for every Codex model.
|
||||
if effort := openAIReasoningEffort(request.ReasoningEffort); effort != "" {
|
||||
req.Reasoning = &responsesReasoning{Effort: effort}
|
||||
// Summary "auto" makes the backend stream reasoning_summary_text deltas so a
|
||||
// long thinking phase shows live progress instead of looking hung.
|
||||
req.Reasoning = &responsesReasoning{Effort: effort, Summary: "auto"}
|
||||
}
|
||||
return req, nil
|
||||
}
|
||||
@@ -382,10 +389,10 @@ func (p *CodexProvider) streamResponses(
|
||||
err = providerio.ScanSSEDataWithContext(streamCtx, cancelStream, response.Body, inner.streamIdleTimeout, func(data string) bool {
|
||||
return p.emitResponsesEvent(ctx, data, state, events)
|
||||
})
|
||||
if errors.Is(err, providerio.ErrStreamIdle) {
|
||||
if errors.Is(err, providerio.ErrStreamIdle) || errors.Is(err, providerio.ErrStreamStalled) {
|
||||
providerio.SendEvent(ctx, events, zeroruntime.StreamEvent{
|
||||
Type: zeroruntime.StreamEventError,
|
||||
Error: p.redact(fmt.Sprintf("provider stream error: idle timeout after %s (upstream stopped sending data)", inner.streamIdleTimeout)),
|
||||
Error: p.redact("provider stream error: " + providerio.StreamTimeoutMessage(err, inner.streamIdleTimeout)),
|
||||
})
|
||||
return
|
||||
}
|
||||
@@ -501,6 +508,17 @@ func (p *CodexProvider) emitResponsesEvent(
|
||||
})
|
||||
}
|
||||
return true
|
||||
case responsesEventReasoningDelta:
|
||||
// Reasoning summary deltas: surface as live "thinking" so a long reasoning
|
||||
// phase shows progress (and keeps the activity clock fresh) instead of
|
||||
// looking like a hang. Requested via reasoning.summary="auto".
|
||||
if event.Delta != "" {
|
||||
providerio.SendEvent(ctx, events, zeroruntime.StreamEvent{
|
||||
Type: zeroruntime.StreamEventReasoning,
|
||||
Content: event.Delta,
|
||||
})
|
||||
}
|
||||
return true
|
||||
case responsesEventFunctionArgsDelta:
|
||||
p.handleFunctionArgsDelta(ctx, &event, state, events)
|
||||
return true
|
||||
|
||||
@@ -485,6 +485,58 @@ func TestCodexProviderForwardsReasoningEffort(t *testing.T) {
|
||||
if reasoning["effort"] != "high" {
|
||||
t.Fatalf("body.reasoning.effort = %#v, want high", reasoning["effort"])
|
||||
}
|
||||
// A reasoning summary must be requested so the backend streams
|
||||
// reasoning_summary_text deltas — otherwise a long think shows no live output.
|
||||
if reasoning["summary"] != "auto" {
|
||||
t.Fatalf("body.reasoning.summary = %#v, want auto", reasoning["summary"])
|
||||
}
|
||||
}
|
||||
|
||||
func TestCodexProviderStreamsReasoningSummaryDeltas(t *testing.T) {
|
||||
// reasoning_summary_text deltas must surface as StreamEventReasoning (live
|
||||
// "thinking"), in order, alongside the normal text output. Without this a long
|
||||
// reasoning phase produces zero visible output and reads as a hang.
|
||||
var rec codexRequest
|
||||
srv := newCodexResponsesServer(t, &rec,
|
||||
`{"type":"response.created","response":{"id":"resp-1","status":"in_progress"}}`,
|
||||
`{"type":"response.reasoning_summary_text.delta","delta":"Thinking. "}`,
|
||||
`{"type":"response.reasoning_summary_text.delta","delta":"Planning."}`,
|
||||
`{"type":"response.output_text.delta","delta":"done"}`,
|
||||
`{"type":"response.completed","response":{"id":"resp-1","status":"completed"}}`,
|
||||
)
|
||||
defer srv.Close()
|
||||
|
||||
provider, err := NewCodexProvider(CodexOptions{
|
||||
Options: Options{APIKey: "sk-test", BaseURL: srv.URL, Model: "gpt-5"},
|
||||
AccountID: "acc-x",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("NewCodexProvider: %v", err)
|
||||
}
|
||||
stream, err := provider.StreamCompletion(context.Background(), zeroruntime.CompletionRequest{
|
||||
Messages: []zeroruntime.Message{{Role: zeroruntime.MessageRoleUser, Content: "solve it"}},
|
||||
ReasoningEffort: "high",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("StreamCompletion: %v", err)
|
||||
}
|
||||
var reasoning, text []string
|
||||
for ev := range stream {
|
||||
switch ev.Type {
|
||||
case zeroruntime.StreamEventReasoning:
|
||||
reasoning = append(reasoning, ev.Content)
|
||||
case zeroruntime.StreamEventText:
|
||||
text = append(text, ev.Content)
|
||||
case zeroruntime.StreamEventError:
|
||||
t.Fatalf("unexpected error event: %q", ev.Error)
|
||||
}
|
||||
}
|
||||
if got := strings.Join(reasoning, ""); got != "Thinking. Planning." {
|
||||
t.Fatalf("reasoning deltas = %q, want %q", got, "Thinking. Planning.")
|
||||
}
|
||||
if got := strings.Join(text, ""); got != "done" {
|
||||
t.Fatalf("text deltas = %q, want %q", got, "done")
|
||||
}
|
||||
}
|
||||
|
||||
func TestCodexProviderOmitsReasoningWhenUnset(t *testing.T) {
|
||||
|
||||
@@ -218,12 +218,12 @@ func (provider *Provider) stream(ctx context.Context, body []byte, events chan<-
|
||||
err = providerio.ScanSSEDataWithContext(streamCtx, cancelStream, response.Body, provider.streamIdleTimeout, func(data string) bool {
|
||||
return provider.emitPayload(ctx, data, state, events)
|
||||
})
|
||||
if errors.Is(err, providerio.ErrStreamIdle) {
|
||||
if errors.Is(err, providerio.ErrStreamIdle) || errors.Is(err, providerio.ErrStreamStalled) {
|
||||
state.flushBufferedContent(events)
|
||||
state.closeBufferedOpen(events)
|
||||
sendEvent(ctx, events, zeroruntime.StreamEvent{
|
||||
Type: zeroruntime.StreamEventError,
|
||||
Error: provider.redact(fmt.Sprintf("provider stream error: idle timeout after %s (upstream stopped sending data)", provider.streamIdleTimeout)),
|
||||
Error: provider.redact("provider stream error: " + providerio.StreamTimeoutMessage(err, provider.streamIdleTimeout)),
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
@@ -23,6 +23,26 @@ const maxSSELineBytes = 16 * 1024 * 1024
|
||||
// closing the connection. Callers surface it as an idle-timeout error.
|
||||
var ErrStreamIdle = errors.New("idle timeout (upstream stopped sending data)")
|
||||
|
||||
// ErrStreamStalled reports that a streaming upstream kept the connection alive
|
||||
// (SSE keep-alives reset the idle watchdog, so it never fired) but produced no
|
||||
// actual output for streamContentStallFactor × the idle timeout. Without this an
|
||||
// upstream that heartbeats-but-stalls — observed on chatgpt/gpt-5.x and ollama
|
||||
// reasoning models — would hang the agent indefinitely.
|
||||
var ErrStreamStalled = errors.New("stream stalled (upstream kept the connection alive but produced no output)")
|
||||
|
||||
// StreamTimeoutMessage returns the human-readable detail for a stream-timeout
|
||||
// error (ErrStreamIdle or ErrStreamStalled), given the configured idle timeout.
|
||||
// Callers prepend their own "provider stream error: " prefix. A stalled stream
|
||||
// gets a distinct, actionable message — it did NOT stop sending data (keep-alives
|
||||
// kept arriving); it just produced no output, which usually means the model is
|
||||
// stuck or very slow.
|
||||
func StreamTimeoutMessage(err error, idleTimeout time.Duration) string {
|
||||
if errors.Is(err, ErrStreamStalled) {
|
||||
return fmt.Sprintf("no output for %s (the model kept the connection alive but produced nothing — it may be stuck; try a faster model or lower reasoning effort)", idleTimeout*streamContentStallFactor)
|
||||
}
|
||||
return fmt.Sprintf("idle timeout after %s (upstream stopped sending data)", idleTimeout)
|
||||
}
|
||||
|
||||
// DefaultStreamIdleTimeout is the single source of truth for how long every
|
||||
// provider waits on a silent stream before aborting it. The watchdog only fires
|
||||
// on genuine silence — SSE keep-alive comments reset it (see
|
||||
@@ -33,6 +53,14 @@ var ErrStreamIdle = errors.New("idle timeout (upstream stopped sending data)")
|
||||
// Override globally with ZERO_STREAM_IDLE_TIMEOUT.
|
||||
const DefaultStreamIdleTimeout = 5 * time.Minute
|
||||
|
||||
// streamContentStallFactor bounds a heartbeat-but-no-output stream. Keep-alives
|
||||
// reset the idle watchdog (a heartbeating upstream is not "dead"), but if NO real
|
||||
// data line arrives for streamContentStallFactor × the idle timeout, the stream is
|
||||
// treated as stalled and aborted (ErrStreamStalled). It is > 1 so a slow-but-
|
||||
// producing request — one that emits data between heartbeats — is never killed; the
|
||||
// content watchdog only catches a stream that heartbeats while producing nothing.
|
||||
const streamContentStallFactor = 2
|
||||
|
||||
// streamIdleTimeoutEnv is the global override for the stream idle timeout. It
|
||||
// accepts a Go duration ("5m", "300s", "90s") or a bare number of seconds
|
||||
// ("300"). A value of "0", "off", "none", or "disabled" turns the watchdog off
|
||||
@@ -202,21 +230,36 @@ func ScanSSEDataWithContext(
|
||||
// The idle watchdog is optional. When idleTimeout <= 0 it is disabled, but we
|
||||
// STILL run the goroutine + select loop so ctx cancellation is always honored
|
||||
// (a nil idleC channel simply never fires in the select).
|
||||
var idleC <-chan time.Time
|
||||
var idleC, contentC <-chan time.Time
|
||||
resetIdle := func() {}
|
||||
resetContent := func() {}
|
||||
if idleTimeout > 0 {
|
||||
// reset drains the timer's channel if it already fired before re-arming, so
|
||||
// a stale tick can't trip the watchdog after activity resumed.
|
||||
reset := func(t *time.Timer, d time.Duration) func() {
|
||||
return func() {
|
||||
if !t.Stop() {
|
||||
select {
|
||||
case <-t.C:
|
||||
default:
|
||||
}
|
||||
}
|
||||
t.Reset(d)
|
||||
}
|
||||
}
|
||||
idle := time.NewTimer(idleTimeout)
|
||||
defer idle.Stop()
|
||||
idleC = idle.C
|
||||
resetIdle = func() {
|
||||
if !idle.Stop() {
|
||||
select {
|
||||
case <-idle.C:
|
||||
default:
|
||||
}
|
||||
}
|
||||
idle.Reset(idleTimeout)
|
||||
}
|
||||
resetIdle = reset(idle, idleTimeout)
|
||||
|
||||
// Content watchdog: only real data lines reset it (keep-alives do not), so a
|
||||
// stream that heartbeats without producing output is bounded instead of
|
||||
// hanging forever.
|
||||
contentTimeout := idleTimeout * streamContentStallFactor
|
||||
content := time.NewTimer(contentTimeout)
|
||||
defer content.Stop()
|
||||
contentC = content.C
|
||||
resetContent = reset(content, contentTimeout)
|
||||
}
|
||||
|
||||
for {
|
||||
@@ -232,6 +275,12 @@ func ScanSSEDataWithContext(
|
||||
// a timeout instead of blocking the agent forever.
|
||||
cancel()
|
||||
return ErrStreamIdle
|
||||
case <-contentC:
|
||||
// Upstream kept heartbeating (idle watchdog never fired) but produced
|
||||
// no output for the content window. Treat as stalled and abort rather
|
||||
// than hanging forever.
|
||||
cancel()
|
||||
return ErrStreamStalled
|
||||
case item, ok := <-payloads:
|
||||
if !ok {
|
||||
// Reader finished: deliver its terminal status (EOF -> nil,
|
||||
@@ -246,8 +295,11 @@ func ScanSSEDataWithContext(
|
||||
}
|
||||
resetIdle()
|
||||
if item.keepAlive {
|
||||
// A heartbeat keeps the connection "alive" (idle watchdog) but is
|
||||
// NOT output, so it must not reset the content watchdog.
|
||||
continue
|
||||
}
|
||||
resetContent()
|
||||
if !handle(item.data) {
|
||||
// The provider asked to stop (e.g. it already emitted an error
|
||||
// for this payload). Abort the read and end like ScanSSEData:
|
||||
|
||||
@@ -201,3 +201,106 @@ func TestUpstreamUnreachable(t *testing.T) {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// A heartbeating-but-output-less upstream must not hang forever: SSE keep-alives
|
||||
// feed the idle watchdog (so it never fires), but the content watchdog
|
||||
// (idleTimeout × streamContentStallFactor) aborts with ErrStreamStalled when no
|
||||
// real data line arrives. This is the gpt-5.x / ollama "still generating forever"
|
||||
// hang.
|
||||
func TestScanSSEDataWithContextAbortsOnContentStall(t *testing.T) {
|
||||
pr, pw := io.Pipe()
|
||||
defer func() { _ = pw.Close() }()
|
||||
|
||||
stop := make(chan struct{})
|
||||
defer close(stop)
|
||||
go func() {
|
||||
// Heartbeat every 30ms — comfortably under the 100ms idle timeout, so idle
|
||||
// never fires under CI/GC jitter — but never send a data line.
|
||||
for {
|
||||
select {
|
||||
case <-stop:
|
||||
return
|
||||
default:
|
||||
}
|
||||
if _, err := io.WriteString(pw, ": keep-alive\n\n"); err != nil {
|
||||
return
|
||||
}
|
||||
time.Sleep(30 * time.Millisecond)
|
||||
}
|
||||
}()
|
||||
|
||||
cancelled := false
|
||||
cancel := func() { cancelled = true }
|
||||
|
||||
done := make(chan error, 1)
|
||||
go func() {
|
||||
// idle 100ms → content stall at 200ms. Keep-alives reset idle but not content.
|
||||
done <- ScanSSEDataWithContext(context.Background(), cancel, pr, 100*time.Millisecond, func(string) bool { return true })
|
||||
}()
|
||||
|
||||
select {
|
||||
case err := <-done:
|
||||
if !errors.Is(err, ErrStreamStalled) {
|
||||
t.Fatalf("err = %v, want ErrStreamStalled", err)
|
||||
}
|
||||
case <-time.After(3 * time.Second):
|
||||
t.Fatal("ScanSSEDataWithContext hung on a heartbeat-but-no-output stream")
|
||||
}
|
||||
if !cancelled {
|
||||
t.Fatal("content stall did not cancel the request context")
|
||||
}
|
||||
}
|
||||
|
||||
// Real data lines reset the content watchdog, so a slow-but-producing stream that
|
||||
// runs longer than the content window is never aborted.
|
||||
func TestScanSSEDataWithContextContentResetsOnData(t *testing.T) {
|
||||
pr, pw := io.Pipe()
|
||||
|
||||
go func() {
|
||||
// One data line every 30ms for ~300ms (well past the 200ms content window,
|
||||
// with comfortable slack under the 100ms idle timeout), so the content
|
||||
// watchdog keeps resetting, then close cleanly.
|
||||
for i := 0; i < 10; i++ {
|
||||
if _, err := io.WriteString(pw, "data: chunk\n\n"); err != nil {
|
||||
return
|
||||
}
|
||||
time.Sleep(30 * time.Millisecond)
|
||||
}
|
||||
_ = pw.Close()
|
||||
}()
|
||||
|
||||
n := 0
|
||||
done := make(chan error, 1)
|
||||
go func() {
|
||||
done <- ScanSSEDataWithContext(context.Background(), func() {}, pr, 100*time.Millisecond, func(string) bool { n++; return true })
|
||||
}()
|
||||
|
||||
select {
|
||||
case err := <-done:
|
||||
if err != nil {
|
||||
t.Fatalf("err = %v, want nil (data kept the stream alive)", err)
|
||||
}
|
||||
case <-time.After(3 * time.Second):
|
||||
t.Fatal("ScanSSEDataWithContext hung on a producing stream")
|
||||
}
|
||||
if n != 10 {
|
||||
t.Fatalf("handled %d data lines, want 10", n)
|
||||
}
|
||||
}
|
||||
|
||||
// StreamTimeoutMessage must give a stalled stream a distinct, accurate detail —
|
||||
// it reports the content window (idle × factor) and must NOT claim the upstream
|
||||
// "stopped sending data" (keep-alives were still arriving).
|
||||
func TestStreamTimeoutMessage(t *testing.T) {
|
||||
idle := 5 * time.Minute
|
||||
if msg := StreamTimeoutMessage(ErrStreamIdle, idle); !strings.Contains(msg, "idle timeout after 5m") {
|
||||
t.Fatalf("idle message = %q, want it to mention the 5m idle timeout", msg)
|
||||
}
|
||||
stalled := StreamTimeoutMessage(ErrStreamStalled, idle)
|
||||
if !strings.Contains(stalled, "no output for 10m") {
|
||||
t.Fatalf("stalled message = %q, want it to report the 10m content window", stalled)
|
||||
}
|
||||
if strings.Contains(stalled, "stopped sending data") {
|
||||
t.Fatalf("stalled message must not claim the upstream stopped sending data: %q", stalled)
|
||||
}
|
||||
}
|
||||
+21
-2
@@ -103,6 +103,25 @@ func errResult(format string, a ...any) tools.Result {
|
||||
return tools.Result{Status: tools.StatusError, Output: "Error: " + fmt.Sprintf(format, a...)}
|
||||
}
|
||||
|
||||
// agentTypeProperty builds the schema for an agent-type parameter, constraining it
|
||||
// to the roster's currently registered types (the built-ins plus any user-registered
|
||||
// agents) via an enum so a model can't spawn or hand off to an unknown type — the
|
||||
// schema teaches the valid set up front instead of the model learning it from a
|
||||
// failed call. Falls back to a free-form string if the roster is unavailable or
|
||||
// empty (so the tool never advertises an empty, un-satisfiable enum).
|
||||
func agentTypeProperty(sw *Swarm, description string) tools.PropertySchema {
|
||||
prop := tools.PropertySchema{Type: "string", Description: description}
|
||||
if sw == nil {
|
||||
return prop
|
||||
}
|
||||
if registry := sw.Registry(); registry != nil {
|
||||
if types := registry.AgentTypes(); len(types) > 0 {
|
||||
prop.Enum = types
|
||||
}
|
||||
}
|
||||
return prop
|
||||
}
|
||||
|
||||
// ---- swarm_spawn -----------------------------------------------------------
|
||||
|
||||
type spawnTool struct {
|
||||
@@ -118,7 +137,7 @@ func (t *spawnTool) Parameters() tools.Schema {
|
||||
return tools.Schema{
|
||||
Type: "object",
|
||||
Properties: map[string]tools.PropertySchema{
|
||||
"agent_type": {Type: "string", Description: "Roster agent type to spawn (e.g. teammate, subagent)."},
|
||||
"agent_type": agentTypeProperty(t.sw, "Roster agent type to spawn."),
|
||||
"task": {Type: "string", Description: "The focused task/briefing for the member."},
|
||||
"team": {Type: "string", Description: "Team to spawn into. Defaults to \"default\"."},
|
||||
},
|
||||
@@ -353,7 +372,7 @@ func (t *handoffTool) Parameters() tools.Schema {
|
||||
Type: "object",
|
||||
Properties: map[string]tools.PropertySchema{
|
||||
"task_id": {Type: "string", Description: "Task id to hand off."},
|
||||
"to_agent_type": {Type: "string", Description: "Roster agent type to take over."},
|
||||
"to_agent_type": agentTypeProperty(t.sw, "Roster agent type to take over."),
|
||||
"note": {Type: "string", Description: "Optional handoff note delivered to the new member's inbox."},
|
||||
"team": {Type: "string", Description: "Team the task belongs to. Defaults to \"default\"."},
|
||||
},
|
||||
|
||||
@@ -2,6 +2,7 @@ package swarm
|
||||
|
||||
import (
|
||||
"context"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
@@ -10,6 +11,46 @@ import (
|
||||
"github.com/Gitlawb/zero/internal/tools"
|
||||
)
|
||||
|
||||
func TestSpawnAndHandoffAgentTypeEnumReflectsRoster(t *testing.T) {
|
||||
sw := newSwarmFor(t, newLauncher(okFor))
|
||||
want := sw.Registry().AgentTypes() // sorted built-ins: subagent, teammate
|
||||
|
||||
spawnEnum := (&spawnTool{sw: sw}).Parameters().Properties["agent_type"].Enum
|
||||
if !slices.Equal(spawnEnum, want) {
|
||||
t.Fatalf("spawn agent_type enum = %v, want roster %v", spawnEnum, want)
|
||||
}
|
||||
if !slices.Contains(spawnEnum, "subagent") || !slices.Contains(spawnEnum, "teammate") {
|
||||
t.Fatalf("enum missing built-in roster types: %v", spawnEnum)
|
||||
}
|
||||
handoffEnum := (&handoffTool{sw: sw}).Parameters().Properties["to_agent_type"].Enum
|
||||
if !slices.Equal(handoffEnum, want) {
|
||||
t.Fatalf("handoff to_agent_type enum = %v, want roster %v", handoffEnum, want)
|
||||
}
|
||||
|
||||
// A user-registered custom type must extend the enum (not a hardcoded list).
|
||||
if err := sw.Registry().Register(Definition{AgentType: "researcher"}); err != nil {
|
||||
t.Fatalf("register custom agent: %v", err)
|
||||
}
|
||||
spawnEnum = (&spawnTool{sw: sw}).Parameters().Properties["agent_type"].Enum
|
||||
if !slices.Contains(spawnEnum, "researcher") {
|
||||
t.Fatalf("custom agent type not reflected in spawn enum: %v", spawnEnum)
|
||||
}
|
||||
handoffEnum = (&handoffTool{sw: sw}).Parameters().Properties["to_agent_type"].Enum
|
||||
if !slices.Contains(handoffEnum, "researcher") {
|
||||
t.Fatalf("custom agent type not reflected in handoff enum: %v", handoffEnum)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAgentTypePropertyFallsBackWithoutRoster(t *testing.T) {
|
||||
prop := agentTypeProperty(nil, "desc")
|
||||
if prop.Type != "string" || prop.Description != "desc" {
|
||||
t.Fatalf("unexpected property: %+v", prop)
|
||||
}
|
||||
if prop.Enum != nil {
|
||||
t.Fatalf("nil swarm must not advertise an (empty) enum, got %v", prop.Enum)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCollapseRuneSafeTruncation(t *testing.T) {
|
||||
// 300 multi-byte runes: truncation must not split a rune (no invalid UTF-8).
|
||||
got := collapse(strings.Repeat("🚀", 300))
|
||||
|
||||
@@ -261,13 +261,11 @@ func (m model) swarmSpawnedAgents() []swarmAgent {
|
||||
switch a.state {
|
||||
case "done", "failed", "completed", "cancelled":
|
||||
a.finishing = true
|
||||
if m.pending {
|
||||
// Run still going: keep it visible and clickable, no fade.
|
||||
a.finishedAt = time.Time{}
|
||||
live = append(live, a)
|
||||
continue
|
||||
}
|
||||
// Turn ended: fade out over the linger window, then drop.
|
||||
// A member drops once ITS OWN task completes — not when the whole turn
|
||||
// ends: fade out over the linger window from when it was first seen
|
||||
// finished (stamped each tick by stampSwarmDone), then remove. This
|
||||
// holds whether or not the overall run is still in flight, mirroring
|
||||
// how finished specialists drop (sidebarSpecialists).
|
||||
doneAt, stamped := m.swarmDoneAt[a.id]
|
||||
if stamped && m.now().Sub(doneAt) >= sidebarAgentLinger {
|
||||
continue // past the linger window — remove
|
||||
|
||||
@@ -391,32 +391,32 @@ func TestSidebarShowsSwarmSpawnedAgents(t *testing.T) {
|
||||
// row (e.g. a resumed transcript that dropped the call): the member is still
|
||||
// A finished member stays visible (and clickable) while the run is still in
|
||||
// flight, so the user can inspect it; only once the turn ends does it drop.
|
||||
func TestSwarmAgentsPersistWhileRunInFlight(t *testing.T) {
|
||||
func TestSwarmAgentDropsOnOwnCompletionEvenMidRun(t *testing.T) {
|
||||
base := time.Date(2026, 6, 24, 12, 0, 0, 0, time.UTC)
|
||||
m := sidebarTestModel()
|
||||
m.now = func() time.Time { return base }
|
||||
m.pending = true // run still going
|
||||
m.pending = true // run STILL going — a member must still drop on its OWN completion
|
||||
m.activeRunID = 7 // exercise the run-scoped filter with a non-zero id
|
||||
m.transcript = append(m.transcript,
|
||||
transcriptRow{kind: rowToolCall, tool: "swarm_spawn", detail: "build homepage", runID: 7},
|
||||
transcriptRow{kind: rowToolResult, tool: "swarm_spawn", detail: "Spawned subagent as task subagent-1 on team default.", runID: 7},
|
||||
transcriptRow{kind: rowToolResult, tool: "swarm_collect", detail: "Results: 1 task(s)\n- subagent-1 [done] build homepage", runID: 7},
|
||||
)
|
||||
// Long past the linger window — but it must still show while pending.
|
||||
m.swarmDoneAt = map[string]time.Time{"subagent-1": base.Add(-10 * sidebarAgentLinger)}
|
||||
|
||||
// Just finished, still within the linger window: shows briefly with a fading ✓.
|
||||
m.swarmDoneAt = map[string]time.Time{"subagent-1": base.Add(-sidebarAgentLinger / 2)}
|
||||
agents := m.swarmSpawnedAgents()
|
||||
if len(agents) != 1 {
|
||||
t.Fatalf("a finished member must stay while the run is in flight, got %d: %+v", len(agents), agents)
|
||||
t.Fatalf("within the linger window a finished member should still show (fading), got %d: %+v", len(agents), agents)
|
||||
}
|
||||
if !agents[0].finishing {
|
||||
t.Fatalf("a finished member should render done (✓), got %+v", agents[0])
|
||||
}
|
||||
|
||||
// Once the turn ends, the long-finished member fades out and drops.
|
||||
m.pending = false
|
||||
// Past the linger window: it drops — even though the overall run is still in
|
||||
// flight (previously a finished member lingered until the whole turn ended).
|
||||
m.swarmDoneAt = map[string]time.Time{"subagent-1": base.Add(-2 * sidebarAgentLinger)}
|
||||
if got := len(m.swarmSpawnedAgents()); got != 0 {
|
||||
t.Fatalf("after the turn ends a long-finished member should drop, got %d", got)
|
||||
t.Fatalf("a member past its linger window must drop on its own completion mid-run, got %d", got)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user