From b2a6cbd85939df4a192ce30d31855b5b025f9028 Mon Sep 17 00:00:00 2001 From: mchwang Date: Sat, 26 Sep 2026 00:51:30 -0700 Subject: [PATCH 1/2] F1a: persist the runner lifecycle in the Store Schema v6 with a v5 backfill: tasks, attempts, user_actions and feedback_events. Guarded attempt transitions, where the Store chooses the terminal state from the first reason, D's result and context currency. State version and context generation counters, cancel task, merge closure, replayable user actions (including refusals) and feedback events. Implements the Store slice of docs/implementation/runner-lifecycle.md. Co-Authored-By: Claude Opus 5.5 --- runner/lifecycle.ts | 91 +++++++ runner/store.ts | 360 +++++++++++++++++++++++++++- test/runner-lifecycle-store.test.ts | 296 +++++++++++++++++++++++ 3 files changed, 738 insertions(+), 9 deletions(-) create mode 100644 runner/lifecycle.ts create mode 100644 test/runner-lifecycle-store.test.ts diff --git a/runner/lifecycle.ts b/runner/lifecycle.ts new file mode 100644 index 0000000..91e7265 --- /dev/null +++ b/runner/lifecycle.ts @@ -0,0 +1,91 @@ +import { createHash } from 'node:crypto'; +import type { InvocationContext, Phase, StopReason } from '../agents/contract.ts'; + +/** F1 runner lifecycle vocabulary. See docs/implementation/runner-lifecycle.md. */ +export type AttemptState = 'pending' | 'running' | 'completed' | 'failed' | 'cancelled' | 'stale'; +export type FirstReason = 'cancelled' | 'shutdown' | 'stale' | 'time-limit'; +export type AttemptKind = 'planning' | 'question' | 'review' | 'check' | 'execute' | 'fix' | 'rebase-fix'; +export type TaskStatus = 'queued' | 'running' | 'needs human' | 'needs amendment' | 'needs approval' + | 'possibly already fixed' | 'in review' | 'approved but merge blocked' | 'merged' | 'cancelled'; + +export const TASK_STATUSES: readonly TaskStatus[] = ['queued', 'running', 'needs human', 'needs amendment', 'needs approval', + 'possibly already fixed', 'in review', 'approved but merge blocked', 'merged', 'cancelled']; +export const CLOSED_STATUSES: readonly TaskStatus[] = ['merged', 'cancelled']; +export const TERMINAL_STATES: readonly AttemptState[] = ['completed', 'failed', 'cancelled', 'stale']; +export const FIRST_REASONS: readonly FirstReason[] = ['cancelled', 'shutdown', 'stale', 'time-limit']; +/** Each F attempt kind runs under exactly one existing D phase. */ +export const ATTEMPT_PHASES = { + planning: 'planning', question: 'questions', review: 'review', check: 'review', + execute: 'execute', fix: 'fix', 'rebase-fix': 'fix', +} as const satisfies Record; +export const WRITABLE_KINDS: readonly AttemptKind[] = ['execute', 'fix', 'rebase-fix']; +export const MAX_REASON = 4000; +export const MAX_RESULT_BYTES = 1024 * 1024; +export const DEFAULT_TASK_BUDGET_MS = 2 * 60 * 60 * 1000; + +const UUID_V4 = /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/; +/** Attempt, action and allocation IDs are lowercase UUID v4s; anything else is refused before use. */ +export function isUuidV4(value: unknown): value is string { return typeof value === 'string' && UUID_V4.test(value); } +export function assertUuidV4(value: unknown, name: string): asserts value is string { + if (!isUuidV4(value)) throw new GuardRefusal(`${name} must be a lowercase UUID v4.`); +} + +/** A guard refused the action. Refusals are definite outcomes and are recorded for replay. */ +export class GuardRefusal extends Error {} +/** Reusing an action ID for a different request. */ +export class ActionIdReused extends GuardRefusal {} + +export function bounded(reason: string): string { + const text = reason.trim() || 'No reason given.'; + return text.length > MAX_REASON ? `${text.slice(0, MAX_REASON - 1)}…` : text; +} + +/** Stable fingerprint of a user action request; key order does not matter. */ +export function requestHash(kind: string, request: unknown): string { + const canonical = (value: unknown): unknown => Array.isArray(value) ? value.map(canonical) + : value && typeof value === 'object' + ? Object.fromEntries(Object.keys(value).sort().map(key => [key, canonical((value as Record)[key])])) + : value; + return createHash('sha256').update(JSON.stringify([kind, canonical(request)])).digest('hex'); +} + +export interface Settlement { + /** F's first reason: the durable one, or the in-memory one if its write failed. */ + firstReason: FirstReason | null; + contextCurrent: boolean; + stopReason?: StopReason; + exitCode: number | null; + /** Output passed validation (schema, size and, for writable attempts, the file-scope audit). */ + valid: boolean; + /** Bounded actionable detail from the provider, used for generic failures. */ + detail?: string; +} +export interface Classification { + state: Exclude; + reason: string | null; + /** The task budget ran out; the task moves to needs human unless it is closed or a cancel is pending. */ + timeLimit: boolean; +} + +/** Terminal-state precedence at settlement. Order matters; see "Rules for the running state", rule 2. */ +export function classifySettlement(s: Settlement): Classification { + const dFailure = s.stopReason === 'timeout' || s.stopReason === 'output-limit' || s.stopReason === 'capture-failure'; + const dReason = () => s.stopReason === 'timeout' ? 'Timed out.' : bounded(s.detail ?? `Agent stopped: ${s.stopReason}.`); + switch (s.firstReason) { + case 'cancelled': return { state: 'cancelled', reason: 'Cancelled.', timeLimit: false }; + // D keeps the first reason it received: a non-shutdown stop reason proves D stopped before shutdown reached it. + case 'shutdown': return dFailure ? { state: 'failed', reason: dReason(), timeLimit: false } + : { state: 'cancelled', reason: 'Stopped by shutdown', timeLimit: false }; + case 'stale': return { state: 'stale', reason: bounded(s.detail ?? 'The plan, snapshot, assignment or referenced code changed.'), timeLimit: false }; + case 'time-limit': return { state: 'cancelled', reason: 'Task time limit reached', timeLimit: true }; + } + if (!s.contextCurrent) return { state: 'stale', reason: 'The plan, snapshot, assignment or referenced code changed.', timeLimit: false }; + if (dFailure) return { state: 'failed', reason: dReason(), timeLimit: false }; + if (s.exitCode === 0 && s.valid) return { state: 'completed', reason: null, timeLimit: false }; + return { state: 'failed', reason: bounded(s.detail ?? `Agent exited with code ${s.exitCode ?? 'none'}.`), timeLimit: false }; +} + +export function sameContext(a: InvocationContext, b: InvocationContext): boolean { + return a.snapshotId === b.snapshotId && a.planId === b.planId && a.planRevision === b.planRevision + && a.assignmentId === b.assignmentId && a.referencedCodeHash === b.referencedCodeHash && a.stateVersion === b.stateVersion; +} diff --git a/runner/store.ts b/runner/store.ts index 1762b42..763fadf 100644 --- a/runner/store.ts +++ b/runner/store.ts @@ -1,9 +1,15 @@ -import type { DatabaseSync, SQLInputValue } from 'node:sqlite'; +import type { DatabaseSync, SQLInputValue, SQLOutputValue } from 'node:sqlite'; import { createRequire } from 'node:module'; import { randomUUID } from 'node:crypto'; import { identityKey, type PlanIdentity } from '../core/identity.ts'; import { importPlan, applySuggestion, assertEditReply, type Plan, type PlanContext, type EditReply } from '../core/plan.ts'; import type { Approval, SegmentChoice } from '../core/approvals.ts'; +import type { InvocationContext, StopReason } from '../agents/contract.ts'; +import { + ATTEMPT_PHASES, CLOSED_STATUSES, DEFAULT_TASK_BUDGET_MS, FIRST_REASONS, GuardRefusal, ActionIdReused, MAX_RESULT_BYTES, TASK_STATUSES, TERMINAL_STATES, + assertUuidV4, bounded, classifySettlement, requestHash, sameContext, + type AttemptKind, type AttemptState, type Classification, type FirstReason, type Settlement, type TaskStatus, +} from './lifecycle.ts'; export function requireSupportedNode(version = process.versions.node): void { const [major, minor] = version.split('.').map(Number); @@ -25,6 +31,22 @@ export interface MergeAttempt { phase: 'AWAITING_CHECKS' | 'LOCKED' | 'MERGEABLE' | 'QUEUED' | null; position: number | null; occurredAt: string | null; createdAt: string; updatedAt: string; } +export interface TaskRecord { + planKey: string; status: TaskStatus; stateVersion: number; contextGeneration: number; assignmentId: string; referencedCodeHash: string; + currentAttemptId: string | null; requeuePending: boolean; cancelRequested: string | null; rebaseInProgress: unknown; budgetDeadline: number | null; + createdAt: string; updatedAt: string; +} +export interface AttemptRecord { + id: string; kind: AttemptKind; phase: string; item: string | null; state: AttemptState; context: InvocationContext; deadline: number; + firstReason: FirstReason | null; stopReason: StopReason | null; exitCode: number | null; signal: string | null; result: unknown; + diagnostic: string | null; diagnosticRef: string | null; createdAt: string; startedAt: string | null; settledAt: string | null; +} +export type FeedbackKind = 'reject' | 'change-request' | 'segment-accept' | 'segment-assign' | 'finding-accept' | 'needs-human-guidance' | 'task-closed'; +const FEEDBACK_KINDS: readonly FeedbackKind[] = ['reject', 'change-request', 'segment-accept', 'segment-assign', 'finding-accept', 'needs-human-guidance', 'task-closed']; +export interface FeedbackEvent { + id: string; planKey: string; actionId: string; planRevision: number; snapshotId: string | null; item: string | null; + kind: FeedbackKind; text: string | null; sourceRef: string; supersedes: string | null; createdAt: string; +} export interface LedgerEntry { sha: string; owner: string | null; origin: 'owned' | 'foreign'; sourceSha: string | null } export interface Checkpoint { id: string; revision: number; snapshotId: string; item: string; @@ -49,9 +71,9 @@ export class Store { try { this.#db.exec('PRAGMA foreign_keys=ON; PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL;'); this.#transaction(() => { - const version = this.#get('PRAGMA user_version')!.user_version; - if (version !== 0 && version !== 1 && version !== 2 && version !== 3 && version !== 4 && version !== 5) throw new Error('Unsupported store schema version.'); - if (version === 5) return; + const version = this.#get('PRAGMA user_version')!.user_version as number; + if (![0, 1, 2, 3, 4, 5, 6].includes(version)) throw new Error('Unsupported store schema version.'); + if (version === 6) return; if (version === 0) this.#db.exec(` CREATE TABLE plans (key TEXT PRIMARY KEY, issue INTEGER NOT NULL, revision INTEGER NOT NULL, snapshot_id TEXT); CREATE TABLE revisions (key TEXT NOT NULL REFERENCES plans(key), revision INTEGER NOT NULL, data TEXT NOT NULL, PRIMARY KEY(key,revision)); @@ -79,6 +101,7 @@ export class Store { data TEXT NOT NULL ); PRAGMA user_version=5;`); + if (version < 6) this.#migrateV6(); }); } catch (error) { this.#db.close(); throw error; } } @@ -93,10 +116,14 @@ export class Store { close(): void { this.#db.close(); } #get(sql: string, ...args: SQLInputValue[]) { return this.#db.prepare(sql).get(...args); } #run(sql: string, ...args: SQLInputValue[]) { return this.#db.prepare(sql).run(...args); } + #depth = 0; + /** Nested calls join the outer transaction, so a user action can wrap existing Store methods atomically. */ #transaction(fn: () => T): T { - this.#db.exec('BEGIN IMMEDIATE'); + if (this.#depth > 0) { this.#depth++; try { return fn(); } finally { this.#depth--; } } + this.#db.exec('BEGIN IMMEDIATE'); this.#depth = 1; try { const result = fn(); this.#db.exec('COMMIT'); return result; } catch (error) { this.#db.exec('ROLLBACK'); throw error; } + finally { this.#depth = 0; } } #current(key: string) { const row = this.#get('SELECT * FROM plans WHERE key=?', key); @@ -114,6 +141,7 @@ export class Store { #savePlan(key: string, plan: Plan, expected: number): void { if (this.#run('UPDATE plans SET revision=? WHERE key=? AND revision=?', plan.revision, key, expected).changes !== 1) throw new Error('Stale plan revision.'); this.#run('INSERT INTO revisions VALUES (?,?,?)', key, plan.revision, encode(plan)); + this.#bumpContext(key); this.#run("UPDATE requests SET state='invalidated',reason='Plan revision changed.' WHERE key=? AND state IN ('pending','ready')", key); const items = new Set(plan.items.map(item => item.id)); for (const row of this.#db.prepare('SELECT item FROM approvals WHERE key=?').all(key)) { @@ -132,6 +160,9 @@ export class Store { this.#run('INSERT INTO plans (key,issue,revision,snapshot_id) VALUES (?,?,?,NULL)', key, plan.issue, 0); this.#savePlan(key, plan, 0); this.#snapshot(key, base, head); + const now = new Date().toISOString(); + this.#run(`INSERT INTO tasks (plan_key,status,state_version,context_generation,assignment_id,referenced_code_hash,created_at,updated_at) + VALUES (?,'in review',0,0,'unassigned',?,?,?)`, key, head, now, now); return plan; }); } @@ -155,6 +186,7 @@ export class Store { const snapshot = { id: randomUUID(), base, head }; this.#run('INSERT INTO snapshots VALUES (?,?,?)', key, snapshot.id, encode(snapshot)); this.#run('UPDATE plans SET snapshot_id=? WHERE key=?', snapshot.id, key); + this.#bumpContext(key); this.#run("UPDATE requests SET state='invalidated',reason='Repository snapshot changed.' WHERE key=? AND state IN ('pending','ready')", key); return snapshot; } @@ -247,10 +279,15 @@ export class Store { if (!['merged','removed','failed'].includes(outcome.state)) throw new Error('Invalid merge-queue outcome.'); if (outcome.state !== 'merged' && (typeof outcome.reason !== 'string' || !outcome.reason.trim() || outcome.reason.length > 4000)) throw new Error('A bounded terminal merge reason is required.'); if (outcome.occurredAt !== undefined && (!Number.isFinite(Date.parse(outcome.occurredAt)) || outcome.occurredAt.length > 64)) throw new Error('Invalid merge-queue timestamp.'); - return this.#changeMergeAttempt(identity, id, ['submitting','queued'], attempt => ({ - ...attempt, state: outcome.state, reason: outcome.state === 'merged' ? null : outcome.reason!.trim(), - occurredAt: outcome.occurredAt ?? null, requiresFreshReview: outcome.requiresFreshReview === true, - })); + // A confirmed merge closes the task and records task-closed in the same transaction (feedback-event rule 2). + return this.#transaction(() => { + const changed = this.#changeMergeAttempt(identity, id, ['submitting','queued'], attempt => ({ + ...attempt, state: outcome.state, reason: outcome.state === 'merged' ? null : outcome.reason!.trim(), + occurredAt: outcome.occurredAt ?? null, requiresFreshReview: outcome.requiresFreshReview === true, + })); + if (changed && outcome.state === 'merged') this.#closeTask(identityKey(identity), 'merged', id); + return changed; + }); } getMergeAttempt(identity: PlanIdentity): MergeAttempt | null { this.#current(identityKey(identity)); @@ -427,4 +464,309 @@ export class Store { this.getCheckpoint(identity, checkpointId); return (this.#get('SELECT MAX(revision) AS revision FROM continuations WHERE key=? AND checkpoint_id=?', identityKey(identity), checkpointId)?.revision as number | null) ?? null; } + // ---- F1 runner lifecycle (docs/implementation/runner-lifecycle.md) ---- + #migrateV6(): void { + const list = (values: readonly string[]) => values.map(value => `'${value}'`).join(','); + this.#db.exec(` + CREATE TABLE IF NOT EXISTS tasks ( + plan_key TEXT PRIMARY KEY REFERENCES plans(key), + status TEXT NOT NULL CHECK (status IN (${list(TASK_STATUSES)})), + state_version INTEGER NOT NULL DEFAULT 0, context_generation INTEGER NOT NULL DEFAULT 0, + assignment_id TEXT NOT NULL, referenced_code_hash TEXT NOT NULL, + current_attempt_id TEXT, requeue_pending INTEGER NOT NULL DEFAULT 0, cancel_requested TEXT, rebase_in_progress TEXT, + budget_deadline INTEGER, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, + FOREIGN KEY (plan_key, current_attempt_id) REFERENCES attempts(plan_key, id)); + CREATE TABLE IF NOT EXISTS attempts ( + id TEXT PRIMARY KEY, plan_key TEXT NOT NULL REFERENCES tasks(plan_key), + kind TEXT NOT NULL, phase TEXT NOT NULL, item TEXT, + state TEXT NOT NULL CHECK (state IN ('pending','running','completed','failed','cancelled','stale')), + context TEXT NOT NULL, deadline INTEGER NOT NULL, + first_reason TEXT CHECK (first_reason IS NULL OR first_reason IN (${list(FIRST_REASONS)})), + stop_reason TEXT, exit_code INTEGER, signal TEXT, result TEXT, diagnostic TEXT, diagnostic_ref TEXT, + preparation_pgid INTEGER, preparation_started_at INTEGER, allocation_id TEXT, + created_at TEXT NOT NULL, started_at TEXT, settled_at TEXT, UNIQUE (plan_key, id)); + CREATE TABLE IF NOT EXISTS user_actions ( + plan_key TEXT NOT NULL REFERENCES plans(key), action_id TEXT NOT NULL, kind TEXT NOT NULL, + request_hash TEXT NOT NULL, response TEXT NOT NULL, created_at TEXT NOT NULL, PRIMARY KEY (plan_key, action_id)); + CREATE TABLE IF NOT EXISTS feedback_events ( + id TEXT PRIMARY KEY, plan_key TEXT NOT NULL REFERENCES plans(key), action_id TEXT NOT NULL, + plan_revision INTEGER NOT NULL, snapshot_id TEXT, item TEXT, + kind TEXT NOT NULL CHECK (kind IN (${list(FEEDBACK_KINDS)})), text TEXT, source_ref TEXT NOT NULL, + supersedes TEXT REFERENCES feedback_events(id), created_at TEXT NOT NULL, UNIQUE (plan_key, kind, action_id));`); + // Backfill: one task per existing plan; a merged plan also gets its task-closed event. + const now = new Date().toISOString(); + const latestMerge = (column: string) => `(SELECT ${column} FROM merge_attempts m WHERE m.key=p.key ORDER BY m.rowid DESC LIMIT 1)`; + this.#run(`INSERT OR IGNORE INTO tasks (plan_key,status,state_version,context_generation,assignment_id,referenced_code_hash,created_at,updated_at) + SELECT p.key, CASE WHEN ${latestMerge("json_extract(m.data,'$.state')")}='merged' THEN 'merged' ELSE 'in review' END, 0, 0, 'unassigned', + COALESCE((SELECT json_extract(s.data,'$.head') FROM snapshots s WHERE s.key=p.key AND s.id=p.snapshot_id),'none'), ?, ? + FROM plans p`, now, now); + this.#run(`INSERT INTO feedback_events (id,plan_key,action_id,plan_revision,snapshot_id,item,kind,text,source_ref,supersedes,created_at) + SELECT lower(hex(randomblob(16))), p.key, ${latestMerge('m.id')}, p.revision, p.snapshot_id, NULL, 'task-closed', NULL, p.key, NULL, ? + FROM plans p JOIN tasks t ON t.plan_key=p.key WHERE t.status='merged' + AND NOT EXISTS (SELECT 1 FROM feedback_events e WHERE e.plan_key=p.key AND e.kind='task-closed')`, now); + this.#db.exec('PRAGMA user_version=6;'); + } + #task(key: string) { + const row = this.#get('SELECT * FROM tasks WHERE plan_key=?', key); + if (!row) throw new Error('Unknown task.'); + return row; + } + #taskRecord(row: Record): TaskRecord { + return { + planKey: row.plan_key as string, status: row.status as TaskStatus, stateVersion: row.state_version as number, + contextGeneration: row.context_generation as number, assignmentId: row.assignment_id as string, + referencedCodeHash: row.referenced_code_hash as string, currentAttemptId: row.current_attempt_id as string | null, + requeuePending: row.requeue_pending === 1, cancelRequested: row.cancel_requested as string | null, + rebaseInProgress: row.rebase_in_progress === null ? null : decode(row.rebase_in_progress), + budgetDeadline: row.budget_deadline as number | null, createdAt: row.created_at as string, updatedAt: row.updated_at as string, + }; + } + #attemptRecord(row: Record): AttemptRecord { + return { + id: row.id as string, kind: row.kind as AttemptKind, phase: row.phase as string, item: row.item as string | null, + state: row.state as AttemptState, context: decode(row.context), deadline: row.deadline as number, + firstReason: row.first_reason as FirstReason | null, stopReason: row.stop_reason as StopReason | null, + exitCode: row.exit_code as number | null, signal: row.signal as string | null, + result: row.result === null ? null : decode(row.result), diagnostic: row.diagnostic as string | null, + diagnosticRef: row.diagnostic_ref as string | null, createdAt: row.created_at as string, + startedAt: row.started_at as string | null, settledAt: row.settled_at as string | null, + }; + } + /** Every durable change to a task or its attempts increases the state version. */ + #touch(key: string): void { + this.#run('UPDATE tasks SET state_version=state_version+1, updated_at=? WHERE plan_key=?', new Date().toISOString(), key); + } + /** A change an attempt depends on increases both counters in the caller's transaction. */ + #bumpContext(key: string): void { + this.#run('UPDATE tasks SET context_generation=context_generation+1, state_version=state_version+1, updated_at=? WHERE plan_key=?', new Date().toISOString(), key); + } + #contextOf(key: string): InvocationContext { + const plan = this.#current(key), task = this.#task(key); + return { + snapshotId: plan.snapshot_id as string, planId: (JSON.parse(key) as string[])[2]!, planRevision: plan.revision as number, + assignmentId: task.assignment_id as string, referencedCodeHash: task.referenced_code_hash as string, + stateVersion: task.context_generation as number, + }; + } + #closed(status: unknown): boolean { return CLOSED_STATUSES.includes(status as TaskStatus); } + #closeTask(key: string, status: 'merged' | 'cancelled', actionId: string): void { + const plan = this.#current(key); + this.#run('UPDATE tasks SET status=?, cancel_requested=NULL WHERE plan_key=?', status, key); + this.#run(`INSERT INTO feedback_events (id,plan_key,action_id,plan_revision,snapshot_id,item,kind,text,source_ref,supersedes,created_at) + VALUES (?,?,?,?,?,NULL,'task-closed',NULL,?,NULL,?) ON CONFLICT(plan_key,kind,action_id) DO NOTHING`, + randomUUID(), key, actionId, plan.revision!, plan.snapshot_id ?? null, key, new Date().toISOString()); + this.#touch(key); + } + getTask(identity: PlanIdentity): TaskRecord { return this.#taskRecord(this.#task(identityKey(identity))); } + currentContext(identity: PlanIdentity): InvocationContext { return this.#contextOf(identityKey(identity)); } + getAttempt(identity: PlanIdentity, id: string): AttemptRecord { + const row = this.#get('SELECT * FROM attempts WHERE plan_key=? AND id=?', identityKey(identity), id); + if (!row) throw new Error('Unknown attempt.'); + return this.#attemptRecord(row); + } + getAttempts(identity: PlanIdentity): AttemptRecord[] { + return this.#db.prepare('SELECT * FROM attempts WHERE plan_key=? ORDER BY rowid').all(identityKey(identity)).map(row => this.#attemptRecord(row)); + } + /** Reassigning work or changing referenced code makes older attempts non-current. */ + setAssignment(identity: PlanIdentity, expectedStateVersion: number, assignmentId: string, referencedCodeHash: string): void { + if (![assignmentId, referencedCodeHash].every(value => typeof value === 'string' && value.length > 0 && value.length <= 200)) + throw new GuardRefusal('Invalid assignment.'); + const key = identityKey(identity); + this.#transaction(() => { + if (this.#task(key).state_version !== expectedStateVersion) throw new GuardRefusal('Stale task state. Reload before writing.'); + this.#run('UPDATE tasks SET assignment_id=?, referenced_code_hash=? WHERE plan_key=?', assignmentId, referencedCodeHash, key); + this.#bumpContext(key); + }); + } + /** Status changes other than closing and admission. Closing uses cancelTask or a confirmed merge; running comes from admission. */ + transitionTask(identity: PlanIdentity, expectedStateVersion: number, to: TaskStatus): void { + if (!TASK_STATUSES.includes(to) || this.#closed(to) || to === 'running') throw new GuardRefusal('Invalid task status change.'); + const key = identityKey(identity); + this.#transaction(() => { + const task = this.#task(key); + if (task.state_version !== expectedStateVersion) throw new GuardRefusal('Stale task state. Reload before writing.'); + if (this.#closed(task.status)) throw new GuardRefusal('A closed task never changes.'); + if (this.#activeAttempt(key)) throw new GuardRefusal('An attempt is still active for this task.'); + this.#run('UPDATE tasks SET status=? WHERE plan_key=?', to, key); + this.#touch(key); + }); + } + #activeAttempt(key: string) { + return this.#get("SELECT * FROM attempts WHERE plan_key=? AND state IN ('pending','running')", key); + } + /** Admission, including retry: status, state version, requeue claim, active attempt and captured context are checked in one transaction. */ + admitAttempt(identity: PlanIdentity, input: { + expectedStateVersion: number; kind: AttemptKind; item?: string | null; expectedContext: InvocationContext; + deadline: number; budgetMs?: number; retryOf?: string; now?: number; + }): AttemptRecord { + const now = input.now ?? Date.now(), budgetMs = input.budgetMs ?? DEFAULT_TASK_BUDGET_MS; + if (!(input.kind in ATTEMPT_PHASES)) throw new GuardRefusal('Unknown attempt kind.'); + if (!Number.isSafeInteger(input.deadline) || input.deadline <= now) throw new GuardRefusal('An attempt needs a finite future deadline.'); + if (!Number.isSafeInteger(budgetMs) || budgetMs < 1) throw new GuardRefusal('Invalid task budget.'); + if (input.retryOf !== undefined) assertUuidV4(input.retryOf, 'Retried attempt ID'); + const key = identityKey(identity); + return this.#transaction(() => { + const task = this.#task(key); + if (task.status !== 'running' && task.status !== 'queued') throw new GuardRefusal(`The task is ${task.status}; it cannot start work.`); + if (task.state_version !== input.expectedStateVersion) throw new GuardRefusal('Stale task state. Reload before writing.'); + if (task.requeue_pending === 1) throw new GuardRefusal('Recovery is requeueing this task.'); + if (task.cancel_requested !== null) throw new GuardRefusal('The task is being cancelled.'); + if (this.#activeAttempt(key)) throw new GuardRefusal('An attempt is already active for this task.'); + const current = this.#contextOf(key); + if (!sameContext(input.expectedContext, current)) throw new GuardRefusal('The plan, snapshot, assignment or referenced code changed. Reload before starting.'); + if (input.retryOf !== undefined) { + const last = task.current_attempt_id === input.retryOf ? this.#get('SELECT * FROM attempts WHERE plan_key=? AND id=?', key, input.retryOf) : undefined; + if (!last || (last.state !== 'failed' && last.state !== 'cancelled')) throw new GuardRefusal('Only the latest failed or cancelled attempt can be retried.'); + if (!sameContext(decode(last.context), current)) throw new GuardRefusal('The retried attempt is out of date. Start a new request on the current code.'); + } + if (input.item !== undefined && input.item !== null && !this.getPlan(identity).items.some(entry => entry.id === input.item)) + throw new GuardRefusal('Unknown plan item.'); + const id = randomUUID(), created = new Date(now).toISOString(); + this.#run(`INSERT INTO attempts (id,plan_key,kind,phase,item,state,context,deadline,created_at) VALUES (?,?,?,?,?,'pending',?,?,?)`, + id, key, input.kind, ATTEMPT_PHASES[input.kind], input.item ?? null, encode(current), input.deadline, created); + this.#run(`UPDATE tasks SET current_attempt_id=?, status='running', budget_deadline=COALESCE(budget_deadline, ?) WHERE plan_key=?`, id, now + budgetMs, key); + this.#touch(key); + return this.getAttempt(identity, id); + }); + } + /** The "Stopping" transition: sets the first reason once, keeps the state. */ + recordFirstReason(identity: PlanIdentity, id: string, reason: FirstReason): boolean { + if (!FIRST_REASONS.includes(reason)) throw new GuardRefusal('Unknown stop reason.'); + const key = identityKey(identity); + return this.#transaction(() => { + const changed = this.#run(`UPDATE attempts SET first_reason=? WHERE plan_key=? AND id=? AND state IN ('pending','running') AND first_reason IS NULL`, reason, key, id).changes === 1; + if (changed) this.#touch(key); + return changed; + }); + } + /** pending -> running after D returns a handle. Refused once a first reason is recorded. */ + markRunning(identity: PlanIdentity, id: string): boolean { + const key = identityKey(identity); + return this.#transaction(() => { + const changed = this.#run(`UPDATE attempts SET state='running', started_at=? WHERE plan_key=? AND id=? AND state='pending' AND first_reason IS NULL + AND id=(SELECT current_attempt_id FROM tasks WHERE plan_key=?)`, new Date().toISOString(), key, id, key).changes === 1; + if (changed) this.#touch(key); + return changed; + }); + } + /** + * Terminal write. The Store, not the caller, chooses the terminal state from the durable first reason + * (or the caller's in-memory one if its write failed), D's result and whether the captured context is still current. + */ + settleAttempt(identity: PlanIdentity, id: string, settlement: Omit & { + signal?: string | null; result?: unknown; diagnosticRef?: string | null; + }): Classification { + if (settlement.firstReason !== null && !FIRST_REASONS.includes(settlement.firstReason)) throw new GuardRefusal('Unknown stop reason.'); + const key = identityKey(identity); + return this.#transaction(() => { + const task = this.#task(key), row = this.#get('SELECT * FROM attempts WHERE plan_key=? AND id=?', key, id); + if (!row || task.current_attempt_id !== id || (row.state !== 'pending' && row.state !== 'running')) throw new GuardRefusal('Attempt is not the active attempt.'); + const firstReason = (row.first_reason as FirstReason | null) ?? settlement.firstReason; + const contextCurrent = sameContext(decode(row.context), this.#contextOf(key)); + let outcome = classifySettlement({ ...settlement, firstReason, contextCurrent }); + if (outcome.state === 'completed' && row.state !== 'running') throw new GuardRefusal('Only a running attempt can complete.'); + if (outcome.state === 'completed' && (this.#closed(task.status) || task.cancel_requested !== null)) + outcome = { state: 'cancelled', reason: 'The task was closed before the result was saved.', timeLimit: false }; + let result: string | null = null; + if (outcome.state === 'completed') { + result = encode(settlement.result ?? null); + if (Buffer.byteLength(result) > MAX_RESULT_BYTES) outcome = { state: 'failed', reason: 'The result exceeds 1 MiB.', timeLimit: false }, result = null; + } + this.#run(`UPDATE attempts SET state=?, first_reason=?, stop_reason=?, exit_code=?, signal=?, result=?, diagnostic=?, diagnostic_ref=?, settled_at=? WHERE id=?`, + outcome.state, firstReason, settlement.stopReason ?? null, settlement.exitCode, settlement.signal ?? null, result, + outcome.reason, settlement.diagnosticRef ?? null, new Date().toISOString(), id); + // A pending cancel task wins over everything, including the time limit. + if (task.cancel_requested !== null && !this.#closed(task.status)) this.#closeTask(key, 'cancelled', task.cancel_requested as string); + else { + if (outcome.timeLimit && !this.#closed(task.status)) this.#run(`UPDATE tasks SET status='needs human' WHERE plan_key=?`, key); + this.#touch(key); + } + return outcome; + }); + } + /** Cancel task: closes now, or, with an active attempt, stops it first and closes when it settles. */ + cancelTask(identity: PlanIdentity, expectedStateVersion: number, actionId: string): 'closed' | 'stopping' { + assertUuidV4(actionId, 'Action ID'); + const key = identityKey(identity); + return this.#transaction(() => { + const task = this.#task(key); + if (task.state_version !== expectedStateVersion) throw new GuardRefusal('Stale task state. Reload before writing.'); + if (this.#closed(task.status)) throw new GuardRefusal('The task is already closed.'); + const merge = this.getMergeAttempt(identity); + if (merge && (merge.state === 'submitting' || merge.state === 'queued')) + throw new GuardRefusal('A merge is in progress. Cancel the task after it finishes or fails.'); + const active = this.#activeAttempt(key); + if (!active) { this.#closeTask(key, 'cancelled', actionId); return 'closed'; } + if (task.cancel_requested !== null) throw new GuardRefusal('The task is already being cancelled.'); + this.#run(`UPDATE attempts SET first_reason='cancelled' WHERE id=? AND first_reason IS NULL`, active.id!); + this.#run('UPDATE tasks SET cancel_requested=? WHERE plan_key=?', actionId, key); + this.#touch(key); + return 'stopping'; + }); + } + /** + * Exact replay for writing user actions. The first definite outcome is recorded, including a guard refusal. + * Storage errors are not recorded, so the UI may resend. The response must be JSON. + */ + userAction(identity: PlanIdentity, action: { actionId: string; kind: string; request: unknown }, apply: () => T): { response: T; replayed: boolean } { + assertUuidV4(action.actionId, 'Action ID'); + if (typeof action.kind !== 'string' || !/^[a-z][a-z-]{0,39}$/.test(action.kind)) throw new GuardRefusal('Invalid action kind.'); + const key = identityKey(identity), hash = requestHash(action.kind, action.request); + const saved = () => { + const row = this.#get('SELECT * FROM user_actions WHERE plan_key=? AND action_id=?', key, action.actionId); + if (!row) return undefined; + if (row.request_hash !== hash) throw new ActionIdReused('Action ID already used for a different request.'); + const outcome = decode<{ ok: boolean; value?: T; error?: string }>(row.response); + if (!outcome.ok) throw new GuardRefusal(outcome.error!); + return { response: outcome.value as T, replayed: true }; + }; + const record = (outcome: object) => { + const response = encode(outcome); + if (response.length > 65536) throw new Error('Action response is too large to record.'); + this.#run('INSERT INTO user_actions VALUES (?,?,?,?,?,?)', key, action.actionId, action.kind, hash, response, new Date().toISOString()); + }; + let replaying = false; + try { + return this.#transaction(() => { + const prior = saved(); if (prior) { replaying = true; return prior; } + const value = apply(); + record({ ok: true, value }); + return { response: value, replayed: false }; + }); + } catch (error) { + const storage = (error as { code?: string }).code === 'ERR_SQLITE_ERROR'; + if (!replaying && !storage && !(error instanceof ActionIdReused) && this.#depth === 0) { + const message = error instanceof Error ? bounded(error.message) : 'Refused.'; + this.#transaction(() => { if (!this.#get('SELECT 1 FROM user_actions WHERE plan_key=? AND action_id=?', key, action.actionId)) record({ ok: false, error: message }); }); + } + throw error; + } + } + /** Append one feedback event. Call inside userAction so the event and its action share one transaction. */ + recordFeedback(identity: PlanIdentity, actionId: string, event: { kind: Exclude; item?: string | null; text?: string | null; sourceRef: string; supersedes?: string | null }): FeedbackEvent { + assertUuidV4(actionId, 'Action ID'); + if (this.#depth === 0) throw new Error('Feedback events are written inside their user action.'); + if (!FEEDBACK_KINDS.includes(event.kind) || event.kind === ('task-closed' as FeedbackKind)) throw new GuardRefusal('Invalid feedback kind.'); + if (event.text != null && (typeof event.text !== 'string' || event.text.length > 4000)) throw new GuardRefusal('Feedback text is limited to 4000 characters.'); + if (typeof event.sourceRef !== 'string' || !event.sourceRef || event.sourceRef.length > 200) throw new GuardRefusal('Invalid feedback source.'); + const key = identityKey(identity), plan = this.#current(key); + if (event.supersedes != null && !this.#get('SELECT 1 FROM feedback_events WHERE plan_key=? AND id=? AND source_ref=?', key, event.supersedes, event.sourceRef)) + throw new GuardRefusal('A superseded event must belong to the same source.'); + const id = randomUUID(), createdAt = new Date().toISOString(); + this.#run(`INSERT INTO feedback_events (id,plan_key,action_id,plan_revision,snapshot_id,item,kind,text,source_ref,supersedes,created_at) VALUES (?,?,?,?,?,?,?,?,?,?,?)`, + id, key, actionId, plan.revision!, plan.snapshot_id ?? null, event.item ?? null, event.kind, event.text ?? null, event.sourceRef, event.supersedes ?? null, createdAt); + return { id, planKey: key, actionId, planRevision: plan.revision as number, snapshotId: plan.snapshot_id as string | null, item: event.item ?? null, + kind: event.kind, text: event.text ?? null, sourceRef: event.sourceRef, supersedes: event.supersedes ?? null, createdAt }; + } + /** Lane J's only read path: available once the task has closed. */ + feedbackEvents(identity: PlanIdentity): FeedbackEvent[] { + const key = identityKey(identity); this.#task(key); + if (!this.#get("SELECT 1 FROM feedback_events WHERE plan_key=? AND kind='task-closed'", key)) throw new GuardRefusal('Feedback is available after the task closes.'); + return this.#db.prepare('SELECT * FROM feedback_events WHERE plan_key=? ORDER BY rowid').all(key).map(row => ({ + id: row.id as string, planKey: row.plan_key as string, actionId: row.action_id as string, planRevision: row.plan_revision as number, + snapshotId: row.snapshot_id as string | null, item: row.item as string | null, kind: row.kind as FeedbackKind, text: row.text as string | null, + sourceRef: row.source_ref as string, supersedes: row.supersedes as string | null, createdAt: row.created_at as string, + })); + } + } diff --git a/test/runner-lifecycle-store.test.ts b/test/runner-lifecycle-store.test.ts new file mode 100644 index 0000000..479a990 --- /dev/null +++ b/test/runner-lifecycle-store.test.ts @@ -0,0 +1,296 @@ +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { randomUUID } from 'node:crypto'; +import { DatabaseSync } from 'node:sqlite'; +import { afterEach, describe, expect, it } from 'vitest'; +import { Store } from '../runner/store.ts'; +import { ActionIdReused, GuardRefusal, classifySettlement, requestHash } from '../runner/lifecycle.ts'; +import type { Plan, PlanContext } from '../core/plan.ts'; + +const identity = { repositoryId: 'repo', taskId: 'task', planId: 'plan' }; +const context: PlanContext = { identity, issue: 1, baseEntries: [{ path: 'a', kind: 'file' }], pathKey: p => p, allowedCommands: [] }; +const plan = (summary = 'Example'): Plan => ({ schema_version: 1, revision: 1, issue: 1, summary, questions: [], items: [{ id: 'P1', title: 'Change', intent: 'Improve', files: [{ path: 'a', kind: 'edit', renamed_from: null, change: 'Change' }], acceptance: [{ type: 'check', text: 'Works' }], depends_on: [] }] }); +const oid = (n: number) => n.toString(16).padStart(40, '0'); +const dirs: string[] = [], stores: Store[] = []; +afterEach(() => { for (const store of stores.splice(0)) store.close(); for (const dir of dirs.splice(0)) rmSync(dir, { recursive: true, force: true }); }); +function open(path: string) { const store = new Store(path); stores.push(store); return store; } +function fixture() { + const dir = mkdtempSync(join(tmpdir(), 'codeboost-lifecycle-')); dirs.push(dir); + const path = join(dir, 'state.sqlite'), store = open(path); + store.createPlan(JSON.stringify(plan()), 'json', context, oid(1), oid(2)); + return { path, store }; +} +/** A task ready for work: moved from review to queued. */ +function queued() { const f = fixture(); f.store.transitionTask(identity, f.store.getTask(identity).stateVersion, 'queued'); return f; } +const later = () => Date.now() + 60_000; +function admit(store: Store, extra: Partial[1]> = {}) { + return store.admitAttempt(identity, { expectedStateVersion: store.getTask(identity).stateVersion, kind: 'execute', item: 'P1', + expectedContext: store.currentContext(identity), deadline: later(), ...extra }); +} +const ok = { firstReason: null, exitCode: 0, valid: true } as const; +const settle = (store: Store, id: string, s: Partial[2]> = {}) => store.settleAttempt(identity, id, { ...ok, ...s }); + +describe('settlement precedence', () => { + it('orders first reason, context currency, D stop reason and exit status', () => { + const base = { contextCurrent: true, exitCode: 0, valid: true }; + expect(classifySettlement({ ...base, firstReason: 'cancelled', exitCode: 1 }).state).toBe('cancelled'); + expect(classifySettlement({ ...base, firstReason: 'stale' }).state).toBe('stale'); + expect(classifySettlement({ ...base, firstReason: 'time-limit' })).toMatchObject({ state: 'cancelled', reason: 'Task time limit reached', timeLimit: true }); + // A D stop that came before shutdown wins; otherwise shutdown cancels. + expect(classifySettlement({ ...base, firstReason: 'shutdown', stopReason: 'timeout' })).toMatchObject({ state: 'failed', reason: 'Timed out.' }); + expect(classifySettlement({ ...base, firstReason: 'shutdown', stopReason: 'shutdown' })).toMatchObject({ state: 'cancelled', reason: 'Stopped by shutdown' }); + // No reason and a changed context is stale, even when D timed out. + expect(classifySettlement({ ...base, firstReason: null, contextCurrent: false, stopReason: 'timeout' }).state).toBe('stale'); + expect(classifySettlement({ ...base, firstReason: null, stopReason: 'output-limit', detail: 'too big' })).toMatchObject({ state: 'failed', reason: 'too big' }); + expect(classifySettlement({ ...base, firstReason: null }).state).toBe('completed'); + expect(classifySettlement({ ...base, firstReason: null, valid: false }).state).toBe('failed'); + }); + it('fingerprints requests independently of key order', () => { + expect(requestHash('note', { a: 1, b: [2, { c: 3, d: 4 }] })).toBe(requestHash('note', { b: [2, { d: 4, c: 3 }], a: 1 })); + expect(requestHash('note', { a: 1 })).not.toBe(requestHash('reject', { a: 1 })); + }); +}); + +describe('schema v6', () => { + it('backfills one task per v5 plan, closes merged plans once, and never treats a null budget as expired', () => { + const { store, path } = fixture(); + const other = { ...identity, planId: 'merged' }; + store.createPlan(JSON.stringify(plan()), 'json', { ...context, identity: other }, oid(1), oid(3)); + const state = { revision: 1, snapshotId: store.getSnapshot(other).id, reviewVersion: store.reviewVersion(other) }; + const merge = store.beginMergeAttempt(other, state, oid(3), null, 'direct'); + store.finishMergeAttempt(other, merge.id, { state: 'merged' }); + store.close(); stores.splice(stores.indexOf(store), 1); + const legacy = new DatabaseSync(path); + legacy.exec('DROP TABLE feedback_events; DROP TABLE user_actions; DROP TABLE tasks; DROP TABLE attempts; PRAGMA user_version=5;'); + legacy.close(); + const migrated = open(path); + const open1 = migrated.getTask(identity), closed = migrated.getTask(other); + expect(open1).toMatchObject({ status: 'in review', stateVersion: 0, contextGeneration: 0, assignmentId: 'unassigned', referencedCodeHash: oid(2), + currentAttemptId: null, requeuePending: false, cancelRequested: null, rebaseInProgress: null, budgetDeadline: null }); + expect(closed.status).toBe('merged'); + const events = migrated.feedbackEvents(other); + expect(events).toHaveLength(1); + expect(events[0]).toMatchObject({ kind: 'task-closed', actionId: merge.id, sourceRef: closed.planKey }); + expect(() => migrated.feedbackEvents(identity)).toThrow(/after the task closes/); + // Reopening a migrated v6 store changes nothing. + expect(open(path).getTask(other).stateVersion).toBe(closed.stateVersion); + }); +}); + +describe('state version and context generation', () => { + it('increase both on context changes, and only the state version on lifecycle changes', () => { + const { store } = queued(); + const start = store.getTask(identity); + store.importRevision(JSON.stringify(plan('Next')), 'json', context, 1); + let now = store.getTask(identity); + expect(now.contextGeneration).toBe(start.contextGeneration + 1); expect(now.stateVersion).toBeGreaterThan(start.stateVersion); + store.recordHistory(identity, { revision: 2, snapshotId: store.getSnapshot(identity).id }, oid(1), oid(4), []); + expect(store.getTask(identity).contextGeneration).toBe(now.contextGeneration + 1); + now = store.getTask(identity); + store.setAssignment(identity, now.stateVersion, 'assignment-2', 'hash-2'); + const assigned = store.getTask(identity); + expect(assigned.contextGeneration).toBe(now.contextGeneration + 1); + const attempt = admit(store); + const admitted = store.getTask(identity); + expect(admitted.contextGeneration).toBe(assigned.contextGeneration); + expect(admitted.stateVersion).toBe(assigned.stateVersion + 1); + expect(attempt.context).toEqual({ snapshotId: store.getSnapshot(identity).id, planId: 'plan', planRevision: 2, assignmentId: 'assignment-2', referencedCodeHash: 'hash-2', stateVersion: assigned.contextGeneration }); + }); +}); + +describe('admission', () => { + it('requires an active task, the current state version, a current context and no active attempt', () => { + const { store } = fixture(); + expect(() => admit(store)).toThrow(/in review/); + store.transitionTask(identity, store.getTask(identity).stateVersion, 'queued'); + expect(() => admit(store, { expectedStateVersion: 0 })).toThrow(/Stale task state/); + const staleContext = store.currentContext(identity); + store.setAssignment(identity, store.getTask(identity).stateVersion, 'other', 'hash'); + expect(() => admit(store, { expectedContext: staleContext })).toThrow(GuardRefusal); + expect(() => admit(store, { deadline: Date.now() - 1 })).toThrow(/future deadline/); + const first = admit(store); + expect(first).toMatchObject({ state: 'pending', kind: 'execute', phase: 'execute', item: 'P1', firstReason: null }); + const task = store.getTask(identity); + expect(task).toMatchObject({ status: 'running', currentAttemptId: first.id }); + expect(task.budgetDeadline).toBeGreaterThan(Date.now()); + expect(() => admit(store)).toThrow(/already active/); + expect(store.getAttempts(identity)).toHaveLength(1); + }); + it('lets exactly one of two processes admit work at the same moment', () => { + const { store, path } = queued(); const other = open(path); + const version = store.getTask(identity).stateVersion, ctx = store.currentContext(identity); + const input = { expectedStateVersion: version, kind: 'execute' as const, expectedContext: ctx, deadline: later() }; + store.admitAttempt(identity, input); + expect(() => other.admitAttempt(identity, input)).toThrow(GuardRefusal); + expect(store.getAttempts(identity)).toHaveLength(1); + }); + it('refuses admission while recovery holds the requeue claim or a cancel is pending', () => { + const { store, path } = queued(); + const db = new DatabaseSync(path); db.exec('UPDATE tasks SET requeue_pending=1'); db.close(); + expect(() => admit(store)).toThrow(/requeueing/); + }); + it('starts the task budget at the first admission and keeps it across later attempts', () => { + const { store } = queued(); + const first = admit(store, { budgetMs: 1000, now: 1_000_000, deadline: later() }); + const budget = store.getTask(identity).budgetDeadline; + expect(budget).toBe(1_001_000); + settle(store, first.id, { exitCode: 1 }); + admit(store, { retryOf: first.id }); + expect(store.getTask(identity).budgetDeadline).toBe(budget); + }); +}); + +describe('attempt transitions', () => { + it('records the first reason once and refuses pending -> running after it', () => { + const { store } = queued(); const attempt = admit(store); + const before = store.getTask(identity).stateVersion; + expect(store.recordFirstReason(identity, attempt.id, 'cancelled')).toBe(true); + expect(store.recordFirstReason(identity, attempt.id, 'shutdown')).toBe(false); + expect(store.getTask(identity).stateVersion).toBe(before + 1); + expect(store.getAttempt(identity, attempt.id)).toMatchObject({ state: 'pending', firstReason: 'cancelled' }); + expect(store.markRunning(identity, attempt.id)).toBe(false); + expect(settle(store, attempt.id, { exitCode: 0 }).state).toBe('cancelled'); + }); + it('keeps the first reason when a later provider error arrives', () => { + const { store } = queued(); const attempt = admit(store); store.markRunning(identity, attempt.id); + store.recordFirstReason(identity, attempt.id, 'cancelled'); + expect(settle(store, attempt.id, { exitCode: 1, valid: false }).state).toBe('cancelled'); + expect(store.getAttempt(identity, attempt.id)).toMatchObject({ state: 'cancelled', firstReason: 'cancelled', exitCode: 1 }); + }); + it('uses the in-memory first reason when its earlier write failed', () => { + const { store } = queued(); const attempt = admit(store); store.markRunning(identity, attempt.id); + expect(settle(store, attempt.id, { firstReason: 'shutdown', stopReason: 'shutdown' }).state).toBe('cancelled'); + expect(store.getAttempt(identity, attempt.id).firstReason).toBe('shutdown'); + }); + it('ends stale, not failed, when the context changed before a provider error or a D timeout', () => { + for (const s of [{ exitCode: 1, valid: false }, { exitCode: null, stopReason: 'timeout' as const }]) { + const { store } = queued(); const attempt = admit(store); store.markRunning(identity, attempt.id); + store.importRevision(JSON.stringify(plan('Moved')), 'json', context, 1); + expect(settle(store, attempt.id, s).state).toBe('stale'); + } + }); + it('publishes only a running attempt with a current context, and stores a bounded result', () => { + const { store } = queued(); const attempt = admit(store); + expect(() => settle(store, attempt.id, { result: { ok: 1 } })).toThrow(/running attempt can complete/); + store.markRunning(identity, attempt.id); + expect(settle(store, attempt.id, { result: { ok: 1 } }).state).toBe('completed'); + expect(store.getAttempt(identity, attempt.id)).toMatchObject({ state: 'completed', result: { ok: 1 } }); + const next = admit(store); store.markRunning(identity, next.id); + expect(settle(store, next.id, { result: 'x'.repeat(1024 * 1024 + 1) })).toMatchObject({ state: 'failed', reason: 'The result exceeds 1 MiB.' }); + }); + it('refuses a late settlement from an old attempt after a retry, without touching the retry', () => { + const { store } = queued(); const old = admit(store); + store.recordFirstReason(identity, old.id, 'cancelled'); settle(store, old.id); + const retry = admit(store, { retryOf: old.id }); store.markRunning(identity, retry.id); + expect(() => settle(store, old.id)).toThrow(/not the active attempt/); + expect(store.getAttempt(identity, retry.id).state).toBe('running'); + expect(store.getTask(identity).currentAttemptId).toBe(retry.id); + }); + it('allows retry only of the latest failed or cancelled attempt with a current context', () => { + const { store } = queued(); const first = admit(store); store.markRunning(identity, first.id); + settle(store, first.id, { exitCode: 1, valid: false }); + store.setAssignment(identity, store.getTask(identity).stateVersion, 'other', 'hash'); + expect(() => admit(store, { retryOf: first.id })).toThrow(/out of date/); + const second = admit(store); store.markRunning(identity, second.id); settle(store, second.id); + expect(() => admit(store, { retryOf: second.id })).toThrow(/failed or cancelled/); + expect(() => admit(store, { retryOf: first.id })).toThrow(/failed or cancelled/); + }); +}); + +describe('task closure', () => { + it('closes a task with no active attempt at once, with one task-closed event', () => { + const { store } = queued(); const actionId = randomUUID(); + expect(store.cancelTask(identity, store.getTask(identity).stateVersion, actionId)).toBe('closed'); + expect(store.getTask(identity).status).toBe('cancelled'); + expect(store.feedbackEvents(identity)).toMatchObject([{ kind: 'task-closed', actionId }]); + expect(() => store.transitionTask(identity, store.getTask(identity).stateVersion, 'queued')).toThrow(/never changes/); + expect(() => admit(store)).toThrow(/cancelled/); + }); + it('stops an active attempt first, then closes the task when it settles, even if it would have completed', () => { + const { store } = queued(); const attempt = admit(store); store.markRunning(identity, attempt.id); + const actionId = randomUUID(); + expect(store.cancelTask(identity, store.getTask(identity).stateVersion, actionId)).toBe('stopping'); + expect(store.getTask(identity)).toMatchObject({ status: 'running', cancelRequested: actionId }); + expect(settle(store, attempt.id, { result: 'late' }).state).toBe('cancelled'); + expect(store.getTask(identity)).toMatchObject({ status: 'cancelled', cancelRequested: null }); + expect(store.feedbackEvents(identity).filter(event => event.kind === 'task-closed')).toHaveLength(1); + }); + it('lets a pending cancel task win over the time limit', () => { + const { store } = queued(); const attempt = admit(store); store.markRunning(identity, attempt.id); + store.recordFirstReason(identity, attempt.id, 'time-limit'); + store.cancelTask(identity, store.getTask(identity).stateVersion, randomUUID()); + settle(store, attempt.id); + expect(store.getTask(identity).status).toBe('cancelled'); + }); + it('moves a timed-out task to needs human, where retry is refused', () => { + const { store } = queued(); const attempt = admit(store); store.markRunning(identity, attempt.id); + store.recordFirstReason(identity, attempt.id, 'time-limit'); + expect(settle(store, attempt.id).state).toBe('cancelled'); + expect(store.getTask(identity).status).toBe('needs human'); + expect(() => admit(store, { retryOf: attempt.id })).toThrow(/needs human/); + }); + it('refuses cancel task during a merge, and closes the task as merged when the merge is confirmed', () => { + const { store } = fixture(); + const state = { revision: 1, snapshotId: store.getSnapshot(identity).id, reviewVersion: store.reviewVersion(identity) }; + const merge = store.beginMergeAttempt(identity, state, oid(2), null, 'direct'); + expect(() => store.cancelTask(identity, store.getTask(identity).stateVersion, randomUUID())).toThrow(/merge is in progress/); + store.finishMergeAttempt(identity, merge.id, { state: 'merged' }); + expect(store.getTask(identity).status).toBe('merged'); + expect(store.feedbackEvents(identity)).toMatchObject([{ kind: 'task-closed', actionId: merge.id }]); + }); +}); + +describe('user actions', () => { + it('replays the saved response without applying the action again', () => { + const { store } = queued(); const actionId = randomUUID(); let applied = 0; + const run = () => store.userAction(identity, { actionId, kind: 'note', request: { text: 'hi' } }, () => ++applied); + expect(run()).toEqual({ response: 1, replayed: false }); + expect(run()).toEqual({ response: 1, replayed: true }); + expect(applied).toBe(1); + expect(() => store.userAction(identity, { actionId, kind: 'reject', request: { text: 'hi' } }, () => 0)).toThrow(ActionIdReused); + }); + it('records a refusal and returns the same refusal after the state changes', () => { + const { store } = queued(); const first = admit(store); store.markRunning(identity, first.id); + const actionId = randomUUID(); + const retry = () => store.userAction(identity, { actionId, kind: 'retry', request: { attemptId: first.id } }, + () => admit(store, { retryOf: first.id }).id); + expect(retry).toThrow(/already active/); + settle(store, first.id, { exitCode: 1, valid: false }); + expect(retry).toThrow(/already active/); + expect(store.getAttempts(identity)).toHaveLength(1); + }); + it('rejects malformed action IDs before storing anything, and does not record storage errors', () => { + const { store } = queued(); + for (const actionId of ['x'.repeat(10_000), 'not-a-uuid', randomUUID().toUpperCase()]) + expect(() => store.userAction(identity, { actionId, kind: 'note', request: {} }, () => 1)).toThrow(/UUID v4/); + const actionId = randomUUID(); + const storage = Object.assign(new Error('disk I/O error'), { code: 'ERR_SQLITE_ERROR' }); + expect(() => store.userAction(identity, { actionId, kind: 'note', request: {} }, () => { throw storage; })).toThrow(/disk/); + expect(store.userAction(identity, { actionId, kind: 'note', request: {} }, () => 7)).toEqual({ response: 7, replayed: false }); + }); + it('writes feedback events only inside their action, and supersedes by source', () => { + const { store } = queued(); + expect(() => store.recordFeedback(identity, randomUUID(), { kind: 'segment-assign', sourceRef: 'choice-1' })).toThrow(/inside their user action/); + const firstId = randomUUID(); + const first = store.userAction(identity, { actionId: firstId, kind: 'assign', request: { item: 'P1' } }, + () => store.recordFeedback(identity, firstId, { kind: 'segment-assign', item: 'P1', sourceRef: 'choice-1' })).response; + const secondId = randomUUID(); + store.userAction(identity, { actionId: secondId, kind: 'assign', request: { item: 'P2' } }, + () => store.recordFeedback(identity, secondId, { kind: 'segment-assign', item: 'P1', sourceRef: 'choice-1', supersedes: first.id })); + const badId = randomUUID(); + expect(() => store.userAction(identity, { actionId: badId, kind: 'assign', request: {} }, + () => store.recordFeedback(identity, badId, { kind: 'segment-assign', sourceRef: 'choice-2', supersedes: first.id }))).toThrow(/same source/); + store.cancelTask(identity, store.getTask(identity).stateVersion, randomUUID()); + expect(store.feedbackEvents(identity).map(event => [event.kind, event.supersedes])).toEqual([['segment-assign', null], ['segment-assign', first.id], ['task-closed', null]]); + }); + it('rolls back the action and its event together', () => { + const { store } = queued(); const actionId = randomUUID(); + expect(() => store.userAction(identity, { actionId, kind: 'note', request: {} }, () => { + store.recordFeedback(identity, actionId, { kind: 'change-request', item: 'P1', text: 'Fix', sourceRef: 'note-1' }); + throw new GuardRefusal('Refused after the event.'); + })).toThrow(/Refused after/); + store.cancelTask(identity, store.getTask(identity).stateVersion, randomUUID()); + expect(store.feedbackEvents(identity).map(event => event.kind)).toEqual(['task-closed']); + }); +}); From 4c83e7822f0ca3d70c7ecdc7489f778e361adcf4 Mon Sep 17 00:00:00 2001 From: mchwang Date: Sat, 26 Sep 2026 01:10:54 -0700 Subject: [PATCH 2/2] Complete only into a running task (contract round 35) Co-Authored-By: Claude Opus 5.5 --- runner/store.ts | 6 ++++-- test/runner-lifecycle-store.test.ts | 7 +++++++ 2 files changed, 11 insertions(+), 2 deletions(-) diff --git a/runner/store.ts b/runner/store.ts index 763fadf..ad8eabd 100644 --- a/runner/store.ts +++ b/runner/store.ts @@ -665,8 +665,10 @@ export class Store { const contextCurrent = sameContext(decode(row.context), this.#contextOf(key)); let outcome = classifySettlement({ ...settlement, firstReason, contextCurrent }); if (outcome.state === 'completed' && row.state !== 'running') throw new GuardRefusal('Only a running attempt can complete.'); - if (outcome.state === 'completed' && (this.#closed(task.status) || task.cancel_requested !== null)) - outcome = { state: 'cancelled', reason: 'The task was closed before the result was saved.', timeLimit: false }; + // The task must still be running; a task never leaves running while an attempt is active, so this is a second safeguard. + if (outcome.state === 'completed' && (task.status !== 'running' || task.cancel_requested !== null)) + outcome = { state: 'cancelled', reason: this.#closed(task.status) || task.cancel_requested !== null + ? 'The task was closed before the result was saved.' : 'The task left the running state before the result was saved.', timeLimit: false }; let result: string | null = null; if (outcome.state === 'completed') { result = encode(settlement.result ?? null); diff --git a/test/runner-lifecycle-store.test.ts b/test/runner-lifecycle-store.test.ts index 479a990..1f4d655 100644 --- a/test/runner-lifecycle-store.test.ts +++ b/test/runner-lifecycle-store.test.ts @@ -179,6 +179,13 @@ describe('attempt transitions', () => { const next = admit(store); store.markRunning(identity, next.id); expect(settle(store, next.id, { result: 'x'.repeat(1024 * 1024 + 1) })).toMatchObject({ state: 'failed', reason: 'The result exceeds 1 MiB.' }); }); + it('freezes the task status while an attempt is active, and never completes into a non-running task', () => { + const { store, path } = queued(); const attempt = admit(store); store.markRunning(identity, attempt.id); + expect(() => store.transitionTask(identity, store.getTask(identity).stateVersion, 'needs amendment')).toThrow(/still active/); + const db = new DatabaseSync(path); db.exec(`UPDATE tasks SET status='needs amendment'`); db.close(); + expect(settle(store, attempt.id, { result: 'late' })).toMatchObject({ state: 'cancelled', reason: 'The task left the running state before the result was saved.' }); + expect(store.getTask(identity).status).toBe('needs amendment'); + }); it('refuses a late settlement from an old attempt after a retry, without touching the retry', () => { const { store } = queued(); const old = admit(store); store.recordFirstReason(identity, old.id, 'cancelled'); settle(store, old.id);