diff --git a/agent/codex_runtime.py b/agent/codex_runtime.py index 91c2af3e995f..59f9bac25a26 100644 --- a/agent/codex_runtime.py +++ b/agent/codex_runtime.py @@ -702,6 +702,16 @@ def run_codex_app_server_turn( except Exception: pass agent._codex_session = None + _user_interrupted = bool( + getattr(agent, "_interrupt_requested", False) + ) + _interrupt_message = ( + getattr(agent, "_interrupt_message", None) + if _user_interrupted + else None + ) + if _user_interrupted: + agent.clear_interrupt() return { "final_response": ( f"Codex app-server turn failed: {exc}. " @@ -711,9 +721,27 @@ def run_codex_app_server_turn( "api_calls": 0, "completed": False, "partial": True, + "interrupted": _user_interrupted, + **( + {"interrupt_message": _interrupt_message} + if _interrupt_message + else {} + ), "error": str(exc), } + # This runtime bypasses the normal conversation-loop finalizer. Mirror its + # interrupt handoff/cleanup so a hard stop cannot poison the next turn and a + # message-bearing compatibility interrupt can still be replayed by callers. + _user_interrupted = bool( + turn.interrupted and getattr(agent, "_interrupt_requested", False) + ) + _interrupt_message = ( + getattr(agent, "_interrupt_message", None) if _user_interrupted else None + ) + if _user_interrupted: + agent.clear_interrupt() + # If the turn signalled the underlying client is wedged (deadline # blown, post-tool watchdog tripped, OAuth refresh died, subprocess # exited), retire the session so the next turn respawns codex @@ -819,6 +847,12 @@ def run_codex_app_server_turn( "api_calls": api_calls, "completed": not turn.interrupted and turn.error is None, "partial": turn.interrupted or turn.error is not None, + "interrupted": _user_interrupted, + **( + {"interrupt_message": _interrupt_message} + if _interrupt_message + else {} + ), "error": turn.error, # The codex app-server runtime IS an early-return path that bypasses # conversation_loop, but we flush the projected assistant/tool messages diff --git a/agent/transports/codex_app_server_session.py b/agent/transports/codex_app_server_session.py index 2e1cc64bdcbc..7954ecbf4d52 100644 --- a/agent/transports/codex_app_server_session.py +++ b/agent/transports/codex_app_server_session.py @@ -300,6 +300,8 @@ class CodexAppServerSession: self._client: Optional[CodexAppServerClient] = None self._thread_id: Optional[str] = None self._interrupt_event = threading.Event() + self._active_turn_id: Optional[str] = None + self._active_turn_lock = threading.Lock() # Pending file-change items, keyed by item id. Populated on # item/started for fileChange items; consumed by the approval # bridge when codex sends item/fileChange/requestApproval. The @@ -374,6 +376,8 @@ class CodexAppServerSession: if self._closed: return self._closed = True + with self._active_turn_lock: + self._active_turn_id = None if self._client is not None: try: self._client.close() @@ -395,6 +399,33 @@ class CodexAppServerSession: and unwind. Called by AIAgent's _interrupt_requested path.""" self._interrupt_event.set() + def request_steer(self, text: str) -> bool: + """Append user guidance to the active Codex turn via ``turn/steer``.""" + cleaned = str(text or "").strip() + if not cleaned: + return False + with self._active_turn_lock: + turn_id = self._active_turn_id + thread_id = self._thread_id + client = self._client + if not turn_id or not thread_id or client is None: + return False + try: + response = client.request( + "turn/steer", + { + "threadId": thread_id, + "input": [{"type": "text", "text": cleaned}], + "expectedTurnId": turn_id, + }, + timeout=10, + ) + except (CodexAppServerError, TimeoutError): + logger.debug("turn/steer rejected for active Codex turn", exc_info=True) + return False + accepted_turn_id = response.get("turnId") if isinstance(response, dict) else None + return accepted_turn_id in {None, turn_id} + # ---------- diagnostics ---------- def _format_error_with_stderr( @@ -469,11 +500,18 @@ class CodexAppServerSession: # Subprocess almost certainly unhealthy — retire so the next # turn re-spawns cleanly. result.should_retire = True + self._interrupt_event.clear() return result assert self._client is not None and self._thread_id is not None result.thread_id = self._thread_id - self._interrupt_event.clear() + # Do not clear here: a hard stop can arrive while ensure_started() is + # spawning/initializing the subprocess. Honor it before launching a + # Codex turn instead of erasing the signal. + if self._interrupt_event.is_set(): + result.interrupted = True + self._interrupt_event.clear() + return result projector = CodexEventProjector() user_input_text = _coerce_turn_input_text(user_input) @@ -505,6 +543,7 @@ class CodexAppServerSession: result.error = self._format_error_with_stderr( "turn/start failed", exc ) + self._interrupt_event.clear() return result except TimeoutError as exc: # turn/start hanging is a strong signal the subprocess is wedged. @@ -514,9 +553,12 @@ class CodexAppServerSession: "turn/start timed out", exc ) result.should_retire = True + self._interrupt_event.clear() return result result.turn_id = (ts.get("turn") or {}).get("id") + with self._active_turn_lock: + self._active_turn_id = result.turn_id deadline = time.monotonic() + turn_timeout turn_complete = False # Post-tool watchdog state. last_tool_completion_at is set whenever @@ -741,6 +783,9 @@ class CodexAppServerSession: ) result.should_retire = True + with self._active_turn_lock: + self._active_turn_id = None + self._interrupt_event.clear() return result def compact_thread( diff --git a/tests/agent/test_codex_app_server_persist.py b/tests/agent/test_codex_app_server_persist.py index 001082e3f0ea..85d4a5757f7e 100644 --- a/tests/agent/test_codex_app_server_persist.py +++ b/tests/agent/test_codex_app_server_persist.py @@ -76,6 +76,34 @@ def test_codex_success_flushes_and_reports_persisted(): assert result["agent_persisted"] is True +def test_codex_user_interrupt_is_reported_and_cleared(): + agent = _make_agent(session_db=None) + turn = _make_turn() + turn.interrupted = True + turn.final_text = "" + agent._codex_session.run_turn.return_value = turn + agent._interrupt_requested = True + agent._interrupt_message = "new correction" + + def clear_interrupt(): + agent._interrupt_requested = False + agent._interrupt_message = None + + agent.clear_interrupt.side_effect = clear_interrupt + result = run_codex_app_server_turn( + agent, + user_message="hello", + original_user_message="hello", + messages=[{"role": "user", "content": "hello"}], + effective_task_id="task-1", + ) + + assert result["interrupted"] is True + assert result["interrupt_message"] == "new correction" + agent.clear_interrupt.assert_called_once_with() + assert agent._interrupt_requested is False + + def test_codex_turn_persists_each_message_exactly_once(): """The user turn (flushed at turn start) must not be duplicated; the projected assistant message must land once. Uses a real SessionDB and the diff --git a/tests/agent/transports/test_codex_app_server_session.py b/tests/agent/transports/test_codex_app_server_session.py index 5479c14ae21e..af850fc0f8d5 100644 --- a/tests/agent/transports/test_codex_app_server_session.py +++ b/tests/agent/transports/test_codex_app_server_session.py @@ -57,6 +57,8 @@ class FakeClient: return {"turn": {"id": "turn-fake-001"}} if method == "turn/interrupt": return {} + if method == "turn/steer": + return {"turnId": (params or {}).get("expectedTurnId")} return {} def notify(self, method: str, params=None): @@ -562,30 +564,53 @@ class TestRunTurn: assert r.should_retire is True assert r.final_text == "" - def test_interrupt_during_turn_issues_turn_interrupt(self): + def test_interrupt_during_startup_skips_turn_start(self): client = FakeClient() - # Don't queue turn/completed — the loop has to interrupt out - client.queue_notification( - "item/completed", - item={"type": "commandExecution", "id": "x", "command": "sleep 60", - "cwd": "/", "status": "inProgress", - "aggregatedOutput": None, "exitCode": None, - "commandActions": []}, - threadId="t", turnId="tu1", - ) s = make_session(client) s.ensure_started() - # Trip the interrupt before run_turn even consumes the notification. - # The loop will see interrupt set on its first iteration and bail. s.request_interrupt() r = s.run_turn("loop forever", turn_timeout=2.0) + + assert r.interrupted is True + assert not any(method == "turn/start" for method, _params in client.requests) + + def test_interrupt_after_turn_start_issues_turn_interrupt(self): + client = FakeClient() + s = make_session(client) + + def request_handler(method, params): + if method == "thread/start": + return {"thread": {"id": "thread-fake-001"}} + if method == "turn/start": + s.request_interrupt() + return {"turn": {"id": "turn-fake-001"}} + return {} + + client._request_handler = request_handler + r = s.run_turn("loop forever", turn_timeout=2.0) + assert r.interrupted is True - # turn/interrupt was requested with the right turnId assert any( method == "turn/interrupt" and params.get("turnId") == "turn-fake-001" for (method, params) in client.requests ) + def test_steer_appends_input_to_active_turn(self): + client = FakeClient() + s = make_session(client) + s.ensure_started() + with s._active_turn_lock: + s._active_turn_id = "turn-live-123" + + assert s.request_steer("Use Postgres instead") is True + method, params = client.requests[-1] + assert method == "turn/steer" + assert params == { + "threadId": "thread-fake-001", + "input": [{"type": "text", "text": "Use Postgres instead"}], + "expectedTurnId": "turn-live-123", + } + def test_deadline_exceeded_records_error(self): client = FakeClient() # No notifications and no completion → must hit deadline