diff --git a/app/api/cron/ai-runs/route.ts b/app/api/cron/ai-runs/route.ts index f77b00e07..921bd1372 100644 --- a/app/api/cron/ai-runs/route.ts +++ b/app/api/cron/ai-runs/route.ts @@ -61,12 +61,16 @@ export async function GET(req: NextRequest) { const remaining = deadline - Date.now(); if (remaining < MIN_SLICE_MS) { // Out of tick. Put it back exactly as it was found so the next tick - // (or an approval decision) picks it up without waiting out the lease. + // (or an approval decision) picks it up without waiting out the lease + // — including the attempt the claim charged. A run that was not given + // a slice has not been tried, and `claim_ai_runs` abandons a run at + // `max_attempts`, so an unexecuted claim must not spend the budget. const { data: requeued, error } = await db .from('family_automation_runs') .update({ state: 'ready', status: legacyStatusFor('ready'), + attempt: Math.max(0, run.attempt - 1), lease_owner: null, lease_expires_at: null, run_after: new Date().toISOString(), diff --git a/docs/audit/a-run-with-no-attempts-left-is-abandoned-check.sql b/docs/audit/a-run-with-no-attempts-left-is-abandoned-check.sql new file mode 100644 index 000000000..ee45c13f7 --- /dev/null +++ b/docs/audit/a-run-with-no-attempts-left-is-abandoned-check.sql @@ -0,0 +1,191 @@ +-- Behavioural proof that `claim_ai_runs` honours the attempt ceiling in its +-- CLAIMING half, not only when a lease expires (finalaudit Q40, second half). +-- +-- `attempt` is the run's abandonment budget: the recovery arm dead-letters an +-- `executing` run whose lease ran out once `attempt >= max_attempts`, and +-- `claimRun` (lib/ai/runs/store.ts) refuses to lease a run at that ceiling. +-- Until the migration this probe accompanies, the candidates query had no such +-- test: a run parked back in the queue with no attempts left was claimed by +-- every tick for ever — attempt 6, 7, 8 … — while every human-initiated path +-- refused it. The dead-letter arm never reached it, because it only ran for an +-- expired lease on an `executing` row, and a parked run holds no lease. +-- +-- Five runs and three pending approvals, one call, one family. Wrapped in a +-- transaction that is rolled back, so the claim it performs leaves nothing +-- behind. +-- +-- THE CONTROL is a run one attempt below the ceiling, under the same predicate +-- otherwise: it MUST be claimed, and its step, request and timeline MUST be +-- untouched. Without it a `claim_ai_runs` that abandons every parked run, or +-- that writes the reconcile for every run it touches, would pass the first +-- block of assertions and be credited with honouring a ceiling it never read. +begin; +do $$ +declare + fam uuid; + -- Abandoned: parked `ready`, no attempts left, with its own request, plan and step. + spent uuid; spent_req uuid; spent_pl uuid; spent_step uuid; + -- CONTROL: the same, one attempt below the ceiling. + below uuid; below_req uuid; below_pl uuid; below_step uuid; + -- Also abandoned: `scheduled_followup` at the ceiling, not yet due. It will + -- never run, so the family is told now rather than on the day. + later uuid; + -- Left alone: at the ceiling, but a cancel is in flight (must end `cancelled`, + -- not `failed`) … + halting uuid; + -- … or a live lease is still held (not the queue's to touch). + held uuid; + -- Approvals in flight: two on the abandoned run (one found by run_id, one + -- only through the step that carries it), one on the CONTROL. + spent_appr uuid; spent_step_appr uuid; below_appr uuid; + claimed uuid[]; + r public.family_automation_runs%rowtype; + st text; + n int; + setup text[] := '{}'; +begin + delete from public.families where name = 'Attempt-ceiling probe'; + insert into public.families (name) values ('Attempt-ceiling probe') returning id into fam; + + insert into public.ai_requests (family_id, kind, request_text, status) + values (fam, 'concierge', 'Sort out the week', 'executing') returning id into spent_req; + insert into public.ai_plans (family_id, request_id, objective) + values (fam, spent_req, 'Week plan') returning id into spent_pl; + insert into public.family_automation_runs + (family_id, plan_id, request_id, state, status, run_type, attempt, max_attempts, run_after, lease_owner, lease_expires_at) + values (fam, spent_pl, spent_req, 'ready', 'approved', 'concierge', 5, 5, now() - interval '1 minute', null, null) + returning id into spent; + insert into public.ai_plan_steps (family_id, plan_id, sequence, step_type, tool_name, description, status) + values (fam, spent_pl, 1, 'act', 'calendar.createEvent', 'Add the dentist', 'ready') returning id into spent_step; + + insert into public.ai_requests (family_id, kind, request_text, status) + values (fam, 'concierge', 'Plan dinner', 'executing') returning id into below_req; + insert into public.ai_plans (family_id, request_id, objective) + values (fam, below_req, 'Dinner plan') returning id into below_pl; + insert into public.family_automation_runs + (family_id, plan_id, request_id, state, status, run_type, attempt, max_attempts, run_after, lease_owner, lease_expires_at) + values (fam, below_pl, below_req, 'ready', 'approved', 'concierge', 4, 5, now() - interval '1 minute', null, null) + returning id into below; + insert into public.ai_plan_steps (family_id, plan_id, sequence, step_type, tool_name, description, status) + values (fam, below_pl, 1, 'act', 'meals.plan', 'Pick a recipe', 'ready') returning id into below_step; + + insert into public.family_automation_runs + (family_id, state, status, run_type, attempt, max_attempts, run_after, lease_owner, lease_expires_at) + values (fam, 'scheduled_followup', 'approved', 'concierge', 5, 5, now() + interval '3 days', null, null) + returning id into later; + insert into public.family_automation_runs + (family_id, state, status, run_type, attempt, max_attempts, run_after, lease_owner, lease_expires_at, cancel_requested_at) + values (fam, 'ready', 'approved', 'concierge', 5, 5, now() - interval '1 minute', null, null, now() - interval '1 second') + returning id into halting; + insert into public.family_automation_runs + (family_id, state, status, run_type, attempt, max_attempts, run_after, lease_owner, lease_expires_at) + values (fam, 'ready', 'approved', 'concierge', 5, 5, now() - interval '1 minute', gen_random_uuid(), now() + interval '2 minutes') + returning id into held; + + -- Approvals the runs opened and nobody has decided. The abandoned run has + -- two: one that names the run (0251's run_id), one that only the step knows + -- about (run_id null, the step's approval_id points at it). The CONTROL has + -- one, and it must survive the call untouched. + insert into public.approval_requests (family_id, domain, capability, title, status, run_id, plan_step_id) + values (fam, 'calendar', 'automate', 'Add the dentist?', 'pending', spent, spent_step) returning id into spent_appr; + insert into public.approval_requests (family_id, domain, capability, title, status, run_id, plan_step_id) + values (fam, 'calendar', 'automate', 'Move the follow-up?', 'pending', null, spent_step) returning id into spent_step_appr; + update public.ai_plan_steps set approval_id = spent_step_appr where id = spent_step; + insert into public.approval_requests (family_id, domain, capability, title, status, run_id, plan_step_id) + values (fam, 'meals', 'automate', 'Order the groceries?', 'pending', below, below_step) returning id into below_appr; + + -- ── The setup really is the setup ──────────────────────────────────────── + -- A fixture that arrived already failed, or a control that was never due, + -- makes the assertions below vacuous; report that as the probe's own fault. + select * into r from public.family_automation_runs where id = spent; + if r.state is distinct from 'ready' or r.attempt < r.max_attempts or r.lease_expires_at is not null or r.cancel_requested_at is not null or r.run_after > now() then + setup := array_append(setup, format('the abandoned fixture was not a due, unleased, uncancelled ready run at the ceiling (state=%s attempt=%s/%s)', r.state, r.attempt, r.max_attempts)); + end if; + select * into r from public.family_automation_runs where id = below; + if r.state is distinct from 'ready' or r.attempt >= r.max_attempts or r.lease_expires_at is not null or r.run_after > now() then + setup := array_append(setup, format('the CONTROL was not a due, unleased ready run below the ceiling (state=%s attempt=%s/%s)', r.state, r.attempt, r.max_attempts)); + end if; + select status into st from public.ai_plan_steps where id = spent_step; + if st is distinct from 'ready' then setup := array_append(setup, format('the abandoned run''s step was %s, not ''ready'', before the call', st)); end if; + select status into st from public.ai_requests where id = spent_req; + if st is distinct from 'executing' then setup := array_append(setup, format('the abandoned run''s request was already %s before the call', st)); end if; + select count(*) into n from public.ai_run_events where run_id in (spent, below, later, halting, held); + if n <> 0 then setup := array_append(setup, format('%s timeline event(s) existed before the call', n)); end if; + select count(*) into n from public.approval_requests where id in (spent_appr, spent_step_appr, below_appr) and status = 'pending'; + if n <> 3 then setup := array_append(setup, format('only %s of the three approvals were pending before the call', n)); end if; + if array_length(setup, 1) is not null then + raise exception E'CONTROL FAILED: the attempt-ceiling probe proved nothing, because its own fixture was wrong:\n - %', array_to_string(setup, E'\n - '); + end if; + + select coalesce(array_agg(c.id), '{}') into claimed from public.claim_ai_runs(50, 120) as c(id); + + -- ── 1. The run with no attempts left is abandoned, not claimed ─────────── + if spent = any(claimed) then + raise exception 'a ready run with no attempts left was claimed again — the claiming half of claim_ai_runs has no attempt ceiling'; + end if; + select * into r from public.family_automation_runs where id = spent; + if r.state is distinct from 'failed' then raise exception 'abandoned run left in state %', r.state; end if; + if r.status is distinct from 'failed' then raise exception 'legacy status left as %', r.status; end if; + if r.error is distinct from 'Run abandoned after the maximum number of attempts.' then raise exception 'abandoned run''s error is %', r.error; end if; + if r.completed_at is null then raise exception 'abandoned run has no completed_at'; end if; + if r.lease_owner is not null or r.lease_expires_at is not null then raise exception 'abandoned run still carries a lease'; end if; + if r.attempt <> 5 then raise exception 'abandoning a run spent an attempt (attempt=%)', r.attempt; end if; + + -- And the rest of the record follows, as 0263 made it follow a dead worker. + select status into st from public.ai_plan_steps where id = spent_step; + if st is distinct from 'failed' then raise exception 'in-flight step left as %', st; end if; + select status into st from public.ai_requests where id = spent_req; + if st is distinct from 'failed' then raise exception 'request ledger left as %', st; end if; + select count(*) into n from public.ai_run_events where run_id = spent and event_type = 'run_failed'; + if n <> 1 then raise exception 'expected one run_failed event, found %', n; end if; + + -- Its approvals are closed, by the run and by the step, with no decider. + select status into st from public.approval_requests where id = spent_appr; + if st is distinct from 'cancelled' then raise exception 'the abandoned run''s approval (by run_id) was left %: it stays in "Needs your decision" for a run that is over', st; end if; + select status into st from public.approval_requests where id = spent_step_appr; + if st is distinct from 'cancelled' then raise exception 'the abandoned run''s approval (reachable only through the step) was left %', st; end if; + select count(*) into n from public.approval_requests where id in (spent_appr, spent_step_appr) and decided_at is not null and decided_by is null; + if n <> 2 then raise exception 'a closed approval is missing decided_at, or names a decider when nobody decided (%/2)', n; end if; + + -- ── 2. The scheduled follow-up at the ceiling is abandoned too, not on the day + select * into r from public.family_automation_runs where id = later; + if later = any(claimed) then raise exception 'a scheduled follow-up that is not due was claimed'; end if; + if r.state is distinct from 'failed' or r.status is distinct from 'failed' or r.completed_at is null then + raise exception 'a scheduled follow-up with no attempts left was left as % — it would have said "Scheduled" until the day, then been abandoned', r.state; + end if; + select count(*) into n from public.ai_run_events where run_id = later and event_type = 'run_failed'; + if n <> 1 then raise exception 'the abandoned follow-up has % run_failed events, not one', n; end if; + + -- ── CONTROL: one attempt below the ceiling, the same call ──────────────── + if not (below = any(claimed)) then + raise exception 'CONTROL FAILED: a due ready run on attempt 4 of 5 was not claimed — the pass is refusing more than the ceiling, so nothing above is attributable to it'; + end if; + select * into r from public.family_automation_runs where id = below; + if r.state <> 'executing' or r.attempt <> 5 or r.lease_owner is null or r.lease_expires_at is null or r.lease_expires_at <= now() then + raise exception 'CONTROL FAILED: the claimed run is not held by a worker (state %, attempt %, lease until %)', r.state, r.attempt, r.lease_expires_at; + end if; + if r.error is not null or r.completed_at is not null or r.status is distinct from 'approved' then + raise exception 'CONTROL FAILED: the claimed run was also dead-lettered (status %, error %, completed_at %) — the reconcile is not bounded to runs at the ceiling', r.status, r.error, r.completed_at; + end if; + select status into st from public.ai_plan_steps where id = below_step; + if st is distinct from 'ready' then raise exception 'CONTROL FAILED: the claimed run''s step was marked % — the step write is a blanket rule, not the ceiling''s', st; end if; + select status into st from public.ai_requests where id = below_req; + if st is distinct from 'executing' then raise exception 'CONTROL FAILED: the claimed run''s request was resolved to %', st; end if; + select count(*) into n from public.ai_run_events where run_id = below; + if n <> 0 then raise exception 'CONTROL FAILED: the claimed run got % timeline event(s) telling the family Bubaly gave up', n; end if; + select status into st from public.approval_requests where id = below_appr; + if st is distinct from 'pending' then raise exception 'CONTROL FAILED: the claimed run''s approval was %: the approval write is a blanket rule, not bounded to the dead runs', st; end if; + + -- ── 3. A cancel in flight, or a live lease, is not the queue's to touch ── + select * into r from public.family_automation_runs where id = halting; + if halting = any(claimed) or r.state is distinct from 'ready' or r.error is not null or r.completed_at is not null then + raise exception 'a run at the ceiling with a cancel in flight was % (error %) — it must end cancelled, not failed', r.state, r.error; + end if; + select * into r from public.family_automation_runs where id = held; + if held = any(claimed) or r.state is distinct from 'ready' or r.attempt <> 5 or r.lease_owner is null or r.error is not null then + raise exception 'a run at the ceiling whose lease is still live was touched (state %, attempt %, error %)', r.state, r.attempt, r.error; + end if; + + raise notice 'OK a run parked with no attempts left is abandoned, its record says so and its approvals are closed; one attempt below, it is claimed and nothing else is touched'; +end $$; +rollback; diff --git a/docs/audit/dead-letter-reconcile-check.sql b/docs/audit/dead-letter-reconcile-check.sql index 6c9e18fe8..96c62a94e 100644 --- a/docs/audit/dead-letter-reconcile-check.sql +++ b/docs/audit/dead-letter-reconcile-check.sql @@ -3,9 +3,14 @@ -- -- WHICH MIGRATION OWNS THIS RULE. `claim_ai_runs` was created by -- `0250_ai_runtime_core.sql`, revoked from the client roles by `0253`, REWRITTEN --- by `0263_dead_letter_reconcile.sql`, and only re-GRANTED — not redefined — by --- `0292_privileged_rpc_grant_reassert.sql`. So 0263's body is what a replayed --- database runs, and its recovery half has two arms: +-- by `0263_dead_letter_reconcile.sql`, only re-GRANTED — not redefined — by +-- `0292_privileged_rpc_grant_reassert.sql`, and REWRITTEN again by +-- `0479_a_run_with_no_attempts_left_is_abandoned_not_reclaimed.sql`, which keeps +-- the expired-lease arms this probe tests word for word, adds a second arm for +-- a run parked in the queue with no attempts left, and closes a dead run's +-- pending approvals (both proved by its own probe, +-- a-run-with-no-attempts-left-is-abandoned-check.sql). So 0479's body is what a +-- replayed database runs, and the recovery half this probe exercises has two arms: -- -- state = case when attempt >= max_attempts then 'failed' else 'ready' end -- diff --git a/docs/audit/privileged-rpc-grants-check.sql b/docs/audit/privileged-rpc-grants-check.sql index ede8a9a8f..821caea2b 100644 --- a/docs/audit/privileged-rpc-grants-check.sql +++ b/docs/audit/privileged-rpc-grants-check.sql @@ -48,16 +48,18 @@ -- `claim_ai_runs`. Functions carry no trigger, CHECK or index to confuse with it. -- -- WHICH STATEMENTS WRITE THAT ACL. Grepping the five names across --- supabase/migrations returns 0204, 0218, 0237, 0250, 0253, 0263 and 0292; 0218 --- only mentions loyalty_redeem_reward in a comment. No migration after 0292 --- names any of them (checked through 0387, the newest file when this was --- written). No migration does a blanket `grant|revoke … on all functions`, sets +-- supabase/migrations returns 0204, 0218, 0237, 0250, 0253, 0263, 0292 and +-- 0479; 0218 only mentions loyalty_redeem_reward in a comment. After 0292 only +-- 0479 names any of them: it rewrites `claim_ai_runs` once more (the claiming +-- half gains the attempt ceiling) and restates 0263:114-115 byte for byte, so +-- the ACL it leaves is the one 0292 left (no other file through 0474, the newest +-- when this was written). No migration does a blanket `grant|revoke … on all functions`, sets -- `alter default privileges` for functions, or builds a function grant/revoke -- with `execute format(…)` — the only two dynamic ACL loops, 0338:154 and -- 0352:123, revoke TABLE DML from anon. So the final ACL of each function is the -- product of these statements, last writer last: -- --- claim_ai_runs 0250 (from public) · 0253 · 0263 · 0292 +-- claim_ai_runs 0250 (from public) · 0253 · 0263 · 0292 · 0479 (restates 0292) -- claim_marketing_generation_jobs 0237 (from public ONLY) · 0292 -- loyalty_award_points / _redeem_reward / _cancel_redemption 0204 · 0292 -- diff --git a/lib/ai/runs/store.ts b/lib/ai/runs/store.ts index 5380e063a..36e006e06 100644 --- a/lib/ai/runs/store.ts +++ b/lib/ai/runs/store.ts @@ -581,7 +581,8 @@ export async function updateRunWhereState( // ─── Leases ───────────────────────────────────────────────────────────────── -export type ClaimedRun = { id: string; familyId: string; leaseOwner: string | null }; +/** `attempt` is the value AFTER the claim charged it, so a caller that hands the run back unexecuted can give the charge back. */ +export type ClaimedRun = { id: string; familyId: string; leaseOwner: string | null; attempt: number }; /** * Lease a batch of due runs through the 0250 RPC. `for update skip locked` @@ -604,13 +605,13 @@ export async function claimRuns( const { data: rows, error: readError } = await db .from('family_automation_runs') - .select('id, family_id, lease_owner') + .select('id, family_id, lease_owner, attempt') .in('id', ids); if (readError) { console.error('[ai/runs] failed to read the claimed runs', readError); return fail(describeDbError(readError, 'Bubaly could not read the claimed runs.'), { code: SERVICE_CODES.db, retryable: true }); } - return ok((rows ?? []).map((r) => ({ id: r.id, familyId: r.family_id, leaseOwner: r.lease_owner }))); + return ok((rows ?? []).map((r) => ({ id: r.id, familyId: r.family_id, leaseOwner: r.lease_owner, attempt: r.attempt }))); } /** diff --git a/supabase/migrations/0479_a_run_with_no_attempts_left_is_abandoned_not_reclaimed.sql b/supabase/migrations/0479_a_run_with_no_attempts_left_is_abandoned_not_reclaimed.sql new file mode 100644 index 000000000..4ca24efca --- /dev/null +++ b/supabase/migrations/0479_a_run_with_no_attempts_left_is_abandoned_not_reclaimed.sql @@ -0,0 +1,187 @@ +-- ============================================================================ +-- 0479_a_run_with_no_attempts_left_is_abandoned_not_reclaimed.sql — the claiming +-- half of `claim_ai_runs` honours the attempt ceiling, as `claimRun` always has. +-- ---------------------------------------------------------------------------- +-- `family_automation_runs.attempt` is the run's abandonment budget, and three +-- readers agree about what it means. The recovery half of `claim_ai_runs` +-- (0250, rewritten by 0263) dead-letters an `executing` run whose lease ran out +-- once `attempt >= max_attempts` — "Run abandoned after the maximum number of +-- attempts." `claimRun` (lib/ai/runs/store.ts), which every human-initiated +-- continuation goes through, refuses to lease a run at that ceiling. And the +-- executor hands the budget back only when a slice made progress, so the +-- counter is the number of slices IN A ROW that completed nothing. +-- +-- The CLAIMING half never read it. Its candidates were every due `ready` / +-- `scheduled_followup` run, and each claim added one. So a run parked back in +-- the queue with no attempts left was claimed by every tick for ever — attempt +-- 6, 7, 8 … — while the resume button, the kick after an approval answer and a +-- step re-run all refused it, silently. The dead-letter arm never reached it, +-- because that arm only ran for an expired lease on an `executing` row, and a +-- parked run holds no lease. Unreachable by people, unkillable by the worker +-- (finalaudit Q40: "the two claim paths disagree"). +-- +-- This rewrites `claim_ai_runs` once more, in the spirit of 0263: +-- +-- 1. The recovery statement also ABANDONS a run sitting in the queue with no +-- attempts left: `ready` or `scheduled_followup`, no live lease, no cancel +-- in flight, `attempt >= max_attempts`. It takes the dead-letter arm — +-- `failed`, 0022's legacy status, the error, `completed_at` — and the same +-- reconcile 0263 added (steps, request ledger, `run_failed` event), so the +-- family is told, not shown "Scheduled" for ever. A run whose cancel is in +-- flight is left for the cancel to finish: it must end `cancelled`, not +-- `failed`. A follow-up that is not yet due is abandoned now, not on the +-- day: it would never have run. +-- 2. The candidates query gains `attempt < max_attempts`, which makes the +-- ceiling the same predicate in both claim paths. After (1) it is also +-- belt-and-braces: a run at the ceiling is `failed` by the time the +-- candidates are read. +-- 3. The reconcile gains a fourth bounded write, for BOTH dead-letter arms: +-- the run's approvals still `pending` are `cancelled`. 0263 left them +-- open, so a run that died with an approval in flight kept a card in +-- "Needs your decision" until `expires_at`, and a parent who decided it +-- approved work that would never start. +-- +-- Nothing else changes: the arm for an expired lease, the lease token's type, +-- the increment, the ordering, `for update skip locked`, the grants. +-- +-- Proof: docs/audit/a-run-with-no-attempts-left-is-abandoned-check.sql runs the +-- real function against five rows, with a control one attempt below the +-- ceiling that must be claimed and otherwise untouched. +-- +-- DEPENDS ON: 0250 (creates `claim_ai_runs` and the queue columns it reads — +-- state, attempt, max_attempts, lease_owner, lease_expires_at, run_after, +-- cancel_requested_at), 0263 (the body this extends), 0253 and 0292 (the grants +-- restated below). Number reserved on #699 (comment 5970781386); nothing in +-- 0475–0478 is referenced. +-- +-- IDEMPOTENT: `create or replace function`; every write is bounded to the rows +-- the same statement just dead-lettered (0093 for `approval_requests`, 0251 for +-- its `run_id`). Safe to re-run. +-- ============================================================================ + +create or replace function public.claim_ai_runs(p_limit integer default 10, p_lease_seconds integer default 120) +returns setof uuid +language plpgsql security definer set search_path = public as $$ +declare + v_limit integer := greatest(1, least(coalesce(p_limit, 10), 50)); + v_lease integer := greatest(30, least(coalesce(p_lease_seconds, 120), 900)); + v_dead uuid[]; +begin + -- Two arms, one statement, so the ids either arm dead-letters are captured + -- for the reconcile below. + -- + -- * A run whose lease expired (0263, unchanged): back to `ready`, or to + -- `failed` once `attempt >= max_attempts`. + -- * A run parked in the queue with no attempts left (new): `ready` or + -- `scheduled_followup`, no live lease, no cancel in flight, and + -- `attempt >= max_attempts`. Every such row takes the `failed` arm — + -- the ceiling test is true by construction — instead of being claimed + -- once more. A cancel in flight is left to finish as `cancelled`. + with recovered as ( + update public.family_automation_runs + set state = case when attempt >= max_attempts then 'failed' else 'ready' end, + status = case when attempt >= max_attempts then 'failed' else status end, + error = case when attempt >= max_attempts + then coalesce(error, 'Run abandoned after the maximum number of attempts.') + else error end, + completed_at = case when attempt >= max_attempts then now() else completed_at end, + lease_owner = null, + lease_expires_at = null, + run_after = now(), + updated_at = now() + where (state = 'executing' + and lease_expires_at is not null + and lease_expires_at < now()) + or (state in ('ready','scheduled_followup') + and attempt >= max_attempts + and (lease_expires_at is null or lease_expires_at < now()) + and cancel_requested_at is null) + returning id, state, plan_id, request_id, family_id + ) + select coalesce(array_agg(id) filter (where state = 'failed'), '{}') into v_dead from recovered; + + if array_length(v_dead, 1) is not null then + -- Steps that had not reached their own end. A step in any other state + -- keeps it. + update public.ai_plan_steps s + set status = 'failed', + error = coalesce(s.error, 'Bubaly stopped part-way through and could not pick this up again.'), + completed_at = now(), + updated_at = now() + from public.family_automation_runs r + where r.id = any(v_dead) + and s.plan_id = r.plan_id + and s.status in ('executing', 'verifying', 'queued', 'ready'); + + -- The request ledger: what the person asked for is resolved, not pending. + update public.ai_requests q + set status = 'failed', + error = coalesce(q.error, 'Bubaly stopped working on this and could not pick it up again.'), + updated_at = now() + from public.family_automation_runs r + where r.id = any(v_dead) + and q.id = r.request_id + and q.status not in ('completed', 'partially_completed', 'failed', 'cancelled'); + + -- And the timeline says why it stops here. + insert into public.ai_run_events (family_id, run_id, request_id, event_type, message, actor_kind) + select r.family_id, r.id, r.request_id, 'run_failed', + 'Bubaly stopped working on this and could not pick it up again.', 'system' + from public.family_automation_runs r + where r.id = any(v_dead); + + -- Approvals still waiting on a run that is over. Left `pending`, each one + -- sits in "Needs your decision" for a run that has already died, and a + -- decision on it marks the row approved while the work it gated never + -- starts (lib/services/approvals: a terminal run is never re-opened by a + -- late decision). Closed the way `cancelRun` closes them — `cancelled`, + -- never `expired`: time did not run out, the run did — found by the run + -- and by the step that carries the approval (0251 lets `run_id` be null), + -- with no decider, because nobody decided. + update public.approval_requests a + set status = 'cancelled', + decided_at = now(), + updated_at = now() + where a.status = 'pending' + and (a.run_id = any(v_dead) + or a.id in (select s.approval_id + from public.ai_plan_steps s + join public.family_automation_runs r on r.plan_id = s.plan_id + where r.id = any(v_dead) + and s.approval_id is not null)); + end if; + + -- The claiming half. `attempt < max_attempts` is the same ceiling `claimRun` + -- applies to one row, so the two claim paths now agree; after the arm above + -- it is also a belt-and-braces test, since a run at the ceiling is already + -- `failed` by the time the candidates are read. + return query + with candidates as ( + select id + from public.family_automation_runs + where state in ('ready','scheduled_followup') + and run_after <= now() + and (lease_expires_at is null or lease_expires_at < now()) + and cancel_requested_at is null + and attempt < max_attempts + order by run_after, created_at + for update skip locked + limit v_limit + ) + update public.family_automation_runs r + set state = 'executing', + attempt = r.attempt + 1, + lease_owner = gen_random_uuid(), + lease_expires_at = now() + make_interval(secs => v_lease), + started_at = coalesce(r.started_at, now()), + updated_at = now() + from candidates c + where r.id = c.id + returning r.id; +end; +$$; + +-- `create or replace` keeps the existing grants; restated so a fresh database +-- built from these files alone lands in the same place as 0253 left it. +revoke all on function public.claim_ai_runs(integer, integer) from public, anon, authenticated; +grant execute on function public.claim_ai_runs(integer, integer) to service_role; diff --git a/tests/a-run-that-threw-is-not-stranded.test.ts b/tests/a-run-that-threw-is-not-stranded.test.ts index 3057e3578..e6fe7418a 100644 --- a/tests/a-run-that-threw-is-not-stranded.test.ts +++ b/tests/a-run-that-threw-is-not-stranded.test.ts @@ -18,7 +18,7 @@ // just failed it, and a repeatable throw burnt every attempt in seconds // (review on #900). The pass itself is SQL; this file drives the real route, // the real `continueRun` and the real `releaseRun` against the in-memory client -// with a line-for-line emulation of 0263's two predicates, and +// with the predicate-for-predicate stand-in in tests/helpers/claim-ai-runs.ts, and // docs/audit/a-run-that-threw-is-not-stranded-check.sql proves the same three // outcomes against the real function. import { randomUUID } from 'node:crypto'; @@ -27,6 +27,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import type { SupabaseClient } from '@supabase/supabase-js'; import type { Database } from '@/lib/database.types'; import { createInMemorySupabase, type InMemorySupabase, type Row } from './helpers/in-memory-supabase'; +import { ABANDONED, claimAiRuns } from './helpers/claim-ai-runs'; const seam = vi.hoisted(() => ({ service: vi.fn(), runGraph: vi.fn() })); vi.mock('@/lib/supabase/server', () => ({ createServiceClient: seam.service })); @@ -41,47 +42,9 @@ import { claimRun, RELEASE_RETRY_DELAY_MS, releaseRun } from '@/lib/ai/runs/stor const CRON_SECRET = 'cron-fixture-secret'; const FAMILY = '00000000-0000-4000-8000-00000000a0f1'; const RUN = '00000000-0000-4000-8000-00000000a0a1'; -const ABANDONED = 'Run abandoned after the maximum number of attempts.'; const DONE = { status: 'completed', completed: 1, failed: 0, pending: 0, awaitingApproval: 0 }; const TICK_CADENCE_MS = 5 * 60_000; -/** - * `claim_ai_runs` as 0263 writes it, predicate for predicate: the recovery pass - * over `executing` rows whose lease is NOT NULL and past, then the candidates - * from `ready` and `scheduled_followup`. Rows are the live table objects. - */ -function claimAiRuns(args: Record, db: InMemorySupabase): string[] { - const limit = Math.max(1, Math.min(Number(args.p_limit ?? 10), 50)); - const leaseSeconds = Math.max(30, Math.min(Number(args.p_lease_seconds ?? 120), 900)); - const now = Date.now(); - const rows = db.table('family_automation_runs') as Array; - for (const r of rows) { - if (r.state !== 'executing' || r.lease_expires_at == null || Date.parse(String(r.lease_expires_at)) >= now) continue; - const dead = Number(r.attempt) >= Number(r.max_attempts); - Object.assign(r, { - state: dead ? 'failed' : 'ready', - status: dead ? 'failed' : r.status, - error: dead ? (r.error ?? ABANDONED) : r.error, - completed_at: dead ? new Date(now).toISOString() : r.completed_at, - lease_owner: null, lease_expires_at: null, run_after: new Date(now).toISOString(), - }); - } - const candidates = rows - .filter((r) => (r.state === 'ready' || r.state === 'scheduled_followup') - && Date.parse(String(r.run_after)) <= now - && (r.lease_expires_at == null || Date.parse(String(r.lease_expires_at)) < now) - && r.cancel_requested_at == null) - .sort((a, b) => String(a.run_after).localeCompare(String(b.run_after))) - .slice(0, limit); - for (const r of candidates) { - Object.assign(r, { - state: 'executing', attempt: Number(r.attempt) + 1, lease_owner: randomUUID(), - lease_expires_at: new Date(now + leaseSeconds * 1000).toISOString(), started_at: r.started_at ?? new Date(now).toISOString(), - }); - } - return candidates.map((r) => r.id); -} - let db: InMemorySupabase; let errors: unknown[][]; diff --git a/tests/a-run-with-no-attempts-left-is-abandoned.test.ts b/tests/a-run-with-no-attempts-left-is-abandoned.test.ts new file mode 100644 index 000000000..958576a21 --- /dev/null +++ b/tests/a-run-with-no-attempts-left-is-abandoned.test.ts @@ -0,0 +1,313 @@ +// A run parked in the queue with no attempts left is abandoned, not claimed +// again (finalaudit Q40, second half). +// +// `attempt` is the run's abandonment budget. The recovery arm of `claim_ai_runs` +// dead-letters an `executing` run whose lease ran out once it reaches +// `max_attempts`; `claimRun` refuses to lease a run at that ceiling; and since +// the first half of Q40 the executor hands the budget back only when a slice +// made progress, so the counter is the number of slices in a row that +// completed nothing. The CLAIMING half of `claim_ai_runs` never read it: a run +// parked back to `ready` at the ceiling was claimed by every tick for ever, +// attempt 6, 7, 8 …, while every human-initiated path refused it — unreachable +// by people, unkillable by the worker. The dead-letter arm could not reach it +// either: it ran only for an expired lease on an `executing` row, and a parked +// run holds no lease. +// +// The migration found by `claimAiRunsMigration()` gives the recovery statement +// a second arm — a queued run at the ceiling is abandoned, with 0263's four-table +// reconcile — and the candidates the same `attempt < max_attempts` test the +// one-row path has. The function is SQL; its proof against the real thing is +// docs/audit/a-run-with-no-attempts-left-is-abandoned-check.sql. This file +// drives the real cron route and the real `claimRun` against the in-memory +// client with the predicate-for-predicate stand-in in tests/helpers, and pins +// the migration's shape. +import { readFileSync } from 'node:fs'; +import { join } from 'node:path'; +import { NextRequest } from 'next/server'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import type { SupabaseClient } from '@supabase/supabase-js'; +import type { Database } from '@/lib/database.types'; +import { createInMemorySupabase, type InMemorySupabase, type Row } from './helpers/in-memory-supabase'; +import { ABANDONED, claimAiRuns, claimAiRunsMigration } from './helpers/claim-ai-runs'; + +const seam = vi.hoisted(() => ({ service: vi.fn(), runGraph: vi.fn() })); +vi.mock('@/lib/supabase/server', () => ({ createServiceClient: seam.service })); +vi.mock('@/lib/i18n/server', () => ({ getTranslations: async () => (key: string) => key })); +vi.mock('@/lib/ai/runs/executor', () => ({ runGraph: seam.runGraph })); +vi.mock('@/lib/ai/planner/replan-port', () => ({ replanPortFor: () => null })); + +import { GET as tick } from '@/app/api/cron/ai-runs/route'; +import { claimRun } from '@/lib/ai/runs/store'; + +const ROOT = join(__dirname, '..'); +const CRON_SECRET = 'cron-fixture-secret'; +const FAMILY = '00000000-0000-4000-8000-00000000a0f2'; +const RUN = '00000000-0000-4000-8000-00000000a0b1'; +const OTHER = '00000000-0000-4000-8000-00000000a0b2'; +const TICK_CADENCE_MS = 5 * 60_000; +/** The route's per-run slice. A slice that takes this long lets a tick hold three of them (3 × 25 s + the 12 s floor > 85 s). */ +const PER_RUN_BUDGET_MS = 25_000; +const PARKED = { status: 'ready', completed: 0, failed: 0, pending: 1, awaitingApproval: 0 }; + +let db: InMemorySupabase; +let slices: number; +const client = () => db as unknown as SupabaseClient; +const rowOf = (id: string) => db.table('family_automation_runs').find((r) => r.id === id) as Row; +const run = () => rowOf(RUN); + +function seedRun(overrides: Row = {}) { + db.seed('family_automation_runs', [{ + id: RUN, family_id: FAMILY, run_type: 'concierge_plan', state: 'ready', status: 'approved', progress: {}, + attempt: 0, max_attempts: 5, lease_owner: null, lease_expires_at: null, cancel_requested_at: null, paused_at: null, + started_at: null, completed_at: null, error: null, summary: null, plan_id: null, request_id: null, + run_after: new Date(Date.now() - 60_000).toISOString(), + ...overrides, + }]); +} + +/** + * What `parkForContinuation` writes when the slice ran out of budget: back to + * `ready`, due now, lease cleared — and `attempt` handed back ONLY when the + * slice left more steps satisfied than it found (executor.ts, budgetReset). + * Each slice spends its 25 s of the tick, so the route's own loop decides how + * many slices a tick holds, exactly as in production. + */ +function parks(progressed: boolean | ((slice: number) => boolean), tookMs = PER_RUN_BUDGET_MS) { + seam.runGraph.mockImplementation(async (runId: string) => { + slices += 1; + const did = typeof progressed === 'function' ? progressed(slices) : progressed; + vi.setSystemTime(new Date(Date.now() + tookMs)); + Object.assign(rowOf(runId), { + state: 'ready', status: 'approved', lease_owner: null, lease_expires_at: null, + run_after: new Date().toISOString(), ...(did ? { attempt: 0 } : {}), + }); + return PARKED; + }); +} + +/** What the executor writes when the slice finishes the run. */ +function completesTheRun() { + seam.runGraph.mockImplementation(async (runId: string) => { + slices += 1; + vi.setSystemTime(new Date(Date.now() + PER_RUN_BUDGET_MS)); + Object.assign(rowOf(runId), { state: 'completed', status: 'executed', lease_owner: null, lease_expires_at: null, completed_at: new Date().toISOString() }); + return { status: 'completed', completed: 1, failed: 0, pending: 0, awaitingApproval: 0 }; + }); +} + +async function cronTick() { + vi.setSystemTime(new Date(Date.now() + TICK_CADENCE_MS)); + const res = await tick(new NextRequest('https://fixture.invalid/api/cron/ai-runs', { headers: { authorization: `Bearer ${CRON_SECRET}` } })); + return { status: res.status, body: await res.json() as { claimed: number; executed: number; returnedToQueue: number; results: Array<{ runId: string; status: string }> } }; +} + +beforeEach(() => { + vi.useFakeTimers({ toFake: ['Date'] }); + vi.setSystemTime(new Date('2026-10-03T12:00:00.000Z')); + vi.stubEnv('CRON_SECRET', CRON_SECRET); + db = createInMemorySupabase({ rpc: { claim_ai_runs: (args, c) => claimAiRuns(args, c) } }); + db.seed('families', [{ id: FAMILY, name: 'Fixture', timezone: 'UTC' }]); + slices = 0; + seam.service.mockImplementation(() => db); + seam.runGraph.mockReset(); + vi.spyOn(console, 'error').mockImplementation(() => undefined); +}); +afterEach(() => { vi.useRealTimers(); vi.restoreAllMocks(); vi.unstubAllEnvs(); }); + +describe('the cron, when a parked run has no attempts left', () => { + it('five slices that complete nothing spend the budget; the next claim abandons the run instead of taking it again', async () => { + seedRun(); + parks(false); + // A tick holds three slices. Each one parks the run `ready` and due, so the + // same tick's next wave takes it again — that is how a run gets more than + // one slice per tick — and each slice that completes nothing is charged. + const first = await cronTick(); + expect(first.body.claimed).toBe(3); + expect(run()).toMatchObject({ state: 'ready', attempt: 3, lease_owner: null, error: null }); + + // Slices four and five, then the third wave finds a run at the ceiling in + // the queue: abandoned, and the claim comes back empty. + const second = await cronTick(); + expect(second.status).toBe(200); + expect(second.body.claimed).toBe(2); + expect(slices, 'exactly max_attempts slices, then no more').toBe(5); + expect(run()).toMatchObject({ state: 'failed', status: 'failed', error: ABANDONED, attempt: 5, lease_owner: null, lease_expires_at: null }); + expect(run().completed_at).toBeTruthy(); + + // And it stays abandoned: a later tick neither claims nor re-fails it. + const third = await cronTick(); + expect(third.body.claimed).toBe(0); + expect(slices).toBe(5); + expect(run()).toMatchObject({ state: 'failed', attempt: 5 }); + }); + + it('a slice that made progress hands the budget back, so the ceiling counts stalls in a row, not slices', async () => { + // The two halves of Q40 together: four stalls, one slice that completed a + // step, five more stalls — abandoned after the tenth slice, not the fifth. + seedRun(); + parks((slice) => slice === 5); + await cronTick(); // slices 1-3: three stalls + expect(run()).toMatchObject({ state: 'ready', attempt: 3 }); + await cronTick(); // 4: stall; 5: progress, back to 0; 6: stall + expect(run()).toMatchObject({ state: 'ready', attempt: 1 }); + await cronTick(); // 7-9: stalls + expect(run()).toMatchObject({ state: 'ready', attempt: 4, error: null }); + await cronTick(); // 10: the fifth stall in a row; the next wave abandons + expect(slices).toBe(10); + expect(run()).toMatchObject({ state: 'failed', error: ABANDONED, attempt: 5 }); + }); + + it('what the cron abandons, claimRun had already refused — the two claim paths agree, and the refusal now has a reason', async () => { + seedRun({ attempt: 5 }); + const before = await claimRun(client(), RUN, 120); + expect(before.ok && before.data.claimed, 'the one-row path refuses at the ceiling').toBe(false); + expect(before.ok && before.data.run?.state, 'and until the cron ran, all a caller saw was "ready"').toBe('ready'); + + const t = await cronTick(); + expect(t.body.claimed).toBe(0); + expect(seam.runGraph).not.toHaveBeenCalled(); + + const after = await claimRun(client(), RUN, 120); + expect(after.ok && after.data.claimed).toBe(false); + expect(after.ok && after.data.run, 'a caller now sees a failed run with its reason').toMatchObject({ state: 'failed', error: ABANDONED }); + }); + + it('abandons a scheduled follow-up at the ceiling now, not on the day it was due', async () => { + seedRun({ state: 'scheduled_followup', attempt: 5, run_after: new Date(Date.now() + 3 * 24 * 3_600_000).toISOString() }); + const t = await cronTick(); + expect(t.body.claimed).toBe(0); + expect(run()).toMatchObject({ state: 'failed', status: 'failed', error: ABANDONED }); + }); + + it.each([ + ['a cancel is in flight — it must end cancelled, not failed', { cancel_requested_at: new Date(Date.now() - 1_000).toISOString() }], + ['its lease is still live — not the queue\'s to touch', { lease_owner: 'someone', lease_expires_at: new Date(Date.now() + 10 * 60_000).toISOString() }], + ])('leaves a run at the ceiling alone when %s', async (_label, overrides) => { + seedRun({ attempt: 5, ...overrides }); + const before = { ...run() }; + const t = await cronTick(); + expect(t.body.claimed).toBe(0); + expect(run()).toEqual(before); + }); + + it('one attempt below the ceiling, the run gets its slice — and if that slice finishes it, it finishes', async () => { + seedRun({ attempt: 4 }); + completesTheRun(); + const t = await cronTick(); + expect(t.body.claimed).toBe(1); + expect(slices).toBe(1); + expect(run()).toMatchObject({ state: 'completed', attempt: 5, error: null }); + }); + + it('one attempt below the ceiling, only a fifth slice that completes nothing abandons it', async () => { + seedRun({ attempt: 4 }); + parks(false); + const t = await cronTick(); + expect(t.body.claimed, 'claimed once for the fifth slice; the next wave abandons rather than claims').toBe(1); + expect(slices).toBe(1); + expect(run()).toMatchObject({ state: 'failed', error: ABANDONED, attempt: 5 }); + }); + + it('a run claimed in a wave but handed back unexecuted is not charged an attempt', async () => { + // The wave claims four runs at once, and the tick gives each a slice only + // while at least 12 s remain. The claim itself charges `attempt`, so a run + // the tick never reached used to carry a charge for a slice it never had. + // With the ceiling enforced by claim_ai_runs, five such hand-backs would + // abandon a run that was never tried. The hand-back now returns the charge. + seedRun(); + db.seed('family_automation_runs', [{ ...run(), id: OTHER, run_after: new Date(Date.now() - 30_000).toISOString() }]); + parks(false, 80_000); // the first run's slice uses up the tick + const t = await cronTick(); + expect(t.body.claimed).toBe(2); + expect(t.body.returnedToQueue).toBe(1); + expect(seam.runGraph).toHaveBeenCalledTimes(1); + expect(seam.runGraph).toHaveBeenCalledWith(RUN, expect.anything()); + expect(rowOf(RUN), 'the run that had its slice is charged').toMatchObject({ state: 'ready', attempt: 1 }); + expect(rowOf(OTHER), 'the run that did not is not').toMatchObject({ state: 'ready', attempt: 0, lease_owner: null, lease_expires_at: null }); + }); +}); + +describe('the migration', () => { + const { path, sql: raw } = claimAiRunsMigration(ROOT); + const sql = raw.split('\n').filter((l) => !l.trimStart().startsWith('--')).join('\n'); + const recovery = sql.slice(sql.indexOf('with recovered as'), sql.indexOf('returning id, state, plan_id, request_id, family_id')); + const claimPass = sql.slice(sql.indexOf('with candidates as'), sql.indexOf('returning r.id')); + + it('rewrites the one function that claims runs', () => { + expect(path).toMatch(/^supabase\/migrations\/\d{4}_a_run_with_no_attempts_left_is_abandoned_not_reclaimed\.sql$/); + expect(sql).toContain('create or replace function public.claim_ai_runs(p_limit integer default 10, p_lease_seconds integer default 120)'); + }); + + it('abandons a queued run at the ceiling in the recovery statement: no live lease, no cancel in flight', () => { + expect(recovery).toMatch(/or \(state in \('ready','scheduled_followup'\)\s+and attempt >= max_attempts\s+and \(lease_expires_at is null or lease_expires_at < now\(\)\)\s+and cancel_requested_at is null\)/); + // and keeps the expired-lease arm as 0263 wrote it + expect(recovery).toMatch(/\(state = 'executing'\s+and lease_expires_at is not null\s+and lease_expires_at < now\(\)\)/); + expect(recovery).toContain("state = case when attempt >= max_attempts then 'failed' else 'ready' end"); + expect(recovery).toContain("status = case when attempt >= max_attempts then 'failed' else status end"); + expect(recovery).toContain("'Run abandoned after the maximum number of attempts.'"); + }); + + it('gives the candidates the same ceiling claimRun applies to one row', () => { + expect(claimPass).toMatch(/and attempt < max_attempts/); + expect(claimPass).toContain("state in ('ready','scheduled_followup')"); + expect(claimPass).toContain('cancel_requested_at is null'); + expect(claimPass).toContain('for update skip locked'); + const store = readFileSync(join(ROOT, 'lib/ai/runs/store.ts'), 'utf8'); + expect(store).toMatch(/run\.attempt < run\.max_attempts/); + }); + + it('keeps the claiming half otherwise as it was — a lease token stays a uuid, and attempt still advances', () => { + expect(sql).toContain('lease_owner = gen_random_uuid()'); + expect(sql).not.toContain('gen_random_uuid()::text'); + expect(sql).toContain('attempt = r.attempt + 1'); + }); + + it("keeps 0263's reconcile, bounded to the runs the statement dead-lettered", () => { + expect(sql).toMatch(/select coalesce\(array_agg\(id\) filter \(where state = 'failed'\), '\{\}'\) into v_dead/); + // Steps, request ledger, timeline — and the approvals write below. + expect(sql.match(/r\.id = any\(v_dead\)/g)).toHaveLength(4); + expect(sql).toMatch(/update public\.ai_plan_steps s[\s\S]*?and s\.status in \('executing', 'verifying', 'queued', 'ready'\)/); + expect(sql).toMatch(/update public\.ai_requests q[\s\S]*?and q\.status not in \('completed', 'partially_completed', 'failed', 'cancelled'\)/); + expect(sql).toMatch(/insert into public\.ai_run_events[\s\S]*?'run_failed'/); + }); + + it("closes a dead run's pending approvals, by the run and by the step, with no decider", () => { + // 0263 left them open: a card in "Needs your decision" for a run that had + // died, and a decision on it approved work that never starts. + const approvals = sql.slice(sql.indexOf('update public.approval_requests a'), sql.indexOf(' end if;')); + expect(approvals.length, 'the approvals write sits inside the dead-letter block').toBeGreaterThan(0); + expect(approvals).toContain("set status = 'cancelled'"); + expect(approvals).toContain('decided_at = now()'); + expect(approvals, 'nobody decided').not.toContain('decided_by'); + expect(approvals).toContain("where a.status = 'pending'"); + expect(approvals).toContain('a.run_id = any(v_dead)'); + expect(approvals).toMatch(/or a\.id in \(select s\.approval_id[\s\S]*?where r\.id = any\(v_dead\)[\s\S]*?and s\.approval_id is not null\)/); + }); + + it('keeps the function service-role only, as 0253 left it', () => { + expect(sql).toContain('revoke all on function public.claim_ai_runs(integer, integer) from public, anon, authenticated'); + expect(sql).toContain('grant execute on function public.claim_ai_runs(integer, integer) to service_role'); + }); + + it('ships a proof with a control one attempt below the ceiling, and pins what is left alone', () => { + const proof = readFileSync(join(ROOT, 'docs/audit/a-run-with-no-attempts-left-is-abandoned-check.sql'), 'utf8'); + for (const claim of [ + 'a ready run with no attempts left was claimed again', + 'CONTROL FAILED: a due ready run on attempt 4 of 5 was not claimed', + 'a scheduled follow-up with no attempts left was left as', + 'a cancel in flight', + 'lease is still live', + 'expected one run_failed event', + 'approval (by run_id) was left', + 'reachable only through the step', + 'names a decider when nobody decided', + // SQL doubles the apostrophe inside its string literal. + "CONTROL FAILED: the claimed run''s approval was", + ]) { + expect(proof, claim).toContain(claim); + } + expect(proof.trim().startsWith('--')).toBe(true); + expect(proof).toMatch(/\nbegin;\n/); + expect(proof.trim().endsWith('rollback;')).toBe(true); + }); +}); diff --git a/tests/a-stale-worker-cannot-write-another-workers-run.test.ts b/tests/a-stale-worker-cannot-write-another-workers-run.test.ts index d45839350..7fed5a102 100644 --- a/tests/a-stale-worker-cannot-write-another-workers-run.test.ts +++ b/tests/a-stale-worker-cannot-write-another-workers-run.test.ts @@ -10,8 +10,8 @@ // awaits `runTool` without a deadline, so a tool can outlive the lease. // // The interleaving the review described, driven here through the real -// `createExecutorPort` over the in-memory client and a predicate-for-predicate -// stand-in for `claim_ai_runs` (0263): +// `createExecutorPort` over the in-memory client and the predicate-for-predicate +// stand-in for `claim_ai_runs` (tests/helpers/claim-ai-runs.ts): // // worker A claims the run at attempt 4 with lease A and blocks inside a tool; // A's lease expires; @@ -43,8 +43,8 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import type { SupabaseClient } from '@supabase/supabase-js'; import type { Database } from '@/lib/database.types'; import type { ToolOutcome } from '@/lib/ai/tools/types'; -import { randomUUID } from 'node:crypto'; import { createInMemorySupabase, type InMemorySupabase, type Row } from './helpers/in-memory-supabase'; +import { ABANDONED, claimAiRuns } from './helpers/claim-ai-runs'; vi.mock('@/lib/supabase/server', () => ({ createServiceClient: () => { throw new Error('the test hands the port its own client'); }, @@ -68,48 +68,6 @@ const NOW = new Date('2026-10-03T12:00:00.000Z'); const CEILING = 5; const DONE: ToolOutcome = { status: 'ok', data: { id: 'ev-1' }, summary: 'Added the dentist.', toolCallId: 'call-1', verified: true }; -const ABANDONED = 'Run abandoned after the maximum number of attempts.'; - -/** - * `claim_ai_runs` as 0263 writes it, predicate for predicate (the same stand-in - * tests/a-run-that-threw-is-not-stranded.test.ts drives the cron with): the - * recovery pass over `executing` rows whose lease is NOT NULL and past — back - * to `ready`, or `failed` once `attempt >= max_attempts` — then the candidates - * from `ready` and `scheduled_followup`, each leased with one more attempt. - * Rows are the live table objects. - */ -function claimAiRuns(args: Record, store: InMemorySupabase): string[] { - const limit = Math.max(1, Math.min(Number(args.p_limit ?? 10), 50)); - const leaseSeconds = Math.max(30, Math.min(Number(args.p_lease_seconds ?? 120), 900)); - const now = Date.now(); - const rows = store.table('family_automation_runs') as Array; - for (const r of rows) { - if (r.state !== 'executing' || r.lease_expires_at == null || Date.parse(String(r.lease_expires_at)) >= now) continue; - const dead = Number(r.attempt) >= Number(r.max_attempts); - Object.assign(r, { - state: dead ? 'failed' : 'ready', - status: dead ? 'failed' : r.status, - error: dead ? (r.error ?? ABANDONED) : r.error, - completed_at: dead ? new Date(now).toISOString() : r.completed_at, - lease_owner: null, lease_expires_at: null, run_after: new Date(now).toISOString(), - }); - } - const candidates = rows - .filter((r) => (r.state === 'ready' || r.state === 'scheduled_followup') - && Date.parse(String(r.run_after)) <= now - && (r.lease_expires_at == null || Date.parse(String(r.lease_expires_at)) < now) - && r.cancel_requested_at == null) - .sort((a, b) => String(a.run_after).localeCompare(String(b.run_after))) - .slice(0, limit); - for (const r of candidates) { - Object.assign(r, { - state: 'executing', attempt: Number(r.attempt) + 1, lease_owner: randomUUID(), - lease_expires_at: new Date(now + leaseSeconds * 1000).toISOString(), started_at: r.started_at ?? new Date(now).toISOString(), - }); - } - return candidates.map((r) => r.id); -} - let db: InMemorySupabase; const client = () => db as unknown as SupabaseClient; const run = () => db.table('family_automation_runs').find((r) => r.id === RUN) as Row; diff --git a/tests/a-successful-slice-must-not-spend-a-retry.test.ts b/tests/a-successful-slice-must-not-spend-a-retry.test.ts index 5233f9d0d..2a79415e9 100644 --- a/tests/a-successful-slice-must-not-spend-a-retry.test.ts +++ b/tests/a-successful-slice-must-not-spend-a-retry.test.ts @@ -5,6 +5,7 @@ import type { SupabaseClient } from '@supabase/supabase-js'; import type { Database } from '@/lib/database.types'; import { createInMemorySupabase, type InMemorySupabase } from './helpers/in-memory-supabase'; import { claimRun, releaseRun } from '@/lib/ai/runs/store'; +import { claimAiRunsMigration } from './helpers/claim-ai-runs'; const ROOT = join(__dirname, '..'); const FAMILY = '00000000-0000-4000-8000-00000000fa01'; @@ -47,8 +48,10 @@ const RUN = '00000000-0000-4000-8000-00000000ru01'; * (run-executor.test.ts pins that through the real `runGraphWith`). A slice that * completed nothing keeps the charge, which is what still lets a run that never * progresses terminate — the proof Q40 asked for is the first case below. The - * second half — the same ceiling inside `claim_ai_runs` — is a migration, and - * is still open; the last case pins that the two claim paths still disagree. + * second half — the same ceiling inside `claim_ai_runs`, so that a run the + * one-row path refuses is abandoned by the cron rather than claimed for ever — + * is the migration the last case pins (tests/a-run-with-no-attempts-left-is- + * abandoned.test.ts drives it). * * This file drives the REAL `claimRun` against the in-memory PostgREST stand-in * — real filters, the real compare-and-set on `attempt` — with the park written @@ -185,23 +188,31 @@ describe('a successful slice must not spend a retry (Q40)', () => { expect(executor.match(/budgetReset\(/g)?.length).toBe(4); }); - it('the two claim paths still disagree — the second half of Q40 is a migration and is still open', () => { - // store.claimRun refuses at the ceiling; the cron RPC has no such filter in - // its claiming pass, so a run past max_attempts is unreachable by every - // human-initiated path while the cron keeps moving it. With the budget now - // spent only on slices that made no progress, that is much harder to reach - // — but the inconsistency itself is unchanged and still a bug. - const rpc = readFileSync(join(ROOT, 'supabase/migrations/0250_ai_runtime_core.sql'), 'utf8'); - const claimPass = rpc.slice(rpc.indexOf('with candidates as'), rpc.indexOf('returning r.id')); - expect(claimPass, 'the claiming pass gained an attempt ceiling — re-check this finding').not.toMatch(/attempt\s*<\s*.*max_attempts/); - expect(claimPass, 'the claiming pass no longer increments attempt').toMatch(/attempt = r\.attempt \+ 1/); - - // And the counter it spends is the one the recovery pass calls abandonment. - expect(rpc).toMatch(/case when attempt >= max_attempts then 'failed'/); - expect(rpc).toContain('Run abandoned after the maximum number of attempts.'); - - // Per-step retries are a different column entirely, which is the evidence + it('the two claim paths agree: the cron abandons a run at the ceiling claimRun refuses at (Q40, second half)', () => { + // store.claimRun refuses at the ceiling. The cron RPC's claiming pass had no + // such filter, so a run past max_attempts was unreachable by every + // human-initiated path while the cron kept moving it — and its dead-letter + // arm never reached a parked run, because that arm only ran for an expired + // lease on an `executing` row. The current definition closes both. + const { sql } = claimAiRunsMigration(ROOT); + const recovery = sql.slice(sql.indexOf('with recovered as'), sql.indexOf('returning id, state')); + expect(recovery, 'a queued run at the ceiling is abandoned').toMatch(/state in \('ready','scheduled_followup'\)\s+and attempt >= max_attempts/); + const claimPass = sql.slice(sql.indexOf('with candidates as'), sql.indexOf('returning r.id')); + expect(claimPass, 'the claiming pass has the ceiling').toMatch(/and attempt < max_attempts/); + expect(claimPass, 'and still increments attempt').toMatch(/attempt = r\.attempt \+ 1/); + const store = readFileSync(join(ROOT, 'lib/ai/runs/store.ts'), 'utf8'); + expect(store, 'the one-row path keeps its ceiling').toMatch(/run\.attempt < run\.max_attempts/); + + // And the counter both paths stop at is the one the recovery arm calls abandonment. + expect(sql).toMatch(/case when attempt >= max_attempts then 'failed'/); + expect(sql).toContain('Run abandoned after the maximum number of attempts.'); + + // The premise, kept as history: 0250's original claiming pass had no ceiling, + // and per-step retries were always a different column, which is the evidence // that run-level `attempt` was never meant to be the retry mechanism. - expect(rpc).toMatch(/max_retries/); + const original = readFileSync(join(ROOT, 'supabase/migrations/0250_ai_runtime_core.sql'), 'utf8'); + const originalPass = original.slice(original.indexOf('with candidates as'), original.indexOf('returning r.id')); + expect(originalPass).not.toMatch(/attempt\s*<\s*.*max_attempts/); + expect(original).toMatch(/max_retries/); }); }); diff --git a/tests/helpers/claim-ai-runs.ts b/tests/helpers/claim-ai-runs.ts new file mode 100644 index 000000000..260fc74d8 --- /dev/null +++ b/tests/helpers/claim-ai-runs.ts @@ -0,0 +1,82 @@ +import { readdirSync, readFileSync } from 'node:fs'; +import { join } from 'node:path'; +import { randomUUID } from 'node:crypto'; +import type { InMemorySupabase, Row } from './in-memory-supabase'; + +/** + * `claim_ai_runs`, predicate for predicate, against the in-memory client. + * + * The real function is SQL (0250, rewritten by 0263 and again by the migration + * `claimAiRunsMigration()` finds), and its behavioural proofs are the SQL + * probes under docs/audit. This is the stand-in a test registers as + * `rpc: { claim_ai_runs }` so the real cron route, `continueRun`, `claimRun` + * and `releaseRun` can be driven end to end. It writes the RUN ROW only; the + * reconcile of a dead-lettered run (steps, request ledger, timeline, pending + * approvals) is the probes' to prove. + * + * The recovery statement has two arms sharing one `failed` outcome: + * - an `executing` run whose lease is NOT NULL and in the past: back to + * `ready`, or `failed` once `attempt >= max_attempts`; + * - a run parked in the queue (`ready` / `scheduled_followup`) with no live + * lease, no cancel in flight and `attempt >= max_attempts`: `failed`. + * The candidates are then the due, unleased, uncancelled queue rows BELOW the + * ceiling, oldest `run_after` first, each claimed with a fresh lease and one + * more attempt. + */ +export const ABANDONED = 'Run abandoned after the maximum number of attempts.'; + +export function claimAiRuns(args: Record, db: InMemorySupabase): string[] { + const limit = Math.max(1, Math.min(Number(args.p_limit ?? 10), 50)); + const leaseSeconds = Math.max(30, Math.min(Number(args.p_lease_seconds ?? 120), 900)); + const now = Date.now(); + const rows = db.table('family_automation_runs') as Array; + + const leaseLive = (r: Row) => r.lease_expires_at != null && Date.parse(String(r.lease_expires_at)) >= now; + const leaseExpired = (r: Row) => r.state === 'executing' && r.lease_expires_at != null && Date.parse(String(r.lease_expires_at)) < now; + const inQueue = (r: Row) => r.state === 'ready' || r.state === 'scheduled_followup'; + const atCeiling = (r: Row) => Number(r.attempt) >= Number(r.max_attempts); + const spentInQueue = (r: Row) => inQueue(r) && atCeiling(r) && !leaseLive(r) && r.cancel_requested_at == null; + + for (const r of rows) { + if (!leaseExpired(r) && !spentInQueue(r)) continue; + const dead = atCeiling(r); + Object.assign(r, { + state: dead ? 'failed' : 'ready', + status: dead ? 'failed' : r.status, + error: dead ? (r.error ?? ABANDONED) : r.error, + completed_at: dead ? new Date(now).toISOString() : r.completed_at, + lease_owner: null, lease_expires_at: null, run_after: new Date(now).toISOString(), + }); + } + + const candidates = rows + .filter((r) => inQueue(r) + && Date.parse(String(r.run_after)) <= now + && !leaseLive(r) + && r.cancel_requested_at == null + && !atCeiling(r)) + .sort((a, b) => String(a.run_after).localeCompare(String(b.run_after))) + .slice(0, limit); + for (const r of candidates) { + Object.assign(r, { + state: 'executing', attempt: Number(r.attempt) + 1, lease_owner: randomUUID(), + lease_expires_at: new Date(now + leaseSeconds * 1000).toISOString(), started_at: r.started_at ?? new Date(now).toISOString(), + }); + } + return candidates.map((r) => r.id); +} + +const SLUG = '_a_run_with_no_attempts_left_is_abandoned_not_reclaimed.sql'; + +/** + * The migration that currently defines `claim_ai_runs`, found by its name + * rather than its number: numbers are reserved by the audit coordinator and a + * file can be renumbered when branches land, and a test that pins the body + * should not break on that. + */ +export function claimAiRunsMigration(root = join(__dirname, '..', '..')): { path: string; sql: string } { + const dir = join(root, 'supabase', 'migrations'); + const file = readdirSync(dir).find((f) => f.endsWith(SLUG)); + if (!file) throw new Error(`no migration named *${SLUG} under ${dir}`); + return { path: join('supabase', 'migrations', file), sql: readFileSync(join(dir, file), 'utf8') }; +} diff --git a/tests/migration-version-safety.test.ts b/tests/migration-version-safety.test.ts index c8e9c53c4..82b9fcaa4 100644 --- a/tests/migration-version-safety.test.ts +++ b/tests/migration-version-safety.test.ts @@ -502,10 +502,12 @@ describe('Supabase migration filename safety', () => { // /family/permissions shows them as read-only on (ROLE-M03). // 0471 adds the per-recipient admin digest delivery store, and 0474 // (reserved for #710) withdraws an admin removed after a digest was - // frozen. This literal tracks the checked-in high-water mark, not - // migration allocation: 0465-0470 remain NWV's, 0472 Support's, and 0473 - // the coordinator's. - expect(audit.nextVersion).toBe('0475'); + // frozen. 0479 (reserved on #699, 5970781386) makes the cron's claim pass + // honour the attempt ceiling and abandon a queued run with no attempts + // left (Q40). This literal tracks the checked-in high-water mark, not + // migration allocation: 0465-0470 remain NWV's, 0472 Support's, 0473 the + // coordinator's, 0475-0476 messaging's, 0477 Surge's and 0478 Meals'. + expect(audit.nextVersion).toBe('0480'); }); it('flags a newly introduced collision instead of silently accepting it', () => {