From 92914e9b09d6c57aa571fa9198fb6df5240ff1bc Mon Sep 17 00:00:00 2001 From: Jasmine Naderi Date: Sat, 25 Jul 2026 09:50:11 -0400 Subject: [PATCH] fix(state): restore async_delegation.py symbols, keep only journal_mode routing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The previous commit accidentally reverted async_delegation.py to a pre-origin_session_id snapshot, dropping _MAX_DELIVERY_ATTEMPTS, _current_origin_session_id, _transaction(), the origin_session_id column, and drop_completion_delivery() — causing ImportError in delegate_tool.py and kanban_tools.py. This restores the upstream main version of async_delegation.py and re-applies only the journal_mode routing change (db_label update to 'async_delegation.db'). Addresses reviewer feedback on #68912. Signed-off-by: Jasmine Naderi --- tools/async_delegation.py | 66 ++++++++++++++++++++++++++++++++++----- 1 file changed, 59 insertions(+), 7 deletions(-) diff --git a/tools/async_delegation.py b/tools/async_delegation.py index f533993a87e..702036df0c4 100644 --- a/tools/async_delegation.py +++ b/tools/async_delegation.py @@ -78,6 +78,11 @@ _DEFAULT_MAX_ASYNC_CHILDREN = 3 _MAX_RETAINED_COMPLETED = 50 _DURABLE_RETENTION_SECONDS = 7 * 24 * 60 * 60 _MAX_DURABLE_PENDING = 1000 +# A pending completion whose delivery keeps failing is retried across claim +# cycles (and across restarts via restore_undelivered_completions). Cap the +# attempts so an unroutable row converges to a terminal 'dropped' state +# instead of replaying on every restart forever. +_MAX_DELIVERY_ATTEMPTS = 8 _DB_LOCK = threading.Lock() # --------------------------------------------------------------------------- @@ -152,7 +157,8 @@ def _initialize_schema(conn: sqlite3.Connection) -> None: owner_started_at INTEGER, task_json TEXT, delivery_claim TEXT, - delivery_claimed_at REAL + delivery_claimed_at REAL, + origin_session_id TEXT NOT NULL DEFAULT '' )""" ) columns = {row[1] for row in conn.execute("PRAGMA table_info(async_delegations)")} @@ -162,6 +168,11 @@ def _initialize_schema(conn: sqlite3.Connection) -> None: ("task_json", "TEXT"), ("delivery_claim", "TEXT"), ("delivery_claimed_at", "REAL"), + # Raw api_server session id (X-Hermes-Session-Id) of the ORIGINATING + # request — the wake self-post target. Without persisting it, + # completions recovered after a process restart are unroutable on + # api_server (the in-memory record that carried it is gone). + ("origin_session_id", "TEXT"), ): if name not in columns: conn.execute(f"ALTER TABLE async_delegations ADD COLUMN {name} {sql_type}") @@ -204,12 +215,13 @@ 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) - VALUES (?, ?, ?, ?, 'running', ?, ?, 'pending', 0, ?, ?, ?)""", + owner_started_at, task_json, origin_session_id) + 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)), + owner_started_at, json.dumps(task_payload), + record.get("origin_session_id", "")), ) _prune_durable_records() @@ -290,11 +302,12 @@ 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 + owner_started_at, task_json, origin_session_id 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 = row + (delegation_id, session_key, origin_ui, parent_id, dispatched_at, + pid, started, task_json, origin_session_id) = row live = False if pid: live = _pid_exists(int(pid)) @@ -306,6 +319,9 @@ def recover_abandoned_delegations() -> int: 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 "", "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"), @@ -482,7 +498,8 @@ def get_durable_delegation(delegation_id: str) -> Optional[Dict[str, Any]]: with _DB_LOCK, _transaction() as conn: row = conn.execute( """SELECT origin_session, state, dispatched_at, completed_at, - result_json, delivery_state, delivery_attempts + result_json, delivery_state, delivery_attempts, + origin_session_id FROM async_delegations WHERE delegation_id=?""", (delegation_id,), ).fetchone() if row is None: @@ -492,6 +509,7 @@ def get_durable_delegation(delegation_id: str) -> Optional[Dict[str, Any]]: "dispatched_at": row[2], "completed_at": row[3], "result": json.loads(row[4]) if row[4] else None, "delivery_state": row[5], "delivery_attempts": row[6], + "origin_session_id": row[7] or "", } @@ -575,6 +593,34 @@ def _prune_completed_locked() -> None: _records.pop(rid, None) +def _current_origin_session_id() -> str: + """Raw session id of the ORIGINATING api_server request, or ``""``. + + The obvious source — ``HERMES_SESSION_ID`` via ``get_session_env`` — is + NOT safe to read at dispatch time: constructing a child agent + (``agent/agent_init.py``) calls ``set_current_session_id(child.session_id)``, + clobbering that ContextVar *and* ``os.environ`` with the subagent's + internal ``{timestamp}_{uuid}`` id moments before the dispatch code reads + it, so the completion wake would self-post into the subagent's own + (unread) session instead of the spawner's. + + The request-scoped ``HERMES_SESSION_CHAT_ID`` binding survives child + construction: ``_bind_api_server_session`` binds ``chat_id`` to the raw + ``X-Hermes-Session-Id``, and its only writer is ``set_session_vars`` — + ``set_current_session_id`` never touches it. Gate on the platform: on + push platforms ``chat_id`` is a chat, not a session, so yield ``""`` + there. + """ + try: + from gateway.session_context import get_session_env + + if get_session_env("HERMES_SESSION_PLATFORM", "") != "api_server": + return "" + return get_session_env("HERMES_SESSION_CHAT_ID", "") or "" + except Exception: + return "" + + def dispatch_async_delegation( *, goal: str, @@ -586,6 +632,7 @@ def dispatch_async_delegation( parent_session_id: Optional[str] = None, runner: Callable[[], Dict[str, Any]], origin_ui_session_id: str = "", + origin_session_id: str = "", interrupt_fn: Optional[Callable[[], None]] = None, max_async_children: int = _DEFAULT_MAX_ASYNC_CHILDREN, progress_fn: Optional[Callable[[], tuple]] = None, @@ -643,6 +690,7 @@ def dispatch_async_delegation( "model": model, "session_key": session_key, "origin_ui_session_id": origin_ui_session_id, + "origin_session_id": origin_session_id, "parent_session_id": parent_session_id, "status": "running", "dispatched_at": dispatched_at, @@ -789,6 +837,7 @@ def _push_completion_event( # session; empty string => CLI (single-session) path. "session_key": record.get("session_key", ""), "origin_ui_session_id": record.get("origin_ui_session_id", ""), + "origin_session_id": record.get("origin_session_id", ""), "parent_session_id": record.get("parent_session_id"), "goal": record.get("goal", ""), "context": record.get("context"), @@ -838,6 +887,7 @@ def dispatch_async_delegation_batch( parent_session_id: Optional[str] = None, runner: Callable[[], Dict[str, Any]], origin_ui_session_id: str = "", + origin_session_id: str = "", interrupt_fn: Optional[Callable[[], None]] = None, max_async_children: int = _DEFAULT_MAX_ASYNC_CHILDREN, delegation_id: Optional[str] = None, @@ -880,6 +930,7 @@ def dispatch_async_delegation_batch( "model": model, "session_key": session_key, "origin_ui_session_id": origin_ui_session_id, + "origin_session_id": origin_session_id, "parent_session_id": parent_session_id, "status": "running", "dispatched_at": dispatched_at, @@ -991,6 +1042,7 @@ def _push_batch_completion_event( "delegation_id": event_record.get("delegation_id"), "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", ""), "parent_session_id": event_record.get("parent_session_id"), "goal": event_record.get("goal", ""), "goals": event_record.get("goals"),