From 071af87944ad26eb67d2de1e84da8b711c8ef871 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Thu, 3 Sep 2026 16:58:38 +0000 Subject: [PATCH 1/2] fix(tests): drive runtime deadlines from injected clocks instead of wall time - rsc-runtime: createAgentRenderEventSequence accepts a time source and the render session accepts an Effect Clock so the maxElapsedMs deadline (elapsed check + pending-boundary sleep) runs on one injectable clock; the dispatcher deadline test now uses TestClock and no longer retries. - agent-bundle: McpProbeService gains a timers seam behind its budget timeout, bounded teardown wait, plugin-data cap, and removal retry; the stalled teardown tests fire those timers in event order. - lifecycle-replay route-unit tests render in-process where the pool already runs under react-server instead of forking a jiti child per replay. - playground-service replay-boundary test waits on subscriber-driven deferreds and awaited appends instead of a fixed polling budget. --- .changeset/deterministic-runtime-deadlines.md | 6 + .../src/dev/playground/mcp-probe-service.ts | 61 ++++++-- .../tests/mcp-probe-service.test.ts | 136 +++++++++++++----- .../tests/playground-service.test.ts | 32 ++++- .../tests/route-unit/lifecycle-replay.test.ts | 7 + packages/rsc-runtime/src/agent-document.ts | 18 ++- packages/rsc-runtime/src/reconciler.ts | 19 ++- packages/rsc-runtime/tests/dispatcher.test.ts | 46 ++++-- 8 files changed, 256 insertions(+), 69 deletions(-) create mode 100644 .changeset/deterministic-runtime-deadlines.md diff --git a/.changeset/deterministic-runtime-deadlines.md b/.changeset/deterministic-runtime-deadlines.md new file mode 100644 index 000000000..eed7beb94 --- /dev/null +++ b/.changeset/deterministic-runtime-deadlines.md @@ -0,0 +1,6 @@ +--- +"@agent-bundle/runtime": patch +"agent-bundle": patch +--- + +Accept an optional `now` time source on `createAgentRenderEventSequence` and run the render dispatcher's `maxElapsedMs` deadline — the event sequence's elapsed check and the pending-boundary deadline sleep — against one injectable clock, so a Flight render's deadline can be driven by a test clock instead of wall-clock time. Add a `timers` option (`McpProbeTimers`) to the Workbench MCP probe service so its total-budget timeout, bounded teardown wait, and detached plugin-data cap can be scheduled without real timers; production behavior is unchanged. (#PR) diff --git a/packages/agent-bundle/src/dev/playground/mcp-probe-service.ts b/packages/agent-bundle/src/dev/playground/mcp-probe-service.ts index 43533c116..437b11cd2 100644 --- a/packages/agent-bundle/src/dev/playground/mcp-probe-service.ts +++ b/packages/agent-bundle/src/dev/playground/mcp-probe-service.ts @@ -46,7 +46,7 @@ export const mcpProbeFailureTextLimit = 2_048; const mcpProbeCapabilityLimit = 32; const mcpProbeNameTextLimit = 256; /** How long a probe response waits for transport teardown before detaching it. */ -const mcpProbeTeardownWaitMs = 50; +export const mcpProbeTeardownWaitMs = 50; /** * Upper bound a detached teardown may hold the plugin-data directory. The * stdio transport's close runs its own TERM/KILL sequence, so this only guards @@ -72,6 +72,33 @@ const connectionErrorCodes = new Set([ export type McpProbeTransport = Transport; +/** + * Timer seam behind every probe delay — the total-budget timeout, the bounded + * teardown wait, the detached plugin-data cap, and the removal retry. Production + * uses Node timers; tests inject a manual scheduler and fire timers in event + * order so no wall-clock time is involved. `schedule` returns the cancel. + */ +export interface McpProbeTimers { + readonly schedule: ( + callback: () => void, + delayMs: number, + options: Readonly<{ + /** A timer that must not keep the process alive on its own (`timer.unref()`). */ + readonly unref: boolean; + }>, + ) => () => void; +} + +const nodeTimers: McpProbeTimers = { + schedule: (callback, delayMs, options) => { + const timer = setTimeout(callback, delayMs); + if (options.unref) timer.unref(); + return () => { + clearTimeout(timer); + }; + }, +}; + export interface McpProbeClient { close(): Promise; connect(transport: Transport): Promise; @@ -100,6 +127,8 @@ export interface McpProbeServiceOptions { readonly projectRoot: string; readonly registry?: TargetRegistry; readonly timeoutMs?: number; + /** Testing seam for every probe delay; production keeps Node timers. */ + readonly timers?: McpProbeTimers; } export class McpProbeTargetNotFoundError extends Error { @@ -314,6 +343,7 @@ export class McpProbeService { readonly #registry: TargetRegistry; readonly #removePluginData: (pluginData: string) => Promise; readonly #timeoutMs: number; + readonly #timers: McpProbeTimers; constructor(options: McpProbeServiceOptions) { this.#clock = options.clock ?? (() => performance.now()); @@ -341,6 +371,7 @@ export class McpProbeService { this.#registry = options.registry ?? createDefaultRegistry(); this.#removePluginData = options.removePluginData ?? removePluginData; this.#timeoutMs = positiveTimeout(options.timeoutMs ?? mcpProbeTimeoutMs); + this.#timers = options.timers ?? nodeTimers; } probe(options: { @@ -448,20 +479,20 @@ export class McpProbeService { * very case the cap bounds. */ #removePluginDataAfter(teardown: Promise, pluginData: string): Promise { - let cap: NodeJS.Timeout | undefined; + let cancelCap: (() => void) | undefined; let capWon = false; // The cap stays referenced on purpose: it is the only handle guaranteeing // the removal runs when a stalled teardown outlives Workbench shutdown, // and it is cleared the moment the teardown settles. const capped = new Promise((resolvePromise) => { - cap = setTimeout(() => { + cancelCap = this.#timers.schedule(() => { capWon = true; resolvePromise(); - }, this.#pluginDataTeardownCapMs); + }, this.#pluginDataTeardownCapMs, { unref: false }); }); const pending = Promise.race([teardown, capped]) .then(() => { - if (cap !== undefined) clearTimeout(cap); + cancelCap?.(); return this.#removePluginData(pluginData); }) .then(() => undefined, () => { @@ -472,7 +503,9 @@ export class McpProbeService { // The teardown settled (a close may have failed fast) but the child // still held the directory for a moment: one bounded, fenced retry. this.#track( - new Promise((resolvePromise) => { setTimeout(resolvePromise, mcpProbePluginDataRetryDelayMs); }) + new Promise((resolvePromise) => { + this.#timers.schedule(resolvePromise, mcpProbePluginDataRetryDelayMs, { unref: false }); + }) .then(() => this.#removePluginData(pluginData)) .then(() => undefined, () => undefined), ); @@ -636,7 +669,7 @@ export class McpProbeService { }); return report; } finally { - let timer: NodeJS.Timeout | undefined; + let cancelWait: (() => void) | undefined; // Keep transport teardown running through its TERM/KILL path without // allowing a stalled close to extend the probe's total time budget. The // plugin-data removal is chained behind that teardown, so a close that @@ -651,13 +684,12 @@ export class McpProbeService { ]); const cleanup = this.#removePluginDataAfter(teardown, options.pluginData); const teardownWait = new Promise((resolvePromise) => { - timer = setTimeout(resolvePromise, mcpProbeTeardownWaitMs); - timer.unref(); + cancelWait = this.#timers.schedule(resolvePromise, mcpProbeTeardownWaitMs, { unref: true }); }); try { await Promise.race([cleanup, teardownWait]); } finally { - if (timer !== undefined) clearTimeout(timer); + cancelWait?.(); } } } @@ -720,18 +752,17 @@ export class McpProbeService { void settledClose(onTimeout); throw new McpProbeTimeoutError(kind); } - let timer: NodeJS.Timeout | undefined; + let cancelTimeout: (() => void) | undefined; const timedOut = new Promise((_resolve, reject) => { - timer = setTimeout(() => { + cancelTimeout = this.#timers.schedule(() => { void settledClose(onTimeout); reject(new McpProbeTimeoutError(kind)); - }, remaining); - timer.unref(); + }, remaining, { unref: true }); }); try { return await Promise.race([operation, timedOut]); } finally { - if (timer !== undefined) clearTimeout(timer); + cancelTimeout?.(); } } } diff --git a/packages/agent-bundle/tests/mcp-probe-service.test.ts b/packages/agent-bundle/tests/mcp-probe-service.test.ts index a2586d860..9517b4e1b 100644 --- a/packages/agent-bundle/tests/mcp-probe-service.test.ts +++ b/packages/agent-bundle/tests/mcp-probe-service.test.ts @@ -9,12 +9,74 @@ import { McpProbeService, McpProbeTargetNotFoundError, mcpProbeInstructionTextLimit, + mcpProbePluginDataTeardownCapMs, + mcpProbeTeardownWaitMs, mcpProbeToolLimit, type McpProbeClient, type McpProbeServiceOptions, + type McpProbeTimers, type McpProbeTransport, } from '../src/dev/playground/mcp-probe-service.ts'; +interface ManualTimer { + readonly callback: () => void; + readonly delayMs: number; +} + +/** + * Manual scheduler for the probe's timer seam: nothing fires on its own. A test + * waits for the service to arm a timer with a given delay (event-ordered — the + * promise settles when the code under test reaches that point), then fires it. + */ +const manualTimers = (): Readonly<{ + readonly fire: (delayMs: number) => Promise; + readonly pending: () => readonly number[]; + readonly timers: McpProbeTimers; +}> => { + const armed = new Set(); + const waiters = new Set<() => void>(); + const find = (delayMs: number): ManualTimer | undefined => + [...armed].find((timer) => timer.delayMs === delayMs); + return Object.freeze({ + fire: async (delayMs) => { + let timer = find(delayMs); + while (timer === undefined) { + await new Promise((resolvePromise) => { waiters.add(resolvePromise); }); + timer = find(delayMs); + } + armed.delete(timer); + timer.callback(); + }, + pending: () => [...armed].map((timer) => timer.delayMs), + timers: { + schedule: (callback, delayMs) => { + const timer: ManualTimer = { callback, delayMs }; + armed.add(timer); + for (const wake of [...waiters]) { + waiters.delete(wake); + wake(); + } + return () => { + armed.delete(timer); + }; + }, + }, + }); +}; + +/** Never settles: a teardown that hangs for the rest of the test. */ +const stalled = (): Promise => new Promise(() => undefined); + +/** + * Whether `promise` has already settled, decided at the next macrotask so every + * microtask chained off the current turn has run first — no wall-clock wait. + */ +const settledBeforeNextTurn = (promise: Promise): Promise => + Promise.race([ + promise.then(() => true, () => true), + new Promise((resolvePromise) => { setImmediate(() => resolvePromise(false)); }), + ]); + const createBundle = async ( servers: Readonly> = { timeline: { @@ -393,40 +455,47 @@ it('returns a timed-out report without awaiting stalled teardown', async () => { const root = await createBundle(); let clientCloses = 0; let transportCloses = 0; - let guard: NodeJS.Timeout | undefined; + const timers = manualTimers(); try { - const stalledClose = async (): Promise => - new Promise((resolvePromise) => setTimeout(resolvePromise, 250)); const service = serviceFor(root, { + // A frozen clock keeps the whole budget for the connect step, so the + // budget timer is armed with exactly `timeoutMs`. + clock: () => 0, createClient: () => client({ - close: async () => { + close: () => { clientCloses += 1; - await stalledClose(); + return stalled(); }, - connect: () => new Promise(() => undefined), + connect: () => stalled(), }), - createStdioTransport: () => transport(async () => { + createStdioTransport: () => transport(() => { transportCloses += 1; - await stalledClose(); + return stalled(); }), timeoutMs: 10, + timers: timers.timers, }); - const report = await Promise.race([ - service.probe({ host: 'claude', serverName: 'timeline' }), - new Promise((_resolve, reject) => { - guard = setTimeout( - () => reject(new Error('The timed-out probe remained blocked on teardown.')), - 150, - ); - }), - ]); + const probe = service.probe({ host: 'claude', serverName: 'timeline' }); + // The budget expires while connect is still pending: the timeout starts + // the transport close, and the report path arms the bounded teardown wait. + await timers.fire(10); + // Both closes hang forever; only the teardown wait may release the report. + await timers.fire(mcpProbeTeardownWaitMs); + expect(await settledBeforeNextTurn(probe)).toBe(true); + const report = await probe; expect(report.status).toBe('timed-out'); expect(clientCloses).toBe(1); expect(transportCloses).toBeGreaterThan(0); + // The detached plugin-data cap is the only timer still armed: teardown is + // running in the background, not on the response path. Firing it releases + // the plugin-data removal, which `settle()` then fences. + expect(timers.pending()).toEqual([mcpProbePluginDataTeardownCapMs]); + await timers.fire(mcpProbePluginDataTeardownCapMs); + await service.settle(); + expect(timers.pending()).toEqual([]); } finally { - if (guard !== undefined) clearTimeout(guard); await rm(root, { force: true, recursive: true }); } }); @@ -434,39 +503,40 @@ it('returns a timed-out report without awaiting stalled teardown', async () => { it('returns a timed-out report without awaiting stalled teardown when the budget is spent before connecting', async () => { const root = await createBundle(); let transportCloses = 0; - let guard: NodeJS.Timeout | undefined; let ticks = 0; + const timers = manualTimers(); try { const service = serviceFor(root, { // The clock reads 0 at probe start and the whole budget later at every // subsequent read, so the connect step finds no time remaining. clock: () => (ticks++ === 0 ? 0 : 10_000), createClient: () => client({ - close: async () => new Promise((resolvePromise) => setTimeout(resolvePromise, 250)), - connect: () => new Promise(() => undefined), + close: () => stalled(), + connect: () => stalled(), }), - createStdioTransport: () => transport(async () => { + createStdioTransport: () => transport(() => { transportCloses += 1; - await new Promise((resolvePromise) => setTimeout(resolvePromise, 250)); + return stalled(); }), timeoutMs: 10, + timers: timers.timers, }); - const report = await Promise.race([ - service.probe({ host: 'claude', serverName: 'timeline' }), - new Promise((_resolve, reject) => { - guard = setTimeout( - () => reject(new Error('The budget-exhausted probe remained blocked on teardown.')), - 150, - ); - }), - ]); + const probe = service.probe({ host: 'claude', serverName: 'timeline' }); + // No budget timer is armed on this path; the report waits only for the + // bounded teardown wait, never for the stalled closes. + await timers.fire(mcpProbeTeardownWaitMs); + expect(await settledBeforeNextTurn(probe)).toBe(true); + const report = await probe; expect(report.status).toBe('timed-out'); expect(report.failure?.kind).toBe('connect'); expect(transportCloses).toBeGreaterThan(0); + expect(timers.pending()).toEqual([mcpProbePluginDataTeardownCapMs]); + await timers.fire(mcpProbePluginDataTeardownCapMs); + await service.settle(); + expect(timers.pending()).toEqual([]); } finally { - if (guard !== undefined) clearTimeout(guard); await rm(root, { force: true, recursive: true }); } }); diff --git a/packages/agent-bundle/tests/playground-service.test.ts b/packages/agent-bundle/tests/playground-service.test.ts index 5f8043606..9378be1ec 100644 --- a/packages/agent-bundle/tests/playground-service.test.ts +++ b/packages/agent-bundle/tests/playground-service.test.ts @@ -13,6 +13,7 @@ import { type PlaygroundEventInput, type PlaygroundJsonObject, type PlaygroundServiceOptions, + type PlaygroundTraceEvent, } from '../src/dev/playground/playground-store.ts'; interface SessionIndex { @@ -549,33 +550,56 @@ it('sets an atomic subscription replay boundary, preserves reentrant order, and await fixture.service.openSession({ ...sessionInput(), sessionId: 'live' }); await fixture.service.append('live', event('project', 'loaded', 'Project loaded.', { revision: 'a' })); const received: number[] = []; + // Every wait below is event-ordered: the deferreds settle from inside the + // subscriber, and appends are awaited, so no polling budget is involved. + const firstDelivered = deferred(); + const secondDelivered = deferred(); + let reentrant: Promise | undefined; const subscription = await fixture.service.subscribe('live', { afterSequence: 0, onEvent: (item) => { received.push(item.sequence); - if (item.sequence === 1) void fixture.service.append('live', event('build', 'completed', 'Build completed.', { epoch: 'epoch-7' })); + if (item.sequence === 1) { + // Issued while the replay backlog is being delivered: it must be + // ordered behind the subscription boundary and reach this subscriber. + reentrant = fixture.service.append('live', event('build', 'completed', 'Build completed.', { epoch: 'epoch-7' })); + firstDelivered.resolve(); + } + if (item.sequence === 2) secondDelivered.resolve(); }, }); - await eventually(() => expect(received).toEqual([1, 2])); + await firstDelivered.promise; + expect(received).toEqual([1]); + await expect(reentrant).resolves.toMatchObject({ sequence: 2 }); + await secondDelivered.promise; + expect(received).toEqual([1, 2]); expect(subscription.closed).toBe(false); await fixture.service.openSession({ ...sessionInput(), sessionId: 'slow' }); const release = deferred(); + const slowFirstDelivered = deferred(); const slowReceived: number[] = []; const slow = await fixture.service.subscribe('slow', { afterSequence: 0, onEvent: async (item) => { slowReceived.push(item.sequence); + if (item.sequence === 1) slowFirstDelivered.resolve(); await release.promise; }, }); await fixture.service.append('slow', event('mcp', 'first', 'First.', { item: 1 })); - await eventually(() => expect(slowReceived).toEqual([1])); + await slowFirstDelivered.promise; + expect(slowReceived).toEqual([1]); + // The subscriber is blocked on `release`; the second event fills the + // one-slot queue and the third overflows it before its append resolves. await fixture.service.append('slow', event('mcp', 'second', 'Second.', { item: 2 })); + expect(slow.closed).toBe(false); await fixture.service.append('slow', event('mcp', 'third', 'Third.', { item: 3 })); - await eventually(() => expect(slow.closed).toBe(true)); + expect(slow.closed).toBe(true); release.resolve(); await expect(fixture.service.replay('slow')).resolves.toMatchObject({ events: [{ sequence: 1 }, { sequence: 2 }, { sequence: 3 }] }); + // Failing closed dropped the queued event instead of delivering it late. + expect(slowReceived).toEqual([1]); } finally { await fixture.close(); } diff --git a/packages/agent-bundle/tests/route-unit/lifecycle-replay.test.ts b/packages/agent-bundle/tests/route-unit/lifecycle-replay.test.ts index 9a8033bda..7cef1bce1 100644 --- a/packages/agent-bundle/tests/route-unit/lifecycle-replay.test.ts +++ b/packages/agent-bundle/tests/route-unit/lifecycle-replay.test.ts @@ -10,6 +10,7 @@ import type { LifecycleReplay } from '../../src/contracts/lifecycles.ts'; import { LifecycleReplayService } from '../../src/dev/playground/lifecycle-replay-service.ts'; import { projectEventDocument } from '../../src/events/project.ts'; import { compileRouteGraph } from '../../src/routes/graph.ts'; +import { renderRouteEvents } from '../../src/test/render.ts'; import type { AgentRouteModule } from '../../src/test/types.ts'; const roots: string[] = []; @@ -104,8 +105,13 @@ const createFixtureProject = async () => { it('replays Claude and Codex PostToolUse through decode, route execution, render, and encode', async () => { const { graph } = await createFixtureProject(); + // This pool already runs under the `react-server` condition, so the route + // renders in-process through the same renderer the dev server's render child + // uses; the child fork itself (a jiti-transpiled Node process per replay) is + // covered by the captured-fixture replays below. const service = new LifecycleReplayService({ prepared: () => ({ graph, targets: ['claude', 'codex'] }), + render: renderRouteEvents, }); const fixtures = [ { @@ -350,6 +356,7 @@ it('replays the Cursor workspaceOpen starter as an observation with no native re const graph = await compileRouteGraph(root, { targets: ['cursor'] } as never); const service = new LifecycleReplayService({ prepared: () => ({ graph, targets: ['cursor'] }), + render: renderRouteEvents, }); const lifecycle = service.list().lifecycles.find((candidate) => candidate.routeId === 'event:workspace/open'); const target = lifecycle?.targets.find((candidate) => candidate.target === 'cursor'); diff --git a/packages/rsc-runtime/src/agent-document.ts b/packages/rsc-runtime/src/agent-document.ts index f9654b69e..4ad5bef43 100644 --- a/packages/rsc-runtime/src/agent-document.ts +++ b/packages/rsc-runtime/src/agent-document.ts @@ -479,11 +479,17 @@ export interface AgentRenderEventSequence { readonly emit: (input: AgentRenderEventInput) => AgentRenderEvent; } +/** + * `now` is the sequence's only time source (elapsed-time and event-rate + * bounds); it defaults to the wall clock and exists so a render pipeline can + * run every deadline against one injected clock. + */ export const createAgentRenderEventSequence = ( limitOverrides: Partial = {}, + now: () => number = Date.now, ): AgentRenderEventSequence => { const limits = resolveAgentRenderLimits(limitOverrides); - const startedAt = Date.now(); + const startedAt = now(); const recentTimes: number[] = []; let completed = false; let nextSequence = 0; @@ -501,12 +507,12 @@ export const createAgentRenderEventSequence = ( 'The render is complete; later work requires a new invocation handoff', ); } - const now = Date.now(); - if (now - startedAt > limits.maxElapsedMs) { + const emittedAt = now(); + if (emittedAt - startedAt > limits.maxElapsedMs) { throw elapsedTimeExceeded(limits.maxElapsedMs); } - recentTimes.push(now); - const windowStart = now - 1000; + recentTimes.push(emittedAt); + const windowStart = emittedAt - 1000; while (recentTimes[0] !== undefined && recentTimes[0] < windowStart) { recentTimes.shift(); } @@ -538,7 +544,7 @@ export const createAgentRenderEventSequence = ( return nextSequence; }, get remainingMs() { - return limits.maxElapsedMs - (Date.now() - startedAt); + return limits.maxElapsedMs - (now() - startedAt); }, }); }; diff --git a/packages/rsc-runtime/src/reconciler.ts b/packages/rsc-runtime/src/reconciler.ts index 92a9445b4..9b9a7636b 100644 --- a/packages/rsc-runtime/src/reconciler.ts +++ b/packages/rsc-runtime/src/reconciler.ts @@ -1,4 +1,4 @@ -import { Deferred, Duration, Effect, Exit, Option, Queue, Stream, type Scope } from 'effect'; +import { Clock, Deferred, Duration, Effect, Exit, Option, Queue, Stream, type Scope } from 'effect'; import { createElement, isValidElement, type ReactElement, type ReactNode } from 'react'; import { createFromReadableStream } from 'react-server-dom-rspack/client.node'; @@ -486,6 +486,12 @@ const progressInput = (update: AgentProgressUpdate): AgentRenderEventInput => ({ }); export interface AgentRenderEventStreamOptions { + /** + * Time source for the `maxElapsedMs` deadline: both the event sequence's + * elapsed check and the pending-boundary deadline sleep read it. Defaults to + * the runtime clock; tests inject a `TestClock` so no real time elapses. + */ + readonly clock?: Clock.Clock; readonly demand: FlightDemand; readonly flight: Promise>; readonly limits?: Partial; @@ -513,7 +519,11 @@ const handoffRequired = (): AgentContractError => export const createAgentRenderEventSession = ( options: AgentRenderEventStreamOptions, ): AgentRenderEventSession => { - const sequence = createAgentRenderEventSequence(options.limits); + const clock = options.clock; + const sequence = createAgentRenderEventSequence( + options.limits, + clock === undefined ? undefined : () => clock.currentTimeMillisUnsafe(), + ); const maxBufferedProgress = resolveAgentRenderLimits(options.limits).maxEvents; let offerProgress: ((input: AgentRenderEventInput) => Effect.Effect) | undefined; let progressFailure: Error | undefined; @@ -606,7 +616,10 @@ export const createAgentRenderEventSession = ( ); }), ); - return { events, progress }; + return { + events: clock === undefined ? events : Stream.provideService(events, Clock.Clock, clock), + progress, + }; }; diff --git a/packages/rsc-runtime/tests/dispatcher.test.ts b/packages/rsc-runtime/tests/dispatcher.test.ts index 39eb9658b..e8e083714 100644 --- a/packages/rsc-runtime/tests/dispatcher.test.ts +++ b/packages/rsc-runtime/tests/dispatcher.test.ts @@ -3,9 +3,12 @@ import { readFileSync } from 'node:fs'; import { join } from 'node:path'; import { Readable } from 'node:stream'; -import { describe, expect, it } from '@rstest/core'; +import { Effect } from 'effect'; +import { TestClock } from 'effect/testing'; +import { describe, expect, it } from 'effect-rstest'; import { createElement } from 'react'; +import { createFlightDemand } from '../src/effect/render-stream.js'; import { AgentContractError, createAgentRenderDispatcher, @@ -14,6 +17,7 @@ import { type AgentProgressReporter, type AgentRenderEvent, } from '../src/index.js'; +import { createAgentRenderEventSession, toPublicEventStream } from '../src/reconciler.js'; describe('decodeAgentDocument', () => { it('decodes protocol host elements into one immutable final document', () => { @@ -436,6 +440,10 @@ const collectEvents = async ( return events; }; +/** The rejection reason of `read`; fails the test effect with the value if it resolves instead. */ +const rejectionOf = (read: () => Promise): Effect.Effect => + Effect.flip(Effect.tryPromise({ catch: (error) => error, try: read })); + const eventTypes = (events: readonly AgentRenderEvent[]): readonly string[] => events.map((event) => { switch (event.type) { @@ -805,15 +813,37 @@ describe('AgentRenderDispatcher streaming', () => { expect(rest.at(-1)?.type).toBe('complete'); }); - it('fails a permanently pending boundary when the elapsed deadline expires', { retry: 2 }, async () => { + it.effect('fails a permanently pending boundary when the elapsed deadline expires', () => Effect.gen(function*() { + // The deadline runs on the injected TestClock, so the Flight worker may + // take any real time to deliver the shell under load; the boundary fails + // exactly when the test advances the clock past `maxElapsedMs`, whether + // the deadline sleep was already armed (it fires) or is armed afterwards + // (no time remains). This is the dispatcher's own `stream()` wiring with + // the clock injected at the session seam. + const clock = yield* TestClock.testClockWith(Effect.succeed); const host = createWorkerHost('single'); - const dispatcher = createAgentRenderDispatcher(host, { limits: { maxElapsedMs: 150 } }); - const reader = dispatcher.stream({ invocation, signal: new AbortController().signal }).getReader(); - const shell = await reader.read(); + const controller = new AbortController(); + yield* Effect.addFinalizer(() => Effect.sync(() => { + controller.abort(); + })); + const demand = createFlightDemand(); + const session = createAgentRenderEventSession({ + clock, + demand, + flight: host.execute({ invocation, signal: controller.signal }), + limits: { maxElapsedMs: 150 }, + signal: controller.signal, + }); + const reader = toPublicEventStream(session.events, demand, controller.signal).getReader(); + const shell = yield* Effect.promise(() => reader.read()); if (shell.value?.type !== 'shell') throw new Error('expected a shell event'); - await expect(reader.read()).rejects.toBeInstanceOf(AgentContractError); - await expect(reader.read()).rejects.toMatchObject({ code: 'elapsed-time-exceeded' }); - }); + + yield* TestClock.adjust(150); + const failure = yield* rejectionOf(() => reader.read()); + expect(failure).toBeInstanceOf(AgentContractError); + expect(failure).toMatchObject({ code: 'elapsed-time-exceeded' }); + expect(yield* rejectionOf(() => reader.read())).toMatchObject({ code: 'elapsed-time-exceeded' }); + })); it('bounds Flight EOF for stream and dispatch and cancels the source', { retry: 2, timeout: 5_000 }, async () => { const finiteReader = ( From 72da7243931918b3292fd4e38524d5040d8f47a8 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Thu, 3 Sep 2026 17:04:28 +0000 Subject: [PATCH 2/2] chore: reference #434 in the changeset --- .changeset/deterministic-runtime-deadlines.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.changeset/deterministic-runtime-deadlines.md b/.changeset/deterministic-runtime-deadlines.md index eed7beb94..f3a57cecc 100644 --- a/.changeset/deterministic-runtime-deadlines.md +++ b/.changeset/deterministic-runtime-deadlines.md @@ -3,4 +3,4 @@ "agent-bundle": patch --- -Accept an optional `now` time source on `createAgentRenderEventSequence` and run the render dispatcher's `maxElapsedMs` deadline — the event sequence's elapsed check and the pending-boundary deadline sleep — against one injectable clock, so a Flight render's deadline can be driven by a test clock instead of wall-clock time. Add a `timers` option (`McpProbeTimers`) to the Workbench MCP probe service so its total-budget timeout, bounded teardown wait, and detached plugin-data cap can be scheduled without real timers; production behavior is unchanged. (#PR) +Accept an optional `now` time source on `createAgentRenderEventSequence` and run the render dispatcher's `maxElapsedMs` deadline — the event sequence's elapsed check and the pending-boundary deadline sleep — against one injectable clock, so a Flight render's deadline can be driven by a test clock instead of wall-clock time. Add a `timers` option (`McpProbeTimers`) to the Workbench MCP probe service so its total-budget timeout, bounded teardown wait, and detached plugin-data cap can be scheduled without real timers; production behavior is unchanged. (#434)