mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-31 19:16:29 +00:00
fix(codex): scope app-server notifications to active turn
This commit is contained in:
Tanmay Dixit
Teknium
parent
453d0ee598
commit
63e363f306
2 changed files with 325 additions and 0 deletions
|
|
@ -92,6 +92,67 @@ class TurnResult:
|
|||
_TURN_ABORTED_MARKERS = ("<turn_aborted>", "<turn_aborted/>")
|
||||
|
||||
|
||||
def _notification_belongs_to_turn(
|
||||
note: dict,
|
||||
*,
|
||||
thread_id: Optional[str],
|
||||
turn_id: Optional[str],
|
||||
) -> bool:
|
||||
"""Return whether a multiplexed notification belongs to this turn.
|
||||
|
||||
Codex app-server can carry parent and hosted subagent threads over one
|
||||
JSON-RPC connection. An explicitly foreign child or
|
||||
stale-turn event must not mutate the active parent's transcript or mark
|
||||
its turn complete. Unscoped notifications remain accepted for protocol
|
||||
compatibility.
|
||||
"""
|
||||
if not isinstance(note, dict):
|
||||
return False
|
||||
params = note.get("params") or {}
|
||||
if not isinstance(params, dict):
|
||||
return True
|
||||
|
||||
nested_turn = params.get("turn") or {}
|
||||
nested_item = params.get("item") or {}
|
||||
|
||||
observed_thread_id = params.get("threadId") or params.get("thread_id")
|
||||
if observed_thread_id is None and isinstance(nested_turn, dict):
|
||||
observed_thread_id = (
|
||||
nested_turn.get("threadId")
|
||||
or nested_turn.get("thread_id")
|
||||
)
|
||||
if observed_thread_id is None and isinstance(nested_item, dict):
|
||||
observed_thread_id = (
|
||||
nested_item.get("threadId")
|
||||
or nested_item.get("thread_id")
|
||||
)
|
||||
|
||||
if (
|
||||
thread_id is not None
|
||||
and observed_thread_id is not None
|
||||
and str(observed_thread_id) != str(thread_id)
|
||||
):
|
||||
return False
|
||||
|
||||
observed_turn_id = params.get("turnId") or params.get("turn_id")
|
||||
if observed_turn_id is None and isinstance(nested_turn, dict):
|
||||
observed_turn_id = nested_turn.get("id") or nested_turn.get("turnId")
|
||||
if observed_turn_id is None and isinstance(nested_item, dict):
|
||||
observed_turn_id = (
|
||||
nested_item.get("turnId")
|
||||
or nested_item.get("turn_id")
|
||||
)
|
||||
|
||||
if (
|
||||
turn_id is not None
|
||||
and observed_turn_id is not None
|
||||
and str(observed_turn_id) != str(turn_id)
|
||||
):
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
|
||||
def _coerce_turn_input_text(user_input: Any) -> str:
|
||||
"""Collapse Hermes/OpenAI rich content into app-server text input.
|
||||
|
||||
|
|
@ -505,6 +566,17 @@ class CodexAppServerSession:
|
|||
pending = self._client.take_notification(timeout=0)
|
||||
if pending is None:
|
||||
break
|
||||
if not _notification_belongs_to_turn(
|
||||
pending,
|
||||
thread_id=self._thread_id,
|
||||
turn_id=result.turn_id,
|
||||
):
|
||||
logger.debug(
|
||||
"ignoring foreign codex notification while draining "
|
||||
"server request: method=%s",
|
||||
pending.get("method"),
|
||||
)
|
||||
continue
|
||||
# Mirror the main notification-handling block below so
|
||||
# display events surface and stay in step with projector
|
||||
# state. Without this, item/started / item/completed
|
||||
|
|
@ -550,6 +622,16 @@ class CodexAppServerSession:
|
|||
continue
|
||||
|
||||
method = note.get("method", "")
|
||||
if not _notification_belongs_to_turn(
|
||||
note,
|
||||
thread_id=self._thread_id,
|
||||
turn_id=result.turn_id,
|
||||
):
|
||||
logger.debug(
|
||||
"ignoring foreign codex notification: method=%s", method
|
||||
)
|
||||
continue
|
||||
|
||||
if self._on_event is not None:
|
||||
try:
|
||||
self._on_event(note)
|
||||
|
|
@ -737,6 +819,16 @@ class CodexAppServerSession:
|
|||
continue
|
||||
|
||||
method = note.get("method", "")
|
||||
if not _notification_belongs_to_turn(
|
||||
note,
|
||||
thread_id=self._thread_id,
|
||||
turn_id=result.turn_id,
|
||||
):
|
||||
logger.debug(
|
||||
"ignoring foreign codex notification: method=%s", method
|
||||
)
|
||||
continue
|
||||
|
||||
if self._on_event is not None:
|
||||
try:
|
||||
self._on_event(note)
|
||||
|
|
|
|||
|
|
@ -95,6 +95,17 @@ class FakeClient:
|
|||
|
||||
# Test helpers
|
||||
def queue_notification(self, method: str, **params):
|
||||
# Keep legacy fixture shorthand aligned with the IDs returned by the
|
||||
# fake thread/start and turn/start responses.
|
||||
if params.get("threadId") in {"t", "th"}:
|
||||
params["threadId"] = "thread-fake-001"
|
||||
if params.get("turnId") == "tu1":
|
||||
params["turnId"] = "turn-fake-001"
|
||||
turn = params.get("turn")
|
||||
if isinstance(turn, dict) and turn.get("id") == "tu1":
|
||||
turn = dict(turn)
|
||||
turn["id"] = "turn-fake-001"
|
||||
params["turn"] = turn
|
||||
self._notifications.append({"method": method, "params": params})
|
||||
|
||||
def queue_server_request(self, method: str, request_id: Any = "srv-1", **params):
|
||||
|
|
@ -196,6 +207,173 @@ class TestRunTurn:
|
|||
# turn_id propagated for downstream session-DB linkage
|
||||
assert r.turn_id == "turn-fake-001"
|
||||
|
||||
def test_subagent_completion_does_not_end_parent_turn(self):
|
||||
"""A child completion must not become the parent's response."""
|
||||
client = FakeClient()
|
||||
client.queue_notification(
|
||||
"item/completed",
|
||||
threadId="thread-child-001",
|
||||
turnId="turn-child-001",
|
||||
item={
|
||||
"type": "agentMessage",
|
||||
"id": "child-message-1",
|
||||
"text": "child summary",
|
||||
},
|
||||
)
|
||||
client.queue_notification(
|
||||
"turn/completed",
|
||||
threadId="thread-child-001",
|
||||
turn={
|
||||
"id": "turn-child-001",
|
||||
"status": "completed",
|
||||
"error": None,
|
||||
},
|
||||
)
|
||||
client.queue_notification(
|
||||
"item/completed",
|
||||
threadId="thread-fake-001",
|
||||
turnId="turn-fake-001",
|
||||
item={
|
||||
"type": "agentMessage",
|
||||
"id": "parent-message-1",
|
||||
"text": "parent synthesis",
|
||||
},
|
||||
)
|
||||
client.queue_notification(
|
||||
"turn/completed",
|
||||
threadId="thread-fake-001",
|
||||
turn={
|
||||
"id": "turn-fake-001",
|
||||
"status": "completed",
|
||||
"error": None,
|
||||
},
|
||||
)
|
||||
|
||||
result = make_session(client).run_turn("delegate this", turn_timeout=2.0)
|
||||
|
||||
assert result.final_text == "parent synthesis"
|
||||
assert result.interrupted is False
|
||||
assert result.error is None
|
||||
assert result.projected_messages == [
|
||||
{"role": "assistant", "content": "parent synthesis"}
|
||||
]
|
||||
|
||||
def test_stale_completion_on_parent_thread_is_ignored(self):
|
||||
"""A late completion from the previous parent turn is not terminal."""
|
||||
client = FakeClient()
|
||||
client.queue_notification(
|
||||
"item/completed",
|
||||
threadId="thread-fake-001",
|
||||
turnId="turn-previous",
|
||||
item={
|
||||
"type": "agentMessage",
|
||||
"id": "stale-message",
|
||||
"text": "stale previous answer",
|
||||
},
|
||||
)
|
||||
client.queue_notification(
|
||||
"turn/completed",
|
||||
threadId="thread-fake-001",
|
||||
turn={
|
||||
"id": "turn-previous",
|
||||
"status": "completed",
|
||||
"error": None,
|
||||
},
|
||||
)
|
||||
client.queue_notification(
|
||||
"item/completed",
|
||||
threadId="thread-fake-001",
|
||||
turnId="turn-fake-001",
|
||||
item={
|
||||
"type": "agentMessage",
|
||||
"id": "current-message",
|
||||
"text": "current parent answer",
|
||||
},
|
||||
)
|
||||
client.queue_notification(
|
||||
"turn/completed",
|
||||
threadId="thread-fake-001",
|
||||
turn={
|
||||
"id": "turn-fake-001",
|
||||
"status": "completed",
|
||||
"error": None,
|
||||
},
|
||||
)
|
||||
|
||||
result = make_session(client).run_turn("new prompt", turn_timeout=2.0)
|
||||
|
||||
assert result.final_text == "current parent answer"
|
||||
assert result.projected_messages == [
|
||||
{"role": "assistant", "content": "current parent answer"}
|
||||
]
|
||||
|
||||
def test_foreign_completion_in_server_request_drain_is_ignored(self):
|
||||
"""Approval draining must not project a child result into the parent."""
|
||||
client = FakeClient()
|
||||
client.queue_server_request(
|
||||
"item/commandExecution/requestApproval",
|
||||
request_id="approval-1",
|
||||
command="pwd",
|
||||
cwd="/tmp",
|
||||
)
|
||||
client.queue_notification(
|
||||
"item/completed",
|
||||
threadId="thread-child-001",
|
||||
turnId="turn-child-001",
|
||||
item={
|
||||
"type": "agentMessage",
|
||||
"id": "child-message",
|
||||
"text": "child drain summary",
|
||||
},
|
||||
)
|
||||
client.queue_notification(
|
||||
"turn/completed",
|
||||
threadId="thread-child-001",
|
||||
turn={
|
||||
"id": "turn-child-001",
|
||||
"status": "completed",
|
||||
"error": None,
|
||||
},
|
||||
)
|
||||
|
||||
original_respond = client.respond
|
||||
|
||||
def respond_and_release_parent(request_id, response):
|
||||
original_respond(request_id, response)
|
||||
client.queue_notification(
|
||||
"item/completed",
|
||||
threadId="thread-fake-001",
|
||||
turnId="turn-fake-001",
|
||||
item={
|
||||
"type": "agentMessage",
|
||||
"id": "parent-message",
|
||||
"text": "parent after approval",
|
||||
},
|
||||
)
|
||||
client.queue_notification(
|
||||
"turn/completed",
|
||||
threadId="thread-fake-001",
|
||||
turn={
|
||||
"id": "turn-fake-001",
|
||||
"status": "completed",
|
||||
"error": None,
|
||||
},
|
||||
)
|
||||
|
||||
client.respond = respond_and_release_parent
|
||||
session = make_session(
|
||||
client,
|
||||
request_routing=_ServerRequestRouting(auto_approve_exec=True),
|
||||
)
|
||||
|
||||
result = session.run_turn("delegate then continue", turn_timeout=2.0)
|
||||
|
||||
assert client.responses == [("approval-1", {"decision": "accept"})]
|
||||
assert result.final_text == "parent after approval"
|
||||
assert result.projected_messages == [
|
||||
{"role": "assistant", "content": "parent after approval"}
|
||||
]
|
||||
|
||||
def test_token_usage_notification_is_captured(self):
|
||||
client = FakeClient()
|
||||
client.queue_notification(
|
||||
|
|
@ -528,6 +706,61 @@ class TestCompactThread:
|
|||
assert r.token_usage_last["totalTokens"] == 12
|
||||
assert r.model_context_window == 200000
|
||||
|
||||
def test_compact_thread_ignores_foreign_child_completion(self):
|
||||
client = FakeClient()
|
||||
client.queue_notification(
|
||||
"item/completed",
|
||||
threadId="thread-child-001",
|
||||
turnId="child-compact-turn",
|
||||
item={
|
||||
"type": "agentMessage",
|
||||
"id": "child-compact-message",
|
||||
"text": "child compact summary",
|
||||
},
|
||||
)
|
||||
client.queue_notification(
|
||||
"turn/completed",
|
||||
threadId="thread-child-001",
|
||||
turn={
|
||||
"id": "child-compact-turn",
|
||||
"status": "completed",
|
||||
"error": None,
|
||||
},
|
||||
)
|
||||
client.queue_notification(
|
||||
"turn/started",
|
||||
threadId="thread-fake-001",
|
||||
turn={"id": "compact-turn-1"},
|
||||
)
|
||||
client.queue_notification(
|
||||
"item/completed",
|
||||
threadId="thread-fake-001",
|
||||
turnId="compact-turn-1",
|
||||
item={
|
||||
"type": "agentMessage",
|
||||
"id": "parent-compact-message",
|
||||
"text": "parent compacted",
|
||||
},
|
||||
)
|
||||
client.queue_notification(
|
||||
"turn/completed",
|
||||
threadId="thread-fake-001",
|
||||
turn={
|
||||
"id": "compact-turn-1",
|
||||
"status": "completed",
|
||||
"error": None,
|
||||
},
|
||||
)
|
||||
|
||||
result = make_session(client).compact_thread(turn_timeout=2.0)
|
||||
|
||||
assert result.error is None
|
||||
assert result.turn_id == "compact-turn-1"
|
||||
assert result.final_text == "parent compacted"
|
||||
assert result.projected_messages == [
|
||||
{"role": "assistant", "content": "parent compacted"}
|
||||
]
|
||||
|
||||
def test_compact_thread_interrupted_returns_non_success(self):
|
||||
client = FakeClient()
|
||||
client.queue_notification(
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue