+ * 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) + } +} 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 { 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 ]