Compare commits

..
Author SHA1 Message Date
anandh8x cd1f1fe786 Document source-build sandbox helpers
Clarify that Linux source builds need the zero-linux-sandbox helper on PATH for native sandboxing, while macOS does not need an extra helper and Windows can self-dispatch from zero.exe.

Also document optional release-style helper builds for Linux and Windows.

Tested: git diff --check
2026-06-30 08:16:03 +05:30
14 changed files with 129 additions and 413 deletions

No files matched your search

+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) || errors.Is(err, providerio.ErrStreamStalled) {
if errors.Is(err, providerio.ErrStreamIdle) {
state.closeOpen(ctx, events)
providerio.SendEvent(ctx, events, zeroruntime.StreamEvent{
Type: zeroruntime.StreamEventError,
Error: provider.redact("provider stream error: " + providerio.StreamTimeoutMessage(err, provider.streamIdleTimeout)),
Error: provider.redact(fmt.Sprintf("provider stream error: idle timeout after %s (upstream stopped sending data)", provider.streamIdleTimeout)),
})
return
}
-7
View File
@@ -264,13 +264,6 @@ 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) || errors.Is(err, providerio.ErrStreamStalled) {
if errors.Is(err, providerio.ErrStreamIdle) {
providerio.SendEvent(ctx, events, zeroruntime.StreamEvent{
Type: zeroruntime.StreamEventError,
Error: provider.redact("provider stream error: " + providerio.StreamTimeoutMessage(err, provider.streamIdleTimeout)),
Error: provider.redact(fmt.Sprintf("provider stream error: idle timeout after %s (upstream stopped sending data)", provider.streamIdleTimeout)),
})
return
}
+3 -21
View File
@@ -59,7 +59,6 @@ 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"
@@ -87,10 +86,6 @@ 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
@@ -261,9 +256,7 @@ 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 != "" {
// 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"}
req.Reasoning = &responsesReasoning{Effort: effort}
}
return req, nil
}
@@ -389,10 +382,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) || errors.Is(err, providerio.ErrStreamStalled) {
if errors.Is(err, providerio.ErrStreamIdle) {
providerio.SendEvent(ctx, events, zeroruntime.StreamEvent{
Type: zeroruntime.StreamEventError,
Error: p.redact("provider stream error: " + providerio.StreamTimeoutMessage(err, inner.streamIdleTimeout)),
Error: p.redact(fmt.Sprintf("provider stream error: idle timeout after %s (upstream stopped sending data)", inner.streamIdleTimeout)),
})
return
}
@@ -508,17 +501,6 @@ 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,58 +485,6 @@ 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) {
+7 -16
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) || errors.Is(err, providerio.ErrStreamStalled) {
if errors.Is(err, providerio.ErrStreamIdle) {
state.flushBufferedContent(events)
state.closeBufferedOpen(events)
sendEvent(ctx, events, zeroruntime.StreamEvent{
Type: zeroruntime.StreamEventError,
Error: provider.redact("provider stream error: " + providerio.StreamTimeoutMessage(err, provider.streamIdleTimeout)),
Error: provider.redact(fmt.Sprintf("provider stream error: idle timeout after %s (upstream stopped sending data)", provider.streamIdleTimeout)),
})
return
}
@@ -385,14 +385,6 @@ func sendBufferedEvent(events chan<- zeroruntime.StreamEvent, event zeroruntime.
func (provider *Provider) openAIRequest(request zeroruntime.CompletionRequest) chatCompletionRequest {
messages := make([]chatMessage, 0, len(request.Messages))
for _, message := range request.Messages {
// Drop a degenerate assistant turn that carries neither text nor tool calls
// (e.g. a sub-agent that failed with no output). The Anthropic/Gemini mappers
// already skip empty turns; without this, the contentless message reaches
// strict OpenAI-compatible servers and is rejected.
if message.Role == zeroruntime.MessageRoleAssistant &&
strings.TrimSpace(message.Content) == "" && len(message.ToolCalls) == 0 {
continue
}
messages = append(messages, mapMessage(message))
}
@@ -453,12 +445,11 @@ func mapMessage(message zeroruntime.Message) chatMessage {
// message that happens to carry Images keeps the plain string/nil content
// path (its images are simply not serialized).
if len(message.Images) == 0 || message.Role != zeroruntime.MessageRoleUser {
// Always set content (to "" when empty) so it serializes as `"content":""`
// rather than being dropped. Strict OpenAI-compatible servers reject a
// message with no content field; tool results and assistant-with-tool-calls
// turns must still carry an (empty) content. Truly empty assistant turns are
// dropped upstream in openAIRequest.
mapped.Content = message.Content
// Preserve the legacy behavior: only non-empty text is serialized, so
// empty content is omitted by the `omitempty` tag.
if message.Content != "" {
mapped.Content = message.Content
}
} else {
parts := make([]contentPart, 0, len(message.Images)+1)
if message.Content != "" {
+85 -65
View File
@@ -1000,6 +1000,57 @@ func TestStreamCompletionEmitsDroppedOnNamelessToolCall(t *testing.T) {
}
}
func TestOpenAIRequestTextOnlyOmitsEmptyContent(t *testing.T) {
provider, err := New(Options{Model: "gpt-test"})
if err != nil {
t.Fatalf("New returned error: %v", err)
}
got, err := json.Marshal(provider.openAIRequest(zeroruntime.CompletionRequest{
Messages: []zeroruntime.Message{
{Role: zeroruntime.MessageRoleSystem, Content: "system"},
{Role: zeroruntime.MessageRoleUser, Content: "hello"},
{
Role: zeroruntime.MessageRoleAssistant,
Content: "", // assistant turn that is *only* a tool call -> empty content must drop
ToolCalls: []zeroruntime.ToolCall{{
ID: "call_1",
Name: "read_file",
Arguments: `{"path":"README.md"}`,
}},
},
{Role: zeroruntime.MessageRoleTool, Content: "contents", ToolCallID: "call_1"},
},
}))
if err != nil {
t.Fatalf("marshal request: %v", err)
}
// String content round-trips as a JSON string, and empty content is omitted.
if !strings.Contains(string(got), `"content":"hello"`) {
t.Fatalf("user content not serialized as string: %s", got)
}
if strings.Contains(string(got), `"content":""`) {
t.Fatalf("empty assistant content must be omitted, got: %s", got)
}
// Decode the assistant message and assert no content key at all.
var decoded chatCompletionRequest
var raw struct {
Messages []map[string]any `json:"messages"`
}
if err := json.Unmarshal(got, &raw); err != nil {
t.Fatalf("unmarshal request: %v", err)
}
_ = decoded
if _, present := raw.Messages[2]["content"]; present {
t.Fatalf("assistant (tool-only) message must omit content key, got: %#v", raw.Messages[2])
}
if raw.Messages[1]["content"] != "hello" {
t.Fatalf("user content = %#v, want \"hello\"", raw.Messages[1]["content"])
}
}
func TestContentPartImageURLMarshalsDataURI(t *testing.T) {
parts := []contentPart{
{Type: "text", Text: "look"},
@@ -1025,6 +1076,38 @@ func TestContentPartImageURLMarshalsDataURI(t *testing.T) {
}
}
func TestOpenAIRequestEmptyContentStaysOmittedAfterAnyRefactor(t *testing.T) {
provider, err := New(Options{Model: "gpt-test"})
if err != nil {
t.Fatalf("New returned error: %v", err)
}
got, err := json.Marshal(provider.openAIRequest(zeroruntime.CompletionRequest{
Messages: []zeroruntime.Message{
{
Role: zeroruntime.MessageRoleAssistant,
Content: "",
ToolCalls: []zeroruntime.ToolCall{{
ID: "call_1",
Name: "read_file",
Arguments: `{}`,
}},
},
},
}))
if err != nil {
t.Fatalf("marshal: %v", err)
}
var raw struct {
Messages []map[string]any `json:"messages"`
}
if err := json.Unmarshal(got, &raw); err != nil {
t.Fatalf("unmarshal: %v", err)
}
if _, present := raw.Messages[0]["content"]; present {
t.Fatalf("empty content boxed in any must stay omitted, got: %#v", raw.Messages[0])
}
}
func TestMapMessageBuildsImageURLContentParts(t *testing.T) {
msg := mapMessage(zeroruntime.Message{
Role: zeroruntime.MessageRoleUser,
@@ -1082,8 +1165,8 @@ func TestMapMessageTextOnlyKeepsStringContent(t *testing.T) {
t.Fatalf("Content = %#v, want string \"hi\"", msg.Content)
}
empty := mapMessage(zeroruntime.Message{Role: zeroruntime.MessageRoleAssistant, Content: ""})
if got, ok := empty.Content.(string); !ok || got != "" {
t.Fatalf("empty text content = %#v, want \"\" so it serializes as content:\"\" (strict servers reject a missing/null content)", empty.Content)
if empty.Content != nil {
t.Fatalf("empty text content = %#v, want nil so omitempty drops it", empty.Content)
}
}
@@ -1163,66 +1246,3 @@ func TestStreamCompletionSerializesImageContentParts(t *testing.T) {
t.Fatalf("image url = %#v, want data:image/png;base64,QUJD", imageURL["url"])
}
}
// TestOpenAIRequestEmptyContentHandling locks in the fix for strict OpenAI-
// compatible servers (e.g. glm-* on Ollama-cloud) that reject a message whose
// `content` is absent/null with "invalid message content type: <nil>": a
// contentless assistant turn with no tool calls is dropped, and every other
// message serializes an explicit `"content"` field (empty string when there is
// no text) instead of omitting it.
func TestOpenAIRequestEmptyContentHandling(t *testing.T) {
provider, err := New(Options{Model: "gpt-test"})
if err != nil {
t.Fatalf("New returned error: %v", err)
}
req := provider.openAIRequest(zeroruntime.CompletionRequest{
Messages: []zeroruntime.Message{
{Role: zeroruntime.MessageRoleUser, Content: "hi"},
// Degenerate empty assistant turn (e.g. a sub-agent that failed with no
// output): must be dropped, not sent.
{Role: zeroruntime.MessageRoleAssistant, Content: " "},
// Assistant with tool calls but no text: kept, content present as "".
{Role: zeroruntime.MessageRoleAssistant, Content: "", ToolCalls: []zeroruntime.ToolCall{{
ID: "call_1", Name: "read_file", Arguments: "{}",
}}},
// Tool result with empty content: kept, content present as "".
{Role: zeroruntime.MessageRoleTool, Content: "", ToolCallID: "call_1"},
},
})
if len(req.Messages) != 3 {
t.Fatalf("empty assistant turn should be dropped; got %d messages: %#v", len(req.Messages), req.Messages)
}
// No message may serialize with an absent/null content field.
for i, message := range req.Messages {
if message.Content == nil {
t.Fatalf("message %d has nil content (would serialize as null/omitted): %#v", i, message)
}
data, err := json.Marshal(message)
if err != nil {
t.Fatalf("marshal message %d: %v", i, err)
}
if !strings.Contains(string(data), `"content":`) {
t.Fatalf("message %d omits the content field: %s", i, data)
}
}
if req.Messages[0].Content != "hi" {
t.Fatalf("user content = %#v, want \"hi\"", req.Messages[0].Content)
}
// The kept-but-textless messages must send content as an explicit empty string.
for _, idx := range []int{1, 2} {
data, err := json.Marshal(req.Messages[idx])
if err != nil {
t.Fatalf("marshal message %d: %v", idx, err)
}
if !strings.Contains(string(data), `"content":""`) {
t.Fatalf("message %d should send content:\"\", got %s", idx, data)
}
}
if len(req.Messages[1].ToolCalls) != 1 {
t.Fatalf("assistant tool calls dropped: %#v", req.Messages[1])
}
}
+2 -7
View File
@@ -18,13 +18,8 @@ type streamOptions struct {
}
type chatMessage struct {
Role string `json:"role"`
// Content has no omitempty: strict OpenAI-compatible servers (e.g. some
// Ollama-cloud models like glm-*) reject a message whose `content` is absent
// or null with "invalid message content type: <nil>". mapMessage always sets
// this (to "" when there's no text), so a contentless message serializes as
// `"content":""` rather than being dropped.
Content any `json:"content"`
Role string `json:"role"`
Content any `json:"content,omitempty"`
ToolCalls []requestToolCall `json:"tool_calls,omitempty"`
ToolCallID string `json:"tool_call_id,omitempty"`
}
+10 -62
View File
@@ -23,26 +23,6 @@ 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
@@ -53,14 +33,6 @@ func StreamTimeoutMessage(err error, idleTimeout time.Duration) string {
// 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
@@ -230,36 +202,21 @@ 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, contentC <-chan time.Time
var idleC <-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 = 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)
resetIdle = func() {
if !idle.Stop() {
select {
case <-idle.C:
default:
}
}
idle.Reset(idleTimeout)
}
}
for {
@@ -275,12 +232,6 @@ 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,
@@ -295,11 +246,8 @@ 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,106 +201,3 @@ 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)
}
}
+2 -21
View File
@@ -103,25 +103,6 @@ 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 {
@@ -137,7 +118,7 @@ func (t *spawnTool) Parameters() tools.Schema {
return tools.Schema{
Type: "object",
Properties: map[string]tools.PropertySchema{
"agent_type": agentTypeProperty(t.sw, "Roster agent type to spawn."),
"agent_type": {Type: "string", Description: "Roster agent type to spawn (e.g. teammate, subagent)."},
"task": {Type: "string", Description: "The focused task/briefing for the member."},
"team": {Type: "string", Description: "Team to spawn into. Defaults to \"default\"."},
},
@@ -372,7 +353,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": agentTypeProperty(t.sw, "Roster agent type to take over."),
"to_agent_type": {Type: "string", Description: "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,7 +2,6 @@ package swarm
import (
"context"
"slices"
"strings"
"testing"
"time"
@@ -11,46 +10,6 @@ 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))
+7 -5
View File
@@ -261,11 +261,13 @@ func (m model) swarmSpawnedAgents() []swarmAgent {
switch a.state {
case "done", "failed", "completed", "cancelled":
a.finishing = true
// 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).
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.
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 TestSwarmAgentDropsOnOwnCompletionEvenMidRun(t *testing.T) {
func TestSwarmAgentsPersistWhileRunInFlight(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 — a member must still drop on its OWN completion
m.pending = true // run still going
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},
)
// Just finished, still within the linger window: shows briefly with a fading ✓.
m.swarmDoneAt = map[string]time.Time{"subagent-1": base.Add(-sidebarAgentLinger / 2)}
// Long past the linger window — but it must still show while pending.
m.swarmDoneAt = map[string]time.Time{"subagent-1": base.Add(-10 * sidebarAgentLinger)}
agents := m.swarmSpawnedAgents()
if len(agents) != 1 {
t.Fatalf("within the linger window a finished member should still show (fading), got %d: %+v", len(agents), agents)
t.Fatalf("a finished member must stay while the run is in flight, got %d: %+v", len(agents), agents)
}
if !agents[0].finishing {
t.Fatalf("a finished member should render done (✓), got %+v", agents[0])
}
// 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)}
// Once the turn ends, the long-finished member fades out and drops.
m.pending = false
if got := len(m.swarmSpawnedAgents()); got != 0 {
t.Fatalf("a member past its linger window must drop on its own completion mid-run, got %d", got)
t.Fatalf("after the turn ends a long-finished member should drop, got %d", got)
}
}