test(runtime-host): assert liveness deadlines on a virtual clock

.github/workflows/ci.yml

@@ -445,7 +445,9 @@ jobs:
       - name: Report CLI Rust build cache
         if: steps.plan.outputs.cli_package == 'true'
         shell: bash
-        run: kache report --format github >> "$GITHUB_STEP_SUMMARY"
+        run: |
+          set -o pipefail
+          kache report --format github | tee -a "$GITHUB_STEP_SUMMARY"
 
       - name: Save CLI Rust build cache
         if: steps.plan.outputs.cli_package == 'true' && github.ref_name == github.event.repository.default_branch

.github/workflows/cli-package-validation.yml

@@ -164,7 +164,9 @@ jobs:
           node native/runtime-host-windows-task-launcher/build.mjs
       - name: Report Rust build cache
         shell: bash
-        run: kache report --format github >> "$GITHUB_STEP_SUMMARY"
+        run: |
+          set -o pipefail
+          kache report --format github | tee -a "$GITHUB_STEP_SUMMARY"
       - name: Save Rust build cache
         if: github.ref_name == github.event.repository.default_branch
         uses: actions/cache/save@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6.1.0

.github/workflows/gitoxide-helper-admission.yml

@@ -97,7 +97,9 @@ jobs:
         run: cargo test --locked
       - name: Report Rust build cache
         shell: bash
-        run: kache report --format github >> "$GITHUB_STEP_SUMMARY"
+        run: |
+          set -o pipefail
+          kache report --format github | tee -a "$GITHUB_STEP_SUMMARY"
       - name: Save Rust build cache
         if: github.ref_name == github.event.repository.default_branch
         uses: actions/cache/save@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6.1.0

.github/workflows/release-windows-check.yml

@@ -172,7 +172,9 @@ jobs:
         run: npm run test:windows-archives
 
       - name: Report Rust build cache
-        run: kache report --format github >> "$GITHUB_STEP_SUMMARY"
+        run: |
+          set -o pipefail
+          kache report --format github | tee -a "$GITHUB_STEP_SUMMARY"
 
       - name: Save Rust build cache
         if: github.ref_name == github.event.repository.default_branch

.github/workflows/runtime-host-peer-admission.yml

@@ -108,7 +108,9 @@ jobs:
         run: cargo test --locked
       - name: Report Rust build cache
         shell: bash
-        run: kache report --format github >> "$GITHUB_STEP_SUMMARY"
+        run: |
+          set -o pipefail
+          kache report --format github | tee -a "$GITHUB_STEP_SUMMARY"
       - name: Save Rust build cache
         if: github.ref_name == github.event.repository.default_branch
         uses: actions/cache/save@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6.1.0

.github/workflows/windows-sandbox-w0.yml

@@ -94,7 +94,9 @@ jobs:
         run: cargo test --locked
       - name: Report Rust build cache
         shell: bash
-        run: kache report --format github >> "$GITHUB_STEP_SUMMARY"
+        run: |
+          set -o pipefail
+          kache report --format github | tee -a "$GITHUB_STEP_SUMMARY"
       - name: Save Rust build cache
         if: github.ref_name == github.event.repository.default_branch
         uses: actions/cache/save@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6.1.0

packages/runtime-host/src/__tests__/resumable-peer-stream.test.ts

@@ -23,6 +23,7 @@ import { setImmediate as tick, setTimeout as delay } from 'node:timers/promises'
 import { test } from 'node:test';
 import { connect, createServer, type Socket } from 'node:net';
 import { once } from 'node:events';
+import { performance } from 'node:perf_hooks';
 import { createRuntimeHostPeerListener } from '../server/peer-listener.js';
 import { RuntimeHostConnectionSession } from '../server/connection-session.js';
 import { LOCAL_OWNER_CONNECTION_AUTHORITY } from '../server/connection-authority.js';
@@ -468,6 +469,19 @@ test('close has a hard deadline even when a healthy peer never drains its receiv
 test('failed proactive upgrade preserves transit; a later direct attachment keeps the logical stream', {
   timeout: 13_000,
 }, async (t) => {
+  let now = 0;
+  t.mock.method(performance, 'now', () => now);
+  t.mock.timers.enable({ apis: ['setInterval'] });
+  const advance = async (milliseconds: number): Promise<void> => {
+    const target = now + milliseconds;
+    while (now < target) {
+      const step = Math.min(250, target - now);
+      now += step;
+      t.mock.timers.tick(step);
+      // Flush real duplex I/O between heartbeats instead of simulating a blackhole.
+      await tick();
+    }
+  };
   let right!: ResumablePeerStream;
   let upgrades = 0;
   const left = new ResumablePeerStream({
@@ -515,10 +529,16 @@ test('failed proactive upgrade preserves transit; a later direct attachment keep
   });
   await left.write(Buffer.from('before-upgrade'));
   assert.deepEqual(await right.read(), Buffer.from('before-upgrade'));
+  await advance(4_999);
+  assert.equal(upgrades, 0);
+  await advance(1);
   assert.deepEqual(await left.read(), Buffer.from('rejected-upgrade-retains-old'));
   assert.equal(left.path?.kind, 'transit');
   await left.write(Buffer.from('old-path-still-live'));
   assert.deepEqual(await right.read(), Buffer.from('old-path-still-live'));
+  await advance(4_999);
+  assert.equal(upgrades, 1);
+  await advance(1);
   assert.deepEqual(await left.read(), Buffer.from('during-path-change'));
   assert.equal(left.path?.kind, 'direct');
   assert.equal(upgrades, 2);

packages/runtime-host/src/__tests__/session-subscription-client.test.ts

@@ -1363,9 +1363,13 @@ test('tolerates a short Host stall without abandoning the connection', {
 
 test('closes an unresponsive request path even while Host notifications continue', {
   timeout: 12_000,
-}, async () => {
+}, async (t) => {
   let received = 0;
   let probes = 0;
+  const probeReceived = deferred<void>();
+  const notificationsReceived = deferred<void>();
+  const finalNotificationReceived = deferred<void>();
+  let sendFinalNotification!: () => Promise<void>;
   await withProtocolPeer(
     async (transport, hostEpoch, rootId) => {
       await transport.read(1_000);
@@ -1381,6 +1385,12 @@ test('closes an unresponsive request path even while Host notifications continue
         state: 'ready',
       });
       let revision = 0;
+      sendFinalNotification = () =>
+        writeProtocolFrame(transport, {
+          kind: 'session.catalog.changed',
+          revision: ++revision,
+          sessionId: 'final-notification',
+        });
       const notifications = setInterval(() => {
         void writeProtocolFrame(transport, {
           kind: 'session.catalog.changed',
@@ -1392,15 +1402,30 @@ test('closes an unresponsive request path even while Host notifications continue
         const probe = decodeClientFrame(await transport.read(1_000));
         assert.ok(!('kind' in probe));
         assert.equal(probe.operation, 'host.status');
+        probeReceived.resolve();
         await transport.closed;
       } finally {
         clearInterval(notifications);
       }
     },
     async (connection) => {
-      connection.subscribeSessionCatalogChanges(() => {
+      let closed = false;
+      void connection.closed.then(() => {
+        closed = true;
+      });
+      connection.subscribeSessionCatalogChanges((event) => {
         received += 1;
+        if (received > 10) notificationsReceived.resolve();
+        if (event.sessionId === 'final-notification') finalNotificationReceived.resolve();
       });
+      t.mock.timers.tick(20);
+      await probeReceived.promise;
+      await notificationsReceived.promise;
+      t.mock.timers.tick(7_999);
+      await sendFinalNotification().catch(() => undefined);
+      await Promise.race([finalNotificationReceived.promise, connection.closed]);
+      assert.equal(closed, false, 'inbound events must not end the pending probe early');
+      t.mock.timers.tick(1);
       await connection.closed;
       assert.ok(received > 10, 'inbound events must remain active during the failed probe');
       assert.equal(probes, 0, 'one-way events cannot acknowledge a probe');
@@ -1411,6 +1436,7 @@ test('closes an unresponsive request path even while Host notifications continue
         probes += 1;
       },
     },
+    () => t.mock.timers.enable({ apis: ['setTimeout'] }),
   );
 });
 
@@ -1422,6 +1448,7 @@ async function withProtocolPeer(
     readonly onLivenessProbe?: () => void;
     readonly onHostStatus?: (status: HostStatusResult) => void;
   } = {},
+  beforeConnect?: () => void,
 ): Promise<void> {
   const base = await mkdtemp(join(tmpdir(), 'maka-runtime-host-subscription-'));
   const capability = await resolveStorageRoot({
@@ -1459,6 +1486,7 @@ async function withProtocolPeer(
       pid: process.pid,
       createdAt: new Date().toISOString(),
     });
+    beforeConnect?.();
     const connected = await connectRuntimeHost({
       rootPath: join(base, 'root'),
       protocol: PROTOCOL,

scripts/ci-workflow-policy.test.mjs

@@ -291,6 +291,13 @@ test('Rust build caches publish immutable source generations only from the defau
       name,
     );
     assert.doesNotMatch(workflow, /kache report [^\n]*--since/u, name);
+    const reports = [...workflow.matchAll(/^\s+(?:run: )?(kache report[^\n]*)$/gmu)].map(
+      ([, command]) => command,
+    );
+    assert.equal(reports.length, 1, name);
+    // The report must reach the raw log, not only the rendered summary panel.
+    assert.equal(reports[0], 'kache report --format github | tee -a "$GITHUB_STEP_SUMMARY"', name);
+    assert.match(workflow, /run: \|\n\s+set -o pipefail\n\s+kache report/u, name);
   }
 });