diff --git a/packages/server/npm/README.md b/packages/server/npm/README.md index 534af32eb..d8d0eb915 100644 --- a/packages/server/npm/README.md +++ b/packages/server/npm/README.md @@ -19,6 +19,8 @@ codex mcp add auto-runtime -- npx -y @on.auto/runtime@0.15.0 mcp When your agent connects, the command starts Auto Runtime in the background if it is not running yet, and relays your agent to it. The runtime it starts keeps running after your agent leaves, so the workflows it runs carry on and other agents reach the same one. It reads its settings from `~/.auto-runtime/.env`, where you put a model provider's key, such as `ANTHROPIC_API_KEY`, and writes its log to `~/.auto-runtime/runtime.log`. When no model provider is set up, your agent is told which file the key goes in and how to restart the runtime once it is there. +If the runtime stops while your agent is connected, the next request starts it again within seconds, and is sent again unless it may already have run, in which case your agent is told to check before it repeats it. A runtime that stops answering is stopped and started again when the command started it; one you started yourself, or another program on the port, is left alone, and your agent is told what to stop. A runtime on the port that asks for a credential is left alone too, and your agent is told to stop it or connect over HTTP. + ## Start it yourself ```bash diff --git a/packages/server/src/local-runtime/agent-recovery.test.ts b/packages/server/src/local-runtime/agent-recovery.test.ts new file mode 100644 index 000000000..a0d9302bc --- /dev/null +++ b/packages/server/src/local-runtime/agent-recovery.test.ts @@ -0,0 +1,144 @@ +import { createServer } from 'node:http'; + +import { createApiKey } from '@on.auto/identity'; +import { Function } from 'effect'; +import { describe, expect, it, onTestFinished } from 'vitest'; + +import { ranToEnd, spawnRuntimeCommand } from '../testing/processes/runtime-command.ts'; +import { + answered, + answerWithId, + initializedLine, + initializeLine, + lineOf, + placeOfTheTest, + readiesIn, +} from '../testing/relays/relayed-processes.ts'; +import { until } from '../testing/waiting/until.ts'; +import { processesOfThisComputer } from './background-process.ts'; +import { restartWaitMs } from './background-start.ts'; + +const homePrefix = 'auto-runtime-recovery-'; + +const listPlaybooks = { + jsonrpc: '2.0', + id: 2, + method: 'tools/call', + params: { name: 'list_playbooks', arguments: {} }, +}; + +const listTools = { jsonrpc: '2.0', id: 2, method: 'tools/list', params: {} }; + +describe('an agent whose runtime in a terminal is stopped', () => { + it('has its next tool call answered by a runtime the relay starts, without waiting as long as a restart can take', async () => { + const { home, port } = await placeOfTheTest(homePrefix); + const terminal = spawnRuntimeCommand([], { home, port }); + await until(() => { + expect(readiesIn(terminal.stdout())).toBe(1); + }); + const relay = spawnRuntimeCommand(['mcp'], { home, port }); + relay.write(lineOf(initializeLine)); + await answered(relay, 1); + relay.write(lineOf(initializedLine)); + terminal.interrupt(); + await terminal.exited; + + const asked = performance.now(); + relay.write(lineOf(listPlaybooks)); + const answers = await answered(relay, 2); + const tookMs = performance.now() - asked; + const status = await ranToEnd(['status'], { home, port }); + relay.end(); + + expect(answerWithId(answers, 2)).toMatchObject({ id: 2, result: { structuredContent: { playbooks: [] } } }); + expect(tookMs).toBeLessThan(restartWaitMs); + expect(status.stdout).toContain('by your agent'); + expect(await relay.exited).toBe(0); + }); +}); + +describe('an agent whose runtime, which it started, stops answering', () => { + it('has the runtime stopped and another started, and its request answered by that one', async () => { + const { home, port, runtimes } = await placeOfTheTest(homePrefix); + const relay = spawnRuntimeCommand(['mcp'], { home, port }); + relay.write(lineOf(initializeLine)); + await answered(relay, 1); + relay.write(lineOf(initializedLine)); + const frozen = await runtimes.pid(); + onTestFinished(() => { + processesOfThisComputer.signal(frozen, 'SIGKILL'); + }); + process.kill(frozen, 'SIGSTOP'); + + relay.write(lineOf(listTools)); + const answers = await answered(relay, 2); + const replacing = await runtimes.pid(); + relay.end(); + + expect(JSON.stringify(answerWithId(answers, 2))).toContain('"name":"list_playbooks"'); + expect(replacing).not.toBe(frozen); + expect(relay.stderr()).toContain(`process ${frozen}, did not answer, so it was stopped and another started`); + expect(await relay.exited).toBe(0); + }); +}); + +describe('an agent whose runtime port a program holds that never answers', () => { + it('is told so, and that it can stop that program or set PORT, and nothing is started', async () => { + const { home, port } = await placeOfTheTest(homePrefix); + const silent = createServer(Function.constVoid).listen(port, '127.0.0.1'); + onTestFinished(() => { + silent.closeAllConnections(); + silent.close(); + }); + const relay = spawnRuntimeCommand(['mcp'], { home, port }); + + relay.write(lineOf(initializeLine)); + const answers = await answered(relay, 1); + relay.end(); + + expect(answers).toEqual([ + { + jsonrpc: '2.0', + id: 1, + error: { + code: -32_603, + message: `The program on port ${port} does not answer; stop it, or set PORT to a free port, and this MCP entry starts Auto Runtime`, + }, + }, + ]); + expect(await relay.exited).toBe(0); + }); +}); + +describe('an agent whose runtime asks for a credential', () => { + it('is told that this entry sends none, and how to reach it instead', async () => { + const { home, port } = await placeOfTheTest(homePrefix); + const { entry } = createApiKey({ id: 'ci-1', org: 'acme', permissions: ['org:read'], playbooks: '*' }); + const guarded = spawnRuntimeCommand([], { + home, + port, + settings: { HOST: '0.0.0.0', API_KEYS: JSON.stringify([entry]) }, + }); + await until(() => { + expect(readiesIn(guarded.stdout())).toBe(1); + }); + const relay = spawnRuntimeCommand(['mcp'], { home, port }); + + relay.write(lineOf(initializeLine)); + const answers = await answered(relay, 1); + relay.end(); + guarded.interrupt(); + + expect(answers).toEqual([ + { + jsonrpc: '2.0', + id: 1, + error: { + code: -32_603, + message: `Auto Runtime on port ${port} asks for a credential, which this MCP entry does not send; stop it, and this entry starts one, or connect your agent to http://localhost:${port}/mcp over HTTP`, + }, + }, + ]); + expect([await relay.exited, await guarded.exited]).toEqual([0, 0]); + }); +}); diff --git a/packages/server/src/local-runtime/agent-relay.test.ts b/packages/server/src/local-runtime/agent-relay.test.ts index a43b24d1d..b0f92de15 100644 --- a/packages/server/src/local-runtime/agent-relay.test.ts +++ b/packages/server/src/local-runtime/agent-relay.test.ts @@ -13,8 +13,15 @@ import { runtimeOnThePortEndedWithTheTest, spawnRuntimeCommand, type RuntimeCommandRun, - type RuntimeCommandPlace, } from '../testing/processes/runtime-command.ts'; +import { + answered, + answerWithId, + initializeLine, + lineOf, + placeOfTheTest, + readiesIn, +} from '../testing/relays/relayed-processes.ts'; import { until } from '../testing/waiting/until.ts'; const homePrefix = 'auto-runtime-agent-'; @@ -25,53 +32,13 @@ async function statusAt(port: number): Promise> { return statusOf(await (await fetch(`http://127.0.0.1:${port}/runtime`)).json()); } -async function placeOfTheTest(): Promise { - const home = homeOfTheTest(homePrefix); - const port = await aPortNothingListensOn(); - runtimeOnThePortEndedWithTheTest(home, port); - return { home, port }; -} - -const initialize = { - jsonrpc: '2.0', - id: 1, - method: 'initialize', - params: { protocolVersion: '2025-11-25', capabilities: {}, clientInfo: { name: 'agent', version: '1.0.0' } }, -}; - -function answersIn(stdout: string): readonly unknown[] { - return stdout - .split('\n') - .filter((line) => line !== '') - .map((line): unknown => JSON.parse(line)); -} - -function answerWithId(answers: readonly unknown[], id: number): unknown { - return answers.find((answer) => typeof answer === 'object' && answer !== null && 'id' in answer && answer.id === id); -} - -function lineOf(message: object): string { - return `${JSON.stringify(message)}\n`; -} - -async function answered(relay: Readonly, count: number): Promise { - await until(() => { - expect(answersIn(relay.stdout())).toHaveLength(count); - }); - return answersIn(relay.stdout()); -} - -function readiesIn(text: string): number { - return text.split('Press Ctrl+C to stop.').length - 1; -} - function markedAsRestartingBy(home: string, pid: number): void { mkdirSync(join(home, '.auto-runtime'), { recursive: true }); writeFileSync(join(home, '.auto-runtime', 'restart-mark.json'), JSON.stringify({ pid, at: Date.now() })); } async function leftAfterItsAgentLeft(relay: Readonly): Promise { - relay.write(lineOf(initialize)); + relay.write(lineOf(initializeLine)); await answered(relay, 1); relay.end(); return relay.exited; @@ -86,7 +53,7 @@ describe('an agent that starts auto-runtime mcp', () => { const starting = spawnRuntimeCommand(['mcp'], { home, port, inAGroupOfItsOwn: true }); const pid = await runtimes.pid(); - starting.write(lineOf(initialize)); + starting.write(lineOf(initializeLine)); await answered(starting, 1); starting.write(lineOf({ jsonrpc: '2.0', method: 'notifications/initialized' })); starting.write(lineOf({ jsonrpc: '2.0', id: 2, method: 'tools/list', params: {} })); @@ -126,13 +93,13 @@ describe('an agent that starts auto-runtime mcp', () => { describe('an agent whose runtime restarts', () => { it('has a request it sends while the runtime is down sent again once the runtime is back', async () => { - const { home, port } = await placeOfTheTest(); + const { home, port } = await placeOfTheTest(homePrefix); const runtime = spawnRuntimeCommand([], { home, port }); await until(() => { expect(readiesIn(runtime.stdout())).toBe(1); }); const relay = spawnRuntimeCommand(['mcp'], { home, port }); - relay.write(lineOf(initialize)); + relay.write(lineOf(initializeLine)); await answered(relay, 1); relay.write(lineOf({ jsonrpc: '2.0', method: 'notifications/initialized' })); @@ -151,10 +118,10 @@ describe('an agent whose runtime restarts', () => { describe('an agent that starts while a runtime restarts', () => { it('waits for that runtime to come back instead of starting a second one', async () => { - const { home, port } = await placeOfTheTest(); + const { home, port } = await placeOfTheTest(homePrefix); markedAsRestartingBy(home, process.pid); const relay = spawnRuntimeCommand(['mcp'], { home, port }); - relay.write(lineOf(initialize)); + relay.write(lineOf(initializeLine)); const runtime = spawnRuntimeCommand([], { home, port }); await answered(relay, 1); @@ -170,7 +137,7 @@ describe('an agent that starts while a runtime restarts', () => { describe('an agent whose runtime port another program holds', () => { it('is told why it cannot reach the runtime, and answered once with what its request met', async () => { - const { home, port } = await placeOfTheTest(); + const { home, port } = await placeOfTheTest(homePrefix); const other = createServer((_request, response) => { response.end('hello'); }).listen(port, '127.0.0.1'); @@ -179,7 +146,7 @@ describe('an agent whose runtime port another program holds', () => { }); const relay = spawnRuntimeCommand(['mcp'], { home, port }); - relay.write(lineOf(initialize)); + relay.write(lineOf(initializeLine)); const answers = await answered(relay, 1); relay.end(); diff --git a/packages/server/src/local-runtime/agent-relay.ts b/packages/server/src/local-runtime/agent-relay.ts index 898d266f5..d209a299b 100644 --- a/packages/server/src/local-runtime/agent-relay.ts +++ b/packages/server/src/local-runtime/agent-relay.ts @@ -5,11 +5,17 @@ import { StdioServerTransport } from '@modelcontextprotocol/server/stdio'; import type { Environment } from '@on.auto/config'; import { release } from '../composition/release.ts'; -import { noticesFor, restartMarkIn, startInBackgroundOf, startingLockIn } from './background-process.ts'; -import { impatient, runningRuntime, throughARestart } from './background-start.ts'; +import { + noticesFor, + processesOfThisComputer, + restartMarkIn, + startInBackgroundOf, + startingLockIn, +} from './background-process.ts'; +import { briefly, impatient, runningRuntime, sightingsOfARelay } from './background-start.ts'; import { logFileIn } from './local-control.ts'; -import { relay } from './mcp-relay.ts'; -import { runtimeAddressOf, type Waiting } from './runtime-probe.ts'; +import { relay, type Timer } from './mcp-relay.ts'; +import { isAnswering, probeAt, runtimeAddressOf, type Waiting } from './runtime-probe.ts'; export interface AgentRelaying { readonly home: string; @@ -21,6 +27,14 @@ function report(line: string): void { process.stderr.write(`${line}\n`); } +const timer: Timer = (milliseconds, then) => { + const timeout = setTimeout(then, milliseconds); + timeout.unref(); + return () => { + clearTimeout(timeout); + }; +}; + export async function relayedAgent(port: number, { home, environment, waiting }: AgentRelaying): Promise { const starting = { ...waiting, @@ -29,6 +43,8 @@ export async function relayedAgent(port: number, { home, environment, waiting }: startingLock: startingLockIn(home), notice: noticesFor(home, report), restartUnderWay: restartMarkIn(home).isUnderWay, + processes: processesOfThisComputer, + sightings: sightingsOfARelay(), startInBackground: startInBackgroundOf({ execPath: process.execPath, execArgv: process.execArgv, @@ -45,7 +61,9 @@ export async function relayedAgent(port: number, { home, environment, waiting }: await relay({ agent: () => new StdioServerTransport(), runtime: () => new StreamableHTTPClientTransport(new URL(`${runtimeAddressOf(port)}/mcp`)), - recover: () => runningRuntime(port, starting, throughARestart), + recover: () => runningRuntime(port, starting, briefly), + answering: async () => isAnswering(await probeAt(port, waiting.fetch)), + timer, report, }); } diff --git a/packages/server/src/local-runtime/background-process.test.ts b/packages/server/src/local-runtime/background-process.test.ts index fb845772e..0b85bdc25 100644 --- a/packages/server/src/local-runtime/background-process.test.ts +++ b/packages/server/src/local-runtime/background-process.test.ts @@ -1,4 +1,4 @@ -import { spawnSync } from 'node:child_process'; +import { spawn, spawnSync } from 'node:child_process'; import { chmodSync, existsSync, @@ -18,10 +18,16 @@ import { join } from 'node:path'; import { fileURLToPath } from 'node:url'; import { Function } from 'effect'; -import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import { afterEach, beforeEach, describe, expect, it, onTestFinished } from 'vitest'; import { until } from '../testing/waiting/until.ts'; -import { noticesFor, restartMarkIn, startInBackgroundOf, startingLockIn } from './background-process.ts'; +import { + noticesFor, + processesOfThisComputer, + restartMarkIn, + startInBackgroundOf, + startingLockIn, +} from './background-process.ts'; import { restartWaitMs, startWaitMs } from './background-start.ts'; const recordingEntry = fileURLToPath(new URL('../testing/entries/record-background-start.ts', import.meta.url)); @@ -225,3 +231,21 @@ describe('the mark a runtime leaves while it restarts', () => { expect(restartMarkIn(home).leave(process.pid)).toBe(false); }); }); + +describe('the processes of this computer', () => { + it('signals one that runs, which is then gone, and signals none once it is gone', async () => { + const child = spawn(process.execPath, ['-e', 'setInterval(() => {}, 1000)'], { stdio: 'ignore' }); + onTestFinished(() => { + child.kill('SIGKILL'); + }); + const running = Number(child.pid); + + const aliveAtFirst = processesOfThisComputer.alive(running); + const killed = processesOfThisComputer.signal(running, 'SIGKILL'); + await until(() => { + expect(processesOfThisComputer.alive(running)).toBe(false); + }); + + expect([aliveAtFirst, killed, processesOfThisComputer.signal(running, 'SIGTERM')]).toEqual([true, true, false]); + }); +}); diff --git a/packages/server/src/local-runtime/background-process.ts b/packages/server/src/local-runtime/background-process.ts index 9d5d8b172..f8edcbbf9 100644 --- a/packages/server/src/local-runtime/background-process.ts +++ b/packages/server/src/local-runtime/background-process.ts @@ -16,7 +16,7 @@ import { join } from 'node:path'; import type { Environment } from '@on.auto/config'; import { Option, Schema } from 'effect'; -import { restartWaitMs, startWaitMs, type StartingLock } from './background-start.ts'; +import { restartWaitMs, startWaitMs, type RuntimeProcesses, type StartingLock } from './background-start.ts'; import { logFileIn, runtimeHomeIn } from './local-control.ts'; export interface BackgroundCommand { @@ -181,3 +181,13 @@ export function restartMarkIn(home: string, now: () => number = Date.now): Resta }, }; } + +function signalled(pid: number, signal: NodeJS.Signals): boolean { + try { + return process.kill(pid, signal); + } catch { + return false; + } +} + +export const processesOfThisComputer: RuntimeProcesses = { signal: signalled, alive: isRunning }; diff --git a/packages/server/src/local-runtime/background-start.test.ts b/packages/server/src/local-runtime/background-start.test.ts index f38b9cbab..daf5d2a6e 100644 --- a/packages/server/src/local-runtime/background-start.test.ts +++ b/packages/server/src/local-runtime/background-start.test.ts @@ -2,104 +2,10 @@ import { describe, expect, it } from 'vitest'; import { defaultServerOptions } from '../lifecycle/lifecycle.ts'; import { longestStopMs } from '../lifecycle/shutdown.ts'; -import { - impatient, - restartWaitMs, - runningRuntime, - startWaitMs, - throughARestart, - type BackgroundStarting, -} from './background-start.ts'; +import { computerWith, startedForAnAgent, startingOn, type Computer } from '../testing/relays/fake-computers.ts'; +import { impatient, restartWaitMs, runningRuntime, startWaitMs, throughARestart } from './background-start.ts'; import { probeEveryMs, type RuntimeFetch } from './runtime-probe.ts'; -const startedForAnAgent = { - version: '0.9.1', - pid: 4242, - started_at: '2026-10-14T09:41:07.000Z', - started_by: 'agent', - directory: '/Users/ada/.auto-runtime', - log_file: '/Users/ada/.auto-runtime/runtime.log', - runs_in_flight: 0, -}; - -interface Computer { - readonly answering: () => object | undefined; - readonly answerWith: (answer?: object) => void; - readonly record: (event: string) => void; - readonly events: () => readonly string[]; -} - -function computerWith(first?: object): Computer { - let answering = first; - const events: string[] = []; - return { - answering: () => answering, - answerWith: (answer) => { - answering = answer; - }, - record: (event) => { - events.push(event); - }, - events: () => [...events], - }; -} - -function fetchOn(computer: Computer): RuntimeFetch { - return (address, { method }) => { - const answering = computer.answering(); - if (answering === undefined) { - return Promise.reject(new TypeError('fetch failed')); - } - if (method === 'POST') { - computer.record(`asked ${address.slice(address.lastIndexOf('/') + 1)}`); - computer.answerWith(); - return Promise.resolve(Response.json({ status: 'stopping' }, { status: 202 })); - } - return Promise.resolve(Response.json(answering)); - }; -} - -interface Options { - readonly version?: string; - readonly startsAs?: object | undefined; - readonly comesBackAfter?: number; - readonly lockTaken?: boolean; - readonly restarting?: boolean; -} - -function startingOn(computer: Computer, options: Options = {}): BackgroundStarting { - let sleeps = 0; - return { - version: options.version ?? '0.9.1', - logFile: '/Users/ada/.auto-runtime/runtime.log', - fetch: fetchOn(computer), - sleep: () => { - sleeps += 1; - if (options.comesBackAfter === sleeps) { - computer.answerWith(startedForAnAgent); - } - return Promise.resolve(); - }, - startInBackground: () => { - computer.record('started'); - computer.answerWith('startsAs' in options ? options.startsAs : startedForAnAgent); - }, - startingLock: { - take: () => { - computer.record('locked'); - return options.lockTaken !== true; - }, - release: () => { - computer.record('unlocked'); - }, - }, - notice: (line) => { - computer.record(`noticed ${line}`); - }, - restartUnderWay: () => options.restarting === true, - }; -} - function refusingToStop(fetch: RuntimeFetch): RuntimeFetch { return (address, request) => request.method === 'POST' ? Promise.reject(new TypeError('fetch failed')) : fetch(address, request); diff --git a/packages/server/src/local-runtime/background-start.ts b/packages/server/src/local-runtime/background-start.ts index 1e880547e..e7e75f358 100644 --- a/packages/server/src/local-runtime/background-start.ts +++ b/packages/server/src/local-runtime/background-start.ts @@ -1,5 +1,6 @@ import { longestStopMs } from '../lifecycle/shutdown.ts'; import { + isAnswering, probeAt, probeEveryMs, stoppedWhenAsked, @@ -14,6 +15,16 @@ export interface StartingLock { readonly release: () => void; } +export interface RuntimeProcesses { + readonly signal: (pid: number, signal: 'SIGTERM' | 'SIGKILL') => boolean; + readonly alive: (pid: number) => boolean; +} + +export interface Sightings { + readonly saw: (status: RunningStatus) => void; + readonly last: () => RunningStatus | undefined; +} + export interface BackgroundStarting extends Waiting { readonly version: string; readonly startInBackground: () => void; @@ -21,6 +32,8 @@ export interface BackgroundStarting extends Waiting { readonly logFile: string; readonly notice: (line: string) => void; readonly restartUnderWay: () => boolean; + readonly processes: RuntimeProcesses; + readonly sightings: Sightings; } export interface Patience { @@ -33,10 +46,30 @@ const startAttempts = Math.ceil(startWaitMs / probeEveryMs); export const impatient: Patience = { attempts: 0 }; +const briefWaitMs = 1000; + +export const briefly: Patience = { attempts: Math.ceil(briefWaitMs / probeEveryMs) }; + export const restartWaitMs = longestStopMs + startWaitMs; export const throughARestart: Patience = { attempts: Math.ceil(restartWaitMs / probeEveryMs) }; +const unansweredRetries = 2; + +const stopAttempts = Math.ceil(longestStopMs / probeEveryMs); + +const killAttempts = 20; + +export function sightingsOfARelay(): Sightings { + let last: RunningStatus | undefined; + return { + saw: (status) => { + last = status; + }, + last: () => last, + }; +} + function isNewer(version: string, than: string): boolean { return version.localeCompare(than, 'en', { numeric: true }) > 0; } @@ -51,8 +84,11 @@ function patienceFor(patience: Patience, starting: BackgroundStarting): Patience return starting.restartUnderWay() ? throughARestart : patience; } -function isAnswering(probe: Probe): boolean { - return probe.state === 'running' || probe.state === 'uncontrolled'; +function seen(probe: Seen, starting: BackgroundStarting): Seen { + if (probe?.state === 'running') { + starting.sightings.saw(probe.status); + } + return probe; } async function startedHere(port: number, starting: BackgroundStarting): Promise { @@ -61,9 +97,9 @@ async function startedHere(port: number, starting: BackgroundStarting): Promise< starting.startInBackground(); } try { - return (await waitedFor(port, starting, startAttempts, isAnswering)) - ? undefined - : `Auto Runtime did not start on port ${port}; its log is ${starting.logFile}`; + return seen(await waitedFor(port, starting, startAttempts, isAnswering), starting) === undefined + ? `Auto Runtime did not start on port ${port}; its log is ${starting.logFile}` + : undefined; } finally { if (took) { starting.startingLock.release(); @@ -87,21 +123,72 @@ async function replaced( : `Auto Runtime ${outdated.version} on port ${port} did not stop, so Auto Runtime ${starting.version} could not replace it`; } +async function gone(pid: number, starting: BackgroundStarting, attempts: number): Promise { + if (!starting.processes.alive(pid)) { + return true; + } + if (attempts <= 0) { + return false; + } + await starting.sleep(probeEveryMs); + return gone(pid, starting, attempts - 1); +} + +async function ended(pid: number, starting: BackgroundStarting): Promise { + starting.processes.signal(pid, 'SIGTERM'); + if (!(await gone(pid, starting, stopAttempts))) { + starting.processes.signal(pid, 'SIGKILL'); + await gone(pid, starting, killAttempts); + } +} + +function unansweredBy(port: number, last: RunningStatus | undefined): string { + return last?.started_by === 'terminal' + ? `Auto Runtime on port ${port}, which runs in a terminal as process ${last.pid}, does not answer; stop it there, and this MCP entry starts another` + : `The program on port ${port} does not answer; stop it, or set PORT to a free port, and this MCP entry starts Auto Runtime`; +} + +async function unanswering(port: number, starting: BackgroundStarting): Promise { + if (seen(await waitedFor(port, starting, unansweredRetries, isAnswering), starting) !== undefined) { + return undefined; + } + const last = starting.sightings.last(); + if (last?.started_by !== 'agent') { + return unansweredBy(port, last); + } + starting.notice( + `Auto Runtime ${last.version} on port ${port}, process ${last.pid}, did not answer, so it was stopped and another started`, + ); + await ended(last.pid, starting); + return startedHere(port, starting); +} + +const refusals: Partial string>> = { + foreign: (port) => `Another program answers on port ${port}; set PORT to start Auto Runtime on a free port`, + guarded: (port) => + `Auto Runtime on port ${port} asks for a credential, which this MCP entry does not send; stop it, and this entry starts one, or connect your agent to http://localhost:${port}/mcp over HTTP`, +}; + export async function runningRuntime( port: number, starting: BackgroundStarting, patience: Patience, ): Promise { - const probe = await probeAt(port, starting.fetch); + const probe = seen(await probeAt(port, starting.fetch), starting); const outdated = outdatedOf(probe, starting.version); if (outdated !== undefined) { return replaced(port, outdated, starting); } - if (probe.state === 'foreign') { - return `Another program answers on port ${port}; set PORT to start Auto Runtime on a free port`; + const refused = refusals[probe.state]; + if (refused !== undefined) { + return refused(port); + } + if (probe.state === 'unresponsive') { + return unanswering(port, starting); } - if (isAnswering(probe) || (await waitedFor(port, starting, patienceFor(patience, starting).attempts, isAnswering))) { + if (isAnswering(probe)) { return undefined; } - return startedHere(port, starting); + const back = await waitedFor(port, starting, patienceFor(patience, starting).attempts, isAnswering); + return seen(back, starting) === undefined ? startedHere(port, starting) : undefined; } diff --git a/packages/server/src/local-runtime/mcp-relay.test.ts b/packages/server/src/local-runtime/mcp-relay.test.ts index e1e6e6d9a..aa9528e33 100644 --- a/packages/server/src/local-runtime/mcp-relay.test.ts +++ b/packages/server/src/local-runtime/mcp-relay.test.ts @@ -1,27 +1,20 @@ -import { InMemoryTransport, parseJSONRPCMessage, type Transport } from '@modelcontextprotocol/server'; +import { parseJSONRPCMessage } from '@modelcontextprotocol/server'; import { describe, expect, it } from 'vitest'; +import { + answeredWithStatus, + callATool, + fakeRuntime, + fakeRuntimeEnd, + idOf, + initialize, + initialized, + listTools, + refusedAtConnect, + relayed, + type Reply, +} from '../testing/relays/fake-relays.ts'; import { until } from '../testing/waiting/until.ts'; -import { relay } from './mcp-relay.ts'; - -const listTools = parseJSONRPCMessage({ jsonrpc: '2.0', id: 2, method: 'tools/list', params: {} }); - -const initialize = parseJSONRPCMessage({ - jsonrpc: '2.0', - id: 1, - method: 'initialize', - params: { protocolVersion: '2025-11-25', capabilities: {}, clientInfo: { name: 'agent', version: '1.0.0' } }, -}); - -const initialized = parseJSONRPCMessage({ jsonrpc: '2.0', method: 'notifications/initialized' }); - -type Reply = (message: unknown) => unknown; - -function idOf(message: unknown): unknown { - return typeof message === 'object' && message !== null && 'id' in message ? message.id : undefined; -} - -const echoTools: Reply = (message) => ({ jsonrpc: '2.0', id: idOf(message), result: { tools: [] } }); const agreeOnAVersion: Reply = (message) => idOf(message) === undefined @@ -30,125 +23,6 @@ const agreeOnAVersion: Reply = (message) => const nameNoVersion: Reply = (message) => ({ jsonrpc: '2.0', id: idOf(message), result: {} }); -interface FakeRuntimeOptions { - readonly reply?: Reply; - readonly failures?: readonly Readonly[]; - readonly negotiates?: boolean; -} - -interface FakeRuntime { - readonly transport: Transport; - readonly received: unknown[]; - readonly negotiated: string[]; - readonly closed: () => boolean; - readonly sends: () => number; -} - -function fakeRuntime({ reply = echoTools, failures = [], negotiates = true }: FakeRuntimeOptions = {}): FakeRuntime { - const received: unknown[] = []; - const negotiated: string[] = []; - const failing = [...failures]; - let closed = false; - let sends = 0; - const transport: Transport = { - start: () => Promise.resolve(), - send: (message: unknown) => { - sends += 1; - const failure = failing.shift(); - if (failure !== undefined) { - return Promise.reject(failure); - } - received.push(message); - const answer = reply(message); - if (answer !== undefined) { - queueMicrotask(() => transport.onmessage?.(parseJSONRPCMessage(answer))); - } - return Promise.resolve(); - }, - close: () => { - closed = true; - return Promise.resolve(); - }, - ...(negotiates - ? { - setProtocolVersion: (version: string) => { - negotiated.push(version); - }, - } - : {}), - }; - return { transport, received, negotiated, closed: () => closed, sends: () => sends }; -} - -const refusedAtConnect = new TypeError('fetch failed', { - cause: Object.assign(new Error('connect ECONNREFUSED 127.0.0.1:5555'), { code: 'ECONNREFUSED', syscall: 'connect' }), -}); - -const cutOffMidAnswer = new TypeError('terminated', { - cause: Object.assign(new Error('other side closed'), { code: 'UND_ERR_SOCKET' }), -}); - -const droppedAfterTheRequest = new TypeError('fetch failed', { - cause: Object.assign(new Error('other side closed'), { code: 'UND_ERR_SOCKET' }), -}); - -function fakeRuntimeEnd(options: FakeRuntimeOptions = {}): () => Transport { - const { transport } = fakeRuntime(options); - return () => transport; -} - -interface Relayed { - readonly agent: Transport; - readonly relayEnd: Transport; - readonly reported: string[]; - readonly recovered: () => number; - readonly ended: Promise; - readonly next: () => Promise; -} - -async function relayed(runtimeEnd: () => Transport, recovery?: string): Promise { - const [agent, relayEnd] = InMemoryTransport.createLinkedPair(); - const reported: string[] = []; - const answers: unknown[] = []; - const waiting: ((message: unknown) => void)[] = []; - let recovered = 0; - Object.assign(agent, { - onmessage: (message: unknown) => { - const waiter = waiting.shift(); - if (waiter === undefined) { - answers.push(message); - } else { - waiter(message); - } - }, - }); - const ended = relay({ - agent: () => relayEnd, - runtime: runtimeEnd, - recover: () => { - recovered += 1; - return Promise.resolve(recovery); - }, - report: (line) => { - reported.push(line); - }, - }); - await agent.start(); - return { - agent, - relayEnd, - reported, - recovered: () => recovered, - ended, - next: () => - answers.length > 0 - ? Promise.resolve(answers.shift()) - : new Promise((resolve) => { - waiting.push(resolve); - }), - }; -} - describe('relaying an agent to the runtime', () => { it('sends what the agent asks to the runtime, and its answer back', async () => { const runtime = fakeRuntime(); @@ -192,6 +66,37 @@ describe('relaying an agent to the runtime', () => { }); }); +const refuseTheCall: Reply = (message) => ({ + jsonrpc: '2.0', + id: idOf(message), + error: { code: -32_602, message: 'Unknown tool: run_nothing' }, +}); + +describe('what the runtime sends the agent', () => { + it('relays an error it answers a request with, as the answer to that request', async () => { + const runtime = fakeRuntime({ reply: refuseTheCall }); + const { agent, next, recovered } = await relayed(() => runtime.transport); + + await agent.send(callATool); + + expect(await next()).toEqual({ + jsonrpc: '2.0', + id: 3, + error: { code: -32_602, message: 'Unknown tool: run_nothing' }, + }); + expect([recovered(), runtime.sends()]).toEqual([0, 1]); + }); + + it('relays a notification it sends of its own accord', async () => { + const runtime = fakeRuntime(); + const { next } = await relayed(() => runtime.transport); + + runtime.transport.onmessage?.(parseJSONRPCMessage({ jsonrpc: '2.0', method: 'notifications/tools/list_changed' })); + + expect(await next()).toEqual({ jsonrpc: '2.0', method: 'notifications/tools/list_changed' }); + }); +}); + describe('the end of a relay', () => { it('closes its connection to the runtime and ends once the agent leaves', async () => { const runtime = fakeRuntime(); @@ -204,23 +109,27 @@ describe('the end of a relay', () => { }); }); -describe('relaying an agent to a runtime that does not answer', () => { - it('makes sure the runtime runs, then sends again a request whose connection could not be made', async () => { - const runtime = fakeRuntime({ failures: [refusedAtConnect] }); +describe('relaying an agent to a runtime its request never reached', () => { + it.each]>([ + ['whose connection could not be made', refusedAtConnect], + ['that asked for a credential', answeredWithStatus(401)], + ])('makes sure the runtime runs, then sends a request %s again', async (_case, failure) => { + const runtime = fakeRuntime({ turns: [failure] }); const { agent, next, recovered } = await relayed(() => runtime.transport); - await agent.send(listTools); + await agent.send(callATool); - expect(await next()).toEqual({ jsonrpc: '2.0', id: 2, result: { tools: [] } }); + expect(await next()).toEqual({ jsonrpc: '2.0', id: 3, result: { tools: [] } }); expect(recovered()).toBe(1); expect(runtime.sends()).toBe(2); }); +}); - it('answers the request with why the runtime could not be started', async () => { - const { agent, next } = await relayed( - fakeRuntimeEnd({ failures: [refusedAtConnect] }), - 'Auto Runtime did not start on port 5555; its log is /Users/ada/.auto-runtime/runtime.log', - ); +describe('relaying an agent to a runtime it could not make sure of', () => { + it('answers the request with why the runtime could not be reached', async () => { + const { agent, next } = await relayed(fakeRuntimeEnd({ turns: [answeredWithStatus(401)] }), { + problem: 'Auto Runtime on port 5555 asks for a credential, which this MCP entry does not send', + }); await agent.send(listTools); @@ -229,26 +138,39 @@ describe('relaying an agent to a runtime that does not answer', () => { id: 2, error: { code: -32_603, - message: 'Auto Runtime did not start on port 5555; its log is /Users/ada/.auto-runtime/runtime.log', + message: 'Auto Runtime on port 5555 asks for a credential, which this MCP entry does not send', }, }); }); it('answers the request with what went wrong when it fails again', async () => { const { agent, next } = await relayed( - fakeRuntimeEnd({ failures: [refusedAtConnect, new Error('Error POSTing to endpoint')] }), + fakeRuntimeEnd({ turns: [refusedAtConnect, new Error('Error POSTing to endpoint')] }), ); await agent.send(listTools); expect(await next()).toMatchObject({ id: 2, error: { message: 'Error: Error POSTing to endpoint' } }); }); + + it('answers the request with what the recovery failed with', async () => { + const { agent, next } = await relayed(fakeRuntimeEnd({ turns: [refusedAtConnect] }), { + recover: () => Promise.reject(new Error('EACCES: permission denied, open runtime.log')), + }); + + await agent.send(listTools); + + expect(await next()).toMatchObject({ + id: 2, + error: { message: 'Error: EACCES: permission denied, open runtime.log' }, + }); + }); }); describe('relaying a notification to a runtime that does not answer', () => { it('reports a notification it could not deliver, which has no one to answer', async () => { const { agent, reported } = await relayed( - fakeRuntimeEnd({ failures: [refusedAtConnect, new Error('refused again')] }), + fakeRuntimeEnd({ turns: [refusedAtConnect, new Error('refused again')] }), ); await agent.send(initialized); @@ -259,27 +181,6 @@ describe('relaying a notification to a runtime that does not answer', () => { }); }); -describe('relaying an agent to a runtime that may have taken the request', () => { - it.each, string]>([ - ['whose answer was cut off', cutOffMidAnswer, 'TypeError: terminated'], - ['whose connection closed after it went out', droppedAfterTheRequest, 'TypeError: fetch failed'], - [ - 'that the runtime answered with an error status', - new Error('Error POSTing to endpoint: busy'), - 'Error: Error POSTing to endpoint: busy', - ], - ])('answers a request %s with what went wrong, and never sends it again', async (_case, failure, problem) => { - const runtime = fakeRuntime({ failures: [failure] }); - const { agent, next, recovered } = await relayed(() => runtime.transport); - - await agent.send(listTools); - - expect(await next()).toEqual({ jsonrpc: '2.0', id: 2, error: { code: -32_603, message: problem } }); - expect(runtime.sends()).toBe(1); - expect(recovered()).toBe(0); - }); -}); - describe('the errors of the two connections of the relay', () => { it('reports those of the connection to the runtime', async () => { const runtime = fakeRuntime(); diff --git a/packages/server/src/local-runtime/mcp-relay.ts b/packages/server/src/local-runtime/mcp-relay.ts index 74919428e..b226b5f84 100644 --- a/packages/server/src/local-runtime/mcp-relay.ts +++ b/packages/server/src/local-runtime/mcp-relay.ts @@ -1,27 +1,83 @@ +import { SdkHttpError } from '@modelcontextprotocol/client'; import { + isJSONRPCErrorResponse, isJSONRPCRequest, isJSONRPCResultResponse, parseJSONRPCMessage, type Transport, } from '@modelcontextprotocol/server'; +export type Timer = (milliseconds: number, then: () => void) => () => void; + export interface Relaying { readonly agent: () => Transport; readonly runtime: () => Transport; readonly recover: () => Promise; + readonly answering: () => Promise; + readonly timer: Timer; readonly report: (line: string) => void; } +export const answerAwaitedMs = 5000; + +export const recoveryAwaitedMs = 45_000; + const internalError = -32_603; -type Sending = { readonly failed: false } | { readonly failed: true; readonly failure: unknown }; +const unauthenticated = 401; + +const unanswered = new Error('Auto Runtime did not answer'); + +const cutOff = new Error('Auto Runtime stopped before it answered'); + +const lostOnTheWay: ReadonlySet = new Set([unanswered, cutOff]); + +export const mayHaveTakenEffect = + 'Auto Runtime stopped answering while it had this request, and runs again now; check whether the request took effect before you send it again'; + +export const stillComing = + 'Auto Runtime is not answering yet, as it is being started or restarted; send the request again in a few seconds'; + +const repeatable: ReadonlySet = new Set([ + 'initialize', + 'ping', + 'tools/list', + 'prompts/list', + 'prompts/get', + 'resources/list', + 'resources/templates/list', + 'resources/read', + 'completion/complete', +]); + +type Delivery = { readonly delivered: true } | { readonly delivered: false; readonly failure: unknown }; + +type Loss = 'never arrived' | 'may have arrived' | 'answered'; function couldNotConnect(cause: unknown): boolean { return typeof cause === 'object' && cause !== null && 'syscall' in cause && cause.syscall === 'connect'; } -function neverWentOut(failure: unknown): boolean { - return failure instanceof TypeError && couldNotConnect(failure.cause); +function neverArrived(failure: unknown): boolean { + return ( + (failure instanceof TypeError && couldNotConnect(failure.cause)) || + (failure instanceof SdkHttpError && failure.status === unauthenticated) + ); +} + +function lossOf(failure: unknown): Loss { + if (neverArrived(failure)) { + return 'never arrived'; + } + return lostOnTheWay.has(failure) || failure instanceof TypeError ? 'may have arrived' : 'answered'; +} + +function isRepeatable(message: unknown): boolean { + return !isJSONRPCRequest(message) || repeatable.has(message.method); +} + +function idOf(message: unknown): unknown { + return isJSONRPCRequest(message) ? message.id : undefined; } class McpRelay { @@ -29,7 +85,9 @@ class McpRelay { readonly #runtime: Transport; readonly #relaying: Relaying; readonly #initializing = new Set(); + readonly #awaited = new Set(); readonly #ended = Promise.withResolvers(); + #recovering: Promise | undefined; constructor(relaying: Relaying) { this.#agent = relaying.agent(); @@ -64,6 +122,9 @@ class McpRelay { } #answered(message: unknown): void { + if (isJSONRPCResultResponse(message) || isJSONRPCErrorResponse(message)) { + this.#awaited.delete(message.id); + } if (isJSONRPCResultResponse(message) && this.#initializing.delete(message.id)) { const { protocolVersion } = message.result; if (typeof protocolVersion === 'string') { @@ -77,32 +138,101 @@ class McpRelay { if (isJSONRPCRequest(message) && message.method === 'initialize') { this.#initializing.add(message.id); } - const sending = await this.#sent(message); - if (!sending.failed) { + const lost = (failure: unknown) => this.#afterALoss(message, failure); + const delivery = await this.#delivered(message, lost); + if (!delivery.delivered) { + await lost(delivery.failure); + } + } + + async #afterALoss(message: unknown, failure: unknown): Promise { + const loss = lossOf(failure); + if (loss === 'answered') { + await this.#refused(message, String(failure)); return; } - const problem = neverWentOut(sending.failure) ? await this.#sentAgain(message) : String(sending.failure); + const problem = await this.#recovered(); if (problem !== undefined) { await this.#refused(message, problem); + return; + } + if (loss === 'may have arrived' && !isRepeatable(message)) { + await this.#refused(message, mayHaveTakenEffect); + return; + } + const lostAgain = (again: unknown) => this.#refused(message, String(again)); + const again = await this.#delivered(message, lostAgain); + if (!again.delivered) { + await lostAgain(again.failure); } } - async #sent(message: unknown): Promise { + async #delivered(message: unknown, lost: (failure: unknown) => Promise): Promise { + const id = idOf(message); + const givingUp = new AbortController(); + const stopWatching = this.#watched(givingUp); + if (id !== undefined) { + this.#awaited.add(id); + } try { - await this.#runtime.send(parseJSONRPCMessage(message)); - return { failed: false }; + await this.#runtime.send(parseJSONRPCMessage(message), { + requestSignal: givingUp.signal, + onRequestStreamEnd: () => { + if (this.#awaited.has(id)) { + void lost(cutOff); + } + }, + }); + return { delivered: true }; } catch (failure) { - return { failed: true, failure }; + return { delivered: false, failure: givingUp.signal.aborted ? unanswered : failure }; + } finally { + stopWatching(); } } - async #sentAgain(message: unknown): Promise { - const recovering = await this.#relaying.recover(); - if (recovering !== undefined) { - return recovering; + #watched(givingUp: AbortController): () => void { + const watch = { + on: true, + cancel: this.#relaying.timer(answerAwaitedMs, () => { + void check(); + }), + }; + const check = async (): Promise => { + const answering = await this.#relaying.answering(); + if (!watch.on) { + return; + } + if (!answering) { + givingUp.abort(); + return; + } + watch.cancel = this.#relaying.timer(answerAwaitedMs, () => { + void check(); + }); + }; + return () => { + watch.on = false; + watch.cancel(); + }; + } + + async #recovered(): Promise { + const recovering = (this.#recovering ??= this.#relaying + .recover() + .catch(String) + .finally(() => { + this.#recovering = undefined; + })); + const late = Promise.withResolvers(); + const cancel = this.#relaying.timer(recoveryAwaitedMs, () => { + late.resolve(stillComing); + }); + try { + return await Promise.race([recovering, late.promise]); + } finally { + cancel(); } - const sending = await this.#sent(message); - return sending.failed ? String(sending.failure) : undefined; } async #refused(message: unknown, problem: string): Promise { @@ -110,6 +240,7 @@ class McpRelay { this.#relaying.report(`Auto Runtime did not take a message: ${problem}`); return; } + this.#awaited.delete(message.id); await this.#agent.send({ jsonrpc: '2.0', id: message.id, error: { code: internalError, message: problem } }); } } diff --git a/packages/server/src/local-runtime/relay-recovery.test.ts b/packages/server/src/local-runtime/relay-recovery.test.ts new file mode 100644 index 000000000..f76728863 --- /dev/null +++ b/packages/server/src/local-runtime/relay-recovery.test.ts @@ -0,0 +1,152 @@ +import { setImmediate } from 'node:timers/promises'; + +import { describe, expect, it } from 'vitest'; + +import { + answeredWithStatus, + callATool, + cutOffMidAnswer, + droppedAfterTheRequest, + fakeRuntime, + listTools, + refusedAtConnect, + relayed, + type Timers, + type Turn, +} from '../testing/relays/fake-relays.ts'; +import { until } from '../testing/waiting/until.ts'; +import { answerAwaitedMs, mayHaveTakenEffect, recoveryAwaitedMs, stillComing } from './mcp-relay.ts'; + +async function firedOnceArmed(timers: Timers): Promise { + await until(() => { + expect(timers.armed(answerAwaitedMs)).toBe(1); + }); + timers.fire(answerAwaitedMs); +} + +describe('relaying an agent to a runtime that may have taken the request', () => { + it.each([ + ['whose answer was cut off', cutOffMidAnswer], + ['whose connection closed after it went out', droppedAfterTheRequest], + ['whose answer never came before its stream ended', 'drops the answer'], + ])( + 'makes sure the runtime runs, then sends a request %s again when sending it twice changes nothing', + async (_case, turn) => { + const runtime = fakeRuntime({ turns: [turn] }); + const { agent, next, recovered } = await relayed(() => runtime.transport); + + await agent.send(listTools); + + expect(await next()).toEqual({ jsonrpc: '2.0', id: 2, result: { tools: [] } }); + expect(recovered()).toBe(1); + expect(runtime.sends()).toBe(2); + }, + ); + + it.each([ + ['whose answer was cut off', cutOffMidAnswer], + ['whose answer never came before its stream ended', 'drops the answer'], + ])('never sends a tool call %s again, and says to check whether it took effect', async (_case, turn) => { + const runtime = fakeRuntime({ turns: [turn] }); + const { agent, next, recovered } = await relayed(() => runtime.transport); + + await agent.send(callATool); + + expect(await next()).toEqual({ jsonrpc: '2.0', id: 3, error: { code: -32_603, message: mayHaveTakenEffect } }); + expect(recovered()).toBe(1); + expect(runtime.sends()).toBe(1); + }); + + it.each, string]>([ + ['without one', new Error('Error POSTing to endpoint: busy'), 'Error: Error POSTing to endpoint: busy'], + ['of its own', answeredWithStatus(503), 'SdkHttpError: Error POSTing to endpoint: 503'], + ])( + 'answers a request the runtime answered with an error status %s with that error, and never sends it again', + async (_case, failure, problem) => { + const runtime = fakeRuntime({ turns: [failure] }); + const { agent, next, recovered } = await relayed(() => runtime.transport); + + await agent.send(listTools); + + expect(await next()).toEqual({ jsonrpc: '2.0', id: 2, error: { code: -32_603, message: problem } }); + expect(runtime.sends()).toBe(1); + expect(recovered()).toBe(0); + }, + ); +}); + +describe('a request the runtime does not start to answer', () => { + it('is given up once the runtime does not answer either, then sent again to the runtime made sure of', async () => { + const runtime = fakeRuntime({ turns: ['hangs'] }); + const { agent, next, recovered, timers } = await relayed(() => runtime.transport, { answering: [false] }); + + await agent.send(listTools); + await firedOnceArmed(timers); + + expect(await next()).toEqual({ jsonrpc: '2.0', id: 2, result: { tools: [] } }); + expect(recovered()).toBe(1); + }); + + it('is waited for as long as the runtime answers, as a long tool call is', async () => { + const runtime = fakeRuntime({ turns: ['resolves late'] }); + const { agent, next, recovered, timers } = await relayed(() => runtime.transport); + + await agent.send(callATool); + await firedOnceArmed(timers); + await firedOnceArmed(timers); + await until(() => { + expect(timers.armed(answerAwaitedMs)).toBe(1); + }); + runtime.resolveLate(); + + expect(await next()).toEqual({ jsonrpc: '2.0', id: 3, result: { tools: [] } }); + expect(recovered()).toBe(0); + await until(() => { + expect(timers.armed(answerAwaitedMs)).toBe(0); + }); + }); + + it('is not given up when its answer starts while the runtime is being asked whether it answers', async () => { + const runtime = fakeRuntime({ turns: ['resolves late'] }); + const asked = Promise.withResolvers(); + const { agent, next, recovered, timers } = await relayed(() => runtime.transport, { + answeringWith: () => asked.promise, + }); + + await agent.send(callATool); + await firedOnceArmed(timers); + runtime.resolveLate(); + const answer = await next(); + await setImmediate(); + asked.resolve(false); + await setImmediate(); + + expect(answer).toEqual({ jsonrpc: '2.0', id: 3, result: { tools: [] } }); + expect([recovered(), runtime.sends(), timers.armed(answerAwaitedMs)]).toEqual([0, 1, 0]); + }); +}); + +describe('a recovery that takes longer than an agent waits', () => { + it('answers the request that the runtime is still coming, and lets another request wait on the same recovery', async () => { + const recovering = Promise.withResolvers(); + const runtime = fakeRuntime({ turns: [refusedAtConnect, refusedAtConnect] }); + const { agent, next, recovered, timers } = await relayed(() => runtime.transport, { + recover: () => recovering.promise, + }); + + await agent.send(listTools); + await agent.send(callATool); + await until(() => { + expect(timers.armed(recoveryAwaitedMs)).toBe(2); + }); + timers.fire(recoveryAwaitedMs); + const answers = [await next(), await next()]; + recovering.resolve('Auto Runtime did not start on port 5555'); + + expect(answers).toEqual([ + { jsonrpc: '2.0', id: 2, error: { code: -32_603, message: stillComing } }, + { jsonrpc: '2.0', id: 3, error: { code: -32_603, message: stillComing } }, + ]); + expect(recovered()).toBe(1); + }); +}); diff --git a/packages/server/src/local-runtime/runtime-commands.test.ts b/packages/server/src/local-runtime/runtime-commands.test.ts index a361486b6..78c30f883 100644 --- a/packages/server/src/local-runtime/runtime-commands.test.ts +++ b/packages/server/src/local-runtime/runtime-commands.test.ts @@ -178,3 +178,30 @@ describe('what cannot be controlled from the command line', () => { expect(errors).toEqual(['Another program answers on port 3000, not Auto Runtime']); }); }); + +const guardedRuntime: RuntimeFetch = (address) => + Promise.resolve( + address.endsWith('/health') + ? Response.json({ status: 'ok' }) + : Response.json({ type: 'https://on.auto/problems/unauthenticated' }, { status: 401 }), + ); + +describe('what cannot be controlled from the command line, though it runs', () => { + it('says a runtime that asks for a credential cannot be controlled', async () => { + const { asking, errors } = askingWith(guardedRuntime); + + await expect(runRuntimeCommand('restart', 5555, asking)).resolves.toBe(1); + expect(errors).toEqual([ + 'The Auto Runtime on port 5555 asks for a credential, so it cannot be controlled from here: only one the auto-runtime command started in local mode can', + ]); + }); + + it('says a runtime that does not answer does not', async () => { + const { asking, errors } = askingWith(() => + Promise.reject(new DOMException('The operation was aborted due to timeout', 'TimeoutError')), + ); + + await expect(runRuntimeCommand('status', 5555, asking)).resolves.toBe(1); + expect(errors).toEqual(['The Auto Runtime on port 5555 does not answer']); + }); +}); diff --git a/packages/server/src/local-runtime/runtime-commands.ts b/packages/server/src/local-runtime/runtime-commands.ts index ff18b7125..291778bab 100644 --- a/packages/server/src/local-runtime/runtime-commands.ts +++ b/packages/server/src/local-runtime/runtime-commands.ts @@ -50,7 +50,7 @@ async function stopped(port: number, asking: RuntimeAsking): Promise { async function restarted(port: number, before: RunningStatus, asking: RuntimeAsking): Promise { const done = (await askedTo('restart', port, asking.fetch)) && - (await waitedFor(port, asking, throughARestart.attempts, startedAgainSince(before))); + (await waitedFor(port, asking, throughARestart.attempts, startedAgainSince(before))) !== undefined; if (!done) { asking.console.error(`Auto Runtime on port ${port} did not come back after it was asked to restart`); return 1; @@ -81,6 +81,9 @@ const refusals: Readonly, (port: numbe down: (port) => `No Auto Runtime is running on port ${port}`, uncontrolled: (port) => `The Auto Runtime on port ${port} cannot be controlled from here: only one the auto-runtime command started in local mode can`, + guarded: (port) => + `The Auto Runtime on port ${port} asks for a credential, so it cannot be controlled from here: only one the auto-runtime command started in local mode can`, + unresponsive: (port) => `The Auto Runtime on port ${port} does not answer`, foreign: (port) => `Another program answers on port ${port}, not Auto Runtime`, }; diff --git a/packages/server/src/local-runtime/runtime-probe.ts b/packages/server/src/local-runtime/runtime-probe.ts index 0e72000fa..e6c62a937 100644 --- a/packages/server/src/local-runtime/runtime-probe.ts +++ b/packages/server/src/local-runtime/runtime-probe.ts @@ -31,38 +31,56 @@ const isHealthy = Schema.is(HealthSchema); export type Probe = | { readonly state: 'running'; readonly status: RunningStatus } | { readonly state: 'uncontrolled' } + | { readonly state: 'guarded' } + | { readonly state: 'unresponsive' } | { readonly state: 'foreign' } | { readonly state: 'down' }; -const probeTimeoutMs = 2000; +const probeTimeoutMs = 1000; + +const unauthorized = 401; export function runtimeAddressOf(port: number): string { return `http://127.0.0.1:${port}`; } -async function bodyOf(fetch: RuntimeFetch, address: string): Promise { +interface Answer { + readonly status: number; + readonly body: unknown; +} + +async function answerOf(fetch: RuntimeFetch, address: string): Promise { const response = await fetch(address, { method: 'GET', timeoutMs: probeTimeoutMs }); - return response.ok ? response.json().catch(Function.constVoid) : undefined; + return { status: response.status, body: response.ok ? await response.json().catch(Function.constVoid) : undefined }; } async function answeringProbe(fetch: RuntimeFetch, port: number): Promise { - const status = await bodyOf(fetch, `${runtimeAddressOf(port)}/runtime`); - if (isRunningStatus(status)) { - return { state: 'running', status }; + const runtime = await answerOf(fetch, `${runtimeAddressOf(port)}/runtime`); + if (isRunningStatus(runtime.body)) { + return { state: 'running', status: runtime.body }; } - return isHealthy(await bodyOf(fetch, `${runtimeAddressOf(port)}/health`)) - ? { state: 'uncontrolled' } - : { state: 'foreign' }; + if (!isHealthy((await answerOf(fetch, `${runtimeAddressOf(port)}/health`)).body)) { + return { state: 'foreign' }; + } + return runtime.status === unauthorized ? { state: 'guarded' } : { state: 'uncontrolled' }; +} + +function isTimeout(failure: unknown): boolean { + return failure instanceof DOMException && failure.name === 'TimeoutError'; } export async function probeAt(port: number, fetch: RuntimeFetch): Promise { try { return await answeringProbe(fetch, port); - } catch { - return { state: 'down' }; + } catch (failure) { + return isTimeout(failure) ? { state: 'unresponsive' } : { state: 'down' }; } } +export function isAnswering(probe: Probe): boolean { + return probe.state === 'running' || probe.state === 'uncontrolled'; +} + export async function askedTo(action: 'stop' | 'restart', port: number, fetch: RuntimeFetch): Promise { try { const response = await fetch(`${runtimeAddressOf(port)}/runtime/${action}`, { @@ -87,12 +105,13 @@ export async function waitedFor( waiting: Waiting, attempts: number, isWhatWeWait: (probe: Probe) => boolean, -): Promise { +): Promise { if (attempts <= 0) { - return false; + return undefined; } - if (isWhatWeWait(await probeAt(port, waiting.fetch))) { - return true; + const probe = await probeAt(port, waiting.fetch); + if (isWhatWeWait(probe)) { + return probe; } await waiting.sleep(probeEveryMs); return waitedFor(port, waiting, attempts - 1, isWhatWeWait); @@ -105,5 +124,7 @@ function isDown(probe: Probe): boolean { } export async function stoppedWhenAsked(port: number, waiting: Waiting): Promise { - return (await askedTo('stop', port, waiting.fetch)) && (await waitedFor(port, waiting, stopAttempts, isDown)); + return ( + (await askedTo('stop', port, waiting.fetch)) && (await waitedFor(port, waiting, stopAttempts, isDown)) !== undefined + ); } diff --git a/packages/server/src/local-runtime/runtime-recovery.test.ts b/packages/server/src/local-runtime/runtime-recovery.test.ts new file mode 100644 index 000000000..88bd7ccca --- /dev/null +++ b/packages/server/src/local-runtime/runtime-recovery.test.ts @@ -0,0 +1,121 @@ +import { describe, expect, it } from 'vitest'; + +import { longestStopMs } from '../lifecycle/shutdown.ts'; +import { computerWith, startedForAnAgent, startingOn } from '../testing/relays/fake-computers.ts'; +import { briefly, impatient, runningRuntime, throughARestart } from './background-start.ts'; +import { probeEveryMs } from './runtime-probe.ts'; + +describe('making sure the runtime runs again for an agent whose request did not reach it', () => { + it('waits a moment for it to come back, then starts one without waiting through a restart', async () => { + const computer = computerWith(); + const starting = startingOn(computer); + + await expect(runningRuntime(5555, starting, briefly)).resolves.toBeUndefined(); + expect(computer.events()).toEqual(['locked', 'started', 'unlocked']); + expect(starting.naps()).toBe(briefly.attempts); + }); + + it('takes the one that comes back within that moment', async () => { + const computer = computerWith(); + + await expect(runningRuntime(5555, startingOn(computer, { comesBackAfter: 4 }), briefly)).resolves.toBeUndefined(); + expect(computer.events()).toEqual([]); + }); + + it('waits through a restart a runtime marked, as long as a restart can take', async () => { + const computer = computerWith(); + const starting = startingOn(computer, { restarting: true, comesBackAfter: throughARestart.attempts - 1 }); + + await expect(runningRuntime(5555, starting, briefly)).resolves.toBeUndefined(); + expect(computer.events()).toEqual([]); + }); +}); + +describe('a runtime on the port that asks for a credential', () => { + it('is not one this entry can reach, and is left running', async () => { + const computer = computerWith('guards'); + + await expect(runningRuntime(5555, startingOn(computer), briefly)).resolves.toBe( + 'Auto Runtime on port 5555 asks for a credential, which this MCP entry does not send; stop it, and this entry starts one, or connect your agent to http://localhost:5555/mcp over HTTP', + ); + expect(computer.events()).toEqual([]); + }); +}); + +describe('a runtime on the port that does not answer', () => { + it('is asked twice more, and taken when it answers', async () => { + const computer = computerWith('hangs'); + const starting = startingOn(computer, { comesBackAfter: 1 }); + + await expect(runningRuntime(5555, starting, briefly)).resolves.toBeUndefined(); + expect(computer.events()).toEqual([]); + }); + + it('is stopped and started again when an agent started it, as this relay saw it answer before', async () => { + const computer = computerWith(startedForAnAgent); + const starting = startingOn(computer); + await runningRuntime(5555, starting, impatient); + computer.answerWith('hangs'); + + await expect(runningRuntime(5555, starting, briefly)).resolves.toBeUndefined(); + expect(computer.events()).toEqual([ + 'noticed Auto Runtime 0.9.1 on port 5555, process 4242, did not answer, so it was stopped and another started', + 'SIGTERM 4242', + 'locked', + 'started', + 'unlocked', + ]); + }); +}); + +describe('a runtime on the port that does not answer, which this relay started', () => { + it('is killed when it outlives the time a stop can take, then started again', async () => { + const computer = computerWith('hangs'); + const starting = startingOn(computer, { sawBefore: startedForAnAgent, outlivesSigterm: true }); + + await expect(runningRuntime(5555, starting, briefly)).resolves.toBeUndefined(); + expect(computer.events().slice(1)).toEqual(['SIGTERM 4242', 'SIGKILL 4242', 'locked', 'started', 'unlocked']); + expect(starting.naps()).toBeGreaterThanOrEqual(longestStopMs / probeEveryMs); + }); + + it('is remembered from the start of the one this relay started, to be stopped once it hangs', async () => { + const computer = computerWith(); + const starting = startingOn(computer); + await runningRuntime(5555, starting, impatient); + computer.answerWith('hangs'); + + await runningRuntime(5555, starting, briefly); + + expect(computer.events()).toEqual([ + 'locked', + 'started', + 'unlocked', + 'noticed Auto Runtime 0.9.1 on port 5555, process 4242, did not answer, so it was stopped and another started', + 'SIGTERM 4242', + 'locked', + 'started', + 'unlocked', + ]); + }); +}); + +describe('a runtime on the port that does not answer, which this relay did not start', () => { + it('is left alone when it runs in a terminal, naming its process', async () => { + const computer = computerWith('hangs'); + const starting = startingOn(computer, { sawBefore: { ...startedForAnAgent, started_by: 'terminal' } }); + + await expect(runningRuntime(5555, starting, briefly)).resolves.toBe( + 'Auto Runtime on port 5555, which runs in a terminal as process 4242, does not answer; stop it there, and this MCP entry starts another', + ); + expect(computer.events()).toEqual([]); + }); + + it('is left alone when this relay never saw it answer', async () => { + const computer = computerWith('hangs'); + + await expect(runningRuntime(5555, startingOn(computer), briefly)).resolves.toBe( + 'The program on port 5555 does not answer; stop it, or set PORT to a free port, and this MCP entry starts Auto Runtime', + ); + expect(computer.events()).toEqual([]); + }); +}); diff --git a/packages/server/src/testing/relays/fake-computers.ts b/packages/server/src/testing/relays/fake-computers.ts new file mode 100644 index 000000000..2b40cc651 --- /dev/null +++ b/packages/server/src/testing/relays/fake-computers.ts @@ -0,0 +1,138 @@ +import { sightingsOfARelay, type BackgroundStarting } from '../../local-runtime/background-start.ts'; +import type { RunningStatus, RuntimeFetch } from '../../local-runtime/runtime-probe.ts'; + +export const startedForAnAgent: RunningStatus = { + version: '0.9.1', + pid: 4242, + started_at: '2026-10-14T09:41:07.000Z', + started_by: 'agent', + directory: '/Users/ada/.auto-runtime', + log_file: '/Users/ada/.auto-runtime/runtime.log', + runs_in_flight: 0, +}; + +export type Answering = object | 'hangs' | 'guards' | undefined; + +export interface Computer { + readonly answering: () => Answering; + readonly answerWith: (answer?: Answering) => void; + readonly record: (event: string) => void; + readonly events: () => readonly string[]; +} + +export function computerWith(first?: Answering): Computer { + let answering = first; + const events: string[] = []; + return { + answering: () => answering, + answerWith: (answer) => { + answering = answer; + }, + record: (event) => { + events.push(event); + }, + events: () => [...events], + }; +} + +const timedOut = new DOMException('The operation was aborted due to timeout', 'TimeoutError'); + +function guarding(address: string): Response { + return address.endsWith('/health') + ? Response.json({ status: 'ok' }) + : Response.json({ type: 'https://on.auto/problems/unauthenticated' }, { status: 401 }); +} + +function fetchOn(computer: Computer): RuntimeFetch { + return (address, { method }) => { + const answering = computer.answering(); + if (answering === undefined) { + return Promise.reject(new TypeError('fetch failed')); + } + if (answering === 'hangs') { + return Promise.reject(timedOut); + } + if (answering === 'guards') { + return Promise.resolve(guarding(address)); + } + if (method === 'POST') { + computer.record(`asked ${address.slice(address.lastIndexOf('/') + 1)}`); + computer.answerWith(); + return Promise.resolve(Response.json({ status: 'stopping' }, { status: 202 })); + } + return Promise.resolve(Response.json(answering)); + }; +} + +export interface Options { + readonly version?: string; + readonly startsAs?: Answering; + readonly comesBackAfter?: number; + readonly lockTaken?: boolean; + readonly restarting?: boolean; + readonly sawBefore?: RunningStatus; + readonly outlivesSigterm?: boolean; +} + +export interface Starting extends BackgroundStarting { + readonly naps: () => number; +} + +function processesOn(computer: Computer, options: Options): BackgroundStarting['processes'] { + let alive = true; + return { + signal: (pid, signal) => { + computer.record(`${signal} ${pid}`); + if (signal === 'SIGKILL' || options.outlivesSigterm !== true) { + alive = false; + computer.answerWith(); + } + return true; + }, + alive: () => alive, + }; +} + +function sightingsBefore(options: Options): BackgroundStarting['sightings'] { + const sightings = sightingsOfARelay(); + if (options.sawBefore !== undefined) { + sightings.saw(options.sawBefore); + } + return sightings; +} + +export function startingOn(computer: Computer, options: Options = {}): Starting { + let sleeps = 0; + return { + version: options.version ?? '0.9.1', + logFile: '/Users/ada/.auto-runtime/runtime.log', + fetch: fetchOn(computer), + sleep: () => { + sleeps += 1; + if (options.comesBackAfter === sleeps) { + computer.answerWith(startedForAnAgent); + } + return Promise.resolve(); + }, + naps: () => sleeps, + startInBackground: () => { + computer.record('started'); + computer.answerWith('startsAs' in options ? options.startsAs : startedForAnAgent); + }, + startingLock: { + take: () => { + computer.record('locked'); + return options.lockTaken !== true; + }, + release: () => { + computer.record('unlocked'); + }, + }, + notice: (line) => { + computer.record(`noticed ${line}`); + }, + restartUnderWay: () => options.restarting === true, + processes: processesOn(computer, options), + sightings: sightingsBefore(options), + }; +} diff --git a/packages/server/src/testing/relays/fake-relays.ts b/packages/server/src/testing/relays/fake-relays.ts new file mode 100644 index 000000000..ae465d1a6 --- /dev/null +++ b/packages/server/src/testing/relays/fake-relays.ts @@ -0,0 +1,255 @@ +import { SdkErrorCode, SdkHttpError } from '@modelcontextprotocol/client'; +import { InMemoryTransport, parseJSONRPCMessage, type Transport } from '@modelcontextprotocol/server'; + +import { relay, type Timer } from '../../local-runtime/mcp-relay.ts'; + +export const listTools = parseJSONRPCMessage({ jsonrpc: '2.0', id: 2, method: 'tools/list', params: {} }); + +export const callATool = parseJSONRPCMessage({ + jsonrpc: '2.0', + id: 3, + method: 'tools/call', + params: { name: 'run_definition', arguments: {} }, +}); + +export const initialize = parseJSONRPCMessage({ + jsonrpc: '2.0', + id: 1, + method: 'initialize', + params: { protocolVersion: '2025-11-25', capabilities: {}, clientInfo: { name: 'agent', version: '1.0.0' } }, +}); + +export const initialized = parseJSONRPCMessage({ jsonrpc: '2.0', method: 'notifications/initialized' }); + +export type Reply = (message: unknown) => unknown; + +export function idOf(message: unknown): unknown { + return typeof message === 'object' && message !== null && 'id' in message ? message.id : undefined; +} + +const echoTools: Reply = (message) => ({ jsonrpc: '2.0', id: idOf(message), result: { tools: [] } }); + +export type Turn = Readonly | 'hangs' | 'drops the answer' | 'resolves late'; + +export interface FakeRuntimeOptions { + readonly reply?: Reply; + readonly turns?: readonly Turn[]; + readonly negotiates?: boolean; +} + +export interface FakeRuntime { + readonly transport: Transport; + readonly received: unknown[]; + readonly negotiated: string[]; + readonly closed: () => boolean; + readonly sends: () => number; + readonly resolveLate: () => void; +} + +function hungUntilGivenUp(signal: Readonly | undefined): Promise { + return new Promise((_resolve, reject) => { + signal?.addEventListener('abort', () => { + reject(new DOMException('This operation was aborted', 'AbortError')); + }); + }); +} + +interface SendOptions { + readonly requestSignal?: Readonly | undefined; + readonly onRequestStreamEnd?: (() => void) | undefined; +} + +interface Turning { + readonly answered: (message: unknown) => Promise; + readonly late: Promise; +} + +function streamEndedFor(options: SendOptions | undefined): void { + queueMicrotask(() => { + options?.onRequestStreamEnd?.(); + }); +} + +function turned(turn: Turn, message: unknown, options: SendOptions | undefined, turning: Turning): Promise { + if (turn === 'hangs') { + return hungUntilGivenUp(options?.requestSignal); + } + if (turn === 'resolves late') { + return turning.late.then(() => turning.answered(message)); + } + if (turn === 'drops the answer') { + streamEndedFor(options); + return Promise.resolve(); + } + return Promise.reject(turn); +} + +function negotiating(negotiates: boolean, negotiate: (version: string) => void): Partial { + return negotiates ? { setProtocolVersion: negotiate } : {}; +} + +export function fakeRuntime({ + reply = echoTools, + turns = [], + negotiates = true, +}: FakeRuntimeOptions = {}): FakeRuntime { + const received: unknown[] = []; + const negotiated: string[] = []; + const coming = [...turns]; + const late = Promise.withResolvers(); + const state = { closed: false, sends: 0 }; + const answered = (message: unknown): Promise => { + received.push(message); + const answer = reply(message); + if (answer !== undefined) { + queueMicrotask(() => transport.onmessage?.(parseJSONRPCMessage(answer))); + } + return Promise.resolve(); + }; + const transport: Transport = { + start: () => Promise.resolve(), + send: (message: unknown, options?: SendOptions) => { + state.sends += 1; + const turn = coming.shift(); + if (turn !== undefined) { + return turned(turn, message, options, { answered, late: late.promise }); + } + const sent = answered(message); + streamEndedFor(options); + return sent; + }, + close: () => { + state.closed = true; + return Promise.resolve(); + }, + ...negotiating(negotiates, (version) => { + negotiated.push(version); + }), + }; + return { + transport, + received, + negotiated, + closed: () => state.closed, + sends: () => state.sends, + resolveLate: () => { + late.resolve(); + }, + }; +} + +export function fakeRuntimeEnd(options: FakeRuntimeOptions = {}): () => Transport { + const { transport } = fakeRuntime(options); + return () => transport; +} + +export const refusedAtConnect = new TypeError('fetch failed', { + cause: Object.assign(new Error('connect ECONNREFUSED 127.0.0.1:5555'), { code: 'ECONNREFUSED', syscall: 'connect' }), +}); + +export function answeredWithStatus(status: number): SdkHttpError { + return new SdkHttpError(SdkErrorCode.ClientHttpNotImplemented, `Error POSTing to endpoint: ${status}`, { status }); +} + +export const cutOffMidAnswer = new TypeError('terminated', { + cause: Object.assign(new Error('other side closed'), { code: 'UND_ERR_SOCKET' }), +}); + +export const droppedAfterTheRequest = new TypeError('fetch failed', { + cause: Object.assign(new Error('other side closed'), { code: 'UND_ERR_SOCKET' }), +}); + +export interface Timers { + readonly timer: Timer; + readonly fire: (milliseconds: number) => void; + readonly armed: (milliseconds: number) => number; +} + +interface Armed { + readonly milliseconds: number; + readonly fires: () => void; +} + +function fakeTimers(): Timers { + const armed = new Set(); + const due = (milliseconds: number): readonly Armed[] => + [...armed].filter((timer) => timer.milliseconds === milliseconds); + return { + timer: (milliseconds, fires) => { + const timer = { milliseconds, fires }; + armed.add(timer); + return () => { + armed.delete(timer); + }; + }, + fire: (milliseconds) => { + for (const timer of due(milliseconds)) { + armed.delete(timer); + timer.fires(); + } + }, + armed: (milliseconds) => due(milliseconds).length, + }; +} + +export interface Recovery { + readonly problem?: string; + readonly answering?: readonly boolean[]; + readonly answeringWith?: () => Promise; + readonly recover?: () => Promise; +} + +export interface Relayed { + readonly agent: Transport; + readonly relayEnd: Transport; + readonly reported: string[]; + readonly recovered: () => number; + readonly timers: Timers; + readonly ended: Promise; + readonly next: () => Promise; +} + +function answersOf(agent: Readonly): () => Promise { + const answers: unknown[] = []; + const waiting: ((message: unknown) => void)[] = []; + Object.assign(agent, { + onmessage: (message: unknown) => { + const waiter = waiting.shift(); + if (waiter === undefined) { + answers.push(message); + } else { + waiter(message); + } + }, + }); + return () => + answers.length > 0 + ? Promise.resolve(answers.shift()) + : new Promise((resolve) => { + waiting.push(resolve); + }); +} + +export async function relayed(runtimeEnd: () => Transport, recovery: Recovery = {}): Promise { + const [agent, relayEnd] = InMemoryTransport.createLinkedPair(); + const reported: string[] = []; + const answering = [...(recovery.answering ?? [])]; + const timers = fakeTimers(); + const next = answersOf(agent); + const counted = { recoveries: 0 }; + const ended = relay({ + agent: () => relayEnd, + runtime: runtimeEnd, + recover: () => { + counted.recoveries += 1; + return recovery.recover?.() ?? Promise.resolve(recovery.problem); + }, + answering: recovery.answeringWith ?? (() => Promise.resolve(answering.shift() ?? true)), + timer: timers.timer, + report: (line) => { + reported.push(line); + }, + }); + await agent.start(); + return { agent, relayEnd, reported, recovered: () => counted.recoveries, timers, ended, next }; +} diff --git a/packages/server/src/testing/relays/relayed-processes.ts b/packages/server/src/testing/relays/relayed-processes.ts new file mode 100644 index 000000000..768cec7f2 --- /dev/null +++ b/packages/server/src/testing/relays/relayed-processes.ts @@ -0,0 +1,57 @@ +import { expect } from 'vitest'; + +import { aPortNothingListensOn } from '../processes/free-port.ts'; +import { + homeOfTheTest, + runtimeOnThePortEndedWithTheTest, + type RuntimeCommandRun, + type RuntimeOnThePort, +} from '../processes/runtime-command.ts'; +import { until } from '../waiting/until.ts'; + +export const initializeLine = { + jsonrpc: '2.0', + id: 1, + method: 'initialize', + params: { protocolVersion: '2025-11-25', capabilities: {}, clientInfo: { name: 'agent', version: '1.0.0' } }, +}; + +export const initializedLine = { jsonrpc: '2.0', method: 'notifications/initialized' }; + +function answersIn(stdout: string): readonly unknown[] { + return stdout + .split('\n') + .filter((line) => line !== '') + .map((line): unknown => JSON.parse(line)); +} + +export function answerWithId(answers: readonly unknown[], id: number): unknown { + return answers.find((answer) => typeof answer === 'object' && answer !== null && 'id' in answer && answer.id === id); +} + +export function lineOf(message: object): string { + return `${JSON.stringify(message)}\n`; +} + +export async function answered(relay: Readonly, count: number): Promise { + await until(() => { + expect(answersIn(relay.stdout())).toHaveLength(count); + }); + return answersIn(relay.stdout()); +} + +export function readiesIn(text: string): number { + return text.split('Press Ctrl+C to stop.').length - 1; +} + +export interface RelayPlace { + readonly home: string; + readonly port: number; + readonly runtimes: RuntimeOnThePort; +} + +export async function placeOfTheTest(prefix: string): Promise { + const home = homeOfTheTest(prefix); + const port = await aPortNothingListensOn(); + return { home, port, runtimes: runtimeOnThePortEndedWithTheTest(home, port) }; +}