From 4ac196c62ede17471de2e1be534556df1bf97c3a Mon Sep 17 00:00:00 2001 From: Zhexuan Yang Date: Fri, 31 Jul 2026 08:31:52 +0800 Subject: [PATCH 1/4] fix(delegation): persist immutable async origins Capture the dispatching turn source, message anchor, profile, and parent session before child construction can mutate context. Persist and restore the snapshot for profile-safe completion routing and compression continuation, while retaining legacy session-key fallback for old rows. --- gateway/run.py | 128 ++++++++-- gateway/session_context.py | 21 +- ...test_async_delegation_origin_durability.py | 241 ++++++++++++++++++ .../test_delegate_apiserver_background.py | 23 ++ tools/async_delegation.py | 101 +++++++- tools/delegate_tool.py | 64 ++--- 6 files changed, 516 insertions(+), 62 deletions(-) create mode 100644 tests/gateway/test_async_delegation_origin_durability.py diff --git a/gateway/run.py b/gateway/run.py index 911021a36ee60..01650938df658 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -13160,6 +13160,7 @@ async def _resolve_async_delegation_session( self, session_entry: SessionEntry, pinned_session_id: str, + origin_profile: str = "", ) -> Optional[SessionEntry]: """Resolve an async completion to its verified owning gateway session. @@ -13195,6 +13196,18 @@ async def _resolve_async_delegation_session( ) return None + if origin_profile and not self._async_origin_profile_matches( + pinned_row, origin_profile + ): + logger.warning( + "Async-delegation spawning session %s belongs to profile %r, " + "not immutable origin profile %r; dropping injection.", + pinned_session_id, + pinned_row.get("profile_name"), + origin_profile, + ) + return None + target_session_id = pinned_session_id follows_compression = False if pinned_row.get("ended_at"): @@ -13241,6 +13254,17 @@ async def _resolve_async_delegation_session( "unknown" if tip_row is None else "ended", ) return None + if origin_profile and not self._async_origin_profile_matches( + tip_row, origin_profile + ): + logger.warning( + "Async-delegation compression continuation %s crossed " + "profiles (%r -> %r); dropping injection.", + target_session_id, + origin_profile, + tip_row.get("profile_name"), + ) + return None route_owns_lineage = session_entry.session_id in { pinned_session_id, @@ -13306,6 +13330,13 @@ async def _resolve_async_delegation_session( ) return switched + @staticmethod + def _async_origin_profile_matches(row: dict, origin_profile: str) -> bool: + """Compare persisted session ownership with an immutable origin profile.""" + expected = str(origin_profile or "").strip() or "default" + actual = str((row or {}).get("profile_name") or "").strip() or "default" + return actual == expected + # ------------------------------------------------------------------ # Mid-run (busy-session) slash command dispatch — "Guard 2". # @@ -15531,9 +15562,16 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g (getattr(event, "metadata", None) or {}).get("gateway_session_id") or "" ).strip() if pinned_session_id: + origin_profile = str( + (getattr(event, "metadata", None) or {}).get( + "gateway_origin_profile" + ) + or "" + ).strip() resolved_entry = await self._resolve_async_delegation_session( session_entry, pinned_session_id, + origin_profile, ) if resolved_entry is None: return @@ -20250,8 +20288,7 @@ def _set_session_env(self, context: SessionContext) -> list: # (api_server) declare supports_async_delivery=False. Use getattr so # bare runners built via object.__new__ (tests) without self.adapters # don't blow up — they simply default to supported. - _adapters = getattr(self, "adapters", None) or {} - _adapter = _adapters.get(context.source.platform) + _adapter = self._adapter_for_source(context.source) _async_delivery = getattr(_adapter, "supports_async_delivery", True) return set_session_vars( platform=context.source.platform.value, @@ -20266,6 +20303,7 @@ def _set_session_env(self, context: SessionContext) -> list: session_key=context.session_key, message_id=str(context.source.message_id) if context.source.message_id else "", profile=getattr(context.source, "profile", "") or "", + source_snapshot=context.source.to_dict(), async_delivery=_async_delivery, ) @@ -20739,6 +20777,35 @@ def _build_process_event_source(self, evt: dict): """ from gateway.session import SessionSource + origin_payload = evt.get("origin_source") + if origin_payload is not None: + if not isinstance(origin_payload, dict): + logger.warning("Synthetic event has invalid immutable origin source") + return None + try: + source = SessionSource.from_dict(dict(origin_payload)) + except Exception: + logger.warning( + "Synthetic event immutable origin source is corrupt", + exc_info=True, + ) + return None + event_profile = str(evt.get("origin_profile") or "").strip() + source_profile = str(getattr(source, "profile", None) or "").strip() + if event_profile and source_profile and event_profile != source_profile: + logger.warning( + "Synthetic event origin profile mismatch: event=%r source=%r", + event_profile, + source_profile, + ) + return None + if event_profile and not source_profile: + source = dataclasses.replace(source, profile=event_profile) + anchor = str(evt.get("origin_message_id") or "").strip() + if anchor: + source = dataclasses.replace(source, message_id=anchor) + return source + session_key = str(evt.get("session_key") or "").strip() derived_platform = "" derived_chat_type = "" @@ -20822,6 +20889,15 @@ async def _inject_watch_notification( """ source = self._build_process_event_source(evt) if not source: + if "origin_source" in evt: + # New durable producers are authoritative. A corrupt or + # inconsistent snapshot must not silently downgrade to legacy + # cache/session-key guessing (or an API self-post) because + # that can deliver to a different profile or conversation. + logger.warning( + "Dropping synthetic event with unusable immutable origin" + ) + return None # API-server-originated sessions bind a RAW session key (the # X-Hermes-Session-Id value — see _bind_api_server_session), not a # structured ``agent:main:...`` key, so _build_process_event_source @@ -20865,11 +20941,7 @@ async def _inject_watch_notification( ) return None platform_name = source.platform.value if hasattr(source.platform, "value") else str(source.platform) - adapter = None - for p, a in self.adapters.items(): - if p.value == platform_name: - adapter = a - break + adapter = self._adapter_for_source(source) if not adapter: return None from gateway.wake import adapter_supports_push as _wake_push_ok @@ -20901,12 +20973,19 @@ async def _inject_watch_notification( parent_session_id = str(evt.get("parent_session_id") or "").strip() if parent_session_id: metadata["gateway_session_id"] = parent_session_id + metadata["gateway_origin_profile"] = str( + evt.get("origin_profile") or "" + ) synth_event = MessageEvent( text=synth_text, message_type=MessageType.TEXT, source=source, internal=True, - message_id=str(evt.get("message_id") or "").strip() or None, + message_id=( + str(evt.get("origin_message_id") or "").strip() + or str(evt.get("message_id") or "").strip() + or None + ), metadata=metadata, ) logger.info( @@ -20942,7 +21021,9 @@ def _completion_delivery_identity(evt: dict) -> Optional[tuple[str, str, object] return (evt_type, producer_id, started_at) return None - async def _classify_completion_target(self, parent_session_id: str) -> str: + async def _classify_completion_target( + self, parent_session_id: str, origin_profile: str = "" + ) -> str: """Classify an async-completion delivery target before adapter acceptance. Returns one of: @@ -20974,6 +21055,10 @@ async def _classify_completion_target(self, parent_session_id: str) -> str: return "retry" if parent is None: return "terminal" + if origin_profile and not self._async_origin_profile_matches( + parent, origin_profile + ): + return "terminal" if not parent.get("ended_at"): return "deliver" if parent.get("end_reason") != "compression": @@ -20993,6 +21078,10 @@ async def _classify_completion_target(self, parent_session_id: str) -> str: return "retry" if tip is None or tip.get("ended_at"): return "retry" + if origin_profile and not self._async_origin_profile_matches( + tip, origin_profile + ): + return "terminal" return "deliver" async def _deliver_completion_notification( @@ -21034,7 +21123,10 @@ async def _deliver_completion_notification( # would falsely acknowledge the durable row as delivered. # Verify the target here, before acceptance, and give drops an # honest durable disposition. - verdict = await self._classify_completion_target(parent_session_id) + verdict = await self._classify_completion_target( + parent_session_id, + str(evt.get("origin_profile") or ""), + ) if verdict == "terminal": logger.warning( "Async delegation %s targets permanently-gone session %s; " @@ -21126,15 +21218,15 @@ async def _deliver_completion_notification( logger.debug("Could not release durable completion claim", exc_info=True) def _enrich_async_delegation_routing(self, evt: dict) -> None: - """Fill platform/chat_id/thread_id/chat_type on an async-delegation event. - - Async-delegation completion events only carry ``session_key`` (the - daemon worker has no access to the per-message routing metadata the - terminal background watcher captures at spawn time). Parse the - session_key into the routing fields ``_build_process_event_source`` - expects. Best-effort: a CLI-origin event (empty session_key) is left - as-is and simply won't route on the gateway. + """Fill legacy async-delegation routes from their session key. + + New events carry an immutable ``origin_source`` snapshot captured by + the dispatching turn and do not need enrichment. This parser remains + only for rows created before that snapshot existed. Best-effort: a + CLI-origin event (empty session_key) is left as-is. """ + if evt.get("origin_source"): + return if evt.get("platform"): return # already enriched parsed = _parse_session_key(evt.get("session_key", "") or "") diff --git a/gateway/session_context.py b/gateway/session_context.py index 24d556d2cba71..4b3a9f92fe89d 100644 --- a/gateway/session_context.py +++ b/gateway/session_context.py @@ -37,7 +37,7 @@ """ from contextvars import ContextVar -from typing import Any +from typing import Any, Dict, Optional # Sentinel to distinguish "never set in this context" from "explicitly set to empty". # When a contextvar holds _UNSET, we fall back to os.environ (CLI/cron compat). @@ -93,6 +93,13 @@ def session_context_engaged() -> bool: _SESSION_MESSAGE_ID: ContextVar = ContextVar("HERMES_SESSION_MESSAGE_ID", default=_UNSET) _SESSION_PROFILE: ContextVar = ContextVar("HERMES_SESSION_PROFILE", default=_UNSET) +# Full SessionSource snapshot for detached work. This is deliberately not in +# _VAR_MAP: it is structured in-process routing state, not an environment +# variable that child processes should inherit. Callers receive a copy so a +# later turn cannot mutate the origin captured by a background producer. +_SESSION_SOURCE_SNAPSHOT: ContextVar = ContextVar( + "HERMES_SESSION_SOURCE_SNAPSHOT", default=_UNSET +) # Whether the current session's delivery channel can route an ASYNC completion # back to the agent AFTER the current turn ends (i.e. wake a fresh turn). @@ -171,6 +178,7 @@ def set_session_vars( cwd: str = "", async_delivery: bool = True, ui_session_id: str = "", + source_snapshot: Optional[Dict[str, Any]] = None, ) -> list: """Set all session context variables and return reset tokens. @@ -206,6 +214,7 @@ def set_session_vars( _SESSION_UI_SESSION_ID.set(ui_session_id), _SESSION_MESSAGE_ID.set(message_id), _SESSION_PROFILE.set(profile), + _SESSION_SOURCE_SNAPSHOT.set(dict(source_snapshot or {})), _SESSION_ASYNC_DELIVERY.set(bool(async_delivery)), ] try: @@ -242,6 +251,7 @@ def clear_session_vars(tokens: list) -> None: _SESSION_UI_SESSION_ID, _SESSION_MESSAGE_ID, _SESSION_PROFILE, + _SESSION_SOURCE_SNAPSHOT, ): var.set("") # Reset async-delivery capability to the "never set" sentinel rather than a @@ -293,6 +303,7 @@ def reset_session_vars() -> None: """ for var in _VAR_MAP.values(): var.set(_UNSET) + _SESSION_SOURCE_SNAPSHOT.set(_UNSET) # Reset the async-delivery capability to "never bound here" (_UNSET) for the # same inheritance-leak reason as the mapped vars above — see clear_session_vars, # which resets this var on the handler-exit path for the symmetric concern. @@ -382,6 +393,14 @@ def session_is_messaging_surface() -> bool: return False +def get_session_source_snapshot() -> Optional[Dict[str, Any]]: + """Return a detached copy of the current turn's SessionSource payload.""" + value = _SESSION_SOURCE_SNAPSHOT.get() + if value is _UNSET or not isinstance(value, dict) or not value: + return None + return dict(value) + + def declare_stateless_channel() -> None: """Declare that this session cannot receive an async background completion. diff --git a/tests/gateway/test_async_delegation_origin_durability.py b/tests/gateway/test_async_delegation_origin_durability.py new file mode 100644 index 0000000000000..8a9acf01065da --- /dev/null +++ b/tests/gateway/test_async_delegation_origin_durability.py @@ -0,0 +1,241 @@ +"""Behavioral coverage for immutable async-delegation origin routing.""" + +import json +import queue +from collections import OrderedDict +from datetime import datetime +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from gateway.config import Platform +from gateway.run import GatewayRunner +from gateway.session import SessionEntry, SessionSource +from gateway.session_context import clear_session_vars, set_session_vars +from tools import async_delegation as ad + + +def _feishu_source(*, chat_id="chat-origin", message_id="om-origin", profile="coder"): + return SessionSource( + platform=Platform.FEISHU, + chat_id=chat_id, + chat_type="dm", + user_id="user-origin", + message_id=message_id, + profile=profile, + ) + + +def _entry(session_id="sess-current"): + return SessionEntry( + session_key="agent:coder:feishu:dm:chat-origin", + session_id=session_id, + created_at=datetime.now(), + updated_at=datetime.now(), + platform=Platform.FEISHU, + chat_type="dm", + ) + + +def test_restart_restore_retains_origin_snapshot_and_additively_migrates(tmp_path, monkeypatch): + db_path = tmp_path / "state.db" + monkeypatch.setattr(ad, "_db_path", lambda: db_path) + + # Exact pre-change table shape: initialization must add columns in place. + import sqlite3 + + conn = sqlite3.connect(db_path) + conn.execute( + """CREATE TABLE async_delegations ( + delegation_id TEXT PRIMARY KEY, origin_session TEXT NOT NULL, + origin_ui_session_id TEXT NOT NULL DEFAULT '', parent_session_id TEXT, + state TEXT NOT NULL, dispatched_at REAL NOT NULL, completed_at REAL, + updated_at REAL NOT NULL, event_json TEXT, result_json TEXT, + delivery_state TEXT NOT NULL DEFAULT 'pending', + delivery_attempts INTEGER NOT NULL DEFAULT 0, delivered_at REAL, + owner_pid INTEGER, owner_started_at INTEGER, task_json TEXT, + delivery_claim TEXT, delivery_claimed_at REAL, + origin_session_id TEXT NOT NULL DEFAULT '' + )""" + ) + conn.commit() + conn.close() + + source = _feishu_source().to_dict() + record = { + "delegation_id": "deleg-restart", + "goal": "finish later", + "context": None, + "toolsets": None, + "role": "leaf", + "model": "m", + "session_key": "agent:coder:feishu:dm:chat-origin", + "origin_ui_session_id": "", + "origin_session_id": "", + "origin_message_id": "om-origin", + "origin_source": source, + "origin_profile": "coder", + "parent_session_id": "sess-parent", + "status": "running", + "dispatched_at": 1.0, + } + ad._persist_dispatch(record) + event = { + "type": "async_delegation", + "delegation_id": "deleg-restart", + "session_key": record["session_key"], + "parent_session_id": "sess-parent", + "origin_message_id": "om-origin", + "origin_source": source, + "origin_profile": "coder", + "status": "completed", + "completed_at": 2.0, + } + ad._persist_completion(event, {"summary": "done"}) + + restored_queue = queue.Queue() + assert ad.restore_undelivered_completions(restored_queue) == 1 + restored = restored_queue.get_nowait() + assert restored["restored"] is True + assert restored["origin_message_id"] == "om-origin" + assert restored["origin_source"] == source + assert restored["origin_profile"] == "coder" + + durable = ad.get_durable_delegation("deleg-restart") + assert durable["origin_message_id"] == "om-origin" + assert durable["origin_source"] == source + assert durable["origin_profile"] == "coder" + + +def test_capture_is_detached_from_mutable_turn_source(monkeypatch): + source = _feishu_source() + payload = source.to_dict() + tokens = set_session_vars( + platform="feishu", + chat_id=source.chat_id, + chat_type=source.chat_type, + user_id=source.user_id, + session_key="agent:coder:feishu:dm:chat-origin", + message_id=source.message_id, + profile="coder", + source_snapshot=payload, + ) + try: + captured = ad.capture_current_origin() + payload["chat_id"] = "chat-mutated" + source.chat_id = "chat-mutated" + finally: + clear_session_vars(tokens) + + assert captured["origin_message_id"] == "om-origin" + assert captured["origin_source"]["chat_id"] == "chat-origin" + assert captured["origin_profile"] == "coder" + + +def test_immutable_event_source_wins_over_mutable_store_and_cache(): + runner = object.__new__(GatewayRunner) + wrong = _feishu_source(chat_id="chat-new", message_id="om-new") + entry = MagicMock(origin=wrong) + runner.session_store = MagicMock() + runner.session_store._entries = { + "agent:coder:feishu:dm:chat-origin": entry, + } + runner._session_sources = OrderedDict( + [("agent:coder:feishu:dm:chat-origin", wrong)] + ) + + resolved = runner._build_process_event_source( + { + "type": "async_delegation", + "session_key": "agent:coder:feishu:dm:chat-origin", + "origin_source": _feishu_source().to_dict(), + "origin_message_id": "om-origin", + "origin_profile": "coder", + } + ) + + assert resolved.chat_id == "chat-origin" + assert resolved.message_id == "om-origin" + assert resolved.profile == "coder" + + +@pytest.mark.asyncio +async def test_synthetic_completion_uses_origin_profile_adapter_and_anchor(): + runner = object.__new__(GatewayRunner) + default_adapter = MagicMock(supports_async_delivery=True) + default_adapter.handle_message = AsyncMock() + coder_adapter = MagicMock(supports_async_delivery=True) + coder_adapter.handle_message = AsyncMock() + runner.adapters = {Platform.FEISHU: default_adapter} + runner._profile_adapters = {"coder": {Platform.FEISHU: coder_adapter}} + runner._active_profile_name = lambda: "default" + + delivered = await runner._inject_watch_notification( + "delegation done", + { + "type": "async_delegation", + "session_key": "agent:coder:feishu:dm:chat-origin", + "origin_source": _feishu_source().to_dict(), + "origin_message_id": "om-origin", + "origin_profile": "coder", + "parent_session_id": "sess-parent", + }, + ) + + assert delivered is True + default_adapter.handle_message.assert_not_awaited() + coder_adapter.handle_message.assert_awaited_once() + event = coder_adapter.handle_message.await_args.args[0] + assert event.message_id == "om-origin" + assert event.source.message_id == "om-origin" + assert event.metadata["gateway_origin_profile"] == "coder" + assert runner._thread_metadata_for_source( + event.source, event.message_id + ) == {"reply_in_thread": True} + + +@pytest.mark.asyncio +async def test_compression_continuation_cannot_cross_origin_profile(): + runner = object.__new__(GatewayRunner) + rows = { + "sess-parent": { + "id": "sess-parent", + "ended_at": "2026-07-31T00:00:00", + "end_reason": "compression", + "profile_name": "coder", + }, + "sess-tip": { + "id": "sess-tip", + "ended_at": None, + "profile_name": "default", + }, + } + runner._session_db = MagicMock() + runner._session_db.get_session = AsyncMock(side_effect=lambda sid: rows.get(sid)) + runner._session_db.get_compression_tip = AsyncMock(return_value="sess-tip") + runner.session_store = MagicMock() + + resolved = await runner._resolve_async_delegation_session( + _entry("sess-parent"), + "sess-parent", + "coder", + ) + + assert resolved is None + runner.session_store.switch_session.assert_not_called() + runner.session_store.advance_compression_session.assert_not_called() + + +def test_corrupt_new_origin_fails_closed_instead_of_using_cache(): + runner = object.__new__(GatewayRunner) + runner.session_store = MagicMock() + runner._session_sources = OrderedDict( + [("agent:coder:feishu:dm:chat-origin", _feishu_source())] + ) + + assert runner._build_process_event_source( + { + "session_key": "agent:coder:feishu:dm:chat-origin", + "origin_source": {"platform": "not-a-platform"}, + } + ) is None diff --git a/tests/tools/test_delegate_apiserver_background.py b/tests/tools/test_delegate_apiserver_background.py index 4c33cf9153bbb..f5c6be59d6670 100644 --- a/tests/tools/test_delegate_apiserver_background.py +++ b/tests/tools/test_delegate_apiserver_background.py @@ -40,6 +40,7 @@ def _clean_queue_and_context(monkeypatch): for var in sc._VAR_MAP.values(): var.set(sc._UNSET) + sc._SESSION_SOURCE_SNAPSHOT.set(sc._UNSET) sc._SESSION_ASYNC_DELIVERY.set(sc._UNSET) # set_current_session_id (invoked by the clobber-reproducing fake child # build) writes os.environ directly — scrub it so it can't leak into @@ -97,8 +98,18 @@ def clobbering_build_child(**kw): # HERMES_SESSION_ID ContextVar + os.environ, clobbering the spawner's # id ~milliseconds before delegate_tool dispatches the batch. from gateway.session_context import set_current_session_id + import gateway.session_context as sc set_current_session_id("20260715_child1") + sc._SESSION_SOURCE_SNAPSHOT.set({ + "platform": "api_server", + "chat_id": "child-route", + "chat_type": "dm", + "message_id": "om-child", + "profile": "other", + }) + sc._SESSION_MESSAGE_ID.set("om-child") + sc._SESSION_PROFILE.set("other") return fake_child monkeypatch.setattr(dt, "_build_child_agent", clobbering_build_child) @@ -118,6 +129,15 @@ def test_apiserver_session_with_id_dispatches_background(monkeypatch): chat_id="raw-sid-7", session_key="raw-sid-7", session_id="raw-sid-7", + message_id="om-parent", + profile="coder", + source_snapshot={ + "platform": "api_server", + "chat_id": "raw-sid-7", + "chat_type": "dm", + "message_id": "om-parent", + "profile": "coder", + }, async_delivery=False, ) @@ -138,6 +158,9 @@ def test_apiserver_session_with_id_dispatches_background(monkeypatch): # id, not the subagent-internal id the child build clobbered # HERMES_SESSION_ID with (see clobbering_build_child). assert evt["origin_session_id"] == "raw-sid-7" + assert evt["origin_message_id"] == "om-parent" + assert evt["origin_source"]["chat_id"] == "raw-sid-7" + assert evt["origin_profile"] == "coder" # --------------------------------------------------------------------------- diff --git a/tools/async_delegation.py b/tools/async_delegation.py index 702036df0c41c..eacff47722007 100644 --- a/tools/async_delegation.py +++ b/tools/async_delegation.py @@ -158,7 +158,10 @@ def _initialize_schema(conn: sqlite3.Connection) -> None: task_json TEXT, delivery_claim TEXT, delivery_claimed_at REAL, - origin_session_id TEXT NOT NULL DEFAULT '' + origin_session_id TEXT NOT NULL DEFAULT '', + origin_message_id TEXT NOT NULL DEFAULT '', + origin_source_json TEXT, + origin_profile TEXT NOT NULL DEFAULT '' )""" ) columns = {row[1] for row in conn.execute("PRAGMA table_info(async_delegations)")} @@ -173,6 +176,9 @@ def _initialize_schema(conn: sqlite3.Connection) -> None: # completions recovered after a process restart are unroutable on # api_server (the in-memory record that carried it is gone). ("origin_session_id", "TEXT"), + ("origin_message_id", "TEXT NOT NULL DEFAULT ''"), + ("origin_source_json", "TEXT"), + ("origin_profile", "TEXT NOT NULL DEFAULT ''"), ): if name not in columns: conn.execute(f"ALTER TABLE async_delegations ADD COLUMN {name} {sql_type}") @@ -215,13 +221,17 @@ def _persist_dispatch(record: Dict[str, Any]) -> None: (delegation_id, origin_session, origin_ui_session_id, parent_session_id, state, dispatched_at, updated_at, delivery_state, delivery_attempts, owner_pid, - owner_started_at, task_json, origin_session_id) - VALUES (?, ?, ?, ?, 'running', ?, ?, 'pending', 0, ?, ?, ?, ?)""", + owner_started_at, task_json, origin_session_id, + origin_message_id, origin_source_json, origin_profile) + VALUES (?, ?, ?, ?, 'running', ?, ?, 'pending', 0, ?, ?, ?, ?, ?, ?, ?)""", (record["delegation_id"], record.get("session_key", ""), record.get("origin_ui_session_id", ""), record.get("parent_session_id"), record["dispatched_at"], now, __import__("os").getpid(), owner_started_at, json.dumps(task_payload), - record.get("origin_session_id", "")), + record.get("origin_session_id", ""), + record.get("origin_message_id", ""), + json.dumps(record.get("origin_source")) if record.get("origin_source") else None, + record.get("origin_profile", "")), ) _prune_durable_records() @@ -302,12 +312,14 @@ def recover_abandoned_delegations() -> int: rows = conn.execute( """SELECT delegation_id, origin_session, origin_ui_session_id, parent_session_id, dispatched_at, owner_pid, - owner_started_at, task_json, origin_session_id + owner_started_at, task_json, origin_session_id, + origin_message_id, origin_source_json, origin_profile FROM async_delegations WHERE state IN ('running','finalizing')""" ).fetchall() for row in rows: (delegation_id, session_key, origin_ui, parent_id, dispatched_at, - pid, started, task_json, origin_session_id) = row + pid, started, task_json, origin_session_id, origin_message_id, + origin_source_json, origin_profile) = row live = False if pid: live = _pid_exists(int(pid)) @@ -316,12 +328,15 @@ def recover_abandoned_delegations() -> int: if live: continue task = json.loads(task_json or "{}") + origin_source = json.loads(origin_source_json) if origin_source_json else None event = { "type": "async_delegation", "delegation_id": delegation_id, "session_key": session_key, "origin_ui_session_id": origin_ui, # Restore the durable wake target so completions recovered # after a restart remain routable to api_server sessions. "origin_session_id": origin_session_id or "", + "origin_message_id": origin_message_id or "", + "origin_profile": origin_profile or "", "parent_session_id": parent_id, "goal": task.get("goal", ""), "goals": task.get("goals"), "context": task.get("context"), "toolsets": task.get("toolsets"), "role": task.get("role"), @@ -330,6 +345,8 @@ def recover_abandoned_delegations() -> int: "error": "Delegation owner exited before recording a terminal result; outcome unknown.", "dispatched_at": dispatched_at, "completed_at": now, } + if origin_source: + event["origin_source"] = origin_source result = {"status": "unknown", "summary": None, "error": event["error"]} conn.execute( """UPDATE async_delegations SET state='unknown', completed_at=?, @@ -499,7 +516,8 @@ def get_durable_delegation(delegation_id: str) -> Optional[Dict[str, Any]]: row = conn.execute( """SELECT origin_session, state, dispatched_at, completed_at, result_json, delivery_state, delivery_attempts, - origin_session_id + origin_session_id, origin_message_id, + origin_source_json, origin_profile FROM async_delegations WHERE delegation_id=?""", (delegation_id,), ).fetchone() if row is None: @@ -510,6 +528,9 @@ def get_durable_delegation(delegation_id: str) -> Optional[Dict[str, Any]]: "result": json.loads(row[4]) if row[4] else None, "delivery_state": row[5], "delivery_attempts": row[6], "origin_session_id": row[7] or "", + "origin_message_id": row[8] or "", + "origin_source": json.loads(row[9]) if row[9] else None, + "origin_profile": row[10] or "", } @@ -621,6 +642,52 @@ def _current_origin_session_id() -> str: return "" +def capture_current_origin() -> Dict[str, Any]: + """Capture immutable routing identity from the currently-bound turn. + + The returned payload is detached from the live ``SessionSource`` and from + the gateway's mutable source cache. Legacy/CLI callers without a bound + source return empty fields and continue through the existing fail-safe + session-key routing path. + """ + try: + from gateway.session import SessionSource + from gateway.session_context import ( + get_session_env, + get_session_source_snapshot, + ) + + source_payload = get_session_source_snapshot() + if source_payload: + # Validate and normalize the snapshot through the public wire + # contract, then copy it again for ownership by the producer. + source_payload = SessionSource.from_dict(source_payload).to_dict() + message_id = get_session_env("HERMES_SESSION_MESSAGE_ID", "") or "" + profile = get_session_env("HERMES_SESSION_PROFILE", "") or "" + if source_payload: + message_id = message_id or str(source_payload.get("message_id") or "") + profile = profile or str(source_payload.get("profile") or "") + if source_payload and not profile: + try: + from hermes_cli.profiles import get_active_profile_name + + profile = get_active_profile_name() or "default" + except Exception: + profile = "default" + return { + "origin_message_id": str(message_id), + "origin_source": dict(source_payload) if source_payload else None, + "origin_profile": str(profile), + } + except Exception: + logger.debug("Could not capture async delegation origin", exc_info=True) + return { + "origin_message_id": "", + "origin_source": None, + "origin_profile": "", + } + + def dispatch_async_delegation( *, goal: str, @@ -633,6 +700,9 @@ def dispatch_async_delegation( runner: Callable[[], Dict[str, Any]], origin_ui_session_id: str = "", origin_session_id: str = "", + origin_message_id: str = "", + origin_source: Optional[Dict[str, Any]] = None, + origin_profile: str = "", interrupt_fn: Optional[Callable[[], None]] = None, max_async_children: int = _DEFAULT_MAX_ASYNC_CHILDREN, progress_fn: Optional[Callable[[], tuple]] = None, @@ -691,6 +761,9 @@ def dispatch_async_delegation( "session_key": session_key, "origin_ui_session_id": origin_ui_session_id, "origin_session_id": origin_session_id, + "origin_message_id": origin_message_id, + "origin_source": dict(origin_source) if origin_source else None, + "origin_profile": origin_profile, "parent_session_id": parent_session_id, "status": "running", "dispatched_at": dispatched_at, @@ -838,6 +911,8 @@ def _push_completion_event( "session_key": record.get("session_key", ""), "origin_ui_session_id": record.get("origin_ui_session_id", ""), "origin_session_id": record.get("origin_session_id", ""), + "origin_message_id": record.get("origin_message_id", ""), + "origin_profile": record.get("origin_profile", ""), "parent_session_id": record.get("parent_session_id"), "goal": record.get("goal", ""), "context": record.get("context"), @@ -855,6 +930,8 @@ def _push_completion_event( "completed_at": completed_at, "exit_reason": result.get("exit_reason"), } + if record.get("origin_source"): + evt["origin_source"] = dict(record["origin_source"]) # Structured stall metadata (#51690) — additive, present only on # stall-monitor finalizations. for _k in ( @@ -888,6 +965,9 @@ def dispatch_async_delegation_batch( runner: Callable[[], Dict[str, Any]], origin_ui_session_id: str = "", origin_session_id: str = "", + origin_message_id: str = "", + origin_source: Optional[Dict[str, Any]] = None, + origin_profile: str = "", interrupt_fn: Optional[Callable[[], None]] = None, max_async_children: int = _DEFAULT_MAX_ASYNC_CHILDREN, delegation_id: Optional[str] = None, @@ -931,6 +1011,9 @@ def dispatch_async_delegation_batch( "session_key": session_key, "origin_ui_session_id": origin_ui_session_id, "origin_session_id": origin_session_id, + "origin_message_id": origin_message_id, + "origin_source": dict(origin_source) if origin_source else None, + "origin_profile": origin_profile, "parent_session_id": parent_session_id, "status": "running", "dispatched_at": dispatched_at, @@ -1043,6 +1126,8 @@ def _push_batch_completion_event( "session_key": event_record.get("session_key", ""), "origin_ui_session_id": event_record.get("origin_ui_session_id", ""), "origin_session_id": event_record.get("origin_session_id", ""), + "origin_message_id": event_record.get("origin_message_id", ""), + "origin_profile": event_record.get("origin_profile", ""), "parent_session_id": event_record.get("parent_session_id"), "goal": event_record.get("goal", ""), "goals": event_record.get("goals"), @@ -1064,6 +1149,8 @@ def _push_batch_completion_event( "dispatched_at": dispatched_at, "completed_at": completed_at, } + if event_record.get("origin_source"): + evt["origin_source"] = dict(event_record["origin_source"]) # Structured stall metadata (#51690) — additive, present only on # stall-monitor finalizations. for _k in ( diff --git a/tools/delegate_tool.py b/tools/delegate_tool.py index 30151b429c425..053117550e833 100644 --- a/tools/delegate_tool.py +++ b/tools/delegate_tool.py @@ -2926,9 +2926,28 @@ def delegate_task( # request-scoped chat_id binding (the raw X-Hermes-Session-Id on # api_server) is untouched by child construction, so read it here and # thread it through the dispatch. - from tools.async_delegation import _current_origin_session_id + from tools.async_delegation import _current_origin_session_id, capture_current_origin + from tools.approval import get_current_session_key _origin_wake_sid = _current_origin_session_id() + _origin_route = capture_current_origin() + _origin_session_key = get_current_session_key(default="") + _origin_ui_session_id = "" + try: + from gateway.session_context import get_session_env + + _origin_surface = get_session_env("HERMES_SESSION_SOURCE", "") + _origin_ui_session_id = get_session_env("HERMES_UI_SESSION_ID", "") + if _origin_surface == "tui": + _agent_session_id = str(getattr(parent_agent, "session_id", "") or "") + if _agent_session_id: + _origin_session_key = _agent_session_id + except Exception: + _origin_ui_session_id = "" + if not _origin_session_key: + _agent_session_id = str(getattr(parent_agent, "session_id", "") or "") + if _agent_session_id: + _origin_session_key = _agent_session_id # Build all child agents on the main thread (thread-safe construction). # _build_child_preserving_parent_tools saves/restores the parent's @@ -3162,7 +3181,6 @@ def _execute_and_aggregate(*, honor_parent_interrupt: bool = True) -> dict: # keep chatting, get the combined summaries back together at the end. if background: from tools.async_delegation import dispatch_async_delegation_batch - from tools.approval import get_current_session_key # Finite sessions cannot route a detached subagent result back to the # agent after their turn/process ends. This includes stateless HTTP @@ -3215,40 +3233,11 @@ def _execute_and_aggregate(*, honor_parent_interrupt: bool = True) -> dict: ) return json.dumps(_sync_result, ensure_ascii=False) - _session_key = get_current_session_key(default="") - _origin_ui_session_id = "" - try: - from gateway.session_context import get_session_env - - _source = get_session_env("HERMES_SESSION_SOURCE", "") - _origin_ui_session_id = get_session_env("HERMES_UI_SESSION_ID", "") - # In desktop/TUI, the routable session key is the durable - # AIAgent.session_id. Context compression can rotate that id during - # the same turn before the TUI-side session dict is re-anchored; - # if we capture the stale approval/session context key here, the - # async completion becomes an orphan and any desktop poller may - # consume it. Gateway chats are different: their session_key is the - # platform conversation key (agent:main:...), so keep it there. - if _source == "tui": - _agent_session_id = str(getattr(parent_agent, "session_id", "") or "") - if _agent_session_id: - _session_key = _agent_session_id - except Exception: - _origin_ui_session_id = "" - if not _session_key: - # CLI (single-process) path: the approval contextvar is only bound - # during gateway/TUI turns and HERMES_SESSION_KEY is not in the CLI - # environment, so the key resolves empty here. Since #64240 the CLI - # drains completions through a positive-ownership filter keyed on - # the durable AIAgent.session_id — an empty session_key would fail - # closed and the CLI could never claim its own completions, while - # a restored foreign event with an empty key could leak into any - # unfiltered consumer (#64484). Stamp the parent's durable session - # id instead; compression rotations are handled on the drain side - # via resolve_resume_session_id lineage resolution. - _agent_session_id = str(getattr(parent_agent, "session_id", "") or "") - if _agent_session_id: - _session_key = _agent_session_id + # All routing identity was captured before child construction. Child + # initialization mutates session-id context, so reading any origin + # field here risks binding the completion to the child or to a later + # mutable gateway cache entry. + _session_key = _origin_session_key _parent_session_id = getattr(parent_agent, "session_id", None) _child_agents = [c for (_, _, c) in children] @@ -3326,6 +3315,9 @@ def _batch_progress(): session_key=_session_key, origin_ui_session_id=_origin_ui_session_id, origin_session_id=_wake_sid, + origin_message_id=_origin_route["origin_message_id"], + origin_source=_origin_route["origin_source"], + origin_profile=_origin_route["origin_profile"], parent_session_id=_parent_session_id, runner=_batch_runner, interrupt_fn=_batch_interrupt, From 7f933307623bd5a6275f72161473b0bd7c877b6c Mon Sep 17 00:00:00 2001 From: Zhexuan Yang Date: Fri, 31 Jul 2026 08:52:51 +0800 Subject: [PATCH 2/4] fix(terminal): persist background task origin and progress Capture immutable terminal routing before dispatch, checkpoint pending completion state for recovery, and route periodic output directly through the owning profile. Periodic updates remain non-conversational status delivery; only final completion re-enters Hermes as a synthetic turn. --- gateway/run.py | 204 +++++++----- gateway/session_context.py | 28 +- .../test_background_process_notifications.py | 105 +++++- tests/gateway/test_completion_delivery.py | 59 ++++ .../tools/test_terminal_origin_durability.py | 148 +++++++++ tools/async_delegation.py | 25 +- tools/process_registry.py | 305 +++++++++++++++--- tools/terminal_tool.py | 137 ++++---- 8 files changed, 787 insertions(+), 224 deletions(-) create mode 100644 tests/tools/test_terminal_origin_durability.py diff --git a/gateway/run.py b/gateway/run.py index 01650938df658..1419bb019b022 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -3115,6 +3115,9 @@ def _drain_gateway_watch_events(completion_queue) -> "list[dict]": return watch_events +_BACKGROUND_PROCESS_STATUS_MIN_INTERVAL_SECONDS = 30.0 + + # Module-level weak reference to the active GatewayRunner instance. # Used by tools (e.g. send_message) that need to route through a live # adapter for plugin platforms. Set in GatewayRunner.__init__(). @@ -20301,6 +20304,7 @@ def _set_session_env(self, context: SessionContext) -> list: user_id=str(context.source.user_id) if context.source.user_id else "", user_name=str(context.source.user_name) if context.source.user_name else "", session_key=context.session_key, + session_id=context.session_id, message_id=str(context.source.message_id) if context.source.message_id else "", profile=getattr(context.source, "profile", "") or "", source_snapshot=context.source.to_dict(), @@ -21115,52 +21119,48 @@ async def _deliver_completion_notification( durable_delegation_id, exc, ) return False - parent_session_id = str(evt.get("parent_session_id") or "").strip() - if parent_session_id: - # Pre-flight (#65838-class): adapter acceptance is NOT proof of - # delivery — the inner #55578 resolver can still fail closed - # inside the message pipeline AFTER the adapter accepted, which - # would falsely acknowledge the durable row as delivered. - # Verify the target here, before acceptance, and give drops an - # honest durable disposition. - verdict = await self._classify_completion_target( - parent_session_id, - str(evt.get("origin_profile") or ""), + parent_session_id = str(evt.get("parent_session_id") or "").strip() + if parent_session_id: + # Adapter acceptance is not proof that the pinned physical session + # will accept the synthetic turn. Pre-flight every durable producer + # (delegation and terminal) before crossing that boundary. + verdict = await self._classify_completion_target( + parent_session_id, + str(evt.get("origin_profile") or ""), + ) + if verdict == "terminal": + logger.warning( + "%s completion targets permanently-gone session %s; " + "terminally dropping delivery.", + evt.get("type", "background"), parent_session_id, ) - if verdict == "terminal": - logger.warning( - "Async delegation %s targets permanently-gone session %s; " - "terminally dropping delivery (result remains in the " - "delegation records).", - durable_delegation_id or "", parent_session_id, - ) - if durable_claim_id: - try: - from tools.async_delegation import drop_completion_delivery + if durable_claim_id: + try: + from tools.async_delegation import drop_completion_delivery - drop_completion_delivery( - durable_delegation_id, durable_claim_id, - ) - except Exception: - logger.debug( - "Could not drop durable completion claim", - exc_info=True, - ) - return None - if verdict == "retry": - if durable_claim_id: - try: - from tools.async_delegation import release_completion_delivery + drop_completion_delivery( + durable_delegation_id, durable_claim_id, + ) + except Exception: + logger.debug( + "Could not drop durable completion claim", + exc_info=True, + ) + return None + if verdict == "retry": + if durable_claim_id: + try: + from tools.async_delegation import release_completion_delivery - release_completion_delivery( - durable_delegation_id, durable_claim_id, - ) - except Exception: - logger.debug( - "Could not release durable completion claim", - exc_info=True, - ) - return False + release_completion_delivery( + durable_delegation_id, durable_claim_id, + ) + except Exception: + logger.debug( + "Could not release durable completion claim", + exc_info=True, + ) + return False if identity is not None: with self._completion_delivery_lock: if ( @@ -21329,7 +21329,9 @@ async def _run_process_watcher(self, watcher: dict) -> None: logger.debug("Process watcher ended (silent): %s", session_id) return - last_output_len = 0 + last_output_snapshot = "" + pending_status = False + next_status_at = 0.0 while True: await asyncio.sleep(interval) @@ -21337,9 +21339,10 @@ async def _run_process_watcher(self, watcher: dict) -> None: if session is None: break - current_output_len = len(session.output_buffer) - has_new_output = current_output_len > last_output_len - last_output_len = current_output_len + current_output = session.output_buffer or "" + if current_output != last_output_snapshot: + last_output_snapshot = current_output + pending_status = True if session.exited: # --- Agent-triggered completion: inject synthetic message --- @@ -21377,6 +21380,12 @@ async def _run_process_watcher(self, watcher: dict) -> None: "user_id": user_id, "user_name": user_name, "message_id": message_id, + "origin_message_id": str( + watcher.get("origin_message_id") or message_id or "" + ), + "origin_source": watcher.get("origin_source"), + "origin_profile": watcher.get("origin_profile", ""), + "parent_session_id": watcher.get("parent_session_id", ""), "started_at": getattr(session, "started_at", None), "command": _command, "exit_code": session.exit_code, @@ -21394,6 +21403,7 @@ async def _run_process_watcher(self, watcher: dict) -> None: # The process remains terminal; retry after failed # adapter injection instead of suppressing the result. continue + _pr_check.mark_notification_delivered(session_id) break # --- Normal text-only notification --- @@ -21430,26 +21440,21 @@ async def _run_process_watcher(self, watcher: dict) -> None: f"[Background process {session_id} finished with exit code {session.exit_code}~ " f"Here's the final output:\n{new_output}]" ) - adapter = None - for p, a in self.adapters.items(): - if p.value == platform_name: - adapter = a - break - if adapter and chat_id: - try: - send_meta = {"thread_id": thread_id} if thread_id else None - await adapter.send( - chat_id, - message_text, - metadata=_non_conversational_metadata(send_meta, platform=platform_name), - ) - except Exception as e: - logger.error("Watcher delivery error: %s", e) + delivered = await self._deliver_process_status_direct( + message_text, watcher, status_key=f"process:{session_id}:final" + ) + if delivered is False: + continue + _pr_check.mark_notification_delivered(session_id) break - elif has_new_output and notify_mode == "all" and not agent_notify: - # New output available -- deliver status update (only in "all" mode) - # Skip periodic updates for agent_notify watchers (they only care about completion) + elif pending_status and notify_mode == "all": + # Periodic output sync is direct status delivery, never a + # synthetic agent turn. Bursts coalesce behind a per-process + # throttle and adapters may edit the same status bubble. + now = time.monotonic() + if now < next_status_at: + continue new_output = session.output_buffer[-500:] if session.output_buffer else "" if new_output: from agent.redact import redact_terminal_output @@ -21460,24 +21465,63 @@ async def _run_process_watcher(self, watcher: dict) -> None: f"[Background process {session_id} is still running~ " f"New output:\n{new_output}]" ) - adapter = None - for p, a in self.adapters.items(): - if p.value == platform_name: - adapter = a - break - if adapter and chat_id: - try: - send_meta = {"thread_id": thread_id} if thread_id else None - await adapter.send( - chat_id, - message_text, - metadata=_non_conversational_metadata(send_meta, platform=platform_name), - ) - except Exception as e: - logger.error("Watcher delivery error: %s", e) + delivered = await self._deliver_process_status_direct( + message_text, watcher, status_key=f"process:{session_id}:running" + ) + if delivered is True: + pending_status = False + next_status_at = ( + now + _BACKGROUND_PROCESS_STATUS_MIN_INTERVAL_SECONDS + ) logger.debug("Process watcher ended: %s", session_id) + async def _deliver_process_status_direct( + self, + message_text: str, + watcher: dict, + *, + status_key: str, + ) -> Optional[bool]: + """Send terminal progress/final text without starting an agent turn.""" + evt = dict(watcher) + if watcher.get("origin_message_id") and not evt.get("message_id"): + evt["message_id"] = watcher["origin_message_id"] + source = self._build_process_event_source(evt) + if source is None: + return None + adapter = self._adapter_for_source(source) + if adapter is None: + return None + anchor = str( + watcher.get("origin_message_id") + or watcher.get("message_id") + or getattr(source, "message_id", "") + or "" + ) or None + metadata = self._thread_metadata_for_source(source, anchor) + if metadata and metadata.get("reply_in_thread") and anchor: + metadata = dict(metadata) + metadata["reply_to_message_id"] = anchor + metadata = _non_conversational_metadata( + metadata, + platform=getattr(source.platform, "value", source.platform), + ) + try: + result = await _send_or_update_status_coro( + adapter, + source.chat_id, + status_key, + message_text, + metadata, + ) + if result is not None and getattr(result, "success", True) is False: + return False + return True + except Exception as exc: + logger.error("Watcher delivery error: %s", exc) + return False + _MAX_INTERRUPT_DEPTH = 3 # Cap recursive interrupt handling (#816) # Config keys whose values MUST invalidate the gateway's cached agent diff --git a/gateway/session_context.py b/gateway/session_context.py index 4b3a9f92fe89d..950a7018871b9 100644 --- a/gateway/session_context.py +++ b/gateway/session_context.py @@ -37,6 +37,7 @@ """ from contextvars import ContextVar +from copy import deepcopy from typing import Any, Dict, Optional # Sentinel to distinguish "never set in this context" from "explicitly set to empty". @@ -398,7 +399,32 @@ def get_session_source_snapshot() -> Optional[Dict[str, Any]]: value = _SESSION_SOURCE_SNAPSHOT.get() if value is _UNSET or not isinstance(value, dict) or not value: return None - return dict(value) + return deepcopy(value) + + +def capture_session_origin() -> Dict[str, Any]: + """Capture immutable routing identity for a detached producer.""" + source_payload = get_session_source_snapshot() + message_id = get_session_env("HERMES_SESSION_MESSAGE_ID", "") or "" + profile = get_session_env("HERMES_SESSION_PROFILE", "") or "" + if source_payload: + message_id = message_id or str(source_payload.get("message_id") or "") + profile = profile or str(source_payload.get("profile") or "") + if source_payload and not profile: + try: + from hermes_cli.profiles import get_active_profile_name + + profile = get_active_profile_name() or "default" + except Exception: + profile = "default" + return { + "origin_message_id": str(message_id), + "origin_source": deepcopy(source_payload) if source_payload else None, + "origin_profile": str(profile), + "parent_session_id": str( + get_session_env("HERMES_SESSION_ID", "") or "" + ), + } def declare_stateless_channel() -> None: diff --git a/tests/gateway/test_background_process_notifications.py b/tests/gateway/test_background_process_notifications.py index f55c92deddb31..8aa8cc98c35ff 100644 --- a/tests/gateway/test_background_process_notifications.py +++ b/tests/gateway/test_background_process_notifications.py @@ -9,12 +9,13 @@ import asyncio from types import SimpleNamespace -from unittest.mock import AsyncMock +from unittest.mock import AsyncMock, MagicMock import pytest from gateway.config import GatewayConfig, Platform from gateway.run import GatewayRunner, _parse_session_key +from gateway.session import SessionSource # --------------------------------------------------------------------------- @@ -36,6 +37,9 @@ def get(self, session_id): def is_completion_consumed(self, session_id): return self._consumed + def mark_notification_delivered(self, session_id): + self.delivered_session_id = session_id + def _build_runner(monkeypatch, tmp_path, mode: str) -> GatewayRunner: """Create a GatewayRunner with a fake config for the given mode.""" @@ -157,6 +161,105 @@ async def _instant_sleep(*_a, **_kw): adapter.send.assert_not_awaited() +@pytest.mark.asyncio +async def test_periodic_output_is_throttled_profile_safe_and_final_only_synthetic( + monkeypatch, tmp_path, +): + """Running output is direct/coalesced; only completion re-enters Hermes.""" + import gateway.run as gateway_run + import tools.process_registry as pr_module + + sessions = [ + SimpleNamespace( + output_buffer="phase 1\n", exited=False, exit_code=None, + command="codex exec task", started_at=1.0, + ), + SimpleNamespace( + output_buffer="phase 1\nphase 2\n", exited=False, exit_code=None, + command="codex exec task", started_at=1.0, + ), + SimpleNamespace( + output_buffer="phase 1\nphase 2\nphase latest\n", exited=False, + exit_code=None, command="codex exec task", started_at=1.0, + ), + SimpleNamespace( + output_buffer="done\n", exited=True, exit_code=0, + command="codex exec task", started_at=1.0, + completion_reason="exited", termination_source="", + ), + ] + registry = _FakeRegistry(sessions) + monkeypatch.setattr(pr_module, "process_registry", registry) + + async def _instant_sleep(*_a, **_kw): + pass + + monkeypatch.setattr(asyncio, "sleep", _instant_sleep) + watcher_clock = MagicMock(side_effect=[100.0, 110.0, 130.0]) + + (tmp_path / "config.yaml").write_text( + "display:\n background_process_notifications: all\n", + encoding="utf-8", + ) + monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path) + runner = GatewayRunner(GatewayConfig()) + monkeypatch.setattr(gateway_run, "time", SimpleNamespace(monotonic=watcher_clock)) + + default_adapter = SimpleNamespace( + supports_async_delivery=True, + send=AsyncMock(), + send_or_update_status=AsyncMock(return_value=SimpleNamespace(success=True)), + handle_message=AsyncMock(), + ) + coder_adapter = SimpleNamespace( + supports_async_delivery=True, + send=AsyncMock(), + send_or_update_status=AsyncMock(return_value=SimpleNamespace(success=True)), + handle_message=AsyncMock(), + ) + runner.adapters = {Platform.FEISHU: default_adapter} + runner._profile_adapters = {"coder": {Platform.FEISHU: coder_adapter}} + runner._active_profile_name = lambda: "default" + + source = SessionSource( + platform=Platform.FEISHU, + chat_id="chat-origin", + chat_type="dm", + user_id="user-origin", + message_id="om-origin", + profile="coder", + ) + await runner._run_process_watcher({ + "session_id": "proc-codex", + "check_interval": 0, + "session_key": "agent:coder:feishu:dm:chat-origin", + "notify_on_complete": True, + "origin_source": source.to_dict(), + "origin_message_id": "om-origin", + "origin_profile": "coder", + }) + + assert coder_adapter.send_or_update_status.await_count == 2 + first, second = coder_adapter.send_or_update_status.await_args_list + assert first.args[1] == second.args[1] == "process:proc-codex:running" + assert "phase 1" in first.args[2] + assert "phase latest" in second.args[2] + assert first.kwargs["metadata"]["reply_in_thread"] is True + assert first.kwargs["metadata"]["reply_to_message_id"] == "om-origin" + + coder_adapter.handle_message.assert_awaited_once() + final_event = coder_adapter.handle_message.await_args.args[0] + assert final_event.internal is True + assert final_event.message_id == "om-origin" + assert final_event.source.profile == "coder" + assert "done" in final_event.text + assert registry.delivered_session_id == "proc-codex" + + default_adapter.send.assert_not_awaited() + default_adapter.send_or_update_status.assert_not_awaited() + default_adapter.handle_message.assert_not_awaited() + + @pytest.mark.asyncio async def test_inject_watch_notification_routes_from_session_store_origin(monkeypatch, tmp_path): from gateway.session import SessionSource diff --git a/tests/gateway/test_completion_delivery.py b/tests/gateway/test_completion_delivery.py index 3c3e09967f415..626e62b014a8f 100644 --- a/tests/gateway/test_completion_delivery.py +++ b/tests/gateway/test_completion_delivery.py @@ -84,6 +84,65 @@ def _completion_event(*, started_at, session_id="proc_reused"): } +@pytest.mark.parametrize( + ("parent", "tip_id", "tip", "expected"), + [ + ( + {"id": "parent", "ended_at": "now", "end_reason": "new", "profile_name": "coder"}, + None, + None, + None, + ), + ( + {"id": "parent", "ended_at": "now", "end_reason": "compression", "profile_name": "coder"}, + None, + None, + False, + ), + ( + {"id": "parent", "ended_at": "now", "end_reason": "compression", "profile_name": "coder"}, + "tip", + {"id": "tip", "ended_at": None, "profile_name": "default"}, + None, + ), + ( + {"id": "parent", "ended_at": "now", "end_reason": "compression", "profile_name": "coder"}, + "tip", + {"id": "tip", "ended_at": None, "profile_name": "coder"}, + True, + ), + ], + ids=["new-boundary", "compression-tip-pending", "cross-profile-tip", "same-profile-tip"], +) +def test_terminal_completion_preflights_compression_lineage( + parent, tip_id, tip, expected, +): + """Terminal completions retry/drop safely until a same-profile tip is live.""" + adapter = SimpleNamespace(handle_message=AsyncMock()) + runner = _runner(adapter) + rows = {"parent": parent} + if tip is not None: + rows["tip"] = tip + runner._session_db = SimpleNamespace( + get_session=AsyncMock(side_effect=lambda session_id: rows.get(session_id)), + get_compression_tip=AsyncMock(return_value=tip_id), + ) + runner._inject_watch_notification = AsyncMock(return_value=True) + event = _completion_event(started_at=123.0, session_id="proc-lineage") + event.update({ + "parent_session_id": "parent", + "origin_profile": "coder", + }) + + result = asyncio.run(runner._deliver_completion_notification("done", event)) + + assert result is expected + if expected is True: + runner._inject_watch_notification.assert_awaited_once_with("done", event) + else: + runner._inject_watch_notification.assert_not_awaited() + + def _stop_after_sleeps(monkeypatch, runner, count): sleep_calls = 0 diff --git a/tests/tools/test_terminal_origin_durability.py b/tests/tools/test_terminal_origin_durability.py new file mode 100644 index 0000000000000..bc1cb7b8b95cd --- /dev/null +++ b/tests/tools/test_terminal_origin_durability.py @@ -0,0 +1,148 @@ +"""Durable immutable origins for terminal(background=True).""" + +import json +import os +from unittest.mock import MagicMock, patch + +from tools.process_registry import ProcessRegistry, ProcessSession + + +def _watcher_config(): + return { + "platform": "feishu", + "chat_id": "oc_chat", + "user_id": "ou_user", + "user_name": "User", + "thread_id": "", + "origin_message_id": "om_origin", + "origin_source": { + "platform": "feishu", + "chat_id": "oc_chat", + "chat_type": "dm", + "user_id": "ou_user", + "message_id": "om_origin", + "profile": "coder", + }, + "origin_profile": "coder", + "parent_session_id": "sess_parent", + "check_interval": 5, + "notify_on_complete": True, + "watch_patterns": [], + } + + +def test_first_spawn_checkpoint_contains_immutable_origin(tmp_path): + registry = ProcessRegistry() + checkpoint = tmp_path / "processes.json" + fake_proc = MagicMock(pid=4242, stdout=MagicMock()) + fake_thread = MagicMock() + config = _watcher_config() + + with patch("tools.process_registry.CHECKPOINT_PATH", checkpoint), \ + patch("tools.process_registry.subprocess.Popen", return_value=fake_proc), \ + patch("tools.process_registry.threading.Thread", return_value=fake_thread), \ + patch.object(registry, "_safe_host_start_time", return_value=99): + session = registry.spawn_local( + "codex exec --full-auto task", + cwd=str(tmp_path), + watcher_config=config, + ) + + config["origin_source"]["chat_id"] = "mutated" + persisted = json.loads(checkpoint.read_text())[0] + assert persisted["notify_on_complete"] is True + assert persisted["watcher_message_id"] == "om_origin" + assert persisted["watcher_origin_source"]["chat_id"] == "oc_chat" + assert persisted["watcher_profile"] == "coder" + assert persisted["watcher_parent_session_id"] == "sess_parent" + assert session.watcher_origin_source["chat_id"] == "oc_chat" + assert session.output_log_path.startswith(str(tmp_path.parent)) + + +def test_finished_notification_survives_restart_until_ack(tmp_path): + checkpoint = tmp_path / "processes.json" + output_log = tmp_path / "proc.log" + output_log.write_text("final output\n") + registry = ProcessRegistry() + session = ProcessSession( + id="proc_final", + command="codex exec task", + pid=os.getpid(), + started_at=1.0, + output_buffer="final output\n", + output_log_path=str(output_log), + notify_on_complete=True, + ) + registry._apply_watcher_config(session, _watcher_config()) + registry._running[session.id] = session + session.exited = True + session.exit_code = 0 + + with patch("tools.process_registry.CHECKPOINT_PATH", checkpoint): + registry._move_to_finished(session) + assert json.loads(checkpoint.read_text())[0]["exited"] is True + + restarted = ProcessRegistry() + assert restarted.recover_from_checkpoint() == 1 + recovered = restarted.get("proc_final") + assert recovered.exited is True + assert recovered.output_buffer == "final output\n" + assert restarted.pending_watchers[0]["origin_message_id"] == "om_origin" + + restarted.mark_notification_delivered("proc_final") + assert json.loads(checkpoint.read_text()) == [] + assert not output_log.exists() + + +def test_dead_pid_recovers_one_lost_final_with_origin(tmp_path): + checkpoint = tmp_path / "processes.json" + entry = { + "session_id": "proc_dead", + "command": "codex exec task", + "pid": 987654321, + "pid_scope": "host", + "host_start_time": 1, + "started_at": 1.0, + "notify_on_complete": True, + "watcher_interval": 5, + "watcher_message_id": "om_origin", + "watcher_origin_source": _watcher_config()["origin_source"], + "watcher_profile": "coder", + "watcher_parent_session_id": "sess_parent", + } + checkpoint.write_text(json.dumps([entry])) + registry = ProcessRegistry() + + with patch("tools.process_registry.CHECKPOINT_PATH", checkpoint), \ + patch.object(registry, "_host_pid_is_ours", return_value=False), \ + patch.object(registry, "_is_host_pid_alive", return_value=False): + assert registry.recover_from_checkpoint() == 1 + recovered = registry.get("proc_dead") + assert recovered.exited is True + assert recovered.completion_reason == "lost" + assert recovered.termination_source == "restart_recovery" + assert registry.pending_watchers[0]["origin_profile"] == "coder" + assert json.loads(checkpoint.read_text())[0]["exited"] is True + + +def test_completion_event_carries_terminal_origin(tmp_path): + registry = ProcessRegistry() + session = ProcessSession( + id="proc_event", + command="codex exec task", + started_at=1.0, + notify_on_complete=True, + output_buffer="done", + ) + registry._apply_watcher_config(session, _watcher_config()) + registry._running[session.id] = session + session.exited = True + session.exit_code = 0 + + with patch.object(registry, "_write_checkpoint"): + registry._move_to_finished(session) + event = registry.completion_queue.get_nowait() + assert event["origin_message_id"] == "om_origin" + assert event["origin_source"]["chat_id"] == "oc_chat" + assert event["origin_profile"] == "coder" + assert event["parent_session_id"] == "sess_parent" diff --git a/tools/async_delegation.py b/tools/async_delegation.py index eacff47722007..05800fedbeea4 100644 --- a/tools/async_delegation.py +++ b/tools/async_delegation.py @@ -651,33 +651,22 @@ def capture_current_origin() -> Dict[str, Any]: session-key routing path. """ try: - from gateway.session import SessionSource from gateway.session_context import ( - get_session_env, - get_session_source_snapshot, + capture_session_origin, ) - source_payload = get_session_source_snapshot() + captured = capture_session_origin() + source_payload = captured.get("origin_source") if source_payload: # Validate and normalize the snapshot through the public wire # contract, then copy it again for ownership by the producer. - source_payload = SessionSource.from_dict(source_payload).to_dict() - message_id = get_session_env("HERMES_SESSION_MESSAGE_ID", "") or "" - profile = get_session_env("HERMES_SESSION_PROFILE", "") or "" - if source_payload: - message_id = message_id or str(source_payload.get("message_id") or "") - profile = profile or str(source_payload.get("profile") or "") - if source_payload and not profile: - try: - from hermes_cli.profiles import get_active_profile_name + from gateway.session import SessionSource - profile = get_active_profile_name() or "default" - except Exception: - profile = "default" + source_payload = SessionSource.from_dict(source_payload).to_dict() return { - "origin_message_id": str(message_id), + "origin_message_id": captured["origin_message_id"], "origin_source": dict(source_payload) if source_payload else None, - "origin_profile": str(profile), + "origin_profile": captured["origin_profile"], } except Exception: logger.debug("Could not capture async delegation origin", exc_info=True) diff --git a/tools/process_registry.py b/tools/process_registry.py index 7daf74b2e3365..fed9584f70627 100644 --- a/tools/process_registry.py +++ b/tools/process_registry.py @@ -44,6 +44,8 @@ from tools.environments.local import _find_shell, _resolve_safe_cwd, _sanitize_subprocess_env from hermes_cli._subprocess_compat import windows_hide_flags from dataclasses import dataclass, field +from copy import deepcopy +from pathlib import Path from typing import Any, Dict, List, Optional from hermes_cli.config import get_hermes_home @@ -105,6 +107,7 @@ class ProcessSession: completion_reason: str = "exited" # exited|killed|lost|failed_start|already_exited termination_source: str = "" # process.kill|kill_all|backend_lost|failed_start output_buffer: str = "" # Rolling output (last MAX_OUTPUT_CHARS) + output_log_path: str = "" # Durable append-only output log max_output_chars: int = MAX_OUTPUT_CHARS detached: bool = False # True if recovered from crash (no pipe) pid_scope: str = "host" # "host" for local/PTY PIDs, "sandbox" for env-local PIDs @@ -115,8 +118,12 @@ class ProcessSession: watcher_user_name: str = "" watcher_thread_id: str = "" watcher_message_id: str = "" # Triggering message id — reply anchor for topic routing + watcher_origin_source: Optional[Dict[str, Any]] = None + watcher_profile: str = "" + watcher_parent_session_id: str = "" watcher_interval: int = 0 # 0 = no watcher configured notify_on_complete: bool = False # Queue agent notification on exit + notification_delivered: bool = False # Durable final accepted/consumed # Watch patterns — trigger agent notification when output matches any pattern watch_patterns: List[str] = field(default_factory=list) _watch_hits: int = field(default=0, repr=False) # total matches delivered @@ -224,6 +231,14 @@ def _clean_shell_noise(text: str) -> str: def _emit_output(self, session: ProcessSession, chunk: str) -> None: """Forward a freshly-read chunk to the live-output sink, if one is set. Called from reader threads; never raise into the read loop.""" + if chunk and session.output_log_path: + try: + path = Path(session.output_log_path) + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("a", encoding="utf-8", errors="replace") as fh: + fh.write(chunk) + except Exception: + logger.debug("Could not persist output for %s", session.id, exc_info=True) sink = self.on_output if sink is None or not chunk: return @@ -313,7 +328,7 @@ def _check_watch_patterns(self, session: ProcessSession, new_text: str) -> None: if should_disable: # Emit exactly one "watch disabled, falling back to notify_on_complete" # summary event so the agent/user sees why things went quiet. - self.completion_queue.put({ + event = { "session_id": session.id, "session_key": session.session_key, "command": session.command, @@ -325,6 +340,9 @@ def _check_watch_patterns(self, session: ProcessSession, new_text: str) -> None: "user_name": session.watcher_user_name, "thread_id": session.watcher_thread_id, "message_id": session.watcher_message_id, + "origin_message_id": session.watcher_message_id, + "origin_profile": session.watcher_profile, + "parent_session_id": session.watcher_parent_session_id, "message": ( f"Watch patterns disabled for process {session.id} — " f"{WATCH_STRIKE_LIMIT} consecutive rate-limit windows triggered " @@ -332,7 +350,10 @@ def _check_watch_patterns(self, session: ProcessSession, new_text: str) -> None: f"Falling back to notify_on_complete semantics; you'll get " f"exactly one notification when the process exits." ), - }) + } + if session.watcher_origin_source: + event["origin_source"] = dict(session.watcher_origin_source) + self.completion_queue.put(event) return # Trim matched output to a reasonable size @@ -344,7 +365,7 @@ def _check_watch_patterns(self, session: ProcessSession, new_text: str) -> None: if not self._global_watch_admit(now): return - self.completion_queue.put({ + event = { "session_id": session.id, "session_key": session.session_key, "command": session.command, @@ -358,7 +379,13 @@ def _check_watch_patterns(self, session: ProcessSession, new_text: str) -> None: "user_name": session.watcher_user_name, "thread_id": session.watcher_thread_id, "message_id": session.watcher_message_id, - }) + "origin_message_id": session.watcher_message_id, + "origin_profile": session.watcher_profile, + "parent_session_id": session.watcher_parent_session_id, + } + if session.watcher_origin_source: + event["origin_source"] = dict(session.watcher_origin_source) + self.completion_queue.put(event) def _global_watch_admit(self, now: float) -> bool: """Return True if this watch_match event is allowed through the global breaker. @@ -491,7 +518,18 @@ def _host_pid_is_ours(cls, pid: Optional[int], expected_start: Optional[int]) -> def _refresh_detached_session(self, session: Optional[ProcessSession]) -> Optional[ProcessSession]: """Update recovered host-PID sessions when the underlying process has exited.""" - if session is None or session.exited or not session.detached or session.pid_scope != "host": + if session is None: + return session + if session.detached and session.output_log_path: + try: + text = Path(session.output_log_path).read_text( + encoding="utf-8", errors="replace" + ) + with session._lock: + session.output_buffer = text[-session.max_output_chars:] + except Exception: + pass + if session.exited or not session.detached or session.pid_scope != "host": return session # Identity-aware liveness: a recycled PID (alive but a different process @@ -686,6 +724,32 @@ def _env_temp_dir(env: Any) -> str: logger.debug("Could not resolve environment temp dir: %s", exc) return "/tmp" + @staticmethod + def _apply_watcher_config( + session: ProcessSession, + config: Optional[Dict[str, Any]], + ) -> None: + """Attach a detached producer's immutable notification origin.""" + if not config: + return + session.watcher_platform = str(config.get("platform") or "") + session.watcher_chat_id = str(config.get("chat_id") or "") + session.watcher_user_id = str(config.get("user_id") or "") + session.watcher_user_name = str(config.get("user_name") or "") + session.watcher_thread_id = str(config.get("thread_id") or "") + session.watcher_message_id = str( + config.get("origin_message_id") or config.get("message_id") or "" + ) + source = config.get("origin_source") + session.watcher_origin_source = deepcopy(source) if isinstance(source, dict) else None + session.watcher_profile = str(config.get("origin_profile") or "") + session.watcher_parent_session_id = str( + config.get("parent_session_id") or "" + ) + session.watcher_interval = int(config.get("check_interval") or 0) + session.notify_on_complete = bool(config.get("notify_on_complete", False)) + session.watch_patterns = list(config.get("watch_patterns") or []) + def spawn_local( self, command: str, @@ -694,6 +758,7 @@ def spawn_local( session_key: str = "", env_vars: dict = None, use_pty: bool = False, + watcher_config: Optional[Dict[str, Any]] = None, ) -> ProcessSession: """ Spawn a background process locally. @@ -722,6 +787,10 @@ def spawn_local( cwd=_resolve_safe_cwd(cwd or os.getcwd()), started_at=time.time(), ) + session.output_log_path = str( + get_hermes_home() / "cache" / "processes" / f"{session.id}.log" + ) + self._apply_watcher_config(session, watcher_config) if use_pty: # Try PTY mode for interactive CLI tools @@ -842,6 +911,7 @@ def spawn_via_env( task_id: str = "", session_key: str = "", timeout: int = 10, + watcher_config: Optional[Dict[str, Any]] = None, ) -> ProcessSession: """ Spawn a background process through a non-local environment backend. @@ -864,6 +934,10 @@ def spawn_via_env( env_ref=env, pid_scope="sandbox", ) + session.output_log_path = str( + get_hermes_home() / "cache" / "processes" / f"{session.id}.log" + ) + self._apply_watcher_config(session, watcher_config) # Run the command in the sandbox with output capture temp_dir = self._env_temp_dir(env) @@ -1158,7 +1232,7 @@ def _move_to_finished(self, session: ProcessSession): if was_running and session.notify_on_complete: from tools.ansi_strip import strip_ansi output_tail = strip_ansi(session.output_buffer[-2000:]) if session.output_buffer else "" - self.completion_queue.put({ + event = { "type": "completion", "session_id": session.id, "session_key": session.session_key, @@ -1171,7 +1245,19 @@ def _move_to_finished(self, session: ProcessSession): # a consumer-observed completion timestamp, this does not vary # based on which watcher notices exit first. "started_at": session.started_at, - }) + "platform": session.watcher_platform, + "chat_id": session.watcher_chat_id, + "user_id": session.watcher_user_id, + "user_name": session.watcher_user_name, + "thread_id": session.watcher_thread_id, + "message_id": session.watcher_message_id, + "origin_message_id": session.watcher_message_id, + "origin_profile": session.watcher_profile, + "parent_session_id": session.watcher_parent_session_id, + } + if session.watcher_origin_source: + event["origin_source"] = deepcopy(session.watcher_origin_source) + self.completion_queue.put(event) # ----- Query Methods ----- @@ -1179,6 +1265,22 @@ def is_completion_consumed(self, session_id: str) -> bool: """Check if a completion notification was already consumed via wait/log.""" return session_id in self._completion_consumed + def mark_notification_delivered(self, session_id: str) -> None: + """Acknowledge a terminal notification after adapter/consumer acceptance.""" + output_log_path = "" + with self._lock: + session = self._running.get(session_id) or self._finished.get(session_id) + if session is not None: + session.notification_delivered = True + output_log_path = session.output_log_path + session.output_log_path = "" + self._delete_output_log(output_log_path) + self._write_checkpoint() + + def _mark_completion_consumed(self, session_id: str) -> None: + self._completion_consumed.add(session_id) + self.mark_notification_delivered(session_id) + def is_session_waiting(self, session_id: str) -> bool: """Whether a goal loop parked on this session should still be parked. @@ -1315,6 +1417,8 @@ def drain_notifications( text = format_process_notification(evt) if text: results.append((evt, text)) + if evt.get("type") == "completion" and _evt_sid: + self.mark_notification_delivered(_evt_sid) for evt in requeue: self.completion_queue.put(evt) return results @@ -1476,7 +1580,7 @@ def read_log(self, session_id: str, offset: int = 0, limit: int = 200) -> dict: "showing": f"{len(selected)} lines", } if session.exited and observed_completion_output: - self._completion_consumed.add(session_id) + self._mark_completion_consumed(session_id) return result def wait(self, session_id: str, timeout: int = None) -> dict: @@ -1526,7 +1630,7 @@ def wait(self, session_id: str, timeout: int = None) -> dict: # child has already exited (issue #17327). self._reconcile_local_exit(session) if session.exited: - self._completion_consumed.add(session_id) + self._mark_completion_consumed(session_id) result = { "status": "exited", "command": session.command, @@ -1599,7 +1703,7 @@ def kill_process( # Only suppress the autonomous turn after its output is present in # the explicit kill result, matching wait/log consumption. if consume_output: - self._completion_consumed.add(session_id) + self._mark_completion_consumed(session_id) return result # Kill via PTY, Popen (local), or env execute (non-local) @@ -1629,7 +1733,7 @@ def kill_process( session.exit_code = None output = strip_ansi(session.output_buffer[-2000:]) if consume_output: - self._completion_consumed.add(session_id) + self._mark_completion_consumed(session_id) self._move_to_finished(session) return { "status": "already_exited", @@ -1651,7 +1755,7 @@ def kill_process( with session._lock: output = strip_ansi(session.output_buffer[-2000:]) if consume_output: - self._completion_consumed.add(session_id) + self._mark_completion_consumed(session_id) session.exited = True session.exit_code = -15 # SIGTERM session.completion_reason = "killed" @@ -1920,7 +2024,8 @@ def _prune_if_needed(self): if (now - s.started_at) > FINISHED_TTL_SECONDS ] for sid in expired: - del self._finished[sid] + removed = self._finished.pop(sid) + self._delete_output_log(removed.output_log_path) self._completion_consumed.discard(sid) self._poll_observed.discard(sid) @@ -1928,7 +2033,8 @@ def _prune_if_needed(self): total = len(self._running) + len(self._finished) if total >= MAX_PROCESSES and self._finished: oldest_id = min(self._finished, key=lambda sid: self._finished[sid].started_at) - del self._finished[oldest_id] + removed = self._finished.pop(oldest_id) + self._delete_output_log(removed.output_log_path) self._completion_consumed.discard(oldest_id) self._poll_observed.discard(oldest_id) @@ -1944,10 +2050,51 @@ def _prune_if_needed(self): if stale_polls: self._poll_observed -= stale_polls + @staticmethod + def _delete_output_log(path_value: str) -> None: + if not path_value: + return + try: + Path(path_value).unlink(missing_ok=True) + except Exception: + logger.debug("Could not remove process output log %s", path_value, exc_info=True) + # ----- Checkpoint (crash recovery) ----- + @staticmethod + def _checkpoint_entry(s: ProcessSession) -> Dict[str, Any]: + return { + "session_id": s.id, + "command": s.command, + "pid": s.pid, + "pid_scope": s.pid_scope, + "host_start_time": s.host_start_time, + "cwd": s.cwd, + "started_at": s.started_at, + "task_id": s.task_id, + "session_key": s.session_key, + "exited": s.exited, + "exit_code": s.exit_code, + "completion_reason": s.completion_reason, + "termination_source": s.termination_source, + "output_tail": s.output_buffer[-2000:] if s.exited else "", + "output_log_path": s.output_log_path, + "watcher_platform": s.watcher_platform, + "watcher_chat_id": s.watcher_chat_id, + "watcher_user_id": s.watcher_user_id, + "watcher_user_name": s.watcher_user_name, + "watcher_thread_id": s.watcher_thread_id, + "watcher_message_id": s.watcher_message_id, + "watcher_origin_source": s.watcher_origin_source, + "watcher_profile": s.watcher_profile, + "watcher_parent_session_id": s.watcher_parent_session_id, + "watcher_interval": s.watcher_interval, + "notify_on_complete": s.notify_on_complete, + "watch_patterns": s.watch_patterns, + } + def _write_checkpoint(self): - """Write running process metadata to checkpoint file atomically.""" + """Write running and undelivered terminal producer state atomically.""" try: with self._lock: entries = [] @@ -1958,26 +2105,14 @@ def _write_checkpoint(self): # for sessions spawned before this field existed. if s.host_start_time is None and s.pid_scope == "host" and s.pid: s.host_start_time = self._safe_host_start_time(s.pid) - entries.append({ - "session_id": s.id, - "command": s.command, - "pid": s.pid, - "pid_scope": s.pid_scope, - "host_start_time": s.host_start_time, - "cwd": s.cwd, - "started_at": s.started_at, - "task_id": s.task_id, - "session_key": s.session_key, - "watcher_platform": s.watcher_platform, - "watcher_chat_id": s.watcher_chat_id, - "watcher_user_id": s.watcher_user_id, - "watcher_user_name": s.watcher_user_name, - "watcher_thread_id": s.watcher_thread_id, - "watcher_message_id": s.watcher_message_id, - "watcher_interval": s.watcher_interval, - "notify_on_complete": s.notify_on_complete, - "watch_patterns": s.watch_patterns, - }) + entries.append(self._checkpoint_entry(s)) + for s in self._finished.values(): + if ( + s.notify_on_complete + and not s.notification_delivered + and s.id not in self._completion_consumed + ): + entries.append(self._checkpoint_entry(s)) # Atomic write to avoid corruption on crash from utils import atomic_json_write @@ -2001,6 +2136,37 @@ def recover_from_checkpoint(self) -> int: recovered = 0 for entry in entries: + if entry.get("exited"): + session = ProcessSession( + id=entry["session_id"], command=entry.get("command", "unknown"), + task_id=entry.get("task_id", ""), session_key=entry.get("session_key", ""), + pid=entry.get("pid"), host_start_time=entry.get("host_start_time"), + pid_scope=entry.get("pid_scope", "host"), cwd=entry.get("cwd"), + started_at=entry.get("started_at", time.time()), detached=True, + exited=True, exit_code=entry.get("exit_code"), + completion_reason=entry.get("completion_reason", "exited"), + termination_source=entry.get("termination_source", ""), + output_buffer=entry.get("output_tail", ""), + output_log_path=entry.get("output_log_path", ""), + watcher_platform=entry.get("watcher_platform", ""), + watcher_chat_id=entry.get("watcher_chat_id", ""), + watcher_user_id=entry.get("watcher_user_id", ""), + watcher_user_name=entry.get("watcher_user_name", ""), + watcher_thread_id=entry.get("watcher_thread_id", ""), + watcher_message_id=entry.get("watcher_message_id", ""), + watcher_origin_source=entry.get("watcher_origin_source"), + watcher_profile=entry.get("watcher_profile", ""), + watcher_parent_session_id=entry.get("watcher_parent_session_id", ""), + watcher_interval=entry.get("watcher_interval", 0), + notify_on_complete=entry.get("notify_on_complete", False), + watch_patterns=entry.get("watch_patterns", []), + ) + with self._lock: + self._finished[session.id] = session + if session.watcher_interval > 0: + self.pending_watchers.append(self._watcher_payload(session)) + recovered += 1 + continue pid = entry.get("pid") if not pid: continue @@ -2033,6 +2199,38 @@ def recover_from_checkpoint(self) -> int: "an unrelated process; refusing to adopt it.", entry.get("session_id", "?"), pid, ) + # The original process is gone or the PID was recycled. Keep + # a terminal lost record when notification was requested so a + # restart never silently drops the user's completion signal. + if not entry.get("notify_on_complete"): + continue + session = ProcessSession( + id=entry["session_id"], command=entry.get("command", "unknown"), + task_id=entry.get("task_id", ""), session_key=entry.get("session_key", ""), + pid=pid, host_start_time=recorded_start, + cwd=entry.get("cwd"), started_at=entry.get("started_at", time.time()), + detached=True, exited=True, exit_code=None, + completion_reason="lost", termination_source="restart_recovery", + output_buffer=entry.get("output_tail", ""), + output_log_path=entry.get("output_log_path", ""), + watcher_platform=entry.get("watcher_platform", ""), + watcher_chat_id=entry.get("watcher_chat_id", ""), + watcher_user_id=entry.get("watcher_user_id", ""), + watcher_user_name=entry.get("watcher_user_name", ""), + watcher_thread_id=entry.get("watcher_thread_id", ""), + watcher_message_id=entry.get("watcher_message_id", ""), + watcher_origin_source=entry.get("watcher_origin_source"), + watcher_profile=entry.get("watcher_profile", ""), + watcher_parent_session_id=entry.get("watcher_parent_session_id", ""), + watcher_interval=entry.get("watcher_interval", 0), + notify_on_complete=entry.get("notify_on_complete", False), + watch_patterns=entry.get("watch_patterns", []), + ) + with self._lock: + self._finished[session.id] = session + if session.watcher_interval > 0: + self.pending_watchers.append(self._watcher_payload(session)) + recovered += 1 continue session = ProcessSession( @@ -2045,6 +2243,7 @@ def recover_from_checkpoint(self) -> int: pid_scope=pid_scope, cwd=entry.get("cwd"), started_at=entry.get("started_at", time.time()), + output_log_path=entry.get("output_log_path", ""), detached=True, # Can't read output, but can report status + kill watcher_platform=entry.get("watcher_platform", ""), watcher_chat_id=entry.get("watcher_chat_id", ""), @@ -2052,6 +2251,9 @@ def recover_from_checkpoint(self) -> int: watcher_user_name=entry.get("watcher_user_name", ""), watcher_thread_id=entry.get("watcher_thread_id", ""), watcher_message_id=entry.get("watcher_message_id", ""), + watcher_origin_source=entry.get("watcher_origin_source"), + watcher_profile=entry.get("watcher_profile", ""), + watcher_parent_session_id=entry.get("watcher_parent_session_id", ""), watcher_interval=entry.get("watcher_interval", 0), notify_on_complete=entry.get("notify_on_complete", False), watch_patterns=entry.get("watch_patterns", []), @@ -2063,23 +2265,32 @@ def recover_from_checkpoint(self) -> int: # Re-enqueue watcher so gateway can resume notifications if session.watcher_interval > 0: - self.pending_watchers.append({ - "session_id": session.id, - "check_interval": session.watcher_interval, - "session_key": session.session_key, - "platform": session.watcher_platform, - "chat_id": session.watcher_chat_id, - "user_id": session.watcher_user_id, - "user_name": session.watcher_user_name, - "thread_id": session.watcher_thread_id, - "message_id": session.watcher_message_id, - "notify_on_complete": session.notify_on_complete, - }) + self.pending_watchers.append(self._watcher_payload(session)) self._write_checkpoint() return recovered + @staticmethod + def _watcher_payload(session: ProcessSession) -> Dict[str, Any]: + payload = { + "session_id": session.id, + "check_interval": session.watcher_interval, + "session_key": session.session_key, + "platform": session.watcher_platform, + "chat_id": session.watcher_chat_id, + "user_id": session.watcher_user_id, + "user_name": session.watcher_user_name, + "thread_id": session.watcher_thread_id, + "message_id": session.watcher_message_id, + "origin_message_id": session.watcher_message_id, + "origin_source": deepcopy(session.watcher_origin_source), + "origin_profile": session.watcher_profile, + "parent_session_id": session.watcher_parent_session_id, + "notify_on_complete": session.notify_on_complete, + } + return payload + # Module-level singleton process_registry = ProcessRegistry() diff --git a/tools/terminal_tool.py b/tools/terminal_tool.py index b02c4bd2f9106..c66c5c3ece6ab 100644 --- a/tools/terminal_tool.py +++ b/tools/terminal_tool.py @@ -2536,6 +2536,50 @@ def terminal_tool( session_key = get_current_session_key(default="") or (task_id or "") + watcher_config = None + notify_unsupported = "" + watch_patterns, conflict_note = _resolve_notification_flag_conflict( + notify_on_complete=bool(notify_on_complete), + watch_patterns=watch_patterns, + background=bool(background), + ) + if background and (notify_on_complete or watch_patterns): + from gateway.session_context import ( + async_delivery_supported as _async_ok, + capture_session_origin, + get_session_env as _gse, + ) + + if not _async_ok(): + notify_on_complete = False + watch_patterns = None + notify_unsupported = ( + "notify_on_complete / watch_patterns are not available in " + "this session — it cannot receive an async completion after " + "the turn ends (a one-shot runner such as `hermes -z`, a " + "cron job, a Kanban worker, or a stateless HTTP endpoint). " + "The process is running in the background; retrieve its " + "result with process(action='poll') or process(action='wait')." + ) + else: + origin = capture_session_origin() + source = origin.get("origin_source") or {} + platform = str(source.get("platform") or _gse("HERMES_SESSION_PLATFORM", "")) + watcher_config = { + "platform": platform, + "chat_id": str(source.get("chat_id") or _gse("HERMES_SESSION_CHAT_ID", "")), + "user_id": str(source.get("user_id") or _gse("HERMES_SESSION_USER_ID", "")), + "user_name": str(source.get("user_name") or _gse("HERMES_SESSION_USER_NAME", "")), + "thread_id": str(source.get("thread_id") or _gse("HERMES_SESSION_THREAD_ID", "")), + "origin_message_id": origin["origin_message_id"], + "origin_source": source or None, + "origin_profile": origin["origin_profile"], + "parent_session_id": origin["parent_session_id"], + "check_interval": 5 if notify_on_complete and platform else 0, + "notify_on_complete": bool(notify_on_complete), + "watch_patterns": list(watch_patterns or []), + } + if background: # Spawn a tracked background process via the process registry. # For local backends: uses subprocess.Popen with output buffering. @@ -2556,6 +2600,7 @@ def terminal_tool( session_key=session_key, env_vars=env.env if hasattr(env, 'env') else None, use_pty=effective_pty, + watcher_config=watcher_config, ) else: proc_session = process_registry.spawn_via_env( @@ -2564,6 +2609,7 @@ def terminal_tool( cwd=effective_cwd, task_id=effective_task_id, session_key=session_key, + watcher_config=watcher_config, ) result_data = { @@ -2580,6 +2626,17 @@ def terminal_tool( result_data["approval"] = approval_note if pty_disabled_reason: result_data["pty_note"] = pty_disabled_reason + if notify_unsupported: + result_data["notify_on_complete"] = False + result_data["notify_unsupported"] = notify_unsupported + logger.info( + "background proc %s: async delivery unsupported on this " + "session; notify_on_complete/watch_patterns disabled", + proc_session.id, + ) + if conflict_note: + logger.warning("background proc %s: %s", proc_session.id, conflict_note) + result_data["watch_patterns_ignored"] = conflict_note # Nudge: background=True without notify_on_complete=True OR # watch_patterns is a silent process. The agent has NO way to @@ -2686,94 +2743,20 @@ def terminal_tool( else canonical_hint ) - # Populate routing metadata on the session so that - # watch-pattern and completion notifications can be - # routed back to the correct chat/thread. - if background and (notify_on_complete or watch_patterns): - from gateway.session_context import ( - async_delivery_supported as _async_ok, - get_session_env as _gse, - ) - - # Finite sessions (stateless HTTP requests and one-shot - # Kanban workers) cannot route a completion back to the - # agent after the turn/process ends. Refuse the promise: - # drop the flags and tell the agent to poll. - if not _async_ok(): - notify_on_complete = False - watch_patterns = None - result_data["notify_on_complete"] = False - result_data["notify_unsupported"] = ( - "notify_on_complete / watch_patterns are not available in " - "this session — it cannot receive an async completion after " - "the turn ends (a one-shot runner such as `hermes -z`, a " - "cron job, a Kanban worker, or a stateless HTTP endpoint). " - "The process is " - "running in the background; retrieve its result with " - "process(action='poll') or process(action='wait')." - ) - logger.info( - "background proc %s: async delivery unsupported on this " - "session; notify_on_complete/watch_patterns disabled", - proc_session.id, - ) - else: - _gw_platform = _gse("HERMES_SESSION_PLATFORM", "") - if _gw_platform: - _gw_chat_id = _gse("HERMES_SESSION_CHAT_ID", "") - _gw_thread_id = _gse("HERMES_SESSION_THREAD_ID", "") - _gw_user_id = _gse("HERMES_SESSION_USER_ID", "") - _gw_user_name = _gse("HERMES_SESSION_USER_NAME", "") - _gw_message_id = _gse("HERMES_SESSION_MESSAGE_ID", "") - proc_session.watcher_platform = _gw_platform - proc_session.watcher_chat_id = _gw_chat_id - proc_session.watcher_user_id = _gw_user_id - proc_session.watcher_user_name = _gw_user_name - proc_session.watcher_thread_id = _gw_thread_id - proc_session.watcher_message_id = _gw_message_id - - # Mutual exclusion: if both notify_on_complete and watch_patterns - # are set, drop watch_patterns. The combination produces duplicate - # notifications (one per match + one on exit) that deliver - # asynchronously and can spam the user long after the process ends. - # notify_on_complete is the more useful signal for "let me know - # when the task finishes"; watch_patterns should be reserved for - # standalone mid-process signals on long-lived processes. - watch_patterns, conflict_note = _resolve_notification_flag_conflict( - notify_on_complete=bool(notify_on_complete), - watch_patterns=watch_patterns, - background=bool(background), - ) - if conflict_note: - logger.warning("background proc %s: %s", proc_session.id, conflict_note) - result_data["watch_patterns_ignored"] = conflict_note - # Mark for agent notification on completion if notify_on_complete and background: - proc_session.notify_on_complete = True result_data["notify_on_complete"] = True # In gateway mode, auto-register a fast watcher so the # gateway can detect completion and trigger a new agent # turn. CLI mode uses the completion_queue directly. if proc_session.watcher_platform: - proc_session.watcher_interval = 5 - process_registry.pending_watchers.append({ - "session_id": proc_session.id, - "check_interval": 5, - "session_key": session_key, - "platform": proc_session.watcher_platform, - "chat_id": proc_session.watcher_chat_id, - "user_id": proc_session.watcher_user_id, - "user_name": proc_session.watcher_user_name, - "thread_id": proc_session.watcher_thread_id, - "message_id": proc_session.watcher_message_id, - "notify_on_complete": True, - }) + process_registry.pending_watchers.append( + process_registry._watcher_payload(proc_session) + ) # Set watch patterns for output monitoring if watch_patterns and background: - proc_session.watch_patterns = list(watch_patterns) result_data["watch_patterns"] = proc_session.watch_patterns return json.dumps(result_data, ensure_ascii=False) From b08fbc56e9a99ff0a6b80d8f05b90d17ec1deb5b Mon Sep 17 00:00:00 2001 From: Zhexuan Yang Date: Fri, 31 Jul 2026 09:00:46 +0800 Subject: [PATCH 3/4] fix(terminal): harden restart delivery durability Register producers before exit observers start, persist local notified output through a child-owned sink and exit sidecar, retain undelivered finals during pruning, and acknowledge final delivery only after explicit acceptance. --- gateway/run.py | 10 +- .../test_background_process_notifications.py | 29 +++ .../tools/test_terminal_origin_durability.py | 95 ++++++++ tools/process_registry.py | 220 +++++++++++++++--- 4 files changed, 313 insertions(+), 41 deletions(-) diff --git a/gateway/run.py b/gateway/run.py index 1419bb019b022..76c3277b62b92 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -21399,9 +21399,11 @@ async def _run_process_watcher(self, watcher: dict) -> None: delivered = await self._deliver_completion_notification( synth_text, completion_evt, ) - if delivered is False: - # The process remains terminal; retry after failed - # adapter injection instead of suppressing the result. + if delivered is not True: + # Only explicit adapter acceptance acknowledges and + # deletes durable producer state. ``None`` includes an + # unroutable/missing-profile adapter and must remain + # pending just like a retryable delivery failure. continue _pr_check.mark_notification_delivered(session_id) break @@ -21443,7 +21445,7 @@ async def _run_process_watcher(self, watcher: dict) -> None: delivered = await self._deliver_process_status_direct( message_text, watcher, status_key=f"process:{session_id}:final" ) - if delivered is False: + if delivered is not True: continue _pr_check.mark_notification_delivered(session_id) break diff --git a/tests/gateway/test_background_process_notifications.py b/tests/gateway/test_background_process_notifications.py index 8aa8cc98c35ff..0c7cda4de6a49 100644 --- a/tests/gateway/test_background_process_notifications.py +++ b/tests/gateway/test_background_process_notifications.py @@ -260,6 +260,35 @@ async def _instant_sleep(*_a, **_kw): default_adapter.handle_message.assert_not_awaited() +@pytest.mark.asyncio +async def test_unroutable_final_is_not_acknowledged(monkeypatch, tmp_path): + """Missing routing/profile adapter must leave durable completion pending.""" + import tools.process_registry as pr_module + + session = SimpleNamespace( + output_buffer="done\n", exited=True, exit_code=0, + command="codex exec task", started_at=1.0, + completion_reason="exited", termination_source="", + ) + registry = _FakeRegistry([session, None]) + monkeypatch.setattr(pr_module, "process_registry", registry) + + async def _instant_sleep(*_a, **_kw): + pass + + monkeypatch.setattr(asyncio, "sleep", _instant_sleep) + runner = _build_runner(monkeypatch, tmp_path, "all") + runner._deliver_completion_notification = AsyncMock(return_value=None) + + await runner._run_process_watcher({ + **_watcher_dict("proc-unroutable"), + "notify_on_complete": True, + }) + + runner._deliver_completion_notification.assert_awaited_once() + assert not hasattr(registry, "delivered_session_id") + + @pytest.mark.asyncio async def test_inject_watch_notification_routes_from_session_store_origin(monkeypatch, tmp_path): from gateway.session import SessionSource diff --git a/tests/tools/test_terminal_origin_durability.py b/tests/tools/test_terminal_origin_durability.py index bc1cb7b8b95cd..48bd9ee10dcda 100644 --- a/tests/tools/test_terminal_origin_durability.py +++ b/tests/tools/test_terminal_origin_durability.py @@ -2,6 +2,8 @@ import json import os +import sys +import time from unittest.mock import MagicMock, patch from tools.process_registry import ProcessRegistry, ProcessSession @@ -146,3 +148,96 @@ def test_completion_event_carries_terminal_origin(tmp_path): assert event["origin_source"]["chat_id"] == "oc_chat" assert event["origin_profile"] == "coder" assert event["parent_session_id"] == "sess_parent" + + +def test_fast_exit_cannot_be_reinserted_as_running(tmp_path): + registry = ProcessRegistry() + checkpoint = tmp_path / "processes.json" + fake_proc = MagicMock(pid=4242, returncode=0) + fake_proc.poll.return_value = 0 + + class ImmediateThread: + def __init__(self, *, target, args, **_kwargs): + self._target = target + self._args = args + + def start(self): + self._target(*self._args) + + with patch("tools.process_registry.CHECKPOINT_PATH", checkpoint), \ + patch("tools.process_registry.get_hermes_home", return_value=tmp_path), \ + patch("tools.process_registry.subprocess.Popen", return_value=fake_proc), \ + patch("tools.process_registry.threading.Thread", ImmediateThread), \ + patch.object(registry, "_safe_host_start_time", return_value=99): + session = registry.spawn_local( + "true", cwd=str(tmp_path), watcher_config=_watcher_config() + ) + + assert session.id not in registry._running + assert registry._finished[session.id] is session + assert registry.completion_queue.get_nowait()["session_id"] == session.id + + +def test_child_owned_log_keeps_growing_after_registry_restart(tmp_path): + checkpoint = tmp_path / "processes.json" + registry = ProcessRegistry() + command = ( + f'{sys.executable} -c "import time; print(\'first\', flush=True); ' + 'time.sleep(0.35); print(\'second\', flush=True)"' + ) + no_reader = MagicMock() + + with patch("tools.process_registry.CHECKPOINT_PATH", checkpoint), \ + patch("tools.process_registry.get_hermes_home", return_value=tmp_path), \ + patch("tools.process_registry.threading.Thread", return_value=no_reader): + session = registry.spawn_local( + command, cwd=str(tmp_path), watcher_config=_watcher_config() + ) + deadline = time.time() + 3 + while time.time() < deadline: + if "first" in open(session.output_log_path, encoding="utf-8").read(): + break + time.sleep(0.02) + + restarted = ProcessRegistry() + with patch.object( + restarted, "_host_pid_is_ours", + side_effect=lambda *_args: session.process.poll() is None, + ): + assert restarted.recover_from_checkpoint() == 1 + assert "first" in restarted.get(session.id).output_buffer + + deadline = time.time() + 3 + while time.time() < deadline: + recovered = restarted.get(session.id) + if "second" in recovered.output_buffer and recovered.exited: + break + time.sleep(0.02) + + assert "second" in recovered.output_buffer + assert recovered.exited is True + assert recovered.exit_code == 0 + + +def test_prune_preserves_undelivered_final_and_artifacts(tmp_path): + registry = ProcessRegistry() + output_log = tmp_path / "pending.log" + exit_file = tmp_path / "pending.exit" + output_log.write_text("done\n") + exit_file.write_text("0\n") + session = ProcessSession( + id="proc_pending", + command="codex exec task", + started_at=1.0, + exited=True, + notify_on_complete=True, + output_log_path=str(output_log), + exit_code_path=str(exit_file), + ) + registry._finished[session.id] = session + + registry._prune_if_needed() + + assert registry._finished[session.id] is session + assert output_log.exists() + assert exit_file.exists() diff --git a/tools/process_registry.py b/tools/process_registry.py index fed9584f70627..0f958f63825f3 100644 --- a/tools/process_registry.py +++ b/tools/process_registry.py @@ -108,6 +108,7 @@ class ProcessSession: termination_source: str = "" # process.kill|kill_all|backend_lost|failed_start output_buffer: str = "" # Rolling output (last MAX_OUTPUT_CHARS) output_log_path: str = "" # Durable append-only output log + exit_code_path: str = "" # Child-written exit status sidecar max_output_chars: int = MAX_OUTPUT_CHARS detached: bool = False # True if recovered from crash (no pipe) pid_scope: str = "host" # "host" for local/PTY PIDs, "sandbox" for env-local PIDs @@ -228,10 +229,12 @@ def _clean_shell_noise(text: str) -> str: lines.pop(0) return "\n".join(lines) - def _emit_output(self, session: ProcessSession, chunk: str) -> None: + def _emit_output( + self, session: ProcessSession, chunk: str, *, persist: bool = True, + ) -> None: """Forward a freshly-read chunk to the live-output sink, if one is set. Called from reader threads; never raise into the read loop.""" - if chunk and session.output_log_path: + if persist and chunk and session.output_log_path: try: path = Path(session.output_log_path) path.parent.mkdir(parents=True, exist_ok=True) @@ -542,13 +545,28 @@ def _refresh_detached_session(self, session: Optional[ProcessSession]) -> Option if session.exited: return session session.exited = True - # Recovered sessions no longer have a waitable handle, so the real - # exit code is unavailable once the original process object is gone. - session.exit_code = None + session.exit_code = self._read_exit_code(session) self._move_to_finished(session) return session + @staticmethod + def _read_exit_code(session: ProcessSession) -> Optional[int]: + """Read the status written by the child-owned durable shell wrapper.""" + if session.exit_code_path: + try: + value = Path(session.exit_code_path).read_text( + encoding="utf-8", errors="replace" + ).strip().splitlines()[-1] + return int(value) + except (OSError, ValueError, IndexError): + pass + if session.process is not None: + return session.process.returncode + if session._pty is not None: + return getattr(session._pty, "exitstatus", None) + return None + @staticmethod def _proc_alive(proc) -> bool: """True if a psutil.Process is running and not a zombie. @@ -790,7 +808,32 @@ def spawn_local( session.output_log_path = str( get_hermes_home() / "cache" / "processes" / f"{session.id}.log" ) + session.exit_code_path = str( + get_hermes_home() / "cache" / "processes" / f"{session.id}.exit" + ) self._apply_watcher_config(session, watcher_config) + durable_child_sink = bool( + session.notify_on_complete or session.watcher_interval > 0 + ) + if durable_child_sink: + output_path = Path(session.output_log_path) + output_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + output_path.touch(mode=0o600, exist_ok=True) + try: + output_path.chmod(0o600) + except OSError: + pass + Path(session.exit_code_path).unlink(missing_ok=True) + + def _shell_command() -> str: + if not durable_child_sink: + return f"set +m; {safe_command}" + quoted_log = shlex.quote(session.output_log_path) + quoted_exit = shlex.quote(session.exit_code_path) + return ( + f"set +m; {{ {safe_command}; }} >> {quoted_log} 2>&1; " + f"rc=$?; printf '%s\\n' \"$rc\" > {quoted_exit}; exit \"$rc\"" + ) if use_pty: # Try PTY mode for interactive CLI tools @@ -803,7 +846,7 @@ def spawn_local( pty_env = _sanitize_subprocess_env(os.environ, env_vars) pty_env["PYTHONUNBUFFERED"] = "1" pty_proc = _PtyProcessCls.spawn( - [user_shell, "-lic", f"set +m; {safe_command}"], + [user_shell, "-lic", _shell_command()], cwd=session.cwd, env=pty_env, dimensions=(30, 120), @@ -813,21 +856,25 @@ def spawn_local( # Store the pty handle on the session for read/write session._pty = pty_proc - # PTY reader thread + # Register and checkpoint before the reader can observe exit. + # This closes the fast-exit race where _move_to_finished saw + # no running producer and the spawn path reinserted it later. + with self._lock: + self._prune_if_needed() + self._running[session.id] = session + self._write_checkpoint() + reader = threading.Thread( - target=self._pty_reader_loop, + target=( + self._local_log_poller_loop + if durable_child_sink else self._pty_reader_loop + ), args=(session,), daemon=True, name=f"proc-pty-reader-{session.id}", ) session._reader_thread = reader reader.start() - - with self._lock: - self._prune_if_needed() - self._running[session.id] = session - - self._write_checkpoint() return session except ImportError: @@ -847,13 +894,13 @@ def spawn_local( _popen_kwargs = {"creationflags": windows_hide_flags()} if _IS_WINDOWS else {} proc = subprocess.Popen( - [user_shell, "-lic", f"set +m; {safe_command}"], + [user_shell, "-lic", _shell_command()], text=True, cwd=session.cwd, env=bg_env, encoding="utf-8", errors="replace", - stdout=subprocess.PIPE, + stdout=(subprocess.DEVNULL if durable_child_sink else subprocess.PIPE), stderr=subprocess.STDOUT, stdin=subprocess.DEVNULL, start_new_session=True, @@ -865,21 +912,24 @@ def spawn_local( session.host_start_time = self._safe_host_start_time(session.pid) try: - # Start output reader thread + # Register and checkpoint before starting any thread capable of + # moving this producer to finished. + with self._lock: + self._prune_if_needed() + self._running[session.id] = session + self._write_checkpoint() + reader = threading.Thread( - target=self._reader_loop, + target=( + self._local_log_poller_loop + if durable_child_sink else self._reader_loop + ), args=(session,), daemon=True, name=f"proc-reader-{session.id}", ) session._reader_thread = reader reader.start() - - with self._lock: - self._prune_if_needed() - self._running[session.id] = session - - self._write_checkpoint() except Exception: # Post-Popen setup failed — kill the orphaned subprocess (and any # descendants spawned via setsid) before re-raising so they do not @@ -899,6 +949,9 @@ def spawn_local( proc.wait(timeout=5) except Exception: pass + with self._lock: + self._running.pop(session.id, None) + self._write_checkpoint() raise return session @@ -988,7 +1041,13 @@ def spawn_via_env( session.output_buffer = f"Failed to start: {e}" if not session.exited: - # Start a poller thread that periodically reads the log file + # Register/checkpoint before the poller can observe an immediate + # sandbox exit and move the producer to finished. + with self._lock: + self._prune_if_needed() + self._running[session.id] = session + self._write_checkpoint() + reader = threading.Thread( target=self._env_poller_loop, args=(session, env, log_path, pid_path, exit_path), @@ -998,18 +1057,66 @@ def spawn_via_env( session._reader_thread = reader reader.start() - with self._lock: - self._prune_if_needed() - if not session.exited: - self._running[session.id] = session - - if not session.exited: - self._write_checkpoint() - return session # ----- Reader / Poller Threads ----- + def _local_log_poller_loop(self, session: ProcessSession) -> None: + """Tail a child-owned log so output survives a gateway process restart.""" + offset = 0 + first_chunk = True + + def _read_delta() -> None: + nonlocal offset, first_chunk + try: + path = Path(session.output_log_path) + with path.open("r", encoding="utf-8", errors="replace") as fh: + fh.seek(offset) + chunk = fh.read() + offset = fh.tell() + except Exception: + return + if not chunk: + return + if first_chunk: + chunk = self._clean_shell_noise(chunk) + first_chunk = False + with session._lock: + session.output_buffer += chunk + if len(session.output_buffer) > session.max_output_chars: + session.output_buffer = session.output_buffer[-session.max_output_chars:] + self._check_watch_patterns(session, chunk) + self._emit_output(session, chunk, persist=False) + + try: + while True: + _read_delta() + proc = session.process + pty = session._pty + alive = pty.isalive() if pty is not None else bool( + proc is not None and proc.poll() is None + ) + if not alive: + break + time.sleep(0.2) + _read_delta() + if session._pty is not None: + try: + session._pty.wait() + except Exception: + pass + elif session.process is not None: + try: + session.process.wait(timeout=5) + except Exception: + pass + finally: + session.exited = True + if session.completion_reason != "killed": + session.exit_code = self._read_exit_code(session) + session.completion_reason = "exited" + self._move_to_finished(session) + def _reader_loop(self, session: ProcessSession): """Background thread: read stdout from a local Popen process. @@ -1268,13 +1375,17 @@ def is_completion_consumed(self, session_id: str) -> bool: def mark_notification_delivered(self, session_id: str) -> None: """Acknowledge a terminal notification after adapter/consumer acceptance.""" output_log_path = "" + exit_code_path = "" with self._lock: session = self._running.get(session_id) or self._finished.get(session_id) if session is not None: session.notification_delivered = True output_log_path = session.output_log_path + exit_code_path = session.exit_code_path session.output_log_path = "" + session.exit_code_path = "" self._delete_output_log(output_log_path) + self._delete_output_log(exit_code_path) self._write_checkpoint() def _mark_completion_consumed(self, session_id: str) -> None: @@ -2022,19 +2133,34 @@ def _prune_if_needed(self): expired = [ sid for sid, s in self._finished.items() if (now - s.started_at) > FINISHED_TTL_SECONDS + and not ( + s.notify_on_complete + and not s.notification_delivered + and sid not in self._completion_consumed + ) ] for sid in expired: removed = self._finished.pop(sid) self._delete_output_log(removed.output_log_path) + self._delete_output_log(removed.exit_code_path) self._completion_consumed.discard(sid) self._poll_observed.discard(sid) # If still over limit, remove oldest finished total = len(self._running) + len(self._finished) - if total >= MAX_PROCESSES and self._finished: - oldest_id = min(self._finished, key=lambda sid: self._finished[sid].started_at) + pruneable = { + sid: session for sid, session in self._finished.items() + if not ( + session.notify_on_complete + and not session.notification_delivered + and sid not in self._completion_consumed + ) + } + if total >= MAX_PROCESSES and pruneable: + oldest_id = min(pruneable, key=lambda sid: pruneable[sid].started_at) removed = self._finished.pop(oldest_id) self._delete_output_log(removed.output_log_path) + self._delete_output_log(removed.exit_code_path) self._completion_consumed.discard(oldest_id) self._poll_observed.discard(oldest_id) @@ -2079,6 +2205,7 @@ def _checkpoint_entry(s: ProcessSession) -> Dict[str, Any]: "termination_source": s.termination_source, "output_tail": s.output_buffer[-2000:] if s.exited else "", "output_log_path": s.output_log_path, + "exit_code_path": s.exit_code_path, "watcher_platform": s.watcher_platform, "watcher_chat_id": s.watcher_chat_id, "watcher_user_id": s.watcher_user_id, @@ -2148,6 +2275,7 @@ def recover_from_checkpoint(self) -> int: termination_source=entry.get("termination_source", ""), output_buffer=entry.get("output_tail", ""), output_log_path=entry.get("output_log_path", ""), + exit_code_path=entry.get("exit_code_path", ""), watcher_platform=entry.get("watcher_platform", ""), watcher_chat_id=entry.get("watcher_chat_id", ""), watcher_user_id=entry.get("watcher_user_id", ""), @@ -2204,15 +2332,32 @@ def recover_from_checkpoint(self) -> int: # restart never silently drops the user's completion signal. if not entry.get("notify_on_complete"): continue + recovered_exit_code = None + exit_code_path = entry.get("exit_code_path", "") + if exit_code_path: + try: + recovered_exit_code = int( + Path(exit_code_path).read_text( + encoding="utf-8", errors="replace" + ).strip().splitlines()[-1] + ) + except (OSError, ValueError, IndexError): + pass session = ProcessSession( id=entry["session_id"], command=entry.get("command", "unknown"), task_id=entry.get("task_id", ""), session_key=entry.get("session_key", ""), pid=pid, host_start_time=recorded_start, cwd=entry.get("cwd"), started_at=entry.get("started_at", time.time()), - detached=True, exited=True, exit_code=None, - completion_reason="lost", termination_source="restart_recovery", + detached=True, exited=True, exit_code=recovered_exit_code, + completion_reason=( + "exited" if recovered_exit_code is not None else "lost" + ), + termination_source=( + "" if recovered_exit_code is not None else "restart_recovery" + ), output_buffer=entry.get("output_tail", ""), output_log_path=entry.get("output_log_path", ""), + exit_code_path=entry.get("exit_code_path", ""), watcher_platform=entry.get("watcher_platform", ""), watcher_chat_id=entry.get("watcher_chat_id", ""), watcher_user_id=entry.get("watcher_user_id", ""), @@ -2244,6 +2389,7 @@ def recover_from_checkpoint(self) -> int: cwd=entry.get("cwd"), started_at=entry.get("started_at", time.time()), output_log_path=entry.get("output_log_path", ""), + exit_code_path=entry.get("exit_code_path", ""), detached=True, # Can't read output, but can report status + kill watcher_platform=entry.get("watcher_platform", ""), watcher_chat_id=entry.get("watcher_chat_id", ""), From 38fa75924097a64732ea6febef0ebdb7f003dea6 Mon Sep 17 00:00:00 2001 From: Zhexuan Yang Date: Fri, 31 Jul 2026 09:05:03 +0800 Subject: [PATCH 4/4] test(async): keep core origin checks adapter-neutral Assert immutable source/profile routing and inject the thread metadata contract explicitly so the durability PR does not depend on Feishu adapter behavior owned by #72024. --- tests/gateway/test_async_delegation_origin_durability.py | 5 ++--- tests/gateway/test_background_process_notifications.py | 6 ++++++ 2 files changed, 8 insertions(+), 3 deletions(-) diff --git a/tests/gateway/test_async_delegation_origin_durability.py b/tests/gateway/test_async_delegation_origin_durability.py index 8a9acf01065da..8154526ab3806 100644 --- a/tests/gateway/test_async_delegation_origin_durability.py +++ b/tests/gateway/test_async_delegation_origin_durability.py @@ -189,9 +189,8 @@ async def test_synthetic_completion_uses_origin_profile_adapter_and_anchor(): assert event.message_id == "om-origin" assert event.source.message_id == "om-origin" assert event.metadata["gateway_origin_profile"] == "coder" - assert runner._thread_metadata_for_source( - event.source, event.message_id - ) == {"reply_in_thread": True} + assert event.source.platform == Platform.FEISHU + assert event.source.profile == "coder" @pytest.mark.asyncio diff --git a/tests/gateway/test_background_process_notifications.py b/tests/gateway/test_background_process_notifications.py index 0c7cda4de6a49..896d61862bfc2 100644 --- a/tests/gateway/test_background_process_notifications.py +++ b/tests/gateway/test_background_process_notifications.py @@ -220,6 +220,11 @@ async def _instant_sleep(*_a, **_kw): runner.adapters = {Platform.FEISHU: default_adapter} runner._profile_adapters = {"coder": {Platform.FEISHU: coder_adapter}} runner._active_profile_name = lambda: "default" + thread_metadata = { + "reply_in_thread": True, + "reply_to_message_id": "om-origin", + } + runner._thread_metadata_for_source = MagicMock(return_value=thread_metadata) source = SessionSource( platform=Platform.FEISHU, @@ -246,6 +251,7 @@ async def _instant_sleep(*_a, **_kw): assert "phase latest" in second.args[2] assert first.kwargs["metadata"]["reply_in_thread"] is True assert first.kwargs["metadata"]["reply_to_message_id"] == "om-origin" + runner._thread_metadata_for_source.assert_called_with(source, "om-origin") coder_adapter.handle_message.assert_awaited_once() final_event = coder_adapter.handle_message.await_args.args[0]