fix(events): cancel orphaned IPC renders, live-socket ownership, streaming UTF-8, ms timeout precision, fresh diagnostic

.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.

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/<server>/{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`)
 

docs/entry-conventions.md

@@ -137,7 +137,7 @@ export default async function inspect({ input, signal }: CliRouteProps<typeof in
 
 The compiler statically projects `inputSchema` onto argv (the bounded grammar
 and every policy rule are documented in
-[Diagnostics](diagnostics.md#route-graph-ab4800ab4816)), generates nested
+[Diagnostics](diagnostics.md#route-graph-ab4800ab4817)), generates nested
 help (`--help` at every level, `--version` at the root), and emits
 `dist/bin/<plugin-name>.js` with the shebang and executable bit through the
 same Rslib synthesis as every other bin. At run time the shell resolves the

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,

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 (

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,
     });
   }

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',

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[];
 }
 

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<unknown>;
+  readonly handle: (request: EventRuntimeRequest, signal: AbortSignal) => Promise<unknown>;
 }
 
 export interface EventRuntimeServer {
@@ -104,6 +105,8 @@ const readOneMessage = Effect.fnUntraced(function*(
   socket: Socket,
 ): Effect.fn.Return<unknown, EventRuntimeTransportError> {
   return yield* Effect.callback<unknown, EventRuntimeTransportError>((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<void, never> {
   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<Socket>;
 }
@@ -212,6 +225,46 @@ class EventSocketService extends Context.Service<EventSocketService, EventSocket
   'agent-bundle/events/EventSocketService',
 ) {}
 
+type EndpointProbe = 'live' | 'missing' | 'stale';
+
+const probeEndpoint = (endpoint: string): Effect.Effect<EndpointProbe, EventRuntimeTransportError> =>
+  Effect.callback<EndpointProbe, EventRuntimeTransportError>((resume) => {
+    const socket = createConnection(endpoint);
+    const cleanup = (): void => {
+      socket.removeListener('connect', onConnect);
+      socket.removeListener('error', onError);
+    };
+    const finish = (effect: Effect.Effect<EndpointProbe, EventRuntimeTransportError>): 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<EventSocketServiceShape, EventRuntimeTransportError> => 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<EventSocketServiceShape, EventRuntimeTransportError>((resume) => {
     const sockets = new Set<Socket>();
     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<void> =>
+  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<void> =>
   Effect.callback<void>((resume) => {
     for (const socket of service.sockets) socket.destroy();
@@ -270,7 +370,7 @@ const closeServer = (service: EventSocketServiceShape): Effect.Effect<void> =>
     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<unknown, EventRuntimeTransportError> => 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,