diff --git a/apps/presentation/dashboard/package.json b/apps/presentation/dashboard/package.json index 4e418a7d49..6829d1b346 100644 --- a/apps/presentation/dashboard/package.json +++ b/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", diff --git a/apps/presentation/dashboard/smoke/bound-conversation-http-fixture.py b/apps/presentation/dashboard/smoke/bound-conversation-http-fixture.py new file mode 100644 index 0000000000..9c1e417ae3 --- /dev/null +++ b/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() diff --git a/apps/presentation/dashboard/smoke/bound-conversation-smoke.ts b/apps/presentation/dashboard/smoke/bound-conversation-smoke.ts new file mode 100644 index 0000000000..a926e0fca0 --- /dev/null +++ b/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 { + for (;;) { + let timer: ReturnType; + const line = await Promise.race([lines.next(), new Promise((_, 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; } +} diff --git a/apps/presentation/dashboard/smoke/conversation-history-smoke.ts b/apps/presentation/dashboard/smoke/conversation-history-smoke.ts index 214efa1743..bcf72d5d46 100644 --- a/apps/presentation/dashboard/smoke/conversation-history-smoke.ts +++ b/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; diff --git a/apps/presentation/dashboard/src/data/chat.ts b/apps/presentation/dashboard/src/data/chat.ts index b560d69753..c3e646afe2 100644 --- a/apps/presentation/dashboard/src/data/chat.ts +++ b/apps/presentation/dashboard/src/data/chat.ts @@ -692,18 +692,32 @@ export async function recordProjectionExchange(options: { export type ConversationContext = | { kind: "manager" } | { kind: "goal"; goalId: string } - | { kind: "project"; projectRef: string }; + | { kind: "project"; projectRef: string } + // A UI address for an existing source-owned Session, never a new Core channel. + | ({ kind: "bound" } & BoundConversationTarget); + +export type BoundConversationTarget = { bindingId: string; sessionId: string }; + +/** Goal projections cannot answer a source-owned or ordinary workspace Chat. */ +export function conversationSupportsStatusProjection(context: ConversationContext) { + return context.kind === "manager" || context.kind === "goal"; +} // Goal ids are single path segments, so a key containing "/" never names a Goal. const PROJECT_CONVERSATION_KEY_PREFIX = "project/"; export function conversationContextKey(context: ConversationContext): string { if (context.kind === "manager") return "manager"; + if (context.kind === "bound") return `bound/${encodeURIComponent(context.bindingId)}/${encodeURIComponent(context.sessionId)}`; return context.kind === "goal" ? context.goalId : `${PROJECT_CONVERSATION_KEY_PREFIX}${context.projectRef}`; } export function conversationContextOfKey(key: string): ConversationContext { if (key === "manager") return { kind: "manager" }; + if (key.startsWith("bound/")) { + const [, bindingId = "", sessionId = ""] = key.split("/"); + return { kind: "bound", bindingId: decodeURIComponent(bindingId), sessionId: decodeURIComponent(sessionId) }; + } if (key.startsWith(PROJECT_CONVERSATION_KEY_PREFIX)) { return { kind: "project", projectRef: key.slice(PROJECT_CONVERSATION_KEY_PREFIX.length) }; } @@ -712,6 +726,7 @@ export function conversationContextOfKey(key: string): ConversationContext { export function conversationChannelId(context: ConversationContext): string { if (context.kind === "manager") return "manager"; + if (context.kind === "bound") return conversationContextKey(context); return context.kind === "goal" ? `goal.${context.goalId}` : `project.${context.projectRef}`; } @@ -723,6 +738,17 @@ export async function createChatSession( agentModelId?: string, agentReasoningEffort?: string, ) { + if (context.kind === "bound") { + if (mode !== "resume_latest" || agentModelId || agentReasoningEffort) { + throw new Error("An original conversation cannot be replaced or reconfigured here."); + } + const session = await resolveBoundChatSession(context, signal); + if (!session.resumable || (agentId && agentId !== session.agent_id)) { + throw new Error("The original conversation is unavailable on its executor."); + } + return { ok: true as const, resumed: true, session_id: session.session_id, + agent_id: session.agent_id, goal_id: session.goal_id, session }; + } return requestJson<{ agent_id: string; goal_id: string | null; @@ -792,8 +818,19 @@ export type ChatSessionSummary = { export type { ChatProject }; +const chatProjectsSchema = z.object({ + ok: z.literal(true), + projects: z.array(z.object({ + project_ref: z.string().min(1), title: z.string(), + grant: z.enum(["workspace_read", "workspace_write"]), + goal_workspace_ref: z.string().optional(), + })), +}); + export async function fetchChatProjects(signal?: AbortSignal) { - return requestJson<{ok: true; projects: ChatProject[]}>("/api/chat/projects", {signal}); + const parsed = chatProjectsSchema.safeParse(await requestJson("/api/chat/projects", {signal})); + if (!parsed.success) throw new Error("The workspace list could not be read. Refresh to retry."); + return parsed.data; } /** ``chat_store`` Session modes; an omitted mode is a managed runtime Session. */ @@ -873,24 +910,58 @@ export type ChatSessionSnapshot = { }; export async function fetchChatSession(sessionId: string, signal?: AbortSignal) { - return requestJson(`/api/chat/sessions/${sessionId}`, { signal }); + return requestJson(`/api/chat/sessions/${encodeURIComponent(sessionId)}`, { signal }); } export async function fetchChatSessions(options: { agentId?: string; channelId?: string; goalId?: string; + bound?: BoundConversationTarget; signal?: AbortSignal; }) { const query = new URLSearchParams(); - if (options.agentId) query.set("agent_id", options.agentId); - if (options.channelId) query.set("channel_id", options.channelId); - if (options.goalId) query.set("goal_id", options.goalId); - return requestJson<{ + if (options.bound) { + if (!/^[a-f0-9]{24}$/.test(options.bound.bindingId) + || !/^[A-Za-z0-9._-]{1,160}$/.test(options.bound.sessionId)) { + throw new Error("Invalid original conversation address."); + } + query.set("conversation_binding_id", options.bound.bindingId); + } else { + if (options.agentId) query.set("agent_id", options.agentId); + if (options.channelId) query.set("channel_id", options.channelId); + if (options.goalId) query.set("goal_id", options.goalId); + } + const listed = await requestJson<{ ok: true; schema_version: "loopx_chat_session_list_v1"; + conversation_binding_id?: string; sessions: ChatSessionSummary[]; }>(`/api/chat/sessions?${query.toString()}`, { signal: options.signal }); + if (!options.bound) return listed; + // An older service may ignore the selector and return unrelated ordinary + // history. The owner's echo is mandatory before any transcript is read. + if (listed.conversation_binding_id !== options.bound.bindingId) { + throw new Error("This service cannot verify the original conversation. Update LoopX and retry."); + } + const sessions = listed.sessions.filter(row => row.session_id === options.bound!.sessionId); + if (sessions.length !== 1) throw new Error("The original conversation is no longer available from this connection."); + return { ...listed, sessions }; +} + +export async function resolveBoundChatSession(target: BoundConversationTarget, signal?: AbortSignal) { + return (await fetchChatSessions({ bound: target, signal })).sessions[0]; +} + +/** Discovery chooses once; subsequent navigation freezes the exact Session. */ +export async function discoverBoundChatSession(bindingId: string, signal?: AbortSignal) { + if (!/^[a-f0-9]{24}$/.test(bindingId)) throw new Error("Invalid conversation connection."); + const listed = await requestJson<{ conversation_binding_id?: string; sessions: ChatSessionSummary[] }>( + `/api/chat/sessions?conversation_binding_id=${bindingId}`, { signal }); + if (listed.conversation_binding_id !== bindingId) throw new Error("This service cannot verify the original conversation. Update LoopX and retry."); + const session = [...listed.sessions].sort((a, b) => b.last_activity_at.localeCompare(a.last_activity_at))[0]; + if (!session || !/^[A-Za-z0-9._-]{1,160}$/.test(session.session_id)) throw new Error("No original conversation is available yet. Start it in Lark first."); + return session; } export function mergeChatSessionMessages(snapshots: ChatSessionSnapshot[]) { @@ -920,13 +991,28 @@ export async function fetchChatHistory(options: { agentId?: string; channelId: string; goalId?: string; + bound?: BoundConversationTarget; }, previous?: ChatHistory): Promise { - const listed = previous ?? await fetchChatSessions({ ...options, signal: AbortSignal.timeout(5000) }); + // Source authorization must be re-read even when recovering an older snapshot. + const listed = (!options.bound ? previous : undefined) ?? await fetchChatSessions({ ...options, signal: AbortSignal.timeout(5000) }); const known = new Map(previous?.snapshots.map((snapshot) => [snapshot.session.session_id, snapshot])); // A failed historical read is not an empty transcript. Retrying only the // missing snapshots keeps this recovery read-only and bounds repeated work. - const missing = listed.sessions.filter((session) => !known.has(session.session_id)); - const results = await Promise.allSettled(missing.map((session) => fetchChatSession(session.session_id, AbortSignal.timeout(5000)))); + const missing = listed.sessions.filter((session) => { + const saved = known.get(session.session_id)?.session; + return !saved || (options.bound && (!session.transcript_revision + || session.transcript_revision !== saved.transcript_revision + || session.updated_at !== saved.updated_at || session.active_turn_id !== saved.active_turn_id + || session.status !== saved.status)); + }); + if (options.bound) for (const session of missing) known.delete(session.session_id); + const results = await Promise.allSettled(missing.map(async (session) => { + const snapshot = await fetchChatSession(session.session_id, AbortSignal.timeout(5000)); + if (snapshot.session.session_id !== session.session_id || snapshot.session.agent_id !== session.agent_id) { + throw new Error("The original Session identity changed during history read."); + } + return snapshot; + })); for (const result of results) { if (result.status === "fulfilled") known.set(result.value.session.session_id, result.value); } @@ -2536,8 +2622,8 @@ export type PrivateConversationStopRecovery = z.infer; -export async function fetchPrivateConversations() { - return privateConversationsSchema.parse(await requestJson("/api/chat/lark/private-conversations")); +export async function fetchPrivateConversations(signal?: AbortSignal) { + return privateConversationsSchema.parse(await requestJson("/api/chat/lark/private-conversations", { signal })); } export async function connectPrivateConversation(appRef: string, projectRef: string, executor: string, contextKind: "project" | "steward" = "project", projectGrant: "workspace_read" | "workspace_write" = "workspace_write", groupIds?: string[]) { return privateConversationsSchema.parse(await requestJson("/api/chat/lark/private-conversations", { diff --git a/apps/presentation/dashboard/src/data/use-bound-conversation.ts b/apps/presentation/dashboard/src/data/use-bound-conversation.ts new file mode 100644 index 0000000000..8c44d0090d --- /dev/null +++ b/apps/presentation/dashboard/src/data/use-bound-conversation.ts @@ -0,0 +1,28 @@ +import { useEffect, useState } from "react"; +import { fetchPrivateConversations, resolveBoundChatSession, + type BoundConversationTarget, type ChatSessionSummary, type PrivateConversation } from "./chat"; + +/** Reads the source's title and executor. Admission remains with the Core owner. */ +export function useBoundConversation(target: BoundConversationTarget | null, enabled: boolean) { + const bindingId = target?.bindingId, sessionId = target?.sessionId; + const key = JSON.stringify([bindingId, sessionId]); + const [state, setState] = useState<{ key: string; session?: ChatSessionSummary; source?: PrivateConversation; error?: string }>(); + const [revision, setRevision] = useState(0); + useEffect(() => { + if (!enabled || bindingId === undefined || sessionId === undefined) return; + let active = true; + const controller = new AbortController(); + const timeout = setTimeout(() => controller.abort(), 5000); + Promise.all([resolveBoundChatSession({ bindingId, sessionId }, controller.signal), fetchPrivateConversations(controller.signal)]) + .then(([session, connections]) => { + if (!active) return; + const source = connections.connections.find(row => row.binding_id === bindingId && row.audience === "owner" && row.context_available); + if (!source) throw new Error("This private conversation connection is unavailable."); + if (source.executor_endpoint_id !== session.agent_id) throw new Error("The original conversation's executor no longer matches its connection."); + setState({ key, session, source }); + }).catch(error => { if (active) setState({ key, error: error instanceof Error ? error.message : String(error) }); }) + .finally(() => clearTimeout(timeout)); + return () => { active = false; controller.abort(); clearTimeout(timeout); }; + }, [bindingId, sessionId, key, enabled, revision]); + return { ...(enabled && state?.key === key ? state : {}), retry: () => setRevision(value => value + 1) }; +} diff --git a/apps/presentation/dashboard/src/data/use-conversation-history.ts b/apps/presentation/dashboard/src/data/use-conversation-history.ts index 95e43a2349..8559fe4b03 100644 --- a/apps/presentation/dashboard/src/data/use-conversation-history.ts +++ b/apps/presentation/dashboard/src/data/use-conversation-history.ts @@ -1,5 +1,5 @@ import { useCallback, useEffect, useRef, useState } from "react"; -import { fetchChatHistory, type ChatHistory } from "./chat"; +import { fetchChatHistory, type ChatHistory, type BoundConversationTarget } from "./chat"; type HistoryState = { scope: string; @@ -16,31 +16,33 @@ export function currentChannelSession(history: ChatHistory | null, agentId: stri } /** Read recovery shared by steward and Goal conversations; never executes a Turn. */ -export function useConversationHistory({ agentId, currentAgentId, channelId, goalId, enabled }: { +export function useConversationHistory({ agentId, currentAgentId, channelId, goalId, enabled, bound }: { agentId?: string; currentAgentId: string; channelId: string; goalId?: string; enabled: boolean; + bound?: BoundConversationTarget; }) { - const scope = JSON.stringify([agentId ?? null, currentAgentId, channelId, goalId ?? null]); + const bindingId = bound?.bindingId, sessionId = bound?.sessionId; + const scope = JSON.stringify([agentId ?? null, currentAgentId, channelId, goalId ?? null, bindingId, sessionId]); const [state, setState] = useState(null); const retryRead = useRef<(() => void) | null>(null); const cached = useRef<{ scope: string; history: ChatHistory } | null>(null); const refreshGeneration = useRef(0); const cachedHistory = (forScope: string) => cached.current?.scope === forScope ? cached.current.history : null; - const params = useRef({ agentId, channelId, goalId, scope }); - params.current = { agentId, channelId, goalId, scope }; + const params = useRef({ agentId, channelId, goalId, scope, bound }); + params.current = { agentId, channelId, goalId, scope, bound }; // A caller that learned of a Turn the cached read cannot show -- the service // refused a send because that Turn is running -- needs what the Session looks // like now, not what this hook read earlier. The re-read stays here so the // conversation keeps a single history owner. const refresh = useCallback(async () => { - const { agentId, channelId, goalId, scope } = params.current; + const { agentId, channelId, goalId, scope, bound } = params.current; const generation = ++refreshGeneration.current; let loaded: ChatHistory; try { - loaded = await fetchChatHistory({ agentId, channelId, goalId }); + loaded = await fetchChatHistory({ agentId, channelId, goalId, bound }); } catch (error) { if (params.current.scope === scope && generation === refreshGeneration.current) { setState({ scope, history: cachedHistory(scope), reading: false, failed: true }); @@ -77,9 +79,11 @@ export function useConversationHistory({ agentId, currentAgentId, channelId, goa // A retry can start during a refresh but still carry the older cached snapshot. const isCurrentRead = () => !cancelled && params.current.scope === scope && generation === refreshGeneration.current && cachedHistory(scope) === previous; - setState({ scope, history: previous, reading: true, failed: false }); + setState(current => ({ scope, history: previous, reading: true, + failed: bindingId !== undefined && current?.scope === scope && current.failed })); try { - const loaded = await fetchChatHistory({ agentId, channelId, goalId }, previous ?? undefined); + const loaded = await fetchChatHistory({ agentId, channelId, goalId, + bound: bindingId === undefined ? undefined : { bindingId, sessionId: sessionId ?? "" } }, previous ?? undefined); if (!isCurrentRead()) return; cached.current = { scope, history: loaded }; setState({ scope, history: loaded, reading: false, failed: false }); @@ -89,9 +93,10 @@ export function useConversationHistory({ agentId, currentAgentId, channelId, goa } finally { reading = false; const known = cachedHistory(scope); - if (!cancelled && (!known || known.unavailableSessionIds.length > 0)) { + if (!cancelled && (bindingId !== undefined || !known || known.unavailableSessionIds.length > 0)) { // Back off sustained transport failures; a user can retry immediately. - retryTimer = setTimeout(() => void read(), Math.min(3000 * 2 ** attempts++, 30_000)); + retryTimer = setTimeout(() => void read(), known && !known.unavailableSessionIds.length + ? 5000 : Math.min(3000 * 2 ** attempts++, 30_000)); } } } @@ -103,7 +108,7 @@ export function useConversationHistory({ agentId, currentAgentId, channelId, goa clearTimeout(retryTimer); retryRead.current = null; }; - }, [agentId, channelId, goalId, enabled, scope]); + }, [agentId, channelId, goalId, enabled, scope, bindingId, sessionId]); const current = enabled && state?.scope === scope ? state : null; const history = current?.history ?? null; // A channel transcript spans executors; readability of another executor's @@ -119,9 +124,10 @@ export function useConversationHistory({ agentId, currentAgentId, channelId, goa : history?.unavailableSessionIds.length ? (history.snapshots.length ? "partial" as const : "unavailable" as const) : current.reading ? "loading" as const : "ready" as const, reading: current?.reading ?? enabled, - sendBlocked: enabled && !currentSessionReadable, + sendBlocked: enabled && (!currentSessionReadable || (bound !== undefined && current?.failed === true)), // Missing older history must not restart a recovered current stream. - connectionKey: currentSessionReadable ? `${scope}:${latest?.session_id ?? "empty"}` : null, + connectionKey: currentSessionReadable && !(bound && current?.failed) + ? `${scope}:${latest?.session_id ?? "empty"}${bound ? `:${latest?.active_turn_id ?? "idle"}` : ""}` : null, retry, }; } diff --git a/apps/presentation/dashboard/src/features/personal-workspace/channel-header.tsx b/apps/presentation/dashboard/src/features/personal-workspace/channel-header.tsx index 841c546aa0..6592dab0e9 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/channel-header.tsx +++ b/apps/presentation/dashboard/src/features/personal-workspace/channel-header.tsx @@ -43,6 +43,7 @@ export function ChannelHeader({ selectedGoalTab, scopeUnavailable = false, workspaceGrantLabel, + originalConversationTitle, }: { agents: WorkspaceAgentOption[]; agentWorkActive?: boolean; @@ -51,6 +52,7 @@ export function ChannelHeader({ goalLifecycleBusy?: boolean; /** Set while the steward conversation is scoped to a workspace. */ workspaceGrantLabel?: string | null; + originalConversationTitle?: string; managerChannelBinding?: ManagerChannelBinding | null; managerChatOpen?: boolean; managerRuntime?: ManagerRuntimeSessionReadback | null; @@ -203,7 +205,7 @@ export function ChannelHeader({ // The steward channel binding describes the steward executor; a workspace // conversation runs on the selected runtime instead. - const stewardBindingVisible = !scopeUnavailable && !selectedGoal && !workspaceGrantLabel; + const stewardBindingVisible = !scopeUnavailable && !selectedGoal && !workspaceGrantLabel && !originalConversationTitle; const scopeControl = !scopeUnavailable && !selectedGoal && managerChatOpen && conversationScope && !readOnlySourceLabel ? ( ) : null; + // Source continuation reports its immutable executor; ordinary conversation + // configuration remains with its own controls. + const sourceRuntimeControl = !originalConversationTitle || scopeUnavailable || agentWorkActive || readOnlySourceLabel + ? runtimeControl : <> + {selectedAgentId} + + ; return (
- {selectedGoal ? undefined)} /> :

{t(scopeUnavailable ? "header.goalView" : "header.manager")}

} + {selectedGoal ? undefined)} /> :

{originalConversationTitle ?? t(scopeUnavailable ? "header.goalView" : "header.manager")}

} {workspaceGrantLabel ?

{t("workspace.conversation", { grant: workspaceGrantLabel })}

: null}
{stewardBindingVisible && managerChannelBinding ? ( @@ -289,7 +299,7 @@ export function ChannelHeader({ ) : null} {scopeControl} - {!selectedGoal ? runtimeControl : null} + {!selectedGoal ? sourceRuntimeControl : null} {onRefresh ? ( {refreshState === "loading" ? {t("header.refreshing")} : refreshState === "error" ? {t("header.refreshFailed")} : null} diff --git a/apps/presentation/dashboard/src/features/personal-workspace/lark-settings-page.tsx b/apps/presentation/dashboard/src/features/personal-workspace/lark-settings-page.tsx index 7839a36b02..80d8ef99cd 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/lark-settings-page.tsx +++ b/apps/presentation/dashboard/src/features/personal-workspace/lark-settings-page.tsx @@ -191,6 +191,7 @@ function larkErrorMessage(cause: unknown, fallback: string, t: WorkspaceTranslat } export function LarkSettingsPage({ + onOpenOriginalConversation, embedded = false, focusGoalConnection = false, goals, @@ -204,6 +205,7 @@ export function LarkSettingsPage({ initialGoalId?: string | null; onChanged?: () => void; onClose: () => void; + onOpenOriginalConversation?: (target: import("../../data/chat").BoundConversationTarget) => void; }) { const { t } = useWorkspaceI18n(); const [tab, setTab] = useState("connections"); @@ -543,7 +545,7 @@ export function LarkSettingsPage({
)} - +