fix(chat): check bound Session identity before input and recovery

loopx/capabilities/native_chat/conversation_bindings.py

@@ -6,17 +6,20 @@
 """
 from __future__ import annotations
 
-from collections.abc import Callable
+from collections.abc import Callable, Iterator
 from contextlib import contextmanager
 from pathlib import Path
-from typing import Any
+from typing import TYPE_CHECKING, Any
 import uuid
 import hashlib
 
 from ...chat_store import TERMINAL_TURN_STATES, _atomic_write_json, _read_json
 from ...control_plane.effect_runtime import EffectRuntimeRejected, effect_runtime_result
 from ...file_lock import exclusive_file_lock
 
+if TYPE_CHECKING:
+    from ...chat_store import ChatSessionStore
+
 
 class ChatConversationBindings:
     def __init__(self, *, root: Path, project_contexts: Any,
@@ -29,7 +32,7 @@ def __init__(self, *, root: Path, project_contexts: Any,
         self.controller: Any | None = None
 
     def observe(self, transport_ref: str, *, audience: str | None = None) -> dict[str, Any]:
-        row = next((item for item in self.read()["bindings"] if item["transport_ref"] == transport_ref), {})
+        row: dict[str, Any] = next((item for item in self.read()["bindings"] if item["transport_ref"] == transport_ref), {})
         if audience is None:
             audience = row.get("audience", "owner")
         if audience == "group" and self._observe_group is not None:
@@ -118,7 +121,7 @@ def configure(self, *, transport_ref: str, project_ref: str,
             readback = self.read()
             if readback != result["state"]:
                 raise OSError("conversation binding publication did not verify")
-            binding = next(row for row in readback["bindings"] if row["transport_ref"] == transport_ref)
+            binding: dict[str, Any] = next(row for row in readback["bindings"] if row["transport_ref"] == transport_ref)
         if previous and previous["binding_id"] != binding["binding_id"]:
             self._stop_revoked_binding(previous["binding_id"])
         return binding
@@ -180,7 +183,7 @@ def resolve(self, *, binding_id: str, source_ref: str, sender_ref: str,
                 private_human_message: bool, session_context: dict[str, Any] | None = None,
                 group_ref: str | None = None, topic_ref: str | None = None,
                 group_human_message: bool = False, include_observation: bool = False,
-                origin: str | None = None) -> dict[str, Any]:
+                origin: str | None = None, session: dict[str, Any] | None = None) -> dict[str, Any]:
         current = self.read()
         row = next((item for item in current.get("bindings", []) if item.get("binding_id") == binding_id), None)
         if row is None:
@@ -197,6 +200,7 @@ def resolve(self, *, binding_id: str, source_ref: str, sender_ref: str,
             "source_ref": source_ref, "sender_ref": sender_ref, "private_human_message": private_human_message,
             **({"origin": origin} if origin is not None else {}),
             **({"session_context": session_context} if session_context is not None else {}),
+            **({"session": self._session_observation(session)} if session is not None else {}),
             **({"group_ref": group_ref, "topic_ref": topic_ref,
                 "group_human_message": group_human_message} if group_ref is not None or topic_ref is not None or group_human_message else {}),
         })
@@ -205,15 +209,17 @@ def resolve(self, *, binding_id: str, source_ref: str, sender_ref: str,
         # or a substitute for the next admission/delivery owner's fresh check.
         return {**selected, "observation": observation} if include_observation else selected
 
-    def _resolve_observed(self, current, row, observation, source):
+    def _resolve_observed(self, current: dict[str, Any], row: dict[str, Any],
+                          observation: dict[str, Any], source: dict[str, Any]) -> dict[str, Any]:
         return self._core("collaboration.conversation.bound_context", {
             "current": current, "binding_id": row["binding_id"], "observation": observation,
             "available_projects": self.projects.available(), **source,
             **({"available_goal_ids": self.goal_scope_ids(row)} if row.get("goal_scope") == "all_registered" else {}),
         })
 
     @contextmanager
-    def input_context(self, saved: dict[str, Any], *, origin: str):
+    def input_context(self, saved: dict[str, Any], *, origin: str, session: dict[str, Any],
+                      store: ChatSessionStore) -> Iterator[dict[str, Any]]:
         """Observe outside the publication fence, then retain it through enqueue.
 
         Saved fields select an identity; fresh Core checks grant it. Local Web
@@ -226,13 +232,25 @@ def input_context(self, saved: dict[str, Any], *, origin: str):
             source = self._session_source(saved)
         else:
             raise ValueError("unsupported bound conversation input origin")
-        observed = self.resolve(binding_id=saved["binding_id"], include_observation=True, **source)
+        observed = self.resolve(binding_id=saved["binding_id"], include_observation=True, session=session, **source)
         with exclusive_file_lock(self.path, operation="admit_bound_input"):
             current = self.read()
             if next((row for row in current.get("bindings", [])
                      if row.get("binding_id") == saved["binding_id"]), None) != observed["binding"]:
                 raise ValueError("conversation binding changed before input admission")
-            yield self._resolve_observed(current, observed["binding"], observed["observation"], source)
+            # The provider observation may have waited. Re-read canonical
+            # identity under the binding fence before Core authorizes input.
+            fresh = store.load_session(session["session_id"])
+            if fresh is None:
+                raise KeyError("chat session was not found")
+            yield self._resolve_observed(current, observed["binding"], observed["observation"],
+                {**source, "session": self._session_observation(fresh)})
+
+    @staticmethod
+    def _session_observation(session: dict[str, Any]) -> dict[str, Any]:
+        return {key: session.get(key) for key in (
+            "session_id", "goal_id", "agent_id", "executor_endpoint_id", "channel_id",
+            "session_mode", "status", "project_context", "steward_context")}
 
     def local_sessions(self, binding_id: str) -> list[dict[str, Any]]:
         """Read original targets; metadata hints never replace Core's grant.
@@ -263,10 +281,7 @@ def local_sessions(self, binding_id: str) -> list[dict[str, Any]]:
             result = self._core("collaboration.conversation.local_sessions", {
                 "current": current, "binding_id": binding_id, "observation": observation,
                 "available_projects": self.projects.available(),
-                "sessions": [{key: session.get(key) for key in (
-                    "session_id", "goal_id", "agent_id", "executor_endpoint_id", "channel_id",
-                    "session_mode", "status", "project_context", "steward_context")}
-                    for session in sessions],
+                "sessions": [self._session_observation(session) for session in sessions],
                 **({"available_goal_ids": self.goal_scope_ids(row)} if row.get("goal_scope") == "all_registered" else {}),
             })
             # Build the lookup only after Core has rejected duplicate identity;
@@ -277,7 +292,7 @@ def local_sessions(self, binding_id: str) -> list[dict[str, Any]]:
     def goal_scope_ids(self, binding: dict[str, Any]) -> list[str]:
         if binding.get("goal_scope") != "all_registered":
             return list(binding.get("goal_ids", []))
-        from ...registry import load_registry
+        from ...control_plane.projects.registry_codec import load_registry
         if self.controller is None or self.controller.registry_path is None:
             raise ValueError("the steward registry is unavailable")
         return [goal["id"] for goal in load_registry(self.controller.registry_path).get("goals", [])]
@@ -315,8 +330,8 @@ def _session_source(saved: dict[str, Any]) -> dict[str, Any]:
         return {"source_ref": saved["source_ref"], "sender_ref": saved["operator_ref"],
                 "private_human_message": True, "session_context": saved}
 
-    def session_context(self, saved: dict[str, Any]) -> dict[str, Any]:
-        return self.resolve(binding_id=saved["binding_id"], **self._session_source(saved))
+    def session_context(self, saved: dict[str, Any], *, session: dict[str, Any] | None = None) -> dict[str, Any]:
+        return self.resolve(binding_id=saved["binding_id"], session=session, **self._session_source(saved))
 
     def manager_runtime_owner(self, saved: dict[str, Any], channel_id: str) -> bool:
         """Fresh Core audience proof; saved Session text is never authority."""
@@ -356,7 +371,8 @@ def agent_observation(self, session_id: str) -> dict[str, Any]:
         No host history, remote thread database or prior identity is discovered.
         """
         from ...attached_session import require_current_attached_session, _require_bound_host
-        from ...registry import find_registry_goal, load_registry
+        from ...control_plane.projects.registry_codec import load_registry
+        from ...registry import find_registry_goal
         if self.controller is None or self.controller.registry_path is None:
             raise ValueError("the attached Session registry is unavailable")
         session = require_current_attached_session(store=self.controller.store,
@@ -421,11 +437,14 @@ def change_agent_target(self, *, binding_id: str, expected_revision: int,
             result = self._core("collaboration.conversation.binding", params)
             if result["changed"]:
                 if session_id is not None:
+                    # agent_observation already requires this exact local controller.
+                    assert self.controller is not None
                     self.controller.store.update_session(session_id, external_conversation_binding_id=binding_id)
                 _atomic_write_json(self.path, result["state"])
             if self.read() != result["state"]:
                 raise OSError("Agent target grant publication did not verify")
-            return result["state"]
+            state: dict[str, Any] = result["state"]
+            return state
 
     def resolve_agent_target(self, selected: dict[str, Any], target: dict[str, Any]) -> dict[str, Any]:
         return self._core("collaboration.conversation.agent_target", {"binding": selected["binding"],

loopx/capabilities/native_chat/project_context.py

@@ -9,11 +9,14 @@
 
 import hashlib
 from pathlib import Path
-from typing import Any, Callable, Mapping
+from typing import TYPE_CHECKING, Any, Callable, Mapping
 
 from ...control_plane.effect_runtime import EffectRuntimeRejected, effect_runtime_result
 from ...presentation.answer_instruction import conversation_answer_instruction
 
+if TYPE_CHECKING:
+    from ...chat_store import ChatSessionStore
+
 
 PROJECT_CONVERSATION_OBJECTIVE = (
     "Have an ordinary conversation about the selected workspace. Preserve this "
@@ -62,10 +65,11 @@ def available(self) -> list[dict[str, str]]:
 
     def resolve(self, project_ref: str, *, session_context: dict[str, Any] | None = None) -> dict[str, Any]:
         try:
-            return effect_runtime_result("collaboration.project.context", {
+            result: dict[str, Any] = effect_runtime_result("collaboration.project.context", {
                 "project_ref": project_ref, "available": self.available(),
                 **({"session_context": session_context} if session_context is not None else {}),
             })
+            return result
         except EffectRuntimeRejected as exc:
             raise ValueError(str(exc)) from exc
 
@@ -74,15 +78,15 @@ def session_context(self, session: dict[str, Any]) -> dict[str, Any]:
         if isinstance(steward, dict):
             if self.conversation_bindings is None:
                 raise ValueError("bound steward authority is unavailable")
-            selected = self.conversation_bindings.session_context(steward)
+            selected = self.conversation_bindings.session_context(steward, session=session)
             return self.runtime_context(session, selected)
         saved = session.get("project_context")
         if not isinstance(saved, dict) or session.get("goal_id") is not None:
             raise ValueError("invalid ordinary project Session")
         if saved.get("audience") in {"bound_owner", "bound_group"}:
             if self.conversation_bindings is None:
                 raise ValueError("bound project conversation authority is unavailable")
-            selected = self.conversation_bindings.session_context(saved)
+            selected = self.conversation_bindings.session_context(saved, session=session)
         else:
             selected = self.resolve(str(saved.get("project_ref") or ""), session_context=saved)
         return self.runtime_context(session, selected)
@@ -115,10 +119,13 @@ def open_bound(self, binding_id: str, source: dict[str, Any], *, executor: str,
         session = {"goal_id": "loopx-manager" if steward else None, "channel_id": selected["channel_id"],
                    "project_context": None if steward else selected["context"],
                    "steward_context": selected["context"] if steward else None}
-        return {**session, **self.session_context(session)}
+        # This is a construction plan, not a canonical Session yet. Subsequent
+        # input/recovery validates the stored identity against a fresh binding.
+        return {**session, **self.runtime_context(session, selected)}
 
     @staticmethod
-    def initialize_bound_scope(store, session, *, runtime_root=None):
+    def initialize_bound_scope(store: ChatSessionStore, session: dict[str, Any], *,
+                               runtime_root: Path | None = None) -> dict[str, Any]:
         steward = session.get("steward_context")
         if not steward:
             return session

loopx/capabilities/native_chat/queued_admission.py

@@ -8,9 +8,9 @@
 from . import project_context as project_context_policy
 
 
-def enqueue_turn(controller, *, session_id: str, client_turn_id: str, message: str,
+def enqueue_turn(controller: Any, *, session_id: str, client_turn_id: str, message: str,
                  attachments: list[dict[str, Any]] | None, work_dir: Path | None, objective: str,
-                 origin: str, external_agent_target: dict | None):
+                 origin: str, external_agent_target: dict[str, Any] | None) -> tuple[dict[str, Any], bool]:
     session = controller.store.load_session(session_id)
     if session is None or session.get("status") == "closed":
         raise KeyError("chat session was not found")
@@ -63,13 +63,14 @@ def enqueue_turn(controller, *, session_id: str, client_turn_id: str, message: s
     return turn, created
 
 
-def _enqueue_bound(controller, session, client_turn_id, message, attachments, origin):
+def _enqueue_bound(controller: Any, session: dict[str, Any], client_turn_id: str, message: str,
+                   attachments: list[dict[str, Any]] | None, origin: str) -> tuple[dict[str, Any], bool]:
     contexts = controller.project_contexts
     bindings = contexts.conversation_bindings
     if bindings is None:
         raise ValueError("bound conversation authority is unavailable")
     saved = session.get("steward_context") or session["project_context"]
-    with bindings.input_context(saved, origin=origin) as selected:
+    with bindings.input_context(saved, origin=origin, session=session, store=controller.store) as selected:
         fresh = controller.store.load_session(session["session_id"])
         if fresh is None or fresh.get("status") == "closed":
             raise KeyError("chat session was not found")
@@ -92,7 +93,8 @@ def _enqueue_bound(controller, session, client_turn_id, message, attachments, or
     return turn, created
 
 
-def _record_local_ingress(runtime_root, session, client_turn_id, message):
+def _record_local_ingress(runtime_root: Path, session: dict[str, Any],
+                          client_turn_id: str, message: str) -> None:
     from ..manager_context import register_ingress
     from ...control_plane.collaboration.inbox import _hash, normalize_source_context
 

loopx/control_plane/collaboration/conversation_binding.ts

@@ -334,13 +334,27 @@ export function resolveBoundConversation(params: JsonObject): JsonObject {
       : JSON.stringify(saved) === JSON.stringify(context);
     if (!matches) throw new EffectRuntimeRequestError("bound Session context changed");
   }
+  const channel = row.context_kind === "steward"
+    ? `manager.external.native.${id}.${source}` : `project.external.${id}.${source}`;
+  if (params.session !== undefined) {
+    const session = requireJsonObject(params.session, "canonical bound Session");
+    const steward = row.context_kind === "steward";
+    sessionIdentity(session.session_id, "bound Session identity");
+    if (session.status === "closed" || session.channel_id !== channel
+        || session.goal_id !== (steward ? "loopx-manager" : null)
+        || session.agent_id !== row.executor_endpoint_id || session.executor_endpoint_id !== row.executor_endpoint_id
+        || session.session_mode !== "managed_runtime" || (steward ? session.project_context : session.steward_context) != null
+        || params.session_context === undefined
+        || JSON.stringify(steward ? session.steward_context : session.project_context) !== JSON.stringify(params.session_context)) {
+      throw new EffectRuntimeRequestError("canonical bound Session identity changed");
+    }
+  }
   // This is an audience proof, not a host-tool grant. Only the independently
   // verified owner in this private source may inherit an explicit machine
   // owner grant. Portfolio scope and project write grants do not imply it.
   return {binding: row, context, owner_manager_audience: row.context_kind === "steward"
       && project.filesystem_scope !== "workspace_only",
-    channel_id: row.context_kind === "steward"
-    ? `manager.external.native.${id}.${source}` : `project.external.${id}.${source}`};
+    channel_id: channel};
 }
 
 /** Discover existing private targets from canonical metadata. This is a read
@@ -363,13 +377,9 @@ export function listLocalBoundSessions(params: JsonObject): JsonObject {
         && (value as JsonObject).binding_id === id) || session.status === "closed") continue;
     const steward = row.context_kind === "steward";
     const saved = requireJsonObject(steward ? session.steward_context : session.project_context, "bound Session context");
-    const selected = resolveBoundConversation({...params, source_ref: saved.source_ref,
+    resolveBoundConversation({...params, source_ref: saved.source_ref,
+      session,
       sender_ref: row.operator_ref, private_human_message: false, origin: "web", session_context: saved});
-    if (session.channel_id !== selected.channel_id || session.goal_id !== (steward ? "loopx-manager" : null)
-        || session.agent_id !== row.executor_endpoint_id || session.executor_endpoint_id !== row.executor_endpoint_id
-        || session.session_mode !== "managed_runtime" || (steward ? session.project_context : session.steward_context) != null) {
-      throw new EffectRuntimeRequestError("canonical bound Session identity changed");
-    }
     ids.push(sessionIdentity(session.session_id, "bound Session identity"));
   }
   if (new Set(ids).size !== ids.length) throw new EffectRuntimeRequestError("canonical bound Session identity is ambiguous");