AI runs: a run a person asks for again gets its attempt budget back (Q40 follow-up)

lib/ai/runs/controls.ts

@@ -22,7 +22,7 @@ import { describeDbError, wroteNoRows } from '@/lib/supabase/errors';
 import { fail, ok, SERVICE_CODES, type ServiceResult, type ServiceScope } from '@/lib/services/types';
 import { canTransitionRun, isTerminalRunState, legacyStatusFor, type RunState, type StepState } from './states';
 import {
-  appendEvent, ledgerClient, loadPlanSteps, loadRun, savePlan, updateRequest, updateRun, updateStep,
+  appendEvent, freshBudget, ledgerClient, loadPlanSteps, loadRun, savePlan, updateRequest, updateRun, updateRunAsObserved, updateStep,
   type PlanStepInput, type RunRow, type StepRow, type StoreOpts,
 } from './store';
 
@@ -116,14 +116,20 @@ export async function resumeRun(scope: ServiceScope, runId: string, opts?: Contr
 
   if (run.state !== 'paused') return fail('That run is not paused.', { code: SERVICE_CODES.invalidInput });
 
-  const updated = await updateRun(scope, runId, {
+  // Conditioned on the run as this call loaded it (store.ts,
+  // updateRunAsObserved): a second resume from another tab, or a claim, that
+  // landed between the read and this write is not overwritten — the run is
+  // simply no longer paused, which is what the caller is told.
+  const updated = await updateRunAsObserved(scope, runId, run, {
     state: 'ready',
     status: legacyStatusFor('ready'),
     paused_at: null,
     run_after: new Date().toISOString(),
     error: null,
+    ...freshBudget(),
   }, { db });
   if (!updated.ok) return updated;
+  if (!updated.data) return fail('That run is not paused.', { code: SERVICE_CODES.invalidInput });
 
   await appendEvent(scope, runId, { eventType: 'resumed', message: 'Resumed - Bubaly will pick this up again shortly.' }, { db });
   if (run.request_id) await updateRequest(scope, run.request_id, { status: 'ready', error: null }, { db });
@@ -223,6 +229,15 @@ export type RerunResult = { rerun: boolean; state: StepState; detail: string };
  * A `failed` ledger row has its attempt bumped so the executor's idempotency
  * ledger takes it over and the retry really re-executes.
  */
+/**
+ * A person's re-entry found the run changed between its read and its write: a
+ * worker claimed it, another tab moved it, a decision folded in. The write was
+ * conditioned on what the person saw (store.ts, updateRunAsObserved) and so
+ * did not land; what they asked for is already in motion or already moot, and
+ * the run shows which.
+ */
+const MOVED_ON = 'That run changed while this was being asked; nothing was overwritten. Check the run and try again if it still needs it.';
+
 export async function rerunStep(
   scope: ServiceScope,
   runId: string,
@@ -308,14 +323,20 @@ export async function rerunStep(
     }
   }
 
-  const resumed = await updateRun(scope, runId, {
+  // Conditioned on the run as this call loaded it. A worker that claimed the
+  // run meanwhile reloads the steps every pass and will pick up the step just
+  // re-queued; writing `ready`, a fresh budget and this snapshot's lease over
+  // its claim is what must not happen.
+  const resumed = await updateRunAsObserved(scope, runId, run, {
     state: 'ready',
     status: legacyStatusFor('ready'),
     run_after: new Date().toISOString(),
     completed_at: null,
     error: null,
+    ...freshBudget(),
   }, { db });
   if (!resumed.ok) return resumed;
+  if (!resumed.data) return fail(MOVED_ON, { code: SERVICE_CODES.invalidInput, retryable: true });
 
   await appendEvent(scope, runId, {
     eventType: 'step_retried',
@@ -417,16 +438,21 @@ export async function editStepInput(
   }, { db });
   if (!saved.ok) return saved;
 
-  const repointed = await updateRun(scope, runId, {
+  // Conditioned on the run as this call loaded it. When the run moved first,
+  // the new plan version stays saved but unreferenced — a version the run
+  // never pointed at — rather than being pointed at over a live claim.
+  const repointed = await updateRunAsObserved(scope, runId, run, {
     plan_id: saved.data.planId,
     state: 'ready',
     status: legacyStatusFor('ready'),
     run_after: new Date().toISOString(),
     completed_at: null,
     error: null,
     current_step_id: null,
+    ...freshBudget(),
   }, { db });
   if (!repointed.ok) return repointed;
+  if (!repointed.data) return fail(MOVED_ON, { code: SERVICE_CODES.invalidInput, retryable: true });
 
   await appendEvent(scope, runId, {
     eventType: 'planned',

lib/ai/runs/store.ts

@@ -579,6 +579,53 @@ export async function updateRunWhereState(
   return ok((data ?? []).length > 0);
 }
 
+/** The run as a person's re-entry found it: what its write is conditioned on. */
+export type RunObservation = Pick<RunRow, 'state' | 'attempt' | 'lease_owner'>;
+
+/**
+ * Compare-and-set on the run as the caller OBSERVED it — state, attempt and
+ * lease together, not the state alone.
+ *
+ * `updateRunWhereState` guards the answer path against a double-submitted
+ * answer. The writes that put a run back in the queue because a person asked
+ * — an approval decision folding into its run, resume, re-run a step, edit a
+ * step — need the wider guard. Each loads the run, decides from that snapshot,
+ * and writes `ready`, a fresh budget and a cleared lease. Two of them can hold
+ * the same snapshot: two final approval folds on one run. The first puts the
+ * run in the queue; a worker claims it (`executing`, attempt charged, a live
+ * lease); the second, still holding the snapshot that said
+ * `awaiting_approval`, wrote `ready`, zeroed the attempt and cleared the
+ * worker's live lease through the plain `updateRun`, which matches on id and
+ * family alone (review on #917, 5973301004). Conditioned on every field the
+ * snapshot showed, that second write matches nothing once anything has moved
+ * — including a run that is `executing` before and after under two different
+ * workers, which the state alone would not tell apart. `false` is "someone
+ * else moved this run first", and the caller leaves the run where that
+ * someone put it.
+ */
+export async function updateRunAsObserved(
+  scope: ServiceScope,
+  runId: string,
+  observed: RunObservation,
+  patch: Database['public']['Tables']['family_automation_runs']['Update'],
+  opts?: StoreOpts,
+): Promise<ServiceResult<boolean>> {
+  const base = ledgerClient(scope, opts)
+    .from('family_automation_runs')
+    .update(patch)
+    .eq('id', runId)
+    .eq('family_id', scope.familyId)
+    .eq('state', observed.state)
+    .eq('attempt', observed.attempt);
+  const fenced = observed.lease_owner === null ? base.is('lease_owner', null) : base.eq('lease_owner', observed.lease_owner);
+  const { data, error } = await fenced.select('id');
+  if (error) {
+    console.error('[ai/runs] failed to update the run as observed', error);
+    return fail(describeDbError(error, 'Bubaly could not update that run.'), { code: SERVICE_CODES.db, retryable: true });
+  }
+  return ok((data ?? []).length > 0);
+}
+
 // ─── Leases ─────────────────────────────────────────────────────────────────
 
 /** `attempt` is the value AFTER the claim charged it, so a caller that hands the run back unexecuted can give the charge back. */
@@ -614,6 +661,23 @@ export async function claimRuns(
   return ok((rows ?? []).map((r) => ({ id: r.id, familyId: r.family_id, leaseOwner: r.lease_owner, attempt: r.attempt })));
 }
 
+/**
+ * The attempt budget a person's own re-entry hands back.
+ *
+ * `attempt` counts the slices in a row that completed nothing; at
+ * `max_attempts` the run is abandoned (`claim_ai_runs`' recovery arm) and
+ * `claimRun` refuses to lease it. Resume, re-run a step, edit a step and an
+ * approval decision (lib/services/approvals) each put the run back to `ready`
+ * because a member of the family asked for it, and each one tells them Bubaly
+ * will pick it up. Written without this, a run that had spent its budget was
+ * promised that and then refused by the kick and dead-lettered by the next
+ * tick, with nothing a person could do about it. A human asking again is a
+ * new budget, the way a slice that made progress is (executor.ts, budgetReset).
+ */
+export function freshBudget(): Pick<Database['public']['Tables']['family_automation_runs']['Update'], 'attempt'> {
+  return { attempt: 0 };
+}
+
 /**
  * Lease ONE run by id.
  *

lib/services/approvals/index.ts

@@ -55,7 +55,7 @@ import { getTranslations } from '@/lib/i18n/server';
 import { kickRun } from '@/lib/ai/runs/continue';
 import { isTerminalRunState, legacyStatusFor, type RunState, type StepState } from '@/lib/ai/runs/states';
 import {
-  appendEvent, ledgerClient, loadPlanSteps, loadRun, updateRequest, updateRun, updateStep, type StepRow,
+  appendEvent, freshBudget, ledgerClient, loadPlanSteps, loadRun, updateRequest, updateRun, updateRunAsObserved, updateStep, type StepRow,
 } from '@/lib/ai/runs/store';
 import {
   availableWriteBackKinds, reminderLeadAt, writeBackTitle, type WriteBackKind,
@@ -462,15 +462,29 @@ async function foldIntoRun(
     return ok({ resumedRunId: null, runId, released: gated.length });
   }
 
-  const resumed = await updateRun(scope, runId, {
+  // A person decided, so the run gets its attempt budget back along with its
+  // place in the queue: at the ceiling the kick below would refuse it and the
+  // next tick would abandon it, decision and all (controls.ts, freshBudget).
+  //
+  // Conditioned on the run AS THIS FOLD LOADED IT. Two final decisions can fold
+  // into one run together; the first puts it in the queue and a worker claims
+  // it, and the second — still holding the snapshot that said
+  // `awaiting_approval` — must not write `ready`, a zero attempt and a cleared
+  // lease over that worker's live claim. Zero rows means the other fold (or
+  // the worker) got there first: the steps this decision released are already
+  // in the graph the running slice reloads every pass, so there is nothing
+  // left for this fold to do and no kick to send.
+  const resumed = await updateRunAsObserved(scope, runId, run.data, {
     state: 'ready',
     status: legacyStatusFor('ready'),
     run_after: nowIso,
     error: null,
     lease_owner: null,
     lease_expires_at: null,
+    ...freshBudget(),
   }, { db });
   if (!resumed.ok) return resumed;
+  if (!resumed.data) return ok({ resumedRunId: null, runId, released: gated.length });
   if (run.data.request_id) await updateRequest(scope, run.data.request_id, { status: 'ready', error: null }, { db });
   kickRun(runId);
   return ok({ resumedRunId: runId, runId, released: gated.length });

tests/a-run-a-person-asks-for-again-gets-its-attempts-back.test.ts

@@ -0,0 +1,151 @@
+// A run a person asks for again gets its attempt budget back (finalaudit Q40,
+// follow-up to its second half).
+//
+// `attempt` is the run's abandonment budget: the slices in a row that completed
+// nothing. At `max_attempts` the recovery arm of `claim_ai_runs` dead-letters
+// the run and `claimRun` refuses to lease it. Four writes put a run back to
+// `ready` because a member of the family asked: resume, re-run a step, edit a
+// step (lib/ai/runs/controls.ts) and an approval decision
+// (lib/services/approvals). Every one of them told the person Bubaly would pick
+// the run up, and not one touched `attempt`. For a run that had spent its
+// budget that promise was false twice over: the kick that follows each write
+// goes through `claimRun`, which answered `claimed: false` without a word, and
+// the next cron tick's recovery arm abandoned the run again. "Try that step
+// again" on an abandoned run produced a run abandoned again, with nothing a
+// person could do about it; an approval granted to a run parked at the ceiling
+// was consumed by a run that then died.
+//
+// `freshBudget()` (lib/ai/runs/store.ts) is the `attempt: 0` each of those
+// writes now carries: a human asking again is a new budget, the way a slice
+// that made progress is (executor.ts, budgetReset). This file drives the real
+// controls, the real `claimRun` and the claim pass's predicate-for-predicate
+// stand-in against the in-memory client; the control case is the same row
+// WITHOUT a person's re-entry, which is what the person's run met before.
+import { readFileSync } from 'node:fs';
+import { join } from 'node:path';
+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 { ServiceScope } from '@/lib/services/types';
+import { rerunStep, resumeRun } from '@/lib/ai/runs/controls';
+import { claimRun } from '@/lib/ai/runs/store';
+import { createInMemorySupabase, type InMemorySupabase, type Row } from './helpers/in-memory-supabase';
+import { ABANDONED, claimAiRuns } from './helpers/claim-ai-runs';
+
+const ROOT = join(__dirname, '..');
+const FAMILY = '00000000-0000-4000-8000-00000000f0a1';
+const MEMBER = '00000000-0000-4000-8000-00000000f0a2';
+const RUN = '00000000-0000-4000-8000-00000000f0b1';
+const PLAN = '00000000-0000-4000-8000-00000000f0c1';
+const STEP = '00000000-0000-4000-8000-00000000f0d1';
+const REQUEST = '00000000-0000-4000-8000-00000000f0e1';
+const CEILING = 5;
+
+let db: InMemorySupabase;
+const client = () => db as unknown as SupabaseClient<Database>;
+const run = () => db.table('family_automation_runs').find((r) => r.id === RUN) as Row;
+
+const scope = (): ServiceScope => ({
+  db: client(), familyId: FAMILY, userId: 'auth-parent', memberId: MEMBER, role: 'parent', actorKind: 'member', tz: 'UTC',
+} as unknown as ServiceScope);
+
+/** A run whose budget is spent, in the state the recovery arm leaves it. */
+function seedAbandoned(over: Row = {}) {
+  db.seed('family_automation_runs', [{
+    id: RUN, family_id: FAMILY, run_type: 'concierge_plan', plan_id: PLAN, request_id: REQUEST, requested_by_member_id: MEMBER,
+    state: 'failed', status: 'failed', error: ABANDONED, progress: {},
+    attempt: CEILING, max_attempts: CEILING, lease_owner: null, lease_expires_at: null, cancel_requested_at: null, paused_at: null,
+    started_at: '2026-10-03T11:00:00.000Z', completed_at: '2026-10-03T11:30:00.000Z', run_after: '2026-10-03T11:30:00.000Z',
+    ...over,
+  }]);
+  db.seed('ai_requests', [{ id: REQUEST, family_id: FAMILY, status: 'failed', error: ABANDONED }]);
+  db.seed('ai_plan_steps', [{
+    id: STEP, family_id: FAMILY, plan_id: PLAN, sequence: 0, step_type: 'act', tool_name: 'calendar.createEvent',
+    description: 'Add the dentist', input_json: { title: 'Dentist' }, dependency_ids: [], condition: null,
+    status: 'failed', approval_required: false, approval_id: null, risk_level: 'low', retry_count: 2, max_retries: 2,
+    result_json: null, error: 'The calendar service timed out.', started_at: null, completed_at: null,
+  }]);
+  db.seed('ai_tool_calls', [{
+    id: '00000000-0000-4000-8000-00000000f0f1', family_id: FAMILY, run_id: RUN, plan_step_id: STEP, tool_name: 'calendar.createEvent',
+    state: 'failed', attempt: 1, resource_id: null, resource_table: null, created_at: '2026-10-03T11:20:00.000Z',
+  }]);
+}
+
+beforeEach(() => {
+  db = createInMemorySupabase({ rpc: { claim_ai_runs: (args, c) => claimAiRuns(args, c) } });
+  vi.spyOn(console, 'error').mockImplementation(() => undefined);
+});
+afterEach(() => { vi.restoreAllMocks(); });
+
+describe('a run a person asks for again', () => {
+  it('re-run a step on an abandoned run: the kick claims it and the next tick takes it, instead of abandoning it again', async () => {
+    seedAbandoned();
+
+    const asked = await rerunStep(scope(), RUN, STEP, { db: client() });
+    expect(asked).toMatchObject({ ok: true, data: { rerun: true } });
+    expect(run()).toMatchObject({ state: 'ready', attempt: 0, error: null, completed_at: null });
+
+    // The kick after the control (intake.ts → continueRun → claimRun).
+    const kick = await claimRun(client(), RUN);
+    expect(kick).toMatchObject({ ok: true, data: { claimed: true } });
+    expect(run()).toMatchObject({ state: 'executing', attempt: 1 });
+    expect(run().lease_owner).toBeTruthy();
+  });
+
+  it("the cron's claim pass, when the kick did not get there first, takes it rather than dead-lettering it", () => {
+    seedAbandoned();
+    return rerunStep(scope(), RUN, STEP, { db: client() }).then((asked) => {
+      expect(asked.ok).toBe(true);
+      expect(claimAiRuns({ p_limit: 10 }, db)).toEqual([RUN]);
+      expect(run()).toMatchObject({ state: 'executing', attempt: 1, error: null });
+    });
+  });
+
+  it('resume a run paused at the ceiling: claimable again', async () => {
+    seedAbandoned({ state: 'paused', status: 'paused', error: null, completed_at: null, paused_at: '2026-10-03T11:40:00.000Z' });
+
+    const asked = await resumeRun(scope(), RUN, { db: client() });
+    expect(asked).toMatchObject({ ok: true, data: { state: 'ready' } });
+    expect(run()).toMatchObject({ state: 'ready', attempt: 0, paused_at: null });
+
+    const kick = await claimRun(client(), RUN);
+    expect(kick).toMatchObject({ ok: true, data: { claimed: true } });
+    expect(run()).toMatchObject({ state: 'executing', attempt: 1 });
+  });
+
+  it('control: the same row at the ceiling without a person asking is refused by the kick and abandoned by the tick', async () => {
+    // What every re-entry produced before: `ready`, due, budget spent.
+    seedAbandoned({ state: 'ready', status: 'approved', error: null, completed_at: null, run_after: '2026-10-03T11:30:00.000Z' });
+
+    const kick = await claimRun(client(), RUN);
+    expect(kick).toMatchObject({ ok: true, data: { claimed: false } });
+    expect(run()).toMatchObject({ state: 'ready', attempt: CEILING });
+
+    expect(claimAiRuns({ p_limit: 10 }, db)).toEqual([]);
+    expect(run()).toMatchObject({ state: 'failed', status: 'failed', error: ABANDONED, attempt: CEILING });
+  });
+
+  it('every write that puts a run back to ready because a person asked carries the fresh budget', () => {
+    const controls = readFileSync(join(ROOT, 'lib/ai/runs/controls.ts'), 'utf8');
+    const approvals = readFileSync(join(ROOT, 'lib/services/approvals/index.ts'), 'utf8');
+
+    // Each `updateRunAsObserved(... { state: 'ready', ... })` patch, split at
+    // the call so a patch is checked against its own fields and nobody else's.
+    // The observed compare-and-set, not the plain `updateRun`: a re-entry
+    // writes from the snapshot it loaded, and two of them (or one and a claim)
+    // can hold the same snapshot (review 5973301004;
+    // tests/a-stale-fold-cannot-re-arm-a-claimed-run.test.ts).
+    const readyPatches = (src: string) => src.split('updateRunAsObserved(').slice(1).filter((chunk) => /state: 'ready',/.test(chunk.split('}, { db })')[0]));
+    const controlPatches = readyPatches(controls);
+    const approvalPatches = readyPatches(approvals);
+    expect(controlPatches, 'resume, re-run a step, edit a step').toHaveLength(3);
+    expect(approvalPatches, 'the decision that returns a run to the queue').toHaveLength(1);
+    for (const patch of [...controlPatches, ...approvalPatches]) {
+      expect(patch.split('}, { db })')[0]).toContain('...freshBudget()');
+    }
+    // And no re-entry write to `ready` is left on the unconditioned path.
+    const bareReady = (src: string) => src.split('updateRun(').slice(1).filter((chunk) => /state: 'ready',/.test(chunk.split('}, { db })')[0]));
+    expect(bareReady(controls), 'controls: every ready write is conditioned').toHaveLength(0);
+    expect(bareReady(approvals), 'approvals: every ready write is conditioned').toHaveLength(0);
+  });
+});