feat(codex): steer busy follow-ups into the active turn

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 {
@@ -177,6 +181,7 @@ type appServerSession struct {
 	wg        sync.WaitGroup
 
 	stateMu      sync.Mutex
+	steerMu      sync.Mutex
 	pendingMsgs  []string
 	currentTurn  string
 	preambleSent bool
@@ -188,6 +193,7 @@ type appServerSession struct {
 
 const (
 	appServerRequestTimeout      = 120 * time.Second
+	appServerSteerTimeout        = 5 * time.Second
 	appServerUsageRefreshTimeout = 1500 * time.Millisecond
 )
 
@@ -511,6 +517,53 @@ func (s *appServerSession) Send(prompt string, messageID string, images []core.I
 	return nil
 }
 
+// SteerTurn appends text to the active Codex turn without starting a second
+// turn. expectedTurnId prevents a late follow-up from being attached to a newer
+// turn if the original one completes while this request is in flight.
+func (s *appServerSession) SteerTurn(prompt string) error {
+	if !s.alive.Load() {
+		return fmt.Errorf("session is closed")
+	}
+	if strings.TrimSpace(prompt) == "" {
+		return fmt.Errorf("codex app-server turn/steer prompt is empty")
+	}
+
+	// Preserve the arrival order of rapid follow-ups and avoid issuing multiple
+	// concurrent turn/steer requests for the same active turn.
+	s.steerMu.Lock()
+	defer s.steerMu.Unlock()
+
+	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,
+		"input": []map[string]any{
+			{
+				"type": "text",
+				"text": prompt,
+			},
+		},
+		"expectedTurnId": turnID,
+	}
+	var resp turnSteerResponse
+	if err := s.requestWithTimeout("turn/steer", params, &resp, appServerSteerTimeout); err != nil {
+		return fmt.Errorf("codex app-server turn/steer: %w", err)
+	}
+	if resp.TurnID != turnID {
+		return fmt.Errorf("codex app-server turn/steer returned turn id %q, want %q", resp.TurnID, turnID)
+	}
+	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

@@ -342,6 +342,89 @@ func TestAppServerSession_HandleRequestUserInputWritesCodexResponse(t *testing.T
 	}
 }
 
+func TestAppServerSession_SteerTurnUsesExpectedActiveTurn(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	stdin := &lockedWriteCloser{}
+	s := &appServerSession{
+		ctx:     ctx,
+		cancel:  cancel,
+		stdin:   stdin,
+		pending: make(map[int64]chan rpcResponseEnvelope),
+	}
+	s.alive.Store(true)
+	s.threadID.Store("thread-1")
+	s.currentTurn = "turn-7"
+
+	done := make(chan error, 1)
+	go func() {
+		done <- s.SteerTurn("add unit tests")
+	}()
+
+	line := waitForWrittenJSONLine(t, stdin)
+	var request struct {
+		ID     int64  `json:"id"`
+		Method string `json:"method"`
+		Params struct {
+			ThreadID       string `json:"threadId"`
+			ExpectedTurnID string `json:"expectedTurnId"`
+			Input          []struct {
+				Type string `json:"type"`
+				Text string `json:"text"`
+			} `json:"input"`
+		} `json:"params"`
+	}
+	if err := json.Unmarshal([]byte(line), &request); err != nil {
+		t.Fatalf("decode request %q: %v", line, err)
+	}
+	if request.Method != "turn/steer" {
+		t.Fatalf("method = %q, want turn/steer", request.Method)
+	}
+	if request.Params.ThreadID != "thread-1" || request.Params.ExpectedTurnID != "turn-7" {
+		t.Fatalf("params = %#v, want thread-1/turn-7", request.Params)
+	}
+	if len(request.Params.Input) != 1 || request.Params.Input[0].Type != "text" || request.Params.Input[0].Text != "add unit tests" {
+		t.Fatalf("input = %#v, want one text item", request.Params.Input)
+	}
+
+	s.pendingMu.Lock()
+	responseCh := s.pending[request.ID]
+	delete(s.pending, request.ID)
+	s.pendingMu.Unlock()
+	if responseCh == nil {
+		t.Fatalf("no pending RPC response channel for id %d", request.ID)
+	}
+	responseCh <- rpcResponseEnvelope{ID: request.ID, Result: json.RawMessage(`{"turnId":"turn-7"}`)}
+
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Fatalf("SteerTurn() error = %v", err)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("SteerTurn() did not finish after RPC response")
+	}
+}
+
+func TestAppServerSession_SteerTurnRejectsMissingActiveTurn(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	stdin := &lockedWriteCloser{}
+	s := &appServerSession{ctx: ctx, cancel: cancel, stdin: stdin}
+	s.alive.Store(true)
+	s.threadID.Store("thread-1")
+
+	err := s.SteerTurn("too late")
+	if err == nil || !strings.Contains(err.Error(), "no active turn") {
+		t.Fatalf("SteerTurn() error = %v, want no active turn", err)
+	}
+	if got := stdin.String(); got != "" {
+		t.Fatalf("unexpected RPC write without active turn: %q", got)
+	}
+}
+
 var _ interface {
 	GetUsage(context.Context) (*core.UsageReport, error)
 } = (*appServerSession)(nil)
@@ -350,6 +433,8 @@ var _ interface {
 	GetContextUsage() *core.ContextUsage
 } = (*appServerSession)(nil)
 
+var _ core.AgentSessionSteerer = (*appServerSession)(nil)
+
 type lockedWriteCloser struct {
 	mu  sync.Mutex
 	buf bytes.Buffer

cmd/cc-connect/main.go

@@ -737,6 +737,7 @@ func main() {
 				engine.SetAgentSessionIdleTimeout(time.Duration(mins) * time.Minute)
 			}
 		}
+		engine.SetBusyMessageMode(proj.BusyMessageMode)
 
 		// Wire sender injection
 		if proj.InjectSender != nil {
@@ -1764,6 +1765,7 @@ func reloadConfig(configPath, projName string, engine *core.Engine) (*core.Confi
 		// explicitly so those stale idle-close timers cannot fire later.
 		engine.SetAgentSessionIdleTimeout(0)
 	}
+	engine.SetBusyMessageMode(proj.BusyMessageMode)
 
 	// Reload instant reply
 	if cfg.InstantReply.Enabled != nil && *cfg.InstantReply.Enabled {

config.example.toml

@@ -1544,17 +1544,25 @@ app_secret = "your-feishu-app-secret"
 # =============================================================================
 # Requires: npm install -g @openai/codex
 # 需要安装：npm install -g @openai/codex
-# Codex uses `codex exec --json` under the hood.
-# Codex 底层使用 `codex exec --json` 命令。
+# Codex defaults to `codex exec --json`; use the app-server backend below for
+# a persistent process and in-flight text steering.
+# Codex 默认使用 `codex exec --json`；如需常驻进程和执行中补充文本，请使用下方
+# app-server 后端。
 
 # [[projects]]
 # name = "my-codex-project"
+# agent_session_idle_timeout_mins = 60  # Close the live process after 1h idle, while preserving the resumable session ID
+#                                       # 空闲 1 小时后关闭 live 进程，但保留可恢复的会话 ID
+# busy_message_mode = "steer"           # Append plain-text follow-ups to the active turn; default is "queue"
+#                                       # 将纯文本补充追加到当前回合；默认值为 "queue"
 #
 # [projects.agent]
 # type = "codex"
 #
 # [projects.agent.options]
 # work_dir = "/path/to/project"
+# backend = "app_server"
+# app_server_url = "stdio"
 # mode = "suggest"  # "suggest" | "auto-edit" | "full-auto" | "yolo"
 #
 # Mode options / 模式说明:

config/config.go

@@ -489,6 +489,10 @@ type ProjectConfig struct {
 	// AgentSessionIdleTimeoutMins 在指定分钟数后关闭空闲的 live agent 进程，
 	// 同时保留已保存的 session ID，便于下一条消息继续恢复。0 或 nil 表示禁用。
 	AgentSessionIdleTimeoutMins *int `toml:"agent_session_idle_timeout_mins,omitempty"`
+	// BusyMessageMode controls plain-text messages received while the current
+	// turn is running: "queue" (default) or "steer". Steering is attempted only
+	// when the active agent session implements the optional steering capability.
+	BusyMessageMode string `toml:"busy_message_mode,omitempty"`
 	// RunAsUser, when set, causes the agent command for this project to be
 	// spawned under a different Unix user via `sudo -n -iu <user> --`. This
 	// provides OS-level file-system isolation from the supervisor user who
@@ -1053,6 +1057,11 @@ func (c *Config) validateInternal(permissive bool) error {
 		if proj.AgentSessionIdleTimeoutMins != nil && *proj.AgentSessionIdleTimeoutMins < 0 {
 			return fmt.Errorf("config: %s.agent_session_idle_timeout_mins must be >= 0", prefix)
 		}
+		switch strings.ToLower(strings.TrimSpace(proj.BusyMessageMode)) {
+		case "", "queue", "steer":
+		default:
+			return fmt.Errorf("config: %s.busy_message_mode must be queue or steer", prefix)
+		}
 		if err := validateRunAsUser(prefix, proj.RunAsUser); err != nil {
 			return err
 		}

config/config_test.go

@@ -1751,6 +1751,30 @@ func TestLoad_RejectsNegativeAgentSessionIdleTimeoutMins(t *testing.T) {
 	}
 }
 
+func TestLoad_ParsesBusyMessageMode(t *testing.T) {
+	configPath := writeConfigFixture(t, projectWithBusyMessageModeFixture)
+
+	cfg, err := Load(configPath)
+	if err != nil {
+		t.Fatalf("Load returned error: %v", err)
+	}
+	if got := cfg.Projects[0].BusyMessageMode; got != "steer" {
+		t.Fatalf("busy_message_mode = %q, want steer", got)
+	}
+}
+
+func TestLoad_RejectsInvalidBusyMessageMode(t *testing.T) {
+	configPath := writeConfigFixture(t, projectWithInvalidBusyMessageModeFixture)
+
+	_, err := Load(configPath)
+	if err == nil {
+		t.Fatal("expected error for invalid busy_message_mode")
+	}
+	if !strings.Contains(err.Error(), "busy_message_mode") {
+		t.Fatalf("error = %q, want busy_message_mode validation", err.Error())
+	}
+}
+
 func TestLoad_ParsesRunAsUser(t *testing.T) {
 	if runtime.GOOS == "windows" {
 		t.Skip("run_as_user is only supported on Linux/macOS")
@@ -2216,6 +2240,42 @@ type = "telegram"
 bot_token = "token_xxx"
 `
 
+const projectWithBusyMessageModeFixture = `
+[[projects]]
+name = "beta"
+busy_message_mode = "steer"
+
+[projects.agent]
+type = "codex"
+
+[projects.agent.options]
+work_dir = "/tmp/beta"
+
+[[projects.platforms]]
+type = "telegram"
+
+[projects.platforms.options]
+bot_token = "token_xxx"
+`
+
+const projectWithInvalidBusyMessageModeFixture = `
+[[projects]]
+name = "beta"
+busy_message_mode = "parallel"
+
+[projects.agent]
+type = "codex"
+
+[projects.agent.options]
+work_dir = "/tmp/beta"
+
+[[projects.platforms]]
+type = "telegram"
+
+[projects.platforms.options]
+bot_token = "token_xxx"
+`
+
 const projectWithRunAsUserFixture = `
 [[projects]]
 name = "sandboxed"