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..ad8eabd 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,311 @@ 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.'); + // 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); + 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..1f4d655 --- /dev/null +++ b/test/runner-lifecycle-store.test.ts @@ -0,0 +1,303 @@ +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('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); + 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']); + }); +});