Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion apps/presentation/dashboard/package.json
Original file line number Diff line number Diff line change
Expand Up @@ -80,7 +80,8 @@
"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",
"smoke:delegation-search": "LOOPX_PLAYWRIGHT_PACKAGE=\"$PWD/node_modules/playwright\" node smoke/delegation-search-browser-smoke.mjs",
"smoke:workspace-command-palette": "node --experimental-strip-types --test src/features/personal-workspace/workspace-command-entries.test.mjs && node --experimental-strip-types smoke/workspace-command-palette-browser-smoke.mjs && node --experimental-strip-types smoke/workspace-navigation-browser-smoke.mjs"
"smoke:workspace-command-palette": "node --experimental-strip-types --test src/features/personal-workspace/workspace-command-entries.test.mjs && node --experimental-strip-types smoke/workspace-command-palette-browser-smoke.mjs && node --experimental-strip-types smoke/workspace-navigation-browser-smoke.mjs",
"smoke:app-steward": "vite build --ssr smoke/app-steward-smoke.ts --outDir node_modules/.cache/loopx-app-steward --emptyOutDir && node node_modules/.cache/loopx-app-steward/app-steward-smoke.js"
},
"dependencies": {
"@fontsource-variable/geist": "^5.3.0",
Expand Down
94 changes: 94 additions & 0 deletions apps/presentation/dashboard/smoke/app-steward-smoke.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,94 @@
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 {fetchAppStewardTarget, fetchPrivateConversations, selectAppSteward,
fetchChatHistory, resolveBoundChatSession} from "../src/data/chat";

const repoRoot = resolve(process.cwd(), "../../..");
const child = spawn(resolveTestPython({repoRoot}), ["-u", "apps/presentation/dashboard/smoke/bound-conversation-http-fixture.py", "--steward"],
{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", value => {stderr += String(value);});
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 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();
globalThis.fetch = (input, init) => originalFetch(new URL(String(input), origin), init);
let connections = await fetchPrivateConversations();
const inspect = async () => {child.stdin.write("inspect\n"); return next();};
const before = await inspect();
assert.equal(before.sessions, 3);
assert.equal(await fetchAppStewardTarget(), null, "No implicit selection between two original stewards");
let peer = connections.connections.find(row => row.app_ref === "synthetic-peer")!;
assert.ok(peer && peer.binding_id !== bindingId);
connections = await selectAppSteward(bindingId, connections.revision);
const target = {bindingId, sessionId};
assert.deepEqual(await fetchAppStewardTarget(), target);
await assert.rejects(() => selectAppSteward(peer.binding_id, connections.revision - 1), /selection could not/);
assert.deepEqual(await fetchAppStewardTarget(), target, "Stale selection cannot retarget the App");
const history = await fetchChatHistory({bound: target});
assert.ok(history.messages.length === 2 && history.messages.every(row => !row.text.includes("Different private steward")));
connections = await selectAppSteward(peer.binding_id, connections.revision);
const peerTarget = await fetchAppStewardTarget();
assert.equal(peerTarget?.bindingId, peer.binding_id);
assert.notEqual(peerTarget?.sessionId, sessionId);
assert.equal((await resolveBoundChatSession(target)).session_id, sessionId,
"An already opened original Session stays fixed after the entry preference changes");
connections = await selectAppSteward(bindingId, connections.revision);
child.stdin.write("deny\n"); await next();
await assert.rejects(() => fetchAppStewardTarget(), /selected original steward is unavailable/);
await assert.rejects(() => selectAppSteward(bindingId, connections.revision), /selection could not/);
child.stdin.write("recover\n"); await next();
assert.deepEqual(await fetchAppStewardTarget(), target);
const after = await inspect();
assert.equal(after.sessions, before.sessions);
assert.deepEqual(after.messages, before.messages);
assert.equal(after.native_requests, before.native_requests, "Entry lookup and preference never invoke a model or replay a Turn");
child.stdin.write("replace-peer\n");
const replaced = await next();
connections = await fetchPrivateConversations();
const futurePeer = connections.connections.find(row => row.app_ref === "synthetic-peer")!;
assert.notEqual(futurePeer.binding_id, peer.binding_id);
assert.equal(futurePeer.binding_id, replaced.new_peer_binding_id);
assert.deepEqual(await fetchAppStewardTarget(), target, "A newly connected steward cannot capture the selected entry");
peer = futurePeer;
child.stdin.write("close\n"); await next();
await assert.rejects(() => fetchAppStewardTarget(), /selected original steward is unavailable/);
child.stdin.write("revoke\n"); await next();
connections = await fetchPrivateConversations();
assert.equal(connections.app_steward_binding_id, bindingId);
await assert.rejects(() => fetchAppStewardTarget(), /selected original steward is unavailable/);
assert.equal((await fetchPrivateConversations()).connections[0].binding_id, peer.binding_id);
connections = await selectAppSteward(null, connections.revision);
assert.equal(await fetchAppStewardTarget(), null, "Only an explicit clear restores the ordinary entry");
const routedFetch = globalThis.fetch;
for (const malformed of [{ok: true}, {ok: true, binding_id: null, session_id: sessionId},
{ok: true, binding_id: bindingId, session_id: "../other"}]) {
globalThis.fetch = async () => new Response(JSON.stringify(malformed));
await assert.rejects(() => fetchAppStewardTarget());
}
globalThis.fetch = routedFetch;
child.stdin.write("stop\n");
const [code] = await exited;
assert.equal(code, 0, stderr);
console.log("app-steward: passed (real HTTP/Core, explicit choice between sources, original Session, stale/identity/revocation refusal, no replay or fallback)");
} finally {
globalThis.fetch = originalFetch;
output.close();
if (child.exitCode === null) {child.kill("SIGINT"); await exited;}
}
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
from __future__ import annotations

import json
import os
from pathlib import Path
import runpy
import socket
Expand All @@ -16,7 +17,8 @@
import pytest

REPO = Path(__file__).resolve().parents[4]
sys.path[:0] = [str(REPO), str(REPO / "tests/control_plane")]
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

Expand All @@ -25,12 +27,16 @@ class SyntheticPrivateSource:
transport_ref = "synthetic-private"
cancelled = staticmethod(lambda: False)

def __init__(self, root):
def __init__(self, root, *, steward=False, peer=False):
self.root, self.verified = root, True
self.steward, self.peer = steward, peer
self.peer_provider = None
if peer:
self.transport_ref = "synthetic-peer"

def observe(self):
return {"transport_ref": self.transport_ref, "provider_ref": "c" * 24,
"operator_ref": "d" * 24, "verified": self.verified}
return {"transport_ref": self.transport_ref, "provider_ref": ("e" if self.peer else "c") * 24,
"operator_ref": ("f" if self.peer else "d") * 24, "verified": self.verified}

def health_snapshot(self):
return {"status": "listening"}
Expand All @@ -45,12 +51,16 @@ def start(self, server):
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}
executor_endpoint_id="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}
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"]
if self.peer:
server.chat_store.append_message(sid, role="agent", text="Different private steward: keep this source separate.")
return
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",
Expand All @@ -73,10 +83,28 @@ def control():
elif command == "close":
server.chat_store.update_session(sid, status="closed")
result = {"closed": True}
elif command == "replace-peer" and self.peer_provider is not None:
old = next(row for row in bindings.read()["bindings"] if row["transport_ref"] == self.peer_provider.transport_ref)
bindings.disconnect(old["binding_id"], expected_revision=bindings.read()["revision"])
self.peer_provider.start(server)
fresh = next(row for row in bindings.read()["bindings"] if row["transport_ref"] == self.peer_provider.transport_ref)
result = {"old_peer_binding_id": old["binding_id"], "new_peer_binding_id": fresh["binding_id"]}
elif command == "revoke":
bindings.disconnect(binding["binding_id"], expected_revision=bindings.read()["revision"])
result = {"revoked": binding["binding_id"]}
elif command == "populate":
for index in range(6):
server.chat_store.append_message(sid, role="user", text=f"Synthetic source request {index + 1}: 保留原任务与尚未确认的结果。")
server.chat_store.append_message(sid, role="agent", text=f"### 合成报告 {index + 1} / Synthetic report\n\n"
+ "\n\n".join(f"**工作 {part + 1}**:已验证的结果保留来源;仍在处理的事项继续保留,不把监听状态当作完成。 "
"Verified results retain their source. Unconfirmed returns remain open for recovery."
for part in range(6)))
result = {"populated": True, "messages": len(server.chat_store.messages(sid))}
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)}
"messages": server.chat_store.messages(sid),
"native_requests": len((self.root / "synthetic-requests.jsonl").read_text().splitlines())}
elif command == "stop":
server.shutdown()
return
Expand All @@ -93,15 +121,22 @@ def main():
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"])
model = runpy.run_path(str(REPO / "examples/loopx-chat-runtime-smoke.py"))["FAKE_CODEX"]
model = model.replace(' method = request.get("method")',
f' with open({str(root / "synthetic-requests.jsonl")!r}, "a") as output:\n'
' output.write(json.dumps(request) + "\\n")\n method = request.get("method")')
fake.write_text(model)
fake.chmod(0o700)
source = SyntheticPrivateSource(root)
steward = "--steward" in sys.argv
source = SyntheticPrivateSource(root, steward=steward)
peer = SyntheticPrivateSource(root, steward=True, peer=True) if steward else None
source.peer_provider = peer
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,))
external_conversation_factories=((lambda _: source, lambda _: peer) if peer else (lambda _: source,)))


if __name__ == "__main__":
Expand Down
17 changes: 17 additions & 0 deletions apps/presentation/dashboard/src/data/chat.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2691,11 +2691,28 @@ const privateConversationStopRecoverySchema = z.object({binding_id: z.string(),
project_title: z.string(), pending_count: z.number().int().positive()});
export type PrivateConversationStopRecovery = z.infer<typeof privateConversationStopRecoverySchema>;
const privateConversationsSchema = z.object({ok: z.literal(true), revision: z.number().int(),
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<typeof privateConversationSchema>;
export async function fetchPrivateConversations(signal?: AbortSignal) {
return privateConversationsSchema.parse(await requestJson<unknown>("/api/chat/lark/private-conversations", { signal }));
}
const appStewardTargetSchema = z.union([
z.object({ok: z.literal(true), binding_id: z.null(), session_id: z.null()}),
z.object({ok: z.literal(true), binding_id: z.string().regex(/^[a-f0-9]{24}$/),
session_id: z.string().regex(/^[A-Za-z0-9._-]{1,160}$/)}),
]);
export async function fetchAppStewardTarget(signal?: AbortSignal): Promise<BoundConversationTarget | null> {
const target = appStewardTargetSchema.parse(await requestJson<unknown>("/api/chat/app-steward", {signal}));
return target.binding_id === null ? null : {bindingId: target.binding_id, sessionId: target.session_id};
}
export async function selectAppSteward(bindingId: string | null, revision: number) {
const result = privateConversationsSchema.parse(await requestJson<unknown>("/api/chat/app-steward", {
method: "POST", headers: {"Content-Type": "application/json"}, body: JSON.stringify({binding_id: bindingId, revision}),
}));
if (typeof window !== "undefined") window.dispatchEvent(new Event("loopx:app-steward-changed"));
return result;
}
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<unknown>("/api/chat/lark/private-conversations", {
method: "POST", headers: {"Content-Type": "application/json"},
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,48 @@
import {useEffect, useRef, useState, type ReactNode} from "react";
import {Button} from "../../components/ui/button";
import {fetchAppStewardTarget, type BoundConversationTarget} from "../../data/chat";
import {useWorkspaceI18n} from "./i18n";

/** Only mount at the local, unbound manager Chat entry. Children must not mount
* until discovery finishes: they otherwise expose the ordinary manager queue.
* Once opened, the route freezes the exact Session and this boundary unmounts.
*/
export function AppStewardEntry({children, onOpen}: {children: ReactNode; onOpen: (target: BoundConversationTarget) => void}) {
const {locale} = useWorkspaceI18n();
const zh = locale === "zh-CN";
const [state, setState] = useState<"loading" | "ordinary" | "unavailable">("loading");
const [revision, setRevision] = useState(0);
const open = useRef(onOpen);
open.current = onOpen;
useEffect(() => {
const changed = () => {setState("loading"); setRevision(value => value + 1);};
window.addEventListener("loopx:app-steward-changed", changed);
return () => window.removeEventListener("loopx:app-steward-changed", changed);
}, []);
useEffect(() => {
let active = true;
const controller = new AbortController();
const timeout = setTimeout(() => controller.abort(), 5000);
void fetchAppStewardTarget(controller.signal).then(target => {
if (!active) return;
if (target) open.current(target);
else setState("ordinary");
}).catch(() => {if (active) setState("unavailable");}).finally(() => clearTimeout(timeout));
return () => {active = false; controller.abort(); clearTimeout(timeout);};
}, [revision]);
if (state === "ordinary") return children;
const settings = new URL(window.location.href);
settings.searchParams.set("view", "overview");
settings.searchParams.set("settings_scope", "machine");
settings.searchParams.set("settings_tab", "lark");
return <main className="mx-auto flex min-h-dvh max-w-xl flex-col justify-center gap-5 p-6 text-slate-900 dark:text-zinc-100">
<h1 className="text-xl font-semibold">{zh ? "LoopX 管家" : "LoopX steward"}</h1>
{state === "loading" ? <p role="status">{zh ? "正在打开管家对话…" : "Opening the steward conversation…"}</p> : <>
<p role="alert">{zh ? "原管家会话暂不可用。请检查所选连接后重试。" : "The original steward conversation is unavailable. Check the selected connection and retry."}</p>
<div className="flex flex-wrap gap-3">
<Button onClick={() => {setState("loading"); setRevision(value => value + 1);}}>{zh ? "重试" : "Retry"}</Button>
<a className="inline-flex min-h-9 items-center underline underline-offset-4" href={settings.href}>{zh ? "打开会话设置" : "Open conversation settings"}</a>
</div>
</>}
</main>;
}
Loading
Loading