# Files changed · PR \#22565 · pingdotgg/t3code [View on GitCafe](https://git.cafe/pingdotgg/t3code/pull/22565/diffs) Repository: [pingdotgg/t3code](https://git.cafe/pingdotgg/t3code) Visibility: public [test: warn on polling loops in tests and make adapter test harnesses event-driven](https://git.cafe/pingdotgg/t3code/pull/22565.md) Version: 31 Head commit: c6593a8410f4cd26803a8b87f1266dc49f86df2e Base commit: a9550e8cf64dfa10a839f19573a6384017669bbe Comparison base: a9550e8cf64dfa10a839f19573a6384017669bbe Showing up to 20 entries from this revision. If the pull request changes, reopen the first page before continuing. ## apps/desktop/src/preview/Manager.test.ts Status: modified; +73; −26 ```diff @@ -376,6 +376,7 @@ }; const makeFaviconWebContents = (options?: { + readonly changes?: ChangeSignal; readonly fetch?: (url: string, init?: RequestInit) => Promise; readonly id?: number; readonly rasterize?: (code: string) => Promise; @@ -386,16 +387,25 @@ let currentUrl = options?.url ?? "http://localhost:3200/"; let destroyed = false; let loading = false; - const fetch = vi.fn( + const fetchImpl = options?.fetch ?? + (async () => + new Response(new Uint8Array(sourcePng), { + headers: { "content-type": "image/png" }, + })); + const fetch = vi.fn((url: string, init?: RequestInit) => { + const response = fetchImpl(url, init); - (async () => + options?.changes?.notify(); + return response; + }); - new Response(new Uint8Array(sourcePng), { - headers: { "content-type": "image/png" }, - })), - ); const executeJavaScriptInIsolatedWorld = vi.fn( - async (_worldId: number, scripts: ReadonlyArray<{ readonly code: string }>) => - options?.rasterize ? options.rasterize(scripts[0]?.code ?? "") : TEST_FAVICON, + (_worldId: number, scripts: ReadonlyArray<{ readonly code: string }>) => { + const result = options?.rasterize + ? options.rasterize(scripts[0]?.code ?? "") + : Promise.resolve(TEST_FAVICON); + options?.changes?.notify(); + return result; + }, ); const reload = vi.fn(); const loadURL = vi.fn(async (url: string) => { @@ -460,10 +470,33 @@ }; }; +// Electron mocks and state subscribers call `notify`, so a wait re-checks its +// predicate exactly when the state it reads changes. +const makeChangeSignal = () => { + const waiters = new Set<() => void>(); + return { + notify: () => { + for (const wake of waiters) wake(); + }, + until: (predicate: () => boolean) => + Effect.callback((resume) => { + if (predicate()) return resume(Effect.void); + const wake = () => { + if (!predicate()) return; + waiters.delete(wake); + resume(Effect.void); + }; + waiters.add(wake); + return Effect.sync(() => void waiters.delete(wake)); + }), + }; +}; +type ChangeSignal = ReturnType; + +// Favicon captures run detached from the Electron listener, so asserting that -const settle = function* (until: () => boolean) { +// one published nothing can only give its promise chain a fixed number of turns. +const flushMicrotasks = function* () { + for (let turn = 0; turn < 30; turn++) yield* Effect.promise(() => Promise.resolve()); - for (let attempt = 0; attempt < 30 && !until(); attempt++) { - yield* Effect.promise(() => Promise.resolve()); - } }; const makeTestPictureInPictureWindow = (loadURL: () => Promise = async () => undefined) => { @@ -843,11 +876,13 @@ effectIt.effect("shares an identical in-flight event and lets a changed event win", () => withManager((manager) => Effect.gen(function* () { + const changes = makeChangeSignal(); let resolveFirst!: (response: Response) => void; const firstResponse = new Promise((resolve) => { resolveFirst = resolve; }); const preview = makeFaviconWebContents({ + changes, fetch: (url) => url.endsWith("first.png") ? firstResponse @@ -862,6 +897,7 @@ yield* manager.subscribeStateChanges((_tabId, state) => Effect.sync(() => { states.push(state); + changes.notify(); }), ); yield* manager.createTab("tab_favicon_latest"); @@ -870,15 +906,15 @@ const faviconUpdated = preview.listeners.get("page-favicon-updated")!; faviconUpdated({} as never, ["http://localhost:3200/first.png"] as never); faviconUpdated({} as never, ["http://localhost:3200/first.png"] as never); - yield* settle(() => preview.fetch.mock.calls.length === 1); + yield* changes.until(() => preview.fetch.mock.calls.length === 1); faviconUpdated({} as never, ["http://localhost:3200/second.png"] as never); + yield* changes.until(() => states.at(-1)?.favicon !== undefined); - yield* settle(() => states.at(-1)?.favicon !== undefined); resolveFirst( new Response(new Uint8Array(makeSourcePng()), { headers: { "content-type": "image/png" }, }), ); + yield* flushMicrotasks(); - yield* settle(() => false); expect(preview.fetch).toHaveBeenCalledTimes(2); expect(states.filter((state) => state.favicon !== undefined)).toHaveLength(1); @@ -889,8 +925,10 @@ effectIt.effect("allows an identical retry after an undecodable capture", () => withManager((manager) => Effect.gen(function* () { + const changes = makeChangeSignal(); let rasterizations = 0; const preview = makeFaviconWebContents({ + changes, rasterize: async () => (++rasterizations === 1 ? null : TEST_FAVICON), }); fromId.mockReturnValue(preview.webContents); @@ -898,6 +936,7 @@ yield* manager.subscribeStateChanges((_tabId, state) => Effect.sync(() => { states.push(state); + changes.notify(); }), ); yield* manager.createTab("tab_favicon_retry"); @@ -905,10 +944,10 @@ const faviconUpdated = preview.listeners.get("page-favicon-updated")!; faviconUpdated({} as never, ["http://localhost:3200/favicon.png"] as never); - yield* settle(() => rasterizations === 1); + yield* changes.until(() => rasterizations === 1); - yield* settle(() => false); + yield* flushMicrotasks(); faviconUpdated({} as never, ["http://localhost:3200/favicon.png"] as never); + yield* changes.until(() => states.at(-1)?.favicon !== undefined); - yield* settle(() => states.at(-1)?.favicon !== undefined); expect(rasterizations).toBe(2); expect(states.at(-1)?.favicon?.dataUrl).toBe(TEST_FAVICON); @@ -919,8 +958,10 @@ effectIt.effect("does not publish a capture invalidated by navigation", () => withManager((manager) => Effect.gen(function* () { + const changes = makeChangeSignal(); let resolveFetch!: (response: Response) => void; const preview = makeFaviconWebContents({ + changes, fetch: () => new Promise((resolve) => { resolveFetch = resolve; @@ -931,6 +972,7 @@ yield* manager.subscribeStateChanges((_tabId, state) => Effect.sync(() => { states.push(state); + changes.notify(); }), ); yield* manager.createTab("tab_favicon_navigation"); @@ -939,7 +981,7 @@ {} as never, ["http://localhost:3200/favicon.png"] as never, ); - yield* settle(() => preview.fetch.mock.calls.length === 1); + yield* changes.until(() => preview.fetch.mock.calls.length === 1); preview.listeners.get("did-start-navigation")?.({ isMainFrame: true, isSameDocument: false, @@ -950,7 +992,7 @@ headers: { "content-type": "image/png" }, }), ); - yield* settle(() => false); + yield* flushMicrotasks(); expect(states.some((state) => state.favicon !== undefined)).toBe(false); }), @@ -960,15 +1002,17 @@ effectIt.effect("ignores an old capture that completes after webview replacement", () => withManager((manager) => Effect.gen(function* () { + const changes = makeChangeSignal(); let resolveFetch!: (response: Response) => void; const initial = makeFaviconWebContents({ + changes, id: 42, fetch: () => new Promise((resolve) => { resolveFetch = resolve; }), }); - const replacement = makeFaviconWebContents({ id: 43 }); + const replacement = makeFaviconWebContents({ changes, id: 43 }); fromId.mockImplementation((id?: number) => id === 42 ? initial.webContents : id === 43 ? replacement.webContents : null, ); @@ -976,6 +1020,7 @@ yield* manager.subscribeStateChanges((_tabId, state) => Effect.sync(() => { states.push(state); + changes.notify(); }), ); yield* manager.createTab("tab_favicon_late_replace"); @@ -984,7 +1029,7 @@ {} as never, ["http://localhost:3200/favicon.png"] as never, ); - yield* settle(() => initial.fetch.mock.calls.length === 1); + yield* changes.until(() => initial.fetch.mock.calls.length === 1); yield* manager.registerWebview("tab_favicon_late_replace", 43); resolveFetch( @@ -992,7 +1037,7 @@ headers: { "content-type": "image/png" }, }), ); - yield* settle(() => false); + yield* flushMicrotasks(); expect(states.at(-1)?.webContentsId).toBe(43); expect( @@ -1005,14 +1050,16 @@ effectIt.effect("treats a reused WebContents id as a new attachment", () => withManager((manager) => Effect.gen(function* () { - const initial = makeFaviconWebContents({ id: 42 }); - const replacement = makeFaviconWebContents({ id: 42 }); + const changes = makeChangeSignal(); + const initial = makeFaviconWebContents({ changes, id: 42 }); + const replacement = makeFaviconWebContents({ changes, id: 42 }); let active = initial.webContents; fromId.mockImplementation(() => active); const states: PreviewManager.PreviewTabState[] = []; yield* manager.subscribeStateChanges((_tabId, state) => Effect.sync(() => { states.push(state); + changes.notify(); }), ); yield* manager.createTab("tab_favicon_reused_id"); @@ -1021,7 +1068,7 @@ {} as never, ["http://localhost:3200/favicon.png"] as never, ); + yield* changes.until(() => states.at(-1)?.favicon !== undefined); - yield* settle(() => states.at(-1)?.favicon !== undefined); active = replacement.webContents; yield* manager.registerWebview("tab_favicon_reused_id", 42); ``` ## apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts Status: modified; +276; −384 ```diff @@ -61,6 +61,7 @@ import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Path from "effect/Path"; +import * as PubSub from "effect/PubSub"; import * as Queue from "effect/Queue"; import * as Schema from "effect/Schema"; import * as Scope from "effect/Scope"; @@ -2273,15 +2274,26 @@ terminalReason: null, }); + // Harnesses notify after every recorded event, continuation, or prompt, so a + // wait re-checks exactly when the state it reads changes. Subscribing before + // the first check means a change that lands in between is not lost. + const makeChangeSignal = Effect.gen(function* () { + const changes = yield* PubSub.unbounded(); + const until = (predicate: () => boolean): Effect.Effect => + Effect.scoped( + Effect.gen(function* () { + const subscription = yield* PubSub.subscribe(changes); + while (!predicate()) yield* PubSub.take(subscription); + }), + ); + return { notify: PubSub.publish(changes, undefined), until }; - const awaitUntil = (predicate: () => boolean, label: string): Effect.Effect => + }); + + // Fixed quiet windows for asserting that nothing happens after input the + // harness cannot observe being handled (a shut-down query, a replaced process). + const yieldTicks = (ticks: number): Effect.Effect => Effect.gen(function* () { - for (let attempt = 0; attempt < 5000; attempt++) { - if (predicate()) { - return; - } - yield* Effect.yieldNow; + for (let tick = 0; tick < ticks; tick++) yield* Effect.yieldNow; - } - return yield* Effect.die(`Timed out waiting for ${label}.`); }); const makeWakeHarnessWithOptions = (options?: { @@ -2321,6 +2333,7 @@ if (offered !== undefined) yield* Deferred.succeed(offered, undefined); yield* Deferred.await(processed); }); + const changes = yield* makeChangeSignal; const offeredMessages: Array = []; const permissionModeChanges: Array = []; const liveSelectionChanges: Array = []; @@ -2362,7 +2375,10 @@ offer: (request) => Effect.sync(() => { continuationRequests.push(request); + }).pipe( + Effect.andThen(Queue.offer(continuationReceipts, request)), + Effect.andThen(changes.notify), - }).pipe(Effect.andThen(Queue.offer(continuationReceipts, request))), + ), }, queryRunner: { allocateSessionId: @@ -2398,7 +2414,7 @@ offer: (message: SDKUserMessage) => Effect.sync(() => { offeredMessages.push(message); + }).pipe(Effect.andThen(changes.notify)), - }), setModel: () => Effect.void, applyFlagSettings: () => Effect.void, setPermissionMode: (mode) => @@ -2419,7 +2435,7 @@ offer: (message) => Effect.sync(() => { offeredMessages.push(message); - }), + }).pipe(Effect.andThen(changes.notify)), setModel: (model) => Effect.sync(() => { liveSelectionChanges.push({ model }); @@ -2475,6 +2491,7 @@ if (event.type === "turn_item.updated" && event.turnItem.type === "system_notice") { yield* Queue.offer(systemNoticeReceipts, event); } + yield* changes.notify; }), ), Effect.forkScoped, @@ -2517,6 +2534,10 @@ getOpenedOptions: () => openedOptions, terminalEvents, hasPendingBackgroundWork, + until: changes.until, + awaitTerminal: (count: number) => changes.until(() => terminalEvents().length >= count), + awaitContinuation: (count: number) => + changes.until(() => continuationRequests.length >= count), }; }); const makeWakeHarness = makeWakeHarnessWithOptions(); @@ -6226,7 +6247,7 @@ ? [event.turnItem] : [], ); + yield* harness.until(() => retryItems().length === 1); - yield* awaitUntil(() => retryItems().length === 1, "Claude retry item"); const runningRetry = retryItems()[0]; assert.equal(runningRetry?.status, "running"); assert.equal(runningRetry?.failure.code, "api_error_529"); @@ -6243,7 +6264,7 @@ text: "Opening GitHub.", }), ); - yield* awaitUntil(() => retryItems().length === 2, "resolved Claude retry item"); + yield* harness.until(() => retryItems().length === 2); assert.lengthOf(harness.terminalEvents(), 0); const recoveredRetry = retryItems()[1]; assert.equal(recoveredRetry?.id, runningRetry?.id); @@ -6257,7 +6278,7 @@ result: "Opened GitHub.", }), ); + yield* harness.awaitTerminal(1); - yield* awaitUntil(() => harness.terminalEvents().length === 1, "recovered Claude turn"); }).pipe( Effect.provide( Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), @@ -6304,7 +6325,7 @@ apiErrorStatus: 529, }), ); + yield* harness.awaitTerminal(1); - yield* awaitUntil(() => harness.terminalEvents().length === 1, "failed Claude turn"); const terminal = harness.terminalEvents()[0]; assert.equal(terminal?.status, "failed"); @@ -6490,7 +6511,7 @@ ); yield* Queue.offer(harness.sdkMessages, wakeTaskStarted); yield* Queue.offer(harness.sdkMessages, turnOneResult); - yield* awaitUntil(() => harness.terminalEvents().length === 1, "first turn terminal"); + yield* harness.awaitTerminal(1); const afterStart = providerThreadRosterEvents(harness.events).filter( (event) => (event.providerThread.pendingBackgroundTasks?.length ?? 0) > 0, @@ -6519,24 +6540,20 @@ session_id: WAKE_NATIVE_SESSION, }), ); - yield* awaitUntil( - () => - providerThreadRosterEvents(harness.events).some( - (event) => - event.providerThread.pendingBackgroundTasks?.[0]?.parentTaskId === "agent-1", + yield* harness.until(() => + providerThreadRosterEvents(harness.events).some( + (event) => + event.providerThread.pendingBackgroundTasks?.[0]?.parentTaskId === "agent-1", + ), - ), - "subagent shell roster", ); yield* Queue.offer(harness.sdkMessages, emptyRoster); - yield* awaitUntil( - () => - providerThreadRosterEvents(harness.events).some( - (event) => - event.providerThread.status === "idle" && - (event.providerThread.pendingBackgroundTasks?.length ?? 0) === 0, - ), - "empty roster clear", + yield* harness.until(() => + providerThreadRosterEvents(harness.events).some( + (event) => + event.providerThread.status === "idle" && + (event.providerThread.pendingBackgroundTasks?.length ?? 0) === 0, + ), ); assert.isFalse(yield* harness.hasPendingBackgroundWork); }).pipe( @@ -6668,7 +6685,7 @@ harness.sdkMessages, makeResultFrame({ uuid: nextUuid(), result: "Watching." }), ); + yield* harness.awaitTerminal(1); - yield* awaitUntil(() => harness.terminalEvents().length === 1, "turn terminal"); const roster = providerThreadRosterEvents(harness.events).at(-1)?.providerThread .pendingBackgroundTasks; @@ -6707,7 +6724,7 @@ ); yield* Queue.offer(harness.sdkMessages, wakeTaskStarted); yield* Queue.offer(harness.sdkMessages, turnOneResult); - yield* awaitUntil(() => harness.terminalEvents().length === 1, "first turn terminal"); + yield* harness.awaitTerminal(1); const settledThread = providerThreadRosterEvents(harness.events).at(-1)?.providerThread; assert.equal(settledThread?.pendingBackgroundTasks?.[0]?.taskId, WAKE_TASK_ID); assert.isTrue(yield* harness.hasPendingBackgroundWork); @@ -6719,11 +6736,10 @@ }); assert.equal(closes, 1, "Stop must close the CLI process that owns the task"); - yield* awaitUntil( + yield* harness.until( () => (providerThreadRosterEvents(harness.events).at(-1)?.providerThread .pendingBackgroundTasks?.length ?? 0) === 0, - "roster clear after Stop", ); assert.isFalse(yield* harness.hasPendingBackgroundWork); assert.lengthOf(harness.continuationRequests, 0); @@ -6738,6 +6754,7 @@ it.effect("a settled Stop leaves a turn that replaced the closing process alone", () => Effect.scoped( Effect.gen(function* () { + const changes = yield* makeChangeSignal; const fileSystem = yield* FileSystem.FileSystem; const idAllocator = yield* IdAllocator.IdAllocatorV2; const attachmentsDir = yield* fileSystem.makeTempDirectoryScoped({ @@ -6761,7 +6778,7 @@ offer: (request) => Effect.sync(() => { continuationRequests.push(request); + }).pipe(Effect.andThen(changes.notify)), - }), }, queryRunner: { allocateSessionId: Effect.succeed(WAKE_NATIVE_SESSION), @@ -6805,7 +6822,7 @@ Stream.runForEach((event) => Effect.sync(() => { events.push(event); - }), + }).pipe(Effect.andThen(changes.notify)), ), Effect.forkScoped, ); @@ -6828,7 +6845,7 @@ ); yield* Queue.offer(processQueues[0]!, wakeTaskStarted); yield* Queue.offer(processQueues[0]!, turnOneResult); + yield* changes.until(() => terminals().length === 1); - yield* awaitUntil(() => terminals().length === 1, "first turn terminal"); const settledThread = providerThreadRosterEvents(events).at(-1)?.providerThread; const settledTurn = terminals()[0]; assert.equal(settledThread?.pendingBackgroundTasks?.[0]?.taskId, WAKE_TASK_ID); @@ -6867,18 +6884,16 @@ uuid: "00000000-0000-4000-8000-000000000905", }), ); + yield* changes.until( - yield* awaitUntil( () => providerThreadRosterEvents(events).at(-1)?.providerThread.pendingBackgroundTasks?.[0] ?.taskId === replacementTaskId, - "replacement roster", ); yield* Queue.shutdown(processQueues[0]!); yield* Fiber.join(stop); assert.isTrue(yield* hasPendingBackgroundWork); + yield* yieldTicks(50); - let quietYields = 0; - yield* awaitUntil(() => quietYields++ >= 50, "late roster events"); const latestThread = providerThreadRosterEvents(events).at(-1)?.providerThread; assert.equal(latestThread?.status, "active"); assert.deepEqual( @@ -6893,7 +6908,7 @@ result: "Started another build.", }), ); - yield* awaitUntil(() => terminals().length === 2, "replacement turn terminal"); + yield* changes.until(() => terminals().length === 2); yield* Queue.offer( processQueues[1]!, claudeSdkFrame({ @@ -6909,7 +6924,7 @@ text: WAKE_ASSISTANT_TEXT, }), ); + yield* changes.until(() => continuationRequests.length === 1); - yield* awaitUntil(() => continuationRequests.length === 1, "replacement wake"); }).pipe( Effect.provide( Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), @@ -6923,6 +6938,7 @@ () => Effect.scoped( Effect.gen(function* () { + const changes = yield* makeChangeSignal; const fileSystem = yield* FileSystem.FileSystem; const idAllocator = yield* IdAllocator.IdAllocatorV2; const attachmentsDir = yield* fileSystem.makeTempDirectoryScoped({ @@ -7007,7 +7023,7 @@ Stream.runForEach((event) => Effect.sync(() => { events.push(event); - }), + }).pipe(Effect.andThen(changes.notify)), ), Effect.forkScoped, ); @@ -7060,15 +7076,13 @@ result: "A settled with background work.", }), ); - yield* awaitUntil( - () => - events.some( - (event) => - event.type === "turn.terminal" && - event.providerThreadId === providerThreadA.id && - event.status === "completed", - ), + yield* changes.until(() => + events.some( + (event) => - "thread A terminal", + event.type === "turn.terminal" && + event.providerThreadId === providerThreadA.id && + event.status === "completed", + ), ); const rosterAAfterSettle = providerThreadRosterEvents(events).findLast( (event) => event.providerThread.id === providerThreadA.id, @@ -7091,15 +7105,13 @@ }), ); assert.equal(processQueues.length, 2); - yield* awaitUntil( - () => + yield* changes.until(() => + providerThreadRosterEvents(events).some( + (event) => + event.providerThread.id === providerThreadA.id && + event.providerThread.status === "idle" && + (event.providerThread.pendingBackgroundTasks?.length ?? 0) === 0, - providerThreadRosterEvents(events).some( + ), - (event) => - event.providerThread.id === providerThreadA.id && - event.providerThread.status === "idle" && - (event.providerThread.pendingBackgroundTasks?.length ?? 0) === 0, - ), - "sibling A roster cleared idle on process replacement", ); assert.isFalse(yield* hasPendingBackgroundWorkForThread(providerThreadA)); @@ -7118,14 +7130,12 @@ session_id: nativeIds[1], }), ); - yield* awaitUntil( + yield* changes.until(() => + providerThreadRosterEvents(events).some( + (event) => + event.providerThread.id === providerThreadB.id && + (event.providerThread.pendingBackgroundTasks?.length ?? 0) > 0, + ), - () => - providerThreadRosterEvents(events).some( - (event) => - event.providerThread.id === providerThreadB.id && - (event.providerThread.pendingBackgroundTasks?.length ?? 0) > 0, - ), - "thread B roster populated", ); assert.isTrue(yield* hasPendingBackgroundWorkForThread(providerThreadB)); assert.isTrue(yield* hasPendingBackgroundWork); @@ -7157,15 +7167,13 @@ session_id: nativeIds[1], }), ); - yield* awaitUntil( - () => - events.some( - (event) => - event.type === "turn.terminal" && - event.providerThreadId === providerThreadB.id && - event.status === "failed", - ), + yield* changes.until(() => + events.some( + (event) => - "thread B failed terminal", + event.type === "turn.terminal" && + event.providerThreadId === providerThreadB.id && + event.status === "failed", + ), ); const rosterBAfterFail = providerThreadRosterEvents(events).findLast( @@ -7264,7 +7272,7 @@ 0, ), ); - yield* awaitUntil(() => harness.terminalEvents().length === 1, "first turn terminal"); + yield* harness.awaitTerminal(1); yield* harness.runtime.startTurn( makeClaudeTestTurnInput({ @@ -7279,14 +7287,8 @@ }), ); yield* Queue.offer(harness.sdkMessages, wakeNotification); - yield* Queue.offer(harness.sdkMessages, planToolUse); // The SDK asks for permission only after streaming the tool_use. - yield* awaitUntil( + yield* harness.offerAndWait(planToolUse); - () => Queue.sizeUnsafe(harness.sdkMessages) === 0, - "the tool_use frame to be consumed", - ); - let settleYields = 0; - yield* awaitUntil(() => settleYields++ >= 50, "the tool_use frame to be handled"); const canUseTool = harness.getOpenedOptions()?.canUseTool; assert.isFunction(canUseTool); const callback = yield* Effect.promise(() => @@ -7307,10 +7309,7 @@ ); yield* Queue.offer(harness.sdkMessages, planToolResult); yield* Queue.offer(harness.sdkMessages, wakeResult); - yield* awaitUntil( - () => harness.continuationRequests.length === 1, - "continuation request", - ); + yield* harness.awaitContinuation(1); yield* Queue.offer( harness.sdkMessages, @@ -7332,7 +7331,7 @@ 1, ), ); + yield* harness.awaitTerminal(2); - yield* awaitUntil(() => harness.terminalEvents().length === 2, "user turn terminal"); yield* harness.runtime.startTurn( makeClaudeTestTurnInput({ @@ -7348,7 +7347,7 @@ messageCreationSource: "provider", }), ); - yield* awaitUntil(() => harness.terminalEvents().length === 3, "continuation terminal"); + yield* harness.awaitTerminal(3); const latestItems = new Map( harness.events.flatMap((event) => @@ -7464,7 +7463,7 @@ harness.sdkMessages, makeResultFrame({ uuid: "00000000-0000-4000-8000-000000000792", result: "Done." }), ); + yield* harness.awaitTerminal(1); - yield* awaitUntil(() => harness.terminalEvents().length === 1, "turn terminal"); const bash = harness.events.findLast( (event) => @@ -7529,7 +7528,7 @@ 0, ), ); - yield* awaitUntil(() => harness.terminalEvents().length === 1, "first turn terminal"); + yield* harness.awaitTerminal(1); yield* harness.runtime.startTurn( makeClaudeTestTurnInput({ @@ -7543,8 +7542,7 @@ runtimePolicy: approvalPolicy, }), ); + yield* harness.offerAndWait( - yield* Queue.offer( - harness.sdkMessages, claudeSdkFrame({ type: "assistant", message: { @@ -7566,12 +7564,6 @@ session_id: WAKE_NATIVE_SESSION, }), ); - yield* awaitUntil( - () => Queue.sizeUnsafe(harness.sdkMessages) === 0, - "the tool_use frame to be consumed", - ); - let settleYields = 0; - yield* awaitUntil(() => settleYields++ >= 50, "the tool_use frame to be held"); const permission = yield* Effect.promise(() => harness.getOpenedOptions()!.canUseTool!( "Bash", @@ -7584,9 +7576,8 @@ ), ).pipe(Effect.forkScoped); - yield* awaitUntil( - () => harness.events.some((event) => event.type === "runtime_request.updated"), + yield* harness.until(() => + harness.events.some((event) => event.type === "runtime_request.updated"), - "the approval request", ); const request = harness.events.findLast( (event) => event.type === "runtime_request.updated", @@ -7651,7 +7642,7 @@ 0, ), ); - yield* awaitUntil(() => harness.terminalEvents().length === 1, "first turn terminal"); + yield* harness.awaitTerminal(1); yield* harness.runtime.startTurn( makeClaudeTestTurnInput({ @@ -7683,7 +7674,7 @@ 1, ), ); - yield* awaitUntil(() => harness.terminalEvents().length === 2, "second turn terminal"); + yield* harness.awaitTerminal(2); assert.equal(harness.terminalEvents()[1]?.status, "completed"); assert.lengthOf(harness.continuationRequests, 0); assert.deepEqual( @@ -7754,7 +7745,7 @@ 0, ), ); - yield* awaitUntil(() => harness.terminalEvents().length === 1, "first turn terminal"); + yield* harness.awaitTerminal(1); yield* harness.runtime.startTurn( makeClaudeTestTurnInput({ @@ -7767,16 +7758,13 @@ providerTurnOrdinal: 2, }), ); + yield* harness.offerAndWait( - yield* Queue.offer( - harness.sdkMessages, makeAssistantTextFrame({ uuid: "00000000-0000-4000-8000-000000000703", text: "Two." }), ); - let heldYields = 0; - yield* awaitUntil(() => heldYields++ >= 50, "unechoed frame to be held"); assert.deepEqual(assistantTexts(), ["One."]); yield* Queue.shutdown(harness.sdkMessages); - yield* awaitUntil(() => harness.terminalEvents().length === 2, "second turn terminal"); + yield* harness.awaitTerminal(2); assert.deepEqual(assistantTexts(), ["One.", "Two."]); assert.equal(harness.terminalEvents()[1]?.status, "failed"); }).pipe( @@ -7814,7 +7802,7 @@ session_id: WAKE_NATIVE_SESSION, }), ); - yield* awaitUntil(() => harness.terminalEvents().length === 1, "turn terminal"); + yield* harness.awaitTerminal(1); const [terminal] = harness.terminalEvents(); assert.equal(terminal?.status, "failed"); if (terminal?.status !== "failed") return; @@ -7871,7 +7859,7 @@ user_message_uuid: harness.promptUuid(0), }), ); + yield* harness.awaitTerminal(1); - yield* awaitUntil(() => harness.terminalEvents().length === 1, "turn terminal"); assert.equal(harness.terminalEvents()[0]?.status, "completed"); }).pipe( Effect.provide( @@ -7899,27 +7887,23 @@ ); yield* Queue.offer(harness.sdkMessages, wakeTaskStarted); yield* Queue.offer(harness.sdkMessages, turnOneResult); - yield* awaitUntil(() => harness.terminalEvents().length === 1, "first turn terminal"); + yield* harness.awaitTerminal(1); assert.equal(harness.terminalEvents()[0]?.status, "completed"); assert.isTrue(yield* harness.hasPendingBackgroundWork); assert.lengthOf(harness.continuationRequests, 0); + yield* harness.offerAndWait(wakeNotification); - yield* Queue.offer(harness.sdkMessages, wakeNotification); - let quietYields = 0; - yield* awaitUntil(() => quietYields++ >= 50, "notification-only quiet window"); assert.lengthOf(harness.continuationRequests, 0); assert.isFalse(yield* harness.hasPendingBackgroundWork); yield* Queue.offer(harness.sdkMessages, wakeAssistant); - yield* awaitUntil(() => harness.continuationRequests.length === 1, "continuation request"); + yield* harness.awaitContinuation(1); assert.equal(harness.continuationRequests[0]?.threadId, harness.threadId); assert.equal(harness.continuationRequests[0]?.providerThreadId, harness.providerThread.id); assert.equal(harness.continuationRequests[0]?.driver, ClaudeAdapterV2.CLAUDE_PROVIDER); assert.equal(harness.continuationRequests[0]?.detail, WAKE_SUMMARY); - yield* Queue.offer(harness.sdkMessages, wakeResult); - let settleYields = 0; - yield* awaitUntil(() => settleYields++ >= 50, "wake result to settle into the buffer"); + yield* harness.offerAndWait(wakeResult); assert.lengthOf(harness.continuationRequests, 1); assert.lengthOf(harness.terminalEvents(), 1); assert.isTrue(yield* harness.hasPendingBackgroundWork); @@ -7949,7 +7933,7 @@ ); yield* Queue.offer(harness.sdkMessages, wakeTaskStarted); yield* Queue.offer(harness.sdkMessages, turnOneResult); + yield* harness.awaitTerminal(1); - yield* awaitUntil(() => harness.terminalEvents().length === 1, "first turn terminal"); yield* harness.offerAndWait(wakeNotification); assert.lengthOf(harness.continuationRequests, 0); @@ -7974,7 +7958,7 @@ yield* harness.offerAndWait(wakeAssistant); yield* harness.offerAndWait(wakeResult); - yield* awaitUntil(() => harness.terminalEvents().length === 2, "wake run terminal"); + yield* harness.awaitTerminal(2); assert.equal(harness.terminalEvents()[1]?.status, "completed"); assert.lengthOf(harness.continuationRequests, 1); assert.isTrue( @@ -8120,10 +8104,10 @@ ); yield* Queue.offer(harness.sdkMessages, wakeTaskStarted); yield* Queue.offer(harness.sdkMessages, turnOneResult); - yield* awaitUntil(() => harness.terminalEvents().length === 1, "first turn terminal"); + yield* harness.awaitTerminal(1); yield* Queue.offer(harness.sdkMessages, wakeNotification); yield* Queue.offer(harness.sdkMessages, wakeResult); + yield* harness.awaitContinuation(1); - yield* awaitUntil(() => harness.continuationRequests.length === 1, "continuation request"); yield* harness.runtime.startTurn( makeClaudeTestTurnInput({ @@ -8137,7 +8121,7 @@ }), ); - yield* awaitUntil(() => harness.offeredMessages.length === 2, "user prompt offered"); + yield* harness.until(() => harness.offeredMessages.length === 2); assert.lengthOf(harness.terminalEvents(), 1); assert.isTrue(yield* harness.hasPendingBackgroundWork); @@ -8148,7 +8132,7 @@ result: "The build passed; nothing else pending.", }), ); + yield* harness.awaitTerminal(2); - yield* awaitUntil(() => harness.terminalEvents().length === 2, "user turn terminal"); assert.equal(harness.terminalEvents()[1]?.status, "completed"); assert.isFalse( harness.events.some( @@ -8169,7 +8153,7 @@ messageCreationSource: "provider", }), ); - yield* awaitUntil(() => harness.terminalEvents().length === 3, "continuation terminal"); + yield* harness.awaitTerminal(3); assert.equal(harness.terminalEvents()[2]?.status, "completed"); assert.lengthOf(harness.offeredMessages, 2); assert.isTrue( @@ -8228,7 +8212,7 @@ origin: { kind: "task-notification" }, }), ); - yield* awaitUntil(() => harness.terminalEvents().length === 1, "server wake terminal"); + yield* harness.awaitTerminal(1); assert.equal(harness.terminalEvents()[0]?.status, "completed"); yield* harness.runtime.startTurn( @@ -8249,7 +8233,7 @@ result: "The background command finished.", }), ); - yield* awaitUntil(() => harness.terminalEvents().length === 2, "queued turn terminal"); + yield* harness.awaitTerminal(2); assert.equal(harness.terminalEvents()[1]?.status, "completed"); }).pipe( Effect.provide( @@ -8282,7 +8266,7 @@ attachments: [], }), ); + yield* harness.until(() => harness.offeredMessages.length === 1); - yield* awaitUntil(() => harness.offeredMessages.length === 1, "recovery prompt offered"); // Live interleaving seen after interrupt recovery: a stale stopped // task_notification and its task-notification-origin result arrive @@ -8310,10 +8294,7 @@ }), ); - yield* awaitUntil( + yield* harness.until(() => hasMessageText(probeAssistantText)); - () => hasMessageText(probeAssistantText), - "probe assistant after stale task-notification result", - ); assert.lengthOf(harness.terminalEvents(), 0); assert.isFalse(hasMessageText(staleResultText)); @@ -8332,7 +8313,7 @@ }), ); - yield* awaitUntil(() => harness.terminalEvents().length === 1, "user turn terminal"); + yield* harness.awaitTerminal(1); assert.equal(harness.terminalEvents()[0]?.status, "completed"); assert.isTrue(hasMessageText(recoveryAssistantText)); assert.isFalse(hasMessageText(staleResultText)); @@ -8390,12 +8371,7 @@ origin: { kind: "task-notification" }, }), ); - const terminalized = Exit.isSuccess( - yield* awaitUntil( - () => harness.terminalEvents().length === 1, - "interrupted terminal", + const terminalized = Exit.isSuccess(yield* harness.awaitTerminal(1).pipe(Effect.exit)); - ).pipe(Effect.exit), - ); yield* Deferred.succeed(closeGate, undefined); assert.isTrue(terminalized); assert.equal(harness.terminalEvents()[0]?.status, "interrupted"); @@ -8427,7 +8403,7 @@ }), ); + yield* harness.awaitTerminal(1); - yield* awaitUntil(() => harness.terminalEvents().length === 1, "spurious terminal"); assert.equal(harness.terminalEvents()[0]?.status, "completed"); assert.lengthOf(harness.offeredMessages, 0); }).pipe( @@ -8590,7 +8566,7 @@ harness.sdkMessages, makeResultFrame({ uuid: "00000000-0000-4000-8000-000000000952", result: "Started." }), ); + yield* harness.awaitTerminal(1); - yield* awaitUntil(() => harness.terminalEvents().length === 1, "first turn terminal"); // Both subagents end together. Claude runs an empty turn for the first // notification and opens the real one a millisecond later, all before @@ -8660,7 +8636,7 @@ yield* harness.offerAndWait(wakeAssistant); yield* harness.offerAndWait(wakeResult); - yield* awaitUntil(() => harness.terminalEvents().length === 2, "wake run terminal"); + yield* harness.awaitTerminal(2); assert.equal(harness.terminalEvents()[1]?.status, "completed"); assert.lengthOf(harness.continuationRequests, 1); assert.isTrue( @@ -8712,9 +8688,9 @@ }), ); yield* Queue.offer(harness.sdkMessages, result("00000000-0000-4000-8000-000000000902")); - yield* awaitUntil(() => harness.terminalEvents().length === 1, "first turn terminal"); + yield* harness.awaitTerminal(1); yield* Queue.offer(harness.sdkMessages, ended("00000000-0000-4000-8000-000000000903")); + yield* harness.awaitContinuation(1); - yield* awaitUntil(() => harness.continuationRequests.length === 1, "subagent wake"); yield* Queue.offer(harness.sdkMessages, result("00000000-0000-4000-8000-000000000904")); yield* harness.runtime.startTurn( makeClaudeTestTurnInput({ @@ -8729,7 +8705,7 @@ messageCreationSource: "provider", }), ); - yield* awaitUntil(() => harness.terminalEvents().length === 2, "wake terminal"); + yield* harness.awaitTerminal(2); yield* userTurn("attempt-rerun-3", 3); yield* Queue.offer( @@ -8746,11 +8722,11 @@ yield* Queue.offer(harness.sdkMessages, ended("00000000-0000-4000-8000-000000000906")); yield* Queue.offer(harness.sdkMessages, wakeTaskStarted); yield* Queue.offer(harness.sdkMessages, result("00000000-0000-4000-8000-000000000907")); + yield* harness.awaitTerminal(3); - yield* awaitUntil(() => harness.terminalEvents().length === 3, "third turn terminal"); yield* Queue.offer(harness.sdkMessages, wakeNotification); yield* Queue.offer(harness.sdkMessages, wakeAssistant); - yield* awaitUntil(() => harness.continuationRequests.length === 2, "command wake"); + yield* harness.awaitContinuation(2); assert.equal( harness.continuationRequests[1]?.notification?.summary, `Command "${WAKE_TASK_DESCRIPTION}" finished`, @@ -8860,7 +8836,7 @@ 0, ), ); - yield* awaitUntil(() => harness.terminalEvents().length === 1, "first turn terminal"); + yield* harness.awaitTerminal(1); yield* harness.runtime.startTurn( makeClaudeTestTurnInput({ @@ -8937,7 +8913,7 @@ for (const frame of wakeFrames) { yield* Queue.offer(harness.sdkMessages, frame); } + yield* harness.awaitTerminal(2); - yield* awaitUntil(() => harness.terminalEvents().length === 2, "user turn terminal"); yield* harness.runtime.startTurn( makeClaudeTestTurnInput({ @@ -8952,7 +8928,7 @@ messageCreationSource: "provider", }), ); - yield* awaitUntil(() => harness.terminalEvents().length === 3, "continuation terminal"); + yield* harness.awaitTerminal(3); const subagentRuns = harness.events.flatMap((event) => event.type === "subagent.updated" && event.subagent.nativeTaskRef?.nativeId === TASK_ID @@ -9008,7 +8984,7 @@ result: "Waiting in the background.", }), ); + yield* harness.awaitTerminal(1); - yield* awaitUntil(() => harness.terminalEvents().length === 1, "first turn terminal"); const idleFrames = [ makeSubagentTaskStartedFrame({ @@ -9042,7 +9018,7 @@ for (const frame of idleFrames) { yield* Queue.offer(harness.sdkMessages, frame); } - yield* awaitUntil(() => harness.continuationRequests.length === 1, "continuation request"); + yield* harness.awaitContinuation(1); yield* Queue.offer( harness.sdkMessages, makeResultFrame({ @@ -9063,7 +9039,7 @@ messageCreationSource: "provider", }), ); + yield* harness.awaitTerminal(2); - yield* awaitUntil(() => harness.terminalEvents().length === 2, "continuation terminal"); const routing = subagentRouting(harness.events, ["toolu-idle-bash"]); assert.isDefined(routing.childThreadId); @@ -9135,7 +9111,7 @@ ]) { yield* Queue.offer(harness.sdkMessages, frame); } + yield* harness.awaitTerminal(1); - yield* awaitUntil(() => harness.terminalEvents().length === 1, "launch turn terminal"); for (const frame of [ makeSubagentToolResultFrame({ @@ -9160,18 +9136,16 @@ ]) { yield* Queue.offer(harness.sdkMessages, frame); } - yield* awaitUntil(() => { + yield* harness.until(() => { const event = latestToolItem(IDLE_BASH); return event?.type === "turn_item.updated" && event.turnItem.status === "running"; + }); + yield* harness.until(() => - }, "idle Bash call"); - yield* awaitUntil( - () => + harness.events.some( + (event) => + event.type === "subagent.updated" && + event.subagent.progress === "Reading the last commits", + ), - harness.events.some( - (event) => - event.type === "subagent.updated" && - event.subagent.progress === "Reading the last commits", - ), - "idle progress", ); const routing = subagentRouting(harness.events, [STRADDLING_BASH, IDLE_BASH]); assert.deepEqual(routing.assistantTexts(routing.childThreadId), ["Still auditing."]); @@ -9200,7 +9174,7 @@ ]) { yield* Queue.offer(harness.sdkMessages, frame); } + yield* harness.awaitContinuation(1); - yield* awaitUntil(() => harness.continuationRequests.length === 1, "continuation request"); yield* Queue.offer( harness.sdkMessages, makeResultFrame({ @@ -9221,7 +9195,7 @@ messageCreationSource: "provider", }), ); - yield* awaitUntil(() => harness.terminalEvents().length === 2, "continuation terminal"); + yield* harness.awaitTerminal(2); const finalRouting = subagentRouting(harness.events, [STRADDLING_BASH, IDLE_BASH]); assert.deepEqual(finalRouting.assistantTexts(finalRouting.childThreadId), [ @@ -9285,7 +9259,7 @@ ]) { yield* Queue.offer(harness.sdkMessages, frame); } - yield* awaitUntil(() => harness.terminalEvents().length === 1, "launch turn terminal"); + yield* harness.awaitTerminal(1); for (const frame of [ claudeSdkFrame({ @@ -9339,12 +9313,10 @@ ]) { yield* Queue.offer(harness.sdkMessages, frame); } - yield* awaitUntil( - () => - harness.events.some( + yield* harness.until(() => + harness.events.some( - (event) => event.type === "message.updated" && event.message.text === "OUTER_WORKING", - ), - "outer work while the nested start waits", + (event) => event.type === "message.updated" && event.message.text === "OUTER_WORKING", + ), ); assert.lengthOf(harness.continuationRequests, 0); for (const frame of [ @@ -9378,15 +9350,13 @@ ]) { yield* Queue.offer(harness.sdkMessages, frame); } + yield* harness.awaitContinuation(1); + yield* harness.until(() => + harness.events.some( + (event) => + event.type === "subagent.updated" && + event.subagent.progress === "Waiting on the nested check", + ), - yield* awaitUntil(() => harness.continuationRequests.length === 1, "continuation request"); - yield* awaitUntil( - () => - harness.events.some( - (event) => - event.type === "subagent.updated" && - event.subagent.progress === "Waiting on the nested check", - ), - "outer progress", ); yield* Queue.offer( harness.sdkMessages, @@ -9408,7 +9378,7 @@ messageCreationSource: "provider", }), ); + yield* harness.awaitTerminal(2); - yield* awaitUntil(() => harness.terminalEvents().length === 2, "continuation terminal"); assert.isFalse( harness.events.some( @@ -9510,13 +9480,13 @@ ]) { yield* Queue.offer(harness.sdkMessages, frame); } - yield* awaitUntil(() => harness.terminalEvents().length === 1, "first turn terminal"); + yield* harness.awaitTerminal(1); yield* userTurn("attempt-claude-earlier-nested-2", 2); yield* Queue.offer( harness.sdkMessages, makeResultFrame({ uuid: "00000000-0000-4000-8000-000000000a05", result: "Still on it." }), ); + yield* harness.awaitTerminal(2); - yield* awaitUntil(() => harness.terminalEvents().length === 2, "second turn terminal"); for (const frame of [ makeSubagentNotificationFrame({ @@ -9538,13 +9508,11 @@ ]) { yield* Queue.offer(harness.sdkMessages, frame); } + yield* harness.until(() => + harness.events.some( + (event) => + event.type === "message.updated" && event.message.text === "OUTER_AFTER_NESTED", + ), - yield* awaitUntil( - () => - harness.events.some( - (event) => - event.type === "message.updated" && event.message.text === "OUTER_AFTER_NESTED", - ), - "outer work after the nested call returns", ); assert.isFalse( @@ -9692,7 +9660,7 @@ ]) { yield* Queue.offer(harness.sdkMessages, frame); } - yield* awaitUntil(() => harness.terminalEvents().length === 1, "launch turn terminal"); + yield* harness.awaitTerminal(1); for (const frame of makeSubagentAssistantFrames({ parentToolUseId: OUTER_TOOL_USE_ID, @@ -9701,7 +9669,7 @@ })) { yield* Queue.offer(harness.sdkMessages, frame); } + yield* harness.until(() => bashItems().length === 1); - yield* awaitUntil(() => bashItems().length === 1, "idle Bash call"); const bashResultFrame = makeSubagentToolResultFrame({ parentToolUseId: OUTER_TOOL_USE_ID, @@ -9770,14 +9738,10 @@ ]) { yield* Queue.offer(harness.sdkMessages, frame); } - yield* awaitUntil( + yield* harness.awaitContinuation(1); - () => harness.continuationRequests.length === 1, - "continuation request", - ); yield* Queue.shutdown(harness.sdkMessages); - let quietYields = 0; + yield* yieldTicks(50); - yield* awaitUntil(() => quietYields++ >= 50, "query exit"); yield* harness.runtime.startTurn( makeClaudeTestTurnInput({ threadId: harness.threadId, @@ -9799,7 +9763,7 @@ result: "The audit finished.", }), ); - yield* awaitUntil(() => harness.terminalEvents().length === 2, "continuation terminal"); + yield* harness.awaitTerminal(2); const input = JSON.stringify("git log -5"); assert.deepEqual(bashItems(), [ @@ -9869,7 +9833,7 @@ ]) { yield* Queue.offer(harness.sdkMessages, frame); } + yield* harness.awaitTerminal(1); - yield* awaitUntil(() => harness.terminalEvents().length === 1, "launch turn terminal"); yield* Queue.offer( harness.sdkMessages, @@ -9887,7 +9851,7 @@ session_id: WAKE_NATIVE_SESSION, }), ); + yield* harness.until(() => sendItems().length >= 1); - yield* awaitUntil(() => sendItems().length >= 1, "idle SendMessage call"); for (const frame of [ claudeSdkFrame({ @@ -9921,15 +9885,11 @@ ]) { yield* Queue.offer(harness.sdkMessages, frame); } - yield* awaitUntil( - () => harness.continuationRequests.length === 1, - "continuation request", - ); + yield* harness.awaitContinuation(1); const beforeExit = sendItems(); yield* Queue.shutdown(harness.sdkMessages); + yield* yieldTicks(50); - let quietYields = 0; - yield* awaitUntil(() => quietYields++ >= 50, "query exit"); yield* harness.runtime.startTurn( makeClaudeTestTurnInput({ threadId: harness.threadId, @@ -9951,7 +9911,7 @@ result: "The audit finished.", }), ); + yield* harness.awaitTerminal(2); - yield* awaitUntil(() => harness.terminalEvents().length === 2, "continuation terminal"); const input = JSON.stringify(SEND_INPUT); assert.deepEqual(beforeExit, [`dynamic_tool:running:${input}`]); @@ -10017,7 +9977,7 @@ ]) { yield* Queue.offer(harness.sdkMessages, frame); } + yield* harness.awaitTerminal(1); - yield* awaitUntil(() => harness.terminalEvents().length === 1, "launch turn terminal"); const otherStartFrame = claudeSdkFrame({ type: "system", @@ -10071,10 +10031,7 @@ ]) { yield* Queue.offer(harness.sdkMessages, frame); } - yield* awaitUntil( - () => harness.continuationRequests.length === 1, - "continuation request", + yield* harness.awaitContinuation(1); - ); yield* Queue.offer( harness.sdkMessages, makeResultFrame({ @@ -10095,7 +10052,7 @@ messageCreationSource: "provider", }), ); - yield* awaitUntil(() => harness.terminalEvents().length === 2, "continuation terminal"); + yield* harness.awaitTerminal(2); const routing = subagentRouting(harness.events, [BASH]); const childTexts = routing.assistantTexts(routing.childThreadId); @@ -10185,7 +10142,7 @@ ? [] : routing.assistantTexts(routing.childThreadId); }; - yield* awaitUntil(() => childTexts().length === 1, "replayed subagent text"); + yield* harness.until(() => childTexts().length === 1); const retryStatuses = harness.events.flatMap((event) => event.type === "turn_item.updated" && event.turnItem.type === "error" && @@ -10271,7 +10228,7 @@ for (const frame of frames) { yield* Queue.offer(harness.sdkMessages, frame); } + yield* harness.awaitTerminal(1); - yield* awaitUntil(() => harness.terminalEvents().length === 1, "turn terminal"); const routing = subagentRouting(harness.events, ["toolu-early-bash"]); assert.isDefined(routing.childThreadId); @@ -10992,7 +10949,7 @@ session_id: WAKE_NATIVE_SESSION, }), ); - yield* awaitUntil(() => subagentEvents().length === 1, "subagent node created"); + yield* harness.until(() => subagentEvents().length === 1); yield* Queue.offer( harness.sdkMessages, @@ -11016,10 +10973,7 @@ session_id: WAKE_NATIVE_SESSION, }), ); + yield* harness.until(() => subagentEvents().at(-1)?.subagent.status === "completed"); - yield* awaitUntil( - () => subagentEvents().at(-1)?.subagent.status === "completed", - "subagent terminal", - ); assert.equal(subagentEvents().at(-1)?.subagent.result, "First line.\nSecond line."); @@ -11030,7 +10984,7 @@ result: "Delegation completed.", }), ); + yield* harness.awaitTerminal(1); - yield* awaitUntil(() => harness.terminalEvents().length === 1, "turn terminal"); }).pipe( Effect.provide( Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), @@ -11090,7 +11044,7 @@ }), ); yield* Queue.offer(harness.sdkMessages, subagentTaskStarted); - yield* awaitUntil(() => subagentEvents().length >= 1, "subagent node created"); + yield* harness.until(() => subagentEvents().length >= 1); yield* Queue.offer( harness.sdkMessages, makeResultFrame({ @@ -11098,11 +11052,11 @@ result: "Spawned the subagent in the background.", }), ); + yield* harness.awaitTerminal(1); - yield* awaitUntil(() => harness.terminalEvents().length === 1, "first turn terminal"); assert.isTrue(yield* harness.hasPendingBackgroundWork); yield* Queue.offer(harness.sdkMessages, subagentStoppedNotification); + yield* harness.awaitContinuation(1); - yield* awaitUntil(() => harness.continuationRequests.length === 1, "continuation request"); yield* Queue.offer( harness.sdkMessages, @@ -11124,7 +11078,7 @@ messageCreationSource: "provider", }), ); - yield* awaitUntil(() => harness.terminalEvents().length === 2, "continuation terminal"); + yield* harness.awaitTerminal(2); assert.equal(subagentEvents().at(-1)?.subagent.status, "cancelled"); assert.isFalse(yield* harness.hasPendingBackgroundWork); @@ -11213,7 +11167,7 @@ }), ); yield* Queue.offer(harness.sdkMessages, subagentTaskStarted); - yield* awaitUntil(() => subagentEvents().length >= 1, "subagent node created"); + yield* harness.until(() => subagentEvents().length >= 1); yield* Queue.offer( harness.sdkMessages, makeResultFrame({ @@ -11221,13 +11175,10 @@ result: "Spawned the subagent in the background.", }), ); + yield* harness.awaitTerminal(1); - yield* awaitUntil(() => harness.terminalEvents().length === 1, "first turn terminal"); yield* Queue.offer(harness.sdkMessages, firstNotification); - yield* awaitUntil( - () => harness.continuationRequests.length === 1, + yield* harness.awaitContinuation(1); - "first continuation request", - ); yield* Queue.offer( harness.sdkMessages, makeResultFrame({ @@ -11248,7 +11199,7 @@ messageCreationSource: "provider", }), ); + yield* harness.awaitTerminal(2); - yield* awaitUntil(() => harness.terminalEvents().length === 2, "continuation terminal"); assert.equal(subagentEvents().at(-1)?.subagent.status, "completed"); assert.equal(subagentEvents().at(-1)?.subagent.result, FIRST_SUMMARY); assert.isFalse(yield* harness.hasPendingBackgroundWork); @@ -11331,10 +11282,7 @@ yield* Queue.offer(harness.sdkMessages, frame); } yield* Queue.offer(harness.sdkMessages, secondNotification); - yield* awaitUntil( - () => harness.continuationRequests.length === 2, - "second continuation request", - ); + yield* harness.awaitContinuation(2); assert.equal(harness.continuationRequests[1]?.detail, SECOND_SUMMARY); assert.isTrue(yield* harness.hasPendingBackgroundWork); @@ -11357,11 +11305,8 @@ messageCreatedBy: "agent", messageCreationSource: "provider", }), - ); - yield* awaitUntil( - () => harness.terminalEvents().length === 3, - "resume continuation terminal", ); + yield* harness.awaitTerminal(3); const statuses = subagentEvents().map((event) => event.subagent.status); const firstCompleted = statuses.indexOf("completed"); @@ -11408,6 +11353,7 @@ it.effect("resets Waiting roster and wake eligibility when the CLI process is replaced", () => Effect.scoped( Effect.gen(function* () { + const changes = yield* makeChangeSignal; const fileSystem = yield* FileSystem.FileSystem; const idAllocator = yield* IdAllocator.IdAllocatorV2; const attachmentsDir = yield* fileSystem.makeTempDirectoryScoped({ @@ -11430,7 +11376,7 @@ offer: (request) => Effect.sync(() => { continuationRequests.push(request); + }).pipe(Effect.andThen(changes.notify)), - }), }, queryRunner: { allocateSessionId: Effect.succeed(WAKE_NATIVE_SESSION), @@ -11471,7 +11417,7 @@ Stream.runForEach((event) => Effect.sync(() => { events.push(event); - }), + }).pipe(Effect.andThen(changes.notify)), ), Effect.forkScoped, ); @@ -11505,10 +11451,7 @@ }), ); yield* Queue.offer(firstProcess, turnOneResult); - yield* awaitUntil( - () => events.some((event) => event.type === "turn.terminal"), - "first turn terminal", - ); + yield* changes.until(() => events.some((event) => event.type === "turn.terminal")); assert.isFalse(yield* hasPendingBackgroundWork); const alternateModel = CLAUDE_TEST_RESTART_MODEL_SELECTION; @@ -11530,15 +11473,13 @@ ); assert.equal(processQueues.length, 2); + yield* changes.until(() => + providerThreadRosterEvents(events).some( + (event) => + event.providerThread.status === "active" && + (event.providerThread.pendingBackgroundTasks?.length ?? 0) === 0 && + event.providerThread.updatedAt !== undefined, + ), - yield* awaitUntil( - () => - providerThreadRosterEvents(events).some( - (event) => - event.providerThread.status === "active" && - (event.providerThread.pendingBackgroundTasks?.length ?? 0) === 0 && - event.providerThread.updatedAt !== undefined, - ), - "roster cleared on process replace while remaining active", ); assert.isFalse(yield* hasPendingBackgroundWork); const emptyActiveRosterEvents = providerThreadRosterEvents(events).filter( @@ -11555,8 +11496,7 @@ const secondProcess = processQueues[1]!; yield* Queue.offer(secondProcess, wakeNotification); + yield* yieldTicks(50); - let settleYields = 0; - yield* awaitUntil(() => settleYields++ >= 50, "stale notification settle"); assert.lengthOf(continuationRequests, 0); yield* Queue.offer( @@ -11566,9 +11506,8 @@ result: "Process restart turn finished.", }), ); + yield* changes.until( - yield* awaitUntil( () => events.filter((event) => event.type === "turn.terminal").length === 2, - "second turn terminal", ); }).pipe( Effect.provide( @@ -11583,6 +11522,7 @@ () => Effect.scoped( Effect.gen(function* () { + const changes = yield* makeChangeSignal; const fileSystem = yield* FileSystem.FileSystem; const idAllocator = yield* IdAllocator.IdAllocatorV2; const attachmentsDir = yield* fileSystem.makeTempDirectoryScoped({ @@ -11605,7 +11545,7 @@ offer: (request) => Effect.sync(() => { continuationRequests.push(request); - }), + }).pipe(Effect.andThen(changes.notify)), }, queryRunner: { allocateSessionId: Effect.succeed(WAKE_NATIVE_SESSION), @@ -11646,7 +11586,7 @@ Stream.runForEach((event) => Effect.sync(() => { events.push(event); - }), + }).pipe(Effect.andThen(changes.notify)), ), Effect.forkScoped, ); @@ -11672,15 +11612,11 @@ const firstProcess = processQueues[0]!; yield* Queue.offer(firstProcess, wakeTaskStarted); yield* Queue.offer(firstProcess, turnOneResult); - yield* awaitUntil( - () => events.some((event) => event.type === "turn.terminal"), - "first turn terminal", - ); + yield* changes.until(() => events.some((event) => event.type === "turn.terminal")); assert.isTrue(yield* hasPendingBackgroundWork); yield* Queue.offer(firstProcess, wakeNotification); + yield* yieldTicks(50); - let quietYields = 0; - yield* awaitUntil(() => quietYields++ >= 50, "notification-only quiet window"); assert.lengthOf(continuationRequests, 0); const alternateModel = CLAUDE_TEST_RESTART_MODEL_SELECTION; @@ -11705,15 +11641,14 @@ result: "User turn finished after model switch.", }), ); - yield* awaitUntil( + yield* changes.until( () => events.filter((event) => event.type === "turn.terminal").length === 2, - "user turn terminal after replace", ); assert.isFalse(yield* hasPendingBackgroundWork); assert.lengthOf(continuationRequests, 0); yield* Queue.offer(secondProcess, wakeResult); - yield* awaitUntil(() => continuationRequests.length === 1, "continuation after result"); + yield* changes.until(() => continuationRequests.length === 1); assert.equal(continuationRequests[0]?.detail, WAKE_SUMMARY); yield* runtime.startTurn( makeClaudeTestTurnInput({ @@ -11729,9 +11664,8 @@ messageCreationSource: "provider", }), ); - yield* awaitUntil( + yield* changes.until( () => events.filter((event) => event.type === "turn.terminal").length === 3, - "continuation terminal after buffered drain", ); assert.isTrue( events.some( @@ -11767,6 +11701,7 @@ () => Effect.scoped( Effect.gen(function* () { + const changes = yield* makeChangeSignal; const SUBAGENT_TASK_ID = "task-buffer-replace-subagent"; const SUBAGENT_TOOL_USE_ID = "toolu-buffer-replace-subagent"; const SUBAGENT_SUMMARY = "SUB_BUFFER_REPLACE_DONE"; @@ -11838,7 +11773,7 @@ offer: (request) => Effect.sync(() => { continuationRequests.push(request); + }).pipe(Effect.andThen(changes.notify)), - }), }, queryRunner: { allocateSessionId: Effect.succeed(WAKE_NATIVE_SESSION), @@ -11881,7 +11816,7 @@ Stream.runForEach((event) => Effect.sync(() => { events.push(event); - }), + }).pipe(Effect.andThen(changes.notify)), ), Effect.forkScoped, ); @@ -11915,7 +11850,7 @@ assert.equal(processQueues.length, 1); const firstProcess = processQueues[0]!; yield* Queue.offer(firstProcess, subagentTaskStarted); + yield* changes.until(() => subagentEvents().length >= 1); - yield* awaitUntil(() => subagentEvents().length >= 1, "subagent node created"); assert.equal(subagentEvents()[0]?.subagent.status, "running"); yield* Queue.offer(firstProcess, subagentAsyncAck); yield* Queue.offer( @@ -11925,14 +11860,11 @@ result: "Spawned the subagent in the background.", }), ); + yield* changes.until(() => events.some((event) => event.type === "turn.terminal")); - yield* awaitUntil( - () => events.some((event) => event.type === "turn.terminal"), - "first turn terminal", - ); assert.isTrue(yield* hasPendingBackgroundWork); yield* Queue.offer(firstProcess, subagentNotification); - yield* awaitUntil(() => continuationRequests.length === 1, "continuation after notify"); + yield* changes.until(() => continuationRequests.length === 1); assert.equal(continuationRequests[0]?.detail, SUBAGENT_SUMMARY); const alternateModel = CLAUDE_TEST_RESTART_MODEL_SELECTION; @@ -11957,9 +11889,8 @@ result: "User turn finished after model switch.", }), ); + yield* changes.until( - yield* awaitUntil( () => events.filter((event) => event.type === "turn.terminal").length === 2, - "user turn terminal after replace", ); assert.isFalse(yield* hasPendingBackgroundWork); assert.equal(subagentEvents().at(-1)?.subagent.status, "completed"); @@ -11986,9 +11917,8 @@ messageCreationSource: "provider", }), ); - yield* awaitUntil( + yield* changes.until( () => events.filter((event) => event.type === "turn.terminal").length === 3, - "continuation terminal after buffered subagent drain", ); const finalSubagent = subagentEvents().at(-1)?.subagent; @@ -12016,6 +11946,7 @@ it.effect("refuses a thinking change that would kill a running background subagent", () => Effect.scoped( Effect.gen(function* () { + const changes = yield* makeChangeSignal; const SUBAGENT_TASK_ID = "task-model-change-running-subagent"; const SUBAGENT_TOOL_USE_ID = "toolu-model-change-running-subagent"; const fileSystem = yield* FileSystem.FileSystem; @@ -12076,7 +12007,7 @@ Stream.runForEach((event) => Effect.sync(() => { events.push(event); - }), + }).pipe(Effect.andThen(changes.notify)), ), Effect.forkScoped, ); @@ -12116,7 +12047,7 @@ result: "Spawned the subagent in the background.", }), ); + yield* changes.until(() => terminals().length === 1); - yield* awaitUntil(() => terminals().length === 1, "first turn terminal"); const settledTurn = terminals()[0]!; const alternateModel = CLAUDE_TEST_RESTART_MODEL_SELECTION; @@ -12157,7 +12088,7 @@ result: "Switched model.", }), ); + yield* changes.until(() => terminals().length === 2); - yield* awaitUntil(() => terminals().length === 2, "switched turn terminal"); yield* runtime.startTurn( makeClaudeTestTurnInput({ threadId, @@ -12313,6 +12244,7 @@ () => Effect.scoped( Effect.gen(function* () { + const changes = yield* makeChangeSignal; const fileSystem = yield* FileSystem.FileSystem; const idAllocator = yield* IdAllocator.IdAllocatorV2; const attachmentsDir = yield* fileSystem.makeTempDirectoryScoped({ @@ -12382,7 +12314,7 @@ Stream.runForEach((event) => Effect.sync(() => { events.push(event); + }).pipe(Effect.andThen(changes.notify)), - }), ), Effect.forkScoped, ); @@ -12407,10 +12339,7 @@ assert.equal(processQueues.length, 1); yield* Queue.offer(processQueues[0]!, wakeTaskStarted); yield* Queue.offer(processQueues[0]!, turnOneResult); - yield* awaitUntil( - () => events.some((event) => event.type === "turn.terminal"), - "first turn terminal", - ); + yield* changes.until(() => events.some((event) => event.type === "turn.terminal")); assert.isTrue(yield* hasPendingBackgroundWork); const alternateModel = CLAUDE_TEST_RESTART_MODEL_SELECTION; @@ -12444,6 +12373,7 @@ () => Effect.scoped( Effect.gen(function* () { + const changes = yield* makeChangeSignal; const fileSystem = yield* FileSystem.FileSystem; const idAllocator = yield* IdAllocator.IdAllocatorV2; const attachmentsDir = yield* fileSystem.makeTempDirectoryScoped({ @@ -12467,7 +12397,7 @@ offer: (request) => Effect.sync(() => { continuationRequests.push(request); + }).pipe(Effect.andThen(changes.notify)), - }), }, queryRunner: { allocateSessionId: Effect.succeed(WAKE_NATIVE_SESSION), @@ -12520,7 +12450,7 @@ Stream.runForEach((event) => Effect.sync(() => { events.push(event); + }).pipe(Effect.andThen(changes.notify)), - }), ), Effect.forkScoped, ); @@ -12544,13 +12474,12 @@ ); yield* Queue.offer(processQueues[0]!, wakeTaskStarted); yield* Queue.offer(processQueues[0]!, turnOneResult); + yield* changes.until( - yield* awaitUntil( () => events.filter((event) => event.type === "turn.terminal").length === 1, - "first turn terminal", ); yield* Queue.offer(processQueues[0]!, wakeNotification); yield* Queue.offer(processQueues[0]!, wakeAssistant); + yield* changes.until(() => continuationRequests.length === 1); - yield* awaitUntil(() => continuationRequests.length === 1, "first continuation request"); assert.isTrue(yield* hasPendingBackgroundWork); const alternateModel = CLAUDE_TEST_RESTART_MODEL_SELECTION; @@ -12606,9 +12535,8 @@ result: "Kicked off the retry build in the background.", }), ); - yield* awaitUntil( + yield* changes.until( () => events.filter((event) => event.type === "turn.terminal").length === 2, - "retry turn terminal", ); yield* Queue.offer( retryProcess, @@ -12631,10 +12559,7 @@ text: "The retry build has finished.", }), ); - yield* awaitUntil( + yield* changes.until(() => continuationRequests.length === 2); - () => continuationRequests.length === 2, - "continuation request after retry", - ); assert.equal(continuationRequests[1]?.detail, "Retry build completed successfully"); }).pipe( Effect.provide( @@ -12828,10 +12753,7 @@ yield* input.harness.offerAndWait( workflowSnapshot({ uuid: input.snapshotUuid, state: input.state ?? "start" }), ); - yield* awaitUntil( - () => workflowMemberEvents(input.harness.events, 2).length === 1, - "workflow members projected", + yield* input.harness.until(() => workflowMemberEvents(input.harness.events, 2).length === 1); - ); }); it.effect.each(["Agent", "Workflow"] as const)( @@ -12961,16 +12883,6 @@ session_id: WAKE_NATIVE_SESSION, }), ); - yield* awaitUntil( - () => - workflowCoordinatorEvents(harness.events).length > 0 && - harness.events.some( - (event) => - event.type === "subagent.updated" && - event.subagent.nativeTaskRef?.nativeId === "task-plain-auditor", - ), - "workflow coordinator and auditor projected", - ); const subagentPrompt = (nativeTaskId: string) => { const subagent = harness.events.flatMap((event) => event.type === "subagent.updated" && @@ -12978,7 +12890,6 @@ ? [event.subagent] : [], )[0]; - assert.isDefined(subagent); const message = threadMessages(harness.events, subagent?.childThreadId).find( (event) => event.message.role === "user", )?.message; @@ -12997,6 +12908,13 @@ : undefined, }; }; + // The user message item lands after the subagent node, so wait on the item itself. + yield* harness.until( + () => + workflowCoordinatorEvents(harness.events).length > 0 && + subagentPrompt(WORKFLOW_TASK_ID).itemMark !== undefined && + subagentPrompt("task-plain-auditor").itemMark !== undefined, + ); assert.deepEqual(subagentPrompt(WORKFLOW_TASK_ID), { text: script, messageMark: true, @@ -13321,10 +13239,7 @@ secondState: "queued", }), ); + yield* harness.until(() => coordinatorTurnItems().length > turnItemsBeforeRepeat); - yield* awaitUntil( - () => coordinatorTurnItems().length > turnItemsBeforeRepeat, - "repeated workflow snapshot", - ); assert.lengthOf(workflowMemberEvents(harness.events, 1), 2); assert.lengthOf(workflowMemberEvents(harness.events, 2), 1); @@ -13350,9 +13265,12 @@ model: "claude-sonnet-4-6", }), ); + // The member's node and app thread pick up the metadata after the coordinator item. + yield* harness.until( + () => + coordinatorTurnItems().length > turnItemsBeforeMetadata && + workflowMemberEvents(harness.events, 1).at(-1)?.subagent.title === "alpha:review" && - yield* awaitUntil( + firstThreadEvents().at(-1)?.appThread.title === "alpha:review", - () => coordinatorTurnItems().length > turnItemsBeforeMetadata, - "updated workflow member metadata", ); assert.equal( workflowMemberEvents(harness.events, 1).at(-1)?.subagent.title, @@ -13575,7 +13493,7 @@ harness.sdkMessages, workflowSnapshot({ uuid: "00000000-0000-4000-8000-000000001121", state: "start" }), ); + yield* harness.until(() => coordinatorCount() > inTurnCount); - yield* awaitUntil(() => coordinatorCount() > inTurnCount, "in-turn workflow snapshot"); assert.equal(coordinatorCount(), inTurnCount + 1); yield* Queue.offer( @@ -13592,11 +13510,8 @@ yield* Queue.offer( harness.sdkMessages, workflowSnapshot({ uuid: "00000000-0000-4000-8000-000000001123", state: "done" }), - ); - yield* awaitUntil( - () => coordinatorCount() > betweenTurnCount, - "between-turn workflow snapshot", ); + yield* harness.until(() => coordinatorCount() > betweenTurnCount); assert.equal(coordinatorCount(), betweenTurnCount + 1); }).pipe( Effect.provide( @@ -13648,7 +13563,7 @@ })) { yield* harness.offerAndWait(frame); } + yield* harness.until(() => toolUpdates(oldBash).length === 1); - yield* awaitUntil(() => toolUpdates(oldBash).length === 1, "old workflow tool running"); readsUntilPause = 2; const stop = yield* harness.runtime .interruptTurn({ @@ -13692,10 +13607,7 @@ })) { yield* Queue.offer(resumedQueue, frame); } - yield* awaitUntil( + yield* harness.until(() => toolUpdates(newBash).length === 1); - () => toolUpdates(newBash).length === 1, - "resumed workflow tool running", - ); yield* Deferred.succeed(releaseCleanup, undefined); yield* Queue.take(harness.terminalReceipts); assert.lengthOf(toolUpdates(oldBash), oldToolCount); @@ -13711,10 +13623,7 @@ toolUseId: newBash, }), ); - yield* awaitUntil( - () => toolUpdates(newBash).at(-1)?.status === "completed", - "resumed workflow tool completed", + yield* harness.until(() => toolUpdates(newBash).at(-1)?.status === "completed"); - ); assert.equal(toolUpdates(newBash).at(-1)?.output, "ok"); }).pipe( Effect.provideService(Clock.Clock, { @@ -14516,25 +14425,20 @@ harness.sdkMessages, workflowSnapshot({ uuid: "00000000-0000-4000-8000-000000001031", state: "done" }), ); - yield* awaitUntil( - () => threadMessages(harness.events, secondThreadId).length === 2, - "workflow member excerpt", + yield* harness.until(() => threadMessages(harness.events, secondThreadId).length === 2); - ); const notification = workflowNotification({ uuid: "00000000-0000-4000-8000-000000001032", status: "completed", }); yield* Queue.offer(harness.sdkMessages, notification); - yield* awaitUntil( - () => - harness.events.some( - (event) => - event.type === "turn_item.updated" && - event.turnItem.type === "subagent" && - event.turnItem.nativeItemRef?.nativeId === WORKFLOW_TASK_ID && + yield* harness.until(() => + harness.events.some( + (event) => + event.type === "turn_item.updated" && + event.turnItem.type === "subagent" && + event.turnItem.nativeItemRef?.nativeId === WORKFLOW_TASK_ID && + event.turnItem.status === "completed", + ), - event.turnItem.status === "completed", - ), - "terminal workflow notification", ); NodeFS.writeFileSync( NodePath.join(harness.transcriptDir, "agent-a2.jsonl"), @@ -14544,11 +14448,10 @@ harness.sdkMessages, claudeSdkFrame({ ...notification, uuid: "00000000-0000-4000-8000-000000001033" }), ); - yield* awaitUntil( + yield* harness.until( () => threadMessages(harness.events, secondThreadId).at(-1)?.message.text === "A2 final answer", - "late terminal transcript answer", ); yield* Queue.offer( harness.sdkMessages, @@ -14625,10 +14528,7 @@ harness.sdkMessages, workflowSnapshot({ uuid: "00000000-0000-4000-8000-000000001009", state: "done" }), ); - yield* awaitUntil( - () => threadMessages(harness.events, secondThreadId).length === 2, - "settled workflow member answers", + yield* harness.until(() => threadMessages(harness.events, secondThreadId).length === 2); - ); const coordinator = workflowCoordinatorEvents(harness.events).at(-1)?.subagent; assert.deepEqual( @@ -14658,14 +14558,12 @@ yield* harness.offerAndWait( workflowSnapshot({ uuid: "00000000-0000-4000-8000-000000001015", state: "done" }), ); - yield* awaitUntil( + yield* harness.until( () => workflowCoordinatorEvents(harness.events).length > coordinatorEventsBeforeRepeat, - "repeated settled workflow snapshot", ); + yield* harness.until( - yield* awaitUntil( () => threadMessages(harness.events, firstThreadId).at(-1)?.message.text === "A1 continued", - "continued workflow transcript answer", ); assert.lengthOf(workflowMemberEvents(harness.events, 2), memberEventsBeforeRepeat); assert.lengthOf(threadMessages(harness.events, secondThreadId), 2); @@ -14677,10 +14575,7 @@ yield* harness.offerAndWait( workflowSnapshot({ uuid: "00000000-0000-4000-8000-000000001016", state: "done" }), ); - yield* awaitUntil( - () => threadMessages(harness.events, secondThreadId).length === 3, + yield* harness.until(() => threadMessages(harness.events, secondThreadId).length === 3); - "late workflow transcript answer", - ); assert.lengthOf(workflowMemberEvents(harness.events, 2), memberEventsBeforeRepeat); assert.equal( threadMessages(harness.events, secondThreadId).at(-1)?.message.text, @@ -14706,9 +14601,8 @@ lastProgressAt: 4432, }), ); - yield* awaitUntil( + yield* harness.until(() => + currentAssistantAnswers().join("\n\n").includes("Second answer"), - () => currentAssistantAnswers().join("\n\n").includes("Second answer"), - "two-turn workflow member answer", ); assert.deepEqual(currentAssistantAnswers().slice(-2), ["First answer", "Second answer"]); const completedMember = workflowMemberEvents(harness.events, 1).at(-1)?.subagent; @@ -14749,11 +14643,10 @@ attempt: 3, }), ); - yield* awaitUntil( + yield* harness.until( () => threadMessages(harness.events, firstThreadId).at(-1)?.message.text === "A1 retry answer", - "retried workflow member answer", ); assert.equal(currentAssistantAnswers().at(-1), "A1 retry answer"); NodeFS.rmSync(NodePath.join(harness.transcriptDir, "agent-a1.jsonl")); @@ -14765,11 +14658,10 @@ attempt: 4, }), ); + yield* harness.until( - yield* awaitUntil( () => workflowCoordinatorEvents(harness.events).at(-1)?.subagent.workflow?.agents[0] ?.attempt === 4, - "fourth workflow member attempt", ); const answersBeforeEmptyAttempt = currentAssistantAnswers(); yield* harness.offerAndWait( ``` ## apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts Status: modified; +130; −155 ```diff @@ -45,6 +45,7 @@ import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as PlatformError from "effect/PlatformError"; +import * as PubSub from "effect/PubSub"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; @@ -1120,16 +1121,20 @@ }); describe("CodexAdapterV2 post-settle continuation", () => { - const awaitUntil = (predicate: () => boolean, label: string): Effect.Effect => - Effect.gen(function* () { - for (let attempt = 0; attempt < 5000; attempt++) { - if (predicate()) { - return; - } - yield* Effect.yieldNow; - } + // The harness notifies after every recorded event or continuation, so a wait + // re-checks exactly when the state it reads changes. Subscribing before the + // first check means a change that lands in between is not lost. + const makeChangeSignal = Effect.gen(function* () { + const changes = yield* PubSub.unbounded(); - return yield* Effect.die(`Timed out waiting for ${label}.`); - }); + const until = (predicate: () => boolean): Effect.Effect => + Effect.scoped( + Effect.gen(function* () { + const subscription = yield* PubSub.subscribe(changes); + while (!predicate()) yield* PubSub.take(subscription); + }), + ); + return { notify: PubSub.publish(changes, undefined), until }; + }); const makeCodexReplayHarness = ( transcript: CodexReplay.CodexAppServerReplayTranscript, @@ -1142,6 +1147,7 @@ const fileSystem = yield* FileSystem.FileSystem; const idAllocator = yield* IdAllocator.IdAllocatorV2; const serverConfig = yield* makeReplayServerConfig(transcript.scenario).pipe(Effect.orDie); + const changes = yield* makeChangeSignal; const continuationRequests: Array = []; const clientFactory: CodexAdapterV2.CodexAppServerClientFactoryShape = { @@ -1188,7 +1194,7 @@ offer: (request) => Effect.sync(() => { continuationRequests.push(request); - }), + }).pipe(Effect.andThen(changes.notify)), }, }); const threadId = ThreadId.make(`thread-${transcript.scenario}`); @@ -1215,6 +1221,7 @@ ? Deferred.succeed(firstTerminal, undefined) : Effect.void, ), + Effect.andThen(changes.notify), Effect.andThen(onEvent(event)), ), ), @@ -1251,6 +1258,8 @@ subagentUpdates, hasPendingBackgroundWork, firstTerminal: Deferred.await(firstTerminal), + until: changes.until, + awaitTerminal: (count: number) => changes.until(() => terminalEvents().length >= count), }; }); @@ -2021,7 +2030,7 @@ decision: ProviderApprovalDecision, ) => Effect.gen(function* () { + yield* harness.until(() => requests().length === count); - yield* awaitUntil(() => requests().length === count, `approval request ${count}`); const request = requests()[count - 1]!; assert.equal(request.kind, kind); yield* harness.runtime.respondToRuntimeRequest({ requestId: request.id, decision }); @@ -2045,15 +2054,15 @@ yield* runTurn(0); yield* answer(1, "command", "acceptForSession"); - yield* awaitUntil(() => requests().length === 2, "file change request"); - yield* awaitUntil(() => harness.terminalEvents().length === 1, "turn 1"); + yield* harness.until(() => requests().length === 2); + yield* harness.awaitTerminal(1); yield* runTurn(1); yield* answer(2, "file-change", "acceptForSession"); + yield* harness.awaitTerminal(2); - yield* awaitUntil(() => harness.terminalEvents().length === 2, "turn 2"); yield* runTurn(2); yield* answer(3, "command", "accept"); yield* answer(4, "file-change", "accept"); - yield* awaitUntil(() => harness.terminalEvents().length === 3, "turn 3"); + yield* harness.awaitTerminal(3); assert.lengthOf(requests(), 4); }).pipe( Effect.provide( @@ -2696,7 +2705,7 @@ text: "Open github.com.", }), ); - yield* awaitUntil(() => harness.terminalEvents().length === 1, "Codex retry recovery"); + yield* harness.awaitTerminal(1); const retryItems = harness.events.flatMap((event) => event.type === "turn_item.updated" && @@ -2817,7 +2826,7 @@ text: "Run two commands.", }), ); - yield* awaitUntil(() => harness.terminalEvents().length === 1, "Codex item start times"); + yield* harness.awaitTerminal(1); const startedAtByItem = harness.events.flatMap((event) => event.type === "turn_item.updated" && @@ -3131,7 +3140,7 @@ text: "Reply with the requested recovery marker.", }), ); - yield* awaitUntil(() => harness.terminalEvents().length === 1, "root turn terminal"); + yield* harness.awaitTerminal(1); assert.deepEqual( assistantMessages(harness.events).map((event) => event.message.text), @@ -3164,7 +3173,7 @@ text: "Reply with the requested recovery marker.", }), ); + yield* harness.awaitTerminal(1); - yield* awaitUntil(() => harness.terminalEvents().length === 1, "root turn terminal"); assert.deepEqual( assistantMessages(harness.events).map((event) => event.message.text), @@ -3969,15 +3978,13 @@ }), ); - yield* awaitUntil( - () => - harness.events.some( - (event) => - event.type === "turn_item.updated" && + yield* harness.until(() => + harness.events.some( + (event) => + event.type === "turn_item.updated" && + event.turnItem.type === "command_execution" && - event.turnItem.type === "command_execution" && - event.turnItem.status === "running", - ), - "running command item", + event.turnItem.status === "running", + ), ); const providerTurnId = harness.events.find( @@ -3995,7 +4002,7 @@ providerTurnId, }); + yield* harness.awaitTerminal(1); - yield* awaitUntil(() => harness.terminalEvents().length === 1, "interrupted terminal"); assert.equal(harness.terminalEvents()[0]?.status, "interrupted"); const terminalIndex = harness.events.findIndex((event) => event.type === "turn.terminal"); @@ -4264,16 +4271,14 @@ text: INTERRUPT_PROMPT, }), ); + yield* harness.until(() => + harness.events.some( + (event) => + event.type === "turn_item.updated" && + event.turnItem.type === "command_execution" && - yield* awaitUntil( - () => - harness.events.some( - (event) => - event.type === "turn_item.updated" && - event.turnItem.type === "command_execution" && - event.turnItem.nativeItemRef?.nativeId === INTERRUPT_CHILD_COMMAND_ITEM && - event.turnItem.status === "running", - ), + event.turnItem.nativeItemRef?.nativeId === INTERRUPT_CHILD_COMMAND_ITEM && + event.turnItem.status === "running", + ), - "running child command item", ); const providerTurnId = harness.events.find( ( @@ -4290,7 +4295,7 @@ providerTurnId, }); + yield* harness.awaitTerminal(1); - yield* awaitUntil(() => harness.terminalEvents().length === 1, "interrupted root terminal"); assert.equal(harness.terminalEvents()[0]?.status, "interrupted"); const childCommandUpdates = harness.events.filter( ( @@ -4367,16 +4372,14 @@ text: INTERRUPT_PROMPT, }), ); - yield* awaitUntil( - () => - harness.events.some( + yield* harness.until(() => + harness.events.some( - (event) => - event.type === "turn_item.updated" && - event.turnItem.type === "command_execution" && - event.turnItem.nativeItemRef?.nativeId === INTERRUPT_CHILD_COMMAND_ITEM && - event.turnItem.status === "running", - ), - "running child command item", + (event) => + event.type === "turn_item.updated" && + event.turnItem.type === "command_execution" && + event.turnItem.nativeItemRef?.nativeId === INTERRUPT_CHILD_COMMAND_ITEM && + event.turnItem.status === "running", + ), ); const providerTurnId = harness.events.find( ( @@ -4393,7 +4396,7 @@ .pipe(Effect.exit); assert.equal(interruptExit._tag, "Failure"); - yield* awaitUntil(() => harness.terminalEvents().length === 1, "interrupted root terminal"); + yield* harness.awaitTerminal(1); assertChildProviderTerminalBeforeRoot(harness.events, harness.threadId); assert.isFalse(yield* harness.hasPendingBackgroundWork); }).pipe( @@ -4462,16 +4465,14 @@ text: INTERRUPT_PROMPT, }), ); + yield* harness.until(() => - yield* awaitUntil( - () => - harness.events.some( - (event) => - event.type === "turn_item.updated" && - event.turnItem.type === "command_execution" && - event.turnItem.nativeItemRef?.nativeId === INTERRUPT_CHILD_COMMAND_ITEM && - event.turnItem.status === "running", - ), + harness.events.some( + (event) => + event.type === "turn_item.updated" && + event.turnItem.type === "command_execution" && + event.turnItem.nativeItemRef?.nativeId === INTERRUPT_CHILD_COMMAND_ITEM && - "running child command item", + event.turnItem.status === "running", + ), ); const providerTurnId = harness.events.find( ( @@ -4486,18 +4487,16 @@ const interruptFiber = yield* harness.runtime .interruptTurn({ providerThread: harness.providerThread, providerTurnId }) .pipe(Effect.forkScoped); + yield* harness.until(() => + harness.events.some( + (event) => + event.type === "provider_turn.updated" && + event.providerTurn.nativeTurnRef?.nativeId === INTERRUPT_LATE_CHILD_NATIVE_TURN, + ), - yield* awaitUntil( - () => - harness.events.some( - (event) => - event.type === "provider_turn.updated" && - event.providerTurn.nativeTurnRef?.nativeId === INTERRUPT_LATE_CHILD_NATIVE_TURN, - ), - "late child provider turn", ); yield* Fiber.join(interruptFiber); - yield* awaitUntil(() => harness.terminalEvents().length === 1, "interrupted root terminal"); + yield* harness.awaitTerminal(1); const lateChildUpdates = harness.events.filter( ( event, @@ -4591,15 +4590,13 @@ text: INTERRUPT_PROMPT, }), ); + yield* harness.until(() => + harness.events.some( - yield* awaitUntil( - () => - harness.events.some( - (event) => - event.type === "turn_item.updated" && - event.turnItem.type === "command_execution" && - event.turnItem.status === "running", - ), - "running command item", + (event) => + event.type === "turn_item.updated" && + event.turnItem.type === "command_execution" && + event.turnItem.status === "running", + ), ); const providerTurnId = harness.events.find( ( @@ -4619,7 +4616,7 @@ .pipe(Effect.exit); assert.equal(interruptExit._tag, "Failure"); - yield* awaitUntil(() => harness.terminalEvents().length === 1, "interrupted terminal"); + yield* harness.awaitTerminal(1); assert.equal(harness.terminalEvents()[0]?.status, "interrupted"); assert.isFalse(yield* harness.hasPendingBackgroundWork); }).pipe( @@ -5070,7 +5067,7 @@ text: FAILED_PROMPT, }), ); - yield* awaitUntil(() => harness.terminalEvents().length === 1, "failed terminal"); + yield* harness.awaitTerminal(1); assert.equal(harness.terminalEvents()[0]?.status, "failed"); const terminalIndex = harness.events.findIndex((event) => event.type === "turn.terminal"); @@ -5785,17 +5782,15 @@ text: RESUME_PROMPT, }), ); + yield* harness.until(() => + harness.subagentUpdates().some((event) => event.subagent.result === "CODEX_FIRST_DONE"), - yield* awaitUntil( - () => - harness.subagentUpdates().some((event) => event.subagent.result === "CODEX_FIRST_DONE"), - "first subagent result", ); assert.lengthOf( harness.subagentUpdates().filter((event) => event.subagent.result === "CODEX_FIRST_DONE"), 1, ); yield* TestClock.adjust("100 millis"); - yield* awaitUntil(() => harness.terminalEvents().length === 1, "root turn terminal"); + yield* harness.awaitTerminal(1); assert.equal(harness.terminalEvents()[0]?.status, "completed"); const settledUpdates = harness.subagentUpdates(); const firstCompletion = settledUpdates[settledUpdates.length - 1]; @@ -5805,10 +5800,7 @@ const settledUpdateCount = settledUpdates.length; yield* TestClock.adjust("30 seconds"); - yield* awaitUntil( - () => harness.subagentUpdates().length > settledUpdateCount, + yield* harness.until(() => harness.subagentUpdates().length > settledUpdateCount); - "subagent re-open", - ); const reopened = harness.subagentUpdates()[settledUpdateCount]; assert.equal(reopened?.subagent.status, "running"); assert.equal(DateTime.toEpochMillis(reopened!.subagent.startedAt!), 1782622470000); @@ -5816,7 +5808,7 @@ assert.isTrue(yield* harness.hasPendingBackgroundWork); yield* TestClock.adjust("30 seconds"); + yield* harness.until(() => { - yield* awaitUntil(() => { const updates = harness.subagentUpdates(); const latest = updates[updates.length - 1]; return ( @@ -5824,7 +5816,7 @@ latest.subagent.status === "completed" && latest.subagent.result === "CODEX_RESUME_DONE" ); - }, "resumed subagent completion"); + }); assert.isFalse(yield* harness.hasPendingBackgroundWork); assert.lengthOf(harness.terminalEvents(), 1); assert.lengthOf(harness.continuationRequests, 0); @@ -6647,14 +6639,12 @@ ? [event.providerTurn.status] : [], ); + yield* harness.until(() => - yield* awaitUntil( + harness.events.some( - () => - harness.events.some( - (event) => - event.type === "provider_thread.updated" && - event.providerThread.goal?.tokensUsed === 11, - ), - "the completed turn to be held", + (event) => + event.type === "provider_thread.updated" && + event.providerThread.goal?.tokensUsed === 11, + ), ); assert.deepEqual(turnStatuses(), ["running"]); assert.lengthOf(harness.terminalEvents(), 0); @@ -6732,14 +6722,12 @@ text: prompt, }), ); + yield* harness.until(() => + harness.events.some( + (event) => + event.type === "provider_thread.updated" && + event.providerThread.goal?.tokensUsed === 11, - yield* awaitUntil( + ), - () => - harness.events.some( - (event) => - event.type === "provider_thread.updated" && - event.providerThread.goal?.tokensUsed === 11, - ), - "the continuation to be adopted", ); yield* harness.runtime.interruptTurn({ providerThread: harness.providerThread, @@ -6827,14 +6815,12 @@ text: prompt, }), ); - yield* awaitUntil( - () => - harness.events.some( - (event) => - event.type === "provider_thread.updated" && - event.providerThread.goal?.tokensUsed === 11, - ), - "the completed turn to be held", + yield* harness.until(() => + harness.events.some( + (event) => + event.type === "provider_thread.updated" && + event.providerThread.goal?.tokensUsed === 11, + ), ); yield* harness.runtime.interruptTurn({ providerThread: harness.providerThread, @@ -6849,15 +6835,13 @@ harness.terminalEvents().map((event) => event.status), ["interrupted"], ); - yield* awaitUntil( - () => - harness.events.some( + yield* harness.until(() => + harness.events.some( + (event) => - (event) => - event.type === "turn_item.updated" && - event.turnItem.type === "command_execution" && - event.turnItem.status === "interrupted", - ), - "the background command to be interrupted", + event.type === "turn_item.updated" && + event.turnItem.type === "command_execution" && + event.turnItem.status === "interrupted", + ), ); }).pipe( Effect.scoped, @@ -6923,14 +6907,12 @@ text: prompt, }), ); - yield* awaitUntil( - () => - harness.events.some( - (event) => - event.type === "provider_thread.updated" && + yield* harness.until(() => + harness.events.some( + (event) => + event.type === "provider_thread.updated" && + event.providerThread.goal?.status === "active", - event.providerThread.goal?.status === "active", - ), - "the goal to be active", + ), ); yield* harness.runtime.interruptTurn({ providerThread: harness.providerThread, @@ -7036,14 +7018,12 @@ text: prompt, }), ); - yield* awaitUntil( + yield* harness.until(() => - () => - harness.events.some( - (event) => - event.type === "provider_thread.updated" && - event.providerThread.goal?.tokensUsed === 11, - ), - "the continuation to be adopted", + harness.events.some( + (event) => + event.type === "provider_thread.updated" && + event.providerThread.goal?.tokensUsed === 11, + ), ); yield* harness.runtime.interruptTurn({ providerThread: harness.providerThread, @@ -7053,15 +7033,13 @@ }), requestRuntimeRestart: true, }); + yield* harness.until(() => + harness.events.some( + (event) => + event.type === "provider_turn.updated" && + event.providerTurn.nativeTurnRef?.nativeId === childTurnId && + event.providerTurn.status === "interrupted", + ), - yield* awaitUntil( - () => - harness.events.some( - (event) => - event.type === "provider_turn.updated" && - event.providerTurn.nativeTurnRef?.nativeId === childTurnId && - event.providerTurn.status === "interrupted", - ), - "the subagent turn to be interrupted", ); }).pipe( Effect.scoped, @@ -7114,10 +7092,7 @@ text: prompt, }), ); - yield* awaitUntil( - () => providerGoals(harness.events).includes("active"), - "the active goal to reach the provider thread", - ); + yield* harness.until(() => providerGoals(harness.events).includes("active")); yield* harness.runtime.interruptTurn({ providerThread: harness.providerThread, providerTurnId: (yield* IdAllocator.IdAllocatorV2).derive.providerTurn({ ``` ## apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts Status: modified; +17; −11 ```diff @@ -4075,12 +4075,14 @@ }), ]); const turns: Array = []; + const firstEnded = yield* Deferred.make(); const bothEnded = yield* Deferred.make(); yield* runtime.events.pipe( Stream.tap((event) => Effect.gen(function* () { if (event.type === "provider_turn.updated" && event.providerTurn.status !== "running") { turns.push(event.providerTurn); + if (turns.length === 1) yield* Deferred.succeed(firstEnded, undefined); if (turns.length === 2) yield* Deferred.succeed(bothEnded, undefined); } }), @@ -4089,9 +4091,7 @@ Effect.forkScoped, ); yield* runtime.startTurn(turnInput(thread)); - yield* Effect.gen(function* () { + yield* Deferred.await(firstEnded); - while (turns.length === 0) yield* Effect.yieldNow; - }); yield* runtime.startTurn(secondTurn(thread)); yield* Deferred.await(bothEnded); const [first, second] = turns; @@ -4319,6 +4319,7 @@ const continued = (after: ReadonlyArray) => Effect.gen(function* () { const offers: Array = []; + const offered = yield* Deferred.make(); const { runtime, thread } = yield* resumed([ ...backgroundLaunch(CHILD), event("session.execution.succeeded", { sessionID: SESSION }), @@ -4353,7 +4354,11 @@ ...after, ]).pipe( Effect.provideService(ProviderContinuationRequests.ProviderContinuationRequests, { + offer: (request) => + Effect.sync(() => void offers.push(request)).pipe( + Effect.andThen(Deferred.succeed(offered, undefined)), + Effect.asVoid, - offer: (request) => Effect.sync(() => void offers.push(request)), + ), take: Effect.never, }), ); @@ -4372,9 +4377,7 @@ Effect.forkScoped, ); yield* runtime.startTurn(withLineage(thread)); - yield* Effect.gen(function* () { - while (offers.length === 0) yield* Effect.yieldNow; - }).pipe(Effect.timeout("2 seconds"), Effect.orDie); + yield* Deferred.await(offered); yield* runtime.startTurn(continuationTurn(thread, offers[0]!)); yield* Deferred.await(bothEnded); const [first, second] = turns; @@ -4821,6 +4824,7 @@ const reportText = `\nCHILD_OK\n`; const steerId = `msg_t3_steer_${SESSION}:message:opencode2-adapter:steer`; const offers: Array = []; + const offered = yield* Deferred.make(); const { runtime, thread } = yield* resumed([ ...backgroundLaunch(CHILD), event("session.execution.succeeded", { sessionID: SESSION }), @@ -4869,7 +4873,11 @@ event("session.execution.succeeded", { sessionID: SESSION }), ]).pipe( Effect.provideService(ProviderContinuationRequests.ProviderContinuationRequests, { + offer: (request) => + Effect.sync(() => void offers.push(request)).pipe( - offer: (request) => Effect.sync(() => void offers.push(request)), + Effect.andThen(Deferred.succeed(offered, undefined)), + Effect.asVoid, + ), take: Effect.never, }), ); @@ -4887,9 +4895,7 @@ Effect.forkScoped, ); yield* runtime.startTurn(withLineage(thread)); + yield* Deferred.await(offered); - yield* Effect.gen(function* () { - while (offers.length === 0) yield* Effect.yieldNow; - }).pipe(Effect.timeout("2 seconds"), Effect.orDie); const followUp = continuationTurn(thread, offers[0]!); yield* runtime.startTurn(followUp); const followUpTurnId = (yield* IdAllocator.IdAllocatorV2).derive.providerTurn({ ``` ## oxlint-plugin-t3code/index.ts Status: modified; +2; −0 ```diff @@ -14,6 +14,7 @@ import noMobilePullRequestGlyphLiterals from "./rules/no-mobile-pull-request-glyph-literals.ts"; import noNativeTitleTooltip from "./rules/no-native-title-tooltip.ts"; import noOutsetStateIndicators from "./rules/no-outset-state-indicators.ts"; +import noPollingLoopInTests from "./rules/no-polling-loop-in-tests.ts"; import noRawMcpRegistration from "./rules/no-raw-mcp-registration.ts"; import noRawResizeObserver from "./rules/no-raw-resize-observer.ts"; import noRealTimeWaitInTests from "./rules/no-real-time-wait-in-tests.ts"; @@ -47,6 +48,7 @@ "no-mobile-pull-request-glyph-literals": noMobilePullRequestGlyphLiterals, "no-native-title-tooltip": noNativeTitleTooltip, "no-outset-state-indicators": noOutsetStateIndicators, + "no-polling-loop-in-tests": noPollingLoopInTests, "no-raw-mcp-registration": noRawMcpRegistration, "no-raw-resize-observer": noRawResizeObserver, "no-real-time-wait-in-tests": noRealTimeWaitInTests, ``` ## oxlint-plugin-t3code/rules/no-polling-loop-in-tests.test.ts Status: added; +231; −0 ```diff @@ -1,0 +1,231 @@ +import { assert, describe } from "@effect/vitest"; + +import { createOxlintRuleHarness } from "../test/utils.ts"; + +const rule = createOxlintRuleHarness("t3code/no-polling-loop-in-tests", { + filename: "fixture.test.ts", +}); + +describe("t3code/no-polling-loop-in-tests", () => { + rule.valid( + "allows a fixed-count flush followed by unrelated error handling", + ` + import { it } from "@effect/vitest"; + import * as Effect from "effect/Effect"; + it.effect("flushes", () => + Effect.gen(function* () { + for (let i = 0; i < 3; i++) yield* Effect.yieldNow; + if (!ready()) return yield* Effect.die("not ready"); + }), + ); + `, + ); + + rule.valid( + "allows a fixed-N flush with a literal bound", + ` + import * as Effect from "effect/Effect"; + + export const flush = Effect.gen(function* () { + for (let i = 0; i < 3; i++) yield* Effect.yieldNow; + }); + `, + ); + + rule.valid( + "allows a fixed-N while flush with a counter bound", + ` + import * as Effect from "effect/Effect"; + + export const flush = Effect.gen(function* () { + let i = 0; + while (i < 3) { + yield* Effect.yieldNow; + i++; + } + }); + `, + ); + + rule.valid( + "allows a fixed-N do-while flush against a named bound", + ` + import * as Effect from "effect/Effect"; + + export const flush = (turns: number) => + Effect.gen(function* () { + let i = 0; + do { + yield* Effect.yieldNow; + i += 1; + } while (i < turns); + }); + `, + ); + + rule.valid( + "allows a loop that takes from a queue", + ` + import * as Effect from "effect/Effect"; + import * as Queue from "effect/Queue"; + + export const until = (events: Queue.Queue) => + Effect.gen(function* () { + while (true) { + const event = yield* Queue.take(events); + if (event > 2) return event; + } + }); + `, + ); + + rule.valid( + "ignores waits inside nested callbacks", + ` + import * as Effect from "effect/Effect"; + + export const run = (items: ReadonlyArray, done: () => boolean) => + Effect.gen(function* () { + for (let i = 0; i < items.length; i++) { + if (done()) return; + yield* Effect.forkChild(Effect.gen(function* () { + yield* Effect.yieldNow; + })); + } + }); + `, + ); + + rule.invalid( + "reports a budgeted yieldNow poll that dies after the loop", + ` + import * as Effect from "effect/Effect"; + + export const awaitUntil = (predicate: () => boolean) => + Effect.gen(function* () { + for (let attempt = 0; attempt < 5000; attempt++) { + if (predicate()) return; + yield* Effect.yieldNow; + } + return yield* Effect.die("Timed out"); + }); + `, + (output) => { + assert.match(output, /Await the `Deferred`/); + }, + ); + + rule.invalid( + "reports a while loop spinning on a predicate", + ` + import * as Effect from "effect/Effect"; + + export const wait = (turns: Array) => + Effect.gen(function* () { + while (turns.length === 0) yield* Effect.yieldNow; + }); + `, + ); + + rule.invalid( + "reports a while loop bounded by a length predicate", + ` + import * as Effect from "effect/Effect"; + + export const wait = (events: Array) => + Effect.gen(function* () { + while (events.length < 1) yield* Effect.yieldNow; + }); + `, + ); + + rule.invalid( + "reports a comparison against an identifier the loop never advances", + ` + import * as Effect from "effect/Effect"; + + export const wait = (state: { count: number }) => + Effect.gen(function* () { + let seen = state.count; + while (seen < 3) { + yield* Effect.yieldNow; + seen = state.count; + } + }); + `, + ); + + rule.invalid( + "reports a for loop bounded by a length predicate", + ` + import * as Effect from "effect/Effect"; + + export const wait = (events: Array) => + Effect.gen(function* () { + for (; events.length < 1; ) yield* Effect.yieldNow; + }); + `, + ); + + rule.invalid( + "reports a sleep poll with a counter bound and an early return", + ` + import * as Effect from "effect/Effect"; + + export const waitFor = (done: () => boolean) => + Effect.gen(function* () { + for (let attempt = 0; attempt < 240; attempt += 1) { + if (done()) return; + yield* Effect.sleep("500 millis"); + } + }); + `, + ); + + rule.invalid( + "reports a microtask poll keyed on the loop condition", + ` + import * as Effect from "effect/Effect"; + + export const settle = function* (until: () => boolean) { + for (let attempt = 0; attempt < 30 && !until(); attempt++) { + yield* Effect.promise(() => Promise.resolve()); + } + }; + `, + ); + + rule.invalid( + "reports an awaited setTimeout poll", + ` + export const waitFor = async (done: () => boolean) => { + do { + await new Promise((resolve) => setTimeout(resolve, 10)); + } while (!done()); + }; + `, + ); + + rule.invalid( + "reports an awaited Promise.resolve poll that breaks", + ` + export const waitFor = async (done: () => boolean) => { + while (true) { + if (done()) break; + await Promise.resolve(); + } + }; + `, + ); +}); + +const productionRule = createOxlintRuleHarness("t3code/no-polling-loop-in-tests"); + +productionRule.valid( + "ignores non-test files", + ` + export const waitFor = async (done: () => boolean) => { + while (!done()) await Promise.resolve(); + }; + `, +); ``` ## oxlint-plugin-t3code/rules/no-polling-loop-in-tests.ts Status: added; +196; −0 ```diff @@ -1,0 +1,196 @@ +import { defineRule, type ESTree } from "@oxlint/plugins"; +import * as Option from "effect/Option"; + +import { getPropertyName, isIdentifier, unwrapExpression } from "../utils.ts"; + +const TEST_FILE_PATTERN = /\.(?:test|spec)\.[cm]?[jt]sx?$/u; +const FUNCTIONS = new Set(["ArrowFunctionExpression", "FunctionExpression", "FunctionDeclaration"]); +const COUNTER_OPERATORS = new Set(["<", "<=", ">", ">="]); +const TIMERS = new Set(["setImmediate", "setTimeout"]); +const MESSAGE = + "Polling loop in a test. Spinning on `yieldNow`/`sleep` is scheduler- and load-dependent. Await the `Deferred`, `Queue.take`, `Stream` element, `drain()`, or `expect.poll` that marks the milestone."; + +type LoopNode = ESTree.ForStatement | ESTree.WhileStatement | ESTree.DoWhileStatement; + +const isNode = (value: unknown): value is ESTree.Node => + typeof value === "object" && value !== null && "type" in value && typeof value.type === "string"; + +// Visits `node` and its descendants, skipping nested functions: a wait or exit +// inside a callback runs on the callback's schedule, not the loop's. +const someDescendant = (node: unknown, match: (node: ESTree.Node) => boolean): boolean => { + if (Array.isArray(node)) return node.some((child) => someDescendant(child, match)); + if (!isNode(node)) return false; + if (match(node)) return true; + if (FUNCTIONS.has(node.type)) return false; + for (const [key, value] of Object.entries(node)) { + if (key === "parent") continue; + if (someDescendant(value, match)) return true; + } + return false; +}; + +const memberName = (node: unknown): Option.Option => { + const expression = unwrapExpression(node); + if (Option.isNone(expression) || expression.value.type !== "MemberExpression") { + return Option.none(); + } + const object = unwrapExpression(expression.value.object); + const property = getPropertyName(expression.value.property); + if (Option.isNone(object) || object.value.type !== "Identifier" || Option.isNone(property)) { + return Option.none(); + } + return Option.some([object.value.name, property.value] as const); +}; + +const isMember = (node: unknown, object: string, property: string): boolean => + Option.match(memberName(node), { + onNone: () => false, + onSome: ([name, key]) => name === object && key === property, + }); + +const isTimerReference = (node: unknown): boolean => { + const expression = unwrapExpression(node); + return ( + Option.isSome(expression) && + expression.value.type === "Identifier" && + TIMERS.has(expression.value.name) + ); +}; + +// `Promise.resolve()`, or a `new Promise` whose executor schedules a timer. +const isTickPromise = (node: unknown): boolean => { + const expression = unwrapExpression(node); + if (Option.isNone(expression)) return false; + const value = expression.value; + if (value.type === "CallExpression") return isMember(value.callee, "Promise", "resolve"); + if (value.type !== "NewExpression" || !isIdentifier(Option.some(value.callee), "Promise")) { + return false; + } + const [executor] = value.arguments; + if (isTimerReference(executor)) return true; + if (!isNode(executor) || !FUNCTIONS.has(executor.type)) return false; + return someDescendant( + (executor as ESTree.ArrowFunctionExpression).body, + (child) => child.type === "CallExpression" && isTimerReference(child.callee), + ); +}; + +const thunkReturnsTick = (node: unknown): boolean => { + if (!isNode(node) || node.type !== "ArrowFunctionExpression") return false; + const body = node.body; + if (body.type !== "BlockStatement") return isTickPromise(body); + const last = body.body.at(-1); + return last?.type === "ReturnStatement" && isTickPromise(last.argument); +}; + +const isWait = (node: ESTree.Node): boolean => { + if (node.type === "MemberExpression") return isMember(node, "Effect", "yieldNow"); + if (node.type === "AwaitExpression") return isTickPromise(node.argument); + if (node.type === "NewExpression") return isTickPromise(node); + if (node.type !== "CallExpression") return false; + if (isMember(node.callee, "Effect", "sleep")) return true; + if (isIdentifier(unwrapExpression(node.callee), "setImmediate")) return true; + if (isMember(node.callee, "Effect", "promise")) return thunkReturnsTick(node.arguments[0]); + return false; +}; + +const hasConditionalExit = (body: ESTree.Node): boolean => + someDescendant( + body, + (node) => + node.type === "IfStatement" && + someDescendant( + [node.consequent, node.alternate], + (child) => + child.type === "ReturnStatement" || + child.type === "BreakStatement" || + child.type === "ThrowStatement", + ), + ); + +const isNumericBound = (node: unknown): boolean => { + const expression = unwrapExpression(node); + if (Option.isNone(expression)) return false; + const value = expression.value; + return ( + value.type === "Identifier" || (value.type === "Literal" && typeof value.value === "number") + ); +}; + +const updatesCounter = (node: unknown, name: string): boolean => + someDescendant( + node, + (child) => + (child.type === "UpdateExpression" && isIdentifier(unwrapExpression(child.argument), name)) || + (child.type === "AssignmentExpression" && + (child.operator === "+=" || child.operator === "-=") && + isIdentifier(unwrapExpression(child.left), name)), + ); + +// A bare counter (`i < 3`, `attempt < max`) that the loop itself advances +// bounds the loop without asking anything about the system under test. +// `events.length < 1` or a counter nobody increments is a predicate. +const isCounterTest = (loop: LoopNode): boolean => { + if (loop.test == null) return false; + const expression = unwrapExpression(loop.test); + if (Option.isNone(expression) || expression.value.type !== "BinaryExpression") return false; + if (!COUNTER_OPERATORS.has(expression.value.operator)) return false; + const { left, right } = expression.value; + const advances = [loop.body, loop.type === "ForStatement" ? loop.update : null]; + return [ + [left, right], + [right, left], + ].some(([counter, bound]) => { + const name = unwrapExpression(counter); + return ( + Option.isSome(name) && + name.value.type === "Identifier" && + isNumericBound(bound) && + updatesCounter(advances, name.value.name) + ); + }); +}; + +const isTrueLiteral = (test: ESTree.Expression): boolean => { + const expression = unwrapExpression(test); + return ( + Option.isSome(expression) && + expression.value.type === "Literal" && + expression.value.value === true + ); +}; + +// `for (...) { yield* Effect.yieldNow } return yield* Effect.die("timed out")` +// keys a failure on the loop running out, which makes the bound a budget. +// A loop is polling when a predicate decides whether it keeps going: a conditional exit in the +// body, or a test that is neither `true` nor a counter the loop advances. A fixed-count flush +// followed by unrelated error handling is not polling, so what comes after the loop is ignored. +const hasPredicateExit = (loop: LoopNode): boolean => + hasConditionalExit(loop.body) || + (loop.test != null && !isTrueLiteral(loop.test) && !isCounterTest(loop)); + +export default defineRule({ + meta: { + type: "problem", + docs: { + description: + "Disallow loops in tests that spin on yieldNow/sleep/timer ticks until a predicate holds; await the milestone instead.", + }, + schema: [], + }, + create(context) { + if (!TEST_FILE_PATTERN.test(context.filename)) return {}; + + const check = (loop: LoopNode) => { + if (!someDescendant(loop.body, isWait)) return; + if (!hasPredicateExit(loop)) return; + context.report({ node: loop, message: MESSAGE }); + }; + + return { + ForStatement: check, + WhileStatement: check, + DoWhileStatement: check, + }; + }, +}); ``` ## vite.config.ts Status: modified; +2; −0 ```diff @@ -187,6 +187,8 @@ "t3code/no-low-value-comments": "error", "t3code/no-manual-effect-runtime-in-tests": "error", "t3code/no-native-title-tooltip": "error", + // Test files only (the rule checks the filename). Becomes an error once existing polls are converted. + "t3code/no-polling-loop-in-tests": "warn", "t3code/no-raw-mcp-registration": "error", "t3code/no-real-time-wait-in-tests": "error", "t3code/no-test-in-loop": "error", ```