feat(agent): add Session.Steer to deliver mid-run user messages

server/pkg/agent/agent.go

@@ -111,6 +111,16 @@ type Session struct {
 	Messages <-chan Message
 	// Result receives exactly one value — the final outcome — then closes.
 	Result <-chan Result
+	// Steer, when non-nil, delivers an additional user message into the
+	// LIVE session while the agent is still working. The message is queued
+	// on the agent's own input stream, so it is considered as the next user
+	// turn rather than interrupting the in-flight one. Nil means the backend
+	// has no mid-run input channel — callers must treat that as "not
+	// supported" and fall back to their existing follow-up-task path, never
+	// as an error. Steer returns an error once the session has finished (or
+	// its input stream is closed); the caller owns that same fallback then,
+	// so a message is never silently dropped. Safe for concurrent use.
+	Steer func(text string) error
 }
 
 // MessageType identifies the kind of Message.

server/pkg/agent/claude.go

@@ -111,6 +111,12 @@ func (b *claudeBackend) Execute(ctx context.Context, prompt string, opts ExecOpt
 	}
 	var closeStdinOnce sync.Once
 	closeStdin := func() { closeStdinOnce.Do(func() { _ = stdin.Close() }) }
+	// Every stream-json frame we send — initial prompt, control responses,
+	// and mid-run Steer messages — goes through this locked writer. Each
+	// frame is a single Write call, so the lock is what guarantees frames
+	// from different goroutines (prompt writer, scanner's control responses,
+	// Steer callers) can never interleave mid-frame on the pipe.
+	stdinW := &lockedFrameWriter{w: stdin}
 	// Capture stderr into both the daemon log (as before) and a bounded tail
 	// buffer so we can include the last few KB in Result.Error when claude
 	// exits unexpectedly. Without the tail, an exit-code-only failure looks
@@ -152,10 +158,16 @@ func (b *claudeBackend) Execute(ctx context.Context, prompt string, opts ExecOpt
 	// leaves the child stuck waiting for a response until its own fallback
 	// timeout.
 	writeDone := make(chan error, 1)
+	// initialWritten gates Steer: a steer frame must never reach claude
+	// before the initial prompt frame, or the steer text would become the
+	// task prompt. Closed only on a successful prompt write.
+	initialWritten := make(chan struct{})
 	go func() {
-		err := writeClaudeInput(stdin, prompt)
+		err := writeClaudeInput(stdinW, prompt)
 		if err != nil {
 			closeStdin()
+		} else {
+			close(initialWritten)
 		}
 		writeDone <- err
 	}()
@@ -268,7 +280,7 @@ func (b *claudeBackend) Execute(ctx context.Context, prompt string, opts ExecOpt
 					})
 				}
 			case "control_request":
-				b.handleControlRequest(msg, stdin)
+				b.handleControlRequest(msg, stdinW)
 			}
 		}
 		scanErr := scanner.Err()
@@ -361,7 +373,31 @@ func (b *claudeBackend) Execute(ctx context.Context, prompt string, opts ExecOpt
 		}
 	}()
 
-	return &Session{Messages: msgCh, Result: resCh}, nil
+	steer := func(text string) error {
+		// Never let a steer frame overtake the initial prompt; if the run
+		// ends (or is cancelled) before the prompt ever lands, report it so
+		// the caller can fall back to its follow-up-task path.
+		select {
+		case <-initialWritten:
+		case <-procDone:
+			return fmt.Errorf("claude session already finished")
+		case <-runCtx.Done():
+			return fmt.Errorf("claude session cancelled: %w", runCtx.Err())
+		}
+		select {
+		case <-procDone:
+			return fmt.Errorf("claude session already finished")
+		default:
+		}
+		if err := writeClaudeInput(stdinW, text); err != nil {
+			// Includes the closed-stdin case after the final result event —
+			// the run is wrapping up and can no longer accept input.
+			return fmt.Errorf("steer claude session: %w", err)
+		}
+		return nil
+	}
+
+	return &Session{Messages: msgCh, Result: resCh, Steer: steer}, nil
 }
 
 func (b *claudeBackend) handleAssistant(msg claudeSDKMessage, ch chan<- Message, usage map[string]TokenUsage) (string, int) {
@@ -725,6 +761,21 @@ func buildClaudeArgs(opts ExecOptions, logger *slog.Logger) []string {
 	return args
 }
 
+// lockedFrameWriter serialises whole-frame writes to claude's stdin. Every
+// producer (initial prompt, control responses, Steer) writes one complete
+// newline-terminated frame per Write call; the mutex makes that write atomic
+// with respect to the other producers.
+type lockedFrameWriter struct {
+	mu sync.Mutex
+	w  io.Writer
+}
+
+func (lw *lockedFrameWriter) Write(p []byte) (int, error) {
+	lw.mu.Lock()
+	defer lw.mu.Unlock()
+	return lw.w.Write(p)
+}
+
 func writeClaudeInput(w io.Writer, prompt string) error {
 	data, err := buildClaudeInput(prompt)
 	if err != nil {

server/pkg/agent/claude_deadlock_test.go

@@ -31,6 +31,9 @@ func TestMain(m *testing.M) {
 	case "async_launched_tool_result":
 		runFakeClaudeAsyncLaunchedToolResult()
 		os.Exit(0)
+	case "steer_echo":
+		runFakeClaudeSteerEcho()
+		os.Exit(0)
 	default:
 		fmt.Fprintf(os.Stderr, "unknown CLAUDE_FAKE_MODE: %q\n", mode)
 		os.Exit(2)

server/pkg/agent/claude_steer_test.go

@@ -0,0 +1,168 @@
+package agent
+
+import (
+	"bufio"
+	"context"
+	"encoding/json"
+	"fmt"
+	"log/slog"
+	"os"
+	"strings"
+	"testing"
+	"time"
+)
+
+// runFakeClaudeSteerEcho reads the initial prompt frame, emits an assistant
+// message (so the caller can prove the run is live before steering), then
+// blocks reading a SECOND stdin frame — the steer message — and echoes its
+// text into the final result. A backend that never delivers the steer frame
+// hangs here until the test deadline instead of passing.
+func runFakeClaudeSteerEcho() {
+	reader := bufio.NewReader(os.Stdin)
+	if _, err := reader.ReadString('\n'); err != nil {
+		fmt.Fprintf(os.Stderr, "read prompt: %v\n", err)
+		os.Exit(51)
+	}
+	fmt.Println(`{"type":"system","session_id":"sess-steer"}`)
+	fmt.Println(`{"type":"assistant","message":{"role":"assistant","content":[{"type":"text","text":"working on it"}]}}`)
+
+	line, err := reader.ReadString('\n')
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "read steer frame: %v\n", err)
+		os.Exit(52)
+	}
+	var frame struct {
+		Type    string `json:"type"`
+		Message struct {
+			Role    string `json:"role"`
+			Content []struct {
+				Type string `json:"type"`
+				Text string `json:"text"`
+			} `json:"content"`
+		} `json:"message"`
+	}
+	if err := json.Unmarshal([]byte(strings.TrimSpace(line)), &frame); err != nil {
+		fmt.Fprintf(os.Stderr, "decode steer frame: %v\n", err)
+		os.Exit(53)
+	}
+	if frame.Type != "user" || frame.Message.Role != "user" || len(frame.Message.Content) != 1 {
+		fmt.Fprintf(os.Stderr, "unexpected steer frame shape: %s\n", line)
+		os.Exit(54)
+	}
+	result := map[string]any{
+		"type":       "result",
+		"subtype":    "success",
+		"is_error":   false,
+		"session_id": "sess-steer",
+		"result":     "steered: " + frame.Message.Content[0].Text,
+	}
+	data, err := json.Marshal(result)
+	if err != nil {
+		os.Exit(55)
+	}
+	fmt.Println(string(data))
+}
+
+// TestClaudeSteerDeliversMidRunUserMessage exercises the full Steer path
+// against a fake child: the steer text must arrive as a well-formed user
+// stream-json frame on the SAME stdin, after the initial prompt, while the
+// run is live.
+func TestClaudeSteerDeliversMidRunUserMessage(t *testing.T) {
+	t.Parallel()
+
+	self, err := os.Executable()
+	if err != nil {
+		t.Fatalf("os.Executable: %v", err)
+	}
+
+	backend, err := New("claude", Config{
+		ExecutablePath: self,
+		Env:            map[string]string{"CLAUDE_FAKE_MODE": "steer_echo", "IS_SANDBOX": "1"},
+		Logger:         slog.Default(),
+	})
+	if err != nil {
+		t.Fatalf("new claude backend: %v", err)
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	session, err := backend.Execute(ctx, "original prompt", ExecOptions{})
+	if err != nil {
+		t.Fatalf("execute: %v", err)
+	}
+	if session.Steer == nil {
+		t.Fatal("claude session must advertise Steer support")
+	}
+
+	// Wait until the fake has emitted its first assistant message, proving
+	// the run is genuinely mid-flight when we steer.
+	sawAssistant := false
+	for msg := range session.Messages {
+		if msg.Type == MessageText && msg.Content == "working on it" {
+			sawAssistant = true
+			if err := session.Steer("switch to plan B"); err != nil {
+				t.Fatalf("steer: %v", err)
+			}
+		}
+	}
+	if !sawAssistant {
+		t.Fatal("never saw the fake's assistant message")
+	}
+
+	result := <-session.Result
+	if result.Error != "" {
+		t.Fatalf("unexpected result error: %q", result.Error)
+	}
+	if result.Output != "steered: switch to plan B" {
+		t.Fatalf("steer text did not round-trip through the live session, got output %q", result.Output)
+	}
+
+	// The session is finished; a late steer must fail loudly so callers
+	// fall back to their follow-up-task path instead of silently dropping
+	// the message.
+	if err := session.Steer("too late"); err == nil {
+		t.Fatal("steer after completion must return an error")
+	}
+}
+
+// TestClaudeSteerAfterCancelFails pins the cancelled-run behaviour: once the
+// context is done and the child killed, Steer reports an error rather than
+// writing into a dead pipe.
+func TestClaudeSteerAfterCancelFails(t *testing.T) {
+	t.Parallel()
+
+	self, err := os.Executable()
+	if err != nil {
+		t.Fatalf("os.Executable: %v", err)
+	}
+
+	backend, err := New("claude", Config{
+		ExecutablePath: self,
+		// The fake blocks waiting for a steer frame we never send, so the
+		// run is alive until cancel fires.
+		Env:    map[string]string{"CLAUDE_FAKE_MODE": "steer_echo", "IS_SANDBOX": "1"},
+		Logger: slog.Default(),
+	})
+	if err != nil {
+		t.Fatalf("new claude backend: %v", err)
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	session, err := backend.Execute(ctx, "original prompt", ExecOptions{})
+	if err != nil {
+		t.Fatalf("execute: %v", err)
+	}
+
+	for msg := range session.Messages {
+		if msg.Type == MessageText && msg.Content == "working on it" {
+			cancel()
+		}
+	}
+	<-session.Result
+
+	if err := session.Steer("into the void"); err == nil {
+		t.Fatal("steer after cancellation must return an error")
+	}
+}