From f07b6dadb917eee72953f7309e6b3f99960f93fe Mon Sep 17 00:00:00 2001 From: Jaisev Sachdev Date: Mon, 3 Aug 2026 16:04:00 +0800 Subject: [PATCH] feat(agent): add Session.Steer to deliver mid-run user messages MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Multica tasks are fire-and-forget today: once an agent starts on an issue, there is no way to talk to it until the run ends. A comment posted mid-run is (correctly, and with care — #5914) deferred and replayed as a follow-up task after completion. This adds the missing capability layer underneath a future live-delivery path: an optional, additive Steer hook on agent.Session that injects an additional user message into the RUNNING session. The claude backend can support this for free: it already runs --input-format stream-json and deliberately keeps stdin open for the whole run (for control_response frames). Steer writes one more user frame to that same pipe; the CLI queues it as the next user turn. Design points: - Steer is a nil-able func field on Session, not a new interface method, so the other 16 backends compile unchanged and callers must treat nil as "unsupported → use the existing follow-up-task path". - A steer frame can never overtake the initial prompt (gated on the prompt write completing), and a steer after the run finishes or is cancelled returns an error instead of silently dropping the message — the caller keeps the deferral fallback in both cases. - All stdin frame producers (initial prompt, control responses, Steer) now share one locked writer, making each frame write atomic. This also hardens a pre-existing latent interleaving between the prompt-writer goroutine and the scanner's control responses. Tested with a fake-CLI round trip: the steer text must arrive as a well-formed user frame on the same stdin, after the prompt, while the run is live (the fake blocks on the second frame, so non-delivery hangs the test rather than passing); plus steer-after-completion and steer-after-cancel error paths. Server-side wiring (routing a mid-run comment to a live session instead of the deferred replay, claim-based so completion reconcile never double-delivers) is deliberately NOT in this PR — proposed separately so the #5914 invariants get their own review. --- server/pkg/agent/agent.go | 10 ++ server/pkg/agent/claude.go | 57 +++++++- server/pkg/agent/claude_deadlock_test.go | 3 + server/pkg/agent/claude_steer_test.go | 168 +++++++++++++++++++++++ 4 files changed, 235 insertions(+), 3 deletions(-) create mode 100644 server/pkg/agent/claude_steer_test.go diff --git a/server/pkg/agent/agent.go b/server/pkg/agent/agent.go index 71aa576621f..6e594a85299 100644 --- a/server/pkg/agent/agent.go +++ b/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. diff --git a/server/pkg/agent/claude.go b/server/pkg/agent/claude.go index ad877a8e22e..0d1dff56f99 100644 --- a/server/pkg/agent/claude.go +++ b/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 { diff --git a/server/pkg/agent/claude_deadlock_test.go b/server/pkg/agent/claude_deadlock_test.go index afb87cc2932..e054aaa6671 100644 --- a/server/pkg/agent/claude_deadlock_test.go +++ b/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) diff --git a/server/pkg/agent/claude_steer_test.go b/server/pkg/agent/claude_steer_test.go new file mode 100644 index 00000000000..d86c6f50395 --- /dev/null +++ b/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") + } +}