From 025f62c9d339b9bac8f9d3e98b729cf16f3b27e7 Mon Sep 17 00:00:00 2001 From: chenkj <1412582379@qq.com> Date: Fri, 7 Aug 2026 19:26:56 +0800 Subject: [PATCH 1/2] feat(codex): support steering the active app-server turn --- agent/codex/appserver_session.go | 53 +++++++++++++++++ agent/codex/appserver_session_test.go | 85 +++++++++++++++++++++++++++ core/interfaces.go | 8 +++ 3 files changed, 146 insertions(+) diff --git a/agent/codex/appserver_session.go b/agent/codex/appserver_session.go index 15d8415c68..1e97476819 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 { @@ -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 diff --git a/agent/codex/appserver_session_test.go b/agent/codex/appserver_session_test.go index ce2b28d32d..d706527a77 100644 --- a/agent/codex/appserver_session_test.go +++ b/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 diff --git a/core/interfaces.go b/core/interfaces.go index 81f4d1c149..4319a6b499 100644 --- a/core/interfaces.go +++ b/core/interfaces.go @@ -581,6 +581,14 @@ type AgentSessionCanceller interface { CancelTurn() error } +// AgentSessionSteerer is an optional interface for agent sessions that can +// append text to the turn currently in flight. Implementations must return an +// error when there is no active turn; callers can then safely fall back to +// queueing the message for the next turn. +type AgentSessionSteerer interface { + SteerTurn(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. From 3dacd5ca2f3c672da03c1275067b278f8084e652 Mon Sep 17 00:00:00 2001 From: chenkj <1412582379@qq.com> Date: Fri, 7 Aug 2026 19:27:32 +0800 Subject: [PATCH 2/2] feat(core): route busy text through active-turn steering --- cmd/cc-connect/main.go | 2 + config.example.toml | 12 +- config/config.go | 9 ++ config/config_test.go | 60 +++++++++ core/cuj_test.go | 116 ++++++++++++++++++ core/engine.go | 102 +++++++++++++++- core/engine_test.go | 269 ++++++++++++++++++++++++++++++++++++++++- core/i18n.go | 8 ++ 8 files changed, 568 insertions(+), 10 deletions(-) diff --git a/cmd/cc-connect/main.go b/cmd/cc-connect/main.go index 96c887aaf6..b008f94988 100644 --- a/cmd/cc-connect/main.go +++ b/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 { diff --git a/config.example.toml b/config.example.toml index 9ce8db32d0..a978111bc7 100644 --- a/config.example.toml +++ b/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 / 模式说明: diff --git a/config/config.go b/config/config.go index 908027db38..d248dac907 100644 --- a/config/config.go +++ b/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 --`. 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 } diff --git a/config/config_test.go b/config/config_test.go index 505067234e..00e95b16d3 100644 --- a/config/config_test.go +++ b/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" diff --git a/core/cuj_test.go b/core/cuj_test.go index f62495f910..9f309ecea3 100644 --- a/core/cuj_test.go +++ b/core/cuj_test.go @@ -211,6 +211,29 @@ func (s *cujAgentSession) getSentPrompts() []string { return out } +// cujSteeringAgent exposes the optional in-flight steering capability while +// retaining the controllable delayed response behavior of cujAgentSession. +type cujSteeringAgent struct { + session *cujSteeringAgentSession +} + +func (a *cujSteeringAgent) Name() string { return "cuj-steering" } +func (a *cujSteeringAgent) StartSession(_ context.Context, _ string) (AgentSession, error) { + return a.session, nil +} +func (a *cujSteeringAgent) ListSessions(_ context.Context) ([]AgentSessionInfo, error) { + return nil, nil +} +func (a *cujSteeringAgent) Stop() error { return nil } + +type cujSteeringAgentSession struct { + *cujAgentSession +} + +func (s *cujSteeringAgentSession) SteerTurn(_ string) error { + return nil +} + // --------------------------------------------------------------------------- // cujEnv bundles the engine + platform stub + agent for a single CUJ run. // --------------------------------------------------------------------------- @@ -1120,6 +1143,99 @@ func TestCUJ_A2_MultiTurnAgentReceivesHistory(t *testing.T) { } } +// CUJ-A8 · A plain-text follow-up sent while a task is running is appended to +// that same turn, acknowledged immediately, and retained in /history. +// +// User actions: (1) start a slow task, (2) send a follow-up while it is busy, +// (3) open /history after completion. +func TestCUJ_A8_BusyFollowUpSteersCurrentTurn(t *testing.T) { + dir := t.TempDir() + platform := &stubPlatformEngine{n: "test"} + baseSession := newCUJAgentSession() + baseSession.reply = "finished with the follow-up" + baseSession.delayMs = 300 + steeringSession := &cujSteeringAgentSession{cujAgentSession: baseSession} + agent := &cujSteeringAgent{session: steeringSession} + engine := NewEngine("test", agent, []Platform{platform}, filepath.Join(dir, "sessions.json"), LangEnglish) + engine.SetBusyMessageMode("steer") + key := "test:steer-cuj" + + // 1. Start a turn that remains in flight long enough for a follow-up. + engine.ReceiveMessage(platform, &Message{ + SessionKey: key, + Platform: "test", + MessageID: "initial", + UserID: "steer-cuj", + UserName: "Steer CUJ", + Content: "prepare the change", + ReplyCtx: "ctx-initial", + }) + waitUntil := func(reason string, cond func() bool) { + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if cond() { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatalf("timed out waiting for %s; sent=%v", reason, platform.getSent()) + } + waitUntil("initial turn to start", func() bool { + return len(baseSession.getSentPrompts()) == 1 + }) + + // 2. Add information while the same session is busy. + engine.ReceiveMessage(platform, &Message{ + SessionKey: key, + Platform: "test", + MessageID: "follow-up", + UserID: "steer-cuj", + UserName: "Steer CUJ", + Content: "also update the documentation", + ReplyCtx: "ctx-follow-up", + }) + waitUntil("steering acknowledgement", func() bool { + for _, sent := range platform.getSent() { + if strings.Contains(sent, engine.i18n.T(MsgMessageSteered)) { + return true + } + } + return false + }) + waitUntil("initial turn completion", func() bool { + return !engine.sessions.GetOrCreateActive(key).Busy() + }) + waitUntil("final response", func() bool { + for _, sent := range platform.getSent() { + if strings.Contains(sent, "finished with the follow-up") { + return true + } + } + return false + }) + for _, sent := range platform.getSent() { + if strings.Contains(sent, engine.i18n.T(MsgMessageQueued)) { + t.Fatalf("user saw a queue acknowledgement instead of steering: %q", sent) + } + } + + // 3. The user opens history and sees both inputs retained in one task. + platform.clearSent() + engine.ReceiveMessage(platform, &Message{ + SessionKey: key, + Platform: "test", + MessageID: "history", + UserID: "steer-cuj", + Content: "/history", + ReplyCtx: "ctx-history", + }) + waitUntil("history response", func() bool { return len(platform.getSent()) > 0 }) + historyOutput := strings.Join(platform.getSent(), "\n") + if !strings.Contains(historyOutput, "prepare the change") || !strings.Contains(historyOutput, "also update the documentation") { + t.Fatalf("history output does not retain both user messages: %q", historyOutput) + } +} + // CUJ-A3 · User uploads image → engine routes it to the agent. // (No real vision LLM; we assert the image attachment reaches the agent.) func TestCUJ_A3_ImageReachesAgent(t *testing.T) { diff --git a/core/engine.go b/core/engine.go index 5abc4f5971..1f6b58ea54 100644 --- a/core/engine.go +++ b/core/engine.go @@ -392,6 +392,7 @@ type Engine struct { // 同时保留已保存的 session ID，便于下次继续恢复。 agentSessionIdleTimeoutNanos atomic.Int64 agentSessionIdleSeq atomic.Uint64 + steerBusyMessages atomic.Bool maxQueuedMessages int dirHistory *DirHistory baseWorkDir string @@ -929,6 +930,13 @@ func (e *Engine) SetAgentSessionIdleTimeout(d time.Duration) { e.agentSessionIdleTimeoutNanos.Store(int64(d)) } +// SetBusyMessageMode controls whether plain-text messages received during an +// active turn are queued (the default) or offered to an agent's optional +// steering capability. Unknown values safely resolve to queue mode. +func (e *Engine) SetBusyMessageMode(mode string) { + e.steerBusyMessages.Store(strings.EqualFold(strings.TrimSpace(mode), "steer")) +} + func (e *Engine) cancelAllAgentSessionIdleCloses() { e.interactiveMu.Lock() states := make([]*interactiveState, 0, len(e.interactiveStates)) @@ -2970,6 +2978,12 @@ func (e *Engine) handleMessage(p Platform, msg *Message) { e.reply(p, msg.ReplyCtx, e.i18n.T(MsgPreviousProcessing)) return } + // Sessions with a native steering capability can append plain-text + // follow-ups to the turn already in flight. Attachments remain queued so + // their storage and delivery semantics stay unchanged. + if e.steerBusyMessages.Load() && e.steerMessageForBusySession(p, msg, interactiveKey, session, sessions) { + return + } // Session is busy — try to queue the message for the running turn // so the agent processes it immediately after the current turn ends. if e.queueMessageForBusySession(p, msg, interactiveKey) { @@ -3160,6 +3174,68 @@ func (e *Engine) queueMessageForBusySession(p Platform, msg *Message, interactiv return true } +// steerMessageForBusySession attempts to append a plain-text message to the +// active agent turn. It returns true only when the message was handled (either +// steered successfully or rejected as stale); unsupported sessions and RPC +// failures return false so the caller can use the existing FIFO queue. +func (e *Engine) steerMessageForBusySession(p Platform, msg *Message, interactiveKey string, session *Session, sessions *SessionManager) bool { + if msg == nil || strings.TrimSpace(msg.Content) == "" || len(msg.Images) > 0 || len(msg.Files) > 0 { + return false + } + + e.interactiveMu.Lock() + state, ok := e.interactiveStates[interactiveKey] + if !ok || state == nil { + e.interactiveMu.Unlock() + return false + } + state.mu.Lock() + e.interactiveMu.Unlock() + + if e.isStaleUserMessageLocked(state, msg.UserMessageTimeMs) { + snap := userMessageWatermarkSnapshotLocked(state) + state.mu.Unlock() + e.logStaleUserMessageDropped("reject_before_steer", msg, interactiveKey, snap) + return true + } + agentSession := state.agentSession + state.mu.Unlock() + + if agentSession == nil || !agentSession.Alive() { + return false + } + steerer, ok := agentSession.(AgentSessionSteerer) + if !ok { + return false + } + + prompt := e.buildSenderPrompt(msg.Content, msg.UserID, msg.UserName, msg.Platform, msg.SessionKey, msg.ChannelKey) + if err := steerer.SteerTurn(prompt); err != nil { + slog.Warn("failed to steer busy-session message; falling back to queue", + "error", err, + "session", msg.SessionKey, + "interactive_key", interactiveKey, + "msg_id", msg.MessageID, + ) + return false + } + + session.TouchUserActivity() + session.AddHistory("user", msg.Content) + sessions.Save() + e.noteUserMessageAccepted(interactiveKey, msg.UserMessageTimeMs) + runMessageAccepted(msg) + + slog.Info("message steered into busy session", + "session", msg.SessionKey, + "interactive_key", interactiveKey, + "user", msg.UserName, + "msg_id", msg.MessageID, + ) + e.reply(p, msg.ReplyCtx, e.i18n.T(MsgMessageSteered)) + return true +} + // ensureInteractiveStateForQueueing creates a placeholder interactiveState // entry if none exists. This allows messages arriving while the agent session // is still starting up to be queued instead of dropped (issue #565). @@ -6264,7 +6340,14 @@ func (e *Engine) cmdPs(p Platform, msg *Message, args []string) { e.interactiveMu.Lock() state, ok := e.interactiveStates[iKey] e.interactiveMu.Unlock() - if !ok || state == nil || state.agentSession == nil || !state.agentSession.Alive() { + if !ok || state == nil { + e.reply(p, msg.ReplyCtx, e.i18n.T(MsgPsNoSession)) + return + } + state.mu.Lock() + agentSession := state.agentSession + state.mu.Unlock() + if agentSession == nil || !agentSession.Alive() { e.reply(p, msg.ReplyCtx, e.i18n.T(MsgPsNoSession)) return } @@ -6273,15 +6356,28 @@ func (e *Engine) cmdPs(p Platform, msg *Message, args []string) { // session lock and races with concurrent normal messages on the CLI's // stdin, so reject instead. _, sessions := e.sessionContextForKey(msg.SessionKey) - if session := sessions.GetOrCreateActive(msg.SessionKey); !session.Busy() { + session := sessions.GetOrCreateActive(msg.SessionKey) + if !session.Busy() { e.reply(p, msg.ReplyCtx, e.i18n.T(MsgPsNoSession)) return } - if err := state.agentSession.Send(text, "", nil, nil); err != nil { + steerer, ok := agentSession.(AgentSessionSteerer) + if !ok { + slog.Warn("ps: agent session does not support turn steering", "session", msg.SessionKey) + e.reply(p, msg.ReplyCtx, e.i18n.T(MsgPsSendFailed)) + return + } + prompt := e.buildSenderPrompt(text, msg.UserID, msg.UserName, msg.Platform, msg.SessionKey, msg.ChannelKey) + if err := steerer.SteerTurn(prompt); err != nil { slog.Error("ps: send failed", "error", err) e.reply(p, msg.ReplyCtx, e.i18n.T(MsgPsSendFailed)) return } + session.TouchUserActivity() + session.AddHistory("user", text) + sessions.Save() + e.noteUserMessageAccepted(iKey, msg.UserMessageTimeMs) + runMessageAccepted(msg) e.reply(p, msg.ReplyCtx, e.i18n.T(MsgPsSent)) } diff --git a/core/engine_test.go b/core/engine_test.go index ce806d11d4..999bddf996 100644 --- a/core/engine_test.go +++ b/core/engine_test.go @@ -8479,6 +8479,13 @@ type queuingAgentSession struct { sendMu sync.Mutex } +type steeringAgentSession struct { + *queuingAgentSession + steerMu sync.Mutex + steerCalls []string + steerErr error +} + func newQueuingSession(id string) *queuingAgentSession { return &queuingAgentSession{ controllableAgentSession: controllableAgentSession{ @@ -8490,6 +8497,10 @@ func newQueuingSession(id string) *queuingAgentSession { } } +func newSteeringSession(id string) *steeringAgentSession { + return &steeringAgentSession{queuingAgentSession: newQueuingSession(id)} +} + func (s *queuingAgentSession) Send(prompt string, _ string, _ []ImageAttachment, _ []FileAttachment) error { s.sendMu.Lock() s.sendCalls = append(s.sendCalls, prompt) @@ -8497,6 +8508,13 @@ func (s *queuingAgentSession) Send(prompt string, _ string, _ []ImageAttachment, return nil } +func (s *steeringAgentSession) SteerTurn(prompt string) error { + s.steerMu.Lock() + defer s.steerMu.Unlock() + s.steerCalls = append(s.steerCalls, prompt) + return s.steerErr +} + // blockingSendAgentSession blocks in Send until unblock is closed, mimicking agents // whose Send does not return until the prompt turn completes (e.g. ACP session/prompt). type blockingSendAgentSession struct { @@ -8847,6 +8865,209 @@ func TestQueueMessageForBusySession_FIFODequeue(t *testing.T) { state.mu.Unlock() } +func TestHandleMessage_BusySteerableSessionSteersPlainText(t *testing.T) { + p := &stubPlatformEngine{n: "test"} + sess := newSteeringSession("steer-success") + e := NewEngine("test", &stubAgent{}, []Platform{p}, "", LangEnglish) + e.SetBusyMessageMode("steer") + + key := "test:steer-user" + state := &interactiveState{ + agentSession: sess, + platform: p, + currentTurnUserMessageTimeMs: 1_000, + } + e.interactiveMu.Lock() + e.interactiveStates[key] = state + e.interactiveMu.Unlock() + + session := e.sessions.GetOrCreateActive(key) + if !session.TryLock() { + t.Fatal("expected session lock") + } + defer session.UnlockWithoutUpdate() + + accepted := false + e.ReceiveMessage(p, &Message{ + SessionKey: key, + Platform: "test", + MessageID: "follow-up-1", + UserID: "steer-user", + UserName: "Steer User", + Content: "also update the docs", + ReplyCtx: "ctx-follow-up", + UserMessageTimeMs: 2_000, + OnAccepted: func() { accepted = true }, + }) + + sess.steerMu.Lock() + steered := append([]string(nil), sess.steerCalls...) + sess.steerMu.Unlock() + if len(steered) != 1 || steered[0] != "also update the docs" { + t.Fatalf("steer calls = %#v, want one follow-up", steered) + } + if !accepted { + t.Fatal("steered message did not run OnAccepted") + } + state.mu.Lock() + queueDepth := len(state.pendingMessages) + watermark := state.currentTurnUserMessageTimeMs + state.mu.Unlock() + if queueDepth != 0 { + t.Fatalf("pending messages = %d, want 0 after steering", queueDepth) + } + if watermark != 2_000 { + t.Fatalf("current turn watermark = %d, want 2000", watermark) + } + + history := session.GetHistory(0) + if len(history) != 1 || history[0].Role != "user" || history[0].Content != "also update the docs" { + t.Fatalf("history = %#v, want steered user message", history) + } + foundAck := false + for _, sent := range p.getSent() { + if strings.Contains(sent, e.i18n.T(MsgMessageSteered)) { + foundAck = true + break + } + } + if !foundAck { + t.Fatalf("expected steering acknowledgement, got %v", p.getSent()) + } +} + +func TestHandleMessage_BusySteerableSessionDefaultsToQueue(t *testing.T) { + p := &stubPlatformEngine{n: "test"} + sess := newSteeringSession("steer-default-queue") + e := NewEngine("test", &stubAgent{}, []Platform{p}, "", LangEnglish) + + key := "test:steer-default-queue" + state := &interactiveState{agentSession: sess, platform: p} + e.interactiveMu.Lock() + e.interactiveStates[key] = state + e.interactiveMu.Unlock() + + session := e.sessions.GetOrCreateActive(key) + if !session.TryLock() { + t.Fatal("expected session lock") + } + defer session.UnlockWithoutUpdate() + + e.ReceiveMessage(p, &Message{ + SessionKey: key, + Platform: "test", + MessageID: "default-queue", + Content: "keep the compatible queue behavior", + ReplyCtx: "ctx-default-queue", + }) + + sess.steerMu.Lock() + steerCount := len(sess.steerCalls) + sess.steerMu.Unlock() + if steerCount != 0 { + t.Fatalf("default mode steered %d time(s), want 0", steerCount) + } + state.mu.Lock() + queueDepth := len(state.pendingMessages) + state.mu.Unlock() + if queueDepth != 1 { + t.Fatalf("pending messages = %d, want 1 in default queue mode", queueDepth) + } +} + +func TestHandleMessage_SteerFailureFallsBackToQueue(t *testing.T) { + p := &stubPlatformEngine{n: "test"} + sess := newSteeringSession("steer-fallback") + sess.steerErr = errors.New("turn changed") + e := NewEngine("test", &stubAgent{}, []Platform{p}, "", LangEnglish) + e.SetBusyMessageMode("steer") + + key := "test:steer-fallback" + state := &interactiveState{agentSession: sess, platform: p} + e.interactiveMu.Lock() + e.interactiveStates[key] = state + e.interactiveMu.Unlock() + + session := e.sessions.GetOrCreateActive(key) + if !session.TryLock() { + t.Fatal("expected session lock") + } + defer session.UnlockWithoutUpdate() + + e.ReceiveMessage(p, &Message{ + SessionKey: key, + Platform: "test", + MessageID: "follow-up-fallback", + Content: "do this next if steering fails", + ReplyCtx: "ctx-fallback", + }) + + state.mu.Lock() + queueDepth := len(state.pendingMessages) + queuedContent := "" + if queueDepth > 0 { + queuedContent = state.pendingMessages[0].content + } + state.mu.Unlock() + if queueDepth != 1 || queuedContent != "do this next if steering fails" { + t.Fatalf("queue = (%d, %q), want fallback message", queueDepth, queuedContent) + } + if got := session.GetHistory(0); len(got) != 0 { + t.Fatalf("history = %#v, queued fallback must not be persisted before drain", got) + } + foundQueued := false + for _, sent := range p.getSent() { + if strings.Contains(sent, e.i18n.T(MsgMessageQueued)) { + foundQueued = true + break + } + } + if !foundQueued { + t.Fatalf("expected queue fallback acknowledgement, got %v", p.getSent()) + } +} + +func TestHandleMessage_BusyAttachmentRemainsQueued(t *testing.T) { + p := &stubPlatformEngine{n: "test"} + sess := newSteeringSession("steer-file") + e := NewEngine("test", &stubAgent{}, []Platform{p}, "", LangEnglish) + e.SetBusyMessageMode("steer") + + key := "test:steer-file" + state := &interactiveState{agentSession: sess, platform: p} + e.interactiveMu.Lock() + e.interactiveStates[key] = state + e.interactiveMu.Unlock() + + session := e.sessions.GetOrCreateActive(key) + if !session.TryLock() { + t.Fatal("expected session lock") + } + defer session.UnlockWithoutUpdate() + + e.ReceiveMessage(p, &Message{ + SessionKey: key, + Platform: "test", + MessageID: "follow-up-file", + Content: "use this file too", + Files: []FileAttachment{{}}, + ReplyCtx: "ctx-file", + }) + + sess.steerMu.Lock() + steerCount := len(sess.steerCalls) + sess.steerMu.Unlock() + if steerCount != 0 { + t.Fatalf("attachment was steered %d time(s), want queue-only", steerCount) + } + state.mu.Lock() + queueDepth := len(state.pendingMessages) + state.mu.Unlock() + if queueDepth != 1 { + t.Fatalf("pending messages = %d, want 1 attachment message", queueDepth) + } +} + func TestQueuedUserMessageStaleForDrainIgnoresOtherPendingMessages(t *testing.T) { e := &Engine{} state := &interactiveState{ @@ -10128,7 +10349,7 @@ func TestCmdPs_IdleSession_RepliesNoSession(t *testing.T) { func TestCmdPs_BusySession_InjectsToAgent(t *testing.T) { p := &stubPlatformEngine{n: "test"} - sess := newQueuingSession("ps-busy") + sess := newSteeringSession("ps-busy") e := NewEngine("test", &stubAgent{}, []Platform{p}, "", LangEnglish) key := "test:user1" @@ -10147,11 +10368,11 @@ func TestCmdPs_BusySession_InjectsToAgent(t *testing.T) { msg := &Message{SessionKey: key, Content: "/ps add unit tests", ReplyCtx: "ctx"} e.cmdPs(p, msg, []string{"add", "unit", "tests"}) - sess.sendMu.Lock() - calls := append([]string(nil), sess.sendCalls...) - sess.sendMu.Unlock() + sess.steerMu.Lock() + calls := append([]string(nil), sess.steerCalls...) + sess.steerMu.Unlock() if len(calls) != 1 || calls[0] != "add unit tests" { - t.Fatalf("expected Send(\"add unit tests\"), got %v", calls) + t.Fatalf("expected SteerTurn(\"add unit tests\"), got %v", calls) } sent := p.getSent() @@ -10167,6 +10388,44 @@ func TestCmdPs_BusySession_InjectsToAgent(t *testing.T) { } } +func TestCmdPs_BusySessionWithoutSteeringCapabilityFailsSafely(t *testing.T) { + p := &stubPlatformEngine{n: "test"} + sess := newQueuingSession("ps-unsupported") + e := NewEngine("test", &stubAgent{}, []Platform{p}, "", LangEnglish) + + key := "test:user1" + state := &interactiveState{agentSession: sess, platform: p} + e.interactiveMu.Lock() + e.interactiveStates[key] = state + e.interactiveMu.Unlock() + + session := e.sessions.GetOrCreateActive(key) + if !session.TryLock() { + t.Fatal("expected TryLock to succeed") + } + defer session.Unlock() + + msg := &Message{SessionKey: key, Content: "/ps do not start another turn", ReplyCtx: "ctx"} + e.cmdPs(p, msg, []string{"do", "not", "start", "another", "turn"}) + + sess.sendMu.Lock() + sendCount := len(sess.sendCalls) + sess.sendMu.Unlock() + if sendCount != 0 { + t.Fatalf("unsupported /ps called Send %d time(s), want 0", sendCount) + } + foundFailure := false + for _, sent := range p.getSent() { + if strings.Contains(sent, e.i18n.T(MsgPsSendFailed)) { + foundFailure = true + break + } + } + if !foundFailure { + t.Fatalf("expected MsgPsSendFailed reply, got %v", p.getSent()) + } +} + // --- 3. executeCardAction routing --- func TestExecuteCardAction_CronEnable(t *testing.T) { diff --git a/core/i18n.go b/core/i18n.go index 410588a041..051c309aed 100644 --- a/core/i18n.go +++ b/core/i18n.go @@ -174,6 +174,7 @@ const ( MsgPreviousProcessing MsgKey = "previous_processing" MsgQueueFull MsgKey = "queue_full" MsgMessageQueued MsgKey = "message_queued" + MsgMessageSteered MsgKey = "message_steered" MsgNoToolsAllowed MsgKey = "no_tools_allowed" MsgCurrentTools MsgKey = "current_tools" MsgCurrentSession MsgKey = "current_session" @@ -744,6 +745,13 @@ var messages = map[MsgKey]map[Language]string{ LangJapanese: "📬 メッセージを受信しました。現在のタスク完了後に処理します。", LangSpanish: "📬 Mensaje recibido — se procesará después de que termine la tarea actual.", }, + MsgMessageSteered: { + LangEnglish: "🧭 Added to the current task.", + LangChinese: "🧭 补充信息已加入当前任务。", + LangTraditionalChinese: "🧭 補充資訊已加入目前任務。", + LangJapanese: "🧭 補足情報を現在のタスクに追加しました。", + LangSpanish: "🧭 Información adicional añadida a la tarea actual.", + }, MsgQueueFull: { LangEnglish: "📬 Message queue is full (%d pending). Please wait for current tasks to complete.", LangChinese: "📬 消息队列已满（%d 条待处理）。请等待当前任务完成。",