feat(chat): continue original private sessions from the workspace

apps/presentation/dashboard/package.json

@@ -76,7 +76,8 @@
     "smoke:conversation-returns-packaged": "LOOPX_PLAYWRIGHT_PACKAGE=\"$PWD/node_modules/playwright\" node smoke/conversation-return-browser-smoke.mjs",
     "smoke:workspace-scope-packaged": "LOOPX_PLAYWRIGHT_PACKAGE=\"$PWD/node_modules/playwright\" node smoke/workspace-scope-browser-smoke.mjs",
     "smoke:recent-completions": "tsc --ignoreConfig --target ES2022 --module CommonJS --moduleResolution Node --ignoreDeprecations 6.0 --resolveJsonModule --esModuleInterop --jsx react-jsx --skipLibCheck --strict --types node --outDir /tmp/loopx-recent-completions-smoke smoke/recent-completions-smoke.ts && NODE_PATH=\"$PWD/node_modules\" node /tmp/loopx-recent-completions-smoke/apps/presentation/dashboard/smoke/recent-completions-smoke.js",
-    "smoke:task-inspector-route": "LOOPX_PLAYWRIGHT_PACKAGE=\"$PWD/node_modules/playwright\" node smoke/task-inspector-route-smoke.mjs"
+    "smoke:task-inspector-route": "LOOPX_PLAYWRIGHT_PACKAGE=\"$PWD/node_modules/playwright\" node smoke/task-inspector-route-smoke.mjs",
+    "smoke:bound-conversation": "vite build --ssr smoke/bound-conversation-smoke.ts --outDir node_modules/.cache/loopx-bound-conversation --emptyOutDir && node node_modules/.cache/loopx-bound-conversation/bound-conversation-smoke.js"
   },
   "dependencies": {
     "@fontsource-variable/geist": "^5.3.0",

apps/presentation/dashboard/smoke/bound-conversation-http-fixture.py

@@ -0,0 +1,108 @@
+"""Packaged UI against real Chat/Core; only provider identity/model are synthetic.
+
+Everything lives in a disposable SQLite runtime. No personal App, source, Goal,
+credentials or model account is read or contacted. Stdin controls source events.
+"""
+from __future__ import annotations
+
+import json
+from pathlib import Path
+import runpy
+import socket
+import sys
+import tempfile
+import threading
+
+import pytest
+
+REPO = Path(__file__).resolve().parents[4]
+sys.path[:0] = [str(REPO), str(REPO / "tests/control_plane")]
+from canonical_authority_fixture import isolate_sqlite_runtime, promoted_create_fixture
+from loopx.chat_server import serve_chat
+
+
+class SyntheticPrivateSource:
+    transport_ref = "synthetic-private"
+    cancelled = staticmethod(lambda: False)
+
+    def __init__(self, root):
+        self.root, self.verified = root, True
+
+    def observe(self):
+        return {"transport_ref": self.transport_ref, "provider_ref": "c" * 24,
+                "operator_ref": "d" * 24, "verified": self.verified}
+
+    def health_snapshot(self):
+        return {"status": "listening"}
+
+    def close(self):
+        pass
+
+    def start(self, server):
+        runtime = server.runtime_controller
+        runtime.codex_home = self.root / "synthetic-model-home"
+        runtime.codex_home.mkdir(exist_ok=True)
+        bindings = runtime.project_contexts.conversation_bindings
+        binding = bindings.configure(transport_ref=self.transport_ref,
+            project_ref=runtime.project_contexts.available()[0]["project_ref"],
+            executor_endpoint_id="codex", context_kind="project", project_grant="workspace_read")
+        source = {"source_ref": "a" * 24, "sender_ref": "d" * 24, "private_human_message": True}
+        original, _ = runtime.open_session(goal_id=None, agent_id="codex", objective="",
+            work_dir=self.root / "project", mode="resume_latest",
+            conversation_binding_id=binding["binding_id"], source_context=source)
+        sid = original["session_id"]
+        server.chat_store.append_message(sid, role="user", text="Synthetic private conversation: review the research notes.")
+        server.chat_store.append_message(sid, role="agent", text="The original conversation and its research notes are ready.")
+        other, _ = runtime.open_session(goal_id=None, agent_id="codex", objective="", mode="new",
+            work_dir=self.root / "project", project_ref=binding["project_ref"])
+        server.chat_store.append_message(other["session_id"], role="agent", text="Unrelated desktop conversation — must remain separate.")
+        print(json.dumps({"origin": f"http://127.0.0.1:{server.server_port}",
+            "bindingId": binding["binding_id"], "sessionId": sid, "ordinarySessionId": other["session_id"]}), flush=True)
+
+        def control():
+            for line in sys.stdin:
+                command = line.strip()
+                if command in {"deny", "recover"}:
+                    self.verified = command == "recover"
+                    result = {"source_verified": self.verified}
+                elif command == "source":
+                    accepted, _ = runtime.enqueue_turn(session_id=sid, message="A new synthetic message from the original private source.",
+                        client_turn_id="synthetic-source-update", origin="lark", work_dir=self.root / "project", objective="")
+                    turn = runtime.wait_for_turn(session_id=sid, turn_id=accepted["turn_id"], timeout_sec=15)
+                    result = {"source_turn_status": turn["status"]}
+                elif command == "close":
+                    server.chat_store.update_session(sid, status="closed")
+                    result = {"closed": True}
+                elif command == "inspect":
+                    saved = server.chat_store.load_session(sid)
+                    result = {"sessions": len(server.chat_store.list_sessions()), "original_thread": saved["upstream_thread_id"],
+                              "messages": server.chat_store.messages(sid)}
+                elif command == "stop":
+                    server.shutdown()
+                    return
+                else:
+                    result = {"unknown_command": command}
+                print(json.dumps(result), flush=True)
+        threading.Thread(target=control, daemon=True).start()
+
+
+def main():
+    with tempfile.TemporaryDirectory(prefix="loopx-bound-ui-") as directory, pytest.MonkeyPatch.context() as patch:
+        root = Path(directory)
+        isolate_sqlite_runtime(root, patch)
+        patch.setenv("CODEX_HOME", str(root / "synthetic-model-home"))
+        registry, _, _ = promoted_create_fixture(root, provider="sqlite")
+        fake = root / "synthetic-codex"
+        fake.write_text(runpy.run_path(str(REPO / "examples/loopx-chat-runtime-smoke.py"))["FAKE_CODEX"])
+        fake.chmod(0o700)
+        source = SyntheticPrivateSource(root)
+        with socket.socket() as sock:
+            sock.bind(("127.0.0.1", 0))
+            port = sock.getsockname()[1]
+        serve_chat(registry_path=registry, runtime_root_override=root / "runtime", scan_roots=[root / "project"],
+            codex_bin=str(fake), lark_cli_bin="missing-synthetic-lark", port=port,
+            external_conversation_factories=(lambda _: source,))
+
+
+if __name__ == "__main__":
+    main()

apps/presentation/dashboard/smoke/bound-conversation-smoke.ts

@@ -0,0 +1,109 @@
+import assert from "node:assert/strict";
+import { spawn } from "node:child_process";
+import { once } from "node:events";
+import { resolve } from "node:path";
+import { createInterface } from "node:readline";
+import { resolveTestPython } from "../../../../scripts/test-python.mjs";
+import { acceptChatTurn, conversationContextKey, conversationContextOfKey, conversationSupportsStatusProjection, createChatSession,
+  discoverBoundChatSession, fetchChatHistory, fetchChatProjects, resolveBoundChatSession } from "../src/data/chat";
+
+// Invariant: continuation and refreshed history retain the verified original
+// Session. An absent/ignored selector, changed source or closed Session cannot
+// create a replacement, disclose unrelated history, or redirect a message.
+const repoRoot = resolve(process.cwd(), "../../..");
+const child = spawn(resolveTestPython({ repoRoot }), ["-u", "apps/presentation/dashboard/smoke/bound-conversation-http-fixture.py"],
+  { cwd: repoRoot, stdio: ["pipe", "pipe", "pipe"] });
+const exited = once(child, "exit");
+const output = createInterface({ input: child.stdout });
+const lines = output[Symbol.asyncIterator]();
+let stderr = "";
+child.stderr.on("data", chunk => { stderr += String(chunk); });
+async function next(): Promise<any> {
+  for (;;) {
+    let timer: ReturnType<typeof setTimeout>;
+    const line = await Promise.race([lines.next(), new Promise<never>((_, reject) => {
+      timer = setTimeout(() => reject(new Error(`Fixture response timed out: ${stderr}`)), 20_000);
+    })]).finally(() => clearTimeout(timer));
+    assert.equal(line.done, false, stderr);
+    if (line.value!.startsWith("{")) return JSON.parse(line.value!);
+  }
+}
+const originalFetch = globalThis.fetch;
+try {
+  const { origin, bindingId, sessionId } = await next();
+  const target = { bindingId, sessionId };
+  const context = { kind: "bound" as const, ...target };
+  for (const [candidate, expected] of [
+    [{ kind: "manager" }, true], [{ kind: "goal", goalId: "research" }, true],
+    [{ kind: "project", projectRef: "workspace" }, false], [context, false],
+  ] as const) assert.equal(conversationSupportsStatusProjection(candidate), expected);
+  const options = { channelId: conversationContextKey(context), bound: target };
+  const calls: { path: string; method: string }[] = [];
+  globalThis.fetch = (input, init) => {
+    calls.push({ path: String(input), method: init?.method ?? "GET" });
+    return originalFetch(new URL(String(input), origin), init);
+  };
+  assert.deepEqual(conversationContextOfKey(conversationContextKey(context)), context);
+  assert.equal((await discoverBoundChatSession(bindingId)).session_id, sessionId);
+  const history = await fetchChatHistory(options);
+  assert.equal(history.sessions.length, 1);
+  assert.equal(history.messages.length, 2);
+  assert.ok(history.messages.every(row => row.session_id === sessionId && !row.text.includes("Unrelated desktop")));
+  assert.equal((await createChatSession(context, "codex")).session_id, sessionId);
+  await assert.rejects(() => createChatSession(context, "codex", "new"));
+  await assert.rejects(() => createChatSession(context, "codex", "resume_latest", undefined, "another-model"));
+  assert.ok(calls.every(row => row.method === "GET"), "Opening original history never starts a Turn or Session");
+  calls.length = 0;
+  await fetchChatHistory(options, history);
+  assert.equal(calls.length, 1, "Quiet refresh verifies the source using only its compact index");
+  assert.equal(calls[0].path, `/api/chat/sessions?conversation_binding_id=${bindingId}`);
+  child.stdin.write("source\n");
+  assert.equal((await next()).source_turn_status, "completed");
+  const updated = await fetchChatHistory(options, history);
+  assert.equal(updated.messages.filter(row => row.text === "A new synthetic message from the original private source.").length, 1);
+  const accepted = await acceptChatTurn(sessionId, "Continue from the desktop.", "synthetic-web-update");
+  assert.equal(accepted.session_id, sessionId);
+  child.stdin.write("deny\n");
+  assert.equal((await next()).source_verified, false);
+  await assert.rejects(() => fetchChatHistory(options, updated));
+  await assert.rejects(() => acceptChatTurn(sessionId, "Must not execute.", "synthetic-denied-web"));
+  child.stdin.write("recover\n");
+  assert.equal((await next()).source_verified, true);
+  assert.equal((await resolveBoundChatSession(target)).session_id, sessionId);
+  const routedFetch = globalThis.fetch;
+  const projects = await fetchChatProjects();
+  assert.ok(projects.projects.length > 0, "The production workspace response is accepted");
+  for (const malformed of [{ok: true}, {ok: true, projects: null},
+    {ok: true, projects: [{project_ref: "workspace", title: "Workspace", grant: "unexpected"}]}]) {
+    globalThis.fetch = async () => new Response(JSON.stringify(malformed));
+    await assert.rejects(() => fetchChatProjects(), /workspace list could not be read/);
+  }
+  globalThis.fetch = async () => new Response(JSON.stringify({ok: true, projects: []}));
+  assert.deepEqual(await fetchChatProjects(), {ok: true, projects: []}, "An observed empty list remains valid");
+  globalThis.fetch = routedFetch;
+  for (const echo of [undefined, "f".repeat(24)]) {
+    globalThis.fetch = async () => new Response(JSON.stringify({ ok: true, sessions: history.sessions, conversation_binding_id: echo }));
+    await assert.rejects(() => resolveBoundChatSession(target), /cannot verify/);
+  }
+  globalThis.fetch = routedFetch;
+  const beforeMalformed = calls.length;
+  await assert.rejects(() => resolveBoundChatSession({ bindingId, sessionId: "../other" }));
+  assert.equal(calls.length, beforeMalformed, "Malformed original addresses never issue a request");
+  child.stdin.write("close\n");
+  assert.equal((await next()).closed, true);
+  await assert.rejects(() => fetchChatHistory(options));
+  await assert.rejects(() => createChatSession(context, "codex"));
+  child.stdin.write("inspect\n");
+  const inspected = await next();
+  assert.equal(inspected.sessions, 2, "No discovery or denial creates a replacement Session");
+  assert.equal(inspected.original_thread, "durable-thread");
+  assert.equal(inspected.messages.filter((row: any) => row.text === "Must not execute.").length, 0);
+  child.stdin.write("stop\n");
+  const [code] = await exited;
+  assert.equal(code, 0, stderr);
+  console.log("bound-conversation: passed (real HTTP/SQLite/Core, exact original history, no fallback/create, source and Web input, denial/recovery, closed and old-service refusal)");
+} finally {
+  globalThis.fetch = originalFetch;
+  output.close();
+  if (child.exitCode === null) { child.kill("SIGINT"); await exited; }
+}

apps/presentation/dashboard/smoke/conversation-history-smoke.ts

@@ -67,7 +67,9 @@ try {
   child.stdin.write("result\n");
   assert.equal((await next()).published, "result");
   await observe();
-  assert.equal(transcript.at(-1)?.text, "First checked result");
+  const firstResult = "First checked result\n\nForm: https://example.org/forms/result。";
+  const revisedResult = "Revised checked result\n\nForm: https://example.org/forms/revision。";
+  assert.equal(transcript.at(-1)?.text, firstResult);
   requests.length = 0;
   await observe();
   assert.equal(requests.length, 1, "Delivery does not trigger continuous full history reads");
@@ -79,17 +81,17 @@ try {
   assert.equal((await next()).published, "revision");
   assert.equal((await next()).unavailable, "old");
   assert.deepEqual(await observe(), ["old"]);
-  assert.equal(transcript.at(-1)?.text, "First checked result", "Failed reads preserve the first result");
+  assert.equal(transcript.at(-1)?.text, firstResult, "Failed reads preserve the complete first result, including its link");
   child.stdin.write("recover\n");
   assert.equal((await next()).recovered, true);
   requests.length = 0;
   await observe();
   assert.deepEqual(requests, ["/api/chat/sessions?", "/api/chat/sessions/old"], "Retry reads only the changed old session");
-  assert.equal(transcript.at(-1)?.text, "Revised checked result");
-  assert.equal(transcript.filter(row => row.text === "Revised checked result").length, 1);
+  assert.equal(transcript.at(-1)?.text, revisedResult);
+  assert.equal(transcript.filter(row => row.text === revisedResult).length, 1);
   assert.equal(transcript.find(row => row.sourceSessionId === "current"), originalCurrent, "Old updates preserve current text and object identity");
   await observe();
-  assert.equal(transcript.filter(row => row.text === "Revised checked result").length, 1);
+  assert.equal(transcript.filter(row => row.text === revisedResult).length, 1);
   child.stdin.end("inspect\n");
   assert.deepEqual(await next(), { store_unchanged: true, turn_count: 0 });
   const [exitCode] = await exited;