From 3936bc91e6ca5761c7a826a4886fef753e06f6f2 Mon Sep 17 00:00:00 2001 From: francisco-orkes Date: Thu, 13 Aug 2026 12:41:59 -0700 Subject: [PATCH 1/4] fix(agent): end the SSE stream on a terminal event MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Streaming an already-terminal execution hung forever: the parse loop returned only on body EOF, and the server holds that connection open emitting nothing but heartbeat comments, which the parser discards. StreamExecution now stops once a done or error event has reached the sink, by cancelling the stream context — the client's documented shutdown path, which the service already treats as a clean stop, rather than a second shutdown mechanism. The CLI no longer depends on the server closing the connection. Cancellation is also what releases the producer: its send onto the event channel now selects on the context, so a consumer that stops reading while the buffer is full can no longer strand that goroutine. Introduces the typed EventPayload mirroring the server's AgentSSEEvent. It is what makes the terminal types checkable here, and what the renderer reads in place of untyped map lookups in the following commit. Refs #102, #116 Co-Authored-By: Claude Opus 5 (1M context) --- internal/agent/client.go | 5 +- internal/agent/client_test.go | 79 +++++++++++++++++++++++ internal/agent/events.go | 110 +++++++++++++++++++++++++++++++-- internal/agent/events_test.go | 94 +++++++++++++++++++++++++++- internal/agent/service.go | 13 ++++ internal/agent/service_test.go | 100 +++++++++++++++++++++++++++--- 6 files changed, 385 insertions(+), 16 deletions(-) diff --git a/internal/agent/client.go b/internal/agent/client.go index 157599c..08cd2d8 100644 --- a/internal/agent/client.go +++ b/internal/agent/client.go @@ -223,7 +223,8 @@ func (c *restClient) Deploy(ctx context.Context, framework string, rawConfig jso // Stream opens an SSE connection for an execution and returns an event channel and // a single-error channel. The caller ranges the events; when it closes, the error // channel carries the terminal error (nil on a clean end). Cancelling ctx ends the -// stream — that surfaces as a context error which the service treats as a clean stop. +// stream — that surfaces as a context error which the service treats as a clean stop +// — and also releases a send blocked on an event channel the caller stopped reading. func (c *restClient) Stream(ctx context.Context, executionID, lastEventID string) (<-chan SSEEvent, <-chan error) { events := make(chan SSEEvent, sseChannelBuffer) errc := make(chan error, 1) @@ -242,7 +243,7 @@ func (c *restClient) Stream(ctx context.Context, executionID, lastEventID string return } defer resp.Body.Close() - errc <- parseSSE(resp.Body, events) + errc <- parseSSE(ctx, resp.Body, events) }() return events, errc } diff --git a/internal/agent/client_test.go b/internal/agent/client_test.go index 7699cf9..7d30a94 100644 --- a/internal/agent/client_test.go +++ b/internal/agent/client_test.go @@ -17,11 +17,13 @@ import ( "context" "encoding/json" "errors" + "io" "net/http" "net/http/httptest" "net/url" "strings" "testing" + "time" "github.com/conductor-oss/conductor-cli/internal/transport" ) @@ -351,6 +353,83 @@ func TestStreamReadsSSE(t *testing.T) { } } +func TestStreamSendsLastEventID(t *testing.T) { + var got string + c := newTestClient(t, func(w http.ResponseWriter, r *http.Request) { + got = r.Header.Get(headerLastEventID) + _, _ = w.Write([]byte("event: done\ndata: {}\n\n")) + }) + + events, errc := c.Stream(context.Background(), "exec-1", "42") + for range events { //nolint:revive // drain + } + if err := <-errc; err != nil { + t.Fatalf("stream error: %v", err) + } + if got != "42" { + t.Errorf("%s = %q, want 42", headerLastEventID, got) + } +} + +// Regression guard for #102, at the seam where it actually bit: the server holds an +// already-terminal execution's connection open and sends only heartbeats, so the body +// never ends. Streaming must still return once the done event has been delivered. +func TestStreamExecutionEndsWhenServerHoldsConnectionOpen(t *testing.T) { + c := newTestClient(t, func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", mimeEventStream) + _, _ = io.WriteString(w, ":connected\n\nid: 1\nevent: done\ndata: {\"output\":\"ok\"}\n\n") + w.(http.Flusher).Flush() + <-r.Context().Done() // the server never closes the stream itself + }) + + sink := &recordingSink{} + done := make(chan error, 1) + go func() { + done <- NewService(c).StreamExecution(context.Background(), "exec-1", "", sink) + }() + + select { + case err := <-done: + if err != nil { + t.Fatalf("StreamExecution: %v", err) + } + case <-time.After(5 * time.Second): + t.Fatal("StreamExecution hung against a server that keeps the connection open") + } + if len(sink.events) != 1 || sink.events[0].Type != EventDone { + t.Fatalf("sink saw %+v, want the done event", sink.events) + } +} + +// A consumer that walks away with the channel buffer full must not strand the +// producer goroutine; cancelling the context is what releases it. +func TestStreamAbandonsSendWhenConsumerStopsReading(t *testing.T) { + c := newTestClient(t, func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", mimeEventStream) + for i := 0; i < sseChannelBuffer*2; i++ { + if _, err := io.WriteString(w, "event: message\ndata: {}\n\n"); err != nil { + return + } + } + w.(http.Flusher).Flush() + <-r.Context().Done() + }) + + ctx, cancel := context.WithCancel(context.Background()) + events, errc := c.Stream(ctx, "exec-1", "") + <-events + cancel() + + select { + case err := <-errc: + if !errors.Is(err, context.Canceled) { + t.Fatalf("stream error = %v, want context.Canceled", err) + } + case <-time.After(5 * time.Second): + t.Fatal("the producer goroutine stayed blocked on a send nobody was reading") + } +} + func keysOf(m map[string]json.RawMessage) []string { ks := make([]string, 0, len(m)) for k := range m { diff --git a/internal/agent/events.go b/internal/agent/events.go index 5a7a679..b91b7ac 100644 --- a/internal/agent/events.go +++ b/internal/agent/events.go @@ -15,6 +15,7 @@ package agent import ( "bufio" + "context" "encoding/json" "io" "strings" @@ -38,6 +39,13 @@ const ( EventDone EventType = "done" ) +// IsTerminal reports whether an event is the last one an execution can emit. The +// server keeps a terminal execution's SSE connection open indefinitely, so the +// stream has to end on the event itself rather than on a close that never comes. +func (t EventType) IsTerminal() bool { + return t == EventDone || t == EventError +} + // SSE framing tokens and limits. const ( sseMaxLineBytes = 1024 * 1024 // generous line buffer for large data frames @@ -70,34 +78,122 @@ func (e SSEEvent) ResolvedType() EventType { return "" } +// EventPayload mirrors the server's AgentSSEEvent: one envelope shared by every event +// kind, each filling only the fields it needs. Renderers read these fields instead of +// looking keys up in a map, so a field the server renames becomes a compile error +// rather than a silently blank line. +// +// Kinds the CLI does not model (context_condensed, subagent_start/stop) still decode +// into the common fields; a renderer that does not know them falls back to raw data. +type EventPayload struct { + ID int64 `json:"id"` + Type EventType `json:"type"` + ExecutionID string `json:"executionId"` + // Content carries the human-readable text of thinking, message, error and + // guardrail_fail events — the server has no separate "message" or "reason" field. + Content string `json:"content"` + // ToolName names the tool of tool_call/tool_result, and the failing task ref of error. + ToolName string `json:"toolName"` + Args RawValue `json:"args"` // tool_call arguments + Result RawValue `json:"result"` // tool_result payload + Target string `json:"target"` // handoff destination agent + Output RawValue `json:"output"` // done payload + GuardrailName string `json:"guardrailName"` + PendingTool map[string]any `json:"pendingTool"` // waiting: the tool awaiting a human + Timestamp int64 `json:"timestamp"` +} + +// RawValue is a payload field the server types as a free-form object — tool +// arguments, tool results, the final output. Keeping it raw lets the payload stay +// typed without pinning down schemas the server does not fix. +type RawValue json.RawMessage + +// UnmarshalJSON and MarshalJSON keep the bytes verbatim, as json.RawMessage does — +// a named type does not inherit its methods, and without them encoding/json would +// treat the underlying []byte as base64. +func (v *RawValue) UnmarshalJSON(data []byte) error { + *v = append((*v)[:0], data...) + return nil +} + +func (v RawValue) MarshalJSON() ([]byte, error) { + if v == nil { + return []byte("null"), nil + } + return v, nil +} + +// String renders the value for display: a JSON string yields its text, anything else +// its compact JSON encoding. Non-strings round-trip through a generic decode so keys +// are ordered deterministically regardless of the order the server sent them in. +func (v RawValue) String() string { + if len(v) == 0 { + return "" + } + var text string + if json.Unmarshal(v, &text) == nil { + return text + } + var generic any + if json.Unmarshal(v, &generic) != nil { + return string(v) + } + encoded, err := json.Marshal(generic) + if err != nil { + return string(v) + } + return string(encoded) +} + +// Payload decodes the event data into the typed payload. A malformed or empty body +// yields the zero payload rather than an error: a renderer shows what arrived, it +// does not police the wire format. +func (e SSEEvent) Payload() EventPayload { + var p EventPayload + _ = json.Unmarshal(e.Data, &p) + return p +} + // parseSSE reads a text/event-stream body and emits one SSEEvent per record onto // out, following WHATWG SSE framing: a blank line ends a record, ":" lines are // comments (heartbeats), and multi-line data is joined with newlines. It returns the // scanner error (or nil) when the stream ends; the caller owns closing out. -func parseSSE(r io.Reader, out chan<- SSEEvent) error { +// +// Every send selects on ctx, so a consumer that walks away mid-stream — the stream +// ends on a terminal event, or the user hits Ctrl-C — cannot strand this goroutine +// on a send into a full buffer. Abandoning a send returns ctx.Err(). +func parseSSE(ctx context.Context, r io.Reader, out chan<- SSEEvent) error { scanner := bufio.NewScanner(r) scanner.Buffer(make([]byte, 0, sseMaxLineBytes), sseMaxLineBytes) var id, event string var dataLines []string - flush := func() { + // flush reports whether the record was delivered; false means ctx is done. + flush := func() bool { if len(dataLines) == 0 && event == "" { - return + return true } - out <- SSEEvent{ + select { + case out <- SSEEvent{ ID: id, Type: EventType(event), Data: json.RawMessage(strings.Join(dataLines, "\n")), + }: + case <-ctx.Done(): + return false } id, event, dataLines = "", "", dataLines[:0] + return true } for scanner.Scan() { line := scanner.Text() switch { case line == "": - flush() + if !flush() { + return ctx.Err() + } case strings.HasPrefix(line, fieldComment): // comment / heartbeat — ignore case strings.HasPrefix(line, fieldID): @@ -108,7 +204,9 @@ func parseSSE(r io.Reader, out chan<- SSEEvent) error { dataLines = append(dataLines, sseFieldValue(line, fieldData)) } } - flush() + if !flush() { + return ctx.Err() + } return scanner.Err() } diff --git a/internal/agent/events_test.go b/internal/agent/events_test.go index 25db014..1329307 100644 --- a/internal/agent/events_test.go +++ b/internal/agent/events_test.go @@ -14,15 +14,19 @@ package agent import ( + "context" + "errors" + "reflect" "strings" "testing" + "time" ) func collectSSE(t *testing.T, body string) []SSEEvent { t.Helper() out := make(chan SSEEvent, 16) go func() { - _ = parseSSE(strings.NewReader(body), out) + _ = parseSSE(context.Background(), strings.NewReader(body), out) close(out) }() var got []SSEEvent @@ -61,3 +65,91 @@ func TestResolvedTypeFallsBackToDataType(t *testing.T) { t.Errorf("ResolvedType = %q, want thinking", e.ResolvedType()) } } + +// Regression guard for #102: a consumer that walks away must not strand the parser +// on a send. Nobody reads out here, so the send can only complete via cancellation. +func TestParseSSEAbandonsSendWhenContextIsDone(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + done := make(chan error, 1) + go func() { + done <- parseSSE(ctx, strings.NewReader("event: done\ndata: {}\n\n"), make(chan SSEEvent)) + }() + + select { + case err := <-done: + if !errors.Is(err, context.Canceled) { + t.Fatalf("parseSSE = %v, want context.Canceled", err) + } + case <-time.After(2 * time.Second): + t.Fatal("parseSSE stayed blocked on a send after the context was cancelled") + } +} + +func TestIsTerminal(t *testing.T) { + for _, tt := range []struct { + typ EventType + want bool + }{ + {EventDone, true}, + {EventError, true}, + {EventMessage, false}, + {EventThinking, false}, + {"", false}, + } { + if got := tt.typ.IsTerminal(); got != tt.want { + t.Errorf("%q.IsTerminal() = %v, want %v", tt.typ, got, tt.want) + } + } +} + +// The payload mirrors the server's AgentSSEEvent; this pins the wire names down. +func TestPayloadDecodesServerFieldNames(t *testing.T) { + e := SSEEvent{Data: []byte(`{"id":7,"type":"tool_call","executionId":"exec-1", + "content":"why","toolName":"lookup","args":{"q":"x"},"result":"ok", + "target":"billing","output":{"result":"done"},"guardrailName":"pii","timestamp":42}`)} + + p := e.Payload() + if p.ID != 7 || p.Type != "tool_call" || p.ExecutionID != "exec-1" || p.Timestamp != 42 { + t.Errorf("envelope = %+v", p) + } + if p.Content != "why" || p.ToolName != "lookup" || p.Target != "billing" || p.GuardrailName != "pii" { + t.Errorf("text fields = %+v", p) + } + if got := p.Args.String(); got != `{"q":"x"}` { + t.Errorf("args = %q", got) + } + if got := p.Result.String(); got != "ok" { + t.Errorf("result = %q", got) + } + if got := p.Output.String(); got != `{"result":"done"}` { + t.Errorf("output = %q", got) + } +} + +func TestPayloadOfMalformedDataIsZero(t *testing.T) { + if p := (SSEEvent{Data: []byte("not json")}).Payload(); !reflect.DeepEqual(p, EventPayload{}) { + t.Errorf("payload = %+v, want zero", p) + } +} + +func TestRawValueString(t *testing.T) { + for _, tt := range []struct { + name string + in RawValue + want string + }{ + {"absent", nil, ""}, + {"string", RawValue(`"hello"`), "hello"}, + {"number", RawValue(`3`), "3"}, + {"object keys sorted", RawValue(`{"b":1,"a":2}`), `{"a":2,"b":1}`}, + {"not json", RawValue(`{oops`), "{oops"}, + } { + t.Run(tt.name, func(t *testing.T) { + if got := tt.in.String(); got != tt.want { + t.Errorf("String() = %q, want %q", got, tt.want) + } + }) + } +} diff --git a/internal/agent/service.go b/internal/agent/service.go index ad9e5b5..27c5ac1 100644 --- a/internal/agent/service.go +++ b/internal/agent/service.go @@ -88,12 +88,25 @@ func (s *service) Deploy(ctx context.Context, framework string, rawConfig json.R // StreamExecution streams an execution's events into the sink until the stream ends. // A context cancellation (e.g. Ctrl-C) is treated as a clean stop, not an error. +// +// A terminal event ends the stream from this side, once the sink has seen it. The +// server holds an already-terminal execution's connection open indefinitely and only +// sends heartbeats, so waiting for the body to end hangs forever. Ending it is a +// cancellation like any other, which is also what releases the client's producer +// goroutine from a send this loop is no longer reading. func (s *service) StreamExecution(ctx context.Context, executionID, lastEventID string, sink EventSink) error { + ctx, cancel := context.WithCancel(ctx) + defer cancel() + events, errc := s.client.Stream(ctx, executionID, lastEventID) for evt := range events { if err := sink.OnEvent(evt); err != nil { return err } + if evt.ResolvedType().IsTerminal() { + cancel() + break + } } if err := <-errc; err != nil && !errors.Is(err, context.Canceled) { return err diff --git a/internal/agent/service_test.go b/internal/agent/service_test.go index 3e4b9a5..386a48b 100644 --- a/internal/agent/service_test.go +++ b/internal/agent/service_test.go @@ -18,6 +18,7 @@ import ( "encoding/json" "errors" "testing" + "time" ) // fakeClient records calls and returns canned values; it implements Client so the @@ -31,6 +32,13 @@ type fakeClient struct { deployResult DeployResult streamEvents []SSEEvent streamErr error + // streamBuffer sizes the event channel; 0 means "big enough for every event", + // so only tests that care about a blocked producer have to set it. + streamBuffer int + // streamDone closes when the producer goroutine exits, so a test can prove it + // was not stranded on a send. + streamDone chan struct{} + lastStreamEventID string } func (f *fakeClient) CheckSupported(ctx context.Context) error { return nil } @@ -46,15 +54,33 @@ func (f *fakeClient) Deploy(ctx context.Context, framework string, rawConfig jso return f.deployResult, nil } +// Stream mirrors the real client's producer: it feeds events through a bounded +// channel from its own goroutine and abandons a send once ctx is done, so a consumer +// that stops reading early is visible to tests instead of silently deadlocking. func (f *fakeClient) Stream(ctx context.Context, id, lastEventID string) (<-chan SSEEvent, <-chan error) { - events := make(chan SSEEvent, len(f.streamEvents)) - errc := make(chan error, 1) - for _, e := range f.streamEvents { - events <- e + f.lastStreamEventID = lastEventID + buffer := f.streamBuffer + if buffer <= 0 { + buffer = len(f.streamEvents) } - close(events) - errc <- f.streamErr - close(errc) + events := make(chan SSEEvent, buffer) + errc := make(chan error, 1) + f.streamDone = make(chan struct{}) + + go func() { + defer close(f.streamDone) + defer close(errc) + defer close(events) + for _, e := range f.streamEvents { + select { + case events <- e: + case <-ctx.Done(): + errc <- ctx.Err() + return + } + } + errc <- f.streamErr + }() return events, errc } @@ -144,6 +170,66 @@ func TestStreamExecutionForwardsEventsAndIgnoresCancel(t *testing.T) { } } +// Regression guard for #102: the stream must end on the terminal event itself, +// because the server never closes an already-terminal execution's connection. +func TestStreamExecutionStopsOnTerminalEvent(t *testing.T) { + for _, terminal := range []EventType{EventDone, EventError} { + t.Run(string(terminal), func(t *testing.T) { + fc := &fakeClient{streamEvents: []SSEEvent{ + {Type: EventMessage}, + {Type: terminal}, + {Type: EventMessage}, // the server would keep the connection open here + }} + sink := &recordingSink{} + + done := make(chan error, 1) + go func() { done <- NewService(fc).StreamExecution(context.Background(), "exec-1", "", sink) }() + + select { + case err := <-done: + if err != nil { + t.Fatalf("StreamExecution: %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("StreamExecution did not return after the terminal event") + } + if len(sink.events) != 2 || sink.events[1].Type != terminal { + t.Fatalf("sink saw %+v, want the message and the terminal event", sink.events) + } + }) + } +} + +// The producer must not be left blocked on a send once the consumer stops reading — +// hence more trailing events than the channel buffer can hold. +func TestStreamExecutionLeavesNoBlockedProducer(t *testing.T) { + fc := &fakeClient{ + streamEvents: []SSEEvent{ + {Type: EventDone}, + {Type: EventMessage}, {Type: EventMessage}, {Type: EventMessage}, + }, + streamBuffer: 1, + } + if err := NewService(fc).StreamExecution(context.Background(), "exec-1", "", &recordingSink{}); err != nil { + t.Fatalf("StreamExecution: %v", err) + } + select { + case <-fc.streamDone: + case <-time.After(2 * time.Second): + t.Fatal("producer goroutine is still blocked after the stream ended") + } +} + +func TestStreamExecutionPassesLastEventID(t *testing.T) { + fc := &fakeClient{streamEvents: []SSEEvent{{Type: EventDone}}} + if err := NewService(fc).StreamExecution(context.Background(), "exec-1", "42", &recordingSink{}); err != nil { + t.Fatalf("StreamExecution: %v", err) + } + if fc.lastStreamEventID != "42" { + t.Errorf("last event id = %q, want 42", fc.lastStreamEventID) + } +} + func TestStreamExecutionReturnsRealError(t *testing.T) { fc := &fakeClient{streamErr: errors.New("boom")} if err := NewService(fc).StreamExecution(context.Background(), "exec-1", "", &recordingSink{}); err == nil { From 3067304d3ed2893ca465b0b8d56d9f3ecb1dfe36 Mon Sep 17 00:00:00 2001 From: francisco-orkes Date: Thu, 13 Aug 2026 12:42:05 -0700 Subject: [PATCH 2/4] fix(cli): render streamed agent events from the typed payload MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The sink looked payload keys up by string literal, and several of those keys are not in what the server sends. thinking and error both read "message" where the server sends "content", so a failed run printed "[error]" with no reason at all. Tool-call arguments arrive as "args", not "input", and a handoff names its destination "target", not "agentName" — both confirmed against the server's AgentSSEEvent. guardrail_fail has no "reason" field either; the server puts the failure detail in "content", so that is what renders, and a payload without one leaves the guardrail name standing rather than trailing an empty separator. Reading the typed payload means the next server-side rename is a compile error instead of a blank line. terminalSink writes to an io.Writer so each event type's rendering is covered by a test. Refs #116 Co-Authored-By: Claude Opus 5 (1M context) --- cmd/agent_stream.go | 72 ++++++++++---------- cmd/agent_stream_test.go | 137 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 173 insertions(+), 36 deletions(-) create mode 100644 cmd/agent_stream_test.go diff --git a/cmd/agent_stream.go b/cmd/agent_stream.go index c3942c3..df3fe10 100644 --- a/cmd/agent_stream.go +++ b/cmd/agent_stream.go @@ -14,8 +14,8 @@ package cmd import ( - "encoding/json" "fmt" + "io" "os" "os/signal" "strings" @@ -79,7 +79,7 @@ events in real time. Use --no-stream to start it and just print the execution id fmt.Println() ctx, stop := signal.NotifyContext(cmd.Context(), os.Interrupt) defer stop() - return svc.StreamExecution(ctx, exec.ID, "", terminalSink{}) + return svc.StreamExecution(ctx, exec.ID, "", newTerminalSink()) }, } @@ -93,65 +93,65 @@ var agentStreamCmd = &cobra.Command{ RunE: func(cmd *cobra.Command, args []string) error { ctx, stop := signal.NotifyContext(cmd.Context(), os.Interrupt) defer stop() - return internal.GetAgentService().StreamExecution(ctx, args[0], streamLastEventID, terminalSink{}) + return internal.GetAgentService().StreamExecution(ctx, args[0], streamLastEventID, newTerminalSink()) }, } -// terminalSink renders streamed agent events to stdout. It is the cmd-layer -// presentation of agent.EventSink; the service and client know nothing about it. -type terminalSink struct{} +// terminalSink renders streamed agent events to a writer, stdout in production. It is +// the cmd-layer presentation of agent.EventSink; the service and client know nothing +// about it. Every field it reads comes from the typed payload, so a server-side rename +// breaks the build instead of quietly printing a blank line. +type terminalSink struct { + w io.Writer +} + +func newTerminalSink() terminalSink { + return terminalSink{w: os.Stdout} +} -func (terminalSink) OnEvent(e agent.SSEEvent) error { - data := map[string]any{} - _ = json.Unmarshal(e.Data, &data) +func (s terminalSink) OnEvent(e agent.SSEEvent) error { + p := e.Payload() switch e.ResolvedType() { case agent.EventThinking: - fmt.Printf(" [thinking] %s\n", truncate(mapStr(data, "message"), truncThinking)) + fmt.Fprintf(s.w, " [thinking] %s\n", truncate(p.Content, truncThinking)) case agent.EventToolCall: - fmt.Printf(" [tool] %s(%s)\n", mapStr(data, "toolName"), truncate(mapStr(data, "input"), truncToolInput)) + fmt.Fprintf(s.w, " [tool] %s(%s)\n", p.ToolName, truncate(p.Args.String(), truncToolInput)) case agent.EventToolResult: - fmt.Printf(" [result] %s -> %s\n", mapStr(data, "toolName"), truncate(mapStr(data, "result"), truncToolResult)) + fmt.Fprintf(s.w, " [result] %s -> %s\n", p.ToolName, truncate(p.Result.String(), truncToolResult)) case agent.EventHandoff: - fmt.Printf(" [handoff] -> %s\n", mapStr(data, "agentName")) + fmt.Fprintf(s.w, " [handoff] -> %s\n", p.Target) case agent.EventMessage: - if content := mapStr(data, "content"); content != "" { - fmt.Print(content) + if p.Content != "" { + fmt.Fprint(s.w, p.Content) } case agent.EventWaiting: - fmt.Printf(" [waiting] human input required (execution: %s)\n", mapStr(data, "executionId")) + fmt.Fprintf(s.w, " [waiting] human input required (execution: %s)\n", p.ExecutionID) case agent.EventGuardrailPass: - fmt.Printf(" [guardrail] PASS %s\n", mapStr(data, "guardrailName")) + fmt.Fprintf(s.w, " [guardrail] PASS %s\n", p.GuardrailName) case agent.EventGuardrailFail: - fmt.Printf(" [guardrail] FAIL %s: %s\n", mapStr(data, "guardrailName"), mapStr(data, "reason")) + // The failure detail rides in content; a server that omits it leaves the + // name standing alone rather than trailing an empty separator. + if p.Content != "" { + fmt.Fprintf(s.w, " [guardrail] FAIL %s: %s\n", p.GuardrailName, p.Content) + } else { + fmt.Fprintf(s.w, " [guardrail] FAIL %s\n", p.GuardrailName) + } case agent.EventError: - fmt.Printf(" [error] %s\n", mapStr(data, "message")) + fmt.Fprintf(s.w, " [error] %s\n", p.Content) case agent.EventDone: - if out := mapStr(data, "output"); out != "" { - fmt.Println() - fmt.Println(out) + if out := p.Output.String(); out != "" { + fmt.Fprintln(s.w) + fmt.Fprintln(s.w, out) } default: if t := e.ResolvedType(); t != "" { - fmt.Printf(" [%s] %s\n", t, truncate(string(e.Data), truncEventData)) + fmt.Fprintf(s.w, " [%s] %s\n", t, truncate(string(e.Data), truncEventData)) } } return nil } -// mapStr returns a string field from a decoded event payload, JSON-encoding non-string values. -func mapStr(data map[string]any, key string) string { - v, ok := data[key] - if !ok { - return "" - } - if s, ok := v.(string); ok { - return s - } - b, _ := json.Marshal(v) - return string(b) -} - func truncate(s string, max int) string { if len(s) <= max { return s diff --git a/cmd/agent_stream_test.go b/cmd/agent_stream_test.go new file mode 100644 index 0000000..c4357d1 --- /dev/null +++ b/cmd/agent_stream_test.go @@ -0,0 +1,137 @@ +/* + * Copyright 2026 Conductor Authors. + *

+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + *

+ * http://www.apache.org/licenses/LICENSE-2.0 + *

+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on + * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the + * specific language governing permissions and limitations under the License. + */ + +package cmd + +import ( + "bytes" + "strings" + "testing" + + "github.com/conductor-oss/conductor-cli/internal/agent" +) + +// Regression guard for #116: every payload below is in the server's AgentSSEEvent +// shape, so a renderer reading a field the server does not send shows up as a blank +// line here rather than in someone's terminal. +func TestTerminalSinkRendersServerPayloads(t *testing.T) { + tests := []struct { + name string + typ agent.EventType + data string + want string + }{ + { + name: "thinking", + typ: agent.EventThinking, + data: `{"type":"thinking","executionId":"e1","content":"probe_agent_llm"}`, + want: " [thinking] probe_agent_llm\n", + }, + { + name: "tool_call", + typ: agent.EventToolCall, + data: `{"type":"tool_call","toolName":"lookup","args":{"q":"orders"}}`, + want: " [tool] lookup({\"q\":\"orders\"})\n", + }, + { + name: "tool_result", + typ: agent.EventToolResult, + data: `{"type":"tool_result","toolName":"lookup","result":"3 orders"}`, + want: " [result] lookup -> 3 orders\n", + }, + { + name: "handoff", + typ: agent.EventHandoff, + data: `{"type":"handoff","target":"billing_agent"}`, + want: " [handoff] -> billing_agent\n", + }, + { + name: "message", + typ: agent.EventMessage, + data: `{"type":"message","content":"Yes."}`, + want: "Yes.", + }, + { + name: "waiting", + typ: agent.EventWaiting, + data: `{"type":"waiting","executionId":"e1","pendingTool":{"name":"approve"}}`, + want: " [waiting] human input required (execution: e1)\n", + }, + { + name: "guardrail_pass", + typ: agent.EventGuardrailPass, + data: `{"type":"guardrail_pass","guardrailName":"pii"}`, + want: " [guardrail] PASS pii\n", + }, + { + name: "guardrail_fail carries its detail in content", + typ: agent.EventGuardrailFail, + data: `{"type":"guardrail_fail","guardrailName":"pii","content":"found an email address"}`, + want: " [guardrail] FAIL pii: found an email address\n", + }, + { + name: "guardrail_fail without detail leaves no dangling separator", + typ: agent.EventGuardrailFail, + data: `{"type":"guardrail_fail","guardrailName":"pii"}`, + want: " [guardrail] FAIL pii\n", + }, + { + name: "error", + typ: agent.EventError, + data: `{"type":"error","toolName":"agent_llm","content":"model call failed: 429"}`, + want: " [error] model call failed: 429\n", + }, + { + name: "done", + typ: agent.EventDone, + data: `{"type":"done","output":{"result":"Yes.","finishReason":"STOP"}}`, + want: "\n{\"finishReason\":\"STOP\",\"result\":\"Yes.\"}\n", + }, + { + name: "unmodelled type falls back to raw data", + typ: "context_condensed", + data: `{"type":"context_condensed","content":"token_limit"}`, + want: " [context_condensed] {\"type\":\"context_condensed\",\"content\":\"token_limit\"}\n", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var out bytes.Buffer + sink := terminalSink{w: &out} + + if err := sink.OnEvent(agent.SSEEvent{Type: tt.typ, Data: []byte(tt.data)}); err != nil { + t.Fatalf("OnEvent: %v", err) + } + if got := out.String(); got != tt.want { + t.Errorf("rendered %q, want %q", got, tt.want) + } + }) + } +} + +func TestTerminalSinkTruncatesLongFields(t *testing.T) { + var out bytes.Buffer + sink := terminalSink{w: &out} + long := strings.Repeat("x", truncThinking+10) + + if err := sink.OnEvent(agent.SSEEvent{ + Type: agent.EventThinking, + Data: []byte(`{"type":"thinking","content":"` + long + `"}`), + }); err != nil { + t.Fatalf("OnEvent: %v", err) + } + if want := " [thinking] " + strings.Repeat("x", truncThinking) + "...\n"; out.String() != want { + t.Errorf("rendered %q, want %q", out.String(), want) + } +} From 9f0830a13c1e175378a98f96462ec5a66f6179fb Mon Sep 17 00:00:00 2001 From: francisco-orkes Date: Thu, 13 Aug 2026 12:42:11 -0700 Subject: [PATCH 3/4] test(e2e): unskip the agent stream termination guard MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The guard for #102 now passes: streaming a just-completed execution returns as soon as the done event arrives. run_bounded needed fixing to make the assertion meaningful. Under bats' errexit, an unguarded `wait` on the child it had just killed aborted the test with the signal status before the helper could return 124, and the `rc=$?` at the call site was equally unreachable — so the timeout the test checks for could never be observed. Refs #102 Co-Authored-By: Claude Opus 5 (1M context) --- test/e2e/agent.bats | 18 +++++++++++------- 1 file changed, 11 insertions(+), 7 deletions(-) diff --git a/test/e2e/agent.bats b/test/e2e/agent.bats index bd33457..f588aa9 100644 --- a/test/e2e/agent.bats +++ b/test/e2e/agent.bats @@ -64,19 +64,24 @@ EOF } # Helper: run a command with a portable time bound (macOS has no GNU timeout). -# Returns 124 on timeout, mirroring timeout(1). +# Returns 124 on timeout, mirroring timeout(1). Every wait is guarded, because a +# bare `wait` on a killed child aborts the test under bats' errexit with the signal +# status — which would hide the timeout the caller is asking about. run_bounded() { local secs="$1"; shift "$@" >"$BATS_TEST_TMPDIR/bounded.out" 2>&1 & local pid=$! - local i=0 + local i=0 rc=0 while [ "$i" -lt "$secs" ]; do - kill -0 "$pid" 2>/dev/null || { wait "$pid"; return $?; } + if ! kill -0 "$pid" 2>/dev/null; then + wait "$pid" 2>/dev/null || rc=$? + return "$rc" + fi sleep 1 i=$((i + 1)) done kill "$pid" 2>/dev/null - wait "$pid" 2>/dev/null + wait "$pid" 2>/dev/null || true return 124 } @@ -281,15 +286,14 @@ run_bounded() { # the fix lands. # bats test_tags=tier:nightly,needs:llm @test "17. Agent stream exits after the terminal event" { - skip "known broken: #102 — agent stream hangs on a terminal execution" require_llm write_agent_config "$BATS_TEST_TMPDIR/run5.yaml" out=$(./conductor agent run --config "$BATS_TEST_TMPDIR/run5.yaml" "Reply with one word" 2>&1) eid=$(echo "$out" | grep -oE '[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}' | head -1) [ -n "$eid" ] - run_bounded 30 ./conductor agent stream "$eid" - rc=$? + rc=0 + run_bounded 30 ./conductor agent stream "$eid" || rc=$? echo "stream exit: $rc" cat "$BATS_TEST_TMPDIR/bounded.out" || true [ "$rc" -ne 124 ] From cc9a51373a804805e9d16810fe92be9415fcd8c7 Mon Sep 17 00:00:00 2001 From: francisco-orkes Date: Fri, 14 Aug 2026 14:14:43 -0700 Subject: [PATCH 4/4] refactor(cli): name the terminal sink constructor for what it does createTerminalSink reads as the verb it is; new- said only that a value came back, which is the ambiguity raised in review on #116. Refs #116 Co-Authored-By: Claude Opus 5 (1M context) --- cmd/agent_stream.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/cmd/agent_stream.go b/cmd/agent_stream.go index df3fe10..60e1524 100644 --- a/cmd/agent_stream.go +++ b/cmd/agent_stream.go @@ -79,7 +79,7 @@ events in real time. Use --no-stream to start it and just print the execution id fmt.Println() ctx, stop := signal.NotifyContext(cmd.Context(), os.Interrupt) defer stop() - return svc.StreamExecution(ctx, exec.ID, "", newTerminalSink()) + return svc.StreamExecution(ctx, exec.ID, "", createTerminalSink()) }, } @@ -93,7 +93,7 @@ var agentStreamCmd = &cobra.Command{ RunE: func(cmd *cobra.Command, args []string) error { ctx, stop := signal.NotifyContext(cmd.Context(), os.Interrupt) defer stop() - return internal.GetAgentService().StreamExecution(ctx, args[0], streamLastEventID, newTerminalSink()) + return internal.GetAgentService().StreamExecution(ctx, args[0], streamLastEventID, createTerminalSink()) }, } @@ -105,7 +105,7 @@ type terminalSink struct { w io.Writer } -func newTerminalSink() terminalSink { +func createTerminalSink() terminalSink { return terminalSink{w: os.Stdout} }