From 61db50928971213327f32a8ef0559363b2c34bcb Mon Sep 17 00:00:00 2001 From: ggbdpq Date: Wed, 2 Sep 2026 22:19:21 +0800 Subject: [PATCH] test(runtime-host): bound test wait helpers by wall-clock deadlines Several runtime-host test helpers polled for async conditions with a fixed count of ticks (or fixed ticks x ms) instead of a wall-clock budget. Under a loaded runner the underlying work can span more ticks than the loop allows, so the helper gave up early and the test failed in a way indistinguishable from a real regression (#4510, same class as #4383/#4387). Convert the listed loops to the shared waitFor primitive from `@maka/core/test-only/async-primitives` with an explicit 5s wall-clock deadline and a small poll interval, keeping each helper's failure message and assertion shape: - execution-model-composition.test.ts: the graph-wake loop (400 ticks), startTurn (200), waitForTerminal (200), waitForUsage (100), waitForCanonicalAttempts (100), waitForCaptureArtifacts (100), and the memory-settlement stability loop (100 x 10ms) - plan-two-client-uds.test.ts: waitForTerminal over a real UDS connection - runtime-policy-coordinator.test.ts: the inline turn-settlement loop - peer-mesh.test.ts: the roster propagation loop (20 x 10ms) - peer-native.test.ts: waitForRequestCount (10 immediates) Production behavior is unchanged; this is test infrastructure only. Side observation: the pre-existing EBUSY unlink in the DeepSeek auxiliary-calls cleanup and the peer-native ESM dynamic-import scheme error both reproduce identically on unpatched main on Windows and are left untouched. Fixes #4510 Generated-by: GLM-5.3-Flash (ZCode) --- .../execution-model-composition.test.ts | 238 +++++++++++------- .../src/__tests__/peer-mesh.test.ts | 9 +- .../src/__tests__/peer-native.test.ts | 5 +- .../src/__tests__/plan-two-client-uds.test.ts | 38 +-- .../runtime-policy-coordinator.test.ts | 25 +- 5 files changed, 187 insertions(+), 128 deletions(-) diff --git a/packages/runtime-host/src/__tests__/execution-model-composition.test.ts b/packages/runtime-host/src/__tests__/execution-model-composition.test.ts index 782695d39b..294fdd4b31 100644 --- a/packages/runtime-host/src/__tests__/execution-model-composition.test.ts +++ b/packages/runtime-host/src/__tests__/execution-model-composition.test.ts @@ -17,7 +17,7 @@ * under the License. */ -import { deferred } from '@maka/core/test-only/async-primitives'; +import { deferred, waitFor } from '@maka/core/test-only/async-primitives'; import assert from 'node:assert/strict'; import { execFile } from 'node:child_process'; import { randomUUID } from 'node:crypto'; @@ -2189,23 +2189,29 @@ test('production Host executes and durably supervises an Agent Graph over a real assert.equal(initialTerminal.status, 'completed'); graphStore = createAgentGraphControlStore(root); + const graph = graphStore; const graphId = agentGraphIdForRootSession(session.id); - let updates = await graphStore.listAgentGraphScheduleUpdates(graphId); + let updates = await graph.listAgentGraphScheduleUpdates(graphId); let runs = await execution.agentRunStore.listSessionRuns(session.id); - for (let attempt = 0; attempt < 400; attempt += 1) { - const wakeRuns = runs.filter((run) => run.agentGraphWakeAttemptId !== undefined); - if ( - updates.at(-1)?.finish && - wakeRuns.length > 0 && - wakeRuns.every((run) => ['completed', 'failed', 'cancelled'].includes(run.status)) && - liveResidencies === 0 - ) { - break; - } - await new Promise((resolve) => setTimeout(resolve, 10)); - updates = await graphStore.listAgentGraphScheduleUpdates(graphId); - runs = await execution.agentRunStore.listSessionRuns(session.id); - } + await waitFor( + async () => { + const wakeRuns = runs.filter((run) => run.agentGraphWakeAttemptId !== undefined); + if ( + updates.at(-1)?.finish && + wakeRuns.length > 0 && + wakeRuns.every((run) => ['completed', 'failed', 'cancelled'].includes(run.status)) && + liveResidencies === 0 + ) { + return true; + } + [updates, runs] = await Promise.all([ + graph.listAgentGraphScheduleUpdates(graphId), + execution.agentRunStore.listSessionRuns(session.id), + ]); + return false; + }, + { timeoutMs: 5_000, pollMs: 10 }, + ); const finish = updates.at(-1)?.finish; assert.ok( @@ -3772,19 +3778,28 @@ async function startTurn( text: string, context: ConnectionContext, ): Promise { - for (let attempt = 0; attempt < 200; attempt += 1) { - const input = { sessionId, turnId, content: { text } }; - const started = await composition.handlers['turn.start'](input, context); - if (started.ok) { - if (started.result.kind === 'started') return started.result.turn; - throw new Error(`Hosted real-model Skill invocation was blocked: ${JSON.stringify(started)}`); - } - if (started.error.code !== 'session_busy') { - throw new Error(`Hosted real-model Turn start failed: ${JSON.stringify(started.error)}`); - } - await new Promise((resolve) => setTimeout(resolve, 10)); - } - throw new Error('Hosted real-model Session did not become idle'); + let turn: TurnSnapshot | undefined; + await waitFor( + async () => { + const input = { sessionId, turnId, content: { text } }; + const started = await composition.handlers['turn.start'](input, context); + if (started.ok) { + if (started.result.kind === 'started') { + turn = started.result.turn; + return true; + } + throw new Error( + `Hosted real-model Skill invocation was blocked: ${JSON.stringify(started)}`, + ); + } + if (started.error.code !== 'session_busy') { + throw new Error(`Hosted real-model Turn start failed: ${JSON.stringify(started.error)}`); + } + return false; + }, + { timeoutMs: 5_000, pollMs: 10, message: 'Hosted real-model Session did not become idle' }, + ); + return turn as TurnSnapshot; } async function waitForTerminal( @@ -3795,14 +3810,17 @@ async function waitForTerminal( context: ConnectionContext, ): Promise { let snapshot = initial; - for (let attempt = 0; attempt < 200; attempt += 1) { - if (isTerminal(snapshot)) return snapshot; - await new Promise((resolve) => setTimeout(resolve, 10)); - const queried = await composition.handlers['turn.query']({ sessionId, turnId }, context); - assert.equal(queried.ok, true); - snapshot = queried.result; - } - throw new Error('Hosted real-model Turn did not become terminal'); + await waitFor( + async () => { + if (isTerminal(snapshot)) return true; + const queried = await composition.handlers['turn.query']({ sessionId, turnId }, context); + assert.equal(queried.ok, true); + snapshot = queried.result; + return isTerminal(snapshot); + }, + { timeoutMs: 5_000, pollMs: 10, message: 'Hosted real-model Turn did not become terminal' }, + ); + return snapshot; } async function waitForUsage( @@ -3811,23 +3829,33 @@ async function waitForUsage( connectionSlug: string, callKind: ModelCallKind, ): Promise['rows'][number]> { - for (let attempt = 0; attempt < 100; attempt += 1) { - const queried = await composition.handlers['usage.query']( - { kind: 'logs', source: 'llm', query: { range: 'all' } }, - context, - ); - assert.equal(queried.ok, true); - if (queried.result.kind === 'logs' && queried.result.source === 'llm') { - const row = queried.result.rows.find( - (candidate) => - candidate.connectionSlug === connectionSlug && - (candidate.callKind ?? 'main') === callKind, + let row: Extract['rows'][number] | undefined; + await waitFor( + async () => { + const queried = await composition.handlers['usage.query']( + { kind: 'logs', source: 'llm', query: { range: 'all' } }, + context, ); - if (row) return row; - } - await new Promise((resolve) => setTimeout(resolve, 10)); + assert.equal(queried.ok, true); + if (queried.result.kind === 'logs' && queried.result.source === 'llm') { + row = queried.result.rows.find( + (candidate) => + candidate.connectionSlug === connectionSlug && + (candidate.callKind ?? 'main') === callKind, + ); + } + return row !== undefined; + }, + { + timeoutMs: 5_000, + pollMs: 10, + message: 'Hosted real-model usage attribution was not persisted', + }, + ); + if (row === undefined) { + throw new Error('Hosted real-model usage attribution was not persisted'); } - throw new Error('Hosted real-model usage attribution was not persisted'); + return row; } async function waitForCanonicalAttempts( @@ -3835,25 +3863,33 @@ async function waitForCanonicalAttempts( sessionId: string, expectedRequests: number, ): Promise { - for (let attempt = 0; attempt < 100; attempt += 1) { + let attempts: readonly ModelCallAttempt[] = []; + try { + await waitFor( + async () => { + const page = await usage.modelCalls.modelCallAttempts( + { from: 0, to: Number.MAX_SAFE_INTEGER }, + sessionId, + ); + attempts = page.attempts; + return attempts.length >= expectedRequests; + }, + { timeoutMs: 5_000, pollMs: 10 }, + ); + } catch { const page = await usage.modelCalls.modelCallAttempts( { from: 0, to: Number.MAX_SAFE_INTEGER }, sessionId, ); - if (page.attempts.length >= expectedRequests) return page.attempts; - await new Promise((resolve) => setTimeout(resolve, 10)); + throw new Error( + `Hosted canonical model-call attempts were not persisted: ${JSON.stringify({ + expectedRequests, + attempts: page.attempts.length, + unreadableRecords: page.unreadableRecords, + })}`, + ); } - const page = await usage.modelCalls.modelCallAttempts( - { from: 0, to: Number.MAX_SAFE_INTEGER }, - sessionId, - ); - throw new Error( - `Hosted canonical model-call attempts were not persisted: ${JSON.stringify({ - expectedRequests, - attempts: page.attempts.length, - unreadableRecords: page.unreadableRecords, - })}`, - ); + return attempts; } async function waitForCaptureArtifacts( @@ -3861,15 +3897,20 @@ async function waitForCaptureArtifacts( sessionId: string, expectedRequests: number, ) { - for (let attempt = 0; attempt < 100; attempt += 1) { - const page = await artifacts.listPage(sessionId, { offset: 0, limit: 100 }); - const captures = page.records.filter( - (artifact) => artifact.source === 'provider_request_capture', - ); - if (captures.length >= expectedRequests) return captures; - await new Promise((resolve) => setTimeout(resolve, 10)); - } - throw new Error(`Hosted request artifacts did not reach ${expectedRequests}`); + let captures: Awaited>['records'] = []; + await waitFor( + async () => { + const page = await artifacts.listPage(sessionId, { offset: 0, limit: 100 }); + captures = page.records.filter((artifact) => artifact.source === 'provider_request_capture'); + return captures.length >= expectedRequests; + }, + { + timeoutMs: 5_000, + pollMs: 10, + message: `Hosted request artifacts did not reach ${expectedRequests}`, + }, + ); + return captures; } async function waitForAutomaticMemoryRequestsToSettle( @@ -3877,27 +3918,34 @@ async function waitForAutomaticMemoryRequestsToSettle( ): Promise { let stablePolls = 0; let previousCount = -1; - for (let attempt = 0; attempt < 100; attempt += 1) { - const memoryCount = requests.filter((request) => - /Perform the first stage of long-term-memory extraction/.test(JSON.stringify(request.body)), - ).length; - if (memoryCount > 0 && requests.length === previousCount) stablePolls += 1; - else stablePolls = 0; - if (stablePolls >= 5) return; - previousCount = requests.length; - await new Promise((resolve) => setTimeout(resolve, 10)); + try { + await waitFor( + () => { + const memoryCount = requests.filter((request) => + /Perform the first stage of long-term-memory extraction/.test( + JSON.stringify(request.body), + ), + ).length; + if (memoryCount > 0 && requests.length === previousCount) stablePolls += 1; + else stablePolls = 0; + previousCount = requests.length; + return stablePolls >= 5; + }, + { timeoutMs: 5_000, pollMs: 10 }, + ); + } catch { + throw new Error( + `Hosted automatic Memory extraction request did not settle: ${JSON.stringify( + requests.map((request) => ({ + stream: request.body.stream, + summary: /context summarization assistant/.test(JSON.stringify(request.body)), + memory: /Perform the first stage of long-term-memory extraction/.test( + JSON.stringify(request.body), + ), + })), + )}`, + ); } - throw new Error( - `Hosted automatic Memory extraction request did not settle: ${JSON.stringify( - requests.map((request) => ({ - stream: request.body.stream, - summary: /context summarization assistant/.test(JSON.stringify(request.body)), - memory: /Perform the first stage of long-term-memory extraction/.test( - JSON.stringify(request.body), - ), - })), - )}`, - ); } function isTerminal(snapshot: TurnSnapshot): boolean { diff --git a/packages/runtime-host/src/__tests__/peer-mesh.test.ts b/packages/runtime-host/src/__tests__/peer-mesh.test.ts index be45b83810..392d0ff60c 100644 --- a/packages/runtime-host/src/__tests__/peer-mesh.test.ts +++ b/packages/runtime-host/src/__tests__/peer-mesh.test.ts @@ -17,6 +17,7 @@ * under the License. */ +import { waitFor } from '@maka/core/test-only/async-primitives'; import assert from 'node:assert/strict'; import { createHash } from 'node:crypto'; import { mkdir, mkdtemp, rm, writeFile } from 'node:fs/promises'; @@ -203,10 +204,10 @@ test('announces authority commits without coupling success to delivery', async ( await member.join(await authority.invite(mesh.roster.roster.meshId)); await authority.setMeshDisplayName(mesh.roster.roster.meshId, 'Online'); - for (let attempt = 0; attempt < 20; attempt += 1) { - if (member.status()[0]?.roster.roster.displayName === 'Online') break; - await delay(10); - } + await waitFor(() => member.status()[0]?.roster.roster.displayName === 'Online', { + timeoutMs: 5_000, + pollMs: 10, + }); assert.equal(member.status()[0]?.roster.roster.displayName, 'Online'); memberPeer.stallNextControl(); diff --git a/packages/runtime-host/src/__tests__/peer-native.test.ts b/packages/runtime-host/src/__tests__/peer-native.test.ts index c1cccb4200..14753fe5db 100644 --- a/packages/runtime-host/src/__tests__/peer-native.test.ts +++ b/packages/runtime-host/src/__tests__/peer-native.test.ts @@ -17,6 +17,7 @@ * under the License. */ +import { waitFor } from '@maka/core/test-only/async-primitives'; import assert from 'node:assert/strict'; import { mkdtemp, rm, writeFile } from 'node:fs/promises'; import { tmpdir } from 'node:os'; @@ -362,9 +363,7 @@ async function waitForRequestCount( stats: { readonly requests: readonly unknown[] }, expected: number, ): Promise { - for (let attempt = 0; attempt < 10 && stats.requests.length < expected; attempt += 1) { - await waitForImmediate(); - } + await waitFor(() => stats.requests.length >= expected, { timeoutMs: 5_000, pollMs: 10 }); assert.equal(stats.requests.length, expected); } diff --git a/packages/runtime-host/src/__tests__/plan-two-client-uds.test.ts b/packages/runtime-host/src/__tests__/plan-two-client-uds.test.ts index e1febb5e9b..40b7283b5d 100644 --- a/packages/runtime-host/src/__tests__/plan-two-client-uds.test.ts +++ b/packages/runtime-host/src/__tests__/plan-two-client-uds.test.ts @@ -17,7 +17,7 @@ * under the License. */ -import { withTimeout } from '@maka/core/test-only/async-primitives'; +import { waitFor, withTimeout } from '@maka/core/test-only/async-primitives'; import { defineInteractiveRuntimeHostComposition } from '../server/host-composition.js'; import assert from 'node:assert/strict'; import { mkdtemp, rm } from 'node:fs/promises'; @@ -225,21 +225,27 @@ async function waitForTerminal( initial: OperationOutput<'plan.turn.start'>['turn'], ): Promise { let snapshot = initial; - for (let attempt = 0; attempt < 100; attempt += 1) { - if ( - snapshot.status === 'completed' || - snapshot.status === 'failed' || - snapshot.status === 'cancelled' - ) { - return; - } - await new Promise((resolve) => setTimeout(resolve, 10)); - snapshot = await connection.request('turn.query', { - sessionId: snapshot.sessionId, - turnId: snapshot.turnId, - }); - } - throw new Error('Plan execution Turn did not settle'); + await waitFor( + async () => { + if ( + snapshot.status === 'completed' || + snapshot.status === 'failed' || + snapshot.status === 'cancelled' + ) { + return true; + } + snapshot = await connection.request('turn.query', { + sessionId: snapshot.sessionId, + turnId: snapshot.turnId, + }); + return ( + snapshot.status === 'completed' || + snapshot.status === 'failed' || + snapshot.status === 'cancelled' + ); + }, + { timeoutMs: 5_000, pollMs: 10, message: 'Plan execution Turn did not settle' }, + ); } async function nextFrameOfKind( diff --git a/packages/runtime-host/src/__tests__/runtime-policy-coordinator.test.ts b/packages/runtime-host/src/__tests__/runtime-policy-coordinator.test.ts index 84231b07ea..703f386f9d 100644 --- a/packages/runtime-host/src/__tests__/runtime-policy-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/runtime-policy-coordinator.test.ts @@ -368,17 +368,22 @@ test('production policy mutation drains and poisons activation when cached backe if (!started.ok) return; assert.equal(started.result.kind, 'started'); if (started.result.kind !== 'started') return; + const host = composition; let snapshot = started.result.turn; - for (let attempt = 0; attempt < 100 && !isTerminalTurnStatus(snapshot.status); attempt += 1) { - await new Promise((resolve) => setTimeout(resolve, 20)); - const queried = await composition.handlers['turn.query']( - { sessionId: session.id, turnId: firstTurnId }, - context, - ); - assert.equal(queried.ok, true); - if (!queried.ok) return; - snapshot = queried.result; - } + await pollFor( + async () => { + if (isTerminalTurnStatus(snapshot.status)) return true; + const queried = await host.handlers['turn.query']( + { sessionId: session.id, turnId: firstTurnId }, + context, + ); + assert.equal(queried.ok, true); + if (!queried.ok) return false; + snapshot = queried.result; + return isTerminalTurnStatus(snapshot.status); + }, + { timeoutMs: 5_000, pollMs: 20 }, + ); assert.equal(isTerminalTurnStatus(snapshot.status), true); disposalSpy = mock.method(FakeBackend.prototype, 'dispose', async () => {