From 4cbd3f8ab3209d2d41aff81566dad8328e026d4d Mon Sep 17 00:00:00 2001 From: kjgbot Date: Thu, 3 Sep 2026 15:41:14 +0200 Subject: [PATCH 1/4] refactor(sdk): share the tick bounds checks with a caller that needs them early No behaviour change. The six checks emitDueTicks applied inline move to an exported assertTickScheduleValid, and emitDueTicks calls it -- same checks, same order, same messages, not a second copy. A runner that accepts a schedule from an operator has to refuse a bad grid at declaration rather than at the first poll: connecting, attaching a worker and only then discovering --interval-ms was 1.5 has already told the operator it started. Sharing the function is what keeps the CLI's refusal and the emit path's refusal from drifting -- a bound added here is enforced at both ends by construction. sleepInterruptible is extracted from hn-monitor.ts for the same reason: a source that sleeps on a bare setTimeout observes SIGINT only after the sleep elapses, so a once-a-minute schedule hangs for a minute on every Ctrl-C. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01FtQSAcGDta5VH9xiZFT4sR Session-Id: c228933d-4f94-4d83-9a9a-daf3c83b94f1 Session-Id: c228933d-4f94-4d83-9a9a-daf3c83b94f1 Session-Id: c228933d-4f94-4d83-9a9a-daf3c83b94f1 Session-Id: c228933d-4f94-4d83-9a9a-daf3c83b94f1 --- sdk/src/cli/interruptible-sleep.ts | 27 +++++++++++++++++++++++++++ sdk/src/tick-source.ts | 30 +++++++++++++++++++++++------- 2 files changed, 50 insertions(+), 7 deletions(-) create mode 100644 sdk/src/cli/interruptible-sleep.ts diff --git a/sdk/src/cli/interruptible-sleep.ts b/sdk/src/cli/interruptible-sleep.ts new file mode 100644 index 000000000..068f67598 --- /dev/null +++ b/sdk/src/cli/interruptible-sleep.ts @@ -0,0 +1,27 @@ +/** + * Sleep that wakes on abort as well as on timeout. + * + * Extracted from `hn-monitor.ts` when `tick-runner.ts` needed the identical + * behaviour. A long-running event source that sleeps on a bare `setTimeout` + * cannot be shut down promptly: SIGINT is observed only after the current + * sleep elapses, which for a schedule polled once a minute means a minute of + * apparent hang on every Ctrl-C. Both runners therefore share one + * implementation rather than each carrying a subtly different copy. + */ + +/** Resolve after `ms`, or immediately when `signal` aborts — whichever is first. */ +export function sleepInterruptible(ms: number, signal?: AbortSignal): Promise { + return new Promise((resolve) => { + if (signal?.aborted) { resolve(); return; } + let timer: ReturnType | undefined; + const onAbort = (): void => { + if (timer !== undefined) clearTimeout(timer); + resolve(); + }; + timer = setTimeout(() => { + signal?.removeEventListener('abort', onAbort); + resolve(); + }, ms); + signal?.addEventListener('abort', onAbort, { once: true }); + }); +} diff --git a/sdk/src/tick-source.ts b/sdk/src/tick-source.ts index 2f9bb5e07..da3b3db9d 100644 --- a/sdk/src/tick-source.ts +++ b/sdk/src/tick-source.ts @@ -225,6 +225,28 @@ function requireNonNegativeInteger(value: number, field: string): void { } } +/** + * Every bounds check `emitDueTicks` applies, in one exported place. + * + * `emitDueTicks` calls this; it is not a second copy of the rules. A runner + * that accepts a schedule from an operator (`sdk/src/cli/tick-runner.ts`) has + * to refuse a bad grid at DECLARATION rather than at the first poll — a + * runner that connects, attaches a worker and only then discovers that + * `--interval-ms` was `1.5` has already told the operator it started. Sharing + * this function rather than re-deriving the rules is what keeps the CLI's + * refusal and the emit path's refusal from drifting apart: a bound added here + * is enforced at both ends by construction. + */ +export function assertTickScheduleValid(schedule: TickSchedule, nowMs: number): void { + requirePositiveInteger(schedule.intervalMs, 'intervalMs'); + requirePositiveInteger(schedule.maxCatchUp ?? DEFAULT_MAX_CATCH_UP, 'maxCatchUp'); + requireNonNegativeInteger(schedule.epochMs ?? 0, 'epochMs'); + requireNonNegativeInteger(nowMs, 'nowMs'); + if (schedule.scheduleId === '') { + throw new Error('tick schedule: scheduleId must be a non-empty string'); + } +} + /** * Submit a `flows.tick` event for every slot that has come due since the cursor * last advanced, and move the cursor. @@ -242,14 +264,8 @@ export async function emitDueTicks( options: { schedule: TickSchedule; cursor: TickCursor; nowMs: number }, ): Promise { const { schedule, cursor, nowMs } = options; - requirePositiveInteger(schedule.intervalMs, 'intervalMs'); + assertTickScheduleValid(schedule, nowMs); const maxCatchUp = schedule.maxCatchUp ?? DEFAULT_MAX_CATCH_UP; - requirePositiveInteger(maxCatchUp, 'maxCatchUp'); - requireNonNegativeInteger(schedule.epochMs ?? 0, 'epochMs'); - requireNonNegativeInteger(nowMs, 'nowMs'); - if (schedule.scheduleId === '') { - throw new Error('tick schedule: scheduleId must be a non-empty string'); - } const currentSlot = slotFor(schedule, nowMs); const firstDue = cursor.lastEmittedSlot === undefined From 60feb2026284fbf35d77198cc9b12483bdf7431e Mon Sep 17 00:00:00 2001 From: kjgbot Date: Thu, 3 Sep 2026 15:48:50 +0200 Subject: [PATCH 2/4] =?UTF-8?q?feat(sdk):=20flows=20tick=20start=20?= =?UTF-8?q?=E2=80=94=20a=20runner=20that=20keeps=20a=20scheduled=20relayfl?= =?UTF-8?q?ow=20firing?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit #151 shipped the scheduled-trigger primitive; nothing drove it. `agent-relay cloud schedules` reported "No workflow schedules found", the CLI had no reference to tick-source, and `grep -niE "tick|schedule" sdk/src/cli.ts` returned nothing. RFC-0001 names that state: a flow that is never triggered is silently zero. The runner is a sibling of hn-monitor, not a new species: connect journal -> hello -> loop emitDueTicks -> drain on abort -> close. ## The cursor is persisted, and that is the point emitDueTicks starts a FRESH cursor at the current slot, which is right for a schedule's first poll -- a new hourly schedule must not backfill from the epoch. But it means an in-memory-only cursor makes a restart SKIP every slot between shutdown and restart, silently. The dedupe key makes re-delivery harmless, so a lost cursor cannot double-fire; nothing in the primitive protects against the skip. So the runner persists the cursor and reloads it, keeping the distinction: no state on disk means first run, start at the current slot; state behind the grid means catch up and report the arrears. Written temp-then-rename, so a crash mid-write leaves the previous good state. A malformed or foreign-schedule state file is a hard error, not a silent reset -- resetting would convert a corrupted file into a silent skip of everything since the last good emit. ## A skip reaches the operator skippedSlots and TickEmitError exist because a skip that is only a return value is a skip a throw can discard. The runner normalises both shapes into one accounting path, so the success case and the failure case cannot be handled separately and one of them forgotten. Each skipped slot is logged individually with its scheduled instant -- a count lets a reader skim past "3 skipped"; an instant is something an operator can look for and fail to find. ## Bounds, not mechanisms The tests pin the three things that can go wrong, each mutation-verified: restart emits exactly one tick per due slot (M1: ignore the persisted cursor -> 4 tests fail, restart yields [6] instead of [4,5,6]); a slot past maxCatchUp is reported and persisted, including when the same poll then fails; a submit failure leaves the unfired slot due, proven by retrying and seeing 4 and 5 fire while 3 is not re-emitted. A test asserting "the runner fired" would pass under all three mutations. ## One defect found in my own parser parseInt('1.5') is 1, so --interval-ms 1.5 silently became a 1ms schedule. takeNumber now requires an exact integer round-trip, refusing 1.5, 1e3, 0x10, '60000ms' and '' as invocation errors. Bounds themselves are not re-derived here: runTickRunner calls assertTickScheduleValid, the same function emitDueTicks uses, so the CLI's refusal cannot drift from the emit path's. Gates: three tsc configs clean; full SDK suite 461 passed / 3 skipped / 0 failed with RELAYFLOWD_BIN pinned to this worktree's build; kernel unchanged and green. Known limit: this makes a relayflow schedulable and runnable locally. Whether `agent-relay cloud schedules` needs a separate registration step is not established here. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01FtQSAcGDta5VH9xiZFT4sR Session-Id: c228933d-4f94-4d83-9a9a-daf3c83b94f1 Session-Id: c228933d-4f94-4d83-9a9a-daf3c83b94f1 Session-Id: c228933d-4f94-4d83-9a9a-daf3c83b94f1 Session-Id: c228933d-4f94-4d83-9a9a-daf3c83b94f1 --- sdk/src/cli.ts | 109 ++++++++++- sdk/src/cli/tick-runner.ts | 346 +++++++++++++++++++++++++++++++++ sdk/tests/tick-runner.test.ts | 348 ++++++++++++++++++++++++++++++++++ 3 files changed, 802 insertions(+), 1 deletion(-) create mode 100644 sdk/src/cli/tick-runner.ts create mode 100644 sdk/tests/tick-runner.test.ts diff --git a/sdk/src/cli.ts b/sdk/src/cli.ts index 6c393ea14..b9569b6ab 100644 --- a/sdk/src/cli.ts +++ b/sdk/src/cli.ts @@ -14,6 +14,7 @@ import { type RunReport, } from './cli/run.js'; import { runHnMonitor } from './cli/hn-monitor.js'; +import { runTickRunner } from './cli/tick-runner.js'; export type { CheckInputDiagnostic, CheckReport } from './cli/check.js'; @@ -26,13 +27,17 @@ type CliExitCode = 0 | 1 | 2 | 3; type ParsedArgs = | { command: 'check'; json: boolean; value: string } | { command: 'run' | 'resume'; dataDir: string; json: boolean; value: string } - | { command: 'hn-monitor'; sub: 'start'; dataDir: string; specPath: string; pollIntervalMs: number | undefined }; + | { command: 'hn-monitor'; sub: 'start'; dataDir: string; specPath: string; pollIntervalMs: number | undefined } + | { command: 'tick'; sub: 'start'; dataDir: string; specPath: string; scheduleId: string; + intervalMs: number; epochMs: number | undefined; maxCatchUp: number | undefined; + pollIntervalMs: number | undefined }; const DEFAULT_DATA_DIR = '.relayflowd'; const USAGE = [ 'Usage:', 'flows check [--json] ', 'flows run [--json] [--data-dir ] ', + 'flows tick start --schedule-id --interval-ms [--epoch-ms ] [--max-catch-up ] [--poll-interval-ms ] [--data-dir ] ', 'flows resume [--json] [--data-dir ] ', 'flows hn-monitor start [--data-dir ] [--poll-interval-ms ] ', ].join(' '); @@ -77,6 +82,30 @@ export async function runCli( } } + if (parsed.command === 'tick') { + const controller = new AbortController(); + const onSignal = (): void => controller.abort(); + process.once('SIGINT', onSignal); + process.once('SIGTERM', onSignal); + try { + return await runTickRunner({ + dataDir: parsed.dataDir, + specPath: parsed.specPath, + schedule: { + scheduleId: parsed.scheduleId, + intervalMs: parsed.intervalMs, + ...(parsed.epochMs === undefined ? {} : { epochMs: parsed.epochMs }), + ...(parsed.maxCatchUp === undefined ? {} : { maxCatchUp: parsed.maxCatchUp }), + }, + pollIntervalMs: parsed.pollIntervalMs, + signal: controller.signal, + }, io) as CliExitCode; + } finally { + process.off('SIGINT', onSignal); + process.off('SIGTERM', onSignal); + } + } + const execution = parsed.command === 'run' ? await runFlow(parsed.value, parsed.dataDir, { onWait: (progress) => emitWait(progress, io) }) : await resumeFlow(parsed.value, parsed.dataDir, { onWait: (progress) => emitWait(progress, io) }); @@ -97,6 +126,7 @@ function emitWait( function parseArgs(args: readonly string[]): ParsedArgs | undefined { const command = args[0]; if (command === 'hn-monitor') return parseHnMonitorArgs(args.slice(1)); + if (command === 'tick') return parseTickArgs(args.slice(1)); if (command !== 'check' && command !== 'run' && command !== 'resume') return undefined; let json = false; @@ -162,6 +192,83 @@ function parseHnMonitorArgs(rest: readonly string[]): ParsedArgs | undefined { return { command: 'hn-monitor', sub: 'start', dataDir, specPath: positionals[0]!, pollIntervalMs }; } +/** + * `flows tick start`. Numeric flags are parsed here but their BOUNDS are not + * re-derived: `runTickRunner` calls `assertTickScheduleValid`, the same + * function `emitDueTicks` uses, so the CLI's refusal and the emit path's + * refusal cannot drift. This parser only rejects shapes it cannot turn into a + * number at all. + */ +function parseTickArgs(rest: readonly string[]): ParsedArgs | undefined { + const sub = rest[0]; + if (sub !== 'start') return undefined; + + let dataDir = DEFAULT_DATA_DIR; + let sawDataDir = false; + let scheduleId: string | undefined; + let intervalMs: number | undefined; + let epochMs: number | undefined; + let maxCatchUp: number | undefined; + let pollIntervalMs: number | undefined; + const positionals: string[] = []; + + // Number, not parseInt: parseInt('1.5') is 1, so a fractional --interval-ms + // would silently become a 1ms schedule instead of being refused. Requiring + // an exact integer round-trip rejects '1.5', '1e3', '0x10' and ' 1' rather + // than coercing them into something the operator did not write. + const takeNumber = (value: string | undefined): number | undefined => { + if (value === undefined || value.startsWith('-')) return undefined; + const parsed = Number(value); + if (!Number.isInteger(parsed) || String(parsed) !== value) return undefined; + return parsed; + }; + + for (let index = 1; index < rest.length; index += 1) { + const argument = rest[index]!; + if (argument === '--data-dir') { + const value = rest[index + 1]; + if (sawDataDir || value === undefined || value.startsWith('-')) return undefined; + dataDir = value; sawDataDir = true; index += 1; continue; + } + if (argument === '--schedule-id') { + const value = rest[index + 1]; + if (scheduleId !== undefined || value === undefined || value.startsWith('-')) return undefined; + scheduleId = value; index += 1; continue; + } + if (argument === '--interval-ms') { + if (intervalMs !== undefined) return undefined; + const value = takeNumber(rest[index + 1]); + if (value === undefined) return undefined; + intervalMs = value; index += 1; continue; + } + if (argument === '--epoch-ms') { + if (epochMs !== undefined) return undefined; + const value = takeNumber(rest[index + 1]); + if (value === undefined) return undefined; + epochMs = value; index += 1; continue; + } + if (argument === '--max-catch-up') { + if (maxCatchUp !== undefined) return undefined; + const value = takeNumber(rest[index + 1]); + if (value === undefined) return undefined; + maxCatchUp = value; index += 1; continue; + } + if (argument === '--poll-interval-ms') { + if (pollIntervalMs !== undefined) return undefined; + const value = takeNumber(rest[index + 1]); + if (value === undefined) return undefined; + pollIntervalMs = value; index += 1; continue; + } + if (argument.startsWith('-')) return undefined; + positionals.push(argument); + } + if (positionals.length !== 1 || scheduleId === undefined || intervalMs === undefined) return undefined; + return { + command: 'tick', sub: 'start', dataDir, specPath: positionals[0]!, + scheduleId, intervalMs, epochMs, maxCatchUp, pollIntervalMs, + }; +} + function emitCheckReport(report: CheckReport, json: boolean, io: CliIo): void { emitDiagnostics(report.diagnostics, io); if (json) { diff --git a/sdk/src/cli/tick-runner.ts b/sdk/src/cli/tick-runner.ts new file mode 100644 index 000000000..98146dd97 --- /dev/null +++ b/sdk/src/cli/tick-runner.ts @@ -0,0 +1,346 @@ +/** + * `flows tick start` — drive a scheduled relayflow. + * + * A sibling of `hn-monitor.ts`, not a new species: a public function that + * composes the primitives directly — connect journal → hello → loop + * `emitDueTicks` → drain on abort → close. The differences from hn-monitor are + * only the ones the grid forces. + * + * ## Why the cursor is persisted, and why that is the whole point + * + * `emitDueTicks` starts a FRESH cursor at the current slot: + * + * const firstDue = cursor.lastEmittedSlot === undefined + * ? currentSlot + * : cursor.lastEmittedSlot + 1; + * + * That is right for a schedule's first ever poll — a new hourly schedule must + * not backfill from the epoch. But it means an in-memory-only cursor makes a + * restart SKIP every slot between shutdown and restart, silently. The dedupe + * key `(schedule_id, scheduled_for_ms)` makes re-delivery of a slot harmless, + * so a lost cursor cannot double-fire; nothing in the primitive protects + * against the skip. The runner is where that is either handled or lost, so it + * persists the cursor and reloads it on start. + * + * The distinction the file keeps: "no cursor on disk" means first run, start + * at the current slot. "A cursor on disk behind the grid" means catch up, and + * report anything past `maxCatchUp` as skipped. Conflating the two either + * backfills a new schedule from 1970 or silently drops a restart's arrears. + * + * ## Why a skip must reach the operator + * + * `skippedSlots` and `TickEmitError` exist because a skip that is only a + * return value is a skip that a `throw` can discard. `emitDueTicks` guarantees + * the accounting survives its own failure; this runner is the consumer that + * makes it visible. A skipped slot is real scheduled work that did not run. + */ + +import { mkdir, readFile, writeFile } from 'node:fs/promises'; +import { dirname, isAbsolute, join, resolve } from 'node:path'; +import { JournalClient } from '../journal-client.js'; +import { + DEFAULT_MAX_CATCH_UP, + TickEmitError, + assertTickScheduleValid, + emitDueTicks, + scheduledForMs, + slotFor, + type TickCursor, + type TickEmitResult, + type TickSchedule, +} from '../tick-source.js'; +import type { EventSubmitResult, HelloResult } from '../protocol.js'; +import type { CliIo } from '../cli.js'; +import { sleepInterruptible } from './interruptible-sleep.js'; + +/** How often to look for due slots when the caller does not say. */ +const DEFAULT_POLL_INTERVAL_MS = 15_000; + +/** + * Minimum client surface the runner uses. Concrete protocol return types + * rather than `unknown`, so a rename in the journal protocol fails to compile + * here instead of drifting silently past the interface — the same reason + * `HnMonitorClient` is shaped this way. + */ +export interface TickRunnerClient { + hello(client: string): Promise; + eventSubmit( + spec: unknown, + event: { type: string; payload?: unknown; key?: string }, + ): Promise; + close(): void; +} + +/** Where the runner's durable cursor lives, and what it holds. */ +export interface TickRunnerState { + scheduleId: string; + /** Mirror of `TickCursor.lastEmittedSlot`. Absent before the first emit. */ + lastEmittedSlot?: number; + /** + * Wall clock of the last successful submit. Not used for scheduling — the + * grid owns that — but it is what makes a dead runner detectable, so + * `staleAfterMs` means something end to end rather than only inside the + * kernel's own sweep. + */ + lastEmittedAtMs?: number; + /** Every slot ever passed over, oldest first. Append-only. */ + skippedSlots: number[]; +} + +interface TickRunnerArgsBase { + /** Data directory containing `relayflowd.sock`. Required. */ + dataDir: string; + /** Absolute path to the canonical flow spec JSON. Required. */ + specPath: string; + /** The grid. Validated at declaration, before anything connects. */ + schedule: TickSchedule; + /** How often to look for due slots. Default 15000ms. */ + pollIntervalMs?: number; + /** + * Where to persist the cursor. Defaults to + * `/tick-state/.json`. + */ + statePath?: string; + /** + * Cap on poll iterations. Undefined = unbounded (production). 0 means + * "connect, poll zero times, drain, exit 0" — used by tests that only need + * the setup and teardown paths. + */ + maxPolls?: number; + /** AbortSignal for external cancellation (tests, SIGINT wiring). */ + signal?: AbortSignal; + /** Injectable clock so tests can place `now` on the grid deliberately. */ + now?: () => number; +} + +/** + * Test injection surface. `connectClient` is supplied alone here — unlike + * hn-monitor, this runner attaches no worker, so there is no pairing to + * enforce: a tick is submitted through `event.submit` and the kernel dispatches + * to whatever worker the flow's trigger names. + */ +export interface TickRunnerArgs extends TickRunnerArgsBase { + connectClient?: (socketPath: string) => TickRunnerClient; +} + +/** Absolute path to the state file for one schedule. */ +export function tickStatePath(dataDir: string, scheduleId: string): string { + return join(dataDir, 'tick-state', `${encodeURIComponent(scheduleId)}.json`); +} + +/** + * Load the durable cursor, distinguishing "no state" from "state behind the + * grid". A missing file is a first run and must start at the current slot; a + * present file must be honoured however far behind it is, so the arrears are + * either caught up or reported. + * + * A malformed or foreign-schedule state file is a hard error rather than a + * silent reset: resetting would convert an operator's corrupted file into a + * silent skip of everything since the last good emit, which is the failure + * this runner exists to prevent. + */ +export async function loadTickState( + path: string, + scheduleId: string, +): Promise { + let raw: string; + try { + raw = await readFile(path, 'utf8'); + } catch (cause) { + if ((cause as { code?: string }).code === 'ENOENT') { + return { scheduleId, skippedSlots: [] }; + } + throw cause; + } + let parsed: unknown; + try { + parsed = JSON.parse(raw); + } catch (cause) { + throw new Error(`tick state at ${path} is not valid JSON`, { cause }); + } + if (typeof parsed !== 'object' || parsed === null) { + throw new Error(`tick state at ${path} is not an object`); + } + const state = parsed as Partial; + if (state.scheduleId !== scheduleId) { + throw new Error( + `tick state at ${path} belongs to schedule ${String(state.scheduleId)}, not ${scheduleId}`, + ); + } + if (state.lastEmittedSlot !== undefined && !Number.isInteger(state.lastEmittedSlot)) { + throw new Error(`tick state at ${path} has a non-integer lastEmittedSlot`); + } + return { + scheduleId, + ...(state.lastEmittedSlot === undefined ? {} : { lastEmittedSlot: state.lastEmittedSlot }), + ...(state.lastEmittedAtMs === undefined ? {} : { lastEmittedAtMs: state.lastEmittedAtMs }), + skippedSlots: Array.isArray(state.skippedSlots) ? state.skippedSlots : [], + }; +} + +/** + * Persist the cursor. Written to a temp file and renamed, so a crash mid-write + * leaves the previous good state rather than a truncated file — a truncated + * file would be a hard error on next start, which is safe but needlessly + * blocks a runner that had a perfectly good cursor a moment earlier. + */ +export async function saveTickState(path: string, state: TickRunnerState): Promise { + await mkdir(dirname(path), { recursive: true }); + const temp = `${path}.tmp`; + await writeFile(temp, `${JSON.stringify(state, null, 2)}\n`, 'utf8'); + const { rename } = await import('node:fs/promises'); + await rename(temp, path); +} + +/** One poll's accounting, for logging and for tests to assert on. */ +export interface TickPollReport { + emittedSlots: number[]; + skippedSlots: number[]; + /** Set when the poll threw. The partial accounting is still present. */ + failure?: unknown; +} + +/** + * Run `flows tick start`. Returns 0 on clean shutdown, 1 on a fatal error. + * + * Fatal means: the schedule is invalid, the state file is unreadable or + * belongs to another schedule, the journal cannot be reached, or a submit + * failed. A submit failure is fatal by design — `emitDueTicks` leaves the slot + * due, so retrying is the next poll's job, but a runner that swallows journal + * failures and keeps looping is the silent-zero this is meant to prevent. + */ +export async function runTickRunner(args: TickRunnerArgs, io: CliIo): Promise { + const now = args.now ?? Date.now; + const pollIntervalMs = args.pollIntervalMs ?? DEFAULT_POLL_INTERVAL_MS; + + // Refuse a bad grid BEFORE connecting. A runner that attaches and only then + // discovers `--interval-ms` was 1.5 has already told the operator it started. + try { + assertTickScheduleValid(args.schedule, now()); + } catch (cause) { + io.stderr(`REFUSED [invalid_schedule] ${(cause as Error).message}`); + return 1; + } + if (!Number.isInteger(pollIntervalMs) || pollIntervalMs <= 0) { + io.stderr('REFUSED [invalid_schedule] pollIntervalMs must be a positive integer'); + return 1; + } + + const specPath = isAbsolute(args.specPath) ? args.specPath : resolve(args.specPath); + const statePath = args.statePath ?? tickStatePath(args.dataDir, args.schedule.scheduleId); + + let spec: unknown; + let state: TickRunnerState; + try { + spec = JSON.parse(await readFile(specPath, 'utf8')); + state = await loadTickState(statePath, args.schedule.scheduleId); + } catch (cause) { + io.stderr(`REFUSED [invalid_state] ${(cause as Error).message}`); + return 1; + } + + const resuming = state.lastEmittedSlot !== undefined; + const cursor: TickCursor = resuming ? { lastEmittedSlot: state.lastEmittedSlot } : {}; + + const socketPath = join(args.dataDir, 'relayflowd.sock'); + const client = args.connectClient + ? args.connectClient(socketPath) + : (new JournalClient(socketPath) as unknown as TickRunnerClient); + + let exitCode = 0; + try { + await client.hello('flows-tick-runner'); + const startSlot = slotFor(args.schedule, now()); + io.stdout( + resuming + ? `TICK_RUNNER schedule=${args.schedule.scheduleId} resuming from slot ${String(state.lastEmittedSlot)}; current slot ${startSlot}` + : `TICK_RUNNER schedule=${args.schedule.scheduleId} first run; starting at slot ${startSlot}`, + ); + + let polls = 0; + while (!(args.signal?.aborted ?? false)) { + if (args.maxPolls !== undefined && polls >= args.maxPolls) break; + polls += 1; + + const report = await pollOnce(spec, client, args.schedule, cursor, now()); + + // Persist before reporting. The cursor is the thing a restart depends + // on; losing it costs slots, whereas losing a log line costs a message. + if (report.emittedSlots.length > 0) { + state.lastEmittedSlot = cursor.lastEmittedSlot; + state.lastEmittedAtMs = now(); + } + if (report.skippedSlots.length > 0) { + state.skippedSlots.push(...report.skippedSlots); + } + if (report.emittedSlots.length > 0 || report.skippedSlots.length > 0) { + await saveTickState(statePath, state); + } + + for (const slot of report.skippedSlots) { + // Loud, per slot, with the instant it stood for. A count would let a + // reader skim past "3 skipped"; an instant is a thing an operator can + // go and look for in the journal and fail to find. + io.stderr( + `TICK_SKIPPED schedule=${args.schedule.scheduleId} slot=${slot} scheduled_for_ms=${scheduledForMs(args.schedule, slot)} — past maxCatchUp=${args.schedule.maxCatchUp ?? DEFAULT_MAX_CATCH_UP}; this scheduled work did NOT run`, + ); + } + for (const slot of report.emittedSlots) { + io.stdout( + `TICK_EMITTED schedule=${args.schedule.scheduleId} slot=${slot} scheduled_for_ms=${scheduledForMs(args.schedule, slot)}`, + ); + } + + if (report.failure !== undefined) { + const failure = report.failure; + const name = failure instanceof Error ? failure.constructor.name : typeof failure; + const message = failure instanceof Error ? failure.message : String(failure); + io.stderr(`TICK_RUNNER_FAILED ${name}: ${message}`); + return 1; + } + + if (args.signal?.aborted ?? false) break; + if (args.maxPolls !== undefined && polls >= args.maxPolls) break; + await sleepInterruptible(pollIntervalMs, args.signal); + } + io.stdout(`TICK_RUNNER_STOPPED schedule=${args.schedule.scheduleId} polls=${polls}`); + } catch (cause) { + const name = cause instanceof Error ? cause.constructor.name : typeof cause; + io.stderr(`TICK_RUNNER_FAILED ${name}: ${(cause as Error).message}`); + exitCode = 1; + } finally { + client.close(); + } + return exitCode; +} + +/** + * One poll. Normalises the two shapes `emitDueTicks` can produce — a result, + * or a `TickEmitError` carrying the partial result — into one report, so the + * caller has exactly one accounting path and cannot handle the success case + * and forget the failure case. + */ +async function pollOnce( + spec: unknown, + sink: TickRunnerClient, + schedule: TickSchedule, + cursor: TickCursor, + nowMs: number, +): Promise { + try { + const result: TickEmitResult = await emitDueTicks(spec, sink, { schedule, cursor, nowMs }); + return { emittedSlots: result.emittedSlots, skippedSlots: result.skippedSlots }; + } catch (cause) { + if (cause instanceof TickEmitError) { + // The partial accounting travelled with the failure. Surface it, then + // let the caller treat the failure as fatal — the unemitted slots are + // still due, because `emitDueTicks` only advances on success. + return { + emittedSlots: cause.emittedSlots, + skippedSlots: cause.skippedSlots, + failure: cause.cause ?? cause, + }; + } + return { emittedSlots: [], skippedSlots: [], failure: cause }; + } +} diff --git a/sdk/tests/tick-runner.test.ts b/sdk/tests/tick-runner.test.ts new file mode 100644 index 000000000..f8bdad06f --- /dev/null +++ b/sdk/tests/tick-runner.test.ts @@ -0,0 +1,348 @@ +/** + * `tick-runner` tests. + * + * Every test here is written against a bound, not a mechanism. "The runner + * fired" is nearly worthless: it passes whatever the cursor, the skip + * accounting or the failure handling do. The three things that can actually go + * wrong are: + * + * 1. a restart skips the slots between shutdown and restart; + * 2. a slot passed over by `maxCatchUp` is not reported anywhere; + * 3. a submit failure advances the cursor past a slot that never fired. + * + * Each has a test that fails if the behaviour is removed while the runner still + * runs, and each is mutation-verified in the repair report. + */ + +import { mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import { + loadTickState, + runTickRunner, + saveTickState, + tickStatePath, + type TickRunnerClient, + type TickRunnerState, +} from '../src/cli/tick-runner.js'; +import type { TickSchedule } from '../src/tick-source.js'; +import type { CliIo } from '../src/cli.js'; + +const INTERVAL = 60_000; +const EPOCH = 1_000_000; + +const SCHEDULE: TickSchedule = { + scheduleId: 'heartbeat', + intervalMs: INTERVAL, + epochMs: EPOCH, + maxCatchUp: 5, +}; + +/** `nowMs` placed exactly on slot `n`'s instant. */ +function atSlot(n: number): number { + return EPOCH + n * INTERVAL; +} + +interface Recorded { + slots: number[]; + scheduledFor: number[]; +} + +/** + * A sink that records what actually reached `event.submit`, and can be told to + * throw on the Nth call. Recording the payload rather than a call count is + * what lets a test assert "exactly one submit per due slot" instead of + * "something was submitted". + */ +function makeClient(throwOnCall?: number): { client: TickRunnerClient; recorded: Recorded } { + const recorded: Recorded = { slots: [], scheduledFor: [] }; + let calls = 0; + const client: TickRunnerClient = { + hello: async () => ({ protocol: 'v0', server: 'test' }) as never, + eventSubmit: async (_spec, event) => { + calls += 1; + if (throwOnCall !== undefined && calls === throwOnCall) { + throw new Error('journal write refused'); + } + const payload = event.payload as { slot: number; scheduled_for_ms: number }; + recorded.slots.push(payload.slot); + recorded.scheduledFor.push(payload.scheduled_for_ms); + return { matched: 0 } as never; + }, + close: () => undefined, + }; + return { client, recorded }; +} + +function makeIo(): { io: CliIo; out: string[]; err: string[] } { + const out: string[] = []; + const err: string[] = []; + return { io: { stdout: (l) => out.push(l), stderr: (l) => err.push(l) }, out, err }; +} + +let dir: string; +let specPath: string; + +beforeEach(async () => { + dir = await mkdtemp(join(tmpdir(), 'tick-runner-')); + specPath = join(dir, 'spec.json'); + await writeFile(specPath, JSON.stringify({ steps: [] }), 'utf8'); +}); + +afterEach(async () => { + await rm(dir, { recursive: true, force: true }); +}); + +describe('bound 1: a restart emits exactly one tick per due slot', () => { + it('resumes from the persisted cursor instead of jumping to the current slot', async () => { + // First runner: one poll at slot 3. Fresh cursor starts AT the current + // slot, so this emits slot 3 only. + const first = makeClient(); + const io1 = makeIo(); + expect( + await runTickRunner( + { dataDir: dir, specPath, schedule: SCHEDULE, maxPolls: 1, + now: () => atSlot(3), connectClient: () => first.client }, + io1.io, + ), + ).toBe(0); + expect(first.recorded.slots).toEqual([3]); + + // The process dies. A NEW runner starts at slot 6 — three slots later. + // Without a persisted cursor it would start at slot 6 and slots 4 and 5 + // would never fire and never be reported: the silent skip. + const second = makeClient(); + const io2 = makeIo(); + expect( + await runTickRunner( + { dataDir: dir, specPath, schedule: SCHEDULE, maxPolls: 1, + now: () => atSlot(6), connectClient: () => second.client }, + io2.io, + ), + ).toBe(0); + + expect(second.recorded.slots).toEqual([4, 5, 6]); + // Exactly one submit per slot across both lifetimes, no duplicates. + expect([...first.recorded.slots, ...second.recorded.slots]).toEqual([3, 4, 5, 6]); + expect(io2.out.some((l) => l.includes('resuming from slot 3'))).toBe(true); + }); + + it('starts at the current slot on a genuine first run, not at the epoch', async () => { + // The other half of the distinction: no state on disk must NOT backfill + // from slot 0, or a new hourly schedule floods the kernel with history. + const { client, recorded } = makeClient(); + const io = makeIo(); + await runTickRunner( + { dataDir: dir, specPath, schedule: SCHEDULE, maxPolls: 1, + now: () => atSlot(500), connectClient: () => client }, + io.io, + ); + expect(recorded.slots).toEqual([500]); + expect(io.out.some((l) => l.includes('first run'))).toBe(true); + }); + + it('persists the cursor and the last-emitted wall clock so a dead runner is detectable', async () => { + const { client } = makeClient(); + await runTickRunner( + { dataDir: dir, specPath, schedule: SCHEDULE, maxPolls: 1, + now: () => atSlot(9), connectClient: () => client }, + makeIo().io, + ); + const state = await loadTickState(tickStatePath(dir, 'heartbeat'), 'heartbeat'); + expect(state.lastEmittedSlot).toBe(9); + expect(state.lastEmittedAtMs).toBe(atSlot(9)); + }); +}); + +describe('bound 2: a slot passed over by maxCatchUp is reported, never lost', () => { + it('names every skipped slot and its scheduled instant, and persists them', async () => { + // Cursor at slot 0, now at slot 20, maxCatchUp 5 → slots 1..15 are due but + // over the bound. They must appear on stderr individually AND in the state + // file; a count alone would let a reader skim past them. + const statePath = tickStatePath(dir, 'heartbeat'); + await saveTickState(statePath, { scheduleId: 'heartbeat', lastEmittedSlot: 0, skippedSlots: [] }); + + const { client, recorded } = makeClient(); + const io = makeIo(); + expect( + await runTickRunner( + { dataDir: dir, specPath, schedule: SCHEDULE, maxPolls: 1, + now: () => atSlot(20), connectClient: () => client }, + io.io, + ), + ).toBe(0); + + // Newest 5 emitted, oldest 15 skipped. + expect(recorded.slots).toEqual([16, 17, 18, 19, 20]); + + const skipLines = io.err.filter((l) => l.startsWith('TICK_SKIPPED')); + expect(skipLines).toHaveLength(15); + // Each line carries the instant, so an operator can go looking for it. + expect(skipLines[0]).toContain('slot=1'); + expect(skipLines[0]).toContain(`scheduled_for_ms=${atSlot(1)}`); + expect(skipLines[0]).toContain('did NOT run'); + + const state = await loadTickState(statePath, 'heartbeat'); + expect(state.skippedSlots).toEqual(Array.from({ length: 15 }, (_, i) => i + 1)); + }); + + it('reports skipped slots even when the poll then fails', async () => { + // The shape the primitive's TickEmitError exists for: a skip and a submit + // failure in the same poll. If the runner only read the success path's + // result, these 15 skips would die with the discarded return value. + const statePath = tickStatePath(dir, 'heartbeat'); + await saveTickState(statePath, { scheduleId: 'heartbeat', lastEmittedSlot: 0, skippedSlots: [] }); + + const { client } = makeClient(1); // throw on the very first submit + const io = makeIo(); + expect( + await runTickRunner( + { dataDir: dir, specPath, schedule: SCHEDULE, maxPolls: 1, + now: () => atSlot(20), connectClient: () => client }, + io.io, + ), + ).toBe(1); + + expect(io.err.filter((l) => l.startsWith('TICK_SKIPPED'))).toHaveLength(15); + expect(io.err.some((l) => l.startsWith('TICK_RUNNER_FAILED'))).toBe(true); + const state = await loadTickState(statePath, 'heartbeat'); + expect(state.skippedSlots).toHaveLength(15); + }); +}); + +describe('bound 3: a submit failure leaves the unfired slot due', () => { + it('does not advance the cursor past a slot whose submit threw', async () => { + const statePath = tickStatePath(dir, 'heartbeat'); + await saveTickState(statePath, { scheduleId: 'heartbeat', lastEmittedSlot: 2, skippedSlots: [] }); + + // Slots 3, 4, 5 due. Throw on the second submit, so 3 lands and 4 does not. + const { client, recorded } = makeClient(2); + const io = makeIo(); + expect( + await runTickRunner( + { dataDir: dir, specPath, schedule: SCHEDULE, maxPolls: 1, + now: () => atSlot(5), connectClient: () => client }, + io.io, + ), + ).toBe(1); + + expect(recorded.slots).toEqual([3]); + // The cursor sits at 3, not 5. Slots 4 and 5 are still due. + const state = await loadTickState(statePath, 'heartbeat'); + expect(state.lastEmittedSlot).toBe(3); + + // Prove it by running again with a working sink: 4 and 5 fire, and 3 is + // not re-emitted. + const retry = makeClient(); + expect( + await runTickRunner( + { dataDir: dir, specPath, schedule: SCHEDULE, maxPolls: 1, + now: () => atSlot(5), connectClient: () => retry.client }, + makeIo().io, + ), + ).toBe(0); + expect(retry.recorded.slots).toEqual([4, 5]); + }); +}); + +describe('fail closed at declaration, before anything connects', () => { + it.each([ + ['non-integer intervalMs', { ...SCHEDULE, intervalMs: 1.5 }], + ['zero intervalMs', { ...SCHEDULE, intervalMs: 0 }], + ['negative epochMs', { ...SCHEDULE, epochMs: -1 }], + ['NaN epochMs', { ...SCHEDULE, epochMs: Number.NaN }], + ['empty scheduleId', { ...SCHEDULE, scheduleId: '' }], + ])('refuses %s without opening a connection', async (_label, schedule) => { + let connected = false; + const io = makeIo(); + expect( + await runTickRunner( + { dataDir: dir, specPath, schedule: schedule as TickSchedule, maxPolls: 1, + now: () => atSlot(1), + connectClient: () => { connected = true; return makeClient().client; } }, + io.io, + ), + ).toBe(1); + // The point of validating early: the operator is not told it started. + expect(connected).toBe(false); + expect(io.err.some((l) => l.startsWith('REFUSED [invalid_schedule]'))).toBe(true); + }); + + it('refuses a state file belonging to another schedule rather than resetting it', async () => { + // A silent reset would convert a corrupted file into a silent skip of + // everything since the last good emit — the failure this runner exists to + // prevent. + const statePath = tickStatePath(dir, 'heartbeat'); + await saveTickState(statePath, { scheduleId: 'something-else', skippedSlots: [] } as TickRunnerState); + const io = makeIo(); + expect( + await runTickRunner( + { dataDir: dir, specPath, schedule: SCHEDULE, maxPolls: 1, + now: () => atSlot(1), connectClient: () => makeClient().client }, + io.io, + ), + ).toBe(1); + expect(io.err.some((l) => l.includes('belongs to schedule something-else'))).toBe(true); + }); + + it('refuses a truncated state file rather than starting from nothing', async () => { + const statePath = tickStatePath(dir, 'heartbeat'); + await saveTickState(statePath, { scheduleId: 'heartbeat', lastEmittedSlot: 4, skippedSlots: [] }); + await writeFile(statePath, '{"scheduleId":"heart', 'utf8'); + const io = makeIo(); + expect( + await runTickRunner( + { dataDir: dir, specPath, schedule: SCHEDULE, maxPolls: 1, + now: () => atSlot(9), connectClient: () => makeClient().client }, + io.io, + ), + ).toBe(1); + expect(io.err.some((l) => l.includes('not valid JSON'))).toBe(true); + }); +}); + +describe('state file durability', () => { + it('leaves the previous good state when a write is interrupted', async () => { + // saveTickState writes a temp file and renames, so a crash mid-write + // cannot truncate the live file. + const statePath = tickStatePath(dir, 'heartbeat'); + await saveTickState(statePath, { scheduleId: 'heartbeat', lastEmittedSlot: 7, skippedSlots: [] }); + const before = await readFile(statePath, 'utf8'); + await writeFile(`${statePath}.tmp`, '{"partial', 'utf8'); + expect(await readFile(statePath, 'utf8')).toBe(before); + expect((await loadTickState(statePath, 'heartbeat')).lastEmittedSlot).toBe(7); + }); +}); + +describe('CLI argument parsing refuses coercion rather than accepting it', () => { + // parseInt('1.5') is 1. A fractional --interval-ms silently becoming a 1ms + // schedule is strictly worse than a refusal: the operator wrote something + // the tool did not do, and nothing says so. These rows pin the refusal. + const { spawnSync } = require('node:child_process') as typeof import('node:child_process'); + const cliPath = join(__dirname, '..', 'dist', 'cli.js'); + + it.each([ + ['fractional', '1.5'], + ['exponent notation', '1e3'], + ['hex', '0x10'], + ['trailing text', '60000ms'], + ['empty', ''], + ])('refuses --interval-ms %s as an invocation error', (_label, value) => { + const result = spawnSync(process.execPath, [ + cliPath, 'tick', 'start', '--schedule-id', 'x', '--interval-ms', value, 'spec.json', + ], { encoding: 'utf8' }); + expect(result.stderr + result.stdout).toContain('invalid_invocation'); + }); + + it('accepts an exact integer and proceeds past parsing', () => { + const result = spawnSync(process.execPath, [ + cliPath, 'tick', 'start', '--schedule-id', 'x', '--interval-ms', '60000', + join(tmpdir(), 'definitely-absent-spec.json'), + ], { encoding: 'utf8' }); + // Past the parser: it fails on the missing spec, not on the invocation. + const output = result.stderr + result.stdout; + expect(output).not.toContain('invalid_invocation'); + expect(output).toContain('invalid_state'); + }); +}); From 756839a6e36e0a20656fb60b4d21c76869b19b76 Mon Sep 17 00:00:00 2001 From: kjgbot Date: Fri, 4 Sep 2026 01:28:17 +0200 Subject: [PATCH 3/4] fix(sdk): make the tick runner's client injection async, and connect before hello The injection surface was synchronous, which hid a missing step. A real JournalClient needs connect() before hello(); the unit tests' fake client has no transport, so it "connected" vacuously and every test passed. The gap only appeared against a live daemon, as: journal client: not connected (hello) Making the seam async and giving the default path ownership of connect+hello means no caller can supply one half of the handshake. A connect failure now reports TICK_RUNNER_FAILED with the error class and message and exits 1, rather than surfacing as an unhandled rejection from inside the poll loop. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01FtQSAcGDta5VH9xiZFT4sR Session-Id: c228933d-4f94-4d83-9a9a-daf3c83b94f1 Session-Id: c228933d-4f94-4d83-9a9a-daf3c83b94f1 Session-Id: c228933d-4f94-4d83-9a9a-daf3c83b94f1 Session-Id: c228933d-4f94-4d83-9a9a-daf3c83b94f1 --- sdk/src/cli/tick-runner.ts | 31 ++++++++++++++++++++++++++----- sdk/tests/tick-runner.test.ts | 22 +++++++++++----------- 2 files changed, 37 insertions(+), 16 deletions(-) diff --git a/sdk/src/cli/tick-runner.ts b/sdk/src/cli/tick-runner.ts index 98146dd97..d56ead5bf 100644 --- a/sdk/src/cli/tick-runner.ts +++ b/sdk/src/cli/tick-runner.ts @@ -120,7 +120,23 @@ interface TickRunnerArgsBase { * to whatever worker the flow's trigger names. */ export interface TickRunnerArgs extends TickRunnerArgsBase { - connectClient?: (socketPath: string) => TickRunnerClient; + /** + * Async, because a real `JournalClient` needs `connect()` before `hello()`. + * A synchronous injection surface hid that: the unit tests' fake client has + * no transport, so it connected vacuously and the missing `connect()` only + * surfaced against a live daemon with `journal client: not connected + * (hello)`. The default path below owns connect+hello so no caller can + * forget half of it. + */ + connectClient?: (socketPath: string) => Promise; +} + +/** Connect and handshake. Both steps, or neither. */ +async function defaultConnectClient(socketPath: string): Promise { + const client = new JournalClient(socketPath); + await client.connect(); + await client.hello('flows-tick-runner'); + return client; } /** Absolute path to the state file for one schedule. */ @@ -243,13 +259,18 @@ export async function runTickRunner(args: TickRunnerArgs, io: CliIo): Promise { expect( await runTickRunner( { dataDir: dir, specPath, schedule: SCHEDULE, maxPolls: 1, - now: () => atSlot(3), connectClient: () => first.client }, + now: () => atSlot(3), connectClient: async () => first.client }, io1.io, ), ).toBe(0); @@ -117,7 +117,7 @@ describe('bound 1: a restart emits exactly one tick per due slot', () => { expect( await runTickRunner( { dataDir: dir, specPath, schedule: SCHEDULE, maxPolls: 1, - now: () => atSlot(6), connectClient: () => second.client }, + now: () => atSlot(6), connectClient: async () => second.client }, io2.io, ), ).toBe(0); @@ -135,7 +135,7 @@ describe('bound 1: a restart emits exactly one tick per due slot', () => { const io = makeIo(); await runTickRunner( { dataDir: dir, specPath, schedule: SCHEDULE, maxPolls: 1, - now: () => atSlot(500), connectClient: () => client }, + now: () => atSlot(500), connectClient: async () => client }, io.io, ); expect(recorded.slots).toEqual([500]); @@ -146,7 +146,7 @@ describe('bound 1: a restart emits exactly one tick per due slot', () => { const { client } = makeClient(); await runTickRunner( { dataDir: dir, specPath, schedule: SCHEDULE, maxPolls: 1, - now: () => atSlot(9), connectClient: () => client }, + now: () => atSlot(9), connectClient: async () => client }, makeIo().io, ); const state = await loadTickState(tickStatePath(dir, 'heartbeat'), 'heartbeat'); @@ -168,7 +168,7 @@ describe('bound 2: a slot passed over by maxCatchUp is reported, never lost', () expect( await runTickRunner( { dataDir: dir, specPath, schedule: SCHEDULE, maxPolls: 1, - now: () => atSlot(20), connectClient: () => client }, + now: () => atSlot(20), connectClient: async () => client }, io.io, ), ).toBe(0); @@ -199,7 +199,7 @@ describe('bound 2: a slot passed over by maxCatchUp is reported, never lost', () expect( await runTickRunner( { dataDir: dir, specPath, schedule: SCHEDULE, maxPolls: 1, - now: () => atSlot(20), connectClient: () => client }, + now: () => atSlot(20), connectClient: async () => client }, io.io, ), ).toBe(1); @@ -222,7 +222,7 @@ describe('bound 3: a submit failure leaves the unfired slot due', () => { expect( await runTickRunner( { dataDir: dir, specPath, schedule: SCHEDULE, maxPolls: 1, - now: () => atSlot(5), connectClient: () => client }, + now: () => atSlot(5), connectClient: async () => client }, io.io, ), ).toBe(1); @@ -238,7 +238,7 @@ describe('bound 3: a submit failure leaves the unfired slot due', () => { expect( await runTickRunner( { dataDir: dir, specPath, schedule: SCHEDULE, maxPolls: 1, - now: () => atSlot(5), connectClient: () => retry.client }, + now: () => atSlot(5), connectClient: async () => retry.client }, makeIo().io, ), ).toBe(0); @@ -260,7 +260,7 @@ describe('fail closed at declaration, before anything connects', () => { await runTickRunner( { dataDir: dir, specPath, schedule: schedule as TickSchedule, maxPolls: 1, now: () => atSlot(1), - connectClient: () => { connected = true; return makeClient().client; } }, + connectClient: async () => { connected = true; return makeClient().client; } }, io.io, ), ).toBe(1); @@ -279,7 +279,7 @@ describe('fail closed at declaration, before anything connects', () => { expect( await runTickRunner( { dataDir: dir, specPath, schedule: SCHEDULE, maxPolls: 1, - now: () => atSlot(1), connectClient: () => makeClient().client }, + now: () => atSlot(1), connectClient: async () => makeClient().client }, io.io, ), ).toBe(1); @@ -294,7 +294,7 @@ describe('fail closed at declaration, before anything connects', () => { expect( await runTickRunner( { dataDir: dir, specPath, schedule: SCHEDULE, maxPolls: 1, - now: () => atSlot(9), connectClient: () => makeClient().client }, + now: () => atSlot(9), connectClient: async () => makeClient().client }, io.io, ), ).toBe(1); From 77e2b9b85fecbe7df32b3583bc7d1982955674de Mon Sep 17 00:00:00 2001 From: kjgbot Date: Fri, 4 Sep 2026 02:08:20 +0200 Subject: [PATCH 4/4] fix(sdk): fail closed on a malformed tick state file `skippedSlots: Array.isArray(x) ? x : []` silently turned a malformed value into "no slots were skipped". That is the exact claim this runner exists to make trustworthy: a state file that cannot say what it missed must stop the runner, not quietly report that it missed nothing. It is also the same silent- fallback shape the runner was written to prevent in emitDueTicks. Refuse a non-array skippedSlots and a non-integer entry within it, matching how loadTickState already refuses a non-integer lastEmittedSlot and a scheduleId that belongs to another schedule. Found by probing my own state loader with hostile-but-valid JSON. The same probe cleared a suspected unbounded-backfill bug: a negative lastEmittedSlot is already caught by the catch-up bound and fails loudly rather than emitting a million ticks. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01FtQSAcGDta5VH9xiZFT4sR Session-Id: c228933d-4f94-4d83-9a9a-daf3c83b94f1 Session-Id: c228933d-4f94-4d83-9a9a-daf3c83b94f1 Session-Id: c228933d-4f94-4d83-9a9a-daf3c83b94f1 --- sdk/src/cli/tick-runner.ts | 13 ++++++++++++- sdk/tests/tick-runner.test.ts | 16 ++++++++++++++++ 2 files changed, 28 insertions(+), 1 deletion(-) diff --git a/sdk/src/cli/tick-runner.ts b/sdk/src/cli/tick-runner.ts index d56ead5bf..9063c616d 100644 --- a/sdk/src/cli/tick-runner.ts +++ b/sdk/src/cli/tick-runner.ts @@ -186,11 +186,22 @@ export async function loadTickState( if (state.lastEmittedSlot !== undefined && !Number.isInteger(state.lastEmittedSlot)) { throw new Error(`tick state at ${path} has a non-integer lastEmittedSlot`); } + // Fail closed rather than coerce. `Array.isArray(...) ? ... : []` silently + // turned a malformed value into "no slots were skipped" — which is the exact + // claim this runner exists to make trustworthy. A state file that cannot say + // what it missed must stop the runner, not quietly report that it missed + // nothing. + if (state.skippedSlots !== undefined && !Array.isArray(state.skippedSlots)) { + throw new Error(`tick state at ${path} has a non-array skippedSlots`); + } + if (state.skippedSlots?.some((slot) => !Number.isInteger(slot))) { + throw new Error(`tick state at ${path} has a non-integer entry in skippedSlots`); + } return { scheduleId, ...(state.lastEmittedSlot === undefined ? {} : { lastEmittedSlot: state.lastEmittedSlot }), ...(state.lastEmittedAtMs === undefined ? {} : { lastEmittedAtMs: state.lastEmittedAtMs }), - skippedSlots: Array.isArray(state.skippedSlots) ? state.skippedSlots : [], + skippedSlots: state.skippedSlots ?? [], }; } diff --git a/sdk/tests/tick-runner.test.ts b/sdk/tests/tick-runner.test.ts index 742ab97d7..3b8b652ae 100644 --- a/sdk/tests/tick-runner.test.ts +++ b/sdk/tests/tick-runner.test.ts @@ -94,6 +94,22 @@ afterEach(async () => { await rm(dir, { recursive: true, force: true }); }); +describe('a malformed state file stops the runner rather than claiming nothing was skipped', () => { + it('refuses a non-array skippedSlots instead of coercing it to []', async () => { + const dir = await mkdtemp(join(tmpdir(), 'tick-state-')); + const path = join(dir, 's.json'); + await writeFile(path, JSON.stringify({ scheduleId: 's1', skippedSlots: 'nope' })); + await expect(loadTickState(path, 's1')).rejects.toThrow(/non-array skippedSlots/); + }); + + it('refuses a non-integer entry inside skippedSlots', async () => { + const dir = await mkdtemp(join(tmpdir(), 'tick-state-')); + const path = join(dir, 's.json'); + await writeFile(path, JSON.stringify({ scheduleId: 's1', skippedSlots: [1, 'two', 3] })); + await expect(loadTickState(path, 's1')).rejects.toThrow(/non-integer entry in skippedSlots/); + }); +}); + describe('bound 1: a restart emits exactly one tick per due slot', () => { it('resumes from the persisted cursor instead of jumping to the current slot', async () => { // First runner: one poll at slot 3. Fresh cursor starts AT the current