fix(gateway): persist async task origins across restart

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.
 

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