diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml index f54c1db6..0450a578 100644 --- a/.github/workflows/publish.yml +++ b/.github/workflows/publish.yml @@ -17,7 +17,7 @@ jobs: with: bun-version: latest - - uses: actions/setup-node@v6 + - uses: actions/setup-node@249970729cb0ef3589644e2896645e5dc5ba9c38 # v6 with: node-version: 22 @@ -42,34 +42,30 @@ jobs: npm --version - name: Install dependencies - run: bun install + run: bun install --frozen-lockfile - - name: Build Packages + - name: Verify release version before building env: - AICTRL_VERSION: ${{ github.event.release.tag_name }} - run: | - export AICTRL_VERSION="${AICTRL_VERSION#v}" - # Expose stripped version to subsequent steps so the publish-if-new.sh - # catalog:/workspace: resolver (and the smoke test below) can see it - # without each step re-doing the v-prefix strip. #72. - echo "AICTRL_VERSION=$AICTRL_VERSION" >> "$GITHUB_ENV" - bun turbo build - - - name: Verify the release tag matches the @aictrl/cli package version - # publish-if-new.sh publishes @aictrl/cli with the version in - # packages/cli/package.json (only the platform binaries are stamped - # with the tag). If the two disagree, the platform binaries publish - # under the tag while @aictrl/cli silently re-tries an already - # published version and the release half-ships (v0.4.4, first run). - # Fail before publishing anything. + RELEASE_TAG: ${{ github.event.release.tag_name }} run: | + set -euo pipefail + AICTRL_VERSION="${RELEASE_TAG#v}" + if [ -z "$AICTRL_VERSION" ]; then + echo "::error::AICTRL_VERSION is empty. This workflow must run from a published release tag." + exit 1 + fi PKG_VERSION=$(node -p "require('./packages/cli/package.json').version") if [ "$PKG_VERSION" != "$AICTRL_VERSION" ]; then - echo "::error::Release tag version $AICTRL_VERSION does not match packages/cli/package.json version $PKG_VERSION. Bump package.json on main, then re-create the release." + echo "::error::Release tag version $AICTRL_VERSION does not match packages/cli/package.json version $PKG_VERSION. Bump package.json on main before creating the release." exit 1 fi + # Platform binaries, wrapper and smoke test must use one version. + echo "AICTRL_VERSION=$AICTRL_VERSION" >> "$GITHUB_ENV" echo "Release version $AICTRL_VERSION matches packages/cli/package.json" + - name: Build Packages + run: bun turbo build + - name: Publish @aictrl/util working-directory: packages/util run: ${{ github.workspace }}/.github/scripts/publish-if-new.sh "@aictrl/util" @@ -173,24 +169,48 @@ jobs: working-directory: packages/cli run: ${{ github.workspace }}/.github/scripts/publish-if-new.sh "@aictrl/cli" + smoke: + needs: publish + runs-on: ubuntu-latest + # Registry install and binary execution receive no publication permissions. + permissions: {} + steps: + - uses: actions/setup-node@249970729cb0ef3589644e2896645e5dc5ba9c38 # v6 + with: + node-version: 22 + - name: Smoke test - verify @aictrl/cli installs cleanly via npm # Catches the v0.3.3-class bug where the published manifest carries # Bun-only protocols (workspace:*, catalog:) that plain `npm install` # rejects with EUNSUPPORTEDPROTOCOL. Without this gate, a broken # publish goes unnoticed until consumers report install failures. # #72. + timeout-minutes: 15 + env: + RELEASE_TAG: ${{ github.event.release.tag_name }} run: | set -euo pipefail + AICTRL_VERSION="${RELEASE_TAG#v}" smoke="${RUNNER_TEMP}/aictrl-publish-smoke" rm -rf "$smoke" mkdir -p "$smoke" cd "$smoke" npm init -y > /dev/null - # CDN propagation can lag ~30-60s after publish; retry briefly. - for attempt in 1 2 3 4 5; do + # v0.4.4 took longer than 2.5 minutes to become installable (#126). + # Ten attempts allow 8.5 minutes of propagation backoff, bounded by + # the step timeout even if npm itself stalls. + for attempt in {1..10}; do if npm install "@aictrl/cli@${AICTRL_VERSION}" 2>&1 | tee install.log; then echo "::notice::npm install @aictrl/cli@${AICTRL_VERSION} succeeded on attempt ${attempt}" + if ! INSTALLED_VERSION=$(./node_modules/.bin/aictrl --version); then + echo "::error::Installed CLI binary could not run for @aictrl/cli@${AICTRL_VERSION}" + exit 1 + fi + if [ "$INSTALLED_VERSION" != "$AICTRL_VERSION" ]; then + echo "::error::Installed CLI reports $INSTALLED_VERSION, expected $AICTRL_VERSION" + exit 1 + fi exit 0 fi # EUNSUPPORTEDPROTOCOL means the resolver in publish-if-new.sh @@ -199,8 +219,11 @@ jobs: echo "::error::@aictrl/cli@${AICTRL_VERSION} ships protocol-prefixed deps that plain npm cannot resolve. The catalog:/workspace: resolver in publish-if-new.sh did not run or did not cover all fields. See aictrl-dev/cli#72." exit 1 fi - echo "::warning::install attempt ${attempt} failed (likely CDN propagation), retrying in 30s..." - sleep 30 + if [ "$attempt" -eq 10 ]; then break; fi + delay=60 + if [ "$attempt" -eq 1 ]; then delay=30; fi + echo "::warning::install attempt ${attempt} failed, retrying in ${delay}s..." + sleep "$delay" done - echo "::error::npm install @aictrl/cli@${AICTRL_VERSION} failed after 5 attempts" + echo "::error::npm install @aictrl/cli@${AICTRL_VERSION} failed after 10 attempts" exit 1 diff --git a/RELEASING.md b/RELEASING.md new file mode 100644 index 00000000..b28be85f --- /dev/null +++ b/RELEASING.md @@ -0,0 +1,53 @@ +# Releasing aictrl CLI + +1. Prepare a PR against `main` with the fixes, the version bump in + `packages/cli/package.json`, and the matching `packages/cli` workspace + version in `bun.lock`. Run `bun install --frozen-lockfile`. +2. Run the release regressions from `packages/cli`: + + ```bash + bun test test/cli/run-mcp-discovery.test.ts test/mcp/discovery-recovery.test.ts test/cli/publish-workflow.test.ts test/cli/run-provider-finish.test.ts test/cli/run-signal-cancellation.test.ts test/cli/classify-session-error.test.ts test/session/idle.test.ts test/session/processor-idle.test.ts + ``` + + Wait for CI build, workspace typecheck and tests to pass and address reviews. + +3. Merge the approved PR and record the resulting `main` commit SHA. Confirm + its package version matches the intended tag. Publish the release against + that exact commit, using reviewed notes: + + ```bash + gh release create v --repo aictrl-dev/cli --target --title v --notes-file + gh run list --repo aictrl-dev/cli --workflow publish.yml --limit 3 + ``` + + Do not use the legacy `script/release` or `script/publish.ts`; they refer to + retired branches/workflows and do not implement these gates. + +4. Require the **Publish to npm** workflow to be green. It checks the tag + against the wrapper package version before building, publishes the platform + packages and wrapper, then retries npm installation through its propagation + window and verifies the installed binary version. If publishing partially + succeeds, inspect the failure before rerunning: npm versions are immutable. +5. Independently install `@aictrl/cli@` in a temporary directory and + verify `./node_modules/.bin/aictrl --version` equals ``. +6. In a separate aictrl application PR, update `docker/executor/Dockerfile` to + the published version. Build the executor image and rerun the headless + discovery and idle-stream fixtures against its binary. Promote through + `sandbox` before `main`; verify a sandbox review persists MCP findings and + a stalled stream reports a timeout rather than waiting for the job limit. + +## 0.4.5 candidate + +- MCP tools are discovered once per connection and retained for both the + startup catalog and model turns (CLI #127). Tool-list change notifications + refresh that catalog; a failed refresh or closed connection stops the next + turn with an actionable error, instead of submitting a reduced toolset. + A successful later notification or reconnection restores discovery health. + Initially unavailable optional MCP servers keep their existing behavior. +- Release version checks and npm propagation handling address CLI #126. +- The executor enables the existing five-minute model stream idle guard in + application #5891. The CLI retains its generic opt-in default; local tools + retain the extended allowance and explicit operator overrides remain valid. + +Publication and executor promotion are separate gates. Preparing this candidate +PR does not publish npm packages or deploy an executor image. diff --git a/bun.lock b/bun.lock index 1123e092..1a07967f 100644 --- a/bun.lock +++ b/bun.lock @@ -25,7 +25,7 @@ }, "packages/cli": { "name": "@aictrl/cli", - "version": "0.3.3", + "version": "0.4.5", "bin": { "aictrl": "./bin/aictrl", }, @@ -210,7 +210,7 @@ }, "packages/util": { "name": "@aictrl/util", - "version": "1.2.16", + "version": "1.2.17", "dependencies": { "zod": "catalog:", }, diff --git a/packages/cli/package.json b/packages/cli/package.json index 0b39a785..ccf8d2c7 100644 --- a/packages/cli/package.json +++ b/packages/cli/package.json @@ -1,6 +1,6 @@ { "$schema": "https://json.schemastore.org/package.json", - "version": "0.4.4", + "version": "0.4.5", "name": "@aictrl/cli", "description": "Headless execution engine for AI agent skills", "type": "module", diff --git a/packages/cli/src/mcp/index.ts b/packages/cli/src/mcp/index.ts index e84863d7..9a2b04e3 100644 --- a/packages/cli/src/mcp/index.ts +++ b/packages/cli/src/mcp/index.ts @@ -62,6 +62,15 @@ export namespace MCP { type MCPClient = Client + type ToolCatalog = { + tools: MCPToolDef[] + refresh: Promise + error?: Error + } + // Definitions belong to the connection that discovered them. Replacing a + // server cannot publish an old client's definitions into its new catalog. + const toolCatalogs = new WeakMap() + export const Status = z .discriminatedUnion("status", [ z @@ -108,11 +117,56 @@ export namespace MCP { export type Status = z.infer // Register notification handlers for MCP client - function registerNotificationHandlers(client: MCPClient, serverName: string) { + function registerNotificationHandlers(client: MCPClient, serverName: string, timeout?: number) { client.setNotificationHandler(ToolListChangedNotificationSchema, async () => { + if (!toolCatalogs.has(client)) return // Disposed connection. log.info("tools list changed notification received", { server: serverName }) - Bus.publish(ToolsChanged, { server: serverName }) + await refreshCatalog(client, serverName, timeout) + await Bus.publish(ToolsChanged, { server: serverName }) + }) + client.onclose = () => { + const catalog = toolCatalogs.get(client) + if (!catalog) return // Explicit disposal removed the catalog first. + catalog.error = new Error(`MCP connection closed for server "${serverName}". Reconnect the MCP server and retry.`) + log.error("MCP connection closed", { serverName }) + } + } + + function refreshCatalog(client: MCPClient, serverName: string, timeout?: number) { + const catalog = toolCatalogs.get(client) + if (!catalog) throw new Error(`Missing MCP tool catalog for server "${serverName}"`) + // Serialize notifications so an older response cannot overwrite a newer + // catalog. Readers await the pending refresh before using its definitions. + catalog.refresh = catalog.refresh.then(async () => { + try { + const result = await client.listTools(undefined, { timeout }) + if (toolCatalogs.get(client) !== catalog) return + if (!client.transport) throw new Error("Connection closed during tool discovery") + catalog.tools = result.tools + catalog.error = undefined + } catch (error) { + if (toolCatalogs.get(client) !== catalog) return + catalog.error = new Error( + `MCP tool discovery failed for server "${serverName}". Reconnect the MCP server and retry.`, + { cause: error }, + ) + log.error("MCP tool discovery failed", { + serverName, + error: error instanceof Error ? error.message : String(error), + }) + } }) + return catalog.refresh + } + + async function catalogTools(client: MCPClient, serverName: string) { + const catalog = toolCatalogs.get(client) + if (!catalog) throw new Error(`Missing MCP tool catalog for server "${serverName}"`) + await catalog.refresh + if (toolCatalogs.get(client) !== catalog) + throw new Error(`MCP connection changed for server "${serverName}". Retry the invocation.`) + if (catalog.error) throw catalog.error + return catalog.tools } // Convert MCP tool definition to AI SDK Tool type @@ -223,6 +277,7 @@ export namespace MCP { // Kill the full descendant tree first so the server exits promptly // and no processes are left behind. for (const client of Object.values(state.clients)) { + toolCatalogs.delete(client) const pid = (client.transport as any)?.pid if (typeof pid !== "number") continue for (const dpid of await descendants(pid)) { @@ -292,6 +347,7 @@ export namespace MCP { export async function add(name: string, mcp: Config.Mcp) { const s = await state() + await disconnect(name) const result = await create(name, mcp) if (!result) { const status = { @@ -309,13 +365,6 @@ export namespace MCP { status: s.status, } } - // Close existing client if present to prevent memory leaks - const existingClient = s.clients[name] - if (existingClient) { - await existingClient.close().catch((error) => { - log.error("Failed to close existing MCP client", { name, error }) - }) - } s.clients[name] = result.mcpClient s.status[name] = result.status @@ -387,7 +436,6 @@ export namespace MCP { version: Installation.VERSION, }) await withTimeout(client.connect(transport), connectTimeout) - registerNotificationHandlers(client, key) mcpClient = client log.info("connected", { key, transport: name }) status = { status: "connected" } @@ -456,7 +504,6 @@ export namespace MCP { version: Installation.VERSION, }) await withTimeout(client.connect(transport), connectTimeout) - registerNotificationHandlers(client, key) mcpClient = client status = { status: "connected", @@ -500,6 +547,7 @@ export namespace MCP { } } + const cfg = await Config.get() const result = await withTimeout(mcpClient.listTools(), mcp.timeout ?? DEFAULT_TIMEOUT).catch((err) => { log.error("failed to get tools from client", { key, error: err }) return undefined @@ -524,12 +572,19 @@ export namespace MCP { } log.info("create() successfully created client", { key, toolCount: result.tools.length }) + toolCatalogs.set(mcpClient, { tools: result.tools, refresh: Promise.resolve() }) + registerNotificationHandlers(mcpClient, key, mcp.timeout ?? cfg.experimental?.mcp_timeout) return { mcpClient, status, } } + function clientStatus(client: MCPClient | undefined, current: Status | undefined): Status { + const error = client && toolCatalogs.get(client)?.error + return error ? { status: "failed", error: error.message } : (current ?? { status: "disabled" }) + } + export async function status() { const s = await state() const cfg = await Config.get() @@ -539,7 +594,7 @@ export namespace MCP { // Include all configured MCPs from config, not just connected ones for (const [key, mcp] of Object.entries(config)) { if (!isMcpConfigured(mcp)) continue - result[key] = s.status[key] ?? { status: "disabled" } + result[key] = clientStatus(s.clients[key], s.status[key]) } return result @@ -563,39 +618,18 @@ export namespace MCP { return } - const result = await create(name, { ...mcp, enabled: true }) - - if (!result) { - const s = await state() - s.status[name] = { - status: "failed", - error: "Unknown error during connection", - } - return - } - - const s = await state() - s.status[name] = result.status - if (result.mcpClient) { - // Close existing client if present to prevent memory leaks - const existingClient = s.clients[name] - if (existingClient) { - await existingClient.close().catch((error) => { - log.error("Failed to close existing MCP client", { name, error }) - }) - } - s.clients[name] = result.mcpClient - } + await add(name, { ...mcp, enabled: true }) } export async function disconnect(name: string) { const s = await state() const client = s.clients[name] if (client) { + delete s.clients[name] + toolCatalogs.delete(client) await client.close().catch((error) => { log.error("Failed to close MCP client", { name, error }) }) - delete s.clients[name] } s.status[name] = { status: "disabled" } } @@ -611,40 +645,26 @@ export namespace MCP { return sanitizedClientName + "_" + sanitizedToolName } + function configuredTimeout(cfg: Config.Info, clientName: string) { + const entry = cfg.mcp?.[clientName] + return (entry && isMcpConfigured(entry) ? entry.timeout : undefined) ?? cfg.experimental?.mcp_timeout + } + export async function tools() { const result: Record = {} const s = await state() const cfg = await Config.get() - const config = cfg.mcp ?? {} - const clientsSnapshot = await clients() - const defaultTimeout = cfg.experimental?.mcp_timeout - - const connectedClients = Object.entries(clientsSnapshot).filter( - ([clientName]) => s.status[clientName]?.status === "connected", - ) - - const toolsResults = await Promise.all( - connectedClients.map(async ([clientName, client]) => { - const toolsResult = await client.listTools().catch((e) => { - log.error("failed to get tools", { clientName, error: e.message }) - const failedStatus = { - status: "failed" as const, - error: e instanceof Error ? e.message : String(e), - } - s.status[clientName] = failedStatus - delete s.clients[clientName] - return undefined - }) - return { clientName, client, toolsResult } - }), + const catalogs = await Promise.all( + Object.entries(s.clients).map(async ([clientName, client]) => ({ + clientName, + client, + tools: await catalogTools(client, clientName), + })), ) - for (const { clientName, client, toolsResult } of toolsResults) { - if (!toolsResult) continue - const mcpConfig = config[clientName] - const entry = isMcpConfigured(mcpConfig) ? mcpConfig : undefined - const timeout = entry?.timeout ?? defaultTimeout - for (const mcpTool of toolsResult.tools) { + for (const { clientName, client, tools } of catalogs) { + const timeout = configuredTimeout(cfg, clientName) + for (const mcpTool of tools) { result[mcpToolKey(clientName, mcpTool.name)] = await convertMcpTool(mcpTool, client, timeout) } } @@ -660,35 +680,16 @@ export namespace MCP { * guaranteed identical to what `resolveTools` dispatches to the model. */ export async function toolEntries(): Promise<{ toolKey: string; serverName: string }[]> { - const result: { toolKey: string; serverName: string }[] = [] const s = await state() - const clientsSnapshot = await clients() - - const connectedClients = Object.entries(clientsSnapshot).filter( - ([clientName]) => s.status[clientName]?.status === "connected", + const catalogs = await Promise.all( + Object.entries(s.clients).map(async ([serverName, client]) => ({ + serverName, + tools: await catalogTools(client, serverName), + })), ) - - const toolsResults = await Promise.all( - connectedClients.map(async ([clientName, client]) => { - const toolsResult = await client.listTools().catch((e) => { - log.error("failed to get tool entries", { clientName, error: e.message }) - return undefined - }) - return { clientName, toolsResult } - }), + return catalogs.flatMap(({ serverName, tools }) => + tools.map((tool) => ({ toolKey: mcpToolKey(serverName, tool.name), serverName })), ) - - for (const { clientName, toolsResult } of toolsResults) { - if (!toolsResult) continue - for (const mcpTool of toolsResult.tools) { - result.push({ - toolKey: mcpToolKey(clientName, mcpTool.name), - serverName: clientName, - }) - } - } - - return result } export async function prompts() { @@ -699,7 +700,7 @@ export namespace MCP { ( await Promise.all( Object.entries(clientsSnapshot).map(async ([clientName, client]) => { - if (s.status[clientName]?.status !== "connected") { + if (clientStatus(client, s.status[clientName]).status !== "connected") { return [] } @@ -720,7 +721,7 @@ export namespace MCP { ( await Promise.all( Object.entries(clientsSnapshot).map(async ([clientName, client]) => { - if (s.status[clientName]?.status !== "connected") { + if (clientStatus(client, s.status[clientName]).status !== "connected") { return [] } diff --git a/packages/cli/test/cli/publish-workflow.test.ts b/packages/cli/test/cli/publish-workflow.test.ts new file mode 100644 index 00000000..a9d4b831 --- /dev/null +++ b/packages/cli/test/cli/publish-workflow.test.ts @@ -0,0 +1,120 @@ +import { describe, expect, test } from "bun:test" +import fs from "fs/promises" +import path from "path" +import { tmpdir } from "../fixture/fixture" + +const workflow = Bun.YAML.parse( + await Bun.file(path.resolve(import.meta.dir, "../../../../.github/workflows/publish.yml")).text(), +) as { + jobs: Record< + string, + { steps: { name?: string; run?: string }[]; needs?: string; permissions?: Record } + > +} +const steps = workflow.jobs.publish.steps +const guard = steps.find((step) => step.name === "Verify release version before building")! +const smoke = workflow.jobs.smoke.steps.find( + (step) => step.name === "Smoke test - verify @aictrl/cli installs cleanly via npm", +)! + +describe("release version gate", () => { + test.each([ + { tag: "v0.4.5", code: 0, message: "matches packages/cli/package.json" }, + { tag: "v0.4.4", code: 1, message: "does not match" }, + { tag: "", code: 1, message: "AICTRL_VERSION is empty" }, + ])("tag '$tag'", async ({ tag, code, message }) => { + await using tmp = await tmpdir() + await fs.mkdir(path.join(tmp.path, "packages/cli"), { recursive: true }) + await Bun.write(path.join(tmp.path, "packages/cli/package.json"), JSON.stringify({ version: "0.4.5" })) + const env = path.join(tmp.path, "github-env") + const proc = Bun.spawn(["bash", "-c", guard.run!], { + cwd: tmp.path, + env: { PATH: process.env.PATH, RELEASE_TAG: tag, GITHUB_ENV: env }, + stdout: "pipe", + stderr: "pipe", + }) + const [stdout, exit] = await Promise.all([new Response(proc.stdout).text(), proc.exited]) + expect(exit, stdout).toBe(code) + expect(stdout).toContain(message) + expect(steps.indexOf(guard)).toBeLessThan(steps.findIndex((step) => step.name === "Build Packages")) + if (!code) expect(await Bun.file(env).text()).toBe("AICTRL_VERSION=0.4.5\n") + else expect(await Bun.file(env).exists()).toBe(false) + }) +}) + +describe("release install smoke gate", () => { + test("published package execution has no publishing permissions", () => { + expect(workflow.jobs.smoke.needs).toBe("publish") + expect(workflow.jobs.smoke.permissions).toEqual({}) + }) + test.each([ + { name: "immediate availability", failures: 0, error: "ETARGET", version: "0.4.5", code: 0, attempts: 1 }, + { name: "delayed beyond old window", failures: 6, error: "ETARGET", version: "0.4.5", code: 0, attempts: 7 }, + { name: "propagation window expires", failures: 10, error: "ETARGET", version: "0.4.5", code: 1, attempts: 10 }, + { + name: "invalid package protocols", + failures: 10, + error: "EUNSUPPORTEDPROTOCOL", + version: "0.4.5", + code: 1, + attempts: 1, + }, + { name: "wrong binary installed", failures: 0, error: "ETARGET", version: "0.4.4", code: 1, attempts: 1 }, + { name: "missing binary", failures: 0, error: "ETARGET", version: "missing", code: 1, attempts: 1 }, + ])("$name", async ({ failures, error, version, code, attempts }) => { + await using tmp = await tmpdir() + const bin = path.join(tmp.path, "bin") + await fs.mkdir(bin) + await Bun.write( + path.join(bin, "npm"), + `#!/bin/bash +set -euo pipefail +if [ "$1" = init ]; then exit 0; fi +count=0 +if [ -f "$FIXTURE_ROOT/count" ]; then count=$(cat "$FIXTURE_ROOT/count"); fi +count=$((count + 1)) +echo "$count" > "$FIXTURE_ROOT/count" +if [ "$count" -le "$FIXTURE_FAILURES" ]; then echo "$FIXTURE_ERROR"; exit 1; fi +mkdir -p node_modules/.bin +if [ "$FIXTURE_VERSION" = missing ]; then exit 0; fi +cat > node_modules/.bin/aictrl <<'BIN' +#!/bin/sh +printf '%s' "$FIXTURE_VERSION" +BIN +chmod +x node_modules/.bin/aictrl +`, + ) + await Bun.write( + path.join(bin, "sleep"), + `#!/bin/sh +echo "$1" >> "$FIXTURE_ROOT/sleeps" +`, + ) + await Promise.all([fs.chmod(path.join(bin, "npm"), 0o755), fs.chmod(path.join(bin, "sleep"), 0o755)]) + const proc = Bun.spawn(["bash", "-c", smoke.run!], { + cwd: tmp.path, + env: { + PATH: `${bin}:${process.env.PATH}`, + RUNNER_TEMP: tmp.path, + RELEASE_TAG: "v0.4.5", + FIXTURE_ROOT: tmp.path, + FIXTURE_FAILURES: String(failures), + FIXTURE_ERROR: error, + FIXTURE_VERSION: version, + }, + stdout: "pipe", + stderr: "pipe", + }) + const [stdout, exit] = await Promise.all([new Response(proc.stdout).text(), proc.exited]) + expect(exit, stdout).toBe(code) + if (version === "missing") expect(stdout).toContain("::error::Installed CLI binary could not run") + expect(await Bun.file(path.join(tmp.path, "count")).text()).toBe(`${attempts}\n`) + const sleeps = Bun.file(path.join(tmp.path, "sleeps")) + if (attempts === 1) expect(await sleeps.exists()).toBe(false) + else { + const delays = (await sleeps.text()).trim().split("\n").map(Number) + expect(delays).toEqual([30, ...Array(attempts - 2).fill(60)]) + if (attempts === 10) expect(delays.reduce((sum, value) => sum + value, 0)).toBe(510) + } + }) +}) diff --git a/packages/cli/test/cli/run-mcp-discovery.test.ts b/packages/cli/test/cli/run-mcp-discovery.test.ts new file mode 100644 index 00000000..ae3e92b0 --- /dev/null +++ b/packages/cli/test/cli/run-mcp-discovery.test.ts @@ -0,0 +1,242 @@ +import { describe, expect, test } from "bun:test" +import path from "path" +import { tmpdir } from "../fixture/fixture" + +const entry = path.resolve(import.meta.dir, "../../src/index.ts") +const command = process.env.AICTRL_TEST_BINARY + ? [process.env.AICTRL_TEST_BINARY] + : ["bun", "run", "--conditions=browser", entry] + +type Event = { + type: string + status?: string + message?: string + tools?: { name: string; source: string }[] +} + +describe("headless MCP discovery", () => { + test.each([ + { + name: "healthy catalog is discovered once across model turns", + notify: false, + failure: false, + hang: false, + turns: 3, + code: 0, + }, + { name: "unrequested rediscovery would fail", notify: false, failure: true, hang: false, turns: 3, code: 0 }, + { name: "unrequested rediscovery would hang", notify: false, failure: true, hang: true, turns: 3, code: 0 }, + { + name: "notified catalog update reaches next model turn", + notify: true, + failure: false, + hang: false, + turns: 3, + code: 0, + }, + { + name: "notified discovery failure stops next model turn", + notify: true, + failure: true, + hang: false, + turns: 1, + code: 1, + }, + { + name: "hung notified discovery stops next model turn", + notify: true, + failure: true, + hang: true, + turns: 1, + code: 1, + }, + ])( + "$name", + async ({ notify, failure, hang, turns, code }) => { + await using tmp = await tmpdir() + let lists = 0 + let records = 0 + const requests: string[][] = [] + const descriptions: string[] = [] + const server = Bun.serve({ + hostname: "127.0.0.1", + port: 0, + idleTimeout: 0, + async fetch(req): Promise { + if (new URL(req.url).pathname === "/mcp") { + if (req.method !== "POST") return new Response(null, { status: 405 }) + const body = (await req.json()) as { id?: number; method: string; params: { protocolVersion: string } } + if (body.method === "notifications/initialized") return new Response(null, { status: 202 }) + if (body.method === "initialize") { + return Response.json({ + jsonrpc: "2.0", + id: body.id, + result: { + protocolVersion: body.params.protocolVersion, + capabilities: { tools: { listChanged: true } }, + serverInfo: { name: "fixture", version: "1" }, + }, + }) + } + if (body.method === "tools/list") { + lists++ + if (failure && lists > 1) { + if (hang) + return new Response(new ReadableStream(), { headers: { "content-type": "text/event-stream" } }) + return Response.json({ + jsonrpc: "2.0", + id: body.id, + error: { code: -32603, message: "injected one-time discovery failure" }, + }) + } + return Response.json({ + jsonrpc: "2.0", + id: body.id, + result: { + tools: [ + { + name: "record_finding", + description: lists === 1 ? "Initial catalog" : "Updated catalog", + inputSchema: { type: "object" }, + }, + ], + }, + }) + } + if (body.method === "tools/call") { + records++ + if (notify) { + const events = [ + { jsonrpc: "2.0", method: "notifications/tools/list_changed" }, + { jsonrpc: "2.0", id: body.id, result: { content: [{ type: "text", text: "recorded" }] } }, + ] + return new Response( + events.map((event) => `event: message\ndata: ${JSON.stringify(event)}\n\n`).join(""), + { + headers: { "content-type": "text/event-stream" }, + }, + ) + } + } + return Response.json({ + jsonrpc: "2.0", + id: body.id, + result: { content: [{ type: "text", text: "recorded" }] }, + }) + } + const body = (await req.json()) as { tools?: { function: { name: string; description?: string } }[] } + const tools = body.tools?.map((tool) => tool.function.name) ?? [] + if (tools.length) { + requests.push(tools) + descriptions.push( + body.tools!.find((tool) => tool.function.name === "aictrl_record_finding")?.function.description ?? + "missing", + ) + } + const name = requests.length === 1 ? "aictrl_record_finding" : "bash" + const call = tools.length && requests.length < 3 + const delta = call + ? { + tool_calls: [ + { + index: 0, + id: `fixture_${requests.length}`, + type: "function", + function: { + name, + arguments: JSON.stringify( + name === "bash" ? { command: "true", description: "fixture no-op" } : {}, + ), + }, + }, + ], + } + : { content: "Fixture complete" } + const chunks = [ + { delta: { role: "assistant", ...delta }, finish_reason: null }, + { delta: {}, finish_reason: call ? "tool_calls" : "stop" }, + ].map( + (choice) => + `data: ${JSON.stringify({ id: "fixture", object: "chat.completion.chunk", created: 1, model: "fixture", choices: [{ index: 0, ...choice }] })}\n\n`, + ) + return new Response(chunks.join("") + "data: [DONE]\n\n", { + headers: { "content-type": "text/event-stream" }, + }) + }, + }) + const config = { + provider: { + fixture: { + npm: "@ai-sdk/openai-compatible", + options: { apiKey: "fixture", baseURL: `http://127.0.0.1:${server.port}/v1` }, + models: { fixture: { name: "fixture", tool_call: true, limit: { context: 100000, output: 1000 } } }, + }, + }, + agent: { title: { disable: true } }, + mcp: { aictrl: { type: "remote", url: `http://127.0.0.1:${server.port}/mcp`, oauth: false, timeout: 1000 } }, + permission: { "*": "allow" }, + } + const proc = Bun.spawn( + [...command, "run", "--format", "json", "--model", "fixture/fixture", "Review this input."], + { + cwd: tmp.path, + env: { + ...process.env, + AICTRL_CONFIG_CONTENT: JSON.stringify(config), + AICTRL_DISABLE_PROJECT_CONFIG: "true", + AICTRL_DISABLE_DEFAULT_PLUGINS: "true", + AICTRL_DISABLE_MODELS_FETCH: "true", + AICTRL_DISABLE_AUTOCOMPACT: "true", + AICTRL_TEST_HOME: tmp.path, + AICTRL_TEST_MANAGED_CONFIG_DIR: path.join(tmp.path, "managed"), + XDG_DATA_HOME: path.join(tmp.path, "data"), + XDG_CONFIG_HOME: path.join(tmp.path, "config"), + XDG_STATE_HOME: path.join(tmp.path, "state"), + XDG_CACHE_HOME: path.join(tmp.path, "cache"), + AICTRL_MODEL_STREAM_IDLE_TIMEOUT_MS: "5000", + }, + stdout: "pipe", + stderr: "pipe", + }, + ) + const timeout = setTimeout(() => proc.kill("SIGKILL"), 15000) + try { + const [stdout, stderr, exit] = await Promise.all([ + new Response(proc.stdout).text(), + new Response(proc.stderr).text(), + proc.exited, + ]) + const events: Event[] = stdout + .split("\n") + .filter((line) => line.startsWith("{")) + .map((line) => JSON.parse(line)) + expect(exit, stderr + stdout).toBe(code) + expect(requests).toHaveLength(turns) + for (const tools of requests) expect(tools).toContain("aictrl_record_finding") + expect(events.find((event) => event.type === "tool_catalog")?.tools).toContainEqual( + expect.objectContaining({ name: "aictrl_record_finding", source: "mcp" }), + ) + expect(lists).toBe(notify ? 2 : 1) + expect(records).toBe(1) + expect(descriptions).toEqual( + notify && !code + ? ["Initial catalog", "Updated catalog", "Updated catalog"] + : Array(turns).fill("Initial catalog"), + ) + expect(events.filter((event) => event.type === "session_error")).toHaveLength(code ? 1 : 0) + if (code) { + expect(events.find((event) => event.type === "session_error")?.message).toContain( + 'MCP tool discovery failed for server "aictrl"', + ) + } + expect(events.filter((event) => event.type === "invocation_complete")).toHaveLength(1) + expect(events.find((event) => event.type === "invocation_complete")?.status).toBe(code ? "error" : "completed") + } finally { + clearTimeout(timeout) + proc.kill("SIGKILL") + server.stop(true) + } + }, + 20000, + ) +}) diff --git a/packages/cli/test/mcp/discovery-recovery.test.ts b/packages/cli/test/mcp/discovery-recovery.test.ts new file mode 100644 index 00000000..b2de0d88 --- /dev/null +++ b/packages/cli/test/mcp/discovery-recovery.test.ts @@ -0,0 +1,290 @@ +import { expect, test } from "bun:test" +import path from "path" +import { tmpdir } from "../fixture/fixture" + +// Real SDK clients run in children so unrelated transport mocks cannot leak in. +async function run(source: string, initial = "initial") { + await using tmp = await tmpdir() + const lists: Record = {} + const rpc: Record = {} + const modes: Record = { "/mcp": initial, "/replacement": "replacement" } + const streams = new Map>() + const closed: Record = {} + const releases: (() => void)[] = [] + let active = 0 + let maxActive = 0 + const server = Bun.serve({ + hostname: "127.0.0.1", + port: 0, + idleTimeout: 0, + async fetch(req): Promise { + const route = new URL(req.url).pathname + if (route === "/inspect") return Response.json({ lists, rpc, closed, maxActive }) + if (route === "/control") { + const body = (await req.json()) as { mode?: string; route?: string; count?: number; release?: boolean } + if (body.release) releases.splice(0).forEach((resolve) => resolve()) + if (body.mode) { + const target = body.route ?? "/mcp" + modes[target] = body.mode + const stream = streams.get(target) + if (!stream) throw new Error(`No notification stream for ${target}`) + for (let i = 0; i < (body.count ?? 1); i++) { + stream.enqueue( + new TextEncoder().encode( + 'event: message\ndata: {"jsonrpc":"2.0","method":"notifications/tools/list_changed"}\n\n', + ), + ) + } + } + return Response.json({ ok: true }) + } + if (route === "/failed") return new Response(null, { status: 503 }) + if (req.method === "GET") { + return new Response( + new ReadableStream({ + start(controller) { + streams.set(route, controller) + }, + cancel() { + streams.delete(route) + closed[route] = (closed[route] ?? 0) + 1 + }, + }), + { headers: { "content-type": "text/event-stream" } }, + ) + } + if (req.method !== "POST") return new Response(null, { status: 405 }) + const body = (await req.json()) as { id?: number; method: string; params: { protocolVersion: string } } + if (body.method === "notifications/initialized") return new Response(null, { status: 202 }) + if (body.method === "initialize") { + return Response.json({ + jsonrpc: "2.0", + id: body.id, + result: { + protocolVersion: body.params.protocolVersion, + capabilities: { tools: { listChanged: true }, prompts: {}, resources: {} }, + serverInfo: { name: "fixture", version: "1" }, + }, + }) + } + if (body.method === "prompts/list" || body.method === "resources/list") { + rpc[body.method] = (rpc[body.method] ?? 0) + 1 + return Response.json({ + jsonrpc: "2.0", + id: body.id, + result: + body.method === "prompts/list" + ? { prompts: [{ name: "review", description: "Review prompt" }] } + : { resources: [{ name: "evidence", uri: "fixture://evidence", description: "Review evidence" }] }, + }) + } + if (body.method === "tools/call") { + return Response.json({ jsonrpc: "2.0", id: body.id, result: { content: [{ type: "text", text: route }] } }) + } + if (body.method !== "tools/list") return new Response(null, { status: 400 }) + lists[route] = (lists[route] ?? 0) + 1 + const mode = modes[route] + active++ + maxActive = Math.max(maxActive, active) + try { + if (mode === "hold") await new Promise((resolve) => releases.push(resolve)) + if (mode === "updated") await Bun.sleep(50) + if (mode === "error") + return Response.json({ jsonrpc: "2.0", id: body.id, error: { code: -32603, message: "refresh failure" } }) + return Response.json({ + jsonrpc: "2.0", + id: body.id, + result: { + tools: + mode === "empty" + ? [] + : [ + { + name: `record_${mode === "hold" ? "stale" : mode}`, + description: mode, + inputSchema: { type: "object" }, + }, + ], + }, + }) + } finally { + active-- + } + }, + }) + const base = `http://127.0.0.1:${server.port}` + await Bun.write( + path.join(tmp.path, "aictrl.json"), + JSON.stringify({ + mcp: { fixture: { type: "remote", url: `${base}/mcp`, oauth: false, timeout: 1000 } }, + }), + ) + const probe = path.join(tmp.path, "probe.ts") + await Bun.write( + probe, + ` + import assert from "node:assert/strict" + import { MCP } from ${JSON.stringify(path.resolve(import.meta.dir, "../../src/mcp/index.ts"))} + import { Instance } from ${JSON.stringify(path.resolve(import.meta.dir, "../../src/project/instance.ts"))} + const base = ${JSON.stringify(base)} + const inspect = async () => (await fetch(base + "/inspect")).json() + const extras = async (healthy, calls) => { + const [prompts, resources] = await Promise.all([MCP.prompts(), MCP.resources()]) + const stats = await inspect() + assert.deepEqual({ + prompts: Object.keys(prompts), resources: Object.keys(resources), + calls: [stats.rpc["prompts/list"], stats.rpc["resources/list"]], + }, { + prompts: healthy ? ["fixture:review"] : [], resources: healthy ? ["fixture:evidence"] : [], + calls: [calls, calls], + }, "prompts and resources must follow the shared MCP health and suppress RPCs while failed") + } + const control = async (body) => { + const response = await fetch(base + "/control", { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify(body) }) + assert.equal(response.status, 200) + } + const wait = async (condition) => { + for (let i = 0; i < 300; i++) { + if (await condition()) return + await Bun.sleep(10) + } + throw new Error("Fixture condition did not become true") + } + await Instance.provide({ directory: process.cwd(), fn: async () => { + try { ${source} } finally { await Instance.dispose() } + } }) + `, + ) + const proc = Bun.spawn(["bun", "run", probe], { + cwd: tmp.path, + env: { + ...process.env, + AICTRL_CONFIG_CONTENT: "{}", + AICTRL_DISABLE_PROJECT_CONFIG: "false", + AICTRL_DISABLE_DEFAULT_PLUGINS: "true", + AICTRL_DISABLE_MODELS_FETCH: "true", + AICTRL_TEST_HOME: tmp.path, + AICTRL_TEST_MANAGED_CONFIG_DIR: path.join(tmp.path, "managed"), + XDG_DATA_HOME: path.join(tmp.path, "data"), + XDG_CONFIG_HOME: path.join(tmp.path, "config"), + XDG_STATE_HOME: path.join(tmp.path, "state"), + XDG_CACHE_HOME: path.join(tmp.path, "cache"), + }, + stdout: "pipe", + stderr: "pipe", + }) + const timeout = setTimeout(() => proc.kill("SIGKILL"), 15000) + try { + const [stderr, exit] = await Promise.all([new Response(proc.stderr).text(), proc.exited]) + expect(exit, stderr).toBe(0) + } finally { + clearTimeout(timeout) + proc.kill("SIGKILL") + releases.splice(0).forEach((resolve) => resolve()) + server.stop(true) + } +} + +test("tools and catalog share startup discovery, including a valid empty catalog", async () => { + for (const initial of ["initial", "empty"]) + await run( + ` + for (let i = 0; i < 3; i++) { + const [tools, entries] = await Promise.all([MCP.tools(), MCP.toolEntries()]) + assert.deepEqual(Object.keys(tools), ${JSON.stringify(initial === "empty" ? [] : ["fixture_record_initial"])}) + assert.deepEqual(entries, ${JSON.stringify(initial === "empty" ? [] : [{ toolKey: "fixture_record_initial", serverName: "fixture" }])}) + assert.equal((await MCP.status()).fixture.status, "connected") + } + assert.equal((await inspect()).lists["/mcp"], 1, "catalog reads must not rediscover tools") + `, + initial, + ) +}, 20000) + +test("notifications serialize refreshes, expose new definitions and recover from discovery errors", async () => { + await run(` + assert.equal((await MCP.status()).fixture.status, "connected") + await extras(true, 1) + await control({ mode: "updated", count: 2 }) + await wait(async () => (await inspect()).lists["/mcp"] >= 2) + assert.deepEqual(Object.keys(await MCP.tools()), ["fixture_record_updated"]) + await wait(async () => (await inspect()).lists["/mcp"] === 3) + assert.deepEqual(await MCP.toolEntries(), [{ toolKey: "fixture_record_updated", serverName: "fixture" }]) + assert.equal((await inspect()).maxActive, 1, "refreshes must serialize") + await control({ mode: "error" }) + await wait(async () => (await inspect()).lists["/mcp"] === 4) + await assert.rejects(MCP.tools(), /MCP tool discovery failed for server "fixture"/) + await assert.rejects(MCP.toolEntries(), /MCP tool discovery failed for server "fixture"/) + assert.equal((await MCP.status()).fixture.status, "failed") + await extras(false, 1) + assert.match((await MCP.status()).fixture.error, /Reconnect the MCP server and retry/) + assert.equal(Object.keys(await MCP.clients()).length, 1) + assert.equal((await inspect()).lists["/mcp"], 4, "failed catalog reads must not silently retry") + await control({ mode: "updated" }) + await wait(async () => (await inspect()).lists["/mcp"] === 5) + assert.deepEqual(Object.keys(await MCP.tools()), ["fixture_record_updated"]) + assert.equal((await MCP.status()).fixture.status, "connected") + await extras(true, 2) + await control({ mode: "empty" }) + await wait(async () => (await inspect()).lists["/mcp"] === 6) + assert.deepEqual(await MCP.tools(), {}) + assert.deepEqual(await MCP.toolEntries(), []) + assert.equal((await MCP.status()).fixture.status, "connected") + assert.equal((await inspect()).lists["/mcp"], 6) + `) +}, 20000) + +test("replacing or disconnecting a client isolates a late refresh and removes failed old connections", async () => { + await run(` + assert.equal((await MCP.status()).fixture.status, "connected") + await control({ mode: "hold" }) + await wait(async () => (await inspect()).lists["/mcp"] === 2) + const pending = MCP.tools().then(() => undefined, () => undefined) + await MCP.add("fixture", { type: "remote", url: base + "/replacement", oauth: false, timeout: 1000 }) + await control({ release: true }) + await pending + assert.deepEqual(Object.keys(await MCP.tools()), ["fixture_record_replacement"]) + assert.deepEqual(await MCP.toolEntries(), [{ toolKey: "fixture_record_replacement", serverName: "fixture" }]) + assert.equal((await MCP.status()).fixture.status, "connected") + const tools = await MCP.tools() + const called = await tools.fixture_record_replacement.execute({}, { toolCallId: "fixture", messages: [] }) + assert.equal(called.content[0].text, "/replacement", "tool bindings must use the replacement client") + await wait(async () => (await inspect()).closed["/mcp"] === 1) + await MCP.add("fixture", { type: "remote", url: base + "/failed", oauth: false, timeout: 1000 }) + assert.equal((await MCP.status()).fixture.status, "failed") + assert.deepEqual(await MCP.tools(), {}) + assert.deepEqual(await MCP.toolEntries(), []) + assert.equal(Object.keys(await MCP.clients()).length, 0) + await wait(async () => (await inspect()).closed["/replacement"] === 1) + await MCP.disconnect("fixture") + assert.equal((await MCP.status()).fixture.status, "disabled") + await MCP.add("dynamic", { type: "remote", url: base + "/replacement", oauth: false, timeout: 1000 }) + assert.deepEqual(Object.keys(await MCP.prompts()), ["dynamic:review"], "programmatic clients need not appear in config") + assert.deepEqual(Object.keys(await MCP.resources()), ["dynamic:evidence"]) + await MCP.disconnect("dynamic") + `) +}, 20000) + +test("transport closure fails catalog reads and reconnect installs a fresh client", async () => { + await run(` + assert.deepEqual(Object.keys(await MCP.tools()), ["fixture_record_initial"]) + await extras(true, 1) + const client = (await MCP.clients()).fixture + await client.close() + assert.equal((await MCP.status()).fixture.status, "failed") + await extras(false, 1) + assert.match((await MCP.status()).fixture.error, /Reconnect the MCP server and retry/) + await assert.rejects(MCP.tools(), /MCP connection closed for server "fixture"/) + await assert.rejects(MCP.toolEntries(), /MCP connection closed for server "fixture"/) + await MCP.connect("fixture") + assert.notEqual((await MCP.clients()).fixture, client) + assert.equal((await MCP.status()).fixture.status, "connected") + await extras(true, 2) + assert.deepEqual(Object.keys(await MCP.tools()), ["fixture_record_initial"]) + assert.equal((await inspect()).lists["/mcp"], 2) + await MCP.disconnect("fixture") + assert.deepEqual(await MCP.tools(), {}) + assert.deepEqual(await MCP.toolEntries(), []) + assert.equal(Object.keys(await MCP.clients()).length, 0) + `) +}, 20000) diff --git a/turbo.json b/turbo.json index 895740e7..7f24c464 100644 --- a/turbo.json +++ b/turbo.json @@ -1,7 +1,15 @@ { "$schema": "https://turborepo.com/schema.json", - "globalEnv": ["CI", "AICTRL_DISABLE_SHARE"], - "globalPassThroughEnv": ["CI", "AICTRL_DISABLE_SHARE", "AICTRL_VERSION", "AICTRL_CHANNEL", "AICTRL_RELEASE", "MODELS_DEV_API_JSON", "AICTRL_MODELS_URL", "GH_REPO"], + "globalEnv": ["CI", "AICTRL_DISABLE_SHARE", "AICTRL_VERSION", "AICTRL_BUILD_SINGLE"], + "globalPassThroughEnv": [ + "CI", + "AICTRL_DISABLE_SHARE", + "AICTRL_CHANNEL", + "AICTRL_RELEASE", + "MODELS_DEV_API_JSON", + "AICTRL_MODELS_URL", + "GH_REPO" + ], "tasks": { "typecheck": { "dependsOn": ["^build"]