From f44aa96a90f93672031833269360e7fa60c31e28 Mon Sep 17 00:00:00 2001 From: Quay Worker Date: Tue, 14 Jul 2026 13:03:41 +0000 Subject: [PATCH 1/2] feat(cli): add task budget increase command (BRIX-1929) --- README.md | 2 + docs/quay-spec.md | 1 + docs/user/cli-reference.md | 13 + docs/user/troubleshooting.md | 7 + packages/cli/src/cli/dispatch.ts | 83 ++++++ packages/cli/src/cli/help.ts | 16 ++ packages/cli/src/core/task_budget.ts | 235 +++++++++++++++++ packages/cli/tests/task/task_budget.test.ts | 276 ++++++++++++++++++++ 8 files changed, 633 insertions(+) create mode 100644 packages/cli/src/core/task_budget.ts create mode 100644 packages/cli/tests/task/task_budget.test.ts diff --git a/README.md b/README.md index af9fefd0..3b015b04 100644 --- a/README.md +++ b/README.md @@ -212,6 +212,8 @@ quay outbox complete --claim-id quay outbox fail --claim-id --error # shared side-effect outbox quay task get | task list # read commands (deterministic JSON) +quay task increase-budget --by --reason + # operator recovery for burned retry budget quay submit-brief | escalate-human | record-human-reply | cancel quay artifact get # raw bytes to stdout ``` diff --git a/docs/quay-spec.md b/docs/quay-spec.md index acc65f18..32921866 100644 --- a/docs/quay-spec.md +++ b/docs/quay-spec.md @@ -1095,6 +1095,7 @@ No `--json` flag exists or is needed; JSON is the only output format. The pull-l | `quay task claim ` | Atomically transition `awaiting-next-brief` → `claimed-by-orchestrator` and **mint a fresh `claim_id`** (UUID v4). Predicate: `state = 'awaiting-next-brief' AND cancel_requested_at IS NULL`. Returns `{"task_id": "...", "claim_id": ""}` on stdout. The orchestrator MUST store `claim_id` and pass it back as `--claim-id ` on every subsequent claim-scoped write. Errors if the task is not in `awaiting-next-brief` or has been cancelled. | Orchestrator | | `quay task release-claim --claim-id ` | Ownership-fenced release: predicate is `state IN ('claimed-by-orchestrator', 'waiting_human') AND claim_id = ?`. On match: clears `claim_id` and transitions to `awaiting-next-brief`. On mismatch: errors with `claim_lost` (the prior claim was timed out and re-claimed by someone else; the caller must abandon and re-claim). Releasing a task already in `awaiting-next-brief` is an idempotent no-op success (no claim_id check needed for that case). | Orchestrator | | `quay task retarget --repo [--base-branch ] --yes` | Clone an existing non-terminal task into another repo. The clone is a fresh `queued` task that reuses the source task's stable objective, ticket snapshot, tags, author metadata, agent/model overrides, screenshot settings, worker execution mode, external ref, Slack thread ref, and retry budget. It gets a new task id, branch, worktree, attempt #1, and task artifacts exactly like `enqueue`, with `tasks.retargeted_from_task_id` pointing at the source. The source task transitions to `cancelled` through a `retargeted` audit event; that event's `event_data` records the target repo, target task id, branch, worktree, and base branch override. If retarget crashes after the clone and cancel intent are durable but before the source terminal transition, the next tick's cancel finalizer recovers the source as `cancelled` and emits the source-side `retargeted` event data from the linked clone. `--yes` is required because the source task is mutated and a running worker session is killed before source worktree/branch cleanup. | Operator or orchestrator | +| `quay task increase-budget (--by \|--set ) --reason [--force]` | Raise an existing task's copied `retry_budget` for operator recovery when control-plane or substrate failures consumed retry budget without useful worker progress. Requires a human-readable reason, records a `task_budget_adjusted` audit event, and recomputes `budget_exhausted` from `attempts_consumed >= retry_budget`. By default it is limited to parked/recovery states (`awaiting-next-brief`, `claimed-by-orchestrator`, `waiting_human`, `non_budget_loop`, `orchestrator_loop`, `worktree_error`); live non-terminal states require `--force`; terminal tasks are refused. | Operator | | `quay submit-brief --claim-id --brief-file --reason ` | Submit a new brief. **Requires `--claim-id` matching the live `tasks.claim_id`** (ownership fence) and `cancel_requested_at IS NULL`. On `claim_id` mismatch: errors with `claim_lost`. On `cancel_requested_at IS NOT NULL`: errors with `cancelled`. On wrong state otherwise: `wrong_state`. On success: persists the brief as an artifact, transitions the task to `queued`, clears `claim_id`, resets `claim_expirations_consecutive` to 0. Tick promotes `queued → running` on its next cycle when capacity allows; budget is consumed at that promotion (unless reason is `advice_answered`). Errors with `budget_exhausted` if `task.budget_exhausted = true` and reason is `blocker_resolved`; orchestrator must use `escalate-human` or `cancel` instead. **Note:** there is no `--reason initial`; the initial brief is supplied at `enqueue` and `attempts.reason = 'initial'` is set internally by Quay on the first spawn. | Orchestrator | | `quay escalate-human --claim-id --question-file [--thread-ref ]` | Record a human question and transition to `waiting_human`. **Requires `--claim-id` matching the live `tasks.claim_id`** (ownership fence) and `cancel_requested_at IS NULL`; same error taxonomy as `submit-brief`. **The CLI does NOT post to Slack** — it persists the `slack_escalation_post` artifact (with `escalation_seq`, `escalation_nonce`, `content_hash` set) and preserves the claim so the orchestrator owns routing, posting, waiting, and reply handling. `--thread-ref` is optional metadata; missing thread refs are allowed so deployments can use fallback routing outside Quay. When provided, it follows the same storage contract as enqueue: canonical `CHANNEL:THREAD_TS` is stored as-is, legacy `slack:CHANNEL:THREAD_TS` is normalized, and malformed refs fail before the task row is updated. Does not consume budget. | Orchestrator | | `quay record-human-reply --claim-id --reply-file [--thread-ref ] [--message-ts ] [--author ]` | Persist a human answer as a `slack_reply` artifact and transition `waiting_human` → `claimed-by-orchestrator` under the same ownership fence. The orchestrator then calls `submit-brief --reason advice_answered`, which completes the original handoff without a second Quay-created handoff round-trip. | Orchestrator | diff --git a/docs/user/cli-reference.md b/docs/user/cli-reference.md index a3b63cca..eae7737f 100644 --- a/docs/user/cli-reference.md +++ b/docs/user/cli-reference.md @@ -428,6 +428,7 @@ quay task release-claim --claim-id quay task retarget --repo [--base-branch ] --yes quay task resnapshot --reason quay task recreate-worktree --yes [--force] +quay task increase-budget (--by |--set ) --reason [--force] ``` `task claim` only succeeds for `awaiting-next-brief` tasks. @@ -446,6 +447,18 @@ branch name. The repo install command runs after recreation, and Quay records a confirming no worker is live; it allows recreation when the path already exists or an active attempt is still recorded. +`task increase-budget` raises an existing task's copied `retry_budget` after an +operator has confirmed that control-plane or substrate failures consumed retry +budget without useful worker progress. It requires a human-readable `--reason`, +records a `task_budget_adjusted` audit event, and recomputes +`budget_exhausted` from `attempts_consumed >= retry_budget`. Use `--by ` to +add attempts or `--set ` to set a higher absolute budget. By default the +command is limited to parked/recovery states such as `awaiting-next-brief`, +`waiting_human`, `claimed-by-orchestrator`, `non_budget_loop`, +`orchestrator_loop`, and `worktree_error`; use `--force` only after confirming a +live non-terminal task should receive more retry budget. Terminal tasks cannot +be adjusted. + `task list`, `task get`, and `task events` accept legacy `task_id` values and return the same task/run rows as before. JSON output now also includes run-aware compatibility fields: diff --git a/docs/user/troubleshooting.md b/docs/user/troubleshooting.md index 32ce325a..df7f6815 100644 --- a/docs/user/troubleshooting.md +++ b/docs/user/troubleshooting.md @@ -168,10 +168,17 @@ is rejected in this state. Use one of: ```bash +quay task increase-budget --by 2 --reason "repaired missing worktree; previous retries did not spawn a worker" quay escalate-human --claim-id --question-file ./question.md quay cancel ``` +Use `task increase-budget` when an operator has fixed a control-plane or +substrate issue that consumed retry budget without useful worker progress, for +example a deleted worktree that caused repeated spawn/no-progress failures. The +command records the reason in a `task_budget_adjusted` audit event and clears +`budget_exhausted` only when the raised budget is above `attempts_consumed`. + After asking a human, use `record-human-reply` to persist the answer, then `submit-brief --reason advice_answered`; that reason is allowed because it does not consume retry budget. diff --git a/packages/cli/src/cli/dispatch.ts b/packages/cli/src/cli/dispatch.ts index f6db6142..3ef57ffc 100644 --- a/packages/cli/src/cli/dispatch.ts +++ b/packages/cli/src/cli/dispatch.ts @@ -65,6 +65,10 @@ import { recreate_task_worktree, type RecreateWorktreeDeps, } from "../core/recreate_worktree.ts"; +import { + adjust_task_budget, + type AdjustTaskBudgetDeps, +} from "../core/task_budget.ts"; import { claim_task, release_claim, @@ -744,6 +748,9 @@ async function handleTask( case "recreate-worktree": if (wantsHelp(rest)) return printHelp(io, ["task", "recreate-worktree"]); return await handleTaskRecreateWorktree(rest, deps, io); + case "increase-budget": + if (wantsHelp(rest)) return printHelp(io, ["task", "increase-budget"]); + return await handleTaskIncreaseBudget(rest, deps, io); default: // A typo'd subcommand benefits from the noun's usage block as much as // a missing one — surface it on stderr alongside the structured envelope. @@ -756,6 +763,82 @@ async function handleTask( } } +async function handleTaskIncreaseBudget( + argv: string[], + deps: CliDeps, + io: CliIO, +): Promise { + const validation = validateFlags(argv, { + boolean: ["--force"], + valued: ["--by", "--set", "--reason"], + }); + if (!validation.ok) { + return writeError(io, "usage_error", validation.message, validation.details); + } + const taskId = positional(argv); + if (!taskId) { + return writeError( + io, + "usage_error", + "task increase-budget requires ", + ); + } + const reason = readFlag(argv, "--reason"); + if (reason === null) { + return writeError( + io, + "usage_error", + "task increase-budget requires --reason ", + ); + } + const byRaw = readFlag(argv, "--by"); + const setRaw = readFlag(argv, "--set"); + if ( + (byRaw === null && setRaw === null) || + (byRaw !== null && setRaw !== null) + ) { + return writeError( + io, + "usage_error", + "task increase-budget requires exactly one of --by or --set", + ); + } + const by = byRaw === null + ? undefined + : parsePositiveIntArg(byRaw, "task increase-budget", "by"); + if (by !== undefined && !by.ok) { + return writeError(io, "usage_error", by.message); + } + const set = setRaw === null + ? undefined + : parsePositiveIntArg(setRaw, "task increase-budget", "set"); + if (set !== undefined && !set.ok) { + return writeError(io, "usage_error", set.message); + } + const budgetDeps: AdjustTaskBudgetDeps = { + db: deps.db, + clock: deps.clock, + supervisorLock: deps.supervisorLock, + }; + const input: { + taskId: string; + by?: number; + set?: number; + reason: string; + force: boolean; + } = { + taskId, + reason, + force: argv.includes("--force"), + }; + if (by !== undefined) input.by = by.value; + if (set !== undefined) input.set = set.value; + return emitServiceResult( + await adjust_task_budget(budgetDeps, input), + io, + ); +} + async function handleTaskRecreateWorktree( argv: string[], deps: CliDeps, diff --git a/packages/cli/src/cli/help.ts b/packages/cli/src/cli/help.ts index d6e2e01d..ee5023aa 100644 --- a/packages/cli/src/cli/help.ts +++ b/packages/cli/src/cli/help.ts @@ -53,6 +53,7 @@ const COMMANDS: Record = { "task retarget", "task resnapshot", "task recreate-worktree", + "task increase-budget", ], }, "task list": { @@ -122,6 +123,21 @@ const COMMANDS: Record = { { flag: "--force", desc: "Allow recreation when the path exists or an active attempt is recorded." }, ], }, + "task increase-budget": { + path: "task increase-budget", + synopsis: + "quay task increase-budget (--by |--set ) --reason [--force]", + summary: + "Raise a task's retry budget and recompute its budget_exhausted flag.", + details: + "Use this for operator recovery when control-plane or substrate failures consumed retry budget without useful worker progress. By default it is limited to parked/recovery states; --force is required for live non-terminal states. Terminal tasks cannot be adjusted.", + flags: [ + { flag: "--by ", desc: "Increase retry_budget by a positive integer." }, + { flag: "--set ", desc: "Set retry_budget to a higher positive integer." }, + { flag: "--reason ", desc: "Required human-readable audit reason." }, + { flag: "--force", desc: "Allow adjustment for live non-terminal states." }, + ], + }, tick: { path: "tick", synopsis: "quay tick", diff --git a/packages/cli/src/core/task_budget.ts b/packages/cli/src/core/task_budget.ts new file mode 100644 index 00000000..8ef05244 --- /dev/null +++ b/packages/cli/src/core/task_budget.ts @@ -0,0 +1,235 @@ +import type { DB } from "../db/connection.ts"; +import type { Clock } from "../ports/clock.ts"; +import type { SupervisorLock } from "./supervisor_lock.ts"; +import { TASK_TERMINAL_STATES } from "./task_state.ts"; + +export type AdjustTaskBudgetErrorCode = + | "unknown_task" + | "validation_error" + | "unsafe_state"; + +export interface AdjustTaskBudgetError { + code: AdjustTaskBudgetErrorCode; + message: string; + details?: Record; +} + +export type AdjustTaskBudgetResult = + | { ok: true; value: AdjustTaskBudgetValue } + | { ok: false; error: AdjustTaskBudgetError }; + +export interface AdjustTaskBudgetValue { + task_id: string; + state: string; + attempts_consumed: number; + previous_retry_budget: number; + retry_budget: number; + previous_budget_exhausted: boolean; + budget_exhausted: boolean; + reason: string; + forced: boolean; + event_id: number; +} + +export interface AdjustTaskBudgetDeps { + db: DB; + clock: Clock; + supervisorLock: SupervisorLock; +} + +export interface AdjustTaskBudgetInput { + taskId: string; + by?: number; + set?: number; + reason: string; + force?: boolean; +} + +interface TaskBudgetRow { + task_id: string; + state: string; + attempts_consumed: number; + retry_budget: number; + budget_exhausted: number; +} + +const SAFE_STATES = new Set([ + "awaiting-next-brief", + "claimed-by-orchestrator", + "waiting_human", + "non_budget_loop", + "orchestrator_loop", + "worktree_error", +]); + +const TERMINAL_STATES = new Set(TASK_TERMINAL_STATES); + +export async function adjust_task_budget( + deps: AdjustTaskBudgetDeps, + input: AdjustTaskBudgetInput, +): Promise { + return deps.supervisorLock.run(() => adjustUnderLock(deps, input)); +} + +function adjustUnderLock( + deps: AdjustTaskBudgetDeps, + input: AdjustTaskBudgetInput, +): AdjustTaskBudgetResult { + const task = loadTask(deps.db, input.taskId); + if (task === null) { + return { + ok: false, + error: { + code: "unknown_task", + message: `task ${input.taskId} not found`, + details: { task_id: input.taskId }, + }, + }; + } + + const reason = input.reason.trim(); + if (reason.length === 0) { + return { + ok: false, + error: { + code: "validation_error", + message: "task increase-budget requires --reason ", + details: { task_id: task.task_id }, + }, + }; + } + + if ( + (input.by === undefined && input.set === undefined) || + (input.by !== undefined && input.set !== undefined) + ) { + return { + ok: false, + error: { + code: "validation_error", + message: "task increase-budget requires exactly one of --by or --set", + details: { task_id: task.task_id }, + }, + }; + } + + if (TERMINAL_STATES.has(task.state)) { + return { + ok: false, + error: { + code: "unsafe_state", + message: `task ${task.task_id} is terminal (${task.state}); budget cannot be adjusted`, + details: { task_id: task.task_id, state: task.state }, + }, + }; + } + + if (!SAFE_STATES.has(task.state) && input.force !== true) { + return { + ok: false, + error: { + code: "unsafe_state", + message: + `task ${task.task_id} is ${task.state}; rerun with --force only after confirming this live state should receive more budget`, + details: { task_id: task.task_id, state: task.state }, + }, + }; + } + + const nextBudget = + input.set !== undefined + ? input.set + : task.retry_budget + (input.by as number); + if (!Number.isInteger(nextBudget) || nextBudget <= 0) { + return { + ok: false, + error: { + code: "validation_error", + message: "retry budget must be a positive integer", + details: { task_id: task.task_id, retry_budget: nextBudget }, + }, + }; + } + if (nextBudget <= task.retry_budget) { + return { + ok: false, + error: { + code: "validation_error", + message: + `task increase-budget must raise retry_budget above ${task.retry_budget}`, + details: { + task_id: task.task_id, + retry_budget: task.retry_budget, + requested_retry_budget: nextBudget, + }, + }, + }; + } + + const now = deps.clock.nowISO(); + const nextBudgetExhausted = task.attempts_consumed >= nextBudget ? 1 : 0; + const eventData = { + reason, + forced: input.force === true, + by: input.by ?? null, + set: input.set ?? null, + attempts_consumed: task.attempts_consumed, + previous_retry_budget: task.retry_budget, + retry_budget: nextBudget, + previous_budget_exhausted: task.budget_exhausted === 1, + budget_exhausted: nextBudgetExhausted === 1, + }; + const event = deps.db.transaction(() => { + deps.db + .query( + `UPDATE tasks + SET retry_budget = ?, + budget_exhausted = ?, + updated_at = ? + WHERE task_id = ?`, + ) + .run(nextBudget, nextBudgetExhausted, now, task.task_id); + + return deps.db + .query<{ event_id: number }, [string, string, string, string, string]>( + `INSERT INTO events ( + task_id, event_type, from_state, to_state, occurred_at, event_data + ) VALUES (?, 'task_budget_adjusted', ?, ?, ?, ?) + RETURNING event_id`, + ) + .get( + task.task_id, + task.state, + task.state, + now, + JSON.stringify(eventData), + ); + })(); + if (!event) throw new Error("task_budget_adjusted event insert returned no row"); + + return { + ok: true, + value: { + task_id: task.task_id, + state: task.state, + attempts_consumed: task.attempts_consumed, + previous_retry_budget: task.retry_budget, + retry_budget: nextBudget, + previous_budget_exhausted: task.budget_exhausted === 1, + budget_exhausted: nextBudgetExhausted === 1, + reason, + forced: input.force === true, + event_id: event.event_id, + }, + }; +} + +function loadTask(db: DB, taskId: string): TaskBudgetRow | null { + return db + .query( + `SELECT task_id, state, attempts_consumed, retry_budget, budget_exhausted + FROM tasks + WHERE task_id = ?`, + ) + .get(taskId) ?? null; +} diff --git a/packages/cli/tests/task/task_budget.test.ts b/packages/cli/tests/task/task_budget.test.ts new file mode 100644 index 00000000..1dc53f04 --- /dev/null +++ b/packages/cli/tests/task/task_budget.test.ts @@ -0,0 +1,276 @@ +import { afterEach, expect, test } from "bun:test"; +import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { dispatch } from "../../src/cli/dispatch.ts"; +import { bufferIO } from "../../src/cli/io.ts"; +import { adjust_task_budget } from "../../src/core/task_budget.ts"; +import { createHarness, type Harness } from "../support/harness.ts"; +import { buildCliDeps, type BuiltCliDeps } from "../support/cli_deps.ts"; +import { + insertAttempt, + insertRepo, + insertTask, + seedTaskObjective, +} from "../support/fixtures.ts"; + +let h: Harness | null = null; +let scratchDirs: string[] = []; +afterEach(() => { + h?.cleanup(); + h = null; + for (const d of scratchDirs.splice(0)) { + try { + rmSync(d, { recursive: true, force: true }); + } catch {} + } +}); + +function writeTemp(contents: string, name = "brief.md"): string { + const dir = mkdtempSync(join(tmpdir(), "quay-budget-test-")); + scratchDirs.push(dir); + const p = join(dir, name); + writeFileSync(p, contents); + return p; +} + +function setupTask(state = "claimed-by-orchestrator"): { + h: Harness; + built: BuiltCliDeps; + taskId: string; +} { + const harness = createHarness(); + const built = buildCliDeps(harness); + const repoId = insertRepo(harness.db, "repo-budget"); + const taskId = insertTask(harness.db, { + taskId: "task-budget", + repoId, + state, + }); + seedTaskObjective(harness, taskId); + insertAttempt(harness.db, { + taskId, + attemptNumber: 1, + reason: "initial", + consumedBudget: 1, + spawnedAt: "2026-01-01T00:00:00.000Z", + }); + harness.db + .query( + `UPDATE tasks + SET attempts_consumed = 5, + retry_budget = 5, + budget_exhausted = 1, + claim_id = CASE WHEN state = 'claimed-by-orchestrator' THEN 'claim-budget' ELSE claim_id END, + claimed_at = CASE WHEN state = 'claimed-by-orchestrator' THEN ? ELSE claimed_at END + WHERE task_id = ?`, + ) + .run(harness.clock.nowISO(), taskId); + h = harness; + return { h: harness, built, taskId }; +} + +test("adjust_task_budget raises budget, clears exhaustion, and records audit reason", async () => { + const { h, built, taskId } = setupTask(); + + const result = await adjust_task_budget( + { + db: h.db, + clock: h.clock, + supervisorLock: built.deps.supervisorLock, + }, + { + taskId, + by: 2, + reason: "missing worktree burned retries without worker progress", + }, + ); + + expect(result).toMatchObject({ + ok: true, + value: { + task_id: taskId, + attempts_consumed: 5, + previous_retry_budget: 5, + retry_budget: 7, + previous_budget_exhausted: true, + budget_exhausted: false, + reason: "missing worktree burned retries without worker progress", + forced: false, + }, + }); + const task = h.db + .query< + { retry_budget: number; budget_exhausted: number }, + [string] + >(`SELECT retry_budget, budget_exhausted FROM tasks WHERE task_id = ?`) + .get(taskId); + expect(task).toEqual({ retry_budget: 7, budget_exhausted: 0 }); + + const event = h.db + .query< + { event_type: string; from_state: string; to_state: string; event_data: string }, + [string] + >( + `SELECT event_type, from_state, to_state, event_data + FROM events + WHERE task_id = ? AND event_type = 'task_budget_adjusted'`, + ) + .get(taskId); + expect(event).toMatchObject({ + event_type: "task_budget_adjusted", + from_state: "claimed-by-orchestrator", + to_state: "claimed-by-orchestrator", + }); + expect(JSON.parse(event!.event_data)).toMatchObject({ + reason: "missing worktree burned retries without worker progress", + previous_retry_budget: 5, + retry_budget: 7, + previous_budget_exhausted: true, + budget_exhausted: false, + }); +}); + +test("task increase-budget CLI lets blocker_resolved resume after budget is raised", async () => { + const { h, built, taskId } = setupTask(); + + const increaseIo = bufferIO(); + const increase = await dispatch( + [ + "task", + "increase-budget", + taskId, + "--set", + "8", + "--reason", + "operator repaired deleted worktree and is allowing real retries", + ], + built.deps, + increaseIo, + ); + expect(increase.exitCode).toBe(0); + expect(increaseIo.err()).toBe(""); + expect(JSON.parse(increaseIo.out())).toMatchObject({ + task_id: taskId, + retry_budget: 8, + budget_exhausted: false, + }); + + const submitIo = bufferIO(); + const submit = await dispatch( + [ + "submit-brief", + taskId, + "--claim-id", + "claim-budget", + "--brief-file", + writeTemp("Retry after budget repair."), + "--reason", + "blocker_resolved", + ], + built.deps, + submitIo, + ); + + expect(submit.exitCode).toBe(0); + expect(submitIo.err()).toBe(""); + expect(JSON.parse(submitIo.out())).toMatchObject({ + state: "queued", + }); + const task = h.db + .query<{ state: string; budget_exhausted: number }, [string]>( + `SELECT state, budget_exhausted FROM tasks WHERE task_id = ?`, + ) + .get(taskId); + expect(task).toEqual({ state: "queued", budget_exhausted: 0 }); +}); + +test("task increase-budget guards live states unless forced", async () => { + const { built, taskId } = setupTask("queued"); + + const refusedIo = bufferIO(); + const refused = await dispatch( + [ + "task", + "increase-budget", + taskId, + "--by", + "1", + "--reason", + "testing guardrail", + ], + built.deps, + refusedIo, + ); + expect(refused.exitCode).toBe(1); + expect(JSON.parse(refusedIo.err())).toMatchObject({ + error: "unsafe_state", + }); + + const forcedIo = bufferIO(); + const forced = await dispatch( + [ + "task", + "increase-budget", + taskId, + "--by", + "1", + "--reason", + "operator confirmed the queued task should receive another retry", + "--force", + ], + built.deps, + forcedIo, + ); + expect(forced.exitCode).toBe(0); + expect(JSON.parse(forcedIo.out())).toMatchObject({ + retry_budget: 6, + forced: true, + }); +}); + +test("task increase-budget refuses terminal tasks", async () => { + const { built, taskId } = setupTask("cancelled"); + + const terminalIo = bufferIO(); + const terminal = await dispatch( + [ + "task", + "increase-budget", + taskId, + "--by", + "1", + "--reason", + "should not reopen terminal task", + ], + built.deps, + terminalIo, + ); + expect(terminal.exitCode).toBe(1); + expect(JSON.parse(terminalIo.err())).toMatchObject({ + error: "unsafe_state", + }); +}); + +test("task increase-budget refuses non-increasing set values", async () => { + const { built, taskId } = setupTask("awaiting-next-brief"); + + const nonIncreasingIo = bufferIO(); + const nonIncreasing = await dispatch( + [ + "task", + "increase-budget", + taskId, + "--set", + "5", + "--reason", + "same value is not an increase", + ], + built.deps, + nonIncreasingIo, + ); + expect(nonIncreasing.exitCode).toBe(1); + expect(JSON.parse(nonIncreasingIo.err())).toMatchObject({ + error: "validation_error", + }); +}); From 172a0c3bfbb935ab73854b03249a1eb99a8f86e3 Mon Sep 17 00:00:00 2001 From: Fabian Scherer Date: Tue, 14 Jul 2026 16:22:23 +0200 Subject: [PATCH 2/2] Fence task budget adjustments --- docs/quay-spec.md | 2 +- packages/cli/src/core/task_budget.ts | 39 +++++++++++--- packages/cli/tests/task/task_budget.test.ts | 60 +++++++++++++++++++++ 3 files changed, 93 insertions(+), 8 deletions(-) diff --git a/docs/quay-spec.md b/docs/quay-spec.md index 32921866..17c09d0f 100644 --- a/docs/quay-spec.md +++ b/docs/quay-spec.md @@ -928,7 +928,7 @@ CREATE TABLE events ( event_id INTEGER PRIMARY KEY AUTOINCREMENT, task_id TEXT NOT NULL REFERENCES tasks(task_id), attempt_id INTEGER REFERENCES attempts(attempt_id), - event_type TEXT NOT NULL, -- spawned / pr_opened / ci_failed / ci_passed / merged / closed / changes_requested / conflict / blocker_ingested / malformed_signal_ingested / no_progress / crashed / spawn_failed / worker_auth_invalid / stale_detected / stale_killed / wall_clock_exceeded / wall_clock_killed / slack_reply_ingested / claimed / claim_expired / orchestrator_loop_parked / non_budget_loop_parked / brief_submitted / human_escalated / budget_exhausted / cancelled / worktree_error / tick_error + event_type TEXT NOT NULL, -- spawned / pr_opened / ci_failed / ci_passed / merged / closed / changes_requested / conflict / blocker_ingested / malformed_signal_ingested / no_progress / crashed / spawn_failed / worker_auth_invalid / stale_detected / stale_killed / wall_clock_exceeded / wall_clock_killed / slack_reply_ingested / claimed / claim_expired / orchestrator_loop_parked / non_budget_loop_parked / brief_submitted / human_escalated / budget_exhausted / task_budget_adjusted / cancelled / worktree_error / tick_error from_state TEXT, to_state TEXT, payload_artifact_id INTEGER REFERENCES artifacts(artifact_id), diff --git a/packages/cli/src/core/task_budget.ts b/packages/cli/src/core/task_budget.ts index 8ef05244..f0340713 100644 --- a/packages/cli/src/core/task_budget.ts +++ b/packages/cli/src/core/task_budget.ts @@ -179,18 +179,24 @@ function adjustUnderLock( previous_budget_exhausted: task.budget_exhausted === 1, budget_exhausted: nextBudgetExhausted === 1, }; - const event = deps.db.transaction(() => { - deps.db + const txResult = deps.db.transaction(() => { + const upd = deps.db .query( `UPDATE tasks SET retry_budget = ?, budget_exhausted = ?, updated_at = ? - WHERE task_id = ?`, + WHERE task_id = ? + AND state = ? + AND cancel_requested_at IS NULL`, ) - .run(nextBudget, nextBudgetExhausted, now, task.task_id); + .run(nextBudget, nextBudgetExhausted, now, task.task_id, task.state); + const changes = (upd as { changes?: number }).changes ?? 0; + if (changes === 0) { + return { ok: false as const }; + } - return deps.db + const event = deps.db .query<{ event_id: number }, [string, string, string, string, string]>( `INSERT INTO events ( task_id, event_type, from_state, to_state, occurred_at, event_data @@ -204,8 +210,27 @@ function adjustUnderLock( now, JSON.stringify(eventData), ); + return { ok: true as const, event }; })(); - if (!event) throw new Error("task_budget_adjusted event insert returned no row"); + if (!txResult.ok) { + const current = loadTask(deps.db, task.task_id); + return { + ok: false, + error: { + code: "unsafe_state", + message: + `task ${task.task_id} changed before budget adjustment; retry after re-checking its current state`, + details: { + task_id: task.task_id, + observed_state: task.state, + current_state: current?.state ?? null, + }, + }, + }; + } + if (!txResult.event) { + throw new Error("task_budget_adjusted event insert returned no row"); + } return { ok: true, @@ -219,7 +244,7 @@ function adjustUnderLock( budget_exhausted: nextBudgetExhausted === 1, reason, forced: input.force === true, - event_id: event.event_id, + event_id: txResult.event.event_id, }, }; } diff --git a/packages/cli/tests/task/task_budget.test.ts b/packages/cli/tests/task/task_budget.test.ts index 1dc53f04..0583edc7 100644 --- a/packages/cli/tests/task/task_budget.test.ts +++ b/packages/cli/tests/task/task_budget.test.ts @@ -131,6 +131,66 @@ test("adjust_task_budget raises budget, clears exhaustion, and records audit rea }); }); +test("adjust_task_budget refuses when state changes before fenced update", async () => { + const { h, built, taskId } = setupTask("awaiting-next-brief"); + let mutated = false; + const racingClock = { + now: () => h.clock.now(), + nowISO: () => { + if (!mutated) { + mutated = true; + h.db + .query(`UPDATE tasks SET state = 'queued' WHERE task_id = ?`) + .run(taskId); + } + return h.clock.nowISO(); + }, + }; + + const result = await adjust_task_budget( + { + db: h.db, + clock: racingClock, + supervisorLock: built.deps.supervisorLock, + }, + { + taskId, + by: 1, + reason: "operator recovery after substrate failure", + }, + ); + + expect(result).toMatchObject({ + ok: false, + error: { + code: "unsafe_state", + details: { + task_id: taskId, + observed_state: "awaiting-next-brief", + current_state: "queued", + }, + }, + }); + const task = h.db + .query<{ state: string; retry_budget: number; budget_exhausted: number }, [string]>( + `SELECT state, retry_budget, budget_exhausted FROM tasks WHERE task_id = ?`, + ) + .get(taskId); + expect(task).toEqual({ + state: "queued", + retry_budget: 5, + budget_exhausted: 1, + }); + const eventCount = h.db + .query<{ count: number }, [string]>( + `SELECT count(*) AS count + FROM events + WHERE task_id = ? AND event_type = 'task_budget_adjusted'`, + ) + .get(taskId); + expect(eventCount).toEqual({ count: 0 }); +}); + test("task increase-budget CLI lets blocker_resolved resume after budget is raised", async () => { const { h, built, taskId } = setupTask();