diff --git a/.changeset/fresh-event-ipc.md b/.changeset/fresh-event-ipc.md new file mode 100644 index 000000000..e04039c51 --- /dev/null +++ b/.changeset/fresh-event-ipc.md @@ -0,0 +1,8 @@ +--- +"agent-bundle": patch +--- + +Cancel shared event renders when their IPC client disconnects, decode split +UTF-8 request bytes incrementally, refuse to unlink live runtime sockets, +preserve event-route timeout milliseconds until native host projection, and +assign a fresh diagnostic to missing shared runtime hosts. diff --git a/docs/diagnostics.md b/docs/diagnostics.md index 34c18faca..438b78650 100644 --- a/docs/diagnostics.md +++ b/docs/diagnostics.md @@ -118,7 +118,7 @@ simply not been built yet is a validation **warning** that only | `AB4749` | error (build) | A payload directory overlaps the artifact `--output` root. | | `AB4750` | info | A payload is older than the newest project source file and may be stale; rerun the project's own build if so. | -## Route graph (`AB4800`–`AB4816`) +## Route graph (`AB4800`–`AB4817`) The route-graph compiler discovers conventional route modules (`src/mcp//{tools,resources,prompts,apps}/*`, `src/events/*/*`, @@ -227,6 +227,7 @@ schema constants), unions, nested objects, transforms, coercions — raises | `AB4814` | error | A CLI route's `inputSchema` leaves the bounded argv grammar (the message names the offending construct and position), a key projects onto a reserved or duplicate option name, a required boolean has no flag expression, or `config.positionals` violates the positional policy. | | `AB4815` | error | A CLI route does not satisfy the routed command contract: missing named `inputSchema`/`resultSchema` exports, a default export that is not an async function, or malformed `config.description`/`aliases`/`exitCode` fields. | | `AB4816` | retired | The stage-2 rendered-command gate. Rendered command routes render through the dispatcher since #102 stage 3; the code is never reused. | +| `AB4817` | error | An event route requires the shared runtime for a target, but no generated MCP entry hosts that runtime and the route does not allow standalone fallback. | ## Development package build (`AB7103`) diff --git a/docs/entry-conventions.md b/docs/entry-conventions.md index 5411eb04c..012ef96fd 100644 --- a/docs/entry-conventions.md +++ b/docs/entry-conventions.md @@ -137,7 +137,7 @@ export default async function inspect({ input, signal }: CliRouteProps.js` with the shebang and executable bit through the same Rslib synthesis as every other bin. At run time the shell resolves the diff --git a/packages/agent-bundle/src/adapters/hook-contract.ts b/packages/agent-bundle/src/adapters/hook-contract.ts index f912fad52..02e3d7891 100644 --- a/packages/agent-bundle/src/adapters/hook-contract.ts +++ b/packages/agent-bundle/src/adapters/hook-contract.ts @@ -23,6 +23,8 @@ export interface TargetHookWrapper { } export interface TargetHookEntry extends TargetHookWrapper { + /** Timeout projected into the native host's seconds unit. */ + readonly timeout?: number; readonly virtualSource: string; } @@ -316,7 +318,7 @@ const eventRouteHookWrapperSource = ( `const target = ${JSON.stringify(entry.target)};`, `const runtimeMode = ${JSON.stringify(route.runtime)};`, `const fallbackMode = ${JSON.stringify(route.fallback)};`, - `const timeoutMs = ${String((entry.hook.timeout ?? 5) * 1_000)};`, + `const timeoutMs = ${String(entry.hook.timeoutMs ?? 5_000)};`, "const endpointId = `${artifactEpoch}:${target}:${dirname(dirname(resolve(process.argv[1])))}`;", '', 'const isRecord = (value) => typeof value === "object" && value !== null && !Array.isArray(value);', @@ -696,16 +698,17 @@ export const planHooks = ( const prebuilt = hook.prebuiltPath !== undefined; const relativePath = hook.prebuiltPath ?? contract.wrapperPath(hook); const command = generatedHookCommand(contract, relativePath, prebuilt ? hook.args ?? [] : []); + const timeout = hook.timeoutMs === undefined ? undefined : Math.ceil(hook.timeoutMs / 1_000); const entryInput: TargetHookDocumentEntryInput = { command, ...(matcher === undefined ? {} : { matcher }), - ...(hook.timeout === undefined ? {} : { timeout: hook.timeout }), + ...(timeout === undefined ? {} : { timeout }), }; const group = contract.documentEntry === undefined ? { hooks: [{ command, - ...(hook.timeout === undefined ? {} : { timeout: hook.timeout }), + ...(timeout === undefined ? {} : { timeout }), type: 'command', }], ...(matcher === undefined ? {} : { matcher }), @@ -721,6 +724,7 @@ export const planHooks = ( ...(matcher === undefined ? {} : { nativeMatcher: matcher }), relativePath, target, + ...(timeout === undefined ? {} : { timeout }), }; hookEntries.push({ ...wrapper, diff --git a/packages/agent-bundle/src/build/entries.ts b/packages/agent-bundle/src/build/entries.ts index 187cb1f7f..ec45ebe8d 100644 --- a/packages/agent-bundle/src/build/entries.ts +++ b/packages/agent-bundle/src/build/entries.ts @@ -442,7 +442,7 @@ export const planCompiledHooks = ( source: entry.hook.source, sourceInputs: Object.freeze([entry.hook.provenance.sourcePath, entry.hook.source]), target: entry.target, - ...(entry.hook.timeout === undefined ? {} : { timeout: entry.hook.timeout }), + ...(entry.timeout === undefined ? {} : { timeout: entry.timeout }), }))); export const compileHooks = async ( diff --git a/packages/agent-bundle/src/config/normalize.ts b/packages/agent-bundle/src/config/normalize.ts index 2cac01ab4..141b7ad99 100644 --- a/packages/agent-bundle/src/config/normalize.ts +++ b/packages/agent-bundle/src/config/normalize.ts @@ -404,6 +404,7 @@ const normalizeHook = ( const targets = sortedUnique(entry.targets ?? defaultTargets); const nativeTools = normalizeNativeHookTools(entry.tools ?? [], registry); const timeout = entry.timeout; + const timeoutMs = timeout === undefined ? undefined : timeout * 1_000; const identity = { ...(args === undefined || args.length === 0 ? {} : { args }), event, @@ -429,7 +430,7 @@ const normalizeHook = ( provenance: prebuilt ? { kind: 'prebuilt', sourcePath: provenance.sourcePath } : { ...provenance }, source, targets, - ...(timeout === undefined ? {} : { timeout }), + ...(timeoutMs === undefined ? {} : { timeoutMs }), tools, }; }; @@ -470,9 +471,9 @@ const normalizeHooks = ( .filter((tool): tool is CanonicalHookTool => typeof tool === 'string' && knownHookTools.has(tool as CanonicalHookTool)) .sort((left, right) => left.localeCompare(right)); - const timeoutMs = route.config['timeoutMs']; - const timeout = typeof timeoutMs === 'number' && Number.isFinite(timeoutMs) && timeoutMs > 0 - ? Math.ceil(timeoutMs / 1_000) + const configuredTimeoutMs = route.config['timeoutMs']; + const timeoutMs = typeof configuredTimeoutMs === 'number' && Number.isFinite(configuredTimeoutMs) && configuredTimeoutMs > 0 + ? configuredTimeoutMs : undefined; const fallback = route.config['fallback'] === 'standalone' ? 'standalone' as const : 'none' as const; const runtime = route.config['runtime'] === 'standalone' ? 'standalone' as const : 'shared' as const; @@ -485,7 +486,7 @@ const normalizeHooks = ( provenance: { kind: 'conventional', sourcePath: route.source }, source: route.source, targets, - ...(timeout === undefined ? {} : { timeout }), + ...(timeoutMs === undefined ? {} : { timeoutMs }), tools, }); } diff --git a/packages/agent-bundle/src/config/validate.ts b/packages/agent-bundle/src/config/validate.ts index 77546dcfb..bbae8e267 100644 --- a/packages/agent-bundle/src/config/validate.ts +++ b/packages/agent-bundle/src/config/validate.ts @@ -1719,7 +1719,7 @@ export const validateModel = ( server.generatedRoutes !== undefined && server.targets.includes(target)); if (runtimeHost) continue; diagnostics.push({ - code: 'AB4816', + code: 'AB4817', message: `Event route ${hook.eventRoute.event} requires the shared runtime on ${target}, but no generated MCP entry hosts it.`, recovery: 'Add a generated MCP route server, or explicitly set event config.runtime to standalone or config.fallback to standalone.', severity: 'error', diff --git a/packages/agent-bundle/src/core/types.ts b/packages/agent-bundle/src/core/types.ts index 5a34d920f..d398d7b09 100644 --- a/packages/agent-bundle/src/core/types.ts +++ b/packages/agent-bundle/src/core/types.ts @@ -425,8 +425,8 @@ export interface NormalizedHook { readonly provenance: SourceProvenance; readonly source: string; readonly targets: readonly string[]; - /** Native hook timeout in seconds. Omit it to use the selected host's default. */ - readonly timeout?: number; + /** Hook execution deadline in milliseconds. Omit it to use the selected host's default. */ + readonly timeoutMs?: number; readonly tools: readonly CanonicalHookTool[]; } diff --git a/packages/agent-bundle/src/events/ipc.ts b/packages/agent-bundle/src/events/ipc.ts index e632a3df4..957e5c564 100644 --- a/packages/agent-bundle/src/events/ipc.ts +++ b/packages/agent-bundle/src/events/ipc.ts @@ -1,12 +1,13 @@ import { createHash } from 'node:crypto'; -import { chmod, mkdir, rm } from 'node:fs/promises'; +import { chmod, mkdir, rm, stat } from 'node:fs/promises'; import { createConnection, createServer, type Server, type Socket } from 'node:net'; import { dirname, join } from 'node:path'; +import { StringDecoder } from 'node:string_decoder'; import { Context, Duration, Effect, Layer } from 'effect'; import { z } from 'zod'; -import { makeScopedEffectRuntime, runPromise, type ScopedEffectRuntime } from '../effect/boundary.ts'; +import { isAbortError, makeScopedEffectRuntime, runPromise, type ScopedEffectRuntime } from '../effect/boundary.ts'; import { liftPromise } from '../effect/lift.ts'; const EVENT_RUNTIME_PROTOCOL_VERSION = 1 as const; @@ -65,7 +66,7 @@ export interface EventRuntimeRequest { export interface CreateEventRuntimeServerOptions { readonly artifactEpoch: string; readonly endpointId: string; - readonly handle: (request: EventRuntimeRequest) => Promise; + readonly handle: (request: EventRuntimeRequest, signal: AbortSignal) => Promise; } export interface EventRuntimeServer { @@ -104,6 +105,8 @@ const readOneMessage = Effect.fnUntraced(function*( socket: Socket, ): Effect.fn.Return { return yield* Effect.callback((resume) => { + const decoder = new StringDecoder('utf8'); + let receivedBytes = 0; let raw = ''; const cleanup = (): void => { socket.removeListener('data', onData); @@ -114,19 +117,27 @@ const readOneMessage = Effect.fnUntraced(function*( cleanup(); resume(effect); }; + const parse = (message: string): void => { + try { + finish(Effect.succeed(JSON.parse(message))); + } catch (error) { + finish(Effect.fail(transportError('invalid-message', 'Event runtime message must be one JSON value.', error))); + } + }; const onData = (chunk: Buffer): void => { - raw += chunk.toString('utf8'); - if (Buffer.byteLength(raw) > MAX_EVENT_MESSAGE_BYTES) { + receivedBytes += chunk.byteLength; + raw += decoder.write(chunk); + if (receivedBytes > MAX_EVENT_MESSAGE_BYTES) { finish(Effect.fail(transportError('invalid-message', 'Event runtime message exceeds the 1 MiB limit.'))); socket.destroy(); + return; } + const delimiter = raw.indexOf('\n'); + if (delimiter !== -1) parse(raw.slice(0, delimiter)); }; const onEnd = (): void => { - try { - finish(Effect.succeed(JSON.parse(raw))); - } catch (error) { - finish(Effect.fail(transportError('invalid-message', 'Event runtime message must be one JSON value.', error))); - } + raw += decoder.end(); + parse(raw); }; const onError = (error: Error): void => { finish(Effect.fail(transportError('runtime-failed', 'Event runtime socket failed.', error))); @@ -144,6 +155,7 @@ const readOneMessage = Effect.fnUntraced(function*( const handleConnection = Effect.fnUntraced(function*( socket: Socket, options: CreateEventRuntimeServerOptions, + signal: AbortSignal, ): Effect.fn.Return { const raw = yield* readOneMessage(socket).pipe(Effect.exit); if (raw._tag === 'Failure') { @@ -183,7 +195,7 @@ const handleConnection = Effect.fnUntraced(function*( hostContractRevision: parsed.data.hostContractRevision, native: parsed.data.native, target: parsed.data.target, - })).pipe(Effect.exit); + }, signal)).pipe(Effect.exit); if (handled._tag === 'Failure') { writeResponse(socket, { artifactEpoch: options.artifactEpoch, @@ -204,6 +216,7 @@ const handleConnection = Effect.fnUntraced(function*( interface EventSocketServiceShape { readonly endpoint: string; + readonly endpointIdentity?: Readonly<{ readonly device: number; readonly inode: number }>; readonly server: Server; readonly sockets: Set; } @@ -212,6 +225,46 @@ class EventSocketService extends Context.Service => + Effect.callback((resume) => { + const socket = createConnection(endpoint); + const cleanup = (): void => { + socket.removeListener('connect', onConnect); + socket.removeListener('error', onError); + }; + const finish = (effect: Effect.Effect): void => { + cleanup(); + socket.destroy(); + resume(effect); + }; + const onConnect = (): void => { + finish(Effect.succeed('live')); + }; + const onError = (error: NodeJS.ErrnoException): void => { + if (error.code === 'ENOENT') { + finish(Effect.succeed('missing')); + return; + } + if (error.code === 'ECONNREFUSED') { + finish(Effect.succeed('stale')); + return; + } + finish(Effect.fail(transportError( + 'runtime-failed', + 'Unable to inspect the existing event runtime endpoint.', + error, + ))); + }; + socket.once('connect', onConnect); + socket.once('error', onError); + return Effect.sync(() => { + cleanup(); + socket.destroy(); + }); + }); + const openServer = ( options: CreateEventRuntimeServerOptions, ): Effect.Effect => Effect.gen(function*() { @@ -220,17 +273,41 @@ const openServer = ( yield* liftPromise(async () => { await mkdir(dirname(endpoint), { mode: 0o700, recursive: true }); await chmod(dirname(endpoint), 0o700); - await rm(endpoint, { force: true }); }).pipe( Effect.mapError((error) => transportError('runtime-failed', 'Unable to prepare the event runtime endpoint.', error)), ); + const endpointState = yield* probeEndpoint(endpoint); + if (endpointState === 'live') { + return yield* Effect.fail(transportError( + 'runtime-failed', + 'Event runtime endpoint already has a live server.', + )); + } + if (endpointState === 'stale') { + yield* liftPromise(() => rm(endpoint)).pipe( + Effect.mapError((error) => transportError('runtime-failed', 'Unable to remove the stale event runtime endpoint.', error)), + ); + } } return yield* Effect.callback((resume) => { const sockets = new Set(); const server = createServer({ allowHalfOpen: true }, (socket) => { sockets.add(socket); socket.once('close', () => sockets.delete(socket)); - void runPromise(handleConnection(socket, options)); + const controller = new AbortController(); + const interrupt = (): void => controller.abort(); + socket.once('close', interrupt); + socket.once('end', interrupt); + socket.once('error', interrupt); + void runPromise(handleConnection(socket, options, controller.signal), { signal: controller.signal }) + .catch((error: unknown) => { + if (!isAbortError(error)) throw error; + }) + .finally(() => { + socket.removeListener('close', interrupt); + socket.removeListener('end', interrupt); + socket.removeListener('error', interrupt); + }); }); const onError = (error: Error): void => { resume(Effect.fail(transportError('runtime-failed', 'Unable to listen on the event runtime endpoint.', error))); @@ -242,8 +319,13 @@ const openServer = ( resume(Effect.succeed({ endpoint, server, sockets })); return; } - void chmod(endpoint, 0o600).then( - () => resume(Effect.succeed({ endpoint, server, sockets })), + void chmod(endpoint, 0o600).then(() => stat(endpoint)).then( + (endpointStat) => resume(Effect.succeed({ + endpoint, + endpointIdentity: { device: endpointStat.dev, inode: endpointStat.ino }, + server, + sockets, + })), (error: unknown) => { server.close(); resume(Effect.fail(transportError('runtime-failed', 'Unable to secure the event runtime endpoint.', error))); @@ -257,6 +339,24 @@ const openServer = ( }); }); +const removeOwnedEndpoint = (service: EventSocketServiceShape): Effect.Effect => + liftPromise(async () => { + if (service.endpointIdentity === undefined) return; + let current; + try { + current = await stat(service.endpoint); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') return; + throw error; + } + if ( + current.dev === service.endpointIdentity.device + && current.ino === service.endpointIdentity.inode + ) { + await rm(service.endpoint, { force: true }); + } + }).pipe(Effect.ignore); + const closeServer = (service: EventSocketServiceShape): Effect.Effect => Effect.callback((resume) => { for (const socket of service.sockets) socket.destroy(); @@ -270,7 +370,7 @@ const closeServer = (service: EventSocketServiceShape): Effect.Effect => Effect.ensuring( process.platform === 'win32' ? Effect.void - : liftPromise(() => rm(service.endpoint, { force: true })).pipe(Effect.ignore), + : removeOwnedEndpoint(service), ), ); @@ -313,7 +413,7 @@ const requestProgram = ( ): Effect.Effect => Effect.acquireUseRelease( connect(eventRuntimeEndpoint(options.endpointId)), (socket) => Effect.gen(function*() { - socket.end(`${JSON.stringify({ + socket.write(`${JSON.stringify({ artifactEpoch: options.artifactEpoch, event: options.event, hostContractRevision: options.hostContractRevision, diff --git a/packages/agent-bundle/src/mcp-server-runtime.ts b/packages/agent-bundle/src/mcp-server-runtime.ts index 05073c160..154526c65 100644 --- a/packages/agent-bundle/src/mcp-server-runtime.ts +++ b/packages/agent-bundle/src/mcp-server-runtime.ts @@ -421,17 +421,16 @@ const startEventRuntime = async ( ): Promise<{ readonly close: () => Promise }> => events.createEventRuntimeServer({ artifactEpoch: events.artifactEpoch, endpointId: events.endpointId, - handle: async (request) => { + handle: async (request, signal) => { const event = canonicalEvent(request.event); const nativeEvent = nativeString(request.native, 'hook_event_name') ?? event; - const controller = new AbortController(); const props = events.createCanonicalEventProps( event, request.native, events.target, nativeEvent, request.hostContractRevision, - controller.signal, + signal, ); const sessionId = nativeString(request.native, 'session_id') ?? nativeString(request.native, 'conversation_id'); @@ -446,7 +445,7 @@ const startEventRuntime = async ( surface: event, }, ...(sessionId === undefined ? {} : { session: available({ sessionId }, 'native') }), - signal: controller.signal, + signal, ...(workspaceRoot === undefined ? {} : { workspace: available({ root: workspaceRoot }, 'native') }), }, async () => events.projectEventDocument( await dispatcher.dispatch({ @@ -456,7 +455,7 @@ const startEventRuntime = async ( // props type is what gives it shape on the other side. props: { event, payload: { canonical: props.canonical, native: props.native } as never }, }, - signal: controller.signal, + signal, }), event, events.target, diff --git a/packages/agent-bundle/tests/cursor-adapter.test.ts b/packages/agent-bundle/tests/cursor-adapter.test.ts index 10adb7511..ae01ca710 100644 --- a/packages/agent-bundle/tests/cursor-adapter.test.ts +++ b/packages/agent-bundle/tests/cursor-adapter.test.ts @@ -260,7 +260,7 @@ it('lowers cursor-targeted hooks into the flat versioned document with dedicated provenance: { kind: 'config', sourcePath: configPath }, source: '/workspace/src/hooks/record-write.ts', targets: ['cursor'], - timeout: 30, + timeoutMs: 30_000, tools: ['file.write'], }, ], diff --git a/packages/agent-bundle/tests/event-ipc.test.ts b/packages/agent-bundle/tests/event-ipc.test.ts index 7cf28699c..25896d7d8 100644 --- a/packages/agent-bundle/tests/event-ipc.test.ts +++ b/packages/agent-bundle/tests/event-ipc.test.ts @@ -1,4 +1,5 @@ -import { stat } from 'node:fs/promises'; +import { stat, writeFile } from 'node:fs/promises'; +import { createConnection, type Socket } from 'node:net'; import { Effect } from 'effect'; import { expect, it } from 'effect-rstest'; @@ -43,6 +44,177 @@ it.live('round-trips a bounded event envelope through the epoch-bound runtime so }); })); +it.live('rejects a second live server without disturbing the endpoint owner', () => Effect.gen(function*() { + if (process.platform === 'win32') return; + const endpointId = `event-ipc-owner-${crypto.randomUUID()}`; + yield* Effect.scoped(Effect.gen(function*() { + yield* Effect.acquireRelease( + Effect.promise(() => createEventRuntimeServer({ + artifactEpoch: 'epoch-1', + endpointId, + handle: async (request) => ({ owner: 'first', target: request.target }), + })), + (server) => Effect.promise(() => server.close()), + ); + + const alreadyRunning = yield* Effect.tryPromise({ + try: () => createEventRuntimeServer({ + artifactEpoch: 'epoch-1', + endpointId, + handle: async () => ({ owner: 'second' }), + }), + catch: (error) => error, + }).pipe(Effect.flip); + expect(alreadyRunning).toMatchObject({ + code: 'runtime-failed', + message: expect.stringMatching(/already has a live server/u), + name: EventRuntimeTransportError.name, + }); + const response = yield* Effect.promise(() => requestEventRuntime({ + artifactEpoch: 'epoch-1', + endpointId, + event: 'session/start', + hostContractRevision: '2.1.250', + native: { hook_event_name: 'SessionStart' }, + signal: new AbortController().signal, + target: 'claude', + timeoutMs: 1_000, + })); + expect(response).toEqual({ owner: 'first', target: 'claude' }); + })); +})); + +it.live('replaces a stale event runtime socket file', () => Effect.gen(function*() { + if (process.platform === 'win32') return; + const endpointId = `event-ipc-stale-${crypto.randomUUID()}`; + const endpoint = yield* Effect.acquireUseRelease( + Effect.promise(() => createEventRuntimeServer({ + artifactEpoch: 'epoch-1', + endpointId, + handle: async () => undefined, + })), + (server) => Effect.succeed(server.endpoint), + (server) => Effect.promise(() => server.close()), + ); + yield* Effect.promise(() => writeFile(endpoint, 'stale socket')); + + yield* Effect.acquireRelease( + Effect.promise(() => createEventRuntimeServer({ + artifactEpoch: 'epoch-1', + endpointId, + handle: async () => ({ replaced: true }), + })), + (server) => Effect.promise(() => server.close()), + ); + const response = yield* Effect.promise(() => requestEventRuntime({ + artifactEpoch: 'epoch-1', + endpointId, + event: 'session/start', + hostContractRevision: '2.1.250', + native: { hook_event_name: 'SessionStart' }, + signal: new AbortController().signal, + target: 'claude', + timeoutMs: 1_000, + })); + expect(response).toEqual({ replaced: true }); +})); + +it.live('interrupts an in-flight event handler when the client disconnects', () => Effect.gen(function*() { + const endpointId = `event-ipc-disconnect-${crypto.randomUUID()}`; + let markStarted!: () => void; + let markAborted!: () => void; + const started = new Promise((resolve) => { markStarted = resolve; }); + const aborted = new Promise((resolve) => { markAborted = resolve; }); + const server = yield* Effect.acquireRelease( + Effect.promise(() => createEventRuntimeServer({ + artifactEpoch: 'epoch-1', + endpointId, + handle: async (_request, signal?: AbortSignal) => { + markStarted(); + await new Promise((resolve) => { + signal?.addEventListener('abort', () => { + markAborted(); + resolve(); + }, { once: true }); + }); + }, + })), + (server) => Effect.promise(() => server.close()), + ); + const socket = yield* Effect.acquireRelease( + Effect.promise(() => new Promise((resolve, reject) => { + const socket = createConnection(server.endpoint); + socket.once('connect', () => resolve(socket)); + socket.once('error', reject); + })), + (socket) => Effect.sync(() => socket.destroy()), + ); + socket.on('error', () => undefined); + socket.write(`${JSON.stringify({ + artifactEpoch: 'epoch-1', + event: 'tool/after', + hostContractRevision: '2.1.250', + native: { hook_event_name: 'PostToolUse' }, + protocolVersion: 1, + target: 'claude', + })}\n`); + yield* Effect.promise(() => started); + socket.destroy(); + const observed = yield* Effect.promise(() => aborted); + expect(observed).toBeUndefined(); +})); + +it.live('decodes a JSON request whose multibyte UTF-8 code point spans socket writes', () => Effect.gen(function*() { + const endpointId = `event-ipc-utf8-${crypto.randomUUID()}`; + const server = yield* Effect.acquireRelease( + Effect.promise(() => createEventRuntimeServer({ + artifactEpoch: 'epoch-1', + endpointId, + handle: async (request) => request.native['label'], + })), + (server) => Effect.promise(() => server.close()), + ); + const socket = yield* Effect.acquireRelease( + Effect.promise(() => new Promise((resolve, reject) => { + const socket = createConnection(server.endpoint); + socket.once('connect', () => resolve(socket)); + socket.once('error', reject); + })), + (socket) => Effect.sync(() => socket.destroy()), + ); + const response = new Promise((resolve, reject) => { + const chunks: Buffer[] = []; + socket.on('data', (chunk: Buffer) => chunks.push(chunk)); + socket.once('end', () => { + try { + resolve(JSON.parse(Buffer.concat(chunks).toString('utf8'))); + } catch (error) { + reject(error); + } + }); + socket.once('error', reject); + }); + const request = Buffer.from(`${JSON.stringify({ + artifactEpoch: 'epoch-1', + event: 'tool/after', + hostContractRevision: '2.1.250', + native: { hook_event_name: 'PostToolUse', label: 'café 🚀' }, + protocolVersion: 1, + target: 'claude', + })}\n`, 'utf8'); + const codePoint = Buffer.from('🚀', 'utf8'); + const codePointOffset = request.indexOf(codePoint); + expect(codePointOffset).toBeGreaterThanOrEqual(0); + socket.write(request.subarray(0, codePointOffset + 1)); + socket.write(request.subarray(codePointOffset + 1)); + + const output = yield* Effect.promise(() => response); + expect(output).toMatchObject({ + output: 'café 🚀', + status: 'ok', + }); +})); + it.live('fails closed on artifact epoch mismatch and missing runtimes', () => Effect.gen(function*() { const endpointId = `event-ipc-${crypto.randomUUID()}`; yield* Effect.scoped(Effect.gen(function*() { diff --git a/packages/agent-bundle/tests/hooks.test.ts b/packages/agent-bundle/tests/hooks.test.ts index b1356b9a1..edc8dfa90 100644 --- a/packages/agent-bundle/tests/hooks.test.ts +++ b/packages/agent-bundle/tests/hooks.test.ts @@ -697,7 +697,7 @@ it('escalates timed-out and aborted wrapper process trees from TERM to KILL befo const base = hookModel(root); const model: NormalizedPlugin = { ...base, - hooks: [{ ...base.hooks[1]!, targets: ['codex'], timeout: 1 }], + hooks: [{ ...base.hooks[1]!, targets: ['codex'], timeoutMs: 1_000 }], targets: [base.targets[0]!], }; const service = new HookService(); @@ -1177,7 +1177,7 @@ const hookModel = (root: string): NormalizedPlugin => ({ provenance: { kind: 'config', sourcePath: join(root, 'agent-bundle.config.ts') }, source: join(root, 'src', 'hooks', 'check-command.ts'), targets: ['claude', 'codex'], - timeout: 7, + timeoutMs: 7_000, tools: ['shell'], }, { @@ -1304,7 +1304,7 @@ it('normalizes a mixed hook fixture and reports malformed hook declarations', as event: hook.event, name: hook.name, targets: hook.targets, - timeout: hook.timeout, + timeoutMs: hook.timeoutMs, tools: hook.tools, // Normalization orders hooks by their stable id, not by declaration // order, so mixed configured and conventional hooks emit deterministically. @@ -1313,35 +1313,35 @@ it('normalizes a mixed hook fixture and reports malformed hook declarations', as event: 'afterTool', name: 'after-tool-record-87785f02', targets: ['claude', 'codex'], - timeout: undefined, + timeoutMs: undefined, tools: ['file.write', 'shell'], }, { event: 'beforeTool', name: 'before-tool-check-command-1f5b5818', targets: ['claude', 'codex'], - timeout: 7, + timeoutMs: 7_000, tools: ['shell'], }, { event: 'beforeTool', name: 'before-tool-check-command-1f5b5818', targets: ['claude', 'codex'], - timeout: 7, + timeoutMs: 7_000, tools: ['shell'], }, { event: 'sessionStart', name: 'session-start-session-start-7ab7e8a5', targets: ['claude', 'codex'], - timeout: undefined, + timeoutMs: undefined, tools: [], }, { event: 'stop', name: 'stop-stop-bb2d7935', targets: ['claude', 'codex'], - timeout: undefined, + timeoutMs: undefined, tools: [], }, ]); diff --git a/packages/agent-bundle/tests/route-graph.test.ts b/packages/agent-bundle/tests/route-graph.test.ts index d8845aaec..645d64aff 100644 --- a/packages/agent-bundle/tests/route-graph.test.ts +++ b/packages/agent-bundle/tests/route-graph.test.ts @@ -5,7 +5,7 @@ import { dirname, join } from 'node:path'; import { afterEach, expect, it } from '@rstest/core'; import ts from 'typescript-5'; -import { inspect, type ReadyInspectResult } from '../src/api.ts'; +import { inspect, type ReadyInspectResult, validate } from '../src/api.ts'; import { runCli } from '../src/cli.ts'; import { discoverProject } from '../src/config/discover.ts'; import type { AgentBundleConfig } from '../src/core/types.ts'; @@ -694,6 +694,32 @@ it('fails unavailable event routes before packaging unless they are target-restr expect(restricted.state).toBe('ready'); }); +it('preserves sub-second event route timeout precision in the normalized model', async () => { + const root = await createRoot(); + await writeTree(root, { + 'agent-bundle.config.ts': [ + 'export default {', + " plugin: { name: 'event-timeout-fixture', version: '1.0.0' },", + " targets: ['cursor'],", + '};', + '', + ].join('\n'), + 'package.json': '{"type":"module"}\n', + 'src/events/workspace/open.tsx': [ + "export const config = { runtime: 'standalone', timeoutMs: 1250 };", + 'export default async function WorkspaceOpen() { return undefined; }', + '', + ].join('\n'), + }); + + const result = await validate({ root }); + expect(result.diagnostics).toEqual([]); + expect(result.model?.hooks).toContainEqual(expect.objectContaining({ + eventRoute: expect.objectContaining({ event: 'workspace/open' }), + timeoutMs: 1_250, + })); +}); + it('requires an explicit standalone mode when no generated runtime can host an event route', async () => { const root = await createRoot(); await writeTree(root, { @@ -711,7 +737,7 @@ it('requires an explicit standalone mode when no generated runtime can host an e const inspected = await inspect({ root }); expect(inspected.state).toBe('invalid'); expect(inspected.diagnostics).toContainEqual(expect.objectContaining({ - code: 'AB4816', + code: 'AB4817', target: 'cursor', })); }); diff --git a/packages/agent-bundle/tests/target-hook-contract.test.ts b/packages/agent-bundle/tests/target-hook-contract.test.ts index 3170f64b9..a7f346b87 100644 --- a/packages/agent-bundle/tests/target-hook-contract.test.ts +++ b/packages/agent-bundle/tests/target-hook-contract.test.ts @@ -293,13 +293,23 @@ it('plans a thin epoch-bound event-route client and keeps standalone execution e const shared: NormalizedHook = { ...planningHook('afterTool', []), eventRoute: { event: 'tool/after', fallback: 'none', runtime: 'shared' }, + timeoutMs: 1_250, }; - const sharedSource = planHooks(planningModel([shared]), 'synthetic', contract).hookEntries[0]!.virtualSource; + const sharedPlan = planHooks(planningModel([shared]), 'synthetic', contract); + const sharedSource = sharedPlan.hookEntries[0]!.virtualSource; expect(sharedSource).toContain('requestEventRuntime'); expect(sharedSource).toContain('__AGENT_BUNDLE_EVENT_ARTIFACT_EPOCH__'); expect(sharedSource).toContain('hostContractRevision: capabilityRevision'); + expect(sharedSource).toContain('const timeoutMs = 1250;'); expect(sharedSource).not.toContain('import * as routeModule'); + expect(sharedPlan.document).toMatchObject({ + hooks: { + SyntheticAfterTool: [{ + hooks: [{ timeout: 2 }], + }], + }, + }); const degraded: NormalizedHook = { ...shared,