From 07530ab83cc3f56db07c5afaeb8c1940fc634b74 Mon Sep 17 00:00:00 2001 From: akemmanuel Date: Sat, 1 Aug 2026 12:03:35 -0300 Subject: [PATCH 1/9] fix(host): harden MCP and prompt queue reliability --- main.ts | 1 + .../src/host/opengui-host.concurrency.test.ts | 100 +++++ .../backend/src/host/opengui-host.mcp.test.ts | 59 ++- packages/backend/src/host/opengui-host.ts | 113 +++-- .../src/mcp/mcp-agent-tool-source.test.ts | 93 ++++ packages/backend/src/mcp/mcp-broker.test.ts | 321 +++++++++++++- packages/backend/src/mcp/mcp-broker.ts | 399 +++++++++++++++--- .../mcp/test-fixtures/stdio-race-server.mjs | 53 +++ .../mcp/test-fixtures/stdio-tool-server.mjs | 23 +- packages/backend/src/routes/host-product.ts | 23 +- packages/backend/src/routes/host-transport.ts | 2 +- packages/harness/src/harness.ts | 2 + packages/harness/src/open-gui-harness.ts | 9 + packages/harness/src/storage/sqlite-store.ts | 25 ++ server/start-web-server.ts | 2 + src/components/PromptBox.test.tsx | 16 +- src/components/PromptBox.tsx | 22 +- src/components/sidebar/ProjectEntry.test.tsx | 6 +- .../useAppKeyboardShortcuts.test.tsx | 4 +- .../HostProvider.render.test.tsx | 37 ++ src/features/host-provider/HostProvider.tsx | 92 +++- src/features/host-provider/host-actions.ts | 24 +- .../host-provider/host-domain-state.ts | 2 +- .../host-provider/host-event-stream.test.ts | 20 + .../host-provider/host-event-stream.ts | 16 + src/features/mcp/McpSettings.render.test.tsx | 64 ++- src/features/mcp/McpSettings.tsx | 40 +- src/i18n/locales/de.json | 17 +- src/i18n/locales/en.json | 17 +- src/i18n/locales/es.json | 17 +- src/protocol/host-client.test.ts | 30 ++ src/protocol/host-client.ts | 1 + src/protocol/host-transcript.test.ts | 77 ++++ src/protocol/host-transcript.ts | 14 + src/protocol/host-types.ts | 19 +- 35 files changed, 1616 insertions(+), 144 deletions(-) create mode 100644 packages/backend/src/mcp/test-fixtures/stdio-race-server.mjs diff --git a/main.ts b/main.ts index 3298a0b1..9575bbf3 100644 --- a/main.ts +++ b/main.ts @@ -31,6 +31,7 @@ import { } from "./main/ipc-security.js"; const __dirname = path.dirname(fileURLToPath(import.meta.url)); +process.env.OPENGUI_VERSION ??= pkg.version; app.setName("OpenGUI"); app.setPath("userData", path.join(app.getPath("appData"), "OpenGUI")); diff --git a/packages/backend/src/host/opengui-host.concurrency.test.ts b/packages/backend/src/host/opengui-host.concurrency.test.ts index 53805116..c4c8d33b 100644 --- a/packages/backend/src/host/opengui-host.concurrency.test.ts +++ b/packages/backend/src/host/opengui-host.concurrency.test.ts @@ -213,6 +213,106 @@ describe("OpenGuiHost concurrent arbitration", () => { await host.close(); }); + test("interrupt prompt aborts the live Run and accepts the new user message immediately", async () => { + const root = await mkdtemp(join(tmpdir(), "opengui-host-interrupt-prompt-")); + temporaryDirectories.push(root); + const project = join(root, "project"); + await mkdir(project); + let requestCount = 0; + const model: ModelTransport = { + async *stream(_request, signal) { + requestCount += 1; + if (requestCount === 1) { + yield { type: "text_delta", delta: "partial " }; + await new Promise((_resolve, reject) => { + signal.addEventListener("abort", () => reject(new Error("aborted")), { once: true }); + }); + } + yield { type: "text_delta", delta: `answer-${requestCount}` }; + yield { type: "completed" }; + }, + }; + const host = new OpenGuiHost(root, { model }); + await host.start(); + const session = await host.createSession({ + projectDirectory: project, + model: { connectionId: "offline", modelId: "fixture-model" }, + reasoning: "none", + }); + await host.prompt(session.id, { text: "active" }); + const queued = await host.prompt(session.id, { text: "stay queued" }); + expect(queued.mode).toBe("follow_up"); + + const interrupted = await host.prompt(session.id, { text: "jump the line" }, undefined, { + interrupt: true, + }); + expect(interrupted.mode).toBe("run"); + await host.waitForIdle(session.id); + + const snapshot = await host.readSession(session.id); + expect( + snapshot.entries + .filter((entry) => entry.kind === "user_message") + .map((entry) => entry.payload.text), + ).toEqual(["active", "jump the line", "stay queued"]); + expect(snapshot.entries.some((entry) => entry.kind === "run_aborted")).toBe(true); + expect(snapshot.entries.some((entry) => entry.payload.text === "partial ")).toBe(false); + expect(snapshot.followUps).toEqual([]); + await host.close(); + }); + + test("send-now keeps durable transcript order and does not duplicate the follow-up", async () => { + const root = await mkdtemp(join(tmpdir(), "opengui-host-send-now-history-")); + temporaryDirectories.push(root); + const project = join(root, "project"); + await mkdir(project); + let requestCount = 0; + const model: ModelTransport = { + async *stream(_request, signal) { + requestCount += 1; + if (requestCount === 1) { + yield { type: "text_delta", delta: "partial answer" }; + await new Promise((_resolve, reject) => { + signal.addEventListener("abort", () => reject(new Error("aborted")), { once: true }); + }); + } + yield { type: "text_delta", delta: `answer-${requestCount}` }; + yield { type: "completed" }; + }, + }; + const host = new OpenGuiHost(root, { model }); + await host.start(); + const session = await host.createSession({ + projectDirectory: project, + model: { connectionId: "offline", modelId: "fixture-model" }, + reasoning: "none", + }); + await host.prompt(session.id, { text: "active" }); + const first = await host.prompt(session.id, { text: "queued first" }); + const second = await host.prompt(session.id, { text: "queued second" }); + expect(first.mode).toBe("follow_up"); + expect(second.mode).toBe("follow_up"); + if (second.mode !== "follow_up") throw new Error("expected queued second"); + + await host.sendFollowUpNow(session.id, second.followUp.id); + await host.waitForIdle(session.id); + + const snapshot = await host.readSession(session.id); + expect( + snapshot.entries + .filter((entry) => entry.kind === "user_message") + .map((entry) => entry.payload.text), + ).toEqual(["active", "queued second", "queued first"]); + expect( + snapshot.entries.filter( + (entry) => entry.kind === "user_message" && entry.payload.text === "queued second", + ), + ).toHaveLength(1); + expect(snapshot.entries.some((entry) => entry.kind === "run_aborted")).toBe(true); + expect(snapshot.followUps).toEqual([]); + await host.close(); + }); + test("concurrent send-now requests dispatch a durable follow-up at most once", async () => { const root = await mkdtemp(join(tmpdir(), "opengui-host-send-now-race-")); temporaryDirectories.push(root); diff --git a/packages/backend/src/host/opengui-host.mcp.test.ts b/packages/backend/src/host/opengui-host.mcp.test.ts index 5aa02613..b50adf82 100644 --- a/packages/backend/src/host/opengui-host.mcp.test.ts +++ b/packages/backend/src/host/opengui-host.mcp.test.ts @@ -3,6 +3,7 @@ import { mkdtemp, rm } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { afterEach, describe, expect, test } from "vitest"; +import { vi } from "vitest"; import type { ModelTransport } from "@opengui/harness"; import { OpenGuiHost } from "./opengui-host.ts"; @@ -37,6 +38,54 @@ function mcpCallingModel(): ModelTransport { } describe("OpenGUI Host MCP connections", () => { + test("an unavailable MCP connection does not prevent a model Run", async () => { + const dataDirectory = await mkdtemp(join(tmpdir(), "opengui-host-mcp-unavailable-")); + temporaryDirectories.push(dataDirectory); + const modelCalled = vi.fn(); + const model: ModelTransport = { + async *stream(request) { + modelCalled(request.toolDefinitions?.map((tool) => tool.name) ?? []); + yield { type: "text_delta" as const, delta: "MCP is optional" }; + yield { type: "completed" as const }; + }, + }; + const host = new OpenGuiHost(dataDirectory, { model }); + await host.start(); + + await host.upsertMcpConnection({ + id: "offline", + label: "Offline server", + enabled: true, + transport: { kind: "http", url: "http://127.0.0.1:1/mcp" }, + }); + const session = await host.createSession({ + projectDirectory: dataDirectory, + model: { connectionId: "fake", modelId: "fake" }, + reasoning: "none", + }); + + await host.prompt(session.id, { text: "Continue without MCP" }); + await host.waitForIdle(session.id); + + const snapshot = await host.readSession(session.id); + expect(snapshot.status).toBe("idle"); + expect(snapshot.entries.some((entry) => entry.kind === "run_failed")).toBe(false); + expect(modelCalled).toHaveBeenCalledWith( + expect.not.arrayContaining([expect.stringMatching(/^mcp__/u)]), + ); + expect(await host.listMcpConnections()).toEqual([ + expect.objectContaining({ + id: "offline", + status: expect.objectContaining({ + state: "offline", + toolCount: 0, + problem: expect.objectContaining({ code: "unavailable", retryable: true }), + }), + }), + ]); + await host.close(); + }); + test("persists a stdio connection, keeps environment values secret, and runs its tool", async () => { const dataDirectory = await mkdtemp(join(tmpdir(), "opengui-host-mcp-")); temporaryDirectories.push(dataDirectory); @@ -54,7 +103,8 @@ describe("OpenGUI Host MCP connections", () => { env: { FIXTURE_SECRET: "hidden-value" }, }, }); - expect(host.listMcpConnections()).toEqual([ + await host.inspectMcpConnection("fixture"); + expect(await host.listMcpConnections()).toEqual([ { id: "fixture", label: "Fixture server", @@ -65,6 +115,11 @@ describe("OpenGUI Host MCP connections", () => { args: [fixture], envKeys: ["FIXTURE_SECRET"], }, + status: { + state: "ready", + toolCount: 1, + lastCheckedAt: expect.any(String), + }, }, ]); @@ -83,7 +138,7 @@ describe("OpenGUI Host MCP connections", () => { const reopened = new OpenGuiHost(dataDirectory, { model: mcpCallingModel() }); await reopened.start(); - expect(reopened.listMcpConnections()).toHaveLength(1); + expect(await reopened.listMcpConnections()).toHaveLength(1); await reopened.close(); }); }); diff --git a/packages/backend/src/host/opengui-host.ts b/packages/backend/src/host/opengui-host.ts index ba1c57ae..252a2d49 100644 --- a/packages/backend/src/host/opengui-host.ts +++ b/packages/backend/src/host/opengui-host.ts @@ -94,7 +94,19 @@ export interface HostHealth { export interface HostModelConnection extends OpenAiCompatibleConnection {} -export type HostMcpConnection = +export interface HostMcpConnectionStatus { + state: "disabled" | "refreshing" | "ready" | "degraded" | "offline"; + toolCount: number; + lastCheckedAt?: string; + problem?: { + code: string; + stage: string; + message: string; + retryable: boolean; + }; +} + +export type HostMcpConnection = ( | { id: string; label: string; @@ -112,7 +124,8 @@ export type HostMcpConnection = label: string; enabled: boolean; transport: { kind: "http"; url: string; bearerTokenConfigured: boolean }; - }; + } +) & { status?: HostMcpConnectionStatus }; export type HostMcpConnectionInput = | { @@ -1019,6 +1032,9 @@ export class OpenGuiHost { async #refreshMcpBroker() { await this.#mcpBroker.replaceConnections(this.#mcpRuntimeConnections()); + void this.#mcpBroker + .refresh({ actorId: "local:settings", sessionId: "mcp-settings" }) + .catch(() => undefined); } async #openCodeGoConnection(apiKey?: string): Promise { @@ -1057,13 +1073,50 @@ export class OpenGuiHost { health(): HostHealth { return { ok: true, - version: process.env.npm_package_version || "0.0.0", + version: process.env.OPENGUI_VERSION || process.env.npm_package_version || "0.0.0", shell: process.env.SHELL || (process.platform === "win32" ? "powershell" : "/bin/sh"), }; } - listMcpConnections(): HostMcpConnection[] { - return structuredClone(this.#settings.mcpConnections); + async listMcpConnections(): Promise { + const catalog = await this.#mcpBroker.catalog({ + actorId: "local:settings", + sessionId: "mcp-settings", + }); + return structuredClone( + this.#settings.mcpConnections.map((connection) => { + const tools = catalog.tools.filter((tool) => tool.ref.connectionId === connection.id); + const problem = catalog.problems.find((item) => item.connectionId === connection.id); + const lastCheckedAt = catalog.checkedAt[connection.id]; + const state: HostMcpConnectionStatus["state"] = !connection.enabled + ? "disabled" + : problem && tools.length > 0 + ? "degraded" + : problem + ? "offline" + : lastCheckedAt + ? "ready" + : "refreshing"; + return { + ...connection, + status: { + state, + toolCount: tools.length, + ...(lastCheckedAt ? { lastCheckedAt } : {}), + ...(problem + ? { + problem: { + code: problem.code, + stage: problem.stage, + message: problem.message, + retryable: problem.retryable, + }, + } + : {}), + }, + }; + }), + ); } async upsertMcpConnection(input: HostMcpConnectionInput): Promise { @@ -1197,10 +1250,12 @@ export class OpenGuiHost { if (this.#resolveExecutionPolicy && (await this.#resolveExecutionPolicy(actor)).restricted) { throw new Error("MCP tools are unavailable for restricted actors"); } - const catalog = await this.#mcpBroker.catalog({ - actorId: actor ? `${actor.type}:${actor.id}` : "local:settings", - sessionId: `mcp-settings:${id}`, - }); + const scope = { actorId: "local:settings", sessionId: "mcp-settings" }; + const catalog = await this.#mcpBroker.refresh(scope, id); + const problem = catalog.problems.find((item) => item.connectionId === id); + if (problem && !catalog.tools.some((tool) => tool.ref.connectionId === id)) { + throw new Error(problem.message); + } return catalog.tools .filter((tool) => tool.ref.connectionId === id) .map((tool) => ({ @@ -1763,6 +1818,7 @@ export class OpenGuiHost { sessionId: string, prompt: PromptInput, actor: DurableActor | undefined = prompt.actor, + options?: { interrupt?: boolean }, ) { const previousAdmission = this.#promptAdmissions.get(sessionId) ?? Promise.resolve(); let releaseAdmission!: () => void; @@ -1775,18 +1831,27 @@ export class OpenGuiHost { try { const { session, snapshot } = await this.#authorizedSession(sessionId, actor, "run"); if (snapshot.status === "running") { - try { - const followUp = await session.followUp(prompt); - return { mode: "follow_up" as const, followUp }; - } catch (error) { - if ( - !(error instanceof Error) || - error.message !== "Follow-ups can only be queued while a Session is running" - ) { - throw error; + if (options?.interrupt) { + // Abort the live Run under admission, then accept this prompt as the next Run. + // Existing Follow-ups stay queued and run after this interrupted turn completes. + if (this.#activeRuns.has(sessionId)) { + await session.abort(); + await this.#activeRuns.get(sessionId); + } + } else { + try { + const followUp = await session.followUp(prompt); + return { mode: "follow_up" as const, followUp }; + } catch (error) { + if ( + !(error instanceof Error) || + error.message !== "Follow-ups can only be queued while a Session is running" + ) { + throw error; + } + // A very short Run may finish between the status read and queue write. + // Admission remains serialized, so it is safe to accept this as the next Run. } - // A very short Run may finish between the status read and queue write. - // Admission remains serialized, so it is safe to accept this as the next Run. } } const iterator = session.run(prompt)[Symbol.asyncIterator](); @@ -1852,15 +1917,13 @@ export class OpenGuiHost { async sendFollowUpNow(sessionId: string, followUpId: string, actor?: DurableActor) { const { session } = await this.#authorizedSession(sessionId, actor, "run"); - const followUps = (await session.read()).followUps; - const selected = followUps.find((item) => item.id === followUpId); - if (!selected) throw new Error(`Pending follow-up not found: ${followUpId}`); - await session.reorderFollowUp(followUpId, 0); + // Take the row first so a finishing Run cannot claim the same Follow-up and + // duplicate it into the transcript while we abort + re-prompt. + const selected = await session.takeFollowUp(followUpId); if (this.#activeRuns.has(sessionId)) { await session.abort(); await this.#activeRuns.get(sessionId); } - await session.removeFollowUp(followUpId); // The caller is authorized to operate the Session above, but execution // belongs to the actor stored with the accepted prompt. Reauthorize that // actor now rather than transferring the caller's grants. diff --git a/packages/backend/src/mcp/mcp-agent-tool-source.test.ts b/packages/backend/src/mcp/mcp-agent-tool-source.test.ts index 595d6ae2..94e419f8 100644 --- a/packages/backend/src/mcp/mcp-agent-tool-source.test.ts +++ b/packages/backend/src/mcp/mcp-agent-tool-source.test.ts @@ -4,6 +4,8 @@ import { createMcpAgentToolSource, type McpBroker, type McpCatalogSnapshot } fro function catalog(): McpCatalogSnapshot { return { generation: "large-generation", + problems: [], + checkedAt: {}, tools: [ { ref: { connectionId: "crm", toolName: "find_customer" }, @@ -34,10 +36,101 @@ function catalog(): McpCatalogSnapshot { } describe("MCP AgentToolSource", () => { + test("turns an unavailable MCP invocation into a recoverable tool result", async () => { + const unavailable = Object.assign(new TypeError("fetch failed"), { + cause: Object.assign(new Error("connect timed out"), { + code: "UND_ERR_CONNECT_TIMEOUT", + }), + }); + const broker: McpBroker = { + catalog: async () => catalog(), + refresh: async () => catalog(), + call: async () => { + throw unavailable; + }, + replaceConnections: async () => undefined, + close: async () => undefined, + }; + const tools = await createMcpAgentToolSource(broker).resolve( + { sessionId: "session", runId: "run", projectDirectory: "/project" }, + new AbortController().signal, + ); + + await expect( + tools.invoke( + { name: "mcp__crm__find_customer__11111111", input: { email: "ada@example.com" } }, + new AbortController().signal, + ), + ).resolves.toEqual({ + status: "error", + summary: "MCP connection timed out", + content: [], + error: { code: "timeout", retryable: true }, + }); + }); + + test.each([ + ["ENOTFOUND", "unavailable"], + ["EAI_AGAIN", "unavailable"], + ["ECONNRESET", "unavailable"], + ["ECONNREFUSED", "unavailable"], + ["EHOSTUNREACH", "unavailable"], + ["ETIMEDOUT", "timeout"], + ])("classifies nested invocation transport code %s", async (code, expectedCode) => { + const broker: McpBroker = { + catalog: async () => catalog(), + refresh: async () => catalog(), + call: async () => { + throw Object.assign(new Error("transport failed"), { + cause: Object.assign(new Error(code), { code }), + }); + }, + replaceConnections: async () => undefined, + close: async () => undefined, + }; + const tools = await createMcpAgentToolSource(broker).resolve( + { sessionId: "session", runId: "run", projectDirectory: "/project" }, + new AbortController().signal, + ); + await expect( + tools.invoke( + { name: "mcp__crm__find_customer__11111111", input: {} }, + new AbortController().signal, + ), + ).resolves.toMatchObject({ + status: "error", + error: { code: expectedCode, retryable: true }, + }); + }); + + test("does not hide an unexpected invocation defect as a discovery error", async () => { + const defect = new Error("Invariant violated"); + const broker: McpBroker = { + catalog: async () => catalog(), + refresh: async () => catalog(), + call: async () => { + throw defect; + }, + replaceConnections: async () => undefined, + close: async () => undefined, + }; + const tools = await createMcpAgentToolSource(broker).resolve( + { sessionId: "session", runId: "run", projectDirectory: "/project" }, + new AbortController().signal, + ); + await expect( + tools.invoke( + { name: "mcp__crm__find_customer__11111111", input: {} }, + new AbortController().signal, + ), + ).rejects.toBe(defect); + }); + test("uses stable search, inspect, and call tools when direct schemas exceed the budget", async () => { const call = vi.fn(async () => ({ status: "ok" as const, summary: "updated", content: [] })); const broker: McpBroker = { catalog: async () => catalog(), + refresh: async () => catalog(), call, replaceConnections: async () => undefined, close: async () => undefined, diff --git a/packages/backend/src/mcp/mcp-broker.test.ts b/packages/backend/src/mcp/mcp-broker.test.ts index a0aeeeca..0099ebc8 100644 --- a/packages/backend/src/mcp/mcp-broker.test.ts +++ b/packages/backend/src/mcp/mcp-broker.test.ts @@ -4,6 +4,7 @@ import { tmpdir } from "node:os"; import { join } from "node:path"; import { once } from "node:events"; import type { AddressInfo } from "node:net"; +import type { Server as HttpServer } from "node:http"; import { createOpenGuiHarness } from "@opengui/harness"; import { FakeModel } from "@opengui/harness/test"; import { Server } from "@modelcontextprotocol/sdk/server/index.js"; @@ -15,8 +16,73 @@ import { describe, expect, test } from "vitest"; import { createMcpAgentToolSource, createMcpBroker } from "./mcp-broker.ts"; const fixture = fileURLToPath(new URL("./test-fixtures/stdio-tool-server.mjs", import.meta.url)); +const raceFixture = fileURLToPath( + new URL("./test-fixtures/stdio-race-server.mjs", import.meta.url), +); describe("McpBroker", () => { + test("bounds discovery time and reports a safe timeout problem", async () => { + const broker = createMcpBroker({ + discoveryTimeoutMs: 200, + connections: [ + { + id: "hanging", + label: "Hanging server", + transport: { + kind: "stdio", + command: process.execPath, + args: ["-e", "process.stdin.resume()"], + }, + }, + ], + }); + + try { + const catalog = await broker.refresh({ actorId: "local:test", sessionId: "timeout" }); + expect(catalog.tools).toEqual([]); + expect(catalog.problems).toEqual([ + { + connectionId: "hanging", + stage: "connect", + code: "timeout", + retryable: true, + message: "MCP connection timed out", + }, + ]); + } finally { + await broker.close(); + } + }); + + test("reports a timeout after initialization as a discovery failure", async () => { + const broker = createMcpBroker({ + // Leave enough time for a cold CI process to complete the MCP handshake; only tools/list + // hangs in this fixture, so the assertion must not race process startup. + discoveryTimeoutMs: 1_000, + connections: [ + { + id: "hanging-list", + label: "Hanging list", + transport: { + kind: "stdio", + command: process.execPath, + args: [fixture], + env: { MCP_HANG_LIST: "1" }, + }, + }, + ], + }); + try { + await expect( + broker.refresh({ actorId: "local:test", sessionId: "discover-timeout" }), + ).resolves.toMatchObject({ + problems: [expect.objectContaining({ stage: "discover", code: "timeout" })], + }); + } finally { + await broker.close(); + } + }); + test("discovers and invokes a configured stdio tool through one actor-scoped interface", async () => { const broker = createMcpBroker({ connections: [ @@ -34,6 +100,7 @@ describe("McpBroker", () => { const scope = { actorId: "local:test", sessionId: "session-test" }; try { + await broker.refresh(scope); const catalog = await broker.catalog(scope); expect(catalog.tools).toEqual([ expect.objectContaining({ @@ -67,6 +134,37 @@ describe("McpBroker", () => { } }); + test("keeps healthy tools when another MCP connection is unavailable", async () => { + const broker = createMcpBroker({ + connections: [ + { + id: "fixture", + label: "Fixture server", + transport: { kind: "stdio", command: process.execPath, args: [fixture] }, + }, + { + id: "offline", + label: "Offline server", + transport: { kind: "http", url: "http://127.0.0.1:1/mcp" }, + }, + ], + }); + + try { + const catalog = await broker.refresh({ actorId: "local:test", sessionId: "partial" }); + expect(catalog.tools.map((tool) => tool.ref.connectionId)).toEqual(["fixture"]); + expect(catalog.problems).toEqual([ + expect.objectContaining({ + connectionId: "offline", + code: "unavailable", + retryable: true, + }), + ]); + } finally { + await broker.close(); + } + }); + test("runs a stdio MCP tool through the Harness and durable Session transcript", async () => { const dataDirectory = await mkdtemp(join(tmpdir(), "opengui-mcp-harness-")); const broker = createMcpBroker({ @@ -79,7 +177,7 @@ describe("McpBroker", () => { ], }); const toolName = ( - await broker.catalog({ actorId: "local:legacy", sessionId: "catalog-inspection" }) + await broker.refresh({ actorId: "local:legacy", sessionId: "catalog-inspection" }) ).tools[0]!.modelName; const model = new FakeModel([ { @@ -135,17 +233,184 @@ describe("McpBroker", () => { const scope = { actorId: "local:test", sessionId: "session-test" }; try { + await broker.refresh(scope); expect((await broker.catalog(scope)).tools).toHaveLength(1); await broker.replaceConnections([]); expect(await broker.catalog(scope)).toEqual({ generation: expect.stringMatching(/^[a-f0-9]{24}$/u), tools: [], + problems: [], + checkedAt: {}, + }); + } finally { + await broker.close(); + } + }); + + test("isolates catalogs between actor and Session runtimes", async () => { + const broker = createMcpBroker({ + connections: [ + { + id: "scoped", + label: "Scoped fixture", + transport: { + kind: "stdio", + command: process.execPath, + args: [fixture], + env: { MCP_PID_TOOL: "1" }, + }, + }, + ], + }); + const first = { actorId: "account:first", sessionId: "one" }; + const second = { actorId: "account:second", sessionId: "two" }; + try { + const firstCatalog = await broker.refresh(first); + const secondCatalog = await broker.refresh(second); + expect(firstCatalog.tools[0]?.ref.toolName).not.toBe(secondCatalog.tools[0]?.ref.toolName); + expect((await broker.catalog(first)).tools).toEqual(firstCatalog.tools); + expect((await broker.catalog(second)).tools).toEqual(secondCatalog.tools); + } finally { + await broker.close(); + } + }); + + test("does not let an older overlapping refresh replace a newer catalog", async () => { + const broker = createMcpBroker({ + connections: [ + { + id: "race", + label: "Race fixture", + transport: { kind: "stdio", command: process.execPath, args: [raceFixture] }, + }, + ], + }); + const scope = { actorId: "local:test", sessionId: "race" }; + try { + const oldRefresh = broker.refresh(scope); + await new Promise((resolve) => setTimeout(resolve, 25)); + const newRefresh = broker.refresh(scope); + await Promise.all([oldRefresh, newRefresh]); + expect((await broker.catalog(scope)).tools.map((tool) => tool.ref.toolName)).toEqual([ + "new_tool", + ]); + } finally { + await broker.close(); + } + }); + + test("does not let an older failure degrade a newer successful refresh", async () => { + const broker = createMcpBroker({ + connections: [ + { + id: "race-failure", + label: "Race failure fixture", + transport: { + kind: "stdio", + command: process.execPath, + args: [raceFixture], + env: { MCP_FIRST_LIST_ERROR: "1" }, + }, + }, + ], + }); + const scope = { actorId: "local:test", sessionId: "race-failure" }; + try { + const oldRefresh = broker.refresh(scope); + await new Promise((resolve) => setTimeout(resolve, 25)); + const newRefresh = broker.refresh(scope); + await Promise.all([oldRefresh, newRefresh]); + expect(await broker.catalog(scope)).toMatchObject({ + tools: [ + expect.objectContaining({ ref: { connectionId: "race-failure", toolName: "new_tool" } }), + ], + problems: [], }); } finally { await broker.close(); } }); + test("bounds the combined catalog across connections", async () => { + const connections = ["one", "two"].map((id) => ({ + id, + label: id, + transport: { + kind: "stdio" as const, + command: process.execPath, + args: [fixture], + env: { MCP_TOOL_COUNT: "300" }, + }, + })); + const broker = createMcpBroker({ connections }); + try { + const catalog = await broker.refresh({ actorId: "local:test", sessionId: "bounded" }); + expect(catalog.tools).toHaveLength(500); + expect(catalog.problems).toContainEqual( + expect.objectContaining({ connectionId: "two", code: "protocol" }), + ); + } finally { + await broker.close(); + } + }); + + test("does not record an aborted first refresh as a completed health check", async () => { + const broker = createMcpBroker({ + connections: [ + { + id: "hanging", + label: "Hanging", + transport: { + kind: "stdio", + command: process.execPath, + args: ["-e", "process.stdin.resume()"], + }, + }, + ], + }); + const scope = { actorId: "local:test", sessionId: "aborted" }; + const controller = new AbortController(); + controller.abort(new Error("Run canceled")); + try { + await expect(broker.refresh(scope, undefined, controller.signal)).rejects.toThrow( + "Run canceled", + ); + expect((await broker.catalog(scope)).checkedAt).toEqual({}); + } finally { + await broker.close(); + } + }); + + test("cancels a caller waiting for connection establishment", async () => { + const broker = createMcpBroker({ + discoveryTimeoutMs: 2_000, + connections: [ + { + id: "hanging", + label: "Hanging", + transport: { + kind: "stdio", + command: process.execPath, + args: ["-e", "process.stdin.resume()"], + }, + }, + ], + }); + const controller = new AbortController(); + const call = broker.call( + { actorId: "local:test", sessionId: "call" }, + { connectionId: "hanging", toolName: "stale" }, + {}, + controller.signal, + ); + setTimeout(() => controller.abort(new Error("Run canceled")), 50); + try { + await expect(call).rejects.toThrow("Run canceled"); + } finally { + await broker.close(); + } + }); + test("discovers and invokes a bearer-authenticated Streamable HTTP tool", async () => { const app = new Hono(); app.all("/mcp", async (context) => { @@ -201,13 +466,63 @@ describe("McpBroker", () => { const scope = { actorId: "local:test", sessionId: "http-session" }; try { - const tool = (await broker.catalog(scope)).tools[0]!; + const tool = (await broker.refresh(scope)).tools[0]!; await expect( broker.call(scope, tool.ref, { name: "Ada" }, new AbortController().signal), ).resolves.toMatchObject({ status: "ok", summary: "Hello Ada" }); + (httpServer as HttpServer).closeAllConnections(); + await new Promise((resolve) => httpServer.close(() => resolve())); + const degraded = await broker.refresh(scope); + expect(degraded.tools.map((item) => item.ref.toolName)).toEqual(["greet"]); + expect(degraded.problems).toEqual([ + expect.objectContaining({ connectionId: "http-fixture", retryable: true }), + ]); + } finally { + await broker.close(); + if (httpServer.listening) httpServer.close(); + } + }); + + test("fails closed when authentication is revoked after healthy discovery", async () => { + let authorized = true; + const app = new Hono(); + app.all("/mcp", async (context) => { + if (!authorized) return new Response("Unauthorized", { status: 401 }); + const transport = new WebStandardStreamableHTTPServerTransport(); + const server = new Server( + { name: "Revocation fixture", version: "1.0.0" }, + { capabilities: { tools: {} } }, + ); + server.setRequestHandler(ListToolsRequestSchema, async () => ({ + tools: [{ name: "private_tool", inputSchema: { type: "object" } }], + })); + await server.connect(transport); + return transport.handleRequest(context.req.raw); + }); + const httpServer = serve({ fetch: app.fetch, port: 0 }); + if (!httpServer.listening) await once(httpServer, "listening"); + const port = (httpServer.address() as AddressInfo).port; + const broker = createMcpBroker({ + connections: [ + { + id: "private", + label: "Private", + transport: { kind: "http", url: `http://127.0.0.1:${port}/mcp` }, + }, + ], + }); + const scope = { actorId: "local:test", sessionId: "revocation" }; + try { + expect((await broker.refresh(scope)).tools).toHaveLength(1); + authorized = false; + const revoked = await broker.refresh(scope); + expect(revoked.tools).toEqual([]); + expect(revoked.problems).toEqual([ + expect.objectContaining({ code: "authentication", retryable: false }), + ]); } finally { await broker.close(); - httpServer.close(); + await new Promise((resolve) => httpServer.close(() => resolve())); } }); }); diff --git a/packages/backend/src/mcp/mcp-broker.ts b/packages/backend/src/mcp/mcp-broker.ts index c2cc3632..825ab7a1 100644 --- a/packages/backend/src/mcp/mcp-broker.ts +++ b/packages/backend/src/mcp/mcp-broker.ts @@ -52,6 +52,16 @@ export interface McpCatalogTool { export interface McpCatalogSnapshot { generation: string; tools: McpCatalogTool[]; + problems: McpConnectionProblem[]; + checkedAt: Record; +} + +export interface McpConnectionProblem { + connectionId: string; + stage: "connect" | "discover" | "invoke"; + code: "timeout" | "authentication" | "permission" | "unavailable" | "protocol" | "unknown"; + retryable: boolean; + message: string; } export type McpResultContent = @@ -62,12 +72,20 @@ export interface McpToolResult { status: "ok" | "error"; summary: string; content: McpResultContent[]; + error?: { code: McpConnectionProblem["code"]; retryable: boolean }; attachments?: Array<{ type: "image"; data: string; mimeType: string }>; structured?: unknown; } export interface McpBroker { + /** Returns cached capabilities only. This method never performs remote I/O. */ catalog(scope: McpActorScope): Promise; + /** Refreshes one or all remote catalogs without discarding last-known-good capabilities. */ + refresh( + scope: McpActorScope, + connectionId?: string, + signal?: AbortSignal, + ): Promise; call( scope: McpActorScope, ref: McpToolRef, @@ -83,9 +101,15 @@ interface Runtime { close(): Promise; } +interface PendingRuntime { + controller: AbortController; + promise: Promise; +} + const MAX_CATALOG_TOOLS = 500; const MAX_CATALOG_BYTES = 2 * 1024 * 1024; const MAX_CATALOG_PAGES = 100; +const DEFAULT_DISCOVERY_TIMEOUT_MS = 5_000; function safeSegment(value: string, fallback: string) { const segment = value @@ -101,10 +125,28 @@ function modelName(ref: McpToolRef) { return `mcp__${safeSegment(ref.connectionId, "server")}__${safeSegment(ref.toolName, "tool")}__${suffix}`; } +function connectionFingerprint(connection: McpConnection | undefined) { + return contentFingerprint(connection ?? null); +} + function runtimeKey(scope: McpActorScope, connectionId: string) { return `${scope.actorId}\u0000${scope.sessionId}\u0000${connectionId}`; } +function cacheKey(scope: McpActorScope, connectionId: string) { + return runtimeKey(scope, connectionId); +} + +function waitWithSignal(promise: Promise, signal?: AbortSignal): Promise { + if (!signal) return promise; + if (signal.aborted) return Promise.reject(signal.reason); + return new Promise((resolve, reject) => { + const aborted = () => reject(signal.reason); + signal.addEventListener("abort", aborted, { once: true }); + void promise.then(resolve, reject).finally(() => signal.removeEventListener("abort", aborted)); + }); +} + function normalizedContent(content: unknown): McpResultContent[] { if (!Array.isArray(content)) return []; return content.flatMap((item) => { @@ -124,12 +166,112 @@ function normalizedContent(content: unknown): McpResultContent[] { }); } +function errorChain(error: unknown): Array> { + const chain: Array> = []; + let current = error; + const seen = new Set(); + while (current && typeof current === "object" && !seen.has(current)) { + seen.add(current); + const record = current as Record; + chain.push(record); + current = record.cause; + } + return chain; +} + +function mcpConnectionProblem( + connectionId: string, + error: unknown, + stage: McpConnectionProblem["stage"] = "discover", +): McpConnectionProblem { + const chain = errorChain(error); + const text = chain + .flatMap((item) => (typeof item.message === "string" ? [item.message] : [])) + .join(" "); + const codes = new Set( + chain.flatMap((item) => (typeof item.code === "string" ? [item.code] : [])), + ); + const status = chain.find((item) => typeof item.status === "number")?.status; + const rpcStatus = chain.find((item) => typeof item.code === "number")?.code; + if (status === 401 || rpcStatus === 401 || /(?:HTTP\s*)?401\b/iu.test(text)) { + return { + connectionId, + stage, + code: "authentication", + retryable: false, + message: "MCP authentication failed", + }; + } + if (status === 403 || rpcStatus === 403 || /(?:HTTP\s*)?403\b/iu.test(text)) { + return { + connectionId, + stage, + code: "permission", + retryable: false, + message: "MCP server denied access", + }; + } + const timeoutCodes = ["UND_ERR_CONNECT_TIMEOUT", "ETIMEDOUT"]; + if (timeoutCodes.some((code) => codes.has(code)) || /timed? ?out|timeout/iu.test(text)) { + return { + connectionId, + stage, + code: "timeout", + retryable: true, + message: "MCP connection timed out", + }; + } + if (/protocol|parse|json|invalid response|malformed/iu.test(text)) { + return { + connectionId, + stage, + code: "protocol", + retryable: false, + message: "MCP server returned an invalid response", + }; + } + const unavailableCodes = [ + "ENOTFOUND", + "EAI_AGAIN", + "ECONNRESET", + "ECONNREFUSED", + "EHOSTUNREACH", + "ENETUNREACH", + "UND_ERR_SOCKET", + ]; + if ( + unavailableCodes.some((code) => codes.has(code)) || + /fetch failed|connect|socket|network|unavailable/iu.test(text) + ) { + return { + connectionId, + stage, + code: "unavailable", + retryable: true, + message: "MCP server is unavailable", + }; + } + return { + connectionId, + stage, + code: "unknown", + retryable: false, + message: "MCP tool discovery failed", + }; +} + class DefaultMcpBroker implements McpBroker { readonly #connections: Map; - readonly #runtimes = new Map>(); + readonly #runtimes = new Map(); + readonly #catalogs = new Map(); + readonly #problems = new Map(); + readonly #checkedAt = new Map(); + readonly #refreshRevisions = new Map(); + readonly #discoveryTimeoutMs: number; - constructor(connections: readonly McpConnection[]) { + constructor(connections: readonly McpConnection[], discoveryTimeoutMs: number) { this.#connections = this.#connectionMap(connections); + this.#discoveryTimeoutMs = discoveryTimeoutMs; } #connectionMap(connections: readonly McpConnection[]) { @@ -138,20 +280,26 @@ class DefaultMcpBroker implements McpBroker { return result; } - async #runtime(scope: McpActorScope, connection: McpConnection) { + async #runtime(scope: McpActorScope, connection: McpConnection, signal?: AbortSignal) { const key = runtimeKey(scope, connection.id); let pending = this.#runtimes.get(key); if (!pending) { - pending = this.#connect(connection, key).catch((error) => { - this.#runtimes.delete(key); + const controller = new AbortController(); + const connectSignal = AbortSignal.any([ + controller.signal, + AbortSignal.timeout(this.#discoveryTimeoutMs), + ]); + const promise = this.#connect(connection, key, connectSignal).catch((error) => { + if (this.#runtimes.get(key)?.promise === promise) this.#runtimes.delete(key); throw error; }); + pending = { controller, promise }; this.#runtimes.set(key, pending); } - return pending; + return waitWithSignal(pending.promise, signal); } - async #connect(connection: McpConnection, key: string): Promise { + async #connect(connection: McpConnection, key: string, signal?: AbortSignal): Promise { const client = new Client({ name: "OpenGUI", version: "0.0.0" }); const transport = connection.transport.kind === "stdio" @@ -175,7 +323,12 @@ class DefaultMcpBroker implements McpBroker { transport.onclose = () => { this.#runtimes.delete(key); }; - await client.connect(transport); + try { + await client.connect(transport, signal ? { signal } : undefined); + } catch (error) { + await client.close().catch(() => undefined); + throw error; + } return { client, close: async () => { @@ -185,47 +338,154 @@ class DefaultMcpBroker implements McpBroker { } async catalog(scope: McpActorScope): Promise { + const connectionIds = [...this.#connections.keys()]; const tools: McpCatalogTool[] = []; + const aggregateProblems: McpConnectionProblem[] = []; let catalogBytes = 0; - for (const connection of this.#connections.values()) { - const runtime = await this.#runtime(scope, connection); - let cursor: string | undefined; - const cursors = new Set(); - let pages = 0; - do { - pages += 1; - if (pages > MAX_CATALOG_PAGES) throw new Error("MCP tool catalog has too many pages"); - const page = await runtime.client.listTools(cursor ? { cursor } : undefined); - for (const tool of page.tools) { - if (tools.length >= MAX_CATALOG_TOOLS) throw new Error("MCP tool catalog is too large"); - const ref = { connectionId: connection.id, toolName: tool.name }; - const inputSchema = tool.inputSchema as Record; - const description = tool.description?.trim() || ""; - catalogBytes += - Buffer.byteLength(JSON.stringify(inputSchema)) + Buffer.byteLength(description); - if (catalogBytes > MAX_CATALOG_BYTES) throw new Error("MCP tool catalog is too large"); - tools.push({ - ref, - modelName: modelName(ref), - title: tool.title?.trim() || tool.annotations?.title?.trim() || tool.name, - description, - inputSchema, - fingerprint: contentFingerprint({ ref, inputSchema, description: tool.description }), + for (const id of connectionIds) { + const key = cacheKey(scope, id); + for (const tool of this.#catalogs.get(key) ?? []) { + const bytes = + Buffer.byteLength(JSON.stringify(tool.inputSchema)) + Buffer.byteLength(tool.description); + if (tools.length >= MAX_CATALOG_TOOLS || catalogBytes + bytes > MAX_CATALOG_BYTES) { + aggregateProblems.push({ + connectionId: id, + stage: "discover", + code: "protocol", + retryable: false, + message: "MCP aggregate tool catalog is too large", }); + break; } - cursor = page.nextCursor; - if (cursor && cursors.has(cursor)) throw new Error("MCP tool catalog repeated a cursor"); - if (cursor) cursors.add(cursor); - } while (cursor); + tools.push(tool); + catalogBytes += bytes; + } } return { generation: contentFingerprint( tools.map(({ ref, fingerprint }) => ({ ref, fingerprint })), ).slice(0, 24), tools, + problems: connectionIds + .flatMap((id) => { + const problem = this.#problems.get(cacheKey(scope, id)); + return problem ? [problem] : []; + }) + .concat(aggregateProblems), + checkedAt: Object.fromEntries( + connectionIds.flatMap((id) => { + const checkedAt = this.#checkedAt.get(cacheKey(scope, id)); + return checkedAt ? [[id, checkedAt]] : []; + }), + ), }; } + async #discover( + scope: McpActorScope, + connection: McpConnection, + signal?: AbortSignal, + onConnected?: () => void, + ): Promise { + const tools: McpCatalogTool[] = []; + let catalogBytes = 0; + const runtime = await this.#runtime(scope, connection, signal); + onConnected?.(); + let cursor: string | undefined; + const cursors = new Set(); + let pages = 0; + do { + pages += 1; + if (pages > MAX_CATALOG_PAGES) throw new Error("MCP tool catalog has too many pages"); + const page = await runtime.client.listTools( + cursor ? { cursor } : undefined, + signal ? { signal } : undefined, + ); + for (const tool of page.tools) { + if (tools.length >= MAX_CATALOG_TOOLS) throw new Error("MCP tool catalog is too large"); + const ref = { connectionId: connection.id, toolName: tool.name }; + const inputSchema = tool.inputSchema as Record; + const description = tool.description?.trim() || ""; + catalogBytes += + Buffer.byteLength(JSON.stringify(inputSchema)) + Buffer.byteLength(description); + if (catalogBytes > MAX_CATALOG_BYTES) throw new Error("MCP tool catalog is too large"); + tools.push({ + ref, + modelName: modelName(ref), + title: tool.title?.trim() || tool.annotations?.title?.trim() || tool.name, + description, + inputSchema, + fingerprint: contentFingerprint({ ref, inputSchema, description: tool.description }), + }); + } + cursor = page.nextCursor; + if (cursor && cursors.has(cursor)) throw new Error("MCP tool catalog repeated a cursor"); + if (cursor) cursors.add(cursor); + } while (cursor); + return tools; + } + + async refresh(scope: McpActorScope, connectionId?: string, signal?: AbortSignal) { + const connections = connectionId + ? [this.#connections.get(connectionId)].filter( + (connection): connection is McpConnection => connection !== undefined, + ) + : [...this.#connections.values()]; + if (connectionId && connections.length === 0) { + throw new Error(`Unknown MCP connection: ${connectionId}`); + } + await Promise.all( + connections.map(async (connection) => { + const key = cacheKey(scope, connection.id); + const revision = (this.#refreshRevisions.get(key) ?? 0) + 1; + this.#refreshRevisions.set(key, revision); + const fingerprint = connectionFingerprint(connection); + const timeoutSignal = AbortSignal.timeout(this.#discoveryTimeoutMs); + const refreshSignal = signal ? AbortSignal.any([signal, timeoutSignal]) : timeoutSignal; + let connected = false; + try { + const tools = await this.#discover(scope, connection, refreshSignal, () => { + connected = true; + }); + if ( + connectionFingerprint(this.#connections.get(connection.id)) !== fingerprint || + this.#refreshRevisions.get(key) !== revision + ) + return; + this.#catalogs.set(key, tools); + this.#problems.delete(key); + } catch (error) { + if (signal?.aborted) throw error; + if ( + connectionFingerprint(this.#connections.get(connection.id)) !== fingerprint || + this.#refreshRevisions.get(key) !== revision + ) + return; + const problem: McpConnectionProblem = timeoutSignal.aborted + ? { + connectionId: connection.id, + stage: connected ? ("discover" as const) : ("connect" as const), + code: "timeout", + retryable: true, + message: connected ? "MCP tool discovery timed out" : "MCP connection timed out", + } + : mcpConnectionProblem(connection.id, error, connected ? "discover" : "connect"); + this.#problems.set(key, problem); + if (!problem.retryable) this.#catalogs.delete(key); + } finally { + if ( + !signal?.aborted && + connectionFingerprint(this.#connections.get(connection.id)) === fingerprint && + this.#refreshRevisions.get(key) === revision + ) { + this.#checkedAt.set(key, new Date().toISOString()); + } + } + }), + ); + return this.catalog(scope); + } + async call( scope: McpActorScope, ref: McpToolRef, @@ -234,7 +494,7 @@ class DefaultMcpBroker implements McpBroker { ): Promise { const connection = this.#connections.get(ref.connectionId); if (!connection) throw new Error(`Unknown MCP connection: ${ref.connectionId}`); - const runtime = await this.#runtime(scope, connection); + const runtime = await this.#runtime(scope, connection, signal); const result = await runtime.client.callTool( { name: ref.toolName, @@ -273,24 +533,53 @@ class DefaultMcpBroker implements McpBroker { async replaceConnections(connections: readonly McpConnection[]) { const next = this.#connectionMap(connections); - await this.#closeRuntimes(); + const changed = new Set( + [...new Set([...this.#connections.keys(), ...next.keys()])].filter( + (id) => + connectionFingerprint(this.#connections.get(id)) !== connectionFingerprint(next.get(id)), + ), + ); + await this.#closeRuntimes(changed); + for (const id of changed) { + for (const key of this.#catalogs.keys()) { + if (key.endsWith(`\u0000${id}`)) this.#catalogs.delete(key); + } + for (const key of this.#problems.keys()) { + if (key.endsWith(`\u0000${id}`)) this.#problems.delete(key); + } + for (const key of this.#checkedAt.keys()) { + if (key.endsWith(`\u0000${id}`)) this.#checkedAt.delete(key); + } + for (const key of this.#refreshRevisions.keys()) { + if (key.endsWith(`\u0000${id}`)) this.#refreshRevisions.delete(key); + } + } this.#connections.clear(); for (const [id, connection] of next) this.#connections.set(id, connection); } - async #closeRuntimes() { - const runtimes = [...this.#runtimes.values()]; - this.#runtimes.clear(); + async #closeRuntimes(connectionIds?: ReadonlySet) { + const runtimes = [...this.#runtimes.entries()].filter( + ([key]) => !connectionIds || [...connectionIds].some((id) => key.endsWith(`\u0000${id}`)), + ); + for (const [key] of runtimes) this.#runtimes.delete(key); + for (const [, pending] of runtimes) pending.controller.abort(new Error("MCP runtime closed")); await Promise.allSettled( - runtimes.map(async (runtime) => { - await (await runtime).close(); + runtimes.map(async ([, pending]) => { + await (await pending.promise).close(); }), ); } } -export function createMcpBroker(input: { connections: readonly McpConnection[] }): McpBroker { - return new DefaultMcpBroker(input.connections); +export function createMcpBroker(input: { + connections: readonly McpConnection[]; + discoveryTimeoutMs?: number; +}): McpBroker { + return new DefaultMcpBroker( + input.connections, + input.discoveryTimeoutMs ?? DEFAULT_DISCOVERY_TIMEOUT_MS, + ); } /** Adapt Host-owned MCP capabilities to the Harness's protocol-neutral tool seam. */ @@ -376,7 +665,7 @@ export function createMcpAgentToolSource( } = {}, ): AgentToolSource { return { - async resolve(scope) { + async resolve(scope, signal) { if (options.authorize && !(await options.authorize(scope))) { return { generation: contentFingerprint({ denied: true, actor: scope.actor?.id }).slice(0, 24), @@ -390,7 +679,9 @@ export function createMcpAgentToolSource( actorId: scope.actor ? `${scope.actor.type}:${scope.actor.id}` : "local:legacy", sessionId: scope.sessionId, }; - const catalog = await broker.catalog(brokerScope); + // Discovery used to happen on every AgentToolSource resolution. Keep that behavior while + // bounding failures per connection and retaining last-known-good tools. + const catalog = await broker.refresh(brokerScope, undefined, signal); const refsByModelName = new Map(catalog.tools.map((tool) => [tool.modelName, tool.ref])); const directDefinitions = catalog.tools.map((tool) => ({ name: tool.modelName, @@ -434,7 +725,19 @@ export function createMcpAgentToolSource( const ref = refsByModelName.get(requestedRef); if (!ref) throw new Error(`Unknown MCP model tool: ${call.name}`); const argumentsInput = progressive ? input.arguments : call.input; - return broker.call(brokerScope, ref, argumentsInput, signal); + try { + return await broker.call(brokerScope, ref, argumentsInput, signal); + } catch (error) { + if (signal.aborted) throw error; + const problem = mcpConnectionProblem(ref.connectionId, error, "invoke"); + if (problem.code === "unknown") throw error; + return { + status: "error", + summary: problem.message, + content: [], + error: { code: problem.code, retryable: problem.retryable }, + } satisfies McpToolResult; + } }, }; }, diff --git a/packages/backend/src/mcp/test-fixtures/stdio-race-server.mjs b/packages/backend/src/mcp/test-fixtures/stdio-race-server.mjs new file mode 100644 index 00000000..43370a43 --- /dev/null +++ b/packages/backend/src/mcp/test-fixtures/stdio-race-server.mjs @@ -0,0 +1,53 @@ +import { createInterface } from "node:readline"; + +const lines = createInterface({ input: process.stdin }); +let lists = 0; + +function send(message) { + process.stdout.write(`${JSON.stringify(message)}\n`); +} + +lines.on("line", (line) => { + const message = JSON.parse(line); + if (message.method === "initialize") { + send({ + jsonrpc: "2.0", + id: message.id, + result: { + protocolVersion: "2025-11-25", + capabilities: { tools: {} }, + serverInfo: { name: "OpenGUI MCP race fixture", version: "1.0.0" }, + }, + }); + return; + } + if (message.method !== "tools/list") return; + lists += 1; + const list = lists; + setTimeout( + () => { + if (list === 1 && process.env.MCP_FIRST_LIST_ERROR === "1") { + send({ + jsonrpc: "2.0", + id: message.id, + error: { code: -32_603, message: "Old discovery failed" }, + }); + return; + } + send({ + jsonrpc: "2.0", + id: message.id, + result: { + tools: [ + { + name: list === 1 ? "old_tool" : "new_tool", + description: `Catalog response ${list}`, + inputSchema: { type: "object" }, + }, + ], + }, + }); + }, + list === 1 ? 100 : 0, + ); +}); diff --git a/packages/backend/src/mcp/test-fixtures/stdio-tool-server.mjs b/packages/backend/src/mcp/test-fixtures/stdio-tool-server.mjs index 5f64b68b..43ace968 100644 --- a/packages/backend/src/mcp/test-fixtures/stdio-tool-server.mjs +++ b/packages/backend/src/mcp/test-fixtures/stdio-tool-server.mjs @@ -21,22 +21,23 @@ lines.on("line", (line) => { return; } if (message.method === "tools/list") { + if (process.env.MCP_HANG_LIST === "1") return; + const toolName = process.env.MCP_PID_TOOL === "1" ? `echo_${process.pid}` : "echo"; + const toolCount = Number.parseInt(process.env.MCP_TOOL_COUNT ?? "1", 10); send({ jsonrpc: "2.0", id: message.id, result: { - tools: [ - { - name: "echo", - title: "Echo", - description: "Return the supplied message.", - inputSchema: { - type: "object", - properties: { message: { type: "string" } }, - required: ["message"], - }, + tools: Array.from({ length: toolCount }, (_, index) => ({ + name: index === 0 ? toolName : `${toolName}_${index}`, + title: "Echo", + description: "Return the supplied message.", + inputSchema: { + type: "object", + properties: { message: { type: "string" } }, + required: ["message"], }, - ], + })), }, }); return; diff --git a/packages/backend/src/routes/host-product.ts b/packages/backend/src/routes/host-product.ts index abb322b3..4bdb60d1 100644 --- a/packages/backend/src/routes/host-product.ts +++ b/packages/backend/src/routes/host-product.ts @@ -381,9 +381,10 @@ export function registerHostProductRoutes( } }); - app.get("/api/host/mcp-connections", async () => - Response.json({ ok: true, value: (await input.getHost()).listMcpConnections() }), - ); + app.get("/api/host/mcp-connections", async () => { + const host = await input.getHost(); + return Response.json({ ok: true, value: await host.listMcpConnections() }); + }); app.post("/api/host/mcp-connections", async (c) => { try { @@ -916,6 +917,7 @@ export function registerHostProductRoutes( .map((item) => item.trim()) .filter((item) => item.length > 0) : undefined; + const interrupt = body.interrupt === true; const actor = c.get("actor") as Actor; const sessionId = c.req.param("sessionId"); if (input.identity) { @@ -930,14 +932,17 @@ export function registerHostProductRoutes( snapshot.model.modelId, ); } + const prompt = { + text, + // `skills: []` must stay distinct from omitted skills (Host defaults). + ...(skills !== undefined ? { skills } : {}), + actor: durableActor(actor), + }; return Response.json({ ok: true, - value: await host.prompt(sessionId, { - text, - // `skills: []` must stay distinct from omitted skills (Host defaults). - ...(skills !== undefined ? { skills } : {}), - actor: durableActor(actor), - }), + value: interrupt + ? await host.prompt(sessionId, prompt, prompt.actor, { interrupt: true }) + : await host.prompt(sessionId, prompt), }); } catch (error) { return sessionError(error); diff --git a/packages/backend/src/routes/host-transport.ts b/packages/backend/src/routes/host-transport.ts index 793e2749..a48910c0 100644 --- a/packages/backend/src/routes/host-transport.ts +++ b/packages/backend/src/routes/host-transport.ts @@ -51,7 +51,7 @@ export function registerHostTransportRoutes(app: BackendApp, deps: HostTransport ok: true, value: { protocolVersion: 1, - appVersion: process.env.npm_package_version || "0.0.0", + appVersion: process.env.OPENGUI_VERSION || process.env.npm_package_version || "0.0.0", }, }), ); diff --git a/packages/harness/src/harness.ts b/packages/harness/src/harness.ts index d470094b..8300f811 100644 --- a/packages/harness/src/harness.ts +++ b/packages/harness/src/harness.ts @@ -113,6 +113,8 @@ export interface HarnessSession { updateFollowUp(followUpId: string, prompt: PromptInput): Promise; reorderFollowUp(followUpId: string, index: number): Promise; removeFollowUp(followUpId: string): Promise; + /** Remove a pending follow-up and return it for immediate dispatch. */ + takeFollowUp(followUpId: string): Promise; abort(): Promise; setModel(selection: ModelSelection): Promise; setReasoning(reasoning: ReasoningLevel): Promise; diff --git a/packages/harness/src/open-gui-harness.ts b/packages/harness/src/open-gui-harness.ts index 252edd62..5fe94b3b 100644 --- a/packages/harness/src/open-gui-harness.ts +++ b/packages/harness/src/open-gui-harness.ts @@ -183,6 +183,10 @@ class HarnessSessionImpl implements HarnessSession { await this.#harness.removeFollowUp(this.#id, followUpId); } + async takeFollowUp(followUpId: string) { + return await this.#harness.takeFollowUp(this.#id, followUpId); + } + async abort() { this.#harness.abort(this.#id); } @@ -491,6 +495,11 @@ class OpenGuiHarnessImpl implements OpenGuiHarness { await this.#store.removeFollowUp(sessionId, followUpId); } + async takeFollowUp(sessionId: string, followUpId: string) { + this.#assertOpen(); + return await this.#store.takePendingFollowUp(sessionId, followUpId); + } + abort(sessionId: string) { this.#assertOpen(); this.#abortControllers.get(sessionId)?.abort(); diff --git a/packages/harness/src/storage/sqlite-store.ts b/packages/harness/src/storage/sqlite-store.ts index bdb194f7..0a58310e 100644 --- a/packages/harness/src/storage/sqlite-store.ts +++ b/packages/harness/src/storage/sqlite-store.ts @@ -352,6 +352,31 @@ export class SqliteSessionStore { if (result.numDeletedRows !== 1n) throw new Error(`Pending follow-up not found: ${followUpId}`); } + /** Atomically claim a pending follow-up for immediate dispatch (send-now). */ + async takePendingFollowUp(sessionId: string, followUpId: string) { + await this.#ready; + return this.#database.transaction().execute(async (transaction) => { + const row = await transaction + .selectFrom("session_follow_ups") + .select(["id", "sequence", "prompt_json", "created_at"]) + .where("session_id", "=", sessionId) + .where("id", "=", followUpId) + .where("state", "=", "pending") + .executeTakeFirst(); + if (!row) throw new Error(`Pending follow-up not found: ${followUpId}`); + const deleted = await transaction + .deleteFrom("session_follow_ups") + .where("id", "=", followUpId) + .where("session_id", "=", sessionId) + .where("state", "=", "pending") + .executeTakeFirst(); + if (deleted.numDeletedRows !== 1n) { + throw new Error(`Pending follow-up not found: ${followUpId}`); + } + return decodeFollowUp(row); + }); + } + async reorderFollowUp(sessionId: string, followUpId: string, requestedIndex: number) { await this.#ready; await this.#database.transaction().execute(async (transaction) => { diff --git a/server/start-web-server.ts b/server/start-web-server.ts index d833891f..c912e3c8 100644 --- a/server/start-web-server.ts +++ b/server/start-web-server.ts @@ -1,5 +1,6 @@ import { serve as honoServe } from "@hono/node-server"; import { createBackendHost as createHost } from "@opengui/backend"; +import packageJson from "../package.json" with { type: "json" }; type BackendHost = ReturnType; type Server = ReturnType; @@ -12,6 +13,7 @@ export interface WebServerDependencies { } export function startWebServer(dependencies: Partial = {}): Server { + process.env.OPENGUI_VERSION ??= packageJson.version; const createBackendHost = dependencies.createBackendHost ?? createHost; const serve = dependencies.serve ?? honoServe; const info = dependencies.info ?? console.info; diff --git a/src/components/PromptBox.test.tsx b/src/components/PromptBox.test.tsx index d3c4a301..539c115b 100644 --- a/src/components/PromptBox.test.tsx +++ b/src/components/PromptBox.test.tsx @@ -147,12 +147,26 @@ describe("PromptBox interactions", () => { "prompt.steerDirection", ); expect(screen.getByText("context 73")).toBeTruthy(); - await userEvent.click(screen.getByRole("button", { name: "prompt.queue" })); + await userEvent.click(screen.getByRole("button", { name: "prompt.steer" })); await userEvent.click(screen.getByRole("button", { name: "prompt.stopGenerating" })); expect(onQueueModeChange).toHaveBeenCalledWith("queue"); expect(fixture.stop).toHaveBeenCalledOnce(); }); + test("toggles from queue mode into steer mode", async () => { + const onQueueModeChange = vi.fn(); + render( + , + ); + await userEvent.click(screen.getByRole("button", { name: "prompt.queue" })); + expect(onQueueModeChange).toHaveBeenCalledWith("after-part"); + }); + test("requires a selected model before enabling send", async () => { fixture.selectedModel = null; const { unmount } = render( diff --git a/src/components/PromptBox.tsx b/src/components/PromptBox.tsx index b2ea15cc..1bd718ce 100644 --- a/src/components/PromptBox.tsx +++ b/src/components/PromptBox.tsx @@ -1,4 +1,4 @@ -import { ArrowUp, ListEnd, Square } from "lucide-react"; +import { ArrowUp, ListEnd, Square, Compass } from "lucide-react"; import * as React from "react"; import { useTranslation } from "react-i18next"; import { FileMentionPopover } from "@/components/FileMentionPopover"; @@ -296,10 +296,10 @@ export const PromptBox = React.forwardRef( isDisabled ? t("prompt.selectOrCreateSession") : isLoading - ? queueMode === "interrupt" - ? t("prompt.interruptAndSend") - : queueMode === "after-part" - ? t("prompt.steerDirection") + ? queueMode === "after-part" + ? t("prompt.steerDirection") + : queueMode === "interrupt" + ? t("prompt.interruptAndSend") : t("prompt.queueMessage") : t("prompt.message") } @@ -326,19 +326,23 @@ export const PromptBox = React.forwardRef( type="button" variant="ghost" size="sm" - title={t("prompt.queueTitle")} + title={queueMode === "after-part" ? t("prompt.steerTitle") : t("prompt.queueTitle")} className="!h-7 shrink-0 gap-1.5 px-1.5 text-xs text-muted-foreground hover:text-foreground" onClick={(e) => { e.stopPropagation(); - onQueueModeChange("queue"); + onQueueModeChange(queueMode === "after-part" ? "queue" : "after-part"); }} > - + {queueMode === "after-part" ? ( + + ) : ( + + )} - {t("prompt.queue")} + {queueMode === "after-part" ? t("prompt.steer") : t("prompt.queue")} )} diff --git a/src/components/sidebar/ProjectEntry.test.tsx b/src/components/sidebar/ProjectEntry.test.tsx index 7d02a2c7..1b081fd0 100644 --- a/src/components/sidebar/ProjectEntry.test.tsx +++ b/src/components/sidebar/ProjectEntry.test.tsx @@ -70,11 +70,13 @@ describe("ProjectEntry", () => { test("creates a Session and collapses an expanded connected Project", async () => { const input = props(); render(); - expect(screen.getByText("sidebar.noSessionsYet")).toBeTruthy(); + expect(screen.getByRole("button", { name: "sidebar.newSession" }).textContent).toContain( + "projectMenu.newSession", + ); await userEvent.click(screen.getByRole("button", { name: "projectMenu.newSession" })); expect(input.setActiveTarget).toHaveBeenCalledWith("/work/Alpha", { newChat: true }); expect(input.closeMobileSidebar).toHaveBeenCalled(); - await userEvent.click(screen.getByRole("button", { name: "Alpha" })); + await userEvent.click(screen.getByText("Alpha").closest("button")!); expect(input.toggleCollapsed).toHaveBeenCalledWith("/work/Alpha"); }); diff --git a/src/features/app-shell/useAppKeyboardShortcuts.test.tsx b/src/features/app-shell/useAppKeyboardShortcuts.test.tsx index 92ace74f..de4cd7ca 100644 --- a/src/features/app-shell/useAppKeyboardShortcuts.test.tsx +++ b/src/features/app-shell/useAppKeyboardShortcuts.test.tsx @@ -43,7 +43,7 @@ describe("application keyboard shortcuts", () => { vi.useRealTimers(); }); - test("toggles queue steering only for a running Session outside dialogs", () => { + test("toggles queue steer only for a running Session outside dialogs", () => { render(); fireEvent.keyDown(window, { key: "d", ctrlKey: true }); expect(screen.getByText("after-part")).toBeTruthy(); @@ -52,6 +52,8 @@ describe("application keyboard shortcuts", () => { ctrlKey: true, }); expect(screen.getByText("after-part")).toBeTruthy(); + fireEvent.keyDown(window, { key: "d", ctrlKey: true }); + expect(screen.getByText("queue")).toBeTruthy(); }); test("protects editable text from undo and sidebar search shortcuts", () => { diff --git a/src/features/host-provider/HostProvider.render.test.tsx b/src/features/host-provider/HostProvider.render.test.tsx index b55eceb0..c2476141 100644 --- a/src/features/host-provider/HostProvider.render.test.tsx +++ b/src/features/host-provider/HostProvider.render.test.tsx @@ -307,6 +307,43 @@ describe("HostProvider render integration", () => { }); }); + test("steer mode queues an after-part Follow-up without interrupting immediately", async () => { + host.prompt.mockResolvedValue({ + mode: "follow_up", + followUp: { + id: "follow-steer", + sequence: 1, + prompt: { text: "steer me" }, + createdAt: "2026-01-01T00:00:02.000Z", + }, + }); + + render( + + + + + , + ); + await waitFor(() => expect(workspace.bootState).toBe("ready")); + await act(() => actions.selectSession("session-1")); + await waitFor(() => + expect(messages[0]?.parts?.[0]).toMatchObject({ text: "hydrated transcript" }), + ); + + await act(() => actions.sendPrompt("steer me", "after-part", ["code-review"])); + + expect(host.prompt).toHaveBeenCalledWith("session-1", "steer me", { + skills: ["code-review"], + }); + expect(host.sendFollowUpNow).not.toHaveBeenCalled(); + expect(actions.getQueuedPrompts("session-1")).toEqual([ + expect.objectContaining({ id: "follow-steer", text: "steer me", mode: "after-part" }), + ]); + }); + test("surfaces bootstrap failure as a stable error state", async () => { host.health.mockRejectedValueOnce(new Error("Host offline")); render( diff --git a/src/features/host-provider/HostProvider.tsx b/src/features/host-provider/HostProvider.tsx index f001fd8c..5ed9d278 100644 --- a/src/features/host-provider/HostProvider.tsx +++ b/src/features/host-provider/HostProvider.tsx @@ -248,6 +248,9 @@ function HostProviderBody({ const hydratingSessionIdsRef = useRef(new Set()); const activeSessionIdRef = useRef(activeSessionId); const queuedPromptsRef = useRef(queuedPrompts); + /** Steer (after-part) Follow-ups waiting for the current model part to end. */ + const afterPartFollowUpsRef = useRef>({}); + const steeringSessionIdsRef = useRef(new Set()); queuedPromptsRef.current = queuedPrompts; const queueController = useMemo( @@ -446,6 +449,38 @@ function HostProviderBody({ [host, transcriptStore], ); + const forgetAfterPartFollowUp = useCallback((sessionId: string, followUpId: string) => { + const pending = afterPartFollowUpsRef.current[sessionId]; + if (!pending?.length) return; + const next = pending.filter((id) => id !== followUpId); + if (next.length === pending.length) return; + if (next.length === 0) delete afterPartFollowUpsRef.current[sessionId]; + else afterPartFollowUpsRef.current[sessionId] = next; + }, []); + + const dispatchAfterPartSteer = useCallback( + (sessionId: string) => { + const followUpId = afterPartFollowUpsRef.current[sessionId]?.[0]; + if (!followUpId || steeringSessionIdsRef.current.has(sessionId)) return; + steeringSessionIdsRef.current.add(sessionId); + void (async () => { + try { + await queueController?.sendNow(sessionId, followUpId); + forgetAfterPartFollowUp(sessionId, followUpId); + if (activeSessionIdRef.current === sessionId) { + await hydrateTranscript(sessionId); + } + } catch { + // Run may have completed and already claimed this Follow-up as FIFO. + forgetAfterPartFollowUp(sessionId, followUpId); + } finally { + steeringSessionIdsRef.current.delete(sessionId); + } + })(); + }, + [forgetAfterPartFollowUp, hydrateTranscript, queueController], + ); + useEffect(() => { let cancelled = false; void (async () => { @@ -500,8 +535,11 @@ function HostProviderBody({ transcriptStore, refreshSessions, hydrateTranscript, - onFollowUpDispatched: (sessionId, followUpId) => - queueController?.recordDispatched(sessionId, followUpId), + onFollowUpDispatched: (sessionId, followUpId) => { + forgetAfterPartFollowUp(sessionId, followUpId); + queueController?.recordDispatched(sessionId, followUpId); + }, + onModelPartEnded: (sessionId) => dispatchAfterPartSteer(sessionId), }); const activeTranscript = useActiveTranscriptSnapshot(); @@ -529,7 +567,7 @@ function HostProviderBody({ items.map((item) => ({ id: item.id, text: item.text, - mode: "queue" as const, + mode: item.mode, createdAt: Date.now(), actor: item.actor, })), @@ -707,7 +745,7 @@ function HostProviderBody({ await requireHost().renameSession(id, title); await refreshSessions(); }, - sendPrompt: async (text, _mode, submittedSkills) => { + sendPrompt: async (text, mode, submittedSkills) => { // The caller captures the visible selection at submit time. Normalize it // before any asynchronous Session creation/refresh can change active keys. const submittedAllowlist = @@ -718,6 +756,8 @@ function HostProviderBody({ submittedSkills.map((name) => name.trim()).filter((name) => name.length > 0), ), ); + // Steer is after-part. interrupt remains available for programmatic send-now paths. + const queueMode = mode === "after-part" || mode === "interrupt" ? mode : "queue"; let optimisticMessage: { scope: { directory: string; sessionId: string }; id: string; @@ -825,18 +865,26 @@ function HostProviderBody({ } } } - const result = await requireHost().prompt( - sessionId, - text, - skillsAllowlist !== undefined ? { skills: skillsAllowlist } : undefined, - ); + const result = await requireHost().prompt(sessionId, text, { + ...(skillsAllowlist !== undefined ? { skills: skillsAllowlist } : {}), + // interrupt = send immediately (used by send-now). Steer/after-part queues + // and dispatches when the current model part ends. + ...(queueMode === "interrupt" ? { interrupt: true } : {}), + }); if (result.mode === "follow_up") { - transcriptStore.dispatch({ - type: "message.removed", - scope, - messageId: optimisticMessageId, - }); - queueController?.recordEnqueued(sessionId, result.followUp); + if (optimisticMessage) { + transcriptStore.dispatch({ + type: "message.removed", + scope, + messageId: optimisticMessageId, + }); + } + const followUpMode = queueMode === "after-part" ? "after-part" : "queue"; + queueController?.recordEnqueued(sessionId, result.followUp, followUpMode); + if (followUpMode === "after-part") { + const pending = afterPartFollowUpsRef.current[sessionId] ?? []; + afterPartFollowUpsRef.current[sessionId] = [...pending, result.followUp.id]; + } } else { let stream = activeStreamRef.current; if (stream?.snapshot.id === sessionId) { @@ -857,6 +905,9 @@ function HostProviderBody({ nextCursor: null, }); } + if (queueMode === "interrupt") { + await hydrateTranscript(sessionId); + } } } catch (error) { if (optimisticMessage) { @@ -983,11 +1034,12 @@ function HostProviderBody({ (queuedPrompts[sessionId] ?? []).map((item) => ({ id: item.id, text: item.text, - mode: "queue" as const, + mode: item.mode, createdAt: Date.now(), actor: item.actor, })), removeFromQueue: (sessionId, promptId) => { + forgetAfterPartFollowUp(sessionId, promptId); void queueController?.remove(sessionId, promptId).catch(notifyUnknownError); }, reorderQueue: (sessionId, fromIndex, toIndex) => { @@ -997,7 +1049,13 @@ function HostProviderBody({ void queueController?.update(sessionId, promptId, text).catch(notifyUnknownError); }, sendQueuedNow: async (sessionId, promptId) => { + forgetAfterPartFollowUp(sessionId, promptId); await queueController?.sendNow(sessionId, promptId); + // send-now aborts the live Run then starts a new one. Rehydrate so any + // in-flight stream buffers cannot leave ghost assistant rows behind. + if (activeSessionIdRef.current === sessionId) { + await hydrateTranscript(sessionId); + } }, setSessionDraft: (key, text) => setSessionDrafts((current) => ({ ...current, [key]: text })), clearSessionDraft: (key) => @@ -1152,10 +1210,12 @@ function HostProviderBody({ busySessionIds, detachedProject, currentActor, + forgetAfterPartFollowUp, host, hydrateTranscript, rememberActiveSession, queuedPrompts, + queueController, refreshModels, replaceProjects, requireHost, diff --git a/src/features/host-provider/host-actions.ts b/src/features/host-provider/host-actions.ts index 47ae5c4d..42667c8e 100644 --- a/src/features/host-provider/host-actions.ts +++ b/src/features/host-provider/host-actions.ts @@ -5,17 +5,20 @@ import type { ActorSnapshot, HostFollowUp, OpenGuiHostClient } from "@/protocol/ export type HostQueueItem = { id: string; text: string; - mode: "queue"; + mode: "queue" | "after-part"; actor?: ActorSnapshot; }; export type HostQueueState = Record; type UpdateHostQueueState = (update: (current: HostQueueState) => HostQueueState) => void; -export function projectHostFollowUps(followUps: HostFollowUp[]): HostQueueItem[] { +export function projectHostFollowUps( + followUps: HostFollowUp[], + mode: HostQueueItem["mode"] = "queue", +): HostQueueItem[] { return followUps.map((item) => ({ id: item.id, text: item.prompt.text, - mode: "queue", + mode, actor: item.prompt.actor, })); } @@ -29,15 +32,24 @@ export class HostQueueController { ) {} #replace(sessionId: string, followUps: HostFollowUp[]) { - this.updateState((current) => ({ ...current, [sessionId]: projectHostFollowUps(followUps) })); + this.updateState((current) => { + const previousModes = new Map((current[sessionId] ?? []).map((item) => [item.id, item.mode])); + return { + ...current, + [sessionId]: projectHostFollowUps(followUps).map((item) => ({ + ...item, + mode: previousModes.get(item.id) ?? "queue", + })), + }; + }); } - recordEnqueued(sessionId: string, followUp: HostFollowUp) { + recordEnqueued(sessionId: string, followUp: HostFollowUp, mode: HostQueueItem["mode"] = "queue") { this.updateState((current) => ({ ...current, [sessionId]: [ ...(current[sessionId] ?? []).filter((item) => item.id !== followUp.id), - ...projectHostFollowUps([followUp]), + ...projectHostFollowUps([followUp], mode), ], })); } diff --git a/src/features/host-provider/host-domain-state.ts b/src/features/host-provider/host-domain-state.ts index 692b3489..f9028d8a 100644 --- a/src/features/host-provider/host-domain-state.ts +++ b/src/features/host-provider/host-domain-state.ts @@ -10,7 +10,7 @@ export type HostBootState = "idle" | "checking-server" | "starting-server" | "re export type HostQueuedPrompt = { id: string; text: string; - mode: "queue"; + mode: "queue" | "after-part"; actor?: ActorSnapshot; }; diff --git a/src/features/host-provider/host-event-stream.test.ts b/src/features/host-provider/host-event-stream.test.ts index f80284a5..649f6de2 100644 --- a/src/features/host-provider/host-event-stream.test.ts +++ b/src/features/host-provider/host-event-stream.test.ts @@ -107,4 +107,24 @@ describe("Host event dispatch", () => { expect(onFollowUpDispatched).toHaveBeenCalledWith("session-1", "follow-up-1"); }); + + it("notifies when a model part ends so Steer can dispatch", () => { + const onModelPartEnded = vi.fn(); + const dispatch = createHostEventDispatcher({ + activeStreamRef: { current: null }, + setActiveSnapshot: vi.fn(), + setBusySessionIds: vi.fn(), + transcriptStore: { dispatch: vi.fn() } as never, + refreshSessions: vi.fn(async () => {}), + onModelPartEnded, + }); + + dispatch(entryEvent("assistant_message")); + dispatch(entryEvent("tool_call")); + dispatch(entryEvent("user_message")); + + expect(onModelPartEnded).toHaveBeenCalledTimes(2); + expect(onModelPartEnded).toHaveBeenNthCalledWith(1, "session-1"); + expect(onModelPartEnded).toHaveBeenNthCalledWith(2, "session-1"); + }); }); diff --git a/src/features/host-provider/host-event-stream.ts b/src/features/host-provider/host-event-stream.ts index e2be301d..5a2736df 100644 --- a/src/features/host-provider/host-event-stream.ts +++ b/src/features/host-provider/host-event-stream.ts @@ -49,6 +49,16 @@ export interface HostEventDispatcherDependencies { transcriptStore: ActiveSessionTranscriptStore; refreshSessions: () => Promise; onFollowUpDispatched?: (sessionId: string, followUpId: string) => void; + /** Fires when the live model part ends (text committed or tool calls begin). */ + onModelPartEnded?: (sessionId: string) => void; +} + +function isModelPartEndEvent(hostEvent: HostEvent): boolean { + return ( + hostEvent.event.type === "entry_appended" && + (hostEvent.event.entry.kind === "assistant_message" || + hostEvent.event.entry.kind === "tool_call") + ); } /** Applies each event synchronously so deltas retain the Host's delivery order. */ @@ -59,6 +69,7 @@ export function createHostEventDispatcher({ transcriptStore, refreshSessions, onFollowUpDispatched, + onModelPartEnded, }: HostEventDispatcherDependencies): (hostEvent: HostEvent) => void { return (hostEvent) => { const terminal = isTerminalHostEvent(hostEvent); @@ -89,6 +100,7 @@ export function createHostEventDispatcher({ }); } + if (isModelPartEndEvent(hostEvent)) onModelPartEnded?.(hostEvent.sessionId); if (terminal) void refreshSessions().catch(notifyUnknownError); }; } @@ -124,12 +136,16 @@ export function useHostEventStream(options: UseHostEventStreamOptions): void { const onFollowUpDispatchedRef = useRef(options.onFollowUpDispatched); onFollowUpDispatchedRef.current = options.onFollowUpDispatched; + const onModelPartEndedRef = useRef(options.onModelPartEnded); + onModelPartEndedRef.current = options.onModelPartEnded; + useEffect(() => { if (!options.host) return; const dispatchEvent = createHostEventDispatcher({ ...options, onFollowUpDispatched: (sessionId, followUpId) => onFollowUpDispatchedRef.current?.(sessionId, followUpId), + onModelPartEnded: (sessionId) => onModelPartEndedRef.current?.(sessionId), }); return options.host.subscribe( dispatchEvent, diff --git a/src/features/mcp/McpSettings.render.test.tsx b/src/features/mcp/McpSettings.render.test.tsx index 92269f7f..bd6c2c9f 100644 --- a/src/features/mcp/McpSettings.render.test.tsx +++ b/src/features/mcp/McpSettings.render.test.tsx @@ -3,9 +3,10 @@ import { cleanup, render, screen, waitFor } from "@testing-library/react"; import userEvent from "@testing-library/user-event"; import { afterEach, describe, expect, test, vi } from "vitest"; +import type { HostMcpConnection } from "@/protocol/host-types"; const host = { - listMcpConnections: vi.fn(async () => []), + listMcpConnections: vi.fn<() => Promise>(async () => []), upsertMcpConnection: vi.fn(async (connection) => connection), inspectMcpConnection: vi.fn(async () => []), removeMcpConnection: vi.fn(async () => undefined), @@ -29,6 +30,33 @@ describe("McpSettings", () => { vi.clearAllMocks(); }); + test("shows actionable health for an unavailable connection", async () => { + host.listMcpConnections.mockResolvedValueOnce([ + { + id: "search", + label: "Search tools", + enabled: true, + transport: { kind: "http", url: "https://example.com/mcp" }, + status: { + state: "offline", + toolCount: 0, + lastCheckedAt: "2026-07-30T12:00:00.000Z", + problem: { + code: "timeout", + stage: "connect", + message: "MCP connection timed out", + retryable: true, + }, + }, + }, + ] as HostMcpConnection[]); + + render(); + + await waitFor(() => expect(screen.getByText("mcp.status.offline")).toBeTruthy()); + expect(screen.getByText("mcp.problem.timeout")).toBeTruthy(); + }); + test("requires explicit approval of the visible stdio command before saving", async () => { render(); await waitFor(() => expect(screen.getByText("mcp.emptyTitle")).toBeTruthy()); @@ -58,4 +86,38 @@ describe("McpSettings", () => { }), ); }); + + test("polls until background discovery leaves the refreshing state", async () => { + const connection = { + id: "calendar", + label: "Calendar", + enabled: true, + transport: { + kind: "http" as const, + url: "https://example.com/mcp", + bearerTokenConfigured: false, + }, + }; + host.listMcpConnections + .mockResolvedValueOnce([ + { ...connection, status: { state: "refreshing" as const, toolCount: 0 } }, + ]) + .mockResolvedValueOnce([ + { + ...connection, + status: { + state: "ready" as const, + toolCount: 2, + lastCheckedAt: "2026-07-31T12:00:00.000Z", + }, + }, + ]); + + render(); + await waitFor(() => expect(screen.getByText("mcp.status.refreshing")).toBeTruthy()); + await waitFor(() => expect(screen.getByText("mcp.status.ready")).toBeTruthy(), { + timeout: 2_000, + }); + expect(host.listMcpConnections).toHaveBeenCalledTimes(2); + }); }); diff --git a/src/features/mcp/McpSettings.tsx b/src/features/mcp/McpSettings.tsx index 5278bd94..9cdae324 100644 --- a/src/features/mcp/McpSettings.tsx +++ b/src/features/mcp/McpSettings.tsx @@ -105,6 +105,14 @@ export function McpSettings() { }); }, []); + useEffect(() => { + if (!connections.some((connection) => connection.status?.state === "refreshing")) return; + const timer = window.setTimeout(() => { + void reload().catch(notifyUnknownError); + }, 1_000); + return () => window.clearTimeout(timer); + }, [connections]); + const updateDraft = (key: K, value: Draft[K]) => { setDraft((current) => (current ? { ...current, [key]: value } : current)); }; @@ -172,8 +180,10 @@ export function McpSettings() { try { const result = await host.inspectMcpConnection(connection.id); setTools((current) => ({ ...current, [connection.id]: result })); + await reload(); } catch (error) { notifyUnknownError(error); + await reload().catch(() => undefined); } finally { setInspecting(null); } @@ -369,14 +379,40 @@ export function McpSettings() {

{connection.label}

- - {t(connection.enabled ? "mcp.enabled" : "mcp.disabled")} + + {t( + connection.status + ? `mcp.status.${connection.status.state}` + : connection.enabled + ? "mcp.enabled" + : "mcp.disabled", + )} {t(`mcp.transport.${connection.transport.kind}`)}

{connection.id}

+ {connection.status && ( +

+ {connection.status.problem + ? t(`mcp.problem.${connection.status.problem.code}`) + : t("mcp.toolCount", { count: connection.status.toolCount })} +

+ )}
{ }); }); +describe("Host prompt client", () => { + test("forwards interrupt prompts without dropping an empty skill allowlist", async () => { + const requests: Array<{ url: string; method: string; body: string | null }> = []; + const client = createHostClient({ + baseUrl: "http://host.test", + fetchImpl: async (input, init) => { + requests.push({ + url: String(input), + method: init?.method ?? "GET", + body: typeof init?.body === "string" ? init.body : null, + }); + return Response.json({ + ok: true, + value: { mode: "run", startedEntries: [] }, + }); + }, + }); + + await client.prompt("session/1", "cut in", { skills: [], interrupt: true }); + + expect(requests).toEqual([ + { + url: "http://host.test/api/host/sessions/session%2F1/prompt", + method: "POST", + body: JSON.stringify({ text: "cut in", skills: [], interrupt: true }), + }, + ]); + }); +}); + describe("Host follow-up client", () => { test("exposes queue management through the Host API", async () => { const requests: Array<{ url: string; method: string; body: string | null }> = []; diff --git a/src/protocol/host-client.ts b/src/protocol/host-client.ts index 055c7235..89647b4d 100644 --- a/src/protocol/host-client.ts +++ b/src/protocol/host-client.ts @@ -240,6 +240,7 @@ export function createHostClient(options: CreateHostClientOptions = {}): OpenGui text, // Preserve empty arrays: they mean "no skills", not "use defaults". ...(options?.skills !== undefined ? { skills: options.skills } : {}), + ...(options?.interrupt ? { interrupt: true } : {}), }), })) as | { mode: "run"; startedEntries: HostSessionSnapshot["entries"] } diff --git a/src/protocol/host-transcript.test.ts b/src/protocol/host-transcript.test.ts index c4b5c40d..6e0b19ee 100644 --- a/src/protocol/host-transcript.test.ts +++ b/src/protocol/host-transcript.test.ts @@ -141,6 +141,83 @@ describe("Host transcript streaming", () => { ]); }); + test("clears live stream buffers on abort so send-now does not leave ghost assistants", () => { + let stream = createHostTranscriptStream(snapshot()); + stream = applyHostTranscriptEvent(stream, { + sessionId: "session-1", + event: { + type: "entry_appended", + entry: { + id: "user-1", + sessionId: "session-1", + sequence: 1, + kind: "user_message", + payload: { text: "first", runId: "run-1" }, + createdAt: "2026-07-10T00:00:01.000Z", + }, + }, + }); + stream = applyHostTranscriptEvent(stream, { + sessionId: "session-1", + event: { type: "assistant_delta", runId: "run-1", delta: "partial answer" }, + }); + stream = applyHostTranscriptEvent(stream, { + sessionId: "session-1", + event: { type: "reasoning_delta", runId: "run-1", delta: "thinking" }, + }); + expect(projectHostTranscriptStream(stream).map((message) => message.info.id)).toEqual([ + "user-1", + "stream:run-1", + ]); + + stream = applyHostTranscriptEvent(stream, { + sessionId: "session-1", + event: { + type: "entry_appended", + entry: { + id: "abort-1", + sessionId: "session-1", + sequence: 2, + kind: "run_aborted", + payload: { runId: "run-1" }, + createdAt: "2026-07-10T00:00:02.000Z", + }, + }, + }); + stream = applyHostTranscriptEvent(stream, { + sessionId: "session-1", + event: { + type: "entry_appended", + entry: { + id: "user-2", + sessionId: "session-1", + sequence: 3, + kind: "user_message", + payload: { text: "send now please", runId: "run-2" }, + createdAt: "2026-07-10T00:00:03.000Z", + }, + }, + }); + stream = applyHostTranscriptEvent(stream, { + sessionId: "session-1", + event: { type: "assistant_delta", runId: "run-2", delta: "fresh answer" }, + }); + + const messages = projectHostTranscriptStream(stream); + expect(messages.map((message) => message.info.id)).toEqual([ + "user-1", + "user-2", + "stream:run-2", + ]); + expect(messages.map((message) => message.parts[0])).toMatchObject([ + { type: "text", text: "first" }, + { type: "text", text: "send now please" }, + { type: "text", text: "fresh answer" }, + ]); + expect(JSON.stringify(messages)).not.toContain("partial answer"); + expect(JSON.stringify(messages)).not.toContain("thinking"); + }); + test("shows assistant deltas immediately and replaces them with the durable message", () => { let stream = createHostTranscriptStream(snapshot()); stream = applyHostTranscriptEvent(stream, { diff --git a/src/protocol/host-transcript.ts b/src/protocol/host-transcript.ts index 8aadad83..87f5a3ed 100644 --- a/src/protocol/host-transcript.ts +++ b/src/protocol/host-transcript.ts @@ -46,6 +46,20 @@ export function applyHostTranscriptEvent( const runId = text(event.entry.payload.runId); if (runId) delete reasoningTextByRun[runId]; } + // Terminal run markers must drop live buffers. Otherwise abort/send-now leaves + // ghost stream:* assistant rows that corrupt message history after the next turn. + if ( + event.entry.kind === "run_completed" || + event.entry.kind === "run_aborted" || + event.entry.kind === "run_failed" || + event.entry.kind === "run_interrupted" + ) { + const runId = text(event.entry.payload.runId); + if (runId) { + delete assistantTextByRun[runId]; + delete reasoningTextByRun[runId]; + } + } return { snapshot: { ...stream.snapshot, diff --git a/src/protocol/host-types.ts b/src/protocol/host-types.ts index 757c8d8e..b5ec5a10 100644 --- a/src/protocol/host-types.ts +++ b/src/protocol/host-types.ts @@ -38,7 +38,19 @@ export interface HostModelOffering { updatedAt: number; } -export type HostMcpConnection = +export interface HostMcpConnectionStatus { + state: "disabled" | "refreshing" | "ready" | "degraded" | "offline"; + toolCount: number; + lastCheckedAt?: string; + problem?: { + code: string; + stage: string; + message: string; + retryable: boolean; + }; +} + +export type HostMcpConnection = ( | { id: string; label: string; @@ -56,7 +68,8 @@ export type HostMcpConnection = label: string; enabled: boolean; transport: { kind: "http"; url: string; bearerTokenConfigured: boolean }; - }; + } +) & { status?: HostMcpConnectionStatus }; export type HostMcpConnectionMutation = | { @@ -251,7 +264,7 @@ export interface OpenGuiHostClient { prompt( sessionId: string, text: string, - options?: { skills?: string[] }, + options?: { skills?: string[]; interrupt?: boolean }, ): Promise< | { mode: "run"; startedEntries: HostSessionEntry[] } | { mode: "follow_up"; followUp: HostFollowUp } From b268b2c42c0893d93e8e67aee841981849ff8f20 Mon Sep 17 00:00:00 2001 From: akemmanuel Date: Sat, 1 Aug 2026 12:04:00 -0300 Subject: [PATCH 2/9] docs: accept model offerings ADR and update architecture --- CONTRIBUTING.md | 9 +- ...exible-users-access-and-model-offerings.md | 12 +- docs/adr/README.md | 5 +- docs/architecture.md | 216 +++++++----------- docs/bridge-lint-cleanup-workflows.md | 6 +- docs/harness-bridge-contract.md | 6 +- docs/plans/first-party-harness-replacement.md | 122 +++++----- ...exible-users-access-and-model-offerings.md | 29 +-- packages/backend/README.md | 6 +- 9 files changed, 197 insertions(+), 214 deletions(-) diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 64c1b3c4..fa97317b 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -2,7 +2,8 @@ ## Setup -Install Node.js 22.5 or newer and pnpm 11.8.0, then run: +The application supports Node.js 22.19 or newer. Contributors and release gates use Node.js 24 +because the complete QA toolchain requires it. Install Node.js 24 and pnpm 11.8.0, then run: ```bash pnpm install @@ -18,13 +19,15 @@ OpenGUI has one product path: 1. The preserved React frontend calls the OpenGUI Host. 2. The Host owns projects, model connections, Sessions, and transport. 3. The first-party Harness owns execution and the append-oriented SQLite Session log. -4. Model turns can invoke only `read`, `write`, `edit`, and `shell`. +4. Model turns can invoke the four native tools (`read`, `write`, `edit`, and `shell`) plus + explicitly configured, Host-authorized MCP tools through the narrow seam in ADR 0015. Do not add external coding-agent SDKs, CLI adapters, compatibility facades, alternate Session -identity schemes, Git/worktree orchestration, MCP, or plugin runtimes. +identity schemes, Git/worktree orchestration, frontend-owned MCP runtimes, or plugin runtimes. Read [`CONTEXT.md`](CONTEXT.md), [`docs/architecture.md`](docs/architecture.md), and [`docs/adr/0010-first-party-opengui-harness.md`](docs/adr/0010-first-party-opengui-harness.md) +and [`docs/adr/0015-host-owned-mcp-tool-connections.md`](docs/adr/0015-host-owned-mcp-tool-connections.md) before changing product boundaries. ## Quality diff --git a/docs/adr/0014-flexible-users-access-and-model-offerings.md b/docs/adr/0014-flexible-users-access-and-model-offerings.md index 88a80451..23777801 100644 --- a/docs/adr/0014-flexible-users-access-and-model-offerings.md +++ b/docs/adr/0014-flexible-users-access-and-model-offerings.md @@ -6,7 +6,17 @@ This ADR decides the target shape for a flexible **users and access** system—i ## Status -proposed +accepted + +Accepted by implementation on 2026-07-28. The merged vertical slice includes model offerings and +entitlements, Host-side offering resolution, `admin` / `viewer` roles, direct-user collaborative +`run` authorization, backend-oriented settings, and the corresponding integration tests. Evidence: +implementation commit `28ca43c`, the +[`flexible users and model offerings` plan](../plans/flexible-users-access-and-model-offerings.md), +and `packages/backend/src/identity/roles-model-offerings.integration.test.ts`. Later +auth strategies, Host login methods, secret-custody upgrades, and named Teams remain phased work; +they do not reopen the object and authorization boundaries decided here. In particular, this +acceptance does not claim that deferred plan phases 2, 4, or 5 have shipped. ## North star diff --git a/docs/adr/README.md b/docs/adr/README.md index 7f73b829..0fa70e76 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -17,7 +17,8 @@ Read these when code or docs disagree about **who owns what**. | [0011](./0011-host-embedded-accounts-and-teams.md) | **Host-embedded Accounts & Team membership**; invite-only Remote Host auth; API keys; shared Sessions | | [0012](./0012-host-path-grants-and-tool-enforcement.md) | **Host path grants**; canonical roots, restricted actors, complete tool/transport mediation | | [0013](./0013-multi-user-host-access-model.md) | **Multi-user Host access**; user-default shares, session ACL, model planes, registration modes | -| [0014](./0014-flexible-users-access-and-model-offerings.md) | **Flexible users & model offerings** (proposed); backends, credentials, slugs, roles/UI | +| [0014](./0014-flexible-users-access-and-model-offerings.md) | **Flexible users & model offerings** (accepted); backends, credentials, slugs, roles/UI | +| [0015](./0015-host-owned-mcp-tool-connections.md) | **Host-owned MCP tools**; scoped runtimes, authorization, bounded discovery, progressive disclosure | Product glossary (no implementation detail): [`CONTEXT.md`](../../CONTEXT.md). @@ -31,4 +32,4 @@ Host identity / Team access plan: [`docs/plans/host-identity-and-teams.md`](../p Host path policy: [ADR 0012](./0012-host-path-grants-and-tool-enforcement.md). -Flexible users / model offerings (proposed): [ADR 0014](./0014-flexible-users-access-and-model-offerings.md), plan [`flexible-users-access-and-model-offerings.md`](../plans/flexible-users-access-and-model-offerings.md). +Flexible users / model offerings: [ADR 0014](./0014-flexible-users-access-and-model-offerings.md), plan [`flexible-users-access-and-model-offerings.md`](../plans/flexible-users-access-and-model-offerings.md). diff --git a/docs/architecture.md b/docs/architecture.md index a898ef13..2f079963 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -1,135 +1,91 @@ -# OpenGUI architecture notes - -Contributor map of the repo **as it exists today**. Product language and ownership: [`CONTEXT.md`](../CONTEXT.md). Accepted decisions: [`docs/adr/`](./adr/). - -## Layers (canonical) - -Four layers — same definitions as [`CONTEXT.md` → Architecture](../CONTEXT.md#architecture): - -| Layer | Owns | Does not own | -| -------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | ---------------------------------------------------------------------------------------------- | -| **OpenGUI Runtime** | Harness Adapters, normalized `HarnessEvent` stream, Harness Inventory, Agent sends on **Harness Scope** (`harnessId` + directory + harness session id). Session/transcript **truth** stays in the Harness. | Queued prompts, multi-Frontend transport, Workspaces, queue UI, `OpenGuiClient` | -| **OpenGUI Backend** | One embedded Runtime per process; HTTP/WebSocket/SSE and Desktop IPC transport; Queued prompts + Queue dispatch; Backend arbitration; Host-embedded Account and API-key authorization; Backend persistence (identity, queue, uploads cleanup). Delegates execution to Runtime. | Workspace connection state, sidebar Project membership, Pending prompts, presentation metadata | -| **OpenGUI Frontend** | Workspaces, Frontend Projects (saved paths), Pending prompts, queue UI, session presentation metadata, UI preferences (via **Frontend persistence**). Talks only to Backend via `OpenGuiClient`. | Harness SDK/CLI, session/transcript source of truth, shared queue storage | -| **Shell** | Bootstrap Frontend (Desktop / Web / Mobile): window chrome, file picker, sidecar lifecycle, static hosting. | Harness execution, session truth, queue dispatch | - -**SDK v1** is in-process only: [`@opengui/runtime`](../packages/runtime/README.md). Target surface: `OpenGUI.create`, `at(directory)`, `SessionHandle` (`send`, `onStream`, `waitUntilIdle`) per [ADR 0007](./adr/0007-runtime-sdk-minimal-surface.md) and [`runtime-sdk-minimal-surface.md`](./plans/runtime-sdk-minimal-surface.md). No queue API in the SDK — use Backend for shared queues ([ADR 0005](./adr/0005-opengui-runtime-backend-split-and-sdk.md)). - -**Harness** = coding-agent CLI/runtime (OpenCode, Claude Code, Codex, Pi). **Provider** = model/API vendor inside a Harness. Never call the OpenGUI server process an “agent backend” ([ADR 0001](./adr/0001-harness-terminology.md)). - -## Where code lives today - -Target layout is in [`plans/runtime-backend-sdk-split.md`](./plans/runtime-backend-sdk-split.md). Current mapping: - -| Layer | Package / entry | Main paths | -| ------------------- | -------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| Shared wire types | `@opengui/protocol` | `packages/protocol/src/` (`HarnessId`, `OpenGuiCapabilities`, `QueueMode`, `SelectedModel`) | -| Runtime | `@opengui/runtime` | `packages/runtime/src/` (`host.ts`, `harness-service.ts`, `harness-runtime.ts`, `open-gui.ts`) | -| Harness Adapters | `packages/runtime/src/adapters/` | `*-bridge.ts`, `harness-adapter-kit.ts` | -| Runtime descriptors | Shared with Frontend protocol | `src/agents/` (`backend.ts`, `cli-harness-factory.ts`, `protocol/`) | -| Backend | `@opengui/backend` + thin entry | `packages/backend/src/` (`createBackendHost`, `host/`, `routes/`, `transport/` — SSE, RPC, FS, static, product API); `server/web-server.ts` (~15 lines, `serve()` only); queue/control services in `server/services/*` until a later pass | -| Frontend | React app | `src/` (`App.tsx`, `components/`, `hooks/`, `features/`, `protocol/`) | -| Desktop Shell | Electron | `main.ts`, `preload.ts`, `main/backend-sidecar.ts` | - -**Rule:** UI and hooks call **Backend** APIs only, not bridge IPC. Bridges register inside the Backend process via Runtime ([ADR 0005](./adr/0005-opengui-runtime-backend-split-and-sdk.md)). - -**Repo map maintenance:** Any PR that moves `server/web-server.ts`, `packages/backend/**`, or Harness bridge modules under `packages/runtime/src/adapters/` must update this section (layer table and paths) in the same PR. CI guard: `pnpm run slop-check` (thin `web-server`, no `lib/harness-adapter-kit`). - -**Storage:** [ADR 0004](./adr/0004-storage-source-of-truth-boundaries.md) — Harness owns sessions/transcripts; Backend SQLite owns queues/uploads; Frontend persistence owns Workspaces/Projects/UI. - -**Session reads:** [ADR 0006](./adr/0006-harness-only-session-and-transcript-reads.md), plan [`session-read-slop-removal.md`](./plans/session-read-slop-removal.md), manual [`session-read-acceptance.md`](./manual/session-read-acceptance.md). - -**Desktop transport:** [ADR 0003](./adr/0003-persistent-desktop-backend-transport.md) — Local Workspace uses private IPC, not loopback HTTP. - -**Host identity / Team access:** [ADR 0011](./adr/0011-host-embedded-accounts-and-teams.md), [ADR 0012](./adr/0012-host-path-grants-and-tool-enforcement.md), [ADR 0013](./adr/0013-multi-user-host-access-model.md), plans [`host-identity-and-teams.md`](./plans/host-identity-and-teams.md) and [`multi-user-host-access.md`](./plans/multi-user-host-access.md). Remote Hosts provide Accounts, invites, Team management, revocable Host API keys, durable Actor attribution, and path grants. Multi-user access is **user-default** with explicit shares (paths, models, sessions)—not ambient Team roommate semantics. Desktop Local remains Account-free. Path enforcement is disabled by default; when enabled, member access is deny-by-default and restricted shell is unavailable. Shell is not a grant jail until a later sandbox ADR. - -**Flexible users & model offerings (proposed):** [ADR 0014](./adr/0014-flexible-users-access-and-model-offerings.md), plan [`flexible-users-access-and-model-offerings.md`](./plans/flexible-users-access-and-model-offerings.md). Target split: **Model backend** + **Provider credentials** + user-facing **Model offering** (slug such as “Company Model”); richer Host roles/capabilities and pluggable Host auth methods; UI for backends, offerings, and entitlements. Do not implement against ad-hoc Settings cards alone—follow the plan phases. - -**MCP tools:** [ADR 0015](./adr/0015-host-owned-mcp-tool-connections.md). MCP connections, credentials, transports, catalog budgeting, and actor/Session isolation are Host-owned in `packages/backend/src/mcp/`. The Harness consumes a protocol-neutral `AgentToolSource`; model adapters never import the MCP SDK. Restricted path-policy actors receive no MCP tools. - -**Host path policy:** [ADR 0012](./adr/0012-host-path-grants-and-tool-enforcement.md). `packages/backend/src/path-policy/` and the identity grant schema/routes are foundation only. Do not describe path grants as enforced until HTTP/RPC/SSE, uploads, Session visibility, and every Harness tool consume `IdentityService.effectivePathPolicy(actor)`. Remote multi-user Hosts are **share-only** for paths (no auto user homes); Desktop keeps device default directories. - -Implementation checklists: - -- [`plans/runtime-backend-sdk-split.md`](./plans/runtime-backend-sdk-split.md) — packages / SDK (Phases 1–3 largely done). -- [`plans/contributor-experience-and-slop-removal.md`](./plans/contributor-experience-and-slop-removal.md) — docs, session index, naming, registry, guardrails. -- [`plans/session-read-slop-removal.md`](./plans/session-read-slop-removal.md) — ADR 0006 detail. -- [`plans/host-identity-and-teams.md`](./plans/host-identity-and-teams.md) — Accounts, Team management, attribution, and optional path-grant enforcement (Phases 0–4 done). -- [`plans/multi-user-host-access.md`](./plans/multi-user-host-access.md) — registration modes, canInvite, session ACL, model planes (ADR 0013). -- [`plans/flexible-users-access-and-model-offerings.md`](./plans/flexible-users-access-and-model-offerings.md) — offerings/slugs, backend auth strategies, roles/UI (ADR 0014, proposed). - -Optional CI: `node scripts/slop-check.mjs`. - -## Harness architecture - -Harness-facing modules live in `src/agents/`: - -- `backend.ts` defines the normalized Harness interface and event shapes used by the app. -- `index.ts` defines supported Harness IDs and routing helpers. -- `cli-harness-factory.ts` contains shared local-CLI defaults and `createCliHarnessNormalizer()` for local CLI Harnesses. -- `claude-code.ts`, `codex.ts`, and `pi.ts` are small descriptors built from the CLI factory. -- `opencode.ts` declares OpenCode capabilities/workspace shape and delegates SDK event translation to `src/agents/protocol/opencode-map.ts`. -- `id-codec.ts` is the Harness ID-codec seam. `shared.ts` re-exports it and keeps session/message tagging helpers. - -When adding or changing a Harness, keep protocol-specific mapping out of UI code. Normalize native events into `HarnessEvent` as close to the adapter as possible, then let the rest of the app consume the normalized event stream. - -## Adding a Harness - -A **Harness Adapter** is a bridge (`setupXBridge`) plus registry metadata. Start with [`docs/harness-bridge-contract.md`](./harness-bridge-contract.md) and `node scripts/scaffold-harness.mjs `. - -1. [`src/agents/harness-registry.ts`](../src/agents/harness-registry.ts) + [`harness-ids.ts`](../src/agents/harness-ids.ts) — id, label, CLI command. -2. [`cli-harness-factory.ts`](../src/agents/cli-harness-factory.ts) — `HARNESS_BACKEND_META`, `normalizeEvent`. -3. [`harness-bridge-registrations.ts`](../packages/runtime/src/harness-bridge-registrations.ts) — register bridge in `BRIDGE_SETUP_BY_HARNESS_ID`. -4. New `*-bridge.ts` under `packages/runtime/src/adapters/`. -5. [`server/harness-inventory.ts`](../server/harness-inventory.ts) — uses registry CLI map. -6. [`session-identity.ts`](../src/lib/session-identity.ts) — parse legacy ids only; new ids via `composeFrontendSessionId`. - -Descriptors in `src/agents/.ts`: use `makeLocalCliCapabilities()` and `createCliHarnessNormalizer()` for tagged CLI streams; custom SDK events go in `src/agents/protocol/` with tests. Session IDs: `composeFrontendSessionId` / codecs in `src/agents/shared.ts`, not ad-hoc strings. - -## Frontend feature slices - -The current frontend is still centered on `src/App.tsx`, but several orchestration concerns have been moved into `src/features/`: - -- `features/app-shell/useAppKeyboardShortcuts.ts` owns app-level keyboard shortcut orchestration. -- `features/session/useActiveSessionQueue.ts` owns active-session queue UI handlers. -- `features/session/useChatSessionSurface.ts` derives the active chat surface state. -- `features/worktree/useActiveWorktreeMerge.ts` owns active worktree merge and pull-request actions. -- `features/local-intent/` owns **Local intent orchestration** (Pending prompt → Agent send, Queued prompt dispatch from PromptBox). `HarnessProvider` (`use-agent-impl-core.tsx`) wires React state and delegates `sendPrompt` / `sendCommand` / queue side effects through `useLocalIntentOrchestration`. -- `features/agent-bootstrap/` — workspace persistence load + post-ready project/server bootstrap. -- `features/agent-resources/` — `loadServerResources` / resource catalog dedupe (`useAgentResourceCatalog`). - -New UI orchestration should follow this direction: keep reusable visual pieces in `src/components/`, keep cross-component state orchestration in a named `src/features//` hook, and keep pure domain utilities in `src/lib/`. - -## Shared UI primitives - -Reusable UI building blocks live in `src/components/ui/`. Recent dialog work uses: - -- `DialogShell` for common dialog layout and footer/body structure. -- `DialogHeader` for consistent dialog titles, descriptions, and icons. -- `ButtonGroup`, `FormField`, and `ToggleSwitch` for repeated form/action patterns. - -Prefer these primitives before adding another one-off dialog header, footer, button group, or toggle implementation. - -## Provider icons - -Provider icons are resolved by `src/components/provider-icons/ProviderIcon.tsx` and `types.ts`. Vite expands the SVG asset manifest from `src/components/provider-icons/svgs/*.svg` with `import.meta.glob`, so adding an icon should only require dropping in a correctly named SVG unless new fallback or alias behavior is needed. - -## Commands +# OpenGUI architecture + +Contributor map of the repository as it exists for the 0.6 release line. Canonical product +language is in [`CONTEXT.md`](../CONTEXT.md); accepted decisions are indexed in +[`docs/adr/`](./adr/README.md). + +## Runtime shape + +```text +Desktop Shell ─┐ +Web Shell ─────┼─ OpenGUI Frontend ── authenticated Host API/events ── OpenGUI Host +Mobile Shell ──┘ ├─ identity + authorization + ├─ model/provider credentials + ├─ MCP connections + └─ first-party Harness + ├─ Session SQLite + ├─ model adapters + └─ built-in + MCP tools +``` -Vite+ (`vp`) is a dev dependency. After `pnpm install`, use **`pnpm vp …`** or **`pnpm run