From 0445dc63f7f98addde2a2d8dbd67fab4e6759443 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 5 Nov 2026 05:46:37 +0800 Subject: [PATCH] fix(chat): trust Core-qualified original conversation after fallback Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../smoke/bound-conversation-http-fixture.py | 46 ++++++++++++++--- .../smoke/bound-conversation-smoke.ts | 50 ++++++++++++++----- apps/presentation/dashboard/src/data/chat.ts | 5 +- 3 files changed, 79 insertions(+), 22 deletions(-) diff --git a/apps/presentation/dashboard/smoke/bound-conversation-http-fixture.py b/apps/presentation/dashboard/smoke/bound-conversation-http-fixture.py index 94ab27b8ac..ea7bcd8302 100644 --- a/apps/presentation/dashboard/smoke/bound-conversation-http-fixture.py +++ b/apps/presentation/dashboard/smoke/bound-conversation-http-fixture.py @@ -19,8 +19,8 @@ REPO = Path(__file__).resolve().parents[4] BACKEND = Path(os.environ.get("LOOPX_CHAT_FIXTURE_BACKEND_ROOT", REPO)).resolve() sys.path[:0] = [str(BACKEND), 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 +from canonical_authority_fixture import isolate_sqlite_runtime, promoted_create_fixture # noqa: E402 +from loopx.chat_server import serve_chat # noqa: E402 class SyntheticPrivateSource: @@ -54,18 +54,40 @@ def start(self, server): runtime.codex_home = self.root / "synthetic-model-home" runtime.codex_home.mkdir(exist_ok=True) bindings = runtime.project_contexts.conversation_bindings + fallback = "--steward-fallback" in sys.argv and not self.peer binding = bindings.configure(transport_ref=self.transport_ref, project_ref=runtime.project_contexts.available()[0]["project_ref"], - executor_endpoint_id="codex", context_kind="steward" if self.steward else "project", project_grant="workspace_read") + executor_endpoint_id="antigravity" if fallback else "codex", + context_kind="steward" if self.steward else "project", project_grant="workspace_read") source = {"source_ref": ("b" if self.peer else "a") * 24, "sender_ref": ("f" if self.peer else "d") * 24, "private_human_message": True} if "--reply-context" in sys.argv and not self.peer: from loopx.extensions.lark.conversation_identity import lark_private_source source = lark_private_source(provider_ref=self.observe()["provider_ref"], event=self.reply_event()) - 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) + if fallback: + scoped = bindings.resolve(binding_id=binding["binding_id"], **source) + original = server.chat_store.create_session(goal_id="loopx-manager", agent_id="codex", + adapter_kind="codex_app_server", upstream_thread_id="durable-thread", upstream_mode="chat", + codex_home=str(runtime.codex_home), channel_id=scoped["channel_id"], steward_context=scoped["context"]) + else: + 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"] + # Seed the canonical post-recovery state in this disposable fixture. + # Core still checks its exact original context, allocation and history; + # this scenario tests UI readback, not the quota transition worker. + if fallback: + original = server.chat_store.update_session(sid, + manager_executor_allocation={"schema_version": "manager_executor_allocation_v0", + "selection_policy": "flexible", "configured_endpoint": "antigravity", + "executor_endpoint": "codex", "allocation_reason": "flexible_availability_fallback", + "executor_endpoint_source": "flexible_pool", "executor_endpoint_default_reason": "", + "model": "gpt-6.1-sol", "model_source": "vendor_default", "reasoning_effort": "high", + "configuration_revision": "synthetic-flexible-routing-v1", + "eligible_endpoints": ["antigravity", "codex"], "available": True}, + upstream_context_history=[{"agent_id": "antigravity", "thread_id": "synthetic-original-argon", + "agent_model_id": "gemini-4-argon-high", "manager_context_version": None}]) if self.peer: server.chat_store.append_message(sid, role="agent", text="Different private steward: keep this source separate.") return @@ -143,6 +165,16 @@ def unavailable(handler): server.RequestHandlerClass._private_conversations = ( unavailable if command == "settings-unavailable" else settings_handler) result = {"settings_available": command == "settings-recover"} + elif fallback and command.startswith("fallback-"): + allocation = dict(original["manager_executor_allocation"]) + if command == "fallback-invalid-allocation": + allocation["allocation_reason"] = "configured_preference" + elif command == "fallback-wrong-configured-endpoint": + allocation["configured_endpoint"] = "codex" + elif command != "fallback-restore": + raise ValueError("unknown synthetic fallback state") + server.chat_store.update_session(sid, manager_executor_allocation=allocation) + result = {"fallback_state": command} elif command == "populate": for index in range(6): server.chat_store.append_message(sid, role="user", text=f"Synthetic source request {index + 1}: 保留原任务与尚未确认的结果。") @@ -188,7 +220,7 @@ def main(): ' output.write(json.dumps(request) + "\\n")\n method = request.get("method")') fake.write_text(model) fake.chmod(0o700) - steward = "--steward" in sys.argv or "--reply-context" in sys.argv + steward = any(flag in sys.argv for flag in ("--steward", "--reply-context", "--steward-fallback")) source = SyntheticPrivateSource(root, steward=steward) peer = SyntheticPrivateSource(root, steward=True, peer=True) if steward and "--reply-context" not in sys.argv else None source.peer_provider = peer diff --git a/apps/presentation/dashboard/smoke/bound-conversation-smoke.ts b/apps/presentation/dashboard/smoke/bound-conversation-smoke.ts index 92727dc223..0b5712e570 100644 --- a/apps/presentation/dashboard/smoke/bound-conversation-smoke.ts +++ b/apps/presentation/dashboard/smoke/bound-conversation-smoke.ts @@ -11,7 +11,9 @@ import { acceptChatTurn, conversationContextKey, conversationContextOfKey, conve // 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"], +const fallback = process.env.LOOPX_BOUND_STEWARD_FALLBACK === "1"; +const child = spawn(resolveTestPython({ repoRoot }), ["-u", "apps/presentation/dashboard/smoke/bound-conversation-http-fixture.py", + ...(fallback ? ["--steward-fallback"] : [])], { cwd: repoRoot, stdio: ["pipe", "pipe", "pipe"] }); const exited = once(child, "exit"); const output = createInterface({ input: child.stdout }); @@ -50,8 +52,10 @@ try { 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.context_kind, fallback ? "steward" : "project"); assert.equal(original.source.project_title, "project"); + assert.equal(original.source.executor_endpoint_id, fallback ? "antigravity" : "codex"); + assert.equal(original.session.agent_id, "codex"); 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"); @@ -74,12 +78,15 @@ try { 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); + let updated = history; + if (!fallback) { + child.stdin.write("source\n"); + assert.equal((await next()).source_turn_status, "completed"); + 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)); @@ -89,13 +96,23 @@ try { 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"}]) { + {...original.source, audience: "group"}, {...original.source, context_available: false}]) { 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; + if (fallback) { + for (const fault of ["fallback-invalid-allocation", "fallback-wrong-configured-endpoint"]) { + child.stdin.write(`${fault}\n`); + assert.equal((await next()).fallback_state, fault); + assert.equal((await originalFetch(new URL(`/api/chat/sessions?conversation_binding_id=${bindingId}`, origin))).status, 409); + await assert.rejects(() => resolveBoundConversation(target)); + child.stdin.write("fallback-restore\n"); + assert.equal((await next()).fallback_state, "fallback-restore"); + assert.equal((await resolveBoundConversation(target)).session.session_id, sessionId); + } + } 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}, @@ -120,15 +137,24 @@ try { 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.sessions, fallback ? 3 : 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)"); + console.log(fallback + ? "bound-conversation: passed (real HTTP/SQLite/Core steward fallback readback, original identity, invalid allocation/configured endpoint and revoked-source refusal; no model transition simulated)" + : "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; } } +if (!fallback) { + const alternate = spawn(process.execPath, [process.argv[1]], { + env: {...process.env, LOOPX_BOUND_STEWARD_FALLBACK: "1"}, stdio: "inherit", + }); + const [code] = await once(alternate, "exit"); + assert.equal(code, 0, "The same original-conversation journey must pass after quota fallback"); +} diff --git a/apps/presentation/dashboard/src/data/chat.ts b/apps/presentation/dashboard/src/data/chat.ts index 556fcf34dd..851e627cff 100644 --- a/apps/presentation/dashboard/src/data/chat.ts +++ b/apps/presentation/dashboard/src/data/chat.ts @@ -2813,9 +2813,8 @@ export async function resolveBoundConversation(target: BoundConversationTarget, 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."); - } + // Core qualifies this exact Session, including its original quota fallback. + // The connection's configured preference need not be its active executor. return {session, source}; }