feat(core): add builtin /steer command

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 {

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()

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

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)

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")

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)
 }