diff --git a/loopx/capabilities/native_chat/conversation_bindings.py b/loopx/capabilities/native_chat/conversation_bindings.py index 128dc6557f..27e55a417f 100644 --- a/loopx/capabilities/native_chat/conversation_bindings.py +++ b/loopx/capabilities/native_chat/conversation_bindings.py @@ -6,10 +6,10 @@ """ 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 @@ -17,6 +17,9 @@ 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,7 +209,8 @@ 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, @@ -213,7 +218,8 @@ def _resolve_observed(self, current, row, observation, source): }) @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"], diff --git a/loopx/capabilities/native_chat/project_context.py b/loopx/capabilities/native_chat/project_context.py index 06c17f9afa..fe56cadb3d 100644 --- a/loopx/capabilities/native_chat/project_context.py +++ b/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,7 +78,7 @@ 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: @@ -82,7 +86,7 @@ def session_context(self, session: dict[str, Any]) -> dict[str, Any]: 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 diff --git a/loopx/capabilities/native_chat/queued_admission.py b/loopx/capabilities/native_chat/queued_admission.py index 17a90fb6a7..2a2b25ef0b 100644 --- a/loopx/capabilities/native_chat/queued_admission.py +++ b/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 diff --git a/loopx/control_plane/collaboration/conversation_binding.ts b/loopx/control_plane/collaboration/conversation_binding.ts index 7db60a86be..8a88ade53a 100644 --- a/loopx/control_plane/collaboration/conversation_binding.ts +++ b/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"); diff --git a/tests/control_plane_ts/conversation_binding.test.ts b/tests/control_plane_ts/conversation_binding.test.ts index c982083440..12c40b3eb2 100644 --- a/tests/control_plane_ts/conversation_binding.test.ts +++ b/tests/control_plane_ts/conversation_binding.test.ts @@ -27,6 +27,8 @@ for (const kind of ["project", "steward"]) { agent_id: "codex", executor_endpoint_id: "codex", session_mode: "managed_runtime", status: "ready", channel_id: `${kind === "steward" ? "manager.external.native" : "project.external"}.${row.binding_id}.${"e".repeat(24)}`, ...(kind === "steward" ? {steward_context: context} : {project_context: context})}; + const continuation = {...lookup, source_ref: "e".repeat(24), sender_ref: row.operator_ref, + private_human_message: false, origin: "web", session_context: context}; test(`${kind} discovery reuses verified saved identity and excludes unrelated or closed Sessions`, () => { const snapshots = [session, {...session, session_id: "historical", status: "closed"}, @@ -36,6 +38,9 @@ for (const kind of ["project", "steward"]) { assert.deepEqual(snapshots, before); assert.deepEqual(listLocalBoundSessions({...lookup, sessions: []}), {session_ids: []}); assert.throws(() => listLocalBoundSessions({...lookup, sessions: [session, session]}), /ambiguous/); + assert.deepEqual(resolveBoundConversation({...continuation, session}).context, context); + assert.throws(() => resolveBoundConversation({...continuation, session: {...session, status: "closed"}}), + /canonical bound Session identity/); }); test(`${kind} discovery refuses changed canonical role, executor, channel or saved source`, () => { @@ -47,6 +52,7 @@ for (const kind of ["project", "steward"]) { {[kind === "steward" ? "steward_context" : "project_context"]: {...context as Record, source_ref: "f".repeat(24)}}]) { assert.throws(() => listLocalBoundSessions({...lookup, sessions: [{...session, ...change}]})); + assert.throws(() => resolveBoundConversation({...continuation, session: {...session, ...change}})); } for (const change of [{current}, {available_projects: []}, {observation: {...observation, verified: false}}, diff --git a/tests/test_chat_conversation_bindings.py b/tests/test_chat_conversation_bindings.py index 36c160f242..551bc8cca2 100644 --- a/tests/test_chat_conversation_bindings.py +++ b/tests/test_chat_conversation_bindings.py @@ -438,6 +438,62 @@ def _bound_project(fixture): return bindings, args, binding, session, turn +@pytest.mark.parametrize("entrypoint", ["http", "runtime_context"]) +def test_bound_session_identity_is_checked_before_input_or_runtime_resolution(ordinary, monkeypatch, entrypoint): # noqa: F811 + from loopx.capabilities.native_chat.external_conversations import ChatExternalConversations + + store, runtime, contexts, request, capture, _, workspace = ordinary + observation = {"transport_ref": "notes-app", "provider_ref": "c" * 24, + "operator_ref": "d" * 24, "verified": True} + bindings = ChatConversationBindings(root=store.root, project_contexts=contexts, + observe=lambda _: observation) + contexts.conversation_bindings = bindings + ChatExternalConversations(runtime) + binding = bindings.configure(transport_ref="notes-app", + project_ref=contexts.available()[0]["project_ref"], executor_endpoint_id="codex") + session, _ = runtime.open_session(goal_id=None, agent_id="codex", work_dir=workspace, + objective="", mode="resume_latest", conversation_binding_id=binding["binding_id"], + source_context={"source_ref": "a" * 24, "sender_ref": observation["operator_ref"], + "private_human_message": True}) + # The canonical store accepts an executor endpoint separately from its + # agent id. A retained Session must still match the current binding. + changed = store.create_session(goal_id=None, agent_id="codex", + executor_endpoint_id="kiro-cli", adapter_kind="codex_app_server", + upstream_thread_id="retained-thread", channel_id=session["channel_id"], + project_context=session["project_context"]) + sid = changed["session_id"] + with pytest.raises(ValueError, match="canonical bound Session identity"): + bindings.local_sessions(binding["binding_id"]) + before_messages = store.messages(sid) + before_requests = capture.read_text() if capture.exists() else "" + resume = runtime.resume_session_queue + dispatches = [] + monkeypatch.setattr(runtime, "resume_session_queue", lambda **args: dispatches.append(args)) + if entrypoint == "http": + status, _ = request(f"/api/chat/sessions/{sid}/turns", + {"message": "Continue here", "client_turn_id": "wrong-recipient"}) + assert status == 400 + else: + with pytest.raises(ValueError, match="canonical bound Session identity"): + contexts.session_context(changed) + assert store.turn_for_client(sid, "wrong-recipient") is None + assert dispatches == [] + assert store.messages(sid) == before_messages + assert (capture.read_text() if capture.exists() else "") == before_requests + # Select the original valid recipient to recover through the same shipped + # HTTP path. No replacement conversation or Goal registration is needed. + monkeypatch.setattr(runtime, "resume_session_queue", resume) + original = session["session_id"] + body = {"message": "Continue here", "client_turn_id": "original-recipient"} + status, accepted = request(f"/api/chat/sessions/{original}/turns", body) + assert status == 202 and accepted["created"] is True + assert runtime.wait_for_turn(session_id=original, + turn_id=accepted["turn_id"], timeout_sec=10)["status"] == "completed" + status, replay = request(f"/api/chat/sessions/{original}/turns", body) + assert status == 202 and replay["created"] is False and replay["turn_id"] == accepted["turn_id"] + assert store.load_session(original)["goal_id"] is None + + def test_binding_reconfiguration_keeps_same_grant_running_but_stops_replaced_grant(ordinary): # noqa: F811 store, runtime, contexts, _, _, _, workspace = ordinary bindings, args, binding, session, turn = _bound_project(ordinary)