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';
@@ -2267,24 +2267,47 @@ 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.runtimeEventStore.listSessionInvocations(session.id);
-    for (let attempt = 0; attempt < 400; attempt += 1) {
-      const wakeRuns = runs.filter(
-        (run) => run.opening.root.kind === 'agent_graph_supervisor_wake',
+    try {
+      await waitFor(
+        async () => {
+          const wakeRuns = runs.filter(
+            (run) => run.opening.root.kind === 'agent_graph_supervisor_wake',
+          );
+          if (
+            updates.at(-1)?.finish &&
+            wakeRuns.length > 0 &&
+            wakeRuns.every((run) => runtimeInvocationOutcome(run) !== undefined) &&
+            liveResidencies === 0
+          ) {
+            return true;
+          }
+          [updates, runs] = await Promise.all([
+            graph.listAgentGraphScheduleUpdates(graphId),
+            execution.runtimeEventStore.listSessionInvocations(session.id),
+          ]);
+          return false;
+        },
+        { timeoutMs: 30_000, pollMs: 10, message: 'graph wake runs did not settle' },
       );
-      if (
-        updates.at(-1)?.finish &&
-        wakeRuns.length > 0 &&
-        wakeRuns.every((run) => runtimeInvocationOutcome(run) !== undefined) &&
-        liveResidencies === 0
-      ) {
-        break;
-      }
-      await new Promise<void>((resolve) => setTimeout(resolve, 10));
-      updates = await graphStore.listAgentGraphScheduleUpdates(graphId);
-      runs = await execution.runtimeEventStore.listSessionInvocations(session.id);
+    } catch (error) {
+      assert.ok(
+        updates.at(-1)?.finish,
+        JSON.stringify({
+          updateCount: updates.length,
+          lastUpdate: updates.at(-1),
+          runs: runs.map((run) => ({
+            runId: run.runId,
+            status: runtimeInvocationOutcome(run) ?? 'running',
+            root: run.opening.root,
+          })),
+          requests: providerRequestTrace(provider.requests),
+        }),
+      );
+      throw error;
     }
 
     const finish = updates.at(-1)?.finish;
@@ -3859,19 +3882,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(
@@ -3882,14 +3914,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(
@@ -3898,23 +3933,33 @@ 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 waitForCanonicalRequests(
@@ -3923,47 +3968,66 @@ async function waitForCanonicalRequests(
   expectedRequests: number,
 ): Promise<number> {
   const ask = () => usage.modelCalls.modelCallSummary({ range: 'all', sessionId }, Date.now());
-  for (let attempt = 0; attempt < 100; attempt += 1) {
-    const { projection } = await ask();
-    if (projection.totalRequests >= expectedRequests) return projection.totalRequests;
-    await new Promise<void>((resolve) => setTimeout(resolve, 10));
+  let totalRequests = -1;
+  try {
+    await waitFor(
+      async () => {
+        const { projection } = await ask();
+        totalRequests = projection.totalRequests;
+        return totalRequests >= expectedRequests;
+      },
+      {
+        timeoutMs: 5_000,
+        pollMs: 10,
+        message: `Hosted model call attempts did not reach ${expectedRequests}`,
+      },
+    );
+  } catch {
+    const { projection, unreadableRecords } = await ask();
+    throw new Error(
+      `Hosted canonical model-call attempts were not persisted: ${JSON.stringify({
+        expectedRequests,
+        totalRequests: projection.totalRequests,
+        unreadableRecords,
+      })}`,
+    );
   }
-  const { projection, unreadableRecords } = await ask();
-  throw new Error(
-    `Hosted canonical model-call attempts were not persisted: ${JSON.stringify({
-      expectedRequests,
-      totalRequests: projection.totalRequests,
-      unreadableRecords,
-    })}`,
-  );
+  return totalRequests;
 }
 
 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, message: 'memory extraction requests did not settle' },
+    );
+  } 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, readFile, rm, writeFile } from 'node:fs/promises';
@@ -280,10 +281,11 @@ 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,
+      message: 'peer mesh member did not reach Online',
+    });
     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';
@@ -515,9 +516,11 @@ 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,
+    message: `peer-native request count did not reach ${expected}`,
+  });
   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']>(