fix: type the agent SSE payload — render event content, end the stream on a terminal event

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, "", createTerminalSink())
 	},
 }
 
@@ -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, createTerminalSink())
 	},
 }
 
-// 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 createTerminalSink() 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

cmd/agent_stream_test.go

@@ -0,0 +1,137 @@
+/*
+ * Copyright 2026 Conductor Authors.
+ * <p>
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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)
+	}
+}

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
 }

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 {