mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-23 16:36:23 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 7

feat(codex): honor redirect and hard stop in the app-server runtime

Browse source 
The Codex app-server runtime bypasses the main conversation loop and drives
its own subprocess turn, so it needs first-class hooks rather than the
OpenAI-loop interrupt path.

- `AIAgent.interrupt()` now forwards a hard stop to
  `CodexAppServerSession.request_interrupt()`, and `redirect()` uses Codex's
  native `turn/steer` protocol instead of cancelling the subprocess.
- `run_turn()` no longer clears an interrupt that arrived during
  `ensure_started()`: a stop landing mid-startup is honored before `turn/start`,
  and the interrupt event is cleared on every exit path.
- `run_codex_app_server_turn()` mirrors the loop finalizer's interrupt handoff
  (surface `interrupted` / `interrupt_message`, then `clear_interrupt()`) on
  both the normal and exception early-return paths, so a hard stop can't leave
  `_interrupt_requested` stale for the next turn.
This commit is contained in:
Brooklyn Nicholson 2026-07-13 02:23:39 -04:00
parent cbf5b05c70
commit f6d2ee4afc
4 changed files with 146 additions and 14 deletions
Download patch file Download diff file Expand all files Collapse all files

34
agent/codex_runtime.py
View file

@ -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

47
agent/transports/codex_app_server_session.py
View file

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

28
tests/agent/test_codex_app_server_persist.py
View file

@ -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

51
tests/agent/transports/test_codex_app_server_session.py
View file

@ -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