diff --git a/agent/claudecode/session.go b/agent/claudecode/session.go index 084554c76..218dbbae3 100644 --- a/agent/claudecode/session.go +++ b/agent/claudecode/session.go @@ -590,6 +590,21 @@ func (cs *claudeSession) Send(prompt string, images []core.ImageAttachment, file }) } +// Steer appends additional guidance to the current in-flight Claude task. +// We map this to a normal user message on the same live session with +// priority=next, which matches Claude's native queue semantics for +// "process after the current step/tool boundary but before the next turn". +func (cs *claudeSession) Steer(prompt string) error { + if !cs.alive.Load() { + return fmt.Errorf("session process is not running") + } + return cs.writeJSON(map[string]any{ + "type": "user", + "priority": "next", + "message": map[string]any{"role": "user", "content": prompt}, + }) +} + func extFromMime(mime string) string { switch mime { case "image/jpeg": @@ -746,7 +761,7 @@ func (cs *claudeSession) Close() error { // Uses single quotes because some splitters (e.g. my_cli) don't support // backslash escapes inside double quotes. For values containing single // quotes, we close the single-quoted segment, add an escaped single -// quote, and reopen: 'it'\”s' → it's +// quote, and reopen: 'it'\”s' -> it's func shellJoinArgs(args []string) string { var b strings.Builder for i, a := range args { diff --git a/agent/claudecode/session_test.go b/agent/claudecode/session_test.go index 53f029a4f..94eca2064 100644 --- a/agent/claudecode/session_test.go +++ b/agent/claudecode/session_test.go @@ -3,6 +3,7 @@ package claudecode import ( "bytes" "context" + "encoding/json" "io" "os" "os/exec" @@ -12,6 +13,12 @@ import ( "github.com/chenhg5/cc-connect/core" ) +type nopWriteCloser struct { + io.Writer +} + +func (nopWriteCloser) Close() error { return nil } + func TestHandleResultParsesUsage(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() @@ -71,6 +78,41 @@ func TestHandleResultNoUsage(t *testing.T) { } } +func TestClaudeSessionSteer_UsesNextPriorityUserMessage(t *testing.T) { + var buf bytes.Buffer + cs := &claudeSession{ + stdin: nopWriteCloser{Writer: &buf}, + } + cs.alive.Store(true) + + if err := cs.Steer("focus on failing tests first"); err != nil { + t.Fatalf("Steer() error = %v", err) + } + + var payload map[string]any + if err := json.Unmarshal(bytes.TrimSpace(buf.Bytes()), &payload); err != nil { + t.Fatalf("decode steer payload: %v", err) + } + + if got := payload["type"]; got != "user" { + t.Fatalf("type = %#v, want user", got) + } + if got := payload["priority"]; got != "next" { + t.Fatalf("priority = %#v, want next", got) + } + + message, ok := payload["message"].(map[string]any) + if !ok { + t.Fatalf("message = %#v, want object", payload["message"]) + } + if got := message["role"]; got != "user" { + t.Fatalf("message.role = %#v, want user", got) + } + if got := message["content"]; got != "focus on failing tests first" { + t.Fatalf("message.content = %#v, want steer text", got) + } +} + func TestReadLoop_ChildHoldsStdoutPipe(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() diff --git a/agent/codex/appserver_session.go b/agent/codex/appserver_session.go index bad0c8949..f0be4a40d 100644 --- a/agent/codex/appserver_session.go +++ b/agent/codex/appserver_session.go @@ -64,6 +64,10 @@ type turnStartResponse struct { } `json:"turn"` } +type turnSteerResponse struct { + TurnID string `json:"turnId"` +} + type turnNotification struct { ThreadID string `json:"threadId"` Turn struct { @@ -471,6 +475,45 @@ func (s *appServerSession) Send(prompt string, images []core.ImageAttachment, fi return nil } +// Steer appends additional guidance to the currently active regular turn. +// This uses Codex app-server's native same-turn steering API rather than +// starting a new turn. +func (s *appServerSession) Steer(prompt string) error { + if !s.alive.Load() { + return fmt.Errorf("session is closed") + } + + threadID := s.CurrentSessionID() + if threadID == "" { + return fmt.Errorf("codex app-server thread id is empty") + } + + s.stateMu.Lock() + turnID := s.currentTurn + s.stateMu.Unlock() + if turnID == "" { + return fmt.Errorf("codex app-server has no active turn to steer") + } + + params := map[string]any{ + "threadId": threadID, + "expectedTurnId": turnID, + "input": []map[string]any{{ + "type": "text", + "text": prompt, + }}, + } + + var resp turnSteerResponse + if err := s.request("turn/steer", params, &resp); err != nil { + return fmt.Errorf("codex app-server turn/steer: %w", err) + } + if resp.TurnID == "" { + return fmt.Errorf("codex app-server turn/steer returned empty turn id") + } + return nil +} + func (s *appServerSession) stageImages(prompt string, images []core.ImageAttachment) (string, []string, error) { if len(images) == 0 { return prompt, nil, nil diff --git a/agent/codex/appserver_session_test.go b/agent/codex/appserver_session_test.go index f865ed9a2..9debe7e46 100644 --- a/agent/codex/appserver_session_test.go +++ b/agent/codex/appserver_session_test.go @@ -1,8 +1,11 @@ package codex import ( + "bytes" "context" "encoding/json" + "io" + "sync" "testing" "github.com/chenhg5/cc-connect/core" @@ -162,6 +165,93 @@ func TestMapAppServerRateLimits_PrefersMultiBucketView(t *testing.T) { } } +func TestAppServerSessionSteer_RequiresActiveTurn(t *testing.T) { + s := &appServerSession{ + ctx: context.Background(), + pending: make(map[int64]chan rpcResponseEnvelope), + } + s.alive.Store(true) + s.threadID.Store("thread-1") + + err := s.Steer("focus on failing tests first") + if err == nil || err.Error() != "codex app-server has no active turn to steer" { + t.Fatalf("Steer() error = %v, want no active turn error", err) + } +} + +func TestAppServerSessionSteer_RequestShape(t *testing.T) { + var buf bytes.Buffer + s := &appServerSession{ + ctx: context.Background(), + stdin: nopAppServerWriteCloser{Writer: &buf}, + pending: make(map[int64]chan rpcResponseEnvelope), + } + s.alive.Store(true) + s.threadID.Store("thread-1") + s.currentTurn = "turn-1" + + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + for { + s.pendingMu.Lock() + ch := s.pending[1] + s.pendingMu.Unlock() + if ch != nil { + ch <- rpcResponseEnvelope{ID: int64(1), Result: json.RawMessage(`{"turnId":"turn-1"}`)} + return + } + } + }() + + if err := s.Steer("focus on failing tests first"); err != nil { + t.Fatalf("Steer() error = %v", err) + } + wg.Wait() + + var payload map[string]any + if err := json.Unmarshal(bytes.TrimSpace(buf.Bytes()), &payload); err != nil { + t.Fatalf("decode steer payload: %v", err) + } + + if got := payload["method"]; got != "turn/steer" { + t.Fatalf("method = %#v, want turn/steer", got) + } + + params, ok := payload["params"].(map[string]any) + if !ok { + t.Fatalf("params = %#v, want object", payload["params"]) + } + if got := params["threadId"]; got != "thread-1" { + t.Fatalf("threadId = %#v, want thread-1", got) + } + if got := params["expectedTurnId"]; got != "turn-1" { + t.Fatalf("expectedTurnId = %#v, want turn-1", got) + } + + input, ok := params["input"].([]any) + if !ok || len(input) != 1 { + t.Fatalf("input = %#v, want single-element array", params["input"]) + } + item, ok := input[0].(map[string]any) + if !ok { + t.Fatalf("input[0] = %#v, want object", input[0]) + } + if got := item["type"]; got != "text" { + t.Fatalf("input[0].type = %#v, want text", got) + } + if got := item["text"]; got != "focus on failing tests first" { + t.Fatalf("input[0].text = %#v, want steer text", got) + } +} + +type nopAppServerWriteCloser struct { + io.Writer +} + +func (nopAppServerWriteCloser) Close() error { return nil } + var _ interface { GetUsage(context.Context) (*core.UsageReport, error) } = (*appServerSession)(nil) diff --git a/core/bridge_capabilities_snapshot_test.go b/core/bridge_capabilities_snapshot_test.go index 9f393a599..bf5070f51 100644 --- a/core/bridge_capabilities_snapshot_test.go +++ b/core/bridge_capabilities_snapshot_test.go @@ -13,7 +13,7 @@ func TestBridgeBuildCapabilitiesSnapshotIncludesProjectCatalog(t *testing.T) { CurrentBuildTime = prevBuildTime }() - bs := NewBridgeServer(0, "", "/bridge/ws", nil) + bs := NewBridgeServer(0, "test-token", "/bridge/ws", nil) bp := bs.NewPlatform("test-proj") e := NewEngine("test-proj", &stubAgent{}, []Platform{bp}, "", LangEnglish) e.AddCommand("deploy", "Deploy app", "ship it", "", "", "config") diff --git a/core/engine.go b/core/engine.go index f15e2cc9a..bb1011afd 100644 --- a/core/engine.go +++ b/core/engine.go @@ -4452,6 +4452,7 @@ var builtinCommands = []struct { {[]string{"heartbeat", "hb"}, "heartbeat"}, {[]string{"compress", "compact"}, "compress"}, {[]string{"stop"}, "stop"}, + {[]string{"steer"}, "steer"}, {[]string{"help"}, "help"}, {[]string{"version"}, "version"}, {[]string{"commands", "command", "cmd"}, "commands"}, @@ -4639,6 +4640,8 @@ func (e *Engine) handleCommand(p Platform, msg *Message, raw string) bool { e.cmdCompress(p, msg) case "stop": e.cmdStop(p, msg) + case "steer": + e.cmdSteer(p, msg, args) case "help": e.cmdHelp(p, msg) case "start": @@ -7026,6 +7029,7 @@ func helpCardGroups() []helpCardGroup { {command: "/alias", action: "nav:/alias"}, {command: "/skills", action: "nav:/skills"}, {command: "/compress", action: "cmd:/compress"}, + {command: "/steer", action: "cmd:/steer"}, {command: "/stop", action: "act:/stop"}, {command: "/ps", action: "cmd:/ps"}, }, @@ -7744,6 +7748,38 @@ func (e *Engine) cmdStop(p Platform, msg *Message) { e.reply(p, msg.ReplyCtx, e.i18n.T(MsgExecutionStopped)) } +func (e *Engine) cmdSteer(p Platform, msg *Message, args []string) { + text := strings.TrimSpace(strings.Join(args, " ")) + if text == "" { + e.reply(p, msg.ReplyCtx, e.i18n.T(MsgSteerEmpty)) + return + } + + iKey := e.interactiveKeyForSessionKey(msg.SessionKey) + e.interactiveMu.Lock() + state, ok := e.interactiveStates[iKey] + e.interactiveMu.Unlock() + + if !ok || state == nil || state.agentSession == nil || !state.agentSession.Alive() { + e.reply(p, msg.ReplyCtx, e.i18n.T(MsgNoExecution)) + return + } + + steerer, ok := state.agentSession.(SessionSteerer) + if !ok { + e.reply(p, msg.ReplyCtx, e.i18n.T(MsgSteerNotSupported)) + return + } + + if err := steerer.Steer(text); err != nil { + slog.Error("steer: send failed", "error", err) + e.reply(p, msg.ReplyCtx, e.i18n.T(MsgSteerSendFailed)) + return + } + + e.reply(p, msg.ReplyCtx, e.i18n.T(MsgSteerSent)) +} + func (e *Engine) stopInteractiveSession(sessionKey string, quietPlatform Platform, quietReplyCtx any) bool { return e.stopInteractiveSessionWithOptions(sessionKey, true) } diff --git a/core/engine_test.go b/core/engine_test.go index b571b1795..dcf282239 100644 --- a/core/engine_test.go +++ b/core/engine_test.go @@ -49,6 +49,17 @@ func (s *recordingAgentSession) RespondPermission(id string, res PermissionResul return nil } +type steerSession struct { + stubAgentSession + lastPrompt string + err error +} + +func (s *steerSession) Steer(prompt string) error { + s.lastPrompt = prompt + return s.err +} + type stubPlatformEngine struct { n string sent []string @@ -8067,6 +8078,106 @@ func TestCmdStop_UsesInteractiveKeyForMultiWorkspace(t *testing.T) { } } +func TestCmdSteer_NoExecution_RepliesNoExecution(t *testing.T) { + p := &stubPlatformEngine{n: "test"} + e := NewEngine("test", &stubAgent{}, []Platform{p}, "", LangEnglish) + msg := &Message{SessionKey: "test:user1", Content: "/steer focus", ReplyCtx: "ctx"} + + e.cmdSteer(p, msg, []string{"focus"}) + + sent := p.getSent() + if len(sent) == 0 { + t.Fatal("expected a reply") + } + if !strings.Contains(sent[0], e.i18n.T(MsgNoExecution)) { + t.Fatalf("expected MsgNoExecution, got %q", sent[0]) + } +} + +func TestCmdSteer_Empty_RepliesUsage(t *testing.T) { + p := &stubPlatformEngine{n: "test"} + e := NewEngine("test", &stubAgent{}, []Platform{p}, "", LangEnglish) + msg := &Message{SessionKey: "test:user1", Content: "/steer", ReplyCtx: "ctx"} + + e.cmdSteer(p, msg, nil) + + sent := p.getSent() + if len(sent) == 0 { + t.Fatal("expected a reply") + } + if !strings.Contains(sent[0], e.i18n.T(MsgSteerEmpty)) { + t.Fatalf("expected MsgSteerEmpty, got %q", sent[0]) + } +} + +func TestCmdSteer_NotSupported_RepliesNotSupported(t *testing.T) { + p := &stubPlatformEngine{n: "test"} + e := NewEngine("test", &stubAgent{}, []Platform{p}, "", LangEnglish) + key := "test:user1" + + e.interactiveMu.Lock() + e.interactiveStates[key] = &interactiveState{agentSession: &stubAgentSession{}} + e.interactiveMu.Unlock() + + msg := &Message{SessionKey: key, Content: "/steer focus", ReplyCtx: "ctx"} + e.cmdSteer(p, msg, []string{"focus"}) + + sent := p.getSent() + if len(sent) == 0 { + t.Fatal("expected a reply") + } + if !strings.Contains(sent[0], e.i18n.T(MsgSteerNotSupported)) { + t.Fatalf("expected MsgSteerNotSupported, got %q", sent[0]) + } +} + +func TestCmdSteer_Success_SendsGuidance(t *testing.T) { + p := &stubPlatformEngine{n: "test"} + e := NewEngine("test", &stubAgent{}, []Platform{p}, "", LangEnglish) + key := "test:user1" + sess := &steerSession{} + + e.interactiveMu.Lock() + e.interactiveStates[key] = &interactiveState{agentSession: sess} + e.interactiveMu.Unlock() + + msg := &Message{SessionKey: key, Content: "/steer focus on tests", ReplyCtx: "ctx"} + e.cmdSteer(p, msg, []string{"focus", "on", "tests"}) + + if sess.lastPrompt != "focus on tests" { + t.Fatalf("steer prompt = %q, want %q", sess.lastPrompt, "focus on tests") + } + sent := p.getSent() + if len(sent) == 0 { + t.Fatal("expected a reply") + } + if !strings.Contains(sent[0], e.i18n.T(MsgSteerSent)) { + t.Fatalf("expected MsgSteerSent, got %q", sent[0]) + } +} + +func TestCmdSteer_Error_RepliesFailed(t *testing.T) { + p := &stubPlatformEngine{n: "test"} + e := NewEngine("test", &stubAgent{}, []Platform{p}, "", LangEnglish) + key := "test:user1" + sess := &steerSession{err: errors.New("boom")} + + e.interactiveMu.Lock() + e.interactiveStates[key] = &interactiveState{agentSession: sess} + e.interactiveMu.Unlock() + + msg := &Message{SessionKey: key, Content: "/steer focus", ReplyCtx: "ctx"} + e.cmdSteer(p, msg, []string{"focus"}) + + sent := p.getSent() + if len(sent) == 0 { + t.Fatal("expected a reply") + } + if !strings.Contains(sent[0], e.i18n.T(MsgSteerSendFailed)) { + t.Fatalf("expected MsgSteerSendFailed, got %q", sent[0]) + } +} + // =========================================================================== // Beta pre-release tests: inject_sender, idle_timeout, /shell, /workspace, // /switch, /memory diff --git a/core/i18n.go b/core/i18n.go index bbf9db3d4..5654193c2 100644 --- a/core/i18n.go +++ b/core/i18n.go @@ -250,8 +250,8 @@ const ( MsgCronBtnUnmute MsgKey = "cron_btn_unmute" MsgCronBtnDelete MsgKey = "cron_btn_delete" - MsgStatusTitle MsgKey = "status_title" - MsgReplyFooterRemaining MsgKey = "reply_footer_remaining" + MsgStatusTitle MsgKey = "status_title" + MsgReplyFooterRemaining MsgKey = "reply_footer_remaining" MsgModelCurrent MsgKey = "model_current" MsgModelChanged MsgKey = "model_changed" MsgModelChangeFailed MsgKey = "model_change_failed" @@ -267,6 +267,10 @@ const ( MsgCompressing MsgKey = "compressing" MsgCompressNoSession MsgKey = "compress_no_session" MsgCompressDone MsgKey = "compress_done" + MsgSteerSent MsgKey = "steer_sent" + MsgSteerSendFailed MsgKey = "steer_send_failed" + MsgSteerEmpty MsgKey = "steer_empty" + MsgSteerNotSupported MsgKey = "steer_not_supported" MsgMemoryNotSupported MsgKey = "memory_not_supported" MsgMemoryShowProject MsgKey = "memory_show_project" @@ -496,6 +500,7 @@ const ( MsgBuiltinCmdQuiet MsgKey = "quiet" MsgBuiltinCmdCompress MsgKey = "compress" MsgBuiltinCmdStop MsgKey = "stop" + MsgBuiltinCmdSteer MsgKey = "steer" MsgBuiltinCmdCron MsgKey = "cron" MsgBuiltinCmdCommands MsgKey = "commands" MsgBuiltinCmdAlias MsgKey = "alias" @@ -652,11 +657,11 @@ var messages = map[MsgKey]map[Language]string{ LangSpanish: "No hay ejecución en progreso.", }, MsgPreviousProcessing: { - LangEnglish: "⏳ Previous request still processing. Use `/ps ` to send a P.S. to the running task.", - LangChinese: "⏳ 上一个请求仍在处理中。使用 `/ps <消息>` 可向正在执行的任务追加补充信息。", - LangTraditionalChinese: "⏳ 上一個請求仍在處理中。使用 `/ps <訊息>` 可向正在執行的任務追加補充資訊。", - LangJapanese: "⏳ 前のリクエストを処理中です。`/ps <メッセージ>` で実行中のタスクに補足情報を送れます。", - LangSpanish: "⏳ La solicitud anterior aún se está procesando. Use `/ps ` para enviar un P.S. a la tarea en curso.", + LangEnglish: "⏳ Previous request still processing. Use `/steer ` to add guidance to the current task.", + LangChinese: "⏳ 上一个请求仍在处理中。使用 `/steer <消息>` 可向当前任务追加引导。", + LangTraditionalChinese: "⏳ 上一個請求仍在處理中。使用 `/steer <訊息>` 可向當前任務追加引導。", + LangJapanese: "⏳ 前のリクエストを処理中です。`/steer <メッセージ>` で現在のタスクに追加の指示を送れます。", + LangSpanish: "⏳ La solicitud anterior aún se está procesando. Use `/steer ` para agregar instrucciones a la tarea actual.", }, MsgMessageQueued: { LangEnglish: "📬 Message received — will process after the current task finishes.", @@ -2099,6 +2104,34 @@ var messages = map[MsgKey]map[Language]string{ LangJapanese: "✅ コンテキスト圧縮完了。", LangSpanish: "✅ Contexto comprimido.", }, + MsgSteerSent: { + LangEnglish: "✅ Guidance sent to the current task.", + LangChinese: "✅ 已向当前任务发送引导。", + LangTraditionalChinese: "✅ 已向當前任務送出引導。", + LangJapanese: "✅ 現在のタスクに追加の指示を送信しました。", + LangSpanish: "✅ Instrucciones enviadas a la tarea actual.", + }, + MsgSteerSendFailed: { + LangEnglish: "❌ Failed to send guidance to the current task.", + LangChinese: "❌ 向当前任务发送引导失败。", + LangTraditionalChinese: "❌ 向當前任務送出引導失敗。", + LangJapanese: "❌ 現在のタスクへの追加指示の送信に失敗しました。", + LangSpanish: "❌ Error al enviar instrucciones a la tarea actual.", + }, + MsgSteerEmpty: { + LangEnglish: "Usage: `/steer `", + LangChinese: "用法：`/steer <消息>`", + LangTraditionalChinese: "用法：`/steer <訊息>`", + LangJapanese: "使い方：`/steer <メッセージ>`", + LangSpanish: "Uso: `/steer `", + }, + MsgSteerNotSupported: { + LangEnglish: "❌ This agent does not support `/steer`.", + LangChinese: "❌ 当前 Agent 不支持 `/steer`。", + LangTraditionalChinese: "❌ 當前 Agent 不支援 `/steer`。", + LangJapanese: "❌ このエージェントは `/steer` をサポートしていません。", + LangSpanish: "❌ Este agente no admite `/steer`.", + }, // Inline strings for engine.go commands MsgStatusMode: { @@ -3382,6 +3415,13 @@ var messages = map[MsgKey]map[Language]string{ LangJapanese: "現在の実行を停止", LangSpanish: "Detener ejecución actual", }, + MsgBuiltinCmdSteer: { + LangEnglish: "Add guidance to the current in-flight task", + LangChinese: "向当前执行中的任务追加引导", + LangTraditionalChinese: "向當前執行中的任務追加引導", + LangJapanese: "現在実行中のタスクに追加の指示を送る", + LangSpanish: "Agregar instrucciones a la tarea en curso", + }, MsgBuiltinCmdCron: { LangEnglish: "Manage scheduled tasks, arg: [add|list|del|enable|disable]", LangChinese: "管理定时任务，参数: [add|list|del|enable|disable]", diff --git a/core/interfaces.go b/core/interfaces.go index fd1692291..aff30084f 100644 --- a/core/interfaces.go +++ b/core/interfaces.go @@ -444,6 +444,14 @@ type ContextCompressor interface { CompressCommand() string } +// SessionSteerer is an optional interface for running agent sessions that can +// append additional user guidance to the current in-flight task without +// starting a new task. Backends should map this to their native same-turn +// steering semantics when available. +type SessionSteerer interface { + Steer(prompt string) error +} + // CommandProvider is an optional interface for agents that expose custom slash // commands via local files (e.g. .claude/commands/*.md). The engine scans the // returned directories for *.md files and registers them as slash commands. diff --git a/docs/usage.md b/docs/usage.md index 65a79a211..82139e0cc 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -43,6 +43,7 @@ Each user gets an independent session with full conversation context. Manage ses | `/allow ` | Pre-allow a tool (next session) | | `/reasoning [level]` | View or switch reasoning effort (Codex) | | `/mode [name]` | View or switch permission mode | +| `/steer ` | Add guidance to the current in-flight task | | `/stop` | Stop current execution | | `/help` | Show available commands | diff --git a/docs/usage.zh-CN.md b/docs/usage.zh-CN.md index 4a049840e..b7feef9e0 100644 --- a/docs/usage.zh-CN.md +++ b/docs/usage.zh-CN.md @@ -45,6 +45,7 @@ cc-connect 完整功能使用指南。 | `/allow <工具名>` | 预授权工具 | | `/reasoning [等级]` | 查看或切换推理强度（Codex）| | `/mode [名称]` | 查看或切换权限模式 | +| `/steer <消息>` | 向当前执行中的任务追加引导 | | `/stop` | 停止当前执行 | | `/help` | 显示可用命令 |