diff --git a/apps/presentation/dashboard/smoke/bound-conversation-http-fixture.py b/apps/presentation/dashboard/smoke/bound-conversation-http-fixture.py index 2b24d60aa9..39a1bd0a99 100644 --- a/apps/presentation/dashboard/smoke/bound-conversation-http-fixture.py +++ b/apps/presentation/dashboard/smoke/bound-conversation-http-fixture.py @@ -68,6 +68,7 @@ def start(self, server): 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) + settings_handler = server.RequestHandlerClass._private_conversations def control(): for line in sys.stdin: @@ -92,6 +93,12 @@ def control(): elif command == "revoke": bindings.disconnect(binding["binding_id"], expected_revision=bindings.read()["revision"]) result = {"revoked": binding["binding_id"]} + elif command in {"settings-unavailable", "settings-recover"}: + def unavailable(handler): + return handler._send_error("Synthetic connection settings are unavailable", status=503) + server.RequestHandlerClass._private_conversations = ( + unavailable if command == "settings-unavailable" else settings_handler) + result = {"settings_available": command == "settings-recover"} elif command == "populate": for index in range(6): server.chat_store.append_message(sid, role="user", text=f"Synthetic source request {index + 1}: 保留原任务与尚未确认的结果。") diff --git a/apps/presentation/dashboard/smoke/bound-conversation-smoke.ts b/apps/presentation/dashboard/smoke/bound-conversation-smoke.ts index a926e0fca0..92727dc223 100644 --- a/apps/presentation/dashboard/smoke/bound-conversation-smoke.ts +++ b/apps/presentation/dashboard/smoke/bound-conversation-smoke.ts @@ -5,7 +5,7 @@ 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"; + discoverBoundChatSession, fetchChatHistory, fetchChatProjects, resolveBoundChatSession, resolveBoundConversation } from "../src/data/chat"; // Invariant: continuation and refreshed history retain the verified original // Session. An absent/ignored selector, changed source or closed Session cannot @@ -41,8 +41,25 @@ try { const calls: { path: string; method: string }[] = []; globalThis.fetch = (input, init) => { calls.push({ path: String(input), method: init?.method ?? "GET" }); + if (String(input).startsWith("/api/chat/lark/private-conversations")) { + throw new Error("Unrelated settings inventory is unavailable"); + } return originalFetch(new URL(String(input), origin), init); }; + const original = await resolveBoundConversation(target); + assert.equal(original.session.session_id, sessionId); + assert.equal(original.source.binding_id, bindingId); + assert.equal(original.source.audience, "owner"); + assert.equal(original.source.context_kind, "project"); + assert.equal(original.source.project_title, "project"); + assert.deepEqual(calls, [{path: `/api/chat/sessions?conversation_binding_id=${bindingId}`, method: "GET"}], + "Opening an original conversation uses one source-qualified read, independent of settings health"); + child.stdin.write("settings-unavailable\n"); + assert.equal((await next()).settings_available, false); + assert.equal((await originalFetch(new URL("/api/chat/lark/private-conversations", origin))).status, 503); + assert.equal((await resolveBoundConversation(target)).session.session_id, sessionId); + child.stdin.write("settings-recover\n"); + assert.equal((await next()).settings_available, true); assert.deepEqual(conversationContextOfKey(conversationContextKey(context)), context); assert.equal((await discoverBoundChatSession(bindingId)).session_id, sessionId); const history = await fetchChatHistory(options); @@ -71,6 +88,14 @@ try { assert.equal((await next()).source_verified, true); assert.equal((await resolveBoundChatSession(target)).session_id, sessionId); const routedFetch = globalThis.fetch; + for (const source of [undefined, {...original.source, binding_id: "f".repeat(24)}, + {...original.source, audience: "group"}, {...original.source, context_available: false}, + {...original.source, executor_endpoint_id: "another-executor"}]) { + globalThis.fetch = async () => new Response(JSON.stringify({ok: true, + conversation_binding_id: bindingId, sessions: history.sessions, conversation_source: source})); + await assert.rejects(() => resolveBoundConversation(target)); + } + globalThis.fetch = routedFetch; 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}, diff --git a/apps/presentation/dashboard/src/data/chat.ts b/apps/presentation/dashboard/src/data/chat.ts index 46cc8dad34..8ff412a6b4 100644 --- a/apps/presentation/dashboard/src/data/chat.ts +++ b/apps/presentation/dashboard/src/data/chat.ts @@ -981,6 +981,7 @@ export async function fetchChatSessions(options: { ok: true; schema_version: "loopx_chat_session_list_v1"; conversation_binding_id?: string; + conversation_source?: unknown; sessions: ChatSessionSummary[]; }>(`/api/chat/sessions?${query.toString()}`, { signal: options.signal }); if (!options.bound) return listed; @@ -2694,6 +2695,25 @@ const privateConversationsSchema = z.object({ok: z.literal(true), revision: z.nu app_steward_binding_id: z.string().regex(/^[a-f0-9]{24}$/).nullable().default(null), connections: z.array(privateConversationSchema), stop_recoveries: z.array(privateConversationStopRecoverySchema).default([])}); export type PrivateConversation = z.infer; +const boundConversationSourceSchema = privateConversationSchema.pick({ + binding_id: true, app_ref: true, context_kind: true, project_title: true, executor_endpoint_id: true, +}).extend({audience: z.literal("owner"), context_available: z.literal(true)}); +export type BoundConversationSource = z.infer; + +/** Session and display source come from one fresh Core-verified read. */ +export async function resolveBoundConversation(target: BoundConversationTarget, signal?: AbortSignal) { + const listed = await fetchChatSessions({bound: target, signal}); + const parsed = boundConversationSourceSchema.safeParse(listed.conversation_source); + if (!parsed.success || parsed.data.binding_id !== target.bindingId) { + throw new Error("This service cannot verify the original conversation. Update LoopX and retry."); + } + const session = listed.sessions[0], source = parsed.data; + if (source.executor_endpoint_id !== session.agent_id) { + throw new Error("The original conversation's executor no longer matches its connection."); + } + return {session, source}; +} + export async function fetchPrivateConversations(signal?: AbortSignal) { return privateConversationsSchema.parse(await requestJson("/api/chat/lark/private-conversations", { signal })); } diff --git a/apps/presentation/dashboard/src/data/use-bound-conversation.ts b/apps/presentation/dashboard/src/data/use-bound-conversation.ts index 8c44d0090d..9db4c27430 100644 --- a/apps/presentation/dashboard/src/data/use-bound-conversation.ts +++ b/apps/presentation/dashboard/src/data/use-bound-conversation.ts @@ -1,24 +1,21 @@ import { useEffect, useState } from "react"; -import { fetchPrivateConversations, resolveBoundChatSession, - type BoundConversationTarget, type ChatSessionSummary, type PrivateConversation } from "./chat"; +import { resolveBoundConversation, + type BoundConversationTarget, type ChatSessionSummary, type BoundConversationSource } 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 [state, setState] = useState<{ key: string; session?: ChatSessionSummary; source?: BoundConversationSource; 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]) => { + resolveBoundConversation({ bindingId, sessionId }, controller.signal) + .then(({session, source}) => { 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)); diff --git a/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-model.ts b/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-model.ts index f13a1af909..36d49babcf 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-model.ts +++ b/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-model.ts @@ -1,7 +1,7 @@ import type { GoalDraft } from "../../../../../../loopx/control_plane/collaboration/goal_draft.js"; import type { TurnStep } from "../../data/turn-steps"; import type { ChatProject, CollaborationReadback, LoopXModeSettings } from "../../data/chat-model"; -import type { BoundConversationTarget, ChatSessionSummary, PrivateConversation } from "../../data/chat"; +import type { BoundConversationTarget, ChatSessionSummary, BoundConversationSource } from "../../data/chat"; import type { TeamPlanAppliedOutcome } from "./team-plan-preview"; import type { ActionReviewPlan } from "../../../../../../loopx/control_plane/presentation/action_review_plan.js"; import type { GoalAcceptanceObservation } from "../../data/goal-acceptance-observation"; @@ -116,7 +116,7 @@ export type WorkspaceConversationDirectory = { onOpenOriginal?: (target: BoundConversationTarget) => void; /** Local manager entry discovery blocks sending, never Goal navigation. */ entryGate?: {unavailable: boolean; retry: () => void}; - bound?: { target: BoundConversationTarget; session?: ChatSessionSummary; source?: PrivateConversation; + bound?: { target: BoundConversationTarget; session?: ChatSessionSummary; source?: BoundConversationSource; error?: string; retry: () => void }; managed?: {target: import("../../data/chat").ManagedConversationTarget; session?: import("../../data/chat").ChatSessionSummary; error?: string; retry: () => void}; }; diff --git a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md index abac098682..62e5f2a3e4 100644 --- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md +++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md @@ -179,6 +179,15 @@ checkpoint remains open until that native path is implemented and read back. Chat storage overrides do not select another registered-work authority. The workspace Status and delivery task map read the coordinator's existing registered runtime root, while private Chat history and its settings keep their own store. +The bound Session index returns its limited display source from the same fresh +Core verification and binding fence. Opening that original conversation no longer +waits for the full connection-settings inventory, unrelated Agent candidates or +listener/recovery diagnostics. The client requires the exact binding echo, +owner audience, available workspace and matching executor; missing or stale +descriptors fail closed without a catalog fallback or replacement Session. The +existing five-second read bound and fresh input admission remain unchanged. +This qualifies a source-read boundary, not complete history/media synchronization +or an installed provider roundtrip. The original-conversation request inventory pins one qualified code revision per HTTP read through the existing Effect request scope. Within that read, provider scope is verified once per distinct original Session, including older Sessions diff --git a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md index 885d64e382..598ae7dcd7 100644 --- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md +++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md @@ -123,7 +123,14 @@ Goal 范围、项目、明确原会话、受管 Agent 以及远端/只读视图 验收不能证明实际原生 App 路由正确。此项保持未闭,直至原生路径实现并独立读回。 Chat 存储覆盖不另选注册工作的权威根目录。工作区状态与交付任务图沿用协调器 -既有的注册 runtime root,私有 Chat 历史与设置仍留在原存储。原会话委派列表通过 +既有的注册 runtime root,私有 Chat 历史与设置仍留在原存储。 +绑定 Session 索引在同一次新 Core 核验和 binding 锁内返回有限的显示来源。 +打开该原会话不再等待全部连接的设置清单、无关 Agent 候选或 listener/恢复诊断。 +客户端仍要求准确的 binding 回显、本人私聊、可用工作区和一致执行器;来源描述 +缺失或失效时拒绝,不回退设置清单或另建 Session。既有五秒读取边界和每次输入的 +新准入不变。此项资格只覆盖来源读取边界,不代表完整历史/媒体同步或安装态的 +真实 provider 往返已通过。 +原会话委派列表通过 既有 Effect 请求范围为每次 HTTP 读取固定一个已核验代码修订。在一次读取内, 每个不同的原 Session(包括重开会话背后的旧 Session)先独立核验 provider 范围, 返回任何内容前再对所有已观察来源做一次新核验。每条记录仍检查当前来源依据、 diff --git a/docs/architecture/rfcs/assets/app-conversation-source-read/after-desktop.png b/docs/architecture/rfcs/assets/app-conversation-source-read/after-desktop.png new file mode 100644 index 0000000000..de7391ea00 Binary files /dev/null and b/docs/architecture/rfcs/assets/app-conversation-source-read/after-desktop.png differ diff --git a/docs/architecture/rfcs/assets/app-conversation-source-read/after-narrow.png b/docs/architecture/rfcs/assets/app-conversation-source-read/after-narrow.png new file mode 100644 index 0000000000..11e97b08e9 Binary files /dev/null and b/docs/architecture/rfcs/assets/app-conversation-source-read/after-narrow.png differ diff --git a/docs/architecture/rfcs/assets/app-conversation-source-read/before-desktop.png b/docs/architecture/rfcs/assets/app-conversation-source-read/before-desktop.png new file mode 100644 index 0000000000..a10ba57c79 Binary files /dev/null and b/docs/architecture/rfcs/assets/app-conversation-source-read/before-desktop.png differ diff --git a/docs/architecture/rfcs/assets/app-conversation-source-read/report-reader-narrow.png b/docs/architecture/rfcs/assets/app-conversation-source-read/report-reader-narrow.png new file mode 100644 index 0000000000..ea46344abc Binary files /dev/null and b/docs/architecture/rfcs/assets/app-conversation-source-read/report-reader-narrow.png differ diff --git a/loopx/capabilities/native_chat/conversation_bindings.py b/loopx/capabilities/native_chat/conversation_bindings.py index 2168754917..6b9e72dbe5 100644 --- a/loopx/capabilities/native_chat/conversation_bindings.py +++ b/loopx/capabilities/native_chat/conversation_bindings.py @@ -301,6 +301,9 @@ def _session_observation(session: dict[str, Any]) -> dict[str, Any]: "session_mode", "status", "project_context", "steward_context")} def local_sessions(self, binding_id: str, *, app_steward_entry: bool = False) -> list[dict[str, Any]]: + return self.local_session_snapshot(binding_id, app_steward_entry=app_steward_entry)["sessions"] + + def local_session_snapshot(self, binding_id: str, *, app_steward_entry: bool = False) -> dict[str, Any]: """Read original targets; metadata hints never replace Core's grant. Provider IO and store inventory stay outside the publication fence. @@ -326,9 +329,10 @@ def local_sessions(self, binding_id: str, *, app_steward_entry: bool = False) -> current = self.read() if next((item for item in current["bindings"] if item["binding_id"] == binding_id), None) != row: raise ValueError("conversation binding changed during identity verification") + projects = self.projects.available() result = self._core("collaboration.conversation.local_sessions", { "current": current, "binding_id": binding_id, "observation": observation, - "available_projects": self.projects.available(), + "available_projects": projects, "sessions": [self._session_observation(session) for session in sessions], **({"app_steward_entry": True} if app_steward_entry else {}), **({"available_goal_ids": self.goal_scope_ids(row)} if row.get("goal_scope") == "all_registered" else {}), @@ -336,7 +340,15 @@ def local_sessions(self, binding_id: str, *, app_steward_entry: bool = False) -> # Build the lookup only after Core has rejected duplicate identity; # an IO dictionary must not silently turn ambiguity into one target. by_id = {session["session_id"]: session for session in sessions} - return [store.public_session(by_id[sid]) for sid in result["session_ids"]] + project = next(project for project in projects if project["project_ref"] == row["project_ref"]) + # Project this exact Core-verified source under the same fence. + # Opening a conversation must not enumerate settings, unrelated + # Agent candidates, listener health or stop-recovery inventories. + return {"sessions": [store.public_session(by_id[sid]) for sid in result["session_ids"]], + "conversation_source": {"binding_id": binding_id, "app_ref": row["transport_ref"], + "context_kind": row["context_kind"], "audience": row.get("audience", "owner"), + "project_title": Path(project["workspace_path"]).name, "context_available": True, + "executor_endpoint_id": row["executor_endpoint_id"]}} def goal_scope_ids(self, binding: dict[str, Any]) -> list[str]: if binding.get("goal_scope") != "all_registered": diff --git a/loopx/chat_server.py b/loopx/chat_server.py index 736cdfec0a..62b21ebbea 100644 --- a/loopx/chat_server.py +++ b/loopx/chat_server.py @@ -1008,7 +1008,7 @@ def _list_sessions(self) -> None: try: if bindings is None: raise ValueError("conversation bindings unavailable") - sessions = bindings.local_sessions(query["conversation_binding_id"][0]) + snapshot = bindings.local_session_snapshot(query["conversation_binding_id"][0]) except (ValueError, OSError, KeyError): # A failed provider/store read may contain private diagnostics. # Refuse this selector; never fall back to an unfiltered list. @@ -1016,7 +1016,7 @@ def _list_sessions(self) -> None: # Older services ignore this new query. A matching selector in the # response lets callers reject their ordinary, unfiltered result. return self._send_json({"ok": True, "schema_version": "loopx_chat_session_list_v1", - "conversation_binding_id": query["conversation_binding_id"][0], "sessions": sessions}) + "conversation_binding_id": query["conversation_binding_id"][0], **snapshot}) goal_id = _compact_text((query.get("goal_id") or [""])[0], limit=160) or None agent_id = _compact_text((query.get("agent_id") or [""])[0], limit=80) or None channel_id = _compact_text((query.get("channel_id") or [""])[0], limit=160) or None diff --git a/tests/test_chat_http_runtime_scope.py b/tests/test_chat_http_runtime_scope.py index 8e342df45a..75886dbae7 100644 --- a/tests/test_chat_http_runtime_scope.py +++ b/tests/test_chat_http_runtime_scope.py @@ -10,6 +10,43 @@ from loopx.control_plane import effect_runtime +@pytest.mark.parametrize("failure", ["owner", "workspace", "revoked"]) +def test_bound_index_carries_only_its_verified_display_source(ordinary, monkeypatch, failure): # noqa: F811 + store, runtime, provider, transport = connect(ordinary) + _, _, contexts, request, capture, _, workspace = ordinary + binding = transport._binding("notes-app") + 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": binding["operator_ref"], + "private_human_message": True}) + bindings = contexts.conversation_bindings + monkeypatch.setattr(bindings, "agent_candidates", lambda *_: pytest.fail("conversation reads must not enumerate settings")) + path = "/api/chat/sessions?conversation_binding_id=" + binding["binding_id"] + before = store.list_sessions() + try: + code, packet = request(path) + assert code == 200, packet + assert packet["conversation_source"] == { + "binding_id": binding["binding_id"], "app_ref": "notes-app", "context_kind": "project", + "project_title": "notes", "context_available": True, "audience": "owner", + "executor_endpoint_id": "codex"} + assert [row["session_id"] for row in packet["sessions"]] == [session["session_id"]] + assert "conversation_source" not in request("/api/chat/sessions")[1] + if failure == "owner": + provider.owner_ready = False + elif failure == "workspace": + contexts.roots = [] + else: + bindings.disconnect(binding["binding_id"], expected_revision=bindings.read()["revision"]) + code, denied = request(path) + assert code == 409 and "conversation_source" not in denied and "sessions" not in denied + assert store.list_sessions() == before + assert '"method": "turn/start"' not in capture.read_text() + assert not provider.writes + finally: + runtime.close() + + @pytest.mark.parametrize("method", ["GET", "POST"]) def test_http_reuses_code_scan_but_rereads_revoked_grants(ordinary, monkeypatch, method): # noqa: F811 store, runtime, provider, transport = connect(ordinary)