fix(delegation): keep nested work on its coordinator route

hermes_state_sessions.py

@@ -230,6 +230,15 @@ def _inherit_col_sql(col: str, extra: str = "") -> str:
     ))
     + "\n                     WHERE id = ? AND parent_session_id IS NOT NULL"
 )
+# A delegate/branch fork of a row that happens to have ended on compression is still not that
+# conversation's continuation. Markers are matched against the QUERIED parent id rather than mere
+# presence, for the same reason as _NON_CONTINUATION_CHILD_FILTER_SQL: a continuation inherits its
+# parent's model_config verbatim, so presence-matching would misclassify it as a delegate.
+_FORK_EDGE_EXCLUSION_SQL = "".join(
+    f"\n                       AND COALESCE({_sql_json_extract('model_config', f'$.{marker}')}, '')"
+    "\n                           != parent_session_id"
+    for marker in ("_delegate_from", "_branched_from")
+)
 _INHERIT_PARENT_ROUTING_SQL = (
     "UPDATE sessions\n                       SET "
     + _INHERIT_SEP.join(_inherit_col_sql(c) for c in (
@@ -241,6 +250,7 @@ def _inherit_col_sql(col: str, extra: str = "") -> str:
     "                           WHERE p.id = sessions.parent_session_id\n"
     "                             AND p.end_reason = 'compression'\n"
     "                       )"
+    + _FORK_EDGE_EXCLUSION_SQL
 )
 
 
@@ -269,7 +279,10 @@ def _inherit_parent_session_metadata(conn, session_id: str) -> None:
         """NULL-fill a child's cwd/git/profile from its parent (profile_name only within the same
         ``agent:<ns>:`` namespace). Gateway routing columns are inherited ONLY by compression forks
         (a crash before the gateway re-records the peer would strand the child unroutable); delegate
-        children must NOT inherit them (peer recovery could repoint traffic into a subagent's session)."""
+        and branch children must NOT inherit them (peer recovery could repoint traffic into a
+        subagent's session), including when their parent row itself ended on compression — a long
+        batch outlives its coordinator's rotation, and two live rows holding one routing key is the
+        shape reported in #92859."""
         conn.execute(_INHERIT_PARENT_META_SQL, (session_id,))
         conn.execute(_INHERIT_PARENT_ROUTING_SQL, (session_id,))
 

tests/state/test_delegate_child_routing_inheritance.py

@@ -0,0 +1,56 @@
+"""A delegate/branch child must never inherit its parent's gateway routing columns.
+
+``_inherit_parent_session_metadata`` states the rule in its own docstring — routing columns are
+inherited ONLY across a compression fork, because "peer recovery could repoint traffic into a
+subagent's session" — but the SQL gated on the PARENT's ``end_reason`` alone.  A delegate child
+whose gateway parent had already rotated on compression (long batch, queued child, detached unit)
+therefore took the chat's ``session_key``/``chat_id``/``user_id``, leaving two live rows holding one
+routing key (NousResearch/hermes-agent#92859).
+
+Real ``SessionDB`` on a temp path, no mocks: the contract asserted here is the RELATIONSHIP between
+the two child kinds — a compression continuation keeps inheriting, a delegate/branch fork does not.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from hermes_state import SessionDB
+
+ROUTING_COLUMNS = ("session_key", "chat_id", "chat_type", "thread_id", "user_id")
+
+
+@pytest.fixture()
+def db(tmp_path):
+    session_db = SessionDB(db_path=tmp_path / "state.db")
+    try:
+        yield session_db
+    finally:
+        session_db.close()
+
+
+def _compressed_gateway_parent(db: SessionDB) -> dict:
+    """A gateway conversation that rotated on compression, i.e. the one case that inherits."""
+    db.create_session(
+        "parent", source="telegram", session_key="agent:main:telegram:dm:42",
+        chat_id="42", chat_type="dm", thread_id="7", user_id="u1",
+    )
+    db.end_session("parent", "compression")
+    return db.get_session("parent")
+
+
+@pytest.mark.parametrize("marker", ["_delegate_from", "_branched_from"])
+def test_delegate_and_branch_children_do_not_take_over_the_parent_route(db: SessionDB, marker: str) -> None:
+    parent = _compressed_gateway_parent(db)
+
+    db.create_session(
+        "worker", source="subagent", parent_session_id="parent", model_config={marker: "parent"},
+    )
+    db.create_session("continuation", source="telegram", parent_session_id="parent")
+
+    worker, continuation = db.get_session("worker"), db.get_session("continuation")
+    for column in ROUTING_COLUMNS:
+        assert parent[column], f"fixture must seed {column}"
+        # The continuation IS the conversation; the worker is an internal transcript.
+        assert continuation[column] == parent[column], column
+        assert worker[column] is None, column

tests/tools/test_nested_delegation_inline_fallback.py

@@ -0,0 +1,89 @@
+"""A delegated worker's own ``delegate_task(background=True)`` must not ride its coordinator's route.
+
+A subagent inherits the spawning chat's session context, so ``_resolve_async_wake_sid`` used to hand
+its nested batch to the async registry: the detached completion is then pushed at the COORDINATOR's
+route while the event is stamped with the worker's internal session id.  The worker itself is gone by
+then — nothing returns the nested result to the turn that asked for it.
+
+Merged NousResearch/hermes-agent#103486 states the intended shape: "A subagent that fans out
+(depth >= 1) calls ``delegate_task`` synchronously by design, since it needs its workers' results
+inside its own turn."  Contract asserted here: the SAME batch dispatches detached from a normal
+session and runs inline from inside ``delegated_child_context`` — only the child worker (which would
+need a live model) is stubbed.
+"""
+
+from __future__ import annotations
+
+import json
+import time
+from types import SimpleNamespace
+
+import pytest
+
+from agent.delegation_context import delegated_child_context
+from gateway.session_context import clear_session_vars, set_session_vars
+from tools import delegate_tool_dispatch as dispatch
+
+
+@pytest.fixture()
+def gateway_session():
+    """Bind a messaging session that CAN receive a detached completion (the coordinator's)."""
+    tokens = set_session_vars(
+        platform="telegram", session_id="coordinator", session_key="agent:main:telegram:dm:42",
+        async_delivery=True,
+    )
+    try:
+        yield
+    finally:
+        clear_session_vars(tokens)
+
+
+@pytest.fixture()
+def batch(monkeypatch):
+    """A one-task batch whose child run is stubbed (no model call), everything else real."""
+    from tools import delegate_tool
+
+    task = {"goal": "nested work"}
+    child = SimpleNamespace(session_id="worker-child")
+    monkeypatch.setattr(
+        delegate_tool, "_run_single_child",
+        lambda **kwargs: {"task_index": 0, "status": "completed", "summary": "nested result"},
+    )
+    monkeypatch.setattr(dispatch, "_finalize_child_results", lambda *args: None)
+    return dispatch._Batch(
+        task_list=[task], children=[(0, task, child)],
+        parent_agent=SimpleNamespace(session_id="worker", _delegate_depth=1),
+        creds={"model": "test-model"}, context=None, top_role="leaf", max_children=1,
+        live_deleg_id=None, live_writers=[], live_paths=[],
+        origin_wake_sid="", origin_ui_session_id="", origin_owner_transport=None,
+        origin_owner_session_record=None, origin_session_history_delivery=False,
+        overall_start=time.monotonic(),
+    )
+
+
+def test_background_delegation_detaches_for_a_normal_session(gateway_session, batch, monkeypatch) -> None:
+    dispatched = []
+    monkeypatch.setattr(
+        "tools.async_delegation.dispatch_async_delegation_batch",
+        lambda **kwargs: dispatched.append(kwargs) or {"status": "dispatched", "delegation_id": "deleg_test"},
+    )
+
+    result = json.loads(dispatch._dispatch_background(batch))
+
+    assert dispatched, "a route-owning session still dispatches detached units"
+    assert "results" not in result
+
+
+def test_delegated_worker_gets_its_nested_batch_back_in_its_own_turn(
+    gateway_session, batch, monkeypatch,
+) -> None:
+    def _must_not_dispatch(**kwargs):
+        raise AssertionError("a delegated worker must not detach onto its coordinator's route")
+
+    monkeypatch.setattr("tools.async_delegation.dispatch_async_delegation_batch", _must_not_dispatch)
+
+    with delegated_child_context("worker"):
+        result = json.loads(dispatch._dispatch_background(batch))
+
+    assert [entry["summary"] for entry in result["results"]] == ["nested result"]
+    assert "SYNCHRONOUSLY" in result["note"]

tools/delegate_tool_dispatch.py

@@ -193,7 +193,7 @@ def _execute_and_aggregate(batch: _Batch, *, honor_parent_interrupt: bool = True
     "no_async": (
         "background=true is not available in this session — it cannot "
         "receive a detached subagent result after the turn ends (a "
-        "finite chat using -Q, --oneshot, or non-TTY stdio, `hermes -z`, a cron job, a Kanban "
+        "delegated worker, a finite chat using -Q, --oneshot, or non-TTY stdio, `hermes -z`, a cron job, a Kanban "
         "worker, or a stateless HTTP endpoint). The subagent(s) ran SYNCHRONOUSLY and the result is included above."
     ),
     "at_capacity": (
@@ -223,6 +223,15 @@ def _resolve_async_wake_sid(origin_wake_sid: str, origin_session_history_deliver
     if get_session_env("HERMES_SINGLE_QUERY_SESSION") == "1":
         return None
 
+    # A delegated worker inherits the coordinator's session context but owns no route of its own:
+    # a detached completion would be pushed at the coordinator's chat (or persisted into its
+    # server history) stamped with the worker's internal session id, while the tool call that
+    # asked for the work has already returned a bare handle. Run the nested batch inline so the
+    # result lands in this worker's own turn (#103486: depth >= 1 fans out synchronously by design).
+    from agent.delegation_context import is_delegated_child_context
+    if is_delegated_child_context():
+        return None
+
     try:
         # Finite sessions cannot route a detached subagent result back to the agent after their turn/process
         # ends. This includes stateless HTTP requests (#10760) and one-shot Kanban workers (#63169). Fall