test(runtime-host): bound test wait helpers by wall-clock deadlines

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<void>((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<TurnSnapshot> {
-  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<void>((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<TurnSnapshot> {
   let snapshot = initial;
-  for (let attempt = 0; attempt < 200; attempt += 1) {
-    if (isTerminal(snapshot)) return snapshot;
-    await new Promise<void>((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,93 +3829,123 @@ async function waitForUsage(
   connectionSlug: string,
   callKind: ModelCallKind,
 ): Promise<Extract<UsageQueryResult, { kind: 'logs'; source: 'llm' }>['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<UsageQueryResult, { kind: 'logs'; source: 'llm' }>['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<void>((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(
   usage: InteractiveUsageStoresWriter,
   sessionId: string,
   expectedRequests: number,
 ): Promise<readonly ModelCallAttempt[]> {
-  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<void>((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(
   artifacts: Awaited<ReturnType<typeof openInteractiveArtifactStoreForWrite>>,
   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<void>((resolve) => setTimeout(resolve, 10));
-  }
-  throw new Error(`Hosted request artifacts did not reach ${expectedRequests}`);
+  let captures: Awaited<ReturnType<typeof artifacts.listPage>>['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(
   requests: readonly ProviderRequest[],
 ): Promise<void> {
   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<void>((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 {

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();

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<void> {
-  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);
 }
 

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<void> {
   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<K extends SubscriptionFrame['kind']>(