Repository Public
1/**2 * Durable waits, owner side: resolveWait / expireWaits.3 * Both lock the RUN row first, then the wait row — the same order as the worker's suspend4 * transaction — so "resolution arrives while the run is still running" cannot be lost: either the5 * worker suspends first (we then re-queue the run) or we resolve first (the worker's suspend6 * transaction sees the resolution and continues without suspending).7 */8import type { RunStatus } from '@aos/shared';9import { canonicalJson } from './canonical.js';10import { getKernel, query, withTx, type Kernel, type Tx } from './db.js';11import { WaitNotFoundError } from './errors.js';12import { announceRunStatus } from './runs.js';1314export interface WaitRow {15 id: string;16 run_id: string;17 step_seq: number;18 mission_id: string;19 kind: 'approval' | 'question' | 'timer' | 'child';20 payload: Record<string, unknown>;21 status: 'pending' | 'resolved' | 'expired' | 'cancelled';22 resolution: Record<string, unknown> | null;23 resolve_by: Date | null;24}2526export type WaitActor = string | { id?: string; name?: string; via?: string };27export interface ResolveResult {28 /** false ⇒ the wait was no longer pending (already resolved / expired / cancelled); nothing changed. */29 resolved: boolean;30 status: WaitRow['status'];31 runId: string;32}3334async function settle(35 tx: Tx,36 waitId: string,37 to: 'resolved' | 'expired',38 resolution: Record<string, unknown> | null,39 actor: WaitActor | null,40 onlyIfDue: boolean,41): Promise<ResolveResult | null> {42 const ref = await tx.one<{ run_id: string }>('SELECT run_id FROM waits WHERE id = $1', [waitId]);43 if (!ref) return null;44 const run = await tx.one<{ id: string; mission_id: string; parent_run_id: string | null; status: RunStatus }>(45 'SELECT id, mission_id, parent_run_id, status FROM runs WHERE id = $1 FOR UPDATE',46 [ref.run_id],47 );48 const w = await tx.one<WaitRow & { due: boolean }>(49 `SELECT *, (resolve_by IS NOT NULL AND resolve_by <= now()) AS due FROM waits WHERE id = $1 FOR UPDATE`,50 [waitId],51 );52 if (!run || !w) return null;53 // Timers are settled by the run itself when due; they cannot be resolved from outside.54 if (w.status !== 'pending' || w.kind === 'timer' || (onlyIfDue && !w.due)) return { resolved: false, status: w.status, runId: run.id };55 await tx.q(`UPDATE waits SET status = $2, resolution = $3::jsonb, resolved_at = now() WHERE id = $1`, [56 waitId,57 to,58 resolution === null ? null : canonicalJson(resolution),59 ]);60 tx.emit({61 missionId: run.mission_id,62 runId: run.id,63 seq: w.step_seq,64 kind: 'wait.resolved',65 trust: to === 'resolved' ? 'owner' : 'system',66 payload: { waitId, kind: w.kind, status: to, resolution, actor },67 });68 if (run.status === 'waiting') {69 await tx.q(`UPDATE runs SET status = 'queued', wake_at = NULL WHERE id = $1`, [run.id]);70 await announceRunStatus(tx, run, 'queued', { reason: `wait ${to}` });71 tx.wake('q');72 }73 return { resolved: true, status: to, runId: run.id };74}7576/**77 * Marks a pending approval/question wait resolved, emits wait.resolved and re-queues the run if it is78 * suspended. `resolution` is what ctx.wait() returns (approval: {decision, remember?, note?};79 * question: {answer}). Idempotent: a wait that is not pending is left untouched (`resolved:false`).80 */81export async function resolveWait(82 waitId: string,83 resolution: Record<string, unknown>,84 actor: WaitActor,85 k: Kernel = getKernel(),86): Promise<ResolveResult> {87 const r = await withTx(k, (tx) => settle(tx, waitId, 'resolved', resolution, actor, false));88 if (!r) throw new WaitNotFoundError(`wait ${waitId} not found`);89 return r;90}9192/** Janitor: pending approval/question waits past `resolve_by` become `expired` and their runs are re-queued. */93export async function expireWaits(k: Kernel = getKernel()): Promise<{ expired: number }> {94 const due = await query<{ id: string }>(95 k,96 `SELECT id FROM waits WHERE status = 'pending' AND kind IN ('approval','question') AND resolve_by IS NOT NULL AND resolve_by <= now()97 ORDER BY resolve_by LIMIT 200`,98 );99 let expired = 0;100 for (const w of due) {101 const r = await withTx(k, (tx) => settle(tx, w.id, 'expired', null, null, true));102 if (r?.resolved) expired++;103 }104 return { expired };105}106