diff --git a/gateway/run.py b/gateway/run.py index 911021a36ee60..76c3277b62b92 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__(). @@ -13160,6 +13163,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 +13199,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 +13257,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 +13333,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 +15565,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 +20291,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, @@ -20264,8 +20304,10 @@ 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(), async_delivery=_async_delivery, ) @@ -20739,6 +20781,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 +20893,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 +20945,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 +20977,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 +21025,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 +21059,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 +21082,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( @@ -21026,49 +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) - 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 + 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 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 ( @@ -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 "") @@ -21237,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) @@ -21245,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 --- @@ -21285,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, @@ -21298,10 +21399,13 @@ 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 # --- Normal text-only notification --- @@ -21338,26 +21442,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 not True: + 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 @@ -21368,24 +21467,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 24d556d2cba71..950a7018871b9 100644 --- a/gateway/session_context.py +++ b/gateway/session_context.py @@ -37,7 +37,8 @@ """ from contextvars import ContextVar -from typing import Any +from copy import deepcopy +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 +94,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 +179,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 +215,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 +252,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 +304,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 +394,39 @@ 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 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: """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..8154526ab3806 --- /dev/null +++ b/tests/gateway/test_async_delegation_origin_durability.py @@ -0,0 +1,240 @@ +"""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 event.source.platform == Platform.FEISHU + assert event.source.profile == "coder" + + +@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/gateway/test_background_process_notifications.py b/tests/gateway/test_background_process_notifications.py index f55c92deddb31..896d61862bfc2 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,140 @@ 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" + 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, + 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" + 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] + 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_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/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_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/tests/tools/test_terminal_origin_durability.py b/tests/tools/test_terminal_origin_durability.py new file mode 100644 index 0000000000000..48bd9ee10dcda --- /dev/null +++ b/tests/tools/test_terminal_origin_durability.py @@ -0,0 +1,243 @@ +"""Durable immutable origins for terminal(background=True).""" + +import json +import os +import sys +import time +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" + + +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/async_delegation.py b/tools/async_delegation.py index 702036df0c41c..05800fedbeea4 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,41 @@ 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_context import ( + capture_session_origin, + ) + + 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. + from gateway.session import SessionSource + + source_payload = SessionSource.from_dict(source_payload).to_dict() + return { + "origin_message_id": captured["origin_message_id"], + "origin_source": dict(source_payload) if source_payload else None, + "origin_profile": captured["origin_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 +689,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 +750,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 +900,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 +919,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 +954,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 +1000,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 +1115,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 +1138,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, diff --git a/tools/process_registry.py b/tools/process_registry.py index 7daf74b2e3365..0f958f63825f3 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,8 @@ 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 + 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 @@ -115,8 +119,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 @@ -221,9 +229,19 @@ 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 persist and 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 +331,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 +343,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 +353,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 +368,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 +382,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 +521,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 @@ -504,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. @@ -686,6 +742,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 +776,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 +805,35 @@ 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" + ) + 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 @@ -734,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), @@ -744,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: @@ -778,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, @@ -796,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 @@ -830,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 @@ -842,6 +964,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 +987,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) @@ -914,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), @@ -924,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. @@ -1158,7 +1339,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 +1352,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 +1372,26 @@ 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 = "" + 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: + 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 +1528,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 +1691,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 +1741,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 +1814,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 +1844,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 +1866,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" @@ -1918,17 +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: - del self._finished[sid] + 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) - del self._finished[oldest_id] + 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) @@ -1944,10 +2176,52 @@ 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, + "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, + "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 +2232,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 +2263,38 @@ 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", ""), + 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", ""), + 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 +2327,55 @@ 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 + 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=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", ""), + 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 +2388,8 @@ 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", ""), + 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", ""), @@ -2052,6 +2397,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 +2411,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)