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

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions docs/agent-managed-compute/reliability.md
Original file line number Diff line number Diff line change
Expand Up @@ -231,6 +231,14 @@ Omit `coordination.port` to allocate a separate port for each concurrent manager
Runtime does not provision a proxy or tunnel.
Remote public endpoints require HTTPS.

Before provider admission, Runtime checks each configured public endpoint with authenticated `initialize` and `tools/list` requests.
The returned grants must match that manager's exact tool names.
The check takes at most 10 seconds, or `coordination.requestTimeoutMs` when shorter, and stops on manager cancellation.
Responses are limited to 1 MiB independently of the incoming request limit, and redirects are refused.
Failure closes the listener and reports a credential-free cause before inference starts.
This verifies the operator's public route; it does not establish reachability from the provider's network.
Omitting `coordination.publicUrl` preserves local-only startup without this network check.

For same-host coordinator restart, configure `authentication.signingKeys` with an active key ID and a secret key map.
Keep the public URL, run ID, actor ID, tool grants, and verification key stable until the retained credential expires.
Stable keys support resumed coordination within the original scope deadline and any explicit credential expiry.
Expand Down
1 change: 1 addition & 0 deletions scripts/check-model-execution-boundary.mjs
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@ const dynamicNonModelFetchOwners = new Set([
'src/platform/auth.ts',
'src/platform/integrations.ts',
'src/runtime/mcp-environment.ts',
'src/runtime/supervise/coordination-preflight.ts',
'bench/src/research-shot.ts',
'bench/src/search-tool.ts',
'bench/src/benchmarks/aec-bench.ts',
Expand Down
8 changes: 8 additions & 0 deletions scripts/check-model-execution-boundary.test.mjs
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,14 @@ describe('model execution boundary source check', () => {
expect(violations[0]?.location).toBe('3:7')
})

it('permits dynamic MCP preflight routes without admitting named model endpoints', () => {
const path = 'src/runtime/supervise/coordination-preflight.ts'
expect(checkJavaScript(path, 'await fetch(input.url)')).toEqual([])
expect(
checkJavaScript(path, "await fetch('https://router.tangle.tools/v1/chat/completions')"),
).toHaveLength(1)
})

it('rejects qualified and aliased global fetch calls', () => {
expect(
checkJavaScript(
Expand Down
17 changes: 16 additions & 1 deletion scripts/release-cohort.test.mjs
Original file line number Diff line number Diff line change
@@ -1,7 +1,9 @@
import { mkdtempSync, rmSync, writeFileSync } from 'node:fs'
import { mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { afterEach, describe, expect, it } from 'vitest'
import { parse } from 'yaml'
import { rangeAdmits } from './lib/packed-package-test.mjs'
import {
assertReleaseCohortArtifacts,
readReleaseCohort,
Expand All @@ -17,6 +19,19 @@ afterEach(() => {
})

describe('release cohort', () => {
it.each(Object.values(readReleaseCohort().packages))(
'selects $name@$version within its workspace catalog range',
({ name, version }) => {
const { catalog } = parse(
readFileSync(new URL('../pnpm-workspace.yaml', import.meta.url), 'utf8'),
)
expect(
rangeAdmits(catalog[name], version),
`${name}@${version} must be admitted by catalog range ${catalog[name]}`,
).toBe(true)
},
)

it('reads one exact source identity for every first-party dependency', () => {
const cohort = readReleaseCohort()
expect(Object.keys(cohort.packages)).toEqual([
Expand Down
2 changes: 1 addition & 1 deletion src/mcp/tool-server.ts
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@ import type { JsonRpcMessage, JsonRpcResponse, McpToolDescriptor, McpTransport }

export type { JsonRpcMessage, JsonRpcResponse, McpToolDescriptor, McpTransport } from './protocol'

const PROTOCOL_VERSION = '2024-11-05'
export const PROTOCOL_VERSION = '2024-11-05'

/** @experimental */
export interface StdioToolServerOptions {
Expand Down
16 changes: 15 additions & 1 deletion src/runtime/supervise/coordination-mcp.ts
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,7 @@ import {
coordinationHttpHandler,
coordinationHttpLimits,
} from './coordination-http'
import { preflightPublicCoordination } from './coordination-preflight'
import { singleFlightTools } from './single-flight-tools'

export type { CoordinationHttpAudit, CoordinationHttpOptions } from './coordination-http'
Expand Down Expand Up @@ -567,8 +568,21 @@ export async function serveCoordinationMcp(
audiences.add(publicAddress.host)
if (host === '0.0.0.0' || host === '::') audiences.add(`127.0.0.1:${port}`)
paths.add(publicAddress.pathname)
if (configured !== undefined) {
await preflightPublicCoordination({
url,
headers,
signal: opts.scope.signal,
requestTimeoutMs,
toolNames: selectedNames,
})
}
} catch (error) {
await new Promise<void>((resolve) => server.close(() => resolve()))
closed = true
await new Promise<void>((resolve) => {
server.close(() => resolve())
server.closeAllConnections()
})
throw error
}

Expand Down
150 changes: 150 additions & 0 deletions src/runtime/supervise/coordination-preflight.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,150 @@
import { createServer, type IncomingMessage, type ServerResponse } from 'node:http'
import { afterEach, describe, expect, it } from 'vitest'
import { preflightPublicCoordination } from './coordination-preflight'

const closes: Array<() => Promise<void>> = []
afterEach(async () => {
await Promise.all(closes.splice(0).map((close) => close()))
})

async function endpoint(handler: (request: IncomingMessage, response: ServerResponse) => void) {
const server = createServer(handler)
await new Promise<void>((resolve) => server.listen(0, '127.0.0.1', resolve))
closes.push(
() =>
new Promise((resolve) => {
server.closeAllConnections()
server.close(() => resolve())
}),
)
const address = server.address()
if (!address || typeof address === 'string') throw new Error('test endpoint did not bind')
return `http://127.0.0.1:${address.port}`
}

function inspect(url: string, signal = new AbortController().signal, requestTimeoutMs = 1000) {
return preflightPublicCoordination({
url,
headers: { Authorization: 'Bearer private-test-credential' },
signal,
requestTimeoutMs,
toolNames: ['stop'],
})
}

describe('public coordination preflight', () => {
it('performs authenticated initialization and checks the exact granted tool names', async () => {
const methods: string[] = []
const url = await endpoint((request, response) => {
expect(request.headers.authorization).toBe('Bearer private-test-credential')
let body = ''
request.on('data', (chunk) => {
body += String(chunk)
})
request.on('end', () => {
const rpc = JSON.parse(body)
methods.push(rpc.method)
response.end(
JSON.stringify({
jsonrpc: '2.0',
id: rpc.id,
result:
rpc.method === 'initialize'
? {
protocolVersion: '2024-11-05',
serverInfo: { name: 'coordination' },
capabilities: { tools: {} },
}
: { tools: [{ name: 'stop' }] },
}),
)
})
})
await inspect(url)
expect(methods).toEqual(['initialize', 'tools/list'])
})

it.each([401, 403, 404, 503])(
'refuses HTTP %s without retaining echoed secrets',
async (status) => {
const url = await endpoint((_request, response) => {
response.writeHead(status).end('private-test-credential private-response')
})
await expect(inspect(url)).rejects.toThrow(
`coordination public endpoint preflight failed: HTTP ${status}`,
)
try {
await inspect(url)
} catch (error) {
expect(String(error)).not.toContain('private-')
expect(String(error)).not.toContain(url)
expect(error).not.toHaveProperty('cause')
}
},
)

it('refuses redirects instead of forwarding actor credentials to another endpoint', async () => {
let redirected = 0
const target = await endpoint((_request, response) => {
redirected++
response.end('{}')
})
const url = await endpoint((_request, response) =>
response.writeHead(307, { location: target }).end(),
)
await expect(inspect(url)).rejects.toThrow('HTTP 307')
expect(redirected).toBe(0)
})

it.each(['not-json', '{}', '{"jsonrpc":"2.0","id":"other","result":{}}'])(
'refuses malformed MCP response %s',
async (body) => {
const url = await endpoint((_request, response) => response.end(body))
await expect(inspect(url)).rejects.toThrow('invalid MCP response')
},
)

it('bounds a hanging response body and distinguishes timeout from cancellation', async () => {
const url = await endpoint((_request, response) => {
response.writeHead(200)
response.write('{')
})
await expect(inspect(url, undefined, 40)).rejects.toThrow('timed out')
const controller = new AbortController()
const pending = inspect(url, controller.signal)
controller.abort('private-test-credential in a caller reason')
await expect(pending).rejects.toThrow('preflight failed: cancelled')
})

it('bounds response bytes before decoding or retaining an upstream body', async () => {
const url = await endpoint((_request, response) => response.end('x'.repeat(1024 * 1024 + 1)))
await expect(inspect(url)).rejects.toThrow('response too large')
})

it('rejects incomplete or extra tool grants', async () => {
const url = await endpoint((request, response) => {
let body = ''
request.on('data', (chunk) => {
body += String(chunk)
})
request.on('end', () => {
const rpc = JSON.parse(body)
response.end(
JSON.stringify({
jsonrpc: '2.0',
id: rpc.id,
result:
rpc.method === 'initialize'
? {
protocolVersion: '2024-11-05',
serverInfo: { name: 'coordination' },
capabilities: { tools: {} },
}
: { tools: [{ name: 'spawn_worker' }] },
}),
)
})
})
await expect(inspect(url)).rejects.toThrow('coordination tool grants differ')
})
})
117 changes: 117 additions & 0 deletions src/runtime/supervise/coordination-preflight.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,117 @@
import { ConfigError } from '../../errors'
import { PROTOCOL_VERSION } from '../../mcp/tool-server'
import { linkAbort, runAbortable } from './abortable'

const MAX_RESPONSE_BYTES = 1024 * 1024

class CoordinationPreflightError extends ConfigError {
constructor(reason: string) {
super(`coordination public endpoint preflight failed: ${reason}`)
}
}

function record(value: unknown): value is Record<string, unknown> {
return typeof value === 'object' && value !== null && !Array.isArray(value)
}

/** Check the operator's route before it becomes a provider attachment. This does not prove cloud egress. */
export async function preflightPublicCoordination(input: {
url: string
headers: Readonly<Record<string, string>>
signal: AbortSignal
requestTimeoutMs: number
toolNames: readonly string[]
}): Promise<void> {
const deadline = new AbortController()
const linked = linkAbort(input.signal, deadline.signal)
const timer = setTimeout(() => deadline.abort(), Math.min(input.requestTimeoutMs, 10_000))
timer.unref()
const rpc = async (method: string, params?: unknown): Promise<Record<string, unknown>> => {
const response = await fetch(input.url, {
method: 'POST',
headers: { ...input.headers, 'content-type': 'application/json', accept: 'application/json' },
body: JSON.stringify({ jsonrpc: '2.0', id: method, method, ...(params ? { params } : {}) }),
redirect: 'manual',
signal: linked.signal,
})
if (response.status !== 200) {
void response.body?.cancel().catch(() => undefined)
throw new CoordinationPreflightError(`HTTP ${response.status}`)
}
const reader = response.body?.getReader()
if (!reader) throw new CoordinationPreflightError('invalid MCP response')
let size = 0
const chunks: Uint8Array[] = []
try {
for (;;) {
const part = await reader.read()
if (part.done) break
size += part.value.byteLength
if (size > MAX_RESPONSE_BYTES) {
void reader.cancel().catch(() => undefined)
throw new CoordinationPreflightError('response too large')
}
chunks.push(part.value)
}
} finally {
reader.releaseLock()
}
let body: unknown
try {
body = JSON.parse(new TextDecoder('utf-8', { fatal: true }).decode(Buffer.concat(chunks)))
} catch {
throw new CoordinationPreflightError('invalid MCP response')
}
if (
!record(body) ||
body.jsonrpc !== '2.0' ||
body.id !== method ||
body.error ||
!record(body.result)
) {
throw new CoordinationPreflightError('invalid MCP response')
}
return body.result
}
try {
await runAbortable(
async () => {
const initialized = await rpc('initialize', {
protocolVersion: PROTOCOL_VERSION,
capabilities: {},
clientInfo: { name: 'agent-runtime-coordination-preflight', version: '1' },
})
if (
initialized.protocolVersion !== PROTOCOL_VERSION ||
!record(initialized.serverInfo) ||
initialized.serverInfo.name !== 'coordination' ||
!record(initialized.capabilities) ||
!record(initialized.capabilities.tools)
)
throw new CoordinationPreflightError('invalid MCP initialization')
const listing = await rpc('tools/list')
if (
!Array.isArray(listing.tools) ||
listing.tools.some((tool) => !record(tool) || typeof tool.name !== 'string')
) {
throw new CoordinationPreflightError('invalid MCP tool list')
}
const names = listing.tools.map((tool) => tool.name).sort()
if (JSON.stringify(names) !== JSON.stringify([...input.toolNames].sort())) {
throw new CoordinationPreflightError('coordination tool grants differ')
}
},
linked.signal,
'coordination public endpoint preflight cancelled',
)
} catch (error) {
// Neither a response body nor an upstream exception is safe to retain: both can echo credentials.
if (input.signal.aborted) throw new CoordinationPreflightError('cancelled')
if (deadline.signal.aborted) throw new CoordinationPreflightError('timed out')
if (error instanceof CoordinationPreflightError) throw error
throw new CoordinationPreflightError('transport unavailable')
} finally {
clearTimeout(timer)
linked.release()
}
}
Loading
Loading