test(e2e): wait for drained Hermes cron schedule

test/e2e/live/rebuild-hermes-cron-restore.ts

@@ -46,11 +46,16 @@ interface SeededCronJob {
   scriptName: string;
 }
 
+interface SeedCronJobOptions {
+  minimumFutureMs?: number;
+  schedule?: string;
+}
+
 interface GatewayEvidence {
   active_agents: number;
   gateway_state: string;
   pid: number;
-  running_pid: number;
+  running_pid: number | null;
   start_time: number;
 }
 
@@ -158,7 +163,7 @@ export function parseHermesCronBeginReceipt(text: string): CronControlReceipt {
   return payload as unknown as CronControlReceipt;
 }
 
-function assertFutureCronJob(job: JsonObject, seed: SeededCronJob): void {
+function assertFutureCronJob(job: JsonObject, seed: SeededCronJob, minimumFutureMs: number): void {
   expect(job).toMatchObject({
     enabled: true,
     id: seed.id,
@@ -174,8 +179,25 @@ function assertFutureCronJob(job: JsonObject, seed: SeededCronJob): void {
   expect(runtime.completed).toBe(0);
   expect(
     normalizeTimestampMs(runtime.nextRunAt, `cron job ${seed.id} next run`),
-    "seeded recurring cron job must remain well in the future during rebuild",
-  ).toBeGreaterThan(Date.now() + 60 * 60_000);
+    "seeded recurring cron job must remain in the future",
+  ).toBeGreaterThan(Date.now() + minimumFutureMs);
+}
+
+export function hermesCronJobIsDue(job: JsonObject, label: string, nowMs = Date.now()): boolean {
+  const runtime = hermesCronJobRuntimeState(job, label);
+  return normalizeTimestampMs(runtime.nextRunAt, `${label} next run`) <= nowMs;
+}
+
+export function parseHermesGatewayEvidence(text: string): GatewayEvidence {
+  const payload = parseJsonObject(text, "Hermes gateway status");
+  for (const field of ["active_agents", "pid", "start_time"] as const) {
+    if (!Number.isSafeInteger(payload[field])) fail(`Hermes gateway ${field} is invalid`);
+  }
+  if (payload.running_pid !== null && !Number.isSafeInteger(payload.running_pid)) {
+    fail("Hermes gateway running_pid is invalid");
+  }
+  if (typeof payload.gateway_state !== "string") fail("Hermes gateway state is invalid");
+  return payload as unknown as GatewayEvidence;
 }
 
 export function hermesRuntimeExecArgs(sandboxName: string, command: string[]): string[] {
@@ -254,7 +276,10 @@ export function createRebuildHermesCronRestoreFixture({
     return parseCronJob(result.stdout, jobId, artifactName);
   }
 
-  async function seedCronJob(label: string): Promise<SeededCronJob> {
+  async function seedCronJob(
+    label: string,
+    { minimumFutureMs = 60 * 60_000, schedule = "every 1d" }: SeedCronJobOptions = {},
+  ): Promise<SeededCronJob> {
     const pending = uniqueSeed(label);
     const scriptPath = `${CRON_SCRIPTS_ROOT}/${pending.scriptName}`;
     const writeScript = await dockerRoot(
@@ -279,7 +304,7 @@ export function createRebuildHermesCronRestoreFixture({
         "hermes",
         "cron",
         "create",
-        "every 1d",
+        schedule,
         "--no-agent",
         "--script",
         pending.scriptName,
@@ -296,7 +321,7 @@ export function createRebuildHermesCronRestoreFixture({
       evidence: await readCronJob(id, `phase-${label}-read-seeded-hermes-cron-job`),
       id,
     };
-    assertFutureCronJob(seed.evidence, seed);
+    assertFutureCronJob(seed.evidence, seed, minimumFutureMs);
     await assertExecutionMarkerAbsent(seed, `phase-${label}-verify-cron-not-yet-executed`);
     return seed;
   }
@@ -344,6 +369,32 @@ export function createRebuildHermesCronRestoreFixture({
     fail(`Hermes cron job ${seed.id} did not complete exactly once: ${lastEvidence}`);
   }
 
+  async function waitUntilDueWhileBlocked(
+    seed: SeededCronJob,
+    artifactPrefix: string,
+  ): Promise<void> {
+    let lastEvidence = "no scheduler evidence";
+    for (let attempt = 1; attempt <= EXECUTION_POLL_ATTEMPTS; attempt += 1) {
+      const job = await readCronJob(seed.id, `${artifactPrefix}-job-attempt-${attempt}`);
+      const count = await executionCount(seed, `${artifactPrefix}-marker-attempt-${attempt}`);
+      const runtime = hermesCronJobRuntimeState(job, `cron job ${seed.id}`);
+      lastEvidence = JSON.stringify({
+        completed: runtime.completed,
+        count,
+        last_status: runtime.lastStatus,
+        next_run_at: runtime.nextRunAt,
+      });
+      if (runtime.completed !== 0 || count !== 0 || runtime.lastStatus !== null) {
+        fail(`Hermes cron job ${seed.id} executed while dispatch was drained: ${lastEvidence}`);
+      }
+      if (hermesCronJobIsDue(job, `cron job ${seed.id}`)) return;
+      await sleep(POLL_INTERVAL_MS);
+    }
+    fail(
+      `Hermes cron job ${seed.id} did not become due while dispatch was drained: ${lastEvidence}`,
+    );
+  }
+
   async function assertControlMarker(present: boolean, artifactName: string): Promise<void> {
     const command = present
       ? ["stat", "-c", "%u:%g %a %s", CRON_CONTROL_MARKER]
@@ -353,15 +404,6 @@ export function createRebuildHermesCronRestoreFixture({
     if (present) expect(result.stdout.trim()).toMatch(/^0:0 400 [1-9]\d*$/u);
   }
 
-  function parseGatewayEvidence(text: string): GatewayEvidence {
-    const payload = parseJsonObject(text, "Hermes gateway status");
-    for (const field of ["active_agents", "pid", "running_pid", "start_time"] as const) {
-      if (!Number.isSafeInteger(payload[field])) fail(`Hermes gateway ${field} is invalid`);
-    }
-    if (typeof payload.gateway_state !== "string") fail("Hermes gateway state is invalid");
-    return payload as unknown as GatewayEvidence;
-  }
-
   async function gatewayEvidence(artifactName: string): Promise<GatewayEvidence | null> {
     const script = [
       "import json",
@@ -381,7 +423,7 @@ export function createRebuildHermesCronRestoreFixture({
     ].join("\n");
     const result = await dockerRoot([HERMES_PYTHON, "-I", "-c", script], artifactName);
     if (result.exitCode !== 0) return null;
-    return parseGatewayEvidence(result.stdout.trim());
+    return parseHermesGatewayEvidence(result.stdout.trim());
   }
 
   async function waitForGatewayState(
@@ -544,12 +586,13 @@ export function createRebuildHermesCronRestoreFixture({
       const receipt = parseHermesCronBeginReceipt(begin.stdout);
       await assertControlMarker(true, "phase-8-verify-cron-restore-marker-before-restart");
 
-      const recoverySeed = await seedCronJob("recovery");
-      await enqueueManualRun(recoverySeed, "phase-8-queue-due-hermes-cron-during-drain");
-      await sleep(POLL_INTERVAL_MS);
-      await assertExecutionMarkerAbsent(
+      const recoverySeed = await seedCronJob("recovery", {
+        minimumFutureMs: 15_000,
+        schedule: "every 1m",
+      });
+      await waitUntilDueWhileBlocked(
         recoverySeed,
-        "phase-8-verify-due-cron-blocked-before-restart",
+        "phase-8-wait-for-due-cron-blocked-before-restart",
       );
       const beforeRestart = await waitForGatewayState(
         "draining",

test/e2e/support/rebuild-hermes-cron-restore.test.ts

@@ -3,8 +3,10 @@
 
 import { describe, expect, it } from "vitest";
 import {
+  hermesCronJobIsDue,
   hermesCronJobRuntimeState,
   parseHermesCronBeginReceipt,
+  parseHermesGatewayEvidence,
 } from "../live/rebuild-hermes-cron-restore.ts";
 
 describe("Hermes rebuild cron restore evidence", () => {
@@ -69,4 +71,35 @@ describe("Hermes rebuild cron restore evidence", () => {
       start_time: 47767,
     });
   });
+
+  it("distinguishes naturally due cron work from future work", () => {
+    const job = {
+      last_run_at: null,
+      last_status: null,
+      next_run_at: "2026-08-06T19:41:01.000Z",
+      repeat: { completed: 0, times: null },
+      state: "scheduled",
+    };
+
+    expect(hermesCronJobIsDue(job, "cron job fixture", Date.parse("2026-08-06T19:41:00Z"))).toBe(
+      false,
+    );
+    expect(hermesCronJobIsDue(job, "cron job fixture", Date.parse("2026-08-06T19:41:01Z"))).toBe(
+      true,
+    );
+  });
+
+  it("accepts a transient missing gateway process while restart polling continues", () => {
+    expect(
+      parseHermesGatewayEvidence(
+        '{"active_agents":0,"gateway_state":"draining","pid":263,"running_pid":null,"start_time":38291}',
+      ),
+    ).toEqual({
+      active_agents: 0,
+      gateway_state: "draining",
+      pid: 263,
+      running_pid: null,
+      start_time: 38291,
+    });
+  });
 });