diff --git a/apps/presentation/dashboard/visual-evidence/collaboration-source-runtime/after-desktop.png b/apps/presentation/dashboard/visual-evidence/collaboration-source-runtime/after-desktop.png new file mode 100644 index 0000000000..24e4d05249 Binary files /dev/null and b/apps/presentation/dashboard/visual-evidence/collaboration-source-runtime/after-desktop.png differ diff --git a/apps/presentation/dashboard/visual-evidence/collaboration-source-runtime/after-narrow.png b/apps/presentation/dashboard/visual-evidence/collaboration-source-runtime/after-narrow.png new file mode 100644 index 0000000000..bbf07372e6 Binary files /dev/null and b/apps/presentation/dashboard/visual-evidence/collaboration-source-runtime/after-narrow.png differ diff --git a/apps/presentation/dashboard/visual-evidence/collaboration-source-runtime/before-desktop.png b/apps/presentation/dashboard/visual-evidence/collaboration-source-runtime/before-desktop.png new file mode 100644 index 0000000000..d69c9a836f Binary files /dev/null and b/apps/presentation/dashboard/visual-evidence/collaboration-source-runtime/before-desktop.png differ diff --git a/loopx/capabilities/manager_context/presentation.py b/loopx/capabilities/manager_context/presentation.py index b296bf9761..bc6c01ef82 100644 --- a/loopx/capabilities/manager_context/presentation.py +++ b/loopx/capabilities/manager_context/presentation.py @@ -3,7 +3,7 @@ from . import _read, _root, _hash from .tracking import _entry, _receipt from .roundtrip import reply_status -from ...control_plane.collaboration.source_chat_observation import original_delivery_facts +from ...control_plane.collaboration.source_chat_observation import original_delivery_facts, is_source_chat_host from ...chat import redact_local_paths from ...control_plane.collaboration import conversation_scope from ...control_plane.collaboration.goal_instance_scope import ( @@ -52,7 +52,8 @@ def project_collaboration(store, root, session_id, messages, *, registry, contex try: route = _read(path) client = route.get("client_turn_id") - if route.get("session_id") != session_id or client not in missing_clients: + if (route.get("session_id") != session_id or client not in missing_clients + or not is_source_chat_host(root, route, store)): continue recovered.setdefault(client, []).append(route) except (OSError, ValueError, TypeError): @@ -87,6 +88,9 @@ def project_collaboration(store, root, session_id, messages, *, registry, contex scope=scope, ) route = _read(_root(root) / "roundtrips" / (row["request_id"] + ".json")) + if not is_source_chat_host(root, route, store): + result.append(message) + continue lifetime = decide_collaboration_lifecycle( scope, operation="original_request_inspect", record=row, route=route, initial_delivery=original_delivery_facts(row, route, turn, source_id=row["source_id"]), diff --git a/loopx/capabilities/manager_context/roundtrip.py b/loopx/capabilities/manager_context/roundtrip.py index 9308e9ea8b..c0361c60eb 100644 --- a/loopx/capabilities/manager_context/roundtrip.py +++ b/loopx/capabilities/manager_context/roundtrip.py @@ -237,9 +237,11 @@ def report( } -def project_chat_return_deliveries(root, session_id, messages): +def project_chat_return_deliveries(root, session_id, messages, *, source_store=None): """Attach public-safe delivery readback to returned Chat transcript rows.""" + from ...control_plane.collaboration.source_chat_observation import is_source_chat_host + pending_message_ids = { message_id for message in messages @@ -254,7 +256,8 @@ def project_chat_return_deliveries(root, session_id, messages): for path in paths: try: route = _read(path) - if route.get("session_id") != session_id: + if (route.get("session_id") != session_id + or (source_store is not None and not is_source_chat_host(root, route, source_store))): continue request_id = str(route.get("request_id") or "") if path.stem != request_id or not BARE_SHA256_PATTERN.fullmatch(request_id): @@ -296,7 +299,7 @@ def project_chat_session_snapshot(root, store, session_id, *, registry, context_messages = context_messages[:page["message_page"]["total"]] snapshot["messages"] = page["messages"] snapshot["messages"] = project_chat_return_deliveries( - root, session_id, snapshot["messages"] + root, session_id, snapshot["messages"], source_store=store ) from .presentation import project_collaboration # Use the frozen context to choose the carrier, projecting only this page. diff --git a/loopx/chat_server.py b/loopx/chat_server.py index 717f0f04a0..d625be9824 100644 --- a/loopx/chat_server.py +++ b/loopx/chat_server.py @@ -934,7 +934,7 @@ def _session_snapshot(self, session_id: str) -> None: self.server.runtime_controller.recover_exited_turn(session_id) snapshot = cast(Callable[..., dict[str, object]], project_chat_session_snapshot) self._send_json(snapshot( - self.server.runtime_root, self.server.chat_store, session_id, + self.server.runtime_controller.coordination_runtime_root, self.server.chat_store, session_id, registry=self.server.registry_path, message_limit=message_limit, message_cursor=query.get("message_cursor", [None])[0])) except KeyError: diff --git a/tests/control_plane/test_chat_collaboration_source_runtime.py b/tests/control_plane/test_chat_collaboration_source_runtime.py new file mode 100644 index 0000000000..b5712c8102 --- /dev/null +++ b/tests/control_plane/test_chat_collaboration_source_runtime.py @@ -0,0 +1,152 @@ +"""Original Chat reads shared receipts without borrowing another Chat host.""" +# ruff: noqa: F811 - pytest injects the imported isolated provider fixture. +import http.client +import json +import shutil +import threading +from contextlib import contextmanager + +import pytest + +from test_local_delegation import service # noqa: F401 - real File/SQLite authority + +from loopx.capabilities.manager_context import _root, acknowledge, deliver +from loopx.capabilities.manager_context.roundtrip import drain, report, reply_status +from loopx.capabilities.native_chat.project_context import ChatProjectContexts +from loopx.chat_runtime import ChatRuntimeController +from loopx.chat_server import ChatHTTPServer, ChatRequestHandler +from loopx.chat_store import ChatSessionStore + + +@contextmanager +def session_api(store, registry, workspace): + controller = ChatRuntimeController( + store=store, registry_path=registry, + project_contexts=ChatProjectContexts([workspace]), + codex_bin=str(workspace / "no-model-executable"), + ) + server = ChatHTTPServer(("127.0.0.1", 0), ChatRequestHandler) + server.chat_store, server.runtime_controller = store, controller + server.runtime_root, server.registry_path = store.root.parent, registry + server.selected_goal_id, server.verbose = None, False + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + + def read(session_id, query=""): + connection = http.client.HTTPConnection(*server.server_address, timeout=15) + try: + connection.request("GET", f"/api/chat/sessions/{session_id}" + query) + response = connection.getresponse() + payload = json.loads(response.read()) + assert response.status == 200, payload + return payload + finally: + connection.close() + + try: + yield read, controller + finally: + server.shutdown() + server.server_close() + thread.join(5) + + +@pytest.fixture(params=[False, True], ids=["colocated", "separate-chat"]) +def returned(service, request): + workspace, delegation = service + root, registry = delegation.root, delegation.registry + store = ChatSessionStore(workspace / "private-chat" if request.param else root) + session = store.create_session(goal_id="loopx-manager", agent_id="codex", + adapter_kind="codex_app_server", upstream_thread_id="original", channel_id="manager") + turn, _ = store.create_turn(session["session_id"], client_turn_id="original-question", + message="Check the original requirement and return the result here.", origin="web") + receipt = deliver(root, registry, session=session, turn=turn, + request={"goal_id": delegation.goal_id, "agent_id": "analyst", "brief": { + "schema_version": "collaboration_brief_v0", "purpose": "Check the original requirement", + "context": "Keep the original conversation and recipient.", "constraints": ["No external actions"], + "inputs": [], "acceptance": ["Return the checked result"], + "return_requirement": "Return here without a second question", + }}, source_store=store) + store.update_turn(session["session_id"], turn["turn_id"], status="completing", + response={"message": "The existing recipient will return here.", "context_handoff_receipt": receipt}) + store.finalize_managed_turn_completion(session["session_id"], turn["turn_id"]) + acknowledge(root, delegation.goal_id, "analyst", receipt["request_id"], "adopt", "Checked the source") + report(root, delegation.goal_id, "analyst", receipt["request_id"], "conclusion", "The checked result is ready.") + assert drain(root, registry, store, None) == 1 + assert reply_status(root, receipt)[0]["status"] == "delivered" + if store.root.parent != root: + # A plausible Chat-local copy must never replace registered receipt truth. + shutil.copytree(_root(root), _root(store.root.parent)) + decoy = _root(store.root.parent) / "replies" / receipt["request_id"] / "conclusion.delivery.json" + decoy.write_text(json.dumps({"status": "verification_required", "error": "provider_delivery_unverified"})) + return workspace, root, registry, store, session, receipt + + +def stored_json(*roots): + return {path: path.read_bytes() for root in roots for path in root.rglob("*.json")} + + +def test_http_keeps_original_return_and_request_readback(returned): + workspace, root, registry, store, session, receipt = returned + before = stored_json(_root(root), _root(store.root.parent), store.root) + with session_api(store, registry, workspace) as (read, controller): + assert controller.coordination_runtime_root == root + snapshot = read(session["session_id"]) + result = next(m for m in snapshot["messages"] if m.get("origin") == "manager_followup") + assert result["text"].endswith("The checked result is ready.") + assert result["return_delivery"]["status"] == "delivered" + commitment = next(m for m in snapshot["messages"] if m.get("collaboration")) + assert commitment["collaboration"]["request_id"] == receipt["request_id"] + # A later history page still carries this result's original identity. + first = read(session["session_id"], "?message_limit=2") + cursor = first["message_page"]["next_cursor"] + assert cursor + page = read(session["session_id"], "?message_limit=1&message_cursor=" + cursor) + assert page["messages"][0]["message_id"] == result["message_id"] + assert page["messages"][0]["return_delivery"]["status"] == "delivered" + assert stored_json(_root(root), _root(store.root.parent), store.root) == before, "Reading must not execute or rewrite the original work" + + +def test_copied_session_cannot_borrow_original_host_receipts(returned): + workspace, root, registry, store, session, _ = returned + foreign = ChatSessionStore(workspace / "foreign-chat") + shutil.copytree(store._session_dir(session["session_id"]), foreign._session_dir(session["session_id"])) + before = stored_json(_root(root), store.root, foreign.root) + with session_api(foreign, registry, workspace) as (read, _): + snapshot = read(session["session_id"]) + assert any(m.get("origin") == "manager_followup" for m in snapshot["messages"]) + assert all("return_delivery" not in m and "collaboration" not in m for m in snapshot["messages"]) + assert stored_json(_root(root), store.root, foreign.root) == before + + +def test_missing_receipt_is_queued_and_recovers_without_reexecuting(returned): + workspace, root, registry, store, session, receipt = returned + path = _root(root) / "replies" / receipt["request_id"] / "conclusion.delivery.json" + saved = path.read_bytes() + path.unlink() + with session_api(store, registry, workspace) as (read, _): + pending = read(session["session_id"]) + result = next(m for m in pending["messages"] if m.get("origin") == "manager_followup") + assert result["return_delivery"]["status"] == "queued" + assert not path.exists(), "A read must not synthesize a delivery receipt" + path.write_bytes(saved) + recovered = read(session["session_id"]) + replies = [m for m in recovered["messages"] if m.get("origin") == "manager_followup"] + assert len(replies) == 1 and replies[0]["message_id"] == result["message_id"] + assert replies[0]["return_delivery"]["status"] == "delivered" + + +def test_unreadable_original_route_does_not_fall_back_to_chat_local_copy(returned): + workspace, root, registry, store, session, receipt = returned + route = _root(root) / "roundtrips" / (receipt["request_id"] + ".json") + saved = route.read_bytes() + route.write_text("invalid-json") + with session_api(store, registry, workspace) as (read, _): + unavailable = read(session["session_id"]) + assert all("return_delivery" not in m and "collaboration" not in m for m in unavailable["messages"]) + assert sum(m.get("origin") == "manager_followup" for m in unavailable["messages"]) == 1 + assert route.read_text() == "invalid-json" + route.write_bytes(saved) + recovered = read(session["session_id"]) + returned_message = next(m for m in recovered["messages"] if m.get("origin") == "manager_followup") + assert returned_message["return_delivery"]["status"] == "delivered" diff --git a/tests/test_collaboration_goal_instance.py b/tests/test_collaboration_goal_instance.py index 8720a47c58..333f0e673f 100644 --- a/tests/test_collaboration_goal_instance.py +++ b/tests/test_collaboration_goal_instance.py @@ -125,6 +125,7 @@ def _http_snapshot(root, registry, session_id): # These fixtures script Host liveness; the real HTTP/store projection below # tests collaboration lifetime and receipts, not exited-process recovery. server.runtime_controller = SimpleNamespace( + coordination_runtime_root=root, recover_exited_turn=lambda _session_id: None, close=lambda: None, ) worker = threading.Thread(target=server.serve_forever, daemon=True)