Compare commits

...
Author SHA1 Message Date
Gnanam ca748c2b77 fix(config): clear ZERO_PROVIDER when switching to an unnamed/default profile
SetActiveProviderEnv("") now unsets ZERO_PROVIDER instead of leaving the previous
value in the environment. Switching from a named provider back to an unnamed/default
profile otherwise kept exporting the stale provider into child zero exec processes,
which would resolve the old credentials. Extended the round-trip test with the
stale-env regression case: seed a value, switch to blank, assert both os.Getenv and
applyEnv no longer see it.

Addresses CodeRabbit review on #348.
2026-07-01 01:50:51 +05:30
Gnanam ef1a95a6ad fix(tui): sub-agents inherit the parent's live provider after a /model switch
Swarm members and Task sub-agents run as child 'zero exec' processes that re-resolve
the provider from config.json themselves — they're passed --model but never the
parent's provider. After switching provider/model in the TUI via /model (an in-memory
change), the child still resolved config.json's stale activeProvider, landed on a
provider whose credentials didn't match the live model, and failed auth in ~3s
(exit 3, 0 tool calls, 'API key missing or invalid / can't reach the provider') — so
the spawned agent appeared then instantly disappeared.

Export ZERO_PROVIDER (config.SetActiveProviderEnv) to the live provider name on a
/model provider switch. Child processes inherit the environment, and config
resolution already maps ZERO_PROVIDER -> activeProvider, so the child resolves the
SAME profile and its credentials (env key / stored key / OAuth) — covering all three
auth methods, for both Task and swarm (both spawn via the specialist executor).

Startup needs no export: the parent and child both read config's default, so they
already match; only an in-memory switch diverges.
2026-07-01 00:01:59 +05:30
KRATOS 02a3d12f34 fix(providerio): bound heartbeat-but-no-output streams so they can't hang forever (#347)
* fix(providerio): bound heartbeat-but-no-output streams (content-stall watchdog)

A stalled-but-alive upstream hung the agent indefinitely: SSE keep-alives reset the
idle watchdog (intentionally — a heartbeating upstream isn't 'dead', and aborting it
killed healthy long requests), so a stream that heartbeats while producing nothing
never tripped the idle timeout and there was no other ceiling. Observed on both
chatgpt/gpt-5.x (Codex) and ollama minimax-m3 — different provider code, same shared
providerio SSE loop — so the fix belongs here, once, for every provider.

Add a second 'content' watchdog alongside the idle one: keep-alives still reset idle
(no regression for heartbeating connections), but only REAL data lines reset the
content watchdog. If no data arrives for streamContentStallFactor (2) × the idle
timeout, abort with ErrStreamStalled instead of hanging. A slow-but-producing stream
(data between heartbeats) keeps resetting content and is never killed.

Contained to internal/providers/providerio — no provider/caller changes.

* fix(providerio): surface ErrStreamStalled in callers + widen flaky test margins

Address review:
- All four provider stream handlers (openai, codex, anthropic, gemini) now treat
  ErrStreamStalled in the same branch as ErrStreamIdle, via a new
  providerio.StreamTimeoutMessage helper that gives the stall a distinct, accurate
  message (it reports the content window and does NOT claim the upstream stopped
  sending data — keep-alives were still arriving). Previously a stalled stream fell
  through to the generic provider-error path.
- Widen the content-watchdog test timing margins (idle 100ms, data/keepalive every
  30ms) so CI/GC jitter can't trip the idle watchdog and mask the content path.
- Add a StreamTimeoutMessage unit test.
2026-06-30 22:19:52 +05:30
KRATOS a0548d707b test(providers): isolate the OAuth store in the chatgpt Codex routing test (#346)
TestNewRoutesChatGPTCatalogToCodexProvider asserts the chatgpt-account-id header is
empty when no OAuth login is stored, but never redirected the token store — so on a
developer machine with a real chatgpt login it read that login and failed (it still
passed in CI, where no token exists). Point ZERO_OAUTH_TOKENS_PATH at an empty temp
path, mirroring TestNewRoutesChatGPTCatalogWithStoredAccountID, so the test is
deterministic regardless of the developer's stored credentials.
2026-06-30 21:21:44 +05:30
KRATOS 83a371d669 fix: stream Codex reasoning + drop swarm agents on their own completion (#345)
Two agent-visibility fixes surfaced diagnosing a 20-minute gpt-5.5 step and an
'agents never disappear' report.

A) Codex Responses path streamed no reasoning, so a long thinking phase produced
   zero visible output and read as a hang (the activity clock never advanced).
   Root cause: the request never asked for a reasoning summary and the parser had
   no case for reasoning deltas. Now request reasoning.summary="auto" and forward
   response.reasoning_summary_text.delta as StreamEventReasoning — the existing TUI
   handler renders it live and refreshes the activity clock.

B) A finished swarm member stayed in the AGENTS panel until the whole turn ended.
   Now it drops once ITS OWN task completes (fade over the linger window, then
   remove), independent of the overall run — matching how finished specialists
   already behave. Running members and mid-flight collect are unaffected.
2026-06-30 21:21:30 +05:30
KRATOS 84823d7be1 fix(swarm): constrain agent_type to the registered roster via a dynamic enum (#344)
* fix(swarm): constrain spawn/handoff agent_type to the registered roster via enum

swarm_spawn and swarm_handoff described agent_type / to_agent_type as a free-form
string with only an 'e.g. teammate, subagent' hint, so a model could pass an
unregistered type (e.g. 'worker') and only discover it was invalid from the runtime
'unknown agent type' error — wasting tool calls, especially on weaker models.

Set the parameter's enum dynamically from the live roster (Registry.AgentTypes()),
so the schema advertises the valid set (built-ins plus any user-registered agents)
up front and an invalid type can't be sent. Falls back to a free-form string when
the roster is unavailable, so the tool never advertises an empty, un-satisfiable
enum. The runtime 'available agent types: …' error is kept as a backstop.

* test(swarm): also assert handoff to_agent_type enum picks up a custom roster type

Address review: the custom-type recheck only re-read spawnTool, so a regression
leaving handoffTool on the built-in list would have passed.
2026-06-30 20:50:11 +05:30
15 changed files with 401 additions and 38 deletions

No files matched your search

@@ -0,0 +1,49 @@
package config
import (
"os"
"testing"
)
// SetActiveProviderEnv must export the provider name so a child process (whose
// resolution reads the inherited environment via the nil-map -> os.Getenv path in
// applyEnv) resolves the SAME active provider. A blank name CLEARS it.
func TestSetActiveProviderEnvRoundTrips(t *testing.T) {
t.Setenv(ActiveProviderEnv, "") // register cleanup; restores the original after
if err := os.Unsetenv(ActiveProviderEnv); err != nil {
t.Fatalf("unset: %v", err)
}
SetActiveProviderEnv("") // blank with nothing set stays unset
if got := os.Getenv(ActiveProviderEnv); got != "" {
t.Fatalf("empty name set the env to %q, want unset", got)
}
// Regression: switching from a named provider back to an unnamed/default profile
// must CLEAR a previously exported value, not leave it stale for children.
if err := os.Setenv(ActiveProviderEnv, "stale-provider"); err != nil {
t.Fatalf("seed stale: %v", err)
}
SetActiveProviderEnv("") // blank must unset the stale value
if got := os.Getenv(ActiveProviderEnv); got != "" {
t.Fatalf("blank name left stale env %q, want cleared", got)
}
cfgStale := &FileConfig{}
applyEnv(cfgStale, nil)
if cfgStale.ActiveProvider != "" {
t.Fatalf("applyEnv still resolved stale provider %q after clear", cfgStale.ActiveProvider)
}
SetActiveProviderEnv("ollama-cloud")
if got := os.Getenv(ActiveProviderEnv); got != "ollama-cloud" {
t.Fatalf("env = %q, want ollama-cloud", got)
}
// applyEnv with a nil map is the exact path a child uses: it falls back to
// os.Getenv, so the inherited value selects the active provider.
cfg := &FileConfig{}
applyEnv(cfg, nil)
if cfg.ActiveProvider != "ollama-cloud" {
t.Fatalf("applyEnv ActiveProvider = %q, want ollama-cloud (inherited env not read)", cfg.ActiveProvider)
}
}
+21 -1
View File
@@ -431,8 +431,28 @@ func mergeProfile(base ProviderProfile, next ProviderProfile) ProviderProfile {
return base
}
// ActiveProviderEnv selects the active provider profile by name (read in applyEnv).
const ActiveProviderEnv = "ZERO_PROVIDER"
// SetActiveProviderEnv exports the active provider name to the process environment
// so a spawned child process (which inherits the environment) resolves the SAME
// provider profile — and therefore the same credentials (env key / stored key /
// OAuth) — as its parent. Without this a sub-agent re-resolves config.json's
// default provider and can land on one whose credentials don't match the parent's
// live selection, failing auth the instant it spawns. A blank name CLEARS the
// variable: switching back to an unnamed/default profile must not keep exporting a
// stale provider to children.
func SetActiveProviderEnv(name string) {
name = strings.TrimSpace(name)
if name == "" {
_ = os.Unsetenv(ActiveProviderEnv)
return
}
_ = os.Setenv(ActiveProviderEnv, name)
}
func applyEnv(cfg *FileConfig, env map[string]string) {
activeProvider := strings.TrimSpace(envValue(env, "ZERO_PROVIDER"))
activeProvider := strings.TrimSpace(envValue(env, ActiveProviderEnv))
if activeProvider != "" {
cfg.ActiveProvider = activeProvider
}
+2 -2
View File
@@ -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
}
+7
View File
@@ -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",
}
+2 -2
View File
@@ -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
}
+21 -3
View File
@@ -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
+52
View File
@@ -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) {
+2 -2
View File
@@ -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
}
+62 -10
View File
@@ -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
View File
@@ -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\"."},
},
+41
View File
@@ -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))
+4
View File
@@ -441,6 +441,8 @@ func (m model) handleModelCommand(args string) (model, string) {
m.providerProfile = nextProfile
m.provider = nextProvider
m.providerName = displayValue(nextProfile.Name, string(metadata.ProviderKind))
// Keep sub-agent child processes on the same provider we just switched to.
config.SetActiveProviderEnv(nextProfile.Name)
m.modelName = target.modelID
resetEffort := false
if m.reasoningEffort != "" && !reasoningEffortAllowed(target.reasoningEfforts, m.reasoningEffort) {
@@ -516,6 +518,8 @@ func (m model) switchProviderModel(providerName, modelID string) (model, string)
m.providerProfile = target
m.providerName = target.Name
m.modelName = target.Model
// Keep sub-agent child processes on the same provider we just switched to.
config.SetActiveProviderEnv(target.Name)
if strings.TrimSpace(m.userConfigPath) != "" {
_, _ = config.SetActiveProvider(m.userConfigPath, target.Name)
_, _ = config.SetProviderModel(m.userConfigPath, target.Name, target.Model)
+5 -7
View File
@@ -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
+9 -9
View File
@@ -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)
}
}