fix(chat): keep manager return delivery on the original host

loopx/capabilities/manager_context/roundtrip.py

@@ -665,11 +665,28 @@ def _resolve_delivery_sender(external_sender):
     return None, False
 
 
+def _belongs_to_chat_host(root, store, path):
+    from ...control_plane.collaboration.source_chat_observation import is_source_chat_host
+
+    route_path = _root(root) / "roundtrips" / (path.parent.name + ".json")
+    try:
+        # Missing legacy routes can only be recovered from the coordination
+        # root's Chat receipts. Private hosts recover their frozen provenance
+        # at startup before draining; a foreign host must never claim them.
+        route = _read(route_path) if route_path.exists() else {}
+        return route.get("kind") != "peer" and is_source_chat_host(root, route, store)
+    except (OSError, ValueError, TypeError):
+        logging.getLogger(__name__).warning("Manager return source provenance unavailable")
+        return False
+
+
 def _drain_exact(root, registry, store, external_sender, *, now, cancelled):
     processed = 0
     for path in iter_result_paths(_root(root) / "replies"):
         if cancelled():
             break
+        if not _belongs_to_chat_host(root, store, path):
+            continue
         state_path = path.with_name(path.stem + ".delivery.json")
         effect_locks = ExitStack()
         try:
@@ -992,6 +1009,8 @@ def drain(root, registry, store, external_sender, *, now=None, cancelled=lambda:
     for path in iter_result_paths(_root(root) / "replies"):
         if cancelled():
             break
+        if not _belongs_to_chat_host(root, store, path):
+            continue
         state_path = path.with_name(path.stem + ".delivery.json")
         with exclusive_file_lock(path.with_suffix(".lock")):
             try:

loopx/control_plane/collaboration/source_chat_observation.py

@@ -11,15 +11,25 @@
 )
 
 
-def observe_source_chat(coordination_root, route):
+def _source_chat_root(coordination_root, route):
     locator = route.get("source_chat_runtime_root", str(Path(coordination_root).resolve()))
     if not isinstance(locator, str) or not locator or not Path(locator).is_absolute():
         raise ValueError("original source Chat store provenance is invalid")
+    return Path(locator).resolve()
+
+
+def is_source_chat_host(coordination_root, route, store):
+    """Match frozen provenance before a host admits or mutates a shared return."""
+    return _source_chat_root(coordination_root, route) == store.root.parent.resolve()
+
+
+def observe_source_chat(coordination_root, route):
+    source_root = _source_chat_root(coordination_root, route)
     session_id = _opaque_id(route.get("session_id"), field="session_id")
     client_turn_id = _opaque_id(route.get("client_turn_id"), field="client_turn_id")
     if session_id in {".", ".."}:
         raise ValueError("original source Chat session identity is invalid")
-    session_dir = Path(locator) / "chat" / "sessions" / session_id
+    session_dir = source_root / "chat" / "sessions" / session_id
     session = _read_json(session_dir / "session.json")
     turns = [value for path in sorted((session_dir / "turns").glob("*.json"))
              if not path.name.endswith(".events.json")

loopx/semantics/project_registry_io_manifest_v1.json

@@ -279,7 +279,7 @@
     },
     {
       "site": "loopx/capabilities/manager_context/roundtrip.py::<module>.drain::codec_read:load_project_registry#1",
-      "line": 979,
+      "line": 996,
       "column": 22,
       "kind": "codec_read",
       "api": "load_project_registry",

tests/test_manager_context_roundtrip.py

@@ -47,15 +47,16 @@ def flow(tmp_path):
     )
     store = ChatSessionStore(tmp_path)
 
-    def create(external=False, project=False, brief=None):
-        session = store.create_session(
+    def create(external=False, project=False, brief=None, source_store=None):
+        actual_store = source_store or store
+        session = actual_store.create_session(
             goal_id="research" if project else "loopx-manager",
             agent_id="codex",
             adapter_kind="codex_app_server",
             upstream_thread_id="test",
             channel_id="goal.research" if project else "manager.external.test" if external else "manager",
         )
-        turn, _ = store.create_turn(
+        turn, _ = actual_store.create_turn(
             session["session_id"],
             client_turn_id="question",
             message="Investigate this new constraint",
@@ -86,9 +87,10 @@ def create(external=False, project=False, brief=None):
             )
         receipt = deliver(
             tmp_path, registry, session=session, turn=turn,
-            request={**target, **({"brief": brief} if brief else {})}
+            request={**target, **({"brief": brief} if brief else {})},
+            source_store=source_store,
         )
-        store.update_turn(
+        actual_store.update_turn(
             session["session_id"],
             turn["turn_id"],
             status="completing",
@@ -97,7 +99,7 @@ def create(external=False, project=False, brief=None):
                 "context_handoff_receipt": receipt,
             },
         )
-        store.finalize_managed_turn_completion(session["session_id"], turn["turn_id"])
+        actual_store.finalize_managed_turn_completion(session["session_id"], turn["turn_id"])
         return session, turn, receipt
 
     return tmp_path, registry, store, create
@@ -141,6 +143,59 @@ def unexpected(*_):
     assert reply_status(root, receipt)[0]["status"] == "delivered"
 
 
+@pytest.mark.parametrize("explicit_source", [False, True], ids=["legacy-root", "pinned-root"])
+def test_other_chat_host_cannot_delay_original_host_return(flow, explicit_source):
+    root, registry, store, create = flow
+    if explicit_source:
+        store = ChatSessionStore(root / "private-chat-host")
+    session, turn, receipt = create(True, source_store=store if explicit_source else None)
+    rid = receipt["request_id"]
+    acknowledge(root, "research", "worker", rid, "adopt", "Checked")
+    report(root, "research", "worker", rid, "conclusion", "The original task is complete.")
+    state_path = _root(root) / "replies" / rid / "conclusion.delivery.json"
+    before = state_path.read_bytes() if state_path.exists() else None
+    foreign = ChatSessionStore(root / "other-chat-host")
+    sent = []
+
+    def sender(route, session, turn, text):
+        sent.append(text)
+        return {"reply_verified": True, "idempotency_key": "sha256:provider-proof"}
+
+    now = datetime.now(timezone.utc)
+    assert drain(root, registry, foreign, sender, now=now) == 0
+    assert (state_path.read_bytes() if state_path.exists() else None) == before
+    assert sent == []
+    # The wrong host cannot impose retry_at and starve the source host's
+    # immediately following tick. A restart must not duplicate the result.
+    assert drain(root, registry, store, sender, now=now) == 1
+    drain(root, registry, ChatSessionStore(store.root.parent), sender, now=now)
+    returned = [m for m in store.messages(session["session_id"])
+                if m.get("origin") == "manager_followup"]
+    assert len(sent) == len(returned) == 1
+    assert returned[0]["turn_id"] == turn["turn_id"]
+    assert reply_status(root, receipt)[0]["status"] == "delivered"
+
+
+def test_exact_return_skips_other_host_before_admission(flow, monkeypatch):
+    from loopx.capabilities.manager_context import roundtrip
+    from loopx.control_plane.projects.registry_codec import SOURCE_SESSION_PROFILE_ID
+
+    root, registry, store, create = flow
+    _, _, receipt = create()
+    acknowledge(root, "research", "worker", receipt["request_id"], "adopt", "Checked")
+    report(root, "research", "worker", receipt["request_id"], "conclusion", "Complete.")
+    profile = json.loads(registry.read_text())
+    profile["profile_id"] = SOURCE_SESSION_PROFILE_ID
+    registry.write_text(json.dumps(profile))
+    foreign = ChatSessionStore(root / "other-chat-host")
+
+    def unexpected(*args, **kwargs):
+        raise AssertionError("A foreign host must not admit or deliver this return")
+
+    monkeypatch.setattr(roundtrip, "_exact_return_context", unexpected)
+    assert drain(root, registry, foreign, unexpected) == 0
+
+
 def test_conclusion_coalesces_unsent_intermediate_decision_and_is_immutable(flow):
     root, registry, store, create = flow
     session, _, r = create()