feat(automation): re-arm suspended wait timers after cold boot + log cleanup

packages/services/service-automation/src/builtin/index.ts

@@ -51,7 +51,7 @@ export { registerScreenNodes } from './screen-nodes.js';
 export { registerHttpNodes } from './http-nodes.js';
 export { registerConnectorNodes } from './connector-nodes.js';
 export { registerNotifyNode } from './notify-node.js';
-export { registerWaitNode, parseIsoDuration } from './wait-node.js';
+export { registerWaitNode, parseIsoDuration, rearmSuspendedWaitTimers } from './wait-node.js';
 export { registerSubflowNode } from './subflow-node.js';
 
 /**

packages/services/service-automation/src/builtin/wait-node.test.ts

@@ -3,7 +3,8 @@
 import { describe, it, expect, beforeEach } from 'vitest';
 import { AutomationEngine } from '../engine.js';
 import type { NodeExecutor } from '../engine.js';
-import { registerWaitNode, parseIsoDuration } from './wait-node.js';
+import { InMemorySuspendedRunStore } from '../suspended-run-store.js';
+import { registerWaitNode, parseIsoDuration, rearmSuspendedWaitTimers } from './wait-node.js';
 import type { IJobService, JobHandler, JobSchedule } from '@objectstack/spec/contracts';
 
 function silentLogger() {
@@ -138,3 +139,93 @@ describe('wait node executor', () => {
  expect(ran).toEqual(['after']);
  });
 });
+
+describe('rearmSuspendedWaitTimers (cold-boot timer re-arm)', () => {
+ /** Boot a fresh engine wired to `store` with the wait flow registered — one "process". */
+ function bootEngine(store: InMemorySuspendedRunStore, ctx: any, waitConfig: Record<string, unknown>) {
+ const engine = new AutomationEngine(silentLogger());
+ const ran: string[] = [];
+ engine.registerNodeExecutor(markerExecutor(ran));
+ registerWaitNode(engine, ctx);
+ engine.setSuspendedRunStore(store);
+ engine.registerFlow('wait_flow', waitFlow(waitConfig));
+ return { engine, ran };
+ }
+
+ it('re-schedules a future timer on a new engine and resumes when it fires', async () => {
+ const store = new InMemorySuspendedRunStore();
+ const config = { eventType: 'timer', timerDuration: 'PT2H' };
+
+ // Process 1: suspend at the wait, then "die" (engine discarded).
+ const boot1 = fakeJobCtx();
+ const a = bootEngine(store, boot1.ctx, config);
+ const paused = await a.engine.execute('wait_flow');
+ expect(paused.status).toBe('paused');
+ const storedAt = boot1.scheduled[0]?.schedule.at;
+ expect(storedAt).toBeTruthy();
+
+ // Process 2: cold boot — fresh engine + fresh (empty) job service.
+ const boot2 = fakeJobCtx();
+ const b = bootEngine(store, boot2.ctx, config);
+ const job = boot2.ctx.getService('job') as IJobService;
+ const rearmed = await rearmSuspendedWaitTimers(b.engine, store, job, silentLogger());
+ expect(rearmed).toBe(1);
+
+ // Same one-shot job name + the persisted deadline (not a fresh now+2h).
+ expect(boot2.scheduled).toHaveLength(1);
+ expect(boot2.scheduled[0].name).toBe(`flow-wait:${paused.runId}:pause`);
+ expect(boot2.scheduled[0].schedule).toMatchObject({ type: 'once', at: storedAt });
+
+ // Firing the re-armed job resumes the run on the new engine.
+ await boot2.scheduled[0].handler({ jobId: boot2.scheduled[0].name });
+ expect(b.ran).toEqual(['after']);
+ expect(await store.list()).toHaveLength(0); // consumed
+ });
+
+ it('resumes an overdue timer immediately (deadline elapsed while down)', async () => {
+ const store = new InMemorySuspendedRunStore();
+ const config = { eventType: 'timer', timeoutMs: 1 };
+
+ const a = bootEngine(store, ctxNoJob(), config); // degraded: no job service
+ const paused = await a.engine.execute('wait_flow');
+ expect(paused.status).toBe('paused');
+ await new Promise((r) => setTimeout(r, 10)); // let the 1ms deadline lapse
+
+ const boot2 = fakeJobCtx();
+ const b = bootEngine(store, boot2.ctx, config);
+ const rearmed = await rearmSuspendedWaitTimers(b.engine, store, undefined, silentLogger());
+ expect(rearmed).toBe(1);
+ expect(b.ran).toEqual(['after']); // resumed inline, no job needed
+ expect(boot2.scheduled).toHaveLength(0);
+ });
+
+ it('skips non-timer pauses (signal waits have no persisted deadline)', async () => {
+ const store = new InMemorySuspendedRunStore();
+ const config = { eventType: 'signal', signalName: 'contract.renewed' };
+
+ const a = bootEngine(store, ctxNoJob(), config);
+ await a.engine.execute('wait_flow');
+
+ const boot2 = fakeJobCtx();
+ const b = bootEngine(store, boot2.ctx, config);
+ const job = boot2.ctx.getService('job') as IJobService;
+ const rearmed = await rearmSuspendedWaitTimers(b.engine, store, job, silentLogger());
+ expect(rearmed).toBe(0);
+ expect(boot2.scheduled).toHaveLength(0);
+ expect(await store.list()).toHaveLength(1); // still suspended, untouched
+ });
+
+ it('leaves a future timer suspended (with a warning) when no job service exists', async () => {
+ const store = new InMemorySuspendedRunStore();
+ const config = { eventType: 'timer', timerDuration: 'PT2H' };
+
+ const a = bootEngine(store, ctxNoJob(), config);
+ await a.engine.execute('wait_flow');
+
+ const b = bootEngine(store, ctxNoJob(), config);
+ const rearmed = await rearmSuspendedWaitTimers(b.engine, store, undefined, silentLogger());
+ expect(rearmed).toBe(0);
+ expect(b.ran).toEqual([]);
+ expect(await store.list()).toHaveLength(1); // resumable externally later
+ });
+});

packages/services/service-automation/src/builtin/wait-node.ts

@@ -3,7 +3,7 @@
 import type { PluginContext } from '@objectstack/core';
 import { defineActionDescriptor } from '@objectstack/spec/automation';
 import type { IJobService } from '@objectstack/spec/contracts';
-import type { AutomationEngine } from '../engine.js';
+import type { AutomationEngine, SuspendedRunStore } from '../engine.js';
 
 /**
  * `wait` built-in node — a durable pause (ADR-0019 suspend/resume), the timer /
@@ -62,10 +62,16 @@ export function registerWaitNode(engine: AutomationEngine, ctx: PluginContext):
  (typeof wec.timeoutMs === 'number' ? wec.timeoutMs : undefined) ??
  (typeof loose.timeoutMs === 'number' ? (loose.timeoutMs as number) : undefined);
 
+ // Persist the wake deadline as node output: the engine writes output
+ // to variables (`<nodeId>.waitUntil`) *before* snapshotting the
+ // suspended run, so a cold-booted kernel can re-arm the timer from the
+ // durable store ({@link rearmSuspendedWaitTimers}).
+ const at = durationMs && durationMs > 0 ? new Date(Date.now() + durationMs).toISOString() : undefined;
+ const output = at ? { waitUntil: at } : undefined;
+
  const job = getJobService();
- if (job && runId != null && durationMs && durationMs > 0) {
+ if (job && runId != null && at) {
  const jobName = `flow-wait:${String(runId)}:${node.id}`;
- const at = new Date(Date.now() + durationMs).toISOString();
  try {
  await job.schedule(jobName, { type: 'once', at }, async () => {
  try {
@@ -79,7 +85,7 @@ export function registerWaitNode(engine: AutomationEngine, ctx: PluginContext):
  }
  }
  });
- return { success: true, suspend: true, correlation: jobName };
+ return { success: true, suspend: true, correlation: jobName, output };
  } catch (err) {
  ctx.logger.warn(
  `[wait] node '${node.id}': failed to schedule timer resume (${(err as Error)?.message ?? err}); ` +
@@ -92,8 +98,9 @@ export function registerWaitNode(engine: AutomationEngine, ctx: PluginContext):
  `(resume it via resume(runId), or install the job service for durable timers)`,
  );
  }
- // Degrade: still suspend; resumption comes from an external resume().
- return { success: true, suspend: true, correlation: `timer:${node.id}` };
+ // Degrade: still suspend; resumption comes from an external resume()
+ // (or a later boot's re-arm pass, when the deadline was persisted).
+ return { success: true, suspend: true, correlation: `timer:${node.id}`, output };
  }
 
  // signal / webhook / manual / condition — suspend; an external producer
@@ -106,6 +113,93 @@ export function registerWaitNode(engine: AutomationEngine, ctx: PluginContext):
  ctx.logger.info('[Wait Node] 1 built-in node executor registered');
 }
 
+/** Minimal logger surface for {@link rearmSuspendedWaitTimers}. */
+interface RearmLogger {
+ info(msg: string, ...args: unknown[]): void;
+ warn(msg: string, ...args: unknown[]): void;
+}
+
+/**
+ * Re-arm auto-resume timers for suspended timer-`wait` runs after a cold boot
+ * (ADR-0019 follow-up). The one-shot job a `wait` node schedules lives in the
+ * job service's process memory unless that service is itself durable — so a
+ * restart loses the wake-up while the suspended run survives in
+ * `sys_automation_run`. This pass walks the durable store and:
+ *
+ * - **overdue** deadlines (`<nodeId>.waitUntil` in the past) → `resume()` now;
+ * - **future** deadlines → re-schedule the same `flow-wait:<runId>:<nodeId>`
+ * one-shot job (no job service → warn and leave it for an external resume);
+ * - runs without a persisted `waitUntil` (approval / screen / signal pauses,
+ * or pre-deadline-persistence rows) → skipped untouched.
+ *
+ * Double-fire safe: if the original job *did* survive (durable job store), the
+ * second `resume(runId)` finds no suspended run — the engine's resume
+ * idempotency absorbs it. Returns how many runs were resumed or re-armed.
+ *
+ * Called by `AutomationServicePlugin.start()` *after* the flow pull, because
+ * `resume()` needs the flow definitions registered.
+ */
+export async function rearmSuspendedWaitTimers(
+ engine: Pick<AutomationEngine, 'resume'>,
+ store: SuspendedRunStore,
+ job: IJobService | undefined,
+ logger: RearmLogger,
+): Promise<number> {
+ let runs;
+ try {
+ runs = await store.list();
+ } catch (err) {
+ logger.warn(`[wait] timer re-arm: failed to list suspended runs: ${(err as Error)?.message ?? err}`);
+ return 0;
+ }
+
+ let rearmed = 0;
+ for (const run of runs) {
+ const wakeAt = run.variables?.[`${run.nodeId}.waitUntil`];
+ if (typeof wakeAt !== 'string' || !wakeAt) continue; // not a timer wait
+ const atMs = Date.parse(wakeAt);
+ if (Number.isNaN(atMs)) continue;
+
+ if (atMs <= Date.now()) {
+ // Deadline elapsed while the process was down — resume immediately.
+ try {
+ await engine.resume(run.runId);
+ rearmed++;
+ } catch (err) {
+ logger.warn(`[wait] timer re-arm: resume of overdue run '${run.runId}' failed: ${(err as Error)?.message ?? err}`);
+ }
+ continue;
+ }
+
+ if (!job) {
+ logger.warn(
+ `[wait] timer re-arm: run '${run.runId}' waits until ${wakeAt} but no job service is registered — ` +
+ `resume it externally via resume(runId)`,
+ );
+ continue;
+ }
+
+ const jobName = `flow-wait:${run.runId}:${run.nodeId}`;
+ try {
+ await job.schedule(jobName, { type: 'once', at: wakeAt }, async () => {
+ try {
+ await engine.resume(run.runId);
+ } finally {
+ try {
+ await job.cancel?.(jobName);
+ } catch {
+ /* best-effort */
+ }
+ }
+ });
+ rearmed++;
+ } catch (err) {
+ logger.warn(`[wait] timer re-arm: failed to re-schedule run '${run.runId}': ${(err as Error)?.message ?? err}`);
+ }
+ }
+ return rearmed;
+}
+
 /**
  * Parse an ISO-8601 duration (the subset flows use — weeks/days + a time part
  * of hours/minutes/seconds, e.g. `PT1H`, `P3D`, `PT90M`, `P1DT12H`) into

packages/services/service-automation/src/plugin.ts

@@ -1,8 +1,9 @@
 // Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license.
 
 import type { Plugin, PluginContext } from '@objectstack/core';
+import type { IJobService } from '@objectstack/spec/contracts';
 import { AutomationEngine } from './engine.js';
-import { installBuiltinNodes } from './builtin/index.js';
+import { installBuiltinNodes, rearmSuspendedWaitTimers } from './builtin/index.js';
 import { SysAutomationRun } from './sys-automation-run.object.js';
 import { ObjectStoreSuspendedRunStore, type SuspendedRunStoreEngine } from './suspended-run-store.js';
 
@@ -112,9 +113,8 @@ export class AutomationServicePlugin implements Plugin {
  }
 
  async start(ctx: PluginContext): Promise<void> {
- console.warn('[Automation:start] entering start()');
  if (!this.engine) {
-console.warn('[Automation:start] engine missing, bailing');
+ctx.logger.warn('[Automation] start() called before init() — engine missing, skipping');
  return;
  }
 
@@ -131,12 +131,14 @@ export class AutomationServicePlugin implements Plugin {
  // services were wired, so we attach the DB-backed store here. Without an
  // engine (or with `suspendedRunStore: 'memory'`) the in-memory default
  // stands — suspended runs simply don't survive a restart.
+ let durableStore: ObjectStoreSuspendedRunStore | null = null;
  if ((this.options.suspendedRunStore ?? 'auto') !== 'memory') {
  let dataEngine: SuspendedRunStoreEngine | null = null;
  try { dataEngine = ctx.getService<SuspendedRunStoreEngine>('objectql'); }
  catch { try { dataEngine = ctx.getService<SuspendedRunStoreEngine>('data'); } catch { /* none */ } }
  if (dataEngine && typeof dataEngine.find === 'function' && typeof dataEngine.insert === 'function') {
- this.engine.setSuspendedRunStore(new ObjectStoreSuspendedRunStore(dataEngine, ctx.logger));
+ durableStore = new ObjectStoreSuspendedRunStore(dataEngine, ctx.logger);
+ this.engine.setSuspendedRunStore(durableStore);
  ctx.logger.info('[Automation] Suspended-run persistence enabled (sys_automation_run)');
  } else {
  ctx.logger.info('[Automation] No ObjectQL engine — suspended runs kept in-memory only');
@@ -152,14 +154,14 @@ export class AutomationServicePlugin implements Plugin {
  registry?: { listItems?: (type: string) => unknown[] };
  }>('objectql');
  if (!ql) {
-console.warn('[Automation] objectql service not found at start()');
+ctx.logger.debug('[Automation] objectql service not found at start()');
  } else if (!ql.registry) {
-console.warn('[Automation] objectql.registry is undefined at start()');
+ctx.logger.debug('[Automation] objectql.registry is undefined at start()');
  } else if (typeof ql.registry.listItems !== 'function') {
-console.warn('[Automation] objectql.registry.listItems is not a function');
+ctx.logger.debug('[Automation] objectql.registry.listItems is not a function');
  }
  const flows = ql?.registry?.listItems?.('flow') ?? [];
-console.warn(`[Automation] flow pull: registry returned ${flows.length} flow(s)`);
+ctx.logger.debug(`[Automation] flow pull: registry returned ${flows.length} flow(s)`);
  let registered = 0;
  for (const f of flows) {
  const def = f as { name?: string };
@@ -179,6 +181,25 @@ export class AutomationServicePlugin implements Plugin {
  const msg = err instanceof Error ? err.message : String(err);
  ctx.logger.warn(`[Automation] flow pull from ObjectQL registry failed: ${msg}`);
  }
+
+ // ADR-0019 follow-up: re-arm auto-resume timers for runs that were
+ // suspended at a timer-`wait` node when the process went down. Must run
+ // *after* the flow pull above — resume() needs the flow definitions
+ // registered. Overdue deadlines resume immediately; future ones get
+ // their one-shot job re-scheduled. Best-effort: a failure here only
+ // means those runs wait for an external resume(runId).
+ if (durableStore) {
+ let job: IJobService | undefined;
+ try { job = ctx.getService<IJobService>('job'); } catch { /* none */ }
+ try {
+ const rearmed = await rearmSuspendedWaitTimers(this.engine, durableStore, job, ctx.logger);
+ if (rearmed > 0) {
+ ctx.logger.info(`[Automation] Re-armed ${rearmed} suspended wait timer(s) after restart`);
+ }
+ } catch (err) {
+ ctx.logger.warn(`[Automation] wait-timer re-arm failed: ${(err as Error).message}`);
+ }
+ }
  }
 
  async destroy(): Promise<void> {