Repository Public
1/**2 * Suspension paths of a run context: durable waits (approval / question), timers, and the owner3 * question raised for a non-idempotent step that may or may not have run before a crash.4 *5 * All three decide "suspend or continue" inside ONE fenced transaction that holds the run row lock6 * and then the wait row lock (the order resolveWait / expireWaits / requestCancel use), with the7 * context gate held exclusively: no step of this run is executing while its lease is released.8 */9import { canonicalJson, idemKeyFor, sha256Hex } from './canonical.js';10import type { Tx } from './db.js';11import { KernelError, ReconcileAborted, RunCancelled, Suspend, type RecordedError } from './errors.js';12import type { RunCtx } from './context.js';13import { completeStep, failStep, insertIntent, suspendRun, type Completed, type StepMeta, type StepRow } from './journal.js';14import type { RunRow } from './runs.js';15import { RECONCILE_OPTIONS, type StepSpec } from './types.js';16import type { WaitRow } from './waits.js';1718export type WaitSpec =19 | { mode: 'wait'; kind: 'approval' | 'question'; payload: Record<string, unknown>; timeoutSec: number | null }20 | { mode: 'timer'; ms: number };2122type Decision = { t: 'done'; c: Completed } | { t: 'suspended' } | { t: 'cancelled' } | { t: 'inline'; ms: number };2324function delay(ms: number, signal: AbortSignal): Promise<void> {25 return new Promise((resolve) => {26 if (signal.aborted) return resolve();27 const t = setTimeout(done, ms);28 function done(): void {29 clearTimeout(t);30 signal.removeEventListener('abort', done);31 resolve();32 }33 signal.addEventListener('abort', done, { once: true });34 });35}3637async function insertWait(tx: Tx, run: RunRow, seq: number, kind: WaitRow['kind'], payload: Record<string, unknown>, timeoutMs: number | null): Promise<void> {38 const w = await tx.one<{ id: string; resolve_by: Date | null }>(39 `INSERT INTO waits (run_id, step_seq, mission_id, kind, payload, resolve_by)40 VALUES ($1,$2,$3,$4,$5::jsonb, CASE WHEN $6::float8 IS NULL THEN NULL ELSE now() + $6::float8 * interval '1 millisecond' END)41 RETURNING id, resolve_by`,42 [run.id, seq, run.mission_id, kind, canonicalJson(payload), timeoutMs],43 );44 const row = w as { id: string; resolve_by: Date | null };45 tx.emit({ missionId: run.mission_id, runId: run.id, seq, kind: 'wait.created', payload: { waitId: row.id, kind, payload, resolveBy: row.resolve_by?.toISOString() ?? null } });46}4748/** ctx.wait / ctx.sleep. The wait is a journaled step (kind 'wait' | 'timer') plus a `waits` row with the same seq. */49export async function durableWait(ctx: RunCtx, seq: number, spec: WaitSpec): Promise<unknown> {50 const request = spec.mode === 'timer' ? { ms: spec.ms } : { kind: spec.kind, payload: spec.payload, timeoutSec: spec.timeoutSec };51 const meta: StepMeta = { kind: spec.mode, name: spec.mode === 'timer' ? 'sleep' : spec.kind, effect: 'pure' };52 const requestJson = canonicalJson(request);53 // The determinism check covers what is asked, not for how long: a timeout that comes from settings54 // edited between two replays must not fail the run (it only matters on first execution).55 const hash = sha256Hex(spec.mode === 'timer' ? requestJson : canonicalJson({ kind: spec.kind, payload: spec.payload }));56 let row = ctx.journal.get(seq);57 if (row) {58 ctx.checkSame(row, meta, hash);59 if (row.status === 'completed') {60 ctx.account(row);61 return JSON.parse(canonicalJson(row.result)) as unknown;62 }63 if (row.status === 'failed') throw ctx.rebuildError(row);64 } else {65 // First execution: intent + waits row + wait.created in one transaction. A crash right after it66 // leaves the run 'running' with a pending wait; the next claim lands in the loop below.67 ctx.assertOpen();68 await ctx.gate.shared();69 try {70 ctx.assertOpen();71 const idemKey = idemKeyFor(ctx.runId, seq);72 await ctx.fenced(async (tx, run) => {73 await insertIntent(tx, run, seq, meta, hash, requestJson, idemKey);74 const timeoutMs = spec.mode === 'timer' ? spec.ms : spec.timeoutSec === null ? null : spec.timeoutSec * 1000;75 await insertWait(tx, run, seq, spec.mode === 'timer' ? 'timer' : spec.kind, spec.mode === 'timer' ? { ms: spec.ms } : spec.payload, timeoutMs);76 }, true);77 row = { seq, ...meta, status: 'started', request_hash: hash, result: null, error: null, usage: null, idem_key: idemKey, completed_at: null };78 ctx.journal.set(seq, row);79 } finally {80 ctx.gate.releaseShared();81 }82 }83 const stepRow: StepRow = row;84 for (;;) {85 await ctx.gate.exclusive();86 let d: Decision;87 try {88 ctx.assertOpen();89 d = await ctx.fenced((tx, run) => decideWait(tx, run, seq, meta, spec.mode, ctx.inlineSleepMs));90 if (d.t === 'suspended') {91 ctx.markSuspended();92 throw new Suspend();93 }94 if (d.t === 'cancelled') {95 ctx.abort(new RunCancelled());96 throw new RunCancelled();97 }98 if (d.t === 'done') return ctx.applyCompleted(stepRow, d.c);99 } finally {100 ctx.gate.releaseExclusive();101 }102 await delay(d.ms, ctx.signal); // short timer: wait in-process (still under the lease), then re-check103 }104}105106async function decideWait(tx: Tx, run: RunRow, seq: number, meta: StepMeta, mode: 'wait' | 'timer', inlineSleepMs: number): Promise<Decision> {107 const w = await tx.one<WaitRow & { due: boolean; remaining_ms: unknown }>(108 `SELECT *, (resolve_by IS NOT NULL AND resolve_by <= now()) AS due,109 GREATEST(0, EXTRACT(EPOCH FROM (resolve_by - now())) * 1000) AS remaining_ms110 FROM waits WHERE run_id = $1 AND step_seq = $2 FOR UPDATE`,111 [run.id, seq],112 );113 if (!w) throw new KernelError(`wait row of step ${seq} is missing`);114 if (run.cancel_requested || w.status === 'cancelled') return { t: 'cancelled' };115 const settle = async (status: 'resolved' | 'expired', result: unknown): Promise<Decision> => {116 if (w.status === 'pending') {117 await tx.q(`UPDATE waits SET status = $2, resolved_at = now() WHERE id = $1`, [w.id, status]);118 tx.emit({ missionId: run.mission_id, runId: run.id, seq, kind: 'wait.resolved', payload: { waitId: w.id, kind: w.kind, status } });119 }120 return { t: 'done', c: await completeStep(tx, run, seq, meta, canonicalJson(result), null) };121 };122 if (mode === 'timer') {123 if (w.due || w.status !== 'pending') return settle('resolved', { wokeAt: (w.resolve_by ?? new Date()).toISOString() });124 const remaining = Math.ceil(Number(w.remaining_ms));125 if (remaining <= inlineSleepMs) return { t: 'inline', ms: Math.max(1, remaining) };126 await suspendRun(tx, run, 'queued', w.resolve_by);127 return { t: 'suspended' };128 }129 if (w.status === 'resolved') return settle('resolved', w.resolution ?? {});130 if (w.status === 'expired' || w.due) return settle('expired', { expired: true });131 await suspendRun(tx, run, 'waiting', null);132 return { t: 'suspended' };133}134135// ------------------------------------------------------------------------------------------------136137export function classifyAnswer(answer: unknown): 'ran' | 'run' | 'abort' | null {138 const a = String(answer ?? '').trim().toLowerCase();139 if (a.startsWith('it did not') || a.startsWith("it didn't") || a === 'run it') return 'run';140 if (a.startsWith('it ran') || a === 'continue') return 'ran';141 if (a.startsWith('abort')) return 'abort';142 return null;143}144145type Asked = { t: 'value'; c: Completed } | { t: 'run' } | { t: 'abort'; at: Date; error: RecordedError } | { t: 'suspended' } | { t: 'cancelled' };146147/**148 * A non_idempotent step was found 'started' and no probe could tell whether its effect happened.149 * The step becomes 'needs_reconcile' and a question wait (same seq) is put to the owner:150 * "it ran — continue" → the step is completed with spec.assumedResult (or a generic marker);151 * "it did not run — run it" → the step goes back to 'started' and is executed (a new crash asks again);152 * "abort" → the step is failed and the run ends 'cancelled'.153 */154export async function askReconcile<Req, Res>(155 ctx: RunCtx, seq: number, spec: StepSpec<Req, Res>, meta: StepMeta, row: StepRow,156): Promise<{ t: 'value'; v: unknown } | { t: 'run' }> {157 await ctx.gate.exclusive();158 try {159 ctx.assertOpen();160 const r = await ctx.fenced(async (tx, run): Promise<Asked> => {161 if (run.cancel_requested) return { t: 'cancelled' };162 const st = await tx.one<{ status: StepRow['status'] }>('SELECT status FROM steps WHERE run_id = $1 AND seq = $2 FOR UPDATE', [run.id, seq]);163 const w = await tx.one<WaitRow>('SELECT * FROM waits WHERE run_id = $1 AND step_seq = $2 FOR UPDATE', [run.id, seq]);164 const ask = async (note?: string): Promise<Asked> => {165 if (w) await tx.q('DELETE FROM waits WHERE id = $1', [w.id]);166 if (st?.status === 'started') {167 await tx.q(`UPDATE steps SET status = 'needs_reconcile' WHERE run_id = $1 AND seq = $2`, [run.id, seq]);168 tx.emit({ missionId: run.mission_id, runId: run.id, seq, kind: 'step.needs_reconcile', payload: { kind: meta.kind, name: meta.name, effect: meta.effect } });169 }170 const text =171 `Step ${seq} (${meta.name}) may or may not have run: the worker stopped after starting it and before recording its result. ` +172 `It is not safe to retry automatically. Check its effect, then choose.${note ? ` (${note})` : ''}`;173 await insertWait(tx, run, seq, 'question', { text, options: [...RECONCILE_OPTIONS], reconcile: true, stepSeq: seq, stepKind: meta.kind, stepName: meta.name }, null);174 await suspendRun(tx, run, 'waiting', null);175 return { t: 'suspended' };176 };177 if (!st) throw new KernelError(`step ${seq} is missing`);178 if (st.status === 'started' || !w) return ask();179 if (st.status !== 'needs_reconcile') throw new KernelError(`step ${seq} is ${st.status}, expected needs_reconcile`);180 if (w.status === 'cancelled') return { t: 'cancelled' };181 if (w.status === 'pending') {182 await suspendRun(tx, run, 'waiting', null);183 return { t: 'suspended' };184 }185 const answer = w.status === 'resolved' ? classifyAnswer(w.resolution?.answer) : null;186 if (answer === 'ran') {187 const assumed = spec.assumedResult188 ? spec.assumedResult(spec.request, row.idem_key)189 : { ok: true, output: 'The owner confirmed that this step had already run before a crash; its original output was lost.', reconciled: 'assumed_done' };190 return { t: 'value', c: await completeStep(tx, run, seq, meta, canonicalJson(assumed), null) };191 }192 if (answer === 'run') {193 await tx.q(`UPDATE steps SET status = 'started' WHERE run_id = $1 AND seq = $2`, [run.id, seq]);194 return { t: 'run' };195 }196 if (answer === 'abort') {197 const error: RecordedError = { name: 'ReconcileAborted', message: `Aborted by the owner while reconciling step ${seq} (${meta.name})` };198 return { t: 'abort', at: await failStep(tx, run, seq, meta, error), error };199 }200 return ask(w.status === 'expired' ? 'the previous question expired' : 'the previous answer was not one of the options');201 });202 switch (r.t) {203 case 'suspended':204 row.status = 'needs_reconcile';205 ctx.markSuspended();206 throw new Suspend();207 case 'cancelled':208 ctx.abort(new RunCancelled());209 throw new RunCancelled();210 case 'abort':211 row.status = 'failed';212 row.error = r.error;213 row.completed_at = r.at;214 throw new ReconcileAborted(seq, meta.name);215 case 'run':216 row.status = 'started';217 return { t: 'run' };218 case 'value':219 return { t: 'value', v: ctx.applyCompleted(row, r.c) };220 }221 } finally {222 ctx.gate.releaseExclusive();223 }224}225