fix(chat): open the Core-qualified steward conversation after quota fallback

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

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");
+}

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};
 }