From 60cd6a7d836a8282eca74e243c3ba7f6c0c3beca Mon Sep 17 00:00:00 2001 From: bo Date: Tue, 25 Aug 2026 19:13:27 +0800 Subject: [PATCH 1/2] fix: harden runtime recovery and QA reliability --- .github/fixtures/smoke-config.json | 4 + .github/scripts/smoke-compiled-binary.ts | 416 ++++++++++++++++++ .github/workflows/ci.yml | 30 +- .github/workflows/release.yml | 39 +- apps/server/src/app.test.ts | 15 + apps/server/src/app.ts | 8 + apps/server/src/routes/projects.test.ts | 23 +- apps/web/src/api/mcp.test.ts | 8 +- apps/web/src/api/mcp.ts | 8 +- .../bootstrap/BootstrapGate.test.tsx | 3 +- .../components/bootstrap/BootstrapGate.tsx | 36 +- .../features/SettingsDialog.interaction.tsx | 48 +- .../components/features/settings-panels.tsx | 36 +- apps/web/src/context/global-sse.test.tsx | 40 ++ apps/web/src/context/global-sse.tsx | 17 + .../query/hooks/todo-continuation.test.ts | 21 +- .../agents/query/hooks/todo-continuation.ts | 2 +- .../agent-core/src/agents/query/loop.test.ts | 12 + packages/agent-core/src/agents/query/loop.ts | 57 ++- .../src/agents/query/recovery.test.ts | 90 ++++ packages/agent-core/src/index.ts | 2 +- .../src/lsp/compat-spike.integration.test.ts | 74 ++-- packages/agent-core/src/main.test.ts | 18 + .../agent-core/src/projects/registry.test.ts | 28 ++ packages/agent-core/src/projects/registry.ts | 35 +- packages/agent-core/src/runtime.ts | 6 + packages/protocol/src/guards.test.ts | 11 + packages/protocol/src/guards.ts | 11 + packages/protocol/src/types.test.ts | 10 + packages/protocol/src/types.ts | 7 + 30 files changed, 949 insertions(+), 166 deletions(-) create mode 100644 .github/scripts/smoke-compiled-binary.ts diff --git a/.github/fixtures/smoke-config.json b/.github/fixtures/smoke-config.json index 29c2bb60..3caf9719 100644 --- a/.github/fixtures/smoke-config.json +++ b/.github/fixtures/smoke-config.json @@ -22,6 +22,10 @@ } } }, + "mcp": { + "disabledBuiltins": ["context7", "grep.app", "exa"], + "servers": {} + }, "profiles": { "principal": { "model": "smoke:test" diff --git a/.github/scripts/smoke-compiled-binary.ts b/.github/scripts/smoke-compiled-binary.ts new file mode 100644 index 00000000..a3b133ac --- /dev/null +++ b/.github/scripts/smoke-compiled-binary.ts @@ -0,0 +1,416 @@ +import { copyFile, chmod, mkdir, mkdtemp, rm } from "node:fs/promises"; +import { createServer } from "node:net"; +import { tmpdir } from "node:os"; +import { join, resolve } from "node:path"; + +const STARTUP_TIMEOUT_MS = 30_000; +const REQUEST_TIMEOUT_MS = 5_000; +const SSE_TIMEOUT_MS = 10_000; + +interface RunningServer { + readonly baseUrl: string; + readonly process: ReturnType; + readonly stdout: Promise; + readonly stderr: Promise; +} + +interface SseEvent { + readonly event: string; + readonly data: unknown; +} + +class SseReader { + readonly #abort: AbortController; + readonly #reader: ReadableStreamDefaultReader; + readonly #decoder = new TextDecoder(); + readonly #queued: SseEvent[] = []; + #buffer = ""; + + private constructor(response: Response, abort: AbortController) { + if (response.body === null) throw new Error("Global SSE response did not include a body"); + this.#abort = abort; + this.#reader = response.body.getReader(); + } + + static async connect(baseUrl: string): Promise { + const abort = new AbortController(); + try { + const response = await withTimeout(fetch(`${baseUrl}/api/events`, { + headers: { accept: "text/event-stream" }, + signal: abort.signal, + }), REQUEST_TIMEOUT_MS, "connecting to global SSE stream"); + if (!response.ok) throw new Error(`Global SSE endpoint returned HTTP ${response.status}`); + return new SseReader(response, abort); + } catch (error) { + abort.abort(); + throw error; + } + } + + async waitFor( + predicate: (event: SseEvent) => boolean, + label: string, + ): Promise { + const deadline = Date.now() + SSE_TIMEOUT_MS; + while (Date.now() < deadline) { + const event = await this.#next(deadline - Date.now()); + if (predicate(event)) return event; + } + throw new Error(`Timed out waiting for SSE event: ${label}`); + } + + async close(): Promise { + this.#abort.abort(); + await this.#reader.cancel().catch(() => undefined); + } + + async #next(timeoutMs: number): Promise { + while (this.#queued.length === 0) { + const result = await withTimeout(this.#reader.read(), timeoutMs, "reading global SSE stream"); + if (result.done) throw new Error("Global SSE stream closed before the expected event"); + this.#buffer += this.#decoder.decode(result.value, { stream: true }); + this.#drainFrames(); + } + return this.#queued.shift()!; + } + + #drainFrames(): void { + this.#buffer = this.#buffer.replaceAll("\r\n", "\n"); + while (true) { + const boundary = this.#buffer.indexOf("\n\n"); + if (boundary === -1) return; + const frame = this.#buffer.slice(0, boundary); + this.#buffer = this.#buffer.slice(boundary + 2); + const lines = frame.split("\n"); + const event = lines.find((line) => line.startsWith("event: "))?.slice(7); + const data = lines + .filter((line) => line.startsWith("data: ")) + .map((line) => line.slice(6)) + .join("\n"); + if (event === undefined || data.length === 0) continue; + this.#queued.push({ event, data: JSON.parse(data) }); + } + } +} + +async function main(): Promise { + const repositoryRoot = resolve(import.meta.dir, "../.."); + const binaryOverride = process.env.ARCHCODE_SMOKE_BINARY?.trim(); + const expectedVersion = process.env.ARCHCODE_SMOKE_EXPECTED_VERSION?.trim(); + const binaryPath = binaryOverride === undefined || binaryOverride.length === 0 + ? join(repositoryRoot, "dist", "archcode") + : resolve(binaryOverride); + const configFixture = join(repositoryRoot, ".github", "fixtures", "smoke-config.json"); + const root = await mkdtemp(join(tmpdir(), "archcode-compiled-smoke-")); + const homeDir = join(root, "home"); + const workspaceRoot = join(root, "workspace"); + let server: RunningServer | undefined; + let events: SseReader | undefined; + + try { + await mkdir(join(homeDir, ".archcode"), { recursive: true }); + await mkdir(workspaceRoot, { recursive: true }); + const configPath = join(homeDir, ".archcode", "config.json"); + await copyFile(configFixture, configPath); + await chmod(configPath, 0o600); + + server = await startServer(binaryPath, homeDir); + const initialHealth = await assertBootstrapReady(server.baseUrl); + assertExpectedVersion(initialHealth, expectedVersion); + await assertEmbeddedWebUi(server.baseUrl); + + const initialProjects = asRecord(await requestJson(server.baseUrl, "/api/projects")); + assert(Array.isArray(initialProjects.projects) && initialProjects.projects.length === 0, + "Expected an isolated empty project registry"); + + const project = asRecord(await requestJson(server.baseUrl, "/api/projects", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ workspaceRoot, name: "Compiled Smoke" }), + }, 201)); + assert(typeof project.slug === "string" && project.slug.length > 0, "Project response did not include a slug"); + assert(project.workspaceRoot === workspaceRoot, "Project response did not preserve the workspace root"); + const slug = encodeURIComponent(project.slug); + + events = await SseReader.connect(server.baseUrl); + await events.waitFor((event) => { + const data = asOptionalRecord(event.data); + return event.event === "session.runtime.snapshot" + && Array.isArray(data?.projectSlugs) + && data.projectSlugs.includes(project.slug); + }, "initial Session runtime snapshot"); + await events.waitFor((event) => { + const data = asOptionalRecord(event.data); + return event.event === "hitl.snapshot" + && Array.isArray(data?.projectSlugs) + && data.projectSlugs.includes(project.slug) + && Array.isArray(data?.entries); + }, "initial HITL snapshot"); + + const renamedProject = asRecord(await requestJson(server.baseUrl, `/api/projects/${slug}`, { + method: "PATCH", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ name: "Compiled Smoke Renamed" }), + })); + assert(renamedProject.name === "Compiled Smoke Renamed", "Project rename did not return the updated name"); + await events.waitFor((event) => { + const data = asOptionalRecord(event.data); + return event.event === "project.catalog_changed" + && data?.type === "project.catalog_changed" + && typeof data?.createdAt === "number"; + }, "live project catalog change"); + + const hitl = asRecord(await requestJson(server.baseUrl, `/api/projects/${slug}/hitl?status=all`)); + assert(Array.isArray(hitl.hitl) && hitl.hitl.length === 0, "Expected a new project to have no HITL records"); + + const createdTodoResponse = asRecord(await requestJson(server.baseUrl, `/api/projects/${slug}/todos`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ content: "Compiled smoke Todo" }), + }, 201)); + const createdTodo = asRecord(createdTodoResponse.todo); + assert(typeof createdTodo.id === "string", "Todo response did not include an id"); + assert(createdTodo.status === "idea" && createdTodo.revision === 1, "Todo was not created in its initial state"); + + await events.waitFor((event) => { + const data = asOptionalRecord(event.data); + return event.event === "resource.changed" + && data?.projectSlug === project.slug + && data?.resourceType === "todo" + && data?.resourceId === createdTodo.id; + }, "live Todo resource change"); + + const updatedTodoResponse = asRecord(await requestJson( + server.baseUrl, + `/api/projects/${slug}/todos/${encodeURIComponent(String(createdTodo.id))}`, + { + method: "PATCH", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ expectedRevision: 1, status: "ready" }), + }, + )); + const updatedTodo = asRecord(updatedTodoResponse.todo); + assert(updatedTodo.status === "ready" && updatedTodo.revision === 2, "Todo mutation was not persisted"); + + const session = asRecord(await requestJson(server.baseUrl, `/api/projects/${slug}/sessions`, { + method: "POST", + }, 201)); + assert(typeof session.sessionId === "string" && session.sessionId.length > 0, + "Session response did not include an id"); + assert(session.rootSessionId === session.sessionId && session.agentName === "lead", + "Direct Session identity was not created correctly"); + assert(Array.isArray(session.messages) && session.messages.length === 0, + "New direct Session should have an empty transcript"); + const sessionId = encodeURIComponent(session.sessionId); + + await assertPersistedApiState(server.baseUrl, slug, "Compiled Smoke Renamed", createdTodo.id, sessionId); + + await events.close(); + events = undefined; + await stopServer(server); + server = undefined; + + server = await startServer(binaryPath, homeDir); + const restartedHealth = await assertBootstrapReady(server.baseUrl); + assertExpectedVersion(restartedHealth, expectedVersion); + await assertPersistedApiState(server.baseUrl, slug, "Compiled Smoke Renamed", createdTodo.id, sessionId); + await assertEmbeddedWebUi(server.baseUrl); + + console.log("Compiled binary smoke passed: bootstrap, project catalog SSE, HITL snapshot, Todo, Session, and restart persistence."); + } catch (error) { + if (server !== undefined) { + await stopServer(server).catch(() => undefined); + const [stdout, stderr] = await Promise.all([server.stdout, server.stderr]); + if (stdout.length > 0) console.error(`\n--- archcode stdout ---\n${stdout}`); + if (stderr.length > 0) console.error(`\n--- archcode stderr ---\n${stderr}`); + } + throw error; + } finally { + await events?.close().catch(() => undefined); + if (server !== undefined) await stopServer(server).catch(() => undefined); + await rm(root, { recursive: true, force: true }); + } +} + +async function assertBootstrapReady(baseUrl: string): Promise> { + const deadline = Date.now() + STARTUP_TIMEOUT_MS; + let lastObservation = "server did not respond"; + while (Date.now() < deadline) { + try { + const health = asRecord(await requestJson(baseUrl, "/api/health")); + assert(health.ok === true, "Health endpoint did not report ok"); + const bootstrap = asRecord(await requestJson(baseUrl, "/api/bootstrap")); + const runtime = asOptionalRecord(bootstrap.runtime); + lastObservation = JSON.stringify(bootstrap); + if (bootstrap.mode === "ready" + && bootstrap.authRequired === false + && bootstrap.authenticated === true + && runtime?.state === "ready") return health; + if (runtime?.state === "error") throw new Error(`Runtime activation failed: ${lastObservation}`); + } catch (error) { + lastObservation = error instanceof Error ? error.message : String(error); + } + await Bun.sleep(100); + } + throw new Error(`Compiled binary did not become ready within ${STARTUP_TIMEOUT_MS}ms: ${lastObservation}`); +} + +function assertExpectedVersion( + health: Record, + expectedVersion: string | undefined, +): void { + if (expectedVersion === undefined || expectedVersion.length === 0) return; + assert( + health.version === expectedVersion, + `Health endpoint reported version ${String(health.version)}; expected ${expectedVersion}`, + ); +} + +async function assertEmbeddedWebUi(baseUrl: string): Promise { + const response = await fetch(`${baseUrl}/`, { signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS) }); + const html = await response.text(); + assert(response.status === 200, `Embedded Web UI returned HTTP ${response.status}`); + assert(html.includes('
'), "Embedded Web UI did not include the React root"); +} + +async function assertPersistedApiState( + baseUrl: string, + slug: string, + projectName: string, + todoId: unknown, + sessionId: string, +): Promise { + const projects = asRecord(await requestJson(baseUrl, "/api/projects")); + assert(Array.isArray(projects.projects) + && projects.projects.some((candidate) => { + const project = asOptionalRecord(candidate); + return project?.slug === decodeURIComponent(slug) && project?.name === projectName; + }), + "Registered project was not readable"); + + const todos = asRecord(await requestJson(baseUrl, `/api/projects/${slug}/todos`)); + assert(Array.isArray(todos.todos) + && todos.todos.some((candidate) => { + const todo = asOptionalRecord(candidate); + return todo?.id === todoId && todo?.status === "ready" && todo?.revision === 2; + }), + "Mutated Todo was not readable"); + + const session = asRecord(await requestJson(baseUrl, `/api/projects/${slug}/sessions/${sessionId}`)); + assert(encodeURIComponent(String(session.sessionId)) === sessionId, "Created Session was not readable"); + + const inventory = asRecord(await requestJson(baseUrl, `/api/projects/${slug}/sessions`)); + assert(Array.isArray(inventory.sessions) + && inventory.sessions.some((candidate) => { + const item = asOptionalRecord(candidate); + return asOptionalRecord(item?.session)?.sessionId === session.sessionId; + }), + "Created Session was missing from project inventory"); + + const tree = asRecord(await requestJson(baseUrl, `/api/projects/${slug}/sessions/${sessionId}/tree`)); + assert(asOptionalRecord(asOptionalRecord(tree.root)?.session)?.sessionId === session.sessionId, + "Created Session was missing from the Agent Tree projection"); + + const hitl = asRecord(await requestJson(baseUrl, `/api/projects/${slug}/hitl?status=all`)); + assert(Array.isArray(hitl.hitl) && hitl.hitl.length === 0, "Unexpected HITL records after restart"); +} + +async function startServer(binaryPath: string, homeDir: string): Promise { + const port = await reservePort(); + const child = Bun.spawn([binaryPath], { + env: { + ...process.env, + HOME: homeDir, + ARCHCODE_PORT: String(port), + ARCHCODE_LOG_LEVEL: "info", + ARCHCODE_ACCESS_LOG: "off", + }, + stdin: "ignore", + stdout: "pipe", + stderr: "pipe", + }); + return { + baseUrl: `http://127.0.0.1:${port}`, + process: child, + stdout: child.stdout instanceof ReadableStream ? new Response(child.stdout).text() : Promise.resolve(""), + stderr: child.stderr instanceof ReadableStream ? new Response(child.stderr).text() : Promise.resolve(""), + }; +} + +async function stopServer(server: RunningServer): Promise { + if (server.process.exitCode !== null) return; + server.process.kill("SIGTERM"); + try { + await withTimeout(server.process.exited, 15_000, "compiled binary shutdown"); + } catch (error) { + if (server.process.exitCode === null) server.process.kill("SIGKILL"); + await withTimeout(server.process.exited, 5_000, "forced compiled binary shutdown").catch(() => undefined); + throw error; + } +} + +async function reservePort(): Promise { + const server = createServer(); + await new Promise((resolvePromise, reject) => { + server.once("error", reject); + server.listen(0, "127.0.0.1", resolvePromise); + }); + const address = server.address(); + assert(address !== null && typeof address === "object", "Could not reserve a local port"); + const port = address.port; + await new Promise((resolvePromise, reject) => server.close((error) => error ? reject(error) : resolvePromise())); + return port; +} + +async function requestJson( + baseUrl: string, + path: string, + init: RequestInit = {}, + expectedStatus = 200, +): Promise { + const response = await fetch(`${baseUrl}${path}`, { + ...init, + signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS), + }); + const text = await response.text(); + if (response.status !== expectedStatus) { + throw new Error(`${init.method ?? "GET"} ${path} returned HTTP ${response.status}: ${text}`); + } + try { + return JSON.parse(text); + } catch { + throw new Error(`${init.method ?? "GET"} ${path} did not return JSON: ${text}`); + } +} + +async function withTimeout(promise: Promise, timeoutMs: number, label: string): Promise { + let timer: ReturnType | undefined; + try { + return await Promise.race([ + promise, + new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error(`Timed out while ${label}`)), Math.max(timeoutMs, 1)); + }), + ]); + } finally { + if (timer !== undefined) clearTimeout(timer); + } +} + +function asRecord(value: unknown): Record { + const record = asOptionalRecord(value); + if (record === undefined) throw new Error(`Expected an object, received ${JSON.stringify(value)}`); + return record; +} + +function asOptionalRecord(value: unknown): Record | undefined { + return typeof value === "object" && value !== null && !Array.isArray(value) + ? value as Record + : undefined; +} + +function assert(condition: unknown, message: string): asserts condition { + if (!condition) throw new Error(message); +} + +await main(); diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 2709a80e..a4a9d093 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -46,32 +46,4 @@ jobs: run: bun run build - name: Smoke test compiled binary - env: - ARCHCODE_PORT: 41967 - run: | - mkdir -p ~/.archcode - install -m 600 .github/fixtures/smoke-config.json ~/.archcode/config.json - ./dist/archcode > "$RUNNER_TEMP/archcode-smoke.log" 2>&1 & - server_pid=$! - cleanup() { - kill "$server_pid" 2>/dev/null || true - wait "$server_pid" 2>/dev/null || true - } - trap cleanup EXIT - - ready=0 - for attempt in 1 2 3 4 5 6 7 8 9 10; do - if curl --silent --fail http://127.0.0.1:41967/api/health > "$RUNNER_TEMP/archcode-health.json"; then - ready=1 - break - fi - sleep 1 - done - - if [ "$ready" -ne 1 ]; then - sed -n '1,160p' "$RUNNER_TEMP/archcode-smoke.log" - exit 1 - fi - - curl --silent --fail http://127.0.0.1:41967/ > "$RUNNER_TEMP/archcode-index.html" - bun -e 'const html = await Bun.file(process.argv[1]).text(); if (!html.includes("
")) process.exit(1)' "$RUNNER_TEMP/archcode-index.html" + run: bun .github/scripts/smoke-compiled-binary.ts diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 10888756..fa171f5c 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -181,42 +181,9 @@ jobs: "$binary" --help | grep -F "Usage: archcode [options]" "$binary" --help | grep -F -- "--port " - smoke_home="$RUNNER_TEMP/archcode-home" - mkdir -p "$smoke_home/.archcode" - install -m 600 .github/fixtures/smoke-config.json "$smoke_home/.archcode/config.json" - - HOME="$smoke_home" "$binary" --port 41967 > "$RUNNER_TEMP/archcode-smoke.log" 2>&1 & - server_pid=$! - cleanup() { - kill "$server_pid" 2>/dev/null || true - wait "$server_pid" 2>/dev/null || true - } - trap cleanup EXIT - - ready=0 - for attempt in 1 2 3 4 5 6 7 8 9 10; do - if curl --silent --fail http://127.0.0.1:41967/api/health > "$RUNNER_TEMP/archcode-health.json"; then - ready=1 - break - fi - sleep 1 - done - - if [[ "$ready" -ne 1 ]]; then - sed -n '1,160p' "$RUNNER_TEMP/archcode-smoke.log" - exit 1 - fi - - bun -e ' - const health = await Bun.file(process.argv[1]).json(); - if (health.ok !== true || health.version !== process.argv[2]) process.exit(1); - ' "$RUNNER_TEMP/archcode-health.json" "$VERSION" - - curl --silent --fail http://127.0.0.1:41967/ > "$RUNNER_TEMP/archcode-index.html" - bun -e ' - const html = await Bun.file(process.argv[1]).text(); - if (!html.includes("
")) process.exit(1); - ' "$RUNNER_TEMP/archcode-index.html" + ARCHCODE_SMOKE_BINARY="$binary" \ + ARCHCODE_SMOKE_EXPECTED_VERSION="$VERSION" \ + bun .github/scripts/smoke-compiled-binary.ts - name: Upload target archive uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 diff --git a/apps/server/src/app.test.ts b/apps/server/src/app.test.ts index f315de92..b0f9d0d4 100644 --- a/apps/server/src/app.test.ts +++ b/apps/server/src/app.test.ts @@ -11,6 +11,7 @@ const mockRuntime = { subscribeSessionRuntimeChanges: mock(() => () => undefined), subscribeMcpStatusChanges: mock(() => () => undefined), subscribeModelRuntimeChanges: mock(() => () => undefined), + subscribeProjectCatalogChanges: mock(() => () => undefined), getMcpServerStatus: mock(() => ({ servers: {} })), getMcpServerInventory: mock(() => ({ servers: {} })), } as unknown as AgentRuntime; @@ -95,4 +96,18 @@ describe("createRuntimeApp", () => { expect(observed[0]).toEqual({ type: "model_runtime.changed", revision: "revision-2", createdAt: 2 }); unsubscribe(); }); + + test("bridges project catalog changes", () => { + let listener: ((event: Extract) => void) | undefined; + const runtime = { + ...mockRuntime, + subscribeProjectCatalogChanges: mock((next: typeof listener) => { listener = next; return () => undefined; }), + } as unknown as AgentRuntime; + const observed: GlobalSSEEvent[] = []; + const unsubscribe = globalEventBus.subscribe((event) => observed.push(event)); + createRuntimeApp(runtime); + listener!({ type: "project.catalog_changed", createdAt: 3 }); + expect(observed[0]).toEqual({ type: "project.catalog_changed", createdAt: 3 }); + unsubscribe(); + }); }); diff --git a/apps/server/src/app.ts b/apps/server/src/app.ts index ae4d1d8f..c6f258bd 100644 --- a/apps/server/src/app.ts +++ b/apps/server/src/app.ts @@ -112,6 +112,7 @@ export function createRuntimeApp( wireSessionRuntimeBridge(serverRuntime, globalEventBus); wireMcpStatusBridge(serverRuntime, globalEventBus); wireModelRuntimeBridge(serverRuntime, globalEventBus); + wireProjectCatalogChangeBridge(serverRuntime, globalEventBus); wireResourceChangeBridge(serverRuntime, globalEventBus); return { app, runtime: serverRuntime }; @@ -166,6 +167,13 @@ function wireModelRuntimeBridge( runtime.subscribeModelRuntimeChanges((event) => bus.emit(event)); } +function wireProjectCatalogChangeBridge( + runtime: AgentRuntime, + bus: { emit(event: GlobalSSEEvent): void }, +): void { + runtime.subscribeProjectCatalogChanges?.((event) => bus.emit(event)); +} + function wireResourceChangeBridge( runtime: AgentRuntime, bus: { emit(event: GlobalSSEEvent): void }, diff --git a/apps/server/src/routes/projects.test.ts b/apps/server/src/routes/projects.test.ts index 05000830..a57a2cc3 100644 --- a/apps/server/src/routes/projects.test.ts +++ b/apps/server/src/routes/projects.test.ts @@ -4,7 +4,7 @@ import { join, resolve } from "node:path"; import type { AgentRuntime } from "@archcode/agent-core"; import { ProjectRegistry, ProjectRuntimeActiveError, silentLogger } from "@archcode/agent-core"; import type { ProjectInfo } from "@archcode/agent-core"; -import type { GlobalSSEEvent } from "@archcode/protocol"; +import type { GlobalSSEEvent, GlobalSSEProjectCatalogChangedEvent } from "@archcode/protocol"; import { createRuntimeApp } from "../app"; import { globalEventBus } from "../events/global-event-bus"; @@ -76,6 +76,10 @@ function createTestRuntime( }, subscribeHitlEvents: () => () => undefined, subscribeSessionRuntimeChanges: () => () => undefined, + subscribeProjectCatalogChanges: (listener: (event: GlobalSSEProjectCatalogChangedEvent) => void) => projectRegistry.subscribeCatalogChanges(() => listener({ + type: "project.catalog_changed", + createdAt: Date.now(), + })), createSession: async () => ({ sessionId: "session", title: null, createdAt: Date.now(), messages: [], steps: [], todos: [], reminders: [] }), getSessionFile: async (_workspaceRoot: string, sessionId: string) => ({ sessionId, title: null, createdAt: Date.now(), messages: [], steps: [], todos: [], reminders: [] }), listSessions: async () => [], @@ -185,6 +189,10 @@ describe("projects routes", () => { expect(response.status).toBe(201); expect(observed).toEqual([ + { + type: "project.catalog_changed", + createdAt: expect.any(Number), + }, { type: "session.runtime.snapshot", projectSlugs: ["alpha"], @@ -375,6 +383,7 @@ describe("projects routes", () => { expect(res.status).toBe(200); expect(await res.json()).toEqual({ ok: true }); expect(events).toEqual([ + expect.objectContaining({ type: "project.catalog_changed" }), expect.objectContaining({ type: "session.runtime.snapshot", projectSlugs: [project.slug], families: [] }), expect.objectContaining({ type: "hitl.snapshot", projectSlugs: [project.slug], entries: [] }), ]); @@ -428,15 +437,21 @@ describe("projects routes", () => { }); const project = (await created.json()) as ProjectInfo; + const events: GlobalSSEEvent[] = []; + const unsubscribe = globalEventBus.subscribe((event) => events.push(event)); const res = await app.request(`/api/projects/${project.slug}`, { method: "PATCH", body: JSON.stringify({ name: "Renamed" }), headers: { "content-type": "application/json" }, }); + unsubscribe(); const body = (await res.json()) as ProjectInfo; expect(res.status).toBe(200); expect(body).toEqual({ ...project, name: "Renamed" }); + expect(events).toEqual([ + expect.objectContaining({ type: "project.catalog_changed" }), + ]); }); test("PATCH /api/projects/:slug rejects unknown body fields", async () => { @@ -544,13 +559,19 @@ describe("projects routes", () => { }); const project = (await created.json()) as ProjectInfo; + const events: GlobalSSEEvent[] = []; + const unsubscribe = globalEventBus.subscribe((event) => events.push(event)); const res = await app.request(`/api/projects/${project.slug}/touch`, { method: "POST" }); + unsubscribe(); const body = (await res.json()) as ProjectInfo; expect(res.status).toBe(200); expect(body.slug).toBe(project.slug); expect(typeof body.lastOpenedAt).toBe("string"); expect(body.lastOpenedAt).not.toBe(project.lastOpenedAt); + expect(events).toEqual([ + expect.objectContaining({ type: "project.catalog_changed" }), + ]); }); test("POST /api/projects/:slug/touch for non-existent slug returns 404 ProjectNotFoundError", async () => { diff --git a/apps/web/src/api/mcp.test.ts b/apps/web/src/api/mcp.test.ts index d7b23281..c3972064 100644 --- a/apps/web/src/api/mcp.test.ts +++ b/apps/web/src/api/mcp.test.ts @@ -30,8 +30,12 @@ describe("MCP control actions", () => { test("loads inventory and reconnects only by saved server identity", async () => { globalThis.document = { cookie: "" } as Document; + const controller = new AbortController(); const fetchMock = mock(async (input: RequestInfo | URL, init?: RequestInit) => { - if (String(input) === "/api/mcp/inventory") return jsonResponse({ servers: { local: [] } }); + if (String(input) === "/api/mcp/inventory") { + expect(init?.signal).toBe(controller.signal); + return jsonResponse({ servers: { local: [] } }); + } expect(String(input)).toBe("/api/mcp/reconnect/local"); expect(init?.method).toBe("POST"); expect(init?.body).toBeUndefined(); @@ -39,7 +43,7 @@ describe("MCP control actions", () => { }); globalThis.fetch = fetchMock as unknown as typeof fetch; - await expect(getMcpInventory()).resolves.toEqual({ local: [] }); + await expect(getMcpInventory({ signal: controller.signal })).resolves.toEqual({ local: [] }); await expect(reconnectMcpServer("local")).resolves.toEqual({ local: { state: "connecting", startedAt: 1 } }); }); }); diff --git a/apps/web/src/api/mcp.ts b/apps/web/src/api/mcp.ts index 2ecc70c2..3ed52ce4 100644 --- a/apps/web/src/api/mcp.ts +++ b/apps/web/src/api/mcp.ts @@ -15,8 +15,12 @@ export async function getMcpStatus(): Promise { return res.servers; } -export async function getMcpInventory(): Promise { - const response = await apiFetch("/api/mcp/inventory"); +export async function getMcpInventory( + options: { signal?: AbortSignal } = {}, +): Promise { + const response = await apiFetch("/api/mcp/inventory", { + signal: options.signal, + }); return response.servers ?? {}; } diff --git a/apps/web/src/components/bootstrap/BootstrapGate.test.tsx b/apps/web/src/components/bootstrap/BootstrapGate.test.tsx index a4e11704..1422237c 100644 --- a/apps/web/src/components/bootstrap/BootstrapGate.test.tsx +++ b/apps/web/src/components/bootstrap/BootstrapGate.test.tsx @@ -163,6 +163,7 @@ describe("BootstrapGate", () => { test("opens Config Recovery inside the restricted Settings shell with a terminal grant", async () => { dom.reconfigure({ url: "http://localhost/config-recovery#token=recovery-token" }); + window.localStorage.setItem("archcodeTheme", "dark"); globalThis.fetch = mock(async (input: RequestInfo | URL, init?: RequestInit) => { if (String(input) === "/api/bootstrap") return Response.json({ mode: "config_error", @@ -180,7 +181,7 @@ describe("BootstrapGate", () => { }) as unknown as typeof fetch; await act(async () => { - root.render(

Workbench mounted

); + root.render(

Workbench mounted

); await Promise.resolve(); }); diff --git a/apps/web/src/components/bootstrap/BootstrapGate.tsx b/apps/web/src/components/bootstrap/BootstrapGate.tsx index 43af89c8..b04d9578 100644 --- a/apps/web/src/components/bootstrap/BootstrapGate.tsx +++ b/apps/web/src/components/bootstrap/BootstrapGate.tsx @@ -16,18 +16,11 @@ type GateState = { kind: "loading" } | { kind: "error"; message: string } | { ki const primaryButton = "inline-flex h-9 items-center justify-center gap-2 rounded-sm bg-brand px-4 text-[12px] font-semibold text-brand-ink transition-colors duration-[var(--motion-fast)] hover:bg-brand-hover focus-visible:outline-none focus-visible:ring-2 focus-visible:ring-brand disabled:cursor-not-allowed disabled:opacity-40"; const secondaryButton = "inline-flex h-9 items-center justify-center gap-2 rounded-sm bg-bg-active px-4 text-[12px] font-semibold text-text-secondary transition-colors duration-[var(--motion-fast)] hover:bg-bg-hover hover:text-text-primary focus-visible:outline-none focus-visible:ring-2 focus-visible:ring-brand disabled:cursor-not-allowed disabled:opacity-40"; -let terminalGrantFromFragment: string | undefined; - -function readTerminalGrant(): string | undefined { - if (terminalGrantFromFragment !== undefined) return terminalGrantFromFragment; +function readTerminalGrantFromFragment(): string | undefined { if (typeof window === "undefined") return undefined; const params = new URLSearchParams(window.location.hash.slice(1)); const token = params.get("token")?.trim(); - if (token) { - terminalGrantFromFragment = token; - window.history.replaceState(null, "", `${window.location.pathname}${window.location.search}`); - } - return terminalGrantFromFragment; + return token || undefined; } function emptySetupConfig(): ServerConfigUpdate { @@ -49,6 +42,7 @@ export function BootstrapGate({ onAuthInvalidated?: () => void; }) { const [state, setState] = useState({ kind: "loading" }); + const [terminalGrant, setTerminalGrant] = useState(readTerminalGrantFromFragment); const reload = useCallback(async () => { setState({ kind: "loading" }); try { @@ -61,6 +55,10 @@ export function BootstrapGate({ }, []); useEffect(() => { void reload(); }, [reload]); + useEffect(() => { + if (terminalGrant === undefined || typeof window === "undefined" || window.location.hash === "") return; + window.history.replaceState(null, "", `${window.location.pathname}${window.location.search}`); + }, [terminalGrant]); useEffect( () => subscribeAuthInvalidation(() => { onAuthInvalidated?.(); @@ -79,16 +77,14 @@ export function BootstrapGate({ const { status } = state; if (status.mode === "setup") { - const grant = readTerminalGrant(); - return grant - ? + return terminalGrant + ? setTerminalGrant(undefined)} onComplete={reload} /> : ; } if (status.mode === "config_error") { - const grant = readTerminalGrant(); - return grant - ? { - if (next.mode === "ready") setTerminalGrantConsumed(); + return terminalGrant + ? { + if (next.mode === "ready") setTerminalGrant(undefined); normalizeBootstrapPath(next); setState({ kind: "status", status: next }); }} /> @@ -178,7 +174,7 @@ function LoginPage({ onLoggedIn }: { onLoggedIn: () => Promise }) { ; } -function SetupPage({ grant, onComplete }: { grant: string; onComplete: () => Promise }) { +function SetupPage({ grant, onGrantConsumed, onComplete }: { grant: string; onGrantConsumed: () => void; onComplete: () => Promise }) { const [config, setConfig] = useState(emptySetupConfig); const [adapterCatalog, setAdapterCatalog] = useState(); const [loadingCatalog, setLoadingCatalog] = useState(true); @@ -228,7 +224,7 @@ function SetupPage({ grant, onComplete }: { grant: string; onComplete: () => Pro ? { config, requireLogin: true, password } : { config, requireLogin: false }; await completeSetup(grant, request); - setTerminalGrantConsumed(); + onGrantConsumed(); await onComplete(); } catch (cause) { setFieldErrors(toFieldErrors(cause)); @@ -262,10 +258,6 @@ function SetupPage({ grant, onComplete }: { grant: string; onComplete: () => Pro ; } -function setTerminalGrantConsumed() { - terminalGrantFromFragment = undefined; -} - function normalizeBootstrapPath(status: BootstrapStatus): void { if (typeof window === "undefined") return; const pathname = window.location.pathname; diff --git a/apps/web/src/components/features/SettingsDialog.interaction.tsx b/apps/web/src/components/features/SettingsDialog.interaction.tsx index e195edca..875f4b78 100644 --- a/apps/web/src/components/features/SettingsDialog.interaction.tsx +++ b/apps/web/src/components/features/SettingsDialog.interaction.tsx @@ -52,6 +52,14 @@ function installDom() { container = document.createElement("div"); document.body.append(container); root = createRoot(container); } function click(label: string) { const element = [...container.querySelectorAll("button")].find((button) => button.textContent === label); if (!element) throw new Error(`Missing ${label}`); act(() => element.click()); } +async function clickAndFlush(label: string) { + const element = [...container.querySelectorAll("button")].find((button) => button.textContent === label); + if (!element) throw new Error(`Missing ${label}`); + await act(async () => { + element.click(); + await new Promise((resolve) => setTimeout(resolve, 0)); + }); +} function input(label: string, index = 0) { const fields = [...container.querySelectorAll("label")].filter((element) => !element.closest("[hidden]") && element.querySelector("span")?.textContent === label); const element = fields[index]?.querySelector("input, textarea, select") as HTMLInputElement | HTMLTextAreaElement | HTMLSelectElement | null; @@ -381,7 +389,7 @@ describe("SettingsDialog interactions", () => { expect(input("Variants JSON")).not.toBeNull(); }); - test("never overwrites sparse generated provider, model, or MCP identifiers", () => { + test("never overwrites sparse generated provider, model, or MCP identifiers", async () => { const sparse = structuredClone(snapshot); sparse.config.provider["provider-3"] = { ...structuredClone(snapshot.config.provider.local), @@ -394,7 +402,7 @@ describe("SettingsDialog interactions", () => { click("Add provider"); click("Add model"); expect(input("Display name", 1).value).toBe("Existing provider three"); - click("MCP"); + await clickAndFlush("MCP"); click("Add MCP server"); expect(container.textContent).toContain("provider-4"); @@ -677,10 +685,10 @@ describe("SettingsDialog interactions", () => { expect(input("Variant").value).toBe("fast"); }); - test("locks secret-bearing identities and still renames entries without preserved secrets", () => { + test("locks secret-bearing identities and still renames entries without preserved secrets", async () => { act(() => root.render( {}} />)); expect((input("Provider ID") as HTMLInputElement).readOnly).toBe(true); - click("MCP"); + await clickAndFlush("MCP"); expect((input("Name") as HTMLInputElement).readOnly).toBe(true); expect((input("Transport") as HTMLSelectElement).disabled).toBe(true); @@ -688,7 +696,7 @@ describe("SettingsDialog interactions", () => { const serverWithoutSecrets = withoutMcpSecrets.config.mcp!.servers.custom; if (serverWithoutSecrets.type === "http") delete serverWithoutSecrets.headers; act(() => root.render( {}} />)); - click("MCP"); + await clickAndFlush("MCP"); const name = input("Name"); change(name, "renamed"); expect(container.textContent).toContain("Delete custom"); @@ -803,9 +811,9 @@ describe("SettingsDialog interactions", () => { }); - test("keeps built-in MCP rows locked in the rendered DOM", () => { + test("keeps built-in MCP rows locked in the rendered DOM", async () => { act(() => root.render( {}} />)); - click("MCP"); + await clickAndFlush("MCP"); expect(container.textContent).toContain("Built-in"); expect(container.textContent).toContain("Ready"); expect(container.textContent).toContain("Not reported"); @@ -815,11 +823,11 @@ describe("SettingsDialog interactions", () => { expect(container.textContent).not.toContain("Delete exa"); }); - test("offers draft Test for built-ins but blocks Reconnect while disabled", () => { + test("offers draft Test for built-ins but blocks Reconnect while disabled", async () => { const disabled = structuredClone(snapshot); disabled.config.mcp!.disabledBuiltins = ["context7"]; act(() => root.render( {}} />)); - click("MCP"); + await clickAndFlush("MCP"); const row = [...container.querySelectorAll("article")].find((article) => article.querySelector("h2")?.textContent === "context7"); if (!row) throw new Error("Missing context7 MCP row"); @@ -831,7 +839,7 @@ describe("SettingsDialog interactions", () => { expect(reconnectButton.title).toContain("Enable and save"); }); - test("drops blank STDIO argument lines from the draft", () => { + test("drops blank STDIO argument lines from the draft", async () => { const stdio = structuredClone(snapshot); stdio.config.mcp!.servers.custom = { type: "stdio", @@ -840,13 +848,31 @@ describe("SettingsDialog interactions", () => { args: [], }; act(() => root.render( {}} />)); - click("MCP"); + await clickAndFlush("MCP"); const args = input("Arguments (one per line)"); change(args, "--first\n\n \n--second\n"); expect(input("Arguments (one per line)").value).toBe("--first\n--second"); }); + test("aborts an in-flight MCP inventory request when leaving the panel", async () => { + let inventorySignal: AbortSignal | undefined; + Object.defineProperty(globalThis, "fetch", { configurable: true, value: mock(async (url: string, init?: RequestInit) => { + if (url !== "/api/mcp/inventory") throw new Error(`Unexpected request: ${url}`); + inventorySignal = init?.signal ?? undefined; + return await new Promise((_resolve, reject) => { + inventorySignal?.addEventListener("abort", () => reject(new DOMException("Aborted", "AbortError")), { once: true }); + }); + }) }); + act(() => root.render( {}} />)); + + await clickAndFlush("MCP"); + expect(inventorySignal?.aborted).toBe(false); + + await clickAndFlush("Models"); + expect(inventorySignal?.aborted).toBe(true); + }); + test("aborts an in-flight MCP draft test when leaving the panel", async () => { let draftSignal: AbortSignal | undefined; Object.defineProperty(globalThis, "fetch", { configurable: true, value: mock(async (url: string, init?: RequestInit) => { diff --git a/apps/web/src/components/features/settings-panels.tsx b/apps/web/src/components/features/settings-panels.tsx index e8dccc1f..68f1dfa3 100644 --- a/apps/web/src/components/features/settings-panels.tsx +++ b/apps/web/src/components/features/settings-panels.tsx @@ -484,6 +484,7 @@ export function SettingsMcpPanel({ config, savedConfig = config, expectedRevisio const [pendingActions, setPendingActions] = useState>(() => new Set()); const pendingActionsRef = useRef(new Set()); const draftTestControllersRef = useRef(new Map()); + const mountedRef = useRef(true); const updateServer = useMcpStatusStore((state) => state.updateServer); const inventoryStatusKey = Object.entries(servers) .sort(([left], [right]) => left.localeCompare(right)) @@ -492,14 +493,16 @@ export function SettingsMcpPanel({ config, savedConfig = config, expectedRevisio useEffect(() => { if (!runtimeAvailable || !active) return; - let mounted = true; + const controller = new AbortController(); setInventoryError(undefined); - void getMcpInventory().then((next) => { - if (mounted) setInventory(next); + void getMcpInventory({ signal: controller.signal }).then((next) => { + if (!controller.signal.aborted) setInventory(next); }).catch((cause) => { - if (mounted) setInventoryError(cause instanceof Error ? cause.message : "Unable to load MCP tool inventory"); + if (!controller.signal.aborted) { + setInventoryError(cause instanceof Error ? cause.message : "Unable to load MCP tool inventory"); + } }); - return () => { mounted = false; }; + return () => controller.abort(); }, [active, expectedRevision, inventoryStatusKey, runtimeAvailable]); useEffect(() => { @@ -508,9 +511,13 @@ export function SettingsMcpPanel({ config, savedConfig = config, expectedRevisio draftTestControllersRef.current.clear(); }, [active, runtimeAvailable]); - useEffect(() => () => { - for (const controller of draftTestControllersRef.current.values()) controller.abort(); - draftTestControllersRef.current.clear(); + useEffect(() => { + mountedRef.current = true; + return () => { + mountedRef.current = false; + for (const controller of draftTestControllersRef.current.values()) controller.abort(); + draftTestControllersRef.current.clear(); + }; }, []); useEffect(() => { @@ -526,10 +533,16 @@ export function SettingsMcpPanel({ config, savedConfig = config, expectedRevisio setPendingActions((current) => new Set(current).add(key)); setActionErrors((current) => { const next = { ...current }; delete next[key]; return next; }); try { await action(); } - catch (cause) { setActionErrors((current) => ({ ...current, [key]: cause instanceof Error ? cause.message : "Action failed" })); } + catch (cause) { + if (mountedRef.current) { + setActionErrors((current) => ({ ...current, [key]: cause instanceof Error ? cause.message : "Action failed" })); + } + } finally { pendingActionsRef.current.delete(key); - setPendingActions((current) => { const next = new Set(current); next.delete(key); return next; }); + if (mountedRef.current) { + setPendingActions((current) => { const next = new Set(current); next.delete(key); return next; }); + } } }; @@ -596,7 +609,8 @@ export function SettingsMcpPanel({ config, savedConfig = config, expectedRevisio {!reconnectEnabled ? Enable and save before reconnecting. diff --git a/apps/web/src/context/global-sse.test.tsx b/apps/web/src/context/global-sse.test.tsx index 14a4f219..109c393a 100644 --- a/apps/web/src/context/global-sse.test.tsx +++ b/apps/web/src/context/global-sse.test.tsx @@ -6,6 +6,7 @@ import type { GlobalSSEHeartbeatEvent, GlobalSSEHitlRealtimeEvent, GlobalSSEMcpStatusEvent, + GlobalSSEProjectCatalogChangedEvent, GlobalSSEModelRuntimeChangedEvent, GlobalSSEResourceChangedEvent, GlobalSSEResetEvent, @@ -52,6 +53,7 @@ import { isSessionSnapshotQueryKey, parseSSEEvent, refreshProjectTodoQueriesAfterSSEOpen, + refreshProjectCatalogAfterSSEOpen, resolveHitlNoticeEntries, requestSSEReconnectOnce, requestSSEShutdownReconnectOnce, @@ -315,9 +317,32 @@ describe("SSE liveness watchdog", () => { expect(isProjectTodoQueryKey(["projects", 1, "todos"])).toBe(false); expect(invalidateQueries).toHaveBeenCalledTimes(1); }); + + test("actively refreshes the project catalog after every SSE open", async () => { + const invalidateQueries = mock(async () => undefined); + + await refreshProjectCatalogAfterSSEOpen({ + invalidateQueries, + } as unknown as Pick); + + expect(invalidateQueries).toHaveBeenCalledWith({ + queryKey: queryKeys.projects, + refetchType: "active", + }); + }); }); describe("parseSSEEvent", () => { + test("accepts only the exact project catalog event contract", () => { + const event: GlobalSSEProjectCatalogChangedEvent = { + type: "project.catalog_changed", + createdAt: 2, + }; + + expect(parseSSEEvent(event.type, JSON.stringify(event))).toEqual(event); + expect(parseSSEEvent(event.type, JSON.stringify({ ...event, projectSlug: "demo" }))).toBeNull(); + }); + test("parses valid event type", () => { const data = JSON.stringify({ type: "event", @@ -1006,6 +1031,21 @@ describe("handleSSEEvent", () => { ]); }); + test("invalidates only the process-wide project catalog for a catalog change", () => { + const event: GlobalSSEProjectCatalogChangedEvent = { + type: "project.catalog_changed", + createdAt: 2, + }; + + handleSSEEvent({ event: event.type, data: JSON.stringify(event) }, deps); + + expect(mockInvalidateQueries).toHaveBeenCalledTimes(1); + expect(mockInvalidateQueries).toHaveBeenCalledWith({ + queryKey: queryKeys.projects, + exact: true, + }); + }); + test("stores the scoped hitl.event view", () => { const event = hitlRealtimeEvent({ projectSlug: "proj", diff --git a/apps/web/src/context/global-sse.tsx b/apps/web/src/context/global-sse.tsx index 59956d0e..89deabc3 100644 --- a/apps/web/src/context/global-sse.tsx +++ b/apps/web/src/context/global-sse.tsx @@ -20,6 +20,7 @@ import { queryKeys } from "../api/queries"; import { isGlobalSSEHitlRealtimeEvent, isGlobalSSEHitlSnapshotEvent, + isGlobalSSEProjectCatalogChangedEvent, isGlobalSSEResourceChangedEvent, isGlobalSSEUpdateChangedEvent, isSessionEventPayload, @@ -365,6 +366,15 @@ export function refreshProjectTodoQueriesAfterSSEOpen( }); } +export function refreshProjectCatalogAfterSSEOpen( + queryClient: Pick, +): Promise { + return queryClient.invalidateQueries({ + queryKey: queryKeys.projects, + refetchType: "active", + }); +} + const GlobalSSEContext = createContext(null); export function parseSSEEvent(_event: string, data: string): GlobalSSEEvent | null { @@ -382,6 +392,8 @@ export function parseSSEEvent(_event: string, data: string): GlobalSSEEvent | nu case "session.runtime.snapshot": case "session.runtime_changed": return parsed; + case "project.catalog_changed": + return isGlobalSSEProjectCatalogChangedEvent(parsed) ? parsed : null; case "update.changed": return isGlobalSSEUpdateChangedEvent(parsed) ? parsed : null; case "hitl.snapshot": @@ -576,6 +588,10 @@ export function handleSSEEvent( invalidateResourceQueries(deps, parsed as GlobalSSEResourceChangedEvent); break; } + case "project.catalog_changed": { + deps.invalidateQueries({ queryKey: queryKeys.projects, exact: true }); + break; + } case "session.runtime.snapshot": { sessionRuntimeStore.getState().applySnapshot(parsed as GlobalSSESessionRuntimeSnapshotEvent); deps.refreshSessionSnapshots(); @@ -789,6 +805,7 @@ export function GlobalSSEProvider({ children }: { children: ReactNode }) { refreshSessionSnapshots(); void queryClient.invalidateQueries({ queryKey: queryKeys.modelRuntime }); void queryClient.invalidateQueries({ queryKey: queryKeys.update }); + void refreshProjectCatalogAfterSSEOpen(queryClient); void refreshProjectTodoQueriesAfterSSEOpen(queryClient); setConnectionState("open"); }, diff --git a/packages/agent-core/src/agents/query/hooks/todo-continuation.test.ts b/packages/agent-core/src/agents/query/hooks/todo-continuation.test.ts index f965bea3..c8b8d5e8 100644 --- a/packages/agent-core/src/agents/query/hooks/todo-continuation.test.ts +++ b/packages/agent-core/src/agents/query/hooks/todo-continuation.test.ts @@ -133,7 +133,7 @@ describe("createTodoContinuationHook - afterLoopEnd (loop continuation)", () => expect(store.getState().reminders).toHaveLength(0); }); - test("updates stagnation count when pending todo count does not decrease", async () => { + test("updates stagnation count when pending todo count stays the same", async () => { const store = createHookStore(); seedTodos(store, [{ id: "todo-1", content: "continue", status: "pending" }]); store.setState({ @@ -161,6 +161,25 @@ describe("createTodoContinuationHook - afterLoopEnd (loop continuation)", () => expect(store.getState().todoContinuationStagnationCount).toBe(0); }); + test("resets stagnation count when pending todo count increases", async () => { + const store = createHookStore(); + seedTodos(store, [ + { id: "todo-1", content: "continue", status: "pending" }, + { id: "todo-2", content: "new work", status: "pending" }, + ]); + store.setState({ + lastTodoContinuationPendingCount: 1, + todoContinuationStagnationCount: 2, + }); + const { afterLoopEnd } = createTodoContinuationHook(); + + await runLoopEnd(afterLoopEnd, store, "completed"); + + expect(store.getState().todoContinuationStagnationCount).toBe(0); + expect(store.getState().lastTodoContinuationPendingCount).toBe(2); + expect(store.getState().reminders).toHaveLength(1); + }); + test("blocks continuation when stagnation threshold reached", async () => { const store = createHookStore(); seedTodos(store, [{ id: "todo-1", content: "continue", status: "pending" }]); diff --git a/packages/agent-core/src/agents/query/hooks/todo-continuation.ts b/packages/agent-core/src/agents/query/hooks/todo-continuation.ts index 745e4c88..6e835297 100644 --- a/packages/agent-core/src/agents/query/hooks/todo-continuation.ts +++ b/packages/agent-core/src/agents/query/hooks/todo-continuation.ts @@ -48,7 +48,7 @@ function createTodoLoopContinuationHook( const pendingCount = checkResult.pendingTodos.length; const lastPendingCount = state.lastTodoContinuationPendingCount; const newStagnationCount = - lastPendingCount !== null && pendingCount >= lastPendingCount + lastPendingCount !== null && pendingCount === lastPendingCount ? state.todoContinuationStagnationCount + 1 : 0; diff --git a/packages/agent-core/src/agents/query/loop.test.ts b/packages/agent-core/src/agents/query/loop.test.ts index 57a85503..f84dfbbe 100644 --- a/packages/agent-core/src/agents/query/loop.test.ts +++ b/packages/agent-core/src/agents/query/loop.test.ts @@ -8,6 +8,7 @@ import { applySessionToolBatchResponse } from "../../execution/session-tool-batc import { HitlBoundaryCodec } from "../../hitl/boundary-codec"; import { setLlmAdapterForTest } from "../../llm/adapter"; import { silentLogger } from "../../logger"; +import { createMockLogger } from "../../logger.test-helper"; import type { ExecutionModelBinding } from "../../models"; import { SkillService } from "../../skills"; import { SessionStoreManager } from "../../store/session-store-manager"; @@ -900,6 +901,8 @@ describe("QueryLoop Tool Output Plane", () => { test("fails closed before model execution and leaves no Step when context preparation fails", async () => { const harness = await createHarness(); + const logger = createMockLogger(); + harness.options.logger = logger; harness.appendUser("prepare"); let modelCalls = 0; harness.options.prepareModelContext = async () => { @@ -918,6 +921,9 @@ describe("QueryLoop Tool Output Plane", () => { expect(streamEvents(harness)).toContain("execution-error"); expect(streamEvents(harness)).not.toContain("step-start"); expect(streamEvents(harness)).not.toContain("step-end"); + expect(logger.error).toHaveBeenCalledWith("query.loop.fatal", expect.objectContaining({ + error: expect.objectContaining({ message: "goal notice persistence failed" }), + })); }); test("projects provider-addressed text and reasoning blocks in their original order", async () => { @@ -1488,11 +1494,13 @@ describe("QueryLoop Tool Output Plane", () => { test("aborts a hung fullStream without waiting for the next chunk", async () => { const harness = await createHarness(); const controller = new AbortController(); + const logger = createMockLogger(); let signalStreamBlocked!: () => void; const streamBlocked = new Promise((resolve) => { signalStreamBlocked = resolve; }); harness.options.abort = controller.signal; + harness.options.logger = logger; harness.appendUser("run"); setLlmAdapterForTest({ @@ -1519,6 +1527,10 @@ describe("QueryLoop Tool Output Plane", () => { await expect(running).resolves.toMatchObject({ status: "aborted" }); expect(streamEvents(harness)).toContain("tool-input-start"); + expect(logger.error).not.toHaveBeenCalled(); + expect(logger.debug).toHaveBeenCalledWith("query.loop.aborted", expect.objectContaining({ + context: expect.objectContaining({ step: 0 }), + })); }); test("aborts hung finalize promises after the stream ends", async () => { diff --git a/packages/agent-core/src/agents/query/loop.ts b/packages/agent-core/src/agents/query/loop.ts index a06025d2..a1f6fcf4 100644 --- a/packages/agent-core/src/agents/query/loop.ts +++ b/packages/agent-core/src/agents/query/loop.ts @@ -62,7 +62,6 @@ type ModelAttemptResult = stepId: string; finalized: FinalizedModelResult; tools: ResolvedToolSet; - streamError?: unknown; } | { outcome: "terminal"; @@ -88,7 +87,7 @@ interface FinalizedModelResult { toolCalls?: ToolCallArray; } -type FinalizationKind = "result" | "toolCalls"; +type FinalizationKind = "stream" | "result" | "toolCalls"; type RetryOrTerminalAttemptResult = | Omit, "stepId"> @@ -230,19 +229,28 @@ async function runModelAttempt(options: ModelAttemptOptions): Promise, store, binding, stepId, abort, ); - const finalized = await finalizeModelResult(result, streamError, store, stepId, step, abort, redactProviderSecrets); + const finalized = await finalizeModelResult( + result, + streamError, + hasStreamError, + store, + stepId, + step, + abort, + redactProviderSecrets, + ); if (finalized.outcome !== "success") { if (finalized.outcome === "retry") await settleUnfinalizedToolParts(); return { ...finalized, stepId }; } - return { outcome: "success", stepId, finalized: finalized.finalized, tools: resolvedTools, streamError }; + return { outcome: "success", stepId, finalized: finalized.finalized, tools: resolvedTools }; } catch (err) { await settleModelResultPromises(result, abort); if (classifyLlmError(err, { boundary: "provider-request" }).kind === "abort") { @@ -652,6 +660,14 @@ export async function runQueryLoop( if (abort.aborted) { const err = abort.reason ?? new DOMException("Aborted", "AbortError"); const classification = classifyLlmError(err); + if (isStepOpen(store, attempt.stepId)) { + store.getState().append({ + type: "step-end", + stepId: attempt.stepId, + step: steps, + finishReason: "interrupted", + }); + } appendTerminalLlmFailureNotice(store, err, classification.kind, { steps, stepId: attempt.stepId, @@ -758,10 +774,16 @@ export async function runQueryLoop( failed = true; runEndStatus = abort.aborted ? "aborted" : "failed"; runEndError = safeError.message; - logger.error("query.loop.fatal", { - error: safeError, - context: { step: steps, sessionId, agentName }, - }); + if (abort.aborted) { + logger.debug("query.loop.aborted", { + context: { step: steps, sessionId, agentName }, + }); + } else { + logger.error("query.loop.fatal", { + error: safeError, + context: { step: steps, sessionId, agentName }, + }); + } store.getState().append(preparationFailed ? { type: "execution-error", @@ -857,8 +879,9 @@ async function consumeFullStream( binding: QueryLoopOptions["binding"], stepId: string, abort?: AbortSignal, -): Promise<{ streamError?: unknown }> { +): Promise<{ streamError?: unknown; hasStreamError: boolean }> { let streamError: unknown; + let hasStreamError = false; type OpenProviderBlock = { redactor: ReturnType; }; @@ -881,6 +904,7 @@ async function consumeFullStream( if (abort?.aborted) break; if (chunk.type === "error") { + hasStreamError = true; streamError = chunk.error; continue; } @@ -984,12 +1008,13 @@ async function consumeFullStream( } } - return { streamError }; + return { streamError, hasStreamError }; } async function finalizeModelResult( result: AnyStreamTextResult, streamError: unknown, + hasStreamError: boolean, store: StoreApi, stepId: string, step: number, @@ -1016,6 +1041,10 @@ async function finalizeModelResult( return handleFinalizationFailure(preferStreamError(streamError, err), store, stepId, step, abort, "result", redactProviderSecrets); } + if (hasStreamError) { + return handleFinalizationFailure(streamError, store, stepId, step, abort, "stream", redactProviderSecrets); + } + if (finishReason !== "tool-calls") { return { outcome: "success", finalized: { finishReason, usage, text } }; } @@ -1174,7 +1203,7 @@ function appendPostStreamTerminalFailure( err: unknown, stepId: string, step: number, - kind: "result" | "toolCalls", + kind: FinalizationKind, ): void { const message = errorMessage(err); const classification = classifyLlmError(err); @@ -1200,7 +1229,9 @@ function appendPostStreamTerminalFailure( }, { terminalNoRetry: true, profile: "post-stream-terminal", - message: `${kind === "toolCalls" ? "Model tool call" : "Model result"} finalization failed: ${message}`, + message: kind === "stream" + ? `Model stream failed: ${message}` + : `${kind === "toolCalls" ? "Model tool call" : "Model result"} finalization failed: ${message}`, }); } diff --git a/packages/agent-core/src/agents/query/recovery.test.ts b/packages/agent-core/src/agents/query/recovery.test.ts index 33c5ec28..7f05260d 100644 --- a/packages/agent-core/src/agents/query/recovery.test.ts +++ b/packages/agent-core/src/agents/query/recovery.test.ts @@ -374,6 +374,62 @@ describe("query loop LLM stream recovery", () => { expect(JSON.stringify(store.getState().messages)).not.toContain("recovery-notice"); }); + test("retryable fullStream error chunk enters recovery even when result promises resolve", async () => { + const store = createStore(); + const events = captureEvents(store); + const streamFn = createMockStreamText([ + { + chunks: [{ type: "error", error: retryableEof("unexpected EOF from error chunk") }], + finishReason: "stop", + text: "must not be accepted", + }, + { text: "Recovered answer" }, + ]); + + const result = await runQueryLoop(makeOptions({ store }), "Recover error chunk"); + + expect(result).toMatchObject({ status: "completed", text: "Recovered answer", steps: 1 }); + expect(streamFn).toHaveBeenCalledTimes(2); + expect(events).toContainEqual(expect.objectContaining({ + type: "llm-retry", + scope: "short", + visibility: "internal", + profile: "zero-output-short", + attempt: 1, + errorKind: "eof", + })); + expect(store.getState().steps.map((step) => step.finishReason)).toEqual(["interrupted", "stop"]); + expect(textParts(store).map((part) => part.text)).toEqual(["Recovered answer"]); + }); + + test("non-retryable fullStream error chunk is terminal even when result promises resolve", async () => { + const store = createStore(); + const events = captureEvents(store); + const streamFn = createMockStreamText([{ + chunks: [{ + type: "error", + error: Object.assign(new Error("Unauthorized error chunk"), { status: 401 }), + }], + finishReason: "stop", + text: "must not be accepted", + }]); + + const result = await runQueryLoop(makeOptions({ store }), "Reject error chunk"); + + expect(result).toMatchObject({ status: "failed", text: "", steps: 0, error: "Unauthorized error chunk" }); + expect(streamFn).toHaveBeenCalledTimes(1); + expect(events.filter((event) => event.type === "llm-retry")).toHaveLength(0); + expect(events).toContainEqual(expect.objectContaining({ + type: "llm-recovery-failed", + profile: "post-stream-terminal", + errorKind: "auth", + statusCode: 401, + message: "Model stream failed: Unauthorized error chunk", + })); + expect(store.getState().steps.map((step) => step.finishReason)).toEqual(["error"]); + expect(textParts(store)).toHaveLength(0); + }); + test("strictly reloads persisted same-run retry attempts", async () => { sessionFileInternals.saveSessionTranscript = realSaveSessionTranscript; const store = createStore(); @@ -620,6 +676,40 @@ describe("query loop LLM stream recovery", () => { })); }); + test("abort after a recovered model result closes the successful attempt Step", async () => { + const store = createStore(); + const abort = new AbortController(); + const append = store.getState().append; + store.setState({ + append: (event) => { + append(event); + if (event.type === "llm-recovery") { + abort.abort(new DOMException("User cancelled after model result", "AbortError")); + } + }, + }); + createMockStreamText([ + { throwBeforeOutput: retryableEof("first attempt failed") }, + { text: "model result raced with cancellation" }, + ]); + + const result = await runQueryLoop( + makeOptions({ store, abort: abort.signal }), + "Cancel after recovered result", + ); + + expect(result).toMatchObject({ outcome: "terminal", status: "aborted", text: "", steps: 0 }); + expect(store.getState().steps).toHaveLength(2); + expect(store.getState().steps.map((step) => step.finishReason)).toEqual(["interrupted", "interrupted"]); + expect(store.getState().steps.every((step) => step.completedAt !== undefined)).toBe(true); + expect(textParts(store)).toEqual([ + expect.objectContaining({ + text: "model result raced with cancellation", + meta: { interrupted: true, discardedFromContext: true }, + }), + ]); + }); + test("continuous session retry is uncapped while delay is capped and next retry time is emitted", async () => { const store = createStore(); const events = captureEvents(store); diff --git a/packages/agent-core/src/index.ts b/packages/agent-core/src/index.ts index 855bfbfb..526f6d12 100644 --- a/packages/agent-core/src/index.ts +++ b/packages/agent-core/src/index.ts @@ -114,7 +114,7 @@ export type { ConsoleLike, LogEntry, LogFields, Logger, LogLevel } from "./logge export { ProjectContextResolver } from "./projects/context-resolver"; export { ProjectRegistry, ProjectRegistryError } from "./projects/registry"; -export type { ProjectRegistrationResult } from "./projects/registry"; +export type { ProjectCatalogChangeListener, ProjectRegistrationResult } from "./projects/registry"; export { SessionLifecycleService } from "./projects/session-lifecycle-service"; export type { SessionLifecycleServiceOptions } from "./projects/session-lifecycle-service"; export type { ProjectContext, ProjectInfo } from "./projects/types"; diff --git a/packages/agent-core/src/lsp/compat-spike.integration.test.ts b/packages/agent-core/src/lsp/compat-spike.integration.test.ts index 5c4614eb..4eda494f 100644 --- a/packages/agent-core/src/lsp/compat-spike.integration.test.ts +++ b/packages/agent-core/src/lsp/compat-spike.integration.test.ts @@ -1,4 +1,4 @@ -import { describe, it, expect, afterAll, beforeAll } from "bun:test"; +import { describe, it, expect, afterAll, afterEach, beforeAll, beforeEach } from "bun:test"; import { mkdir } from "node:fs/promises"; import path from "node:path"; import type { Disposable } from "vscode-jsonrpc"; @@ -68,53 +68,61 @@ interface RALWritable { let child: ReturnType | undefined; let connection: ReturnType | undefined; +let initializeResult: any; + +beforeEach(async () => { + const serverPath = path.join(import.meta.dir, "fake-server.ts"); + child = Bun.spawn(["bun", "run", serverPath], { + cwd: testTempRoot.path, + stdin: "pipe", + stdout: "pipe", + stderr: "inherit", + env: { ...process.env, FAKE_LSP_CONFIG: JSON.stringify(DEFAULT_FAKE_LSP_CONFIG) }, + }); -afterAll(async () => { + const outStream = child.stdout as ReadableStream; + const inSink = child.stdin as any; + const reader = new ReadableStreamMessageReader(adaptReader(outStream)); + const writer = new WriteableStreamMessageWriter(adaptWriter(inSink)); + connection = createMessageConnection(reader, writer); + connection.listen(); + initializeResult = await connection.sendRequest("initialize", { + processId: null, + capabilities: {}, + rootUri: null, + }); +}); + +afterEach(async () => { if (connection) { try { connection.end(); } catch {} + connection = undefined; } if (child && child.exitCode === null) { child.kill(); - child = undefined; + await child.exited; } + child = undefined; + initializeResult = undefined; +}); + +afterAll(async () => { await testTempRoot.cleanup(); }); describe("vscode-jsonrpc compatibility spike", () => { it("can spawn fake server, send initialize, and receive response", async () => { - const serverPath = path.join(import.meta.dir, "fake-server.ts"); - child = Bun.spawn(["bun", "run", serverPath], { - cwd: testTempRoot.path, - stdin: "pipe", - stdout: "pipe", - stderr: "inherit", - env: { ...process.env, FAKE_LSP_CONFIG: JSON.stringify(DEFAULT_FAKE_LSP_CONFIG) }, - }); - - const outStream = child.stdout as ReadableStream; - const inSink = child.stdin as any; - const reader = new ReadableStreamMessageReader(adaptReader(outStream)); - const writer = new WriteableStreamMessageWriter(adaptWriter(inSink)); - connection = createMessageConnection(reader, writer); - connection.listen(); - - const result: any = await connection.sendRequest("initialize", { - processId: null, - capabilities: {}, - rootUri: null, - }); - - expect(result).toBeDefined(); - expect(result).toHaveProperty("capabilities"); - expect(result).toHaveProperty("serverInfo"); - expect(result.serverInfo).toEqual({ + expect(initializeResult).toBeDefined(); + expect(initializeResult).toHaveProperty("capabilities"); + expect(initializeResult).toHaveProperty("serverInfo"); + expect(initializeResult.serverInfo).toEqual({ name: "fake-lsp-server", version: "0.0.1", }); - expect(result.capabilities).toHaveProperty("textDocumentSync", 1); - expect(result.capabilities).toHaveProperty("definitionProvider", true); - expect(result.capabilities).not.toHaveProperty("hoverProvider"); - expect(result.capabilities).not.toHaveProperty("completionProvider"); + expect(initializeResult.capabilities).toHaveProperty("textDocumentSync", 1); + expect(initializeResult.capabilities).toHaveProperty("definitionProvider", true); + expect(initializeResult.capabilities).not.toHaveProperty("hoverProvider"); + expect(initializeResult.capabilities).not.toHaveProperty("completionProvider"); }); it("can send a notification", async () => { diff --git a/packages/agent-core/src/main.test.ts b/packages/agent-core/src/main.test.ts index 5df69e7d..7edec54a 100644 --- a/packages/agent-core/src/main.test.ts +++ b/packages/agent-core/src/main.test.ts @@ -5,6 +5,7 @@ import { join } from "node:path"; import { tmpdir } from "node:os"; import type { GlobalSessionEventEnvelope, + GlobalSSEProjectCatalogChangedEvent, GlobalSSESessionRuntimeChangedEvent, RequestedModelSelection, ServerConfigUpdate, @@ -1060,6 +1061,23 @@ describe("createRuntime", () => { expect(runtime.getMcpServerInventory()).toEqual({ servers: { docs: [] } }); }); test("exposes project registry and shared context resolver", async () => { const runtime = await createRuntime({ configService: await writeConfig(makeConfig()), mcpRuntimeFactory: () => makeFakeMcpRuntime() }); expect(runtime.projectRegistry).toBeDefined(); expect(runtime.contextResolver).toBeDefined(); }); + test("publishes project catalog changes from the Runtime registry facade", async () => { + const workspaceRoot = await makeTempRoot(); + const runtime = await createRuntime({ + configService: await writeConfig(makeConfig()), + mcpRuntimeFactory: () => makeFakeMcpRuntime(), + }); + const events: GlobalSSEProjectCatalogChangedEvent[] = []; + const unsubscribe = runtime.subscribeProjectCatalogChanges?.((event) => events.push(event)); + + await runtime.projectRegistry.add({ workspaceRoot, name: "Catalog event" }); + + expect(events).toEqual([{ + type: "project.catalog_changed", + createdAt: expect.any(Number), + }]); + unsubscribe?.(); + }); test("emits runtime snapshot without idle families", async () => { const workspaceRoot = await makeTempRoot(); const runtime = await createRuntime({ configService: await writeConfig(makeConfig()), mcpRuntimeFactory: () => makeFakeMcpRuntime() }); const project = await runtime.projectRegistry.add({ workspaceRoot, name: "Runtime snapshot" }); const session = await runtime.createSession(workspaceRoot, { agentName: "lead", source: { kind: "direct" } }); const changes: GlobalSSESessionRuntimeChangedEvent[] = []; const unsubscribe = runtime.subscribeSessionRuntimeChanges((event) => changes.push(event)); const events = await runtime.listSessionRuntimeEvents(); expect(events[0]).toMatchObject({ type: "session.runtime.snapshot", projectSlugs: [project.slug], families: [] }); await expect(runtime.stopSessionFamily(workspaceRoot, session.sessionId)).resolves.toBeUndefined(); expect(changes.map(({ activity }) => activity)).toEqual(["stopping", "idle"]); unsubscribe(); }); test("startup continuation recovery preserves persisted Session recency and content", async () => { diff --git a/packages/agent-core/src/projects/registry.test.ts b/packages/agent-core/src/projects/registry.test.ts index 1375fa5c..13a0e2dd 100644 --- a/packages/agent-core/src/projects/registry.test.ts +++ b/packages/agent-core/src/projects/registry.test.ts @@ -135,6 +135,34 @@ describe("ProjectRegistry", () => { await expect(registry.remove(projectA.slug)).resolves.toBeUndefined(); }); + test("publishes catalog changes only after durable add, rename, touch, and remove mutations", async () => { + const registry = new ProjectRegistry({ homeDir: tmpHome, logger: silentLogger }); + let changes = 0; + const unsubscribe = registry.subscribeCatalogChanges(() => { changes += 1; }); + + const project = await registry.add({ workspaceRoot: tmpWorkspaceA, name: "Original" }); + expect(changes).toBe(1); + await registry.add({ workspaceRoot: tmpWorkspaceA, name: "Ignored" }); + expect(changes).toBe(1); + const sameName = await registry.updateName(project.slug, "Original"); + expect(changes).toBe(1); + const persistedSameName = await registry.get(project.slug); + expect(persistedSameName).toBeDefined(); + expect(persistedSameName!).toEqual(sameName); + await registry.updateName(project.slug, "Renamed"); + expect(changes).toBe(2); + await registry.touch(project.slug); + expect(changes).toBe(3); + await registry.remove("missing"); + expect(changes).toBe(3); + await registry.remove(project.slug); + expect(changes).toBe(4); + + unsubscribe(); + await registry.add({ workspaceRoot: tmpWorkspaceB, name: "Not observed" }); + expect(changes).toBe(4); + }); + test("touch on missing slug returns undefined", async () => { const registry = new ProjectRegistry({ homeDir: tmpHome, logger: silentLogger }); diff --git a/packages/agent-core/src/projects/registry.ts b/packages/agent-core/src/projects/registry.ts index 9b20967b..926f5e73 100644 --- a/packages/agent-core/src/projects/registry.ts +++ b/packages/agent-core/src/projects/registry.ts @@ -52,8 +52,11 @@ const MAX_PROJECT_NAME_LENGTH = 80; type MutationResult = { result: T; updated: ProjectInfo[]; + changed: boolean; }; +export type ProjectCatalogChangeListener = () => void; + export interface ProjectRegistrationResult { readonly project: ProjectInfo; readonly created: boolean; @@ -105,6 +108,7 @@ export class ProjectRegistry { #cache: ProjectInfo[] | null = null; #writeQueue: Promise = Promise.resolve(); #logger: Logger; + #catalogChangeListeners = new Set(); constructor(options: ProjectRegistryOptions) { this.#indexFile = join(options.homeDir ?? homedir(), USER_DATA_DIR_NAME, "projects", "index.json"); @@ -115,6 +119,13 @@ export class ProjectRegistry { return sortProjects(await this.#load()); } + subscribeCatalogChanges(listener: ProjectCatalogChangeListener): () => void { + this.#catalogChangeListeners.add(listener); + return () => { + this.#catalogChangeListeners.delete(listener); + }; + } + async get(slug: string): Promise { const project = (await this.#load()).find((item) => item.slug === slug); return project ? cloneProject(project) : undefined; @@ -155,6 +166,7 @@ export class ProjectRegistry { return { result: { project: cloneProject(existing), created: false }, updated: current, + changed: false, }; } @@ -175,6 +187,7 @@ export class ProjectRegistry { return { result: { project: cloneProject(project), created: true }, updated: [...current, project], + changed: true, }; }); } @@ -187,6 +200,7 @@ export class ProjectRegistry { updated: removed === undefined ? current : current.filter((project) => project.slug !== slug), + changed: removed !== undefined, }; }); } @@ -216,6 +230,7 @@ export class ProjectRegistry { return { result: cloneProject(updatedProject), updated, + changed: updatedProject.name !== current[index]!.name, }; }); } @@ -224,7 +239,7 @@ export class ProjectRegistry { return await this.#mutate((current) => { const index = current.findIndex((project) => project.slug === slug); if (index === -1) { - return { result: undefined, updated: current }; + return { result: undefined, updated: current, changed: false }; } const updatedProject: ProjectInfo = { @@ -237,6 +252,7 @@ export class ProjectRegistry { return { result: cloneProject(updatedProject), updated, + changed: true, }; }); } @@ -278,7 +294,10 @@ export class ProjectRegistry { const operation = this.#writeQueue.then(async () => { const current = cloneProjects(await this.#load()); const mutation = await fn(current); - await this.#persist(mutation.updated); + if (mutation.changed) { + await this.#persist(mutation.updated); + this.#publishCatalogChanged(); + } result = mutation.result; }); @@ -290,4 +309,16 @@ export class ProjectRegistry { await operation; return result as T; } + + #publishCatalogChanged(): void { + for (const listener of this.#catalogChangeListeners) { + try { + listener(); + } catch (error) { + this.#logger.warn("project.registry.catalog_change.listener.failed", { + error, + }); + } + } + } } diff --git a/packages/agent-core/src/runtime.ts b/packages/agent-core/src/runtime.ts index 7d10649f..41e7775a 100644 --- a/packages/agent-core/src/runtime.ts +++ b/packages/agent-core/src/runtime.ts @@ -57,6 +57,7 @@ import type { GlobalSSEHitlEntry, GlobalSSEHitlSnapshotEvent, GlobalSSEModelRuntimeChangedEvent, + GlobalSSEProjectCatalogChangedEvent, GlobalSSEResourceChangedEvent, GlobalSessionEventEnvelope, GlobalSSESessionRuntimeChangedEvent, @@ -393,6 +394,7 @@ export interface AgentRuntime { getMemorySnapshot(workspaceRoot: string): Promise; subscribeSessionRuntimeChanges(listener: (event: GlobalSSESessionRuntimeChangedEvent) => void): () => void; subscribeModelRuntimeChanges(listener: (event: GlobalSSEModelRuntimeChangedEvent) => void): () => void; + subscribeProjectCatalogChanges?(listener: (event: GlobalSSEProjectCatalogChangedEvent) => void): () => void; subscribeResourceChanges?(listener: (event: GlobalSSEResourceChangedEvent) => void): () => void; subscribeMcpStatusChanges(listener: (serverName: string, status: McpServerStatus) => void): () => void; getMcpServerStatus(): McpServerStatusResponse; @@ -2431,6 +2433,10 @@ export async function createRuntime( revision: snapshot.revision, createdAt: Date.now(), })), + subscribeProjectCatalogChanges: (listener) => projectRegistry.subscribeCatalogChanges(() => listener({ + type: "project.catalog_changed", + createdAt: Date.now(), + })), subscribeResourceChanges: (listener) => { resourceChangeListeners.add(listener); return () => { diff --git a/packages/protocol/src/guards.test.ts b/packages/protocol/src/guards.test.ts index 8d927956..fd209597 100644 --- a/packages/protocol/src/guards.test.ts +++ b/packages/protocol/src/guards.test.ts @@ -2,6 +2,7 @@ import { describe, expect, test } from "bun:test"; import { isGlobalSSEHitlRealtimeEvent, isGlobalSSEHitlSnapshotEvent, + isGlobalSSEProjectCatalogChangedEvent, isGlobalSSEResourceChangedEvent, isGlobalSSEUpdateChangedEvent, isSessionEventPayload, @@ -745,6 +746,16 @@ describe("protocol event guards", () => { expect(isGlobalSSEResourceChangedEvent({ ...resourceEvent, resourceType: "goal", resourceId: "goal-1" })).toBe(false); expect(isGlobalSSEResourceChangedEvent({ ...resourceEvent, reason: "created" })).toBe(false); expect(isGlobalSSEResourceChangedEvent({ type: "resource.changed" })).toBe(false); + expect(isGlobalSSEProjectCatalogChangedEvent({ + type: "project.catalog_changed", + createdAt: 2, + })).toBe(true); + expect(isGlobalSSEProjectCatalogChangedEvent({ + type: "project.catalog_changed", + projectSlug: "project", + createdAt: 2, + })).toBe(false); + expect(isGlobalSSEProjectCatalogChangedEvent({ type: "project.catalog_changed" })).toBe(false); }); test("accepts only coherent update status change events", () => { diff --git a/packages/protocol/src/guards.ts b/packages/protocol/src/guards.ts index e4c5d19b..7aea4b0f 100644 --- a/packages/protocol/src/guards.ts +++ b/packages/protocol/src/guards.ts @@ -1,6 +1,7 @@ import type { GlobalSSEHitlRealtimeEvent, GlobalSSEHitlSnapshotEvent, + GlobalSSEProjectCatalogChangedEvent, GlobalSSEResourceChangedEvent, FinalizedToolResult, SessionEventPayload, @@ -589,6 +590,16 @@ export function isGlobalSSEResourceChangedEvent(value: unknown): value is Global && isFiniteNumber(event.createdAt); } +export function isGlobalSSEProjectCatalogChangedEvent( + value: unknown, +): value is GlobalSSEProjectCatalogChangedEvent { + const event = record(value); + return event !== undefined + && exact(event, ["type", "createdAt"]) + && event.type === "project.catalog_changed" + && isFiniteNumber(event.createdAt); +} + export function isGlobalSSEUpdateChangedEvent(value: unknown): value is GlobalSSEUpdateChangedEvent { const event = record(value); return event !== undefined diff --git a/packages/protocol/src/types.test.ts b/packages/protocol/src/types.test.ts index eb6f2f92..123054f9 100644 --- a/packages/protocol/src/types.test.ts +++ b/packages/protocol/src/types.test.ts @@ -16,6 +16,7 @@ import type { GlobalSSEEvent, GlobalSSEResourceChangedEvent, GlobalSSEHeartbeatEvent, + GlobalSSEProjectCatalogChangedEvent, GlobalSSELaggedEvent, GlobalSSEResetEvent, GlobalSSESessionRuntimeChangedEvent, @@ -223,6 +224,15 @@ describe("global SSE wire protocol types", () => { expect(serializeRoundTrip(events)).toEqual(events); }); + test("uses one process-wide project catalog invalidation event", () => { + const event: GlobalSSEProjectCatalogChangedEvent = { + type: "project.catalog_changed", + createdAt: 4, + }; + + expect(serializeRoundTrip(event)).toEqual(event); + }); + test("round-trips a global session event envelope", () => { const event: GlobalSessionEventEnvelope = { type: "event", diff --git a/packages/protocol/src/types.ts b/packages/protocol/src/types.ts index c3375a04..4bcdd861 100644 --- a/packages/protocol/src/types.ts +++ b/packages/protocol/src/types.ts @@ -1208,6 +1208,12 @@ export type GlobalSSEResourceChangedEvent = createdAt: number; }; +/** The process-wide registered project catalog changed and must be re-read. */ +export interface GlobalSSEProjectCatalogChangedEvent { + type: "project.catalog_changed"; + createdAt: number; +} + export type GlobalSSEEvent = | GlobalSessionEventEnvelope | GlobalSSEHeartbeatEvent @@ -1220,6 +1226,7 @@ export type GlobalSSEEvent = | GlobalSSESessionRuntimeChangedEvent | GlobalSSEHitlSnapshotEvent | GlobalSSEHitlRealtimeEvent + | GlobalSSEProjectCatalogChangedEvent | GlobalSSEResourceChangedEvent | GlobalSSEUpdateChangedEvent; From beb21331ffacecebcbb9dd99ef88e7f9021ca222 Mon Sep 17 00:00:00 2001 From: bo Date: Tue, 25 Aug 2026 19:28:41 +0800 Subject: [PATCH 2/2] fix(web): constrain reconnect refresh work --- .../features/SettingsDialog.interaction.tsx | 35 ++++++++++++++++ .../components/features/settings-panels.tsx | 41 ++++++++++++------- apps/web/src/context/global-sse.test.tsx | 1 + apps/web/src/context/global-sse.tsx | 1 + 4 files changed, 63 insertions(+), 15 deletions(-) diff --git a/apps/web/src/components/features/SettingsDialog.interaction.tsx b/apps/web/src/components/features/SettingsDialog.interaction.tsx index 875f4b78..8cb75307 100644 --- a/apps/web/src/components/features/SettingsDialog.interaction.tsx +++ b/apps/web/src/components/features/SettingsDialog.interaction.tsx @@ -873,6 +873,41 @@ describe("SettingsDialog interactions", () => { expect(inventorySignal?.aborted).toBe(true); }); + test("aborts the inventory refresh started by MCP reconnect when leaving the panel", async () => { + let inventoryRequests = 0; + let reconnectInventorySignal: AbortSignal | undefined; + Object.defineProperty(globalThis, "fetch", { configurable: true, value: mock(async (url: string, init?: RequestInit) => { + if (url === "/api/mcp/reconnect/custom") { + return Response.json({ servers: { custom: { state: "connecting", startedAt: 1 } } }); + } + if (url === "/api/mcp/inventory") { + inventoryRequests += 1; + if (inventoryRequests === 1) return Response.json({ servers: {} }); + reconnectInventorySignal = init?.signal ?? undefined; + return await new Promise((_resolve, reject) => { + reconnectInventorySignal?.addEventListener("abort", () => reject(new DOMException("Aborted", "AbortError")), { once: true }); + }); + } + throw new Error(`Unexpected request: ${url}`); + }) }); + act(() => root.render( {}} />)); + await clickAndFlush("MCP"); + + const customRow = [...container.querySelectorAll("article")] + .find((article) => article.querySelector("h2")?.textContent === "custom"); + const reconnectButton = [...(customRow?.querySelectorAll("button") ?? [])] + .find((button) => button.textContent === "Reconnect"); + if (!reconnectButton) throw new Error("Missing custom MCP Reconnect button"); + await act(async () => { + reconnectButton.click(); + await new Promise((resolve) => setTimeout(resolve, 0)); + }); + expect(reconnectInventorySignal?.aborted).toBe(false); + + await clickAndFlush("Models"); + expect(reconnectInventorySignal?.aborted).toBe(true); + }); + test("aborts an in-flight MCP draft test when leaving the panel", async () => { let draftSignal: AbortSignal | undefined; Object.defineProperty(globalThis, "fetch", { configurable: true, value: mock(async (url: string, init?: RequestInit) => { diff --git a/apps/web/src/components/features/settings-panels.tsx b/apps/web/src/components/features/settings-panels.tsx index 68f1dfa3..8915bf38 100644 --- a/apps/web/src/components/features/settings-panels.tsx +++ b/apps/web/src/components/features/settings-panels.tsx @@ -483,7 +483,7 @@ export function SettingsMcpPanel({ config, savedConfig = config, expectedRevisio const [inventoryError, setInventoryError] = useState(); const [pendingActions, setPendingActions] = useState>(() => new Set()); const pendingActionsRef = useRef(new Set()); - const draftTestControllersRef = useRef(new Map()); + const actionControllersRef = useRef(new Map()); const mountedRef = useRef(true); const updateServer = useMcpStatusStore((state) => state.updateServer); const inventoryStatusKey = Object.entries(servers) @@ -507,22 +507,22 @@ export function SettingsMcpPanel({ config, savedConfig = config, expectedRevisio useEffect(() => { if (active && runtimeAvailable) return; - for (const controller of draftTestControllersRef.current.values()) controller.abort(); - draftTestControllersRef.current.clear(); + for (const controller of actionControllersRef.current.values()) controller.abort(); + actionControllersRef.current.clear(); }, [active, runtimeAvailable]); useEffect(() => { mountedRef.current = true; return () => { mountedRef.current = false; - for (const controller of draftTestControllersRef.current.values()) controller.abort(); - draftTestControllersRef.current.clear(); + for (const controller of actionControllersRef.current.values()) controller.abort(); + actionControllersRef.current.clear(); }; }, []); useEffect(() => { - for (const controller of draftTestControllersRef.current.values()) controller.abort(); - draftTestControllersRef.current.clear(); + for (const controller of actionControllersRef.current.values()) controller.abort(); + actionControllersRef.current.clear(); setTestResults({}); setActionErrors({}); }, [config]); @@ -592,25 +592,36 @@ export function SettingsMcpPanel({ config, savedConfig = config, expectedRevisio
{!reconnectEnabled ? Enable and save before reconnecting. diff --git a/apps/web/src/context/global-sse.test.tsx b/apps/web/src/context/global-sse.test.tsx index 109c393a..2982cd33 100644 --- a/apps/web/src/context/global-sse.test.tsx +++ b/apps/web/src/context/global-sse.test.tsx @@ -327,6 +327,7 @@ describe("SSE liveness watchdog", () => { expect(invalidateQueries).toHaveBeenCalledWith({ queryKey: queryKeys.projects, + exact: true, refetchType: "active", }); }); diff --git a/apps/web/src/context/global-sse.tsx b/apps/web/src/context/global-sse.tsx index 89deabc3..a59aae87 100644 --- a/apps/web/src/context/global-sse.tsx +++ b/apps/web/src/context/global-sse.tsx @@ -371,6 +371,7 @@ export function refreshProjectCatalogAfterSSEOpen( ): Promise { return queryClient.invalidateQueries({ queryKey: queryKeys.projects, + exact: true, refetchType: "active", }); }