diff --git a/README.md b/README.md index 214fe230..7fb4db8e 100644 --- a/README.md +++ b/README.md @@ -51,7 +51,7 @@ A configuration with a `github` block may also add a `runner` block. The runner - `diagnosticsCapBytes`: the size retention trims that directory back to. It is a target, not a hard limit: the file just saved is always kept, even when it alone passes the cap. The default is 256 MiB. - `limits`: task storage limits (`workBytes`, `workInodes`, `metadataBytes`, `metadataInodes`). -The runner needs Docker and `CLAUDE_CODE_OAUTH_TOKEN`. At start, before the server opens, codeboost recovers what an earlier run left and builds the agent image; a recovery it cannot finish safely stops startup with what to do. Nothing in the review screen starts a run yet (#91 part 2). +The runner needs Docker and `CLAUDE_CODE_OAUTH_TOKEN`. At start, before the server opens, codeboost recovers what an earlier run left and builds the agent image; a recovery it cannot finish safely stops startup with what to do. `POST /api/runner` with `start` or `resume` runs the plan; the review screen has no buttons for them yet. ## Library diff --git a/docs/implementation/runner-lifecycle.md b/docs/implementation/runner-lifecycle.md index 79107500..9813e299 100644 --- a/docs/implementation/runner-lifecycle.md +++ b/docs/implementation/runner-lifecycle.md @@ -319,11 +319,17 @@ This runs before the coordinator opens. 7. **Unowned preparation check.** If any attempt has `preparation_started_at` set and `preparation_pgid` null (see "Launch"), stop here: do not open the coordinator, name each attempt and directory, and wait for `--release-preparation`. Step 4 does not remove those directories. 8. Open the coordinator. -**Production wiring (#91).** The runner is opt-in: the review configuration's `runner` block names its root, an optional diagnostics directory and cap, the committer and optional storage limits (`runner/production.ts`, `parseRunnerConfig`). Without the block, or in a demo, there is no coordinator and runner actions answer that the runner is not configured. With it, `web/cli.ts` takes the lock, `startServer` opens the `Store` and then, before it listens, runs `setUpRunner`: it checks for the github block (the prompt carries the issue) and `CLAUDE_CODE_OAUTH_TOKEN`, calls the lock's `verify()` (decision 1, step 4), reads the runner token for the locked file, and runs `recoverStartup` with D's own `recoverLeftovers`, `exportTaskDiff` and `removeTaskFilesystemsAsync`. The agent image is built after D's recovery has stopped every leftover agent, so no agent keeps writing to its storage during a long build, and before the first export's deadline starts (the build blocks the event loop, so a deadline armed before it would expire as soon as it returned). Any failure closes the `Store`, releases the lock and exits with its message. To export a recovered storage, F saves the storage's `metadataBaseline` and the commit it was seeded from (`attempts.metadata_baseline`, `attempts.storage_base`, schema v9) as soon as D's allocation returns; a row without them gets "Partial output could not be exported: its storage baseline was never saved". Diagnostics are kept per database, in `//diagnostics`: retention reads references only from its own `Store`, so runners of databases sharing a directory must never see each other's files. Recovered diffs are saved through `saveDiagnostic`, with the same owner-only file and retention as the live path; retention keeps every diff this recovery saved, because none is referenced until the finalization transaction. A diff D cut at its byte limit ends with a `codeboost:` line that says so, on this path and the live one. The issue text a prompt carries (title, body and collaborators' comments) is read once and reused for later items for up to five minutes; the base tree listing is read once per base commit. A first Ctrl+C during startup waits for startup to finish; a second stops the process at once, which recovery treats like a crash at the next start. The deps accept execute attempts only (`RunnerDeps.kinds`): admission refuses any other kind before it writes anything. A plan item runs only through `ItemExecutor`, so `/api/runner` refuses to retry an execute attempt on its own; resuming a task comes in #91 part 2. +**Production wiring (#91).** The runner is opt-in: the review configuration's `runner` block names its root, an optional diagnostics directory and cap, the committer and optional storage limits (`runner/production.ts`, `parseRunnerConfig`). Without the block, or in a demo, there is no coordinator and runner actions answer that the runner is not configured. With it, `web/cli.ts` takes the lock, `startServer` opens the `Store` and then, before it listens, runs `setUpRunner`: it checks for the github block (the prompt carries the issue) and `CLAUDE_CODE_OAUTH_TOKEN`, calls the lock's `verify()` (decision 1, step 4), reads the runner token for the locked file, and runs `recoverStartup` with D's own `recoverLeftovers`, `exportTaskDiff` and `removeTaskFilesystemsAsync`. The agent image is built after D's recovery has stopped every leftover agent, so no agent keeps writing to its storage during a long build, and before the first export's deadline starts (the build blocks the event loop, so a deadline armed before it would expire as soon as it returned). Any failure closes the `Store`, releases the lock and exits with its message. To export a recovered storage, F saves the storage's `metadataBaseline` and the commit it was seeded from (`attempts.metadata_baseline`, `attempts.storage_base`, schema v9) as soon as D's allocation returns; a row without them gets "Partial output could not be exported: its storage baseline was never saved". Diagnostics are kept per database, in `//diagnostics`: retention reads references only from its own `Store`, so runners of databases sharing a directory must never see each other's files. Recovered diffs are saved through `saveDiagnostic`, with the same owner-only file and retention as the live path; retention keeps every diff this recovery saved, because none is referenced until the finalization transaction. A diff D cut at its byte limit ends with a `codeboost:` line that says so, on this path and the live one. The issue text a prompt carries (title, body and collaborators' comments) is read once and reused for later items for up to five minutes; the base tree listing is read once per base commit. A first Ctrl+C during startup waits for startup to finish; a second stops the process at once, which recovery treats like a crash at the next start. The deps accept execute attempts only (`RunnerDeps.kinds`): admission refuses any other kind before it writes anything. A plan item runs only through `ItemExecutor`, so `/api/runner` refuses to retry an execute attempt on its own; a task's plan runs again through `resume` (below). ## HTTP and UI contract -**Status reads.** `GET /api/runner` returns only the task and attempt rows plus `stateVersion`, `retryable`, `unresolved` and `stopRequested`. `stopRequested` comes from the in-memory job: null, or `{ attemptId, reason, saved }`, where `saved` is false while the first-reason write has failed. The UI shows "Stopping (not saved yet)" only from this field. It does not change `stateVersion`, so user actions still compare against the durable state version. `unresolved` is computed by the server from the in-memory marker: null, or `{ attemptId, reason: "result-not-saved" | "start-not-saved" | "preparation-not-removed" | "storage-not-removed" }`: the terminal write or the start could not be saved; the host-side preparation files could not be removed (before the terminal write when D never ran, after it once D settled); or the task storage could not be removed after a saved terminal write. The UI shows "Needs restart" with that cause only from this field, because the durable row alone may still look active. It does not rebuild Git history or the full review. +**Status reads.** `GET /api/runner` returns only the task and attempt rows plus `available`, `stateVersion`, `retryable`, `startable`, `resumable`, `unresolved` and `stopRequested`. `stopRequested` comes from the in-memory job: null, or `{ attemptId, reason, saved }`, where `saved` is false while the first-reason write has failed. The UI shows "Stopping (not saved yet)" only from this field. It does not change `stateVersion`, so user actions still compare against the durable state version. `unresolved` is computed by the server from the in-memory marker: null, or `{ attemptId, reason: "result-not-saved" | "start-not-saved" | "preparation-not-removed" | "storage-not-removed" }`: the terminal write or the start could not be saved; the host-side preparation files could not be removed (before the terminal write when D never ran, after it once D settled); or the task storage could not be removed after a saved terminal write. The UI shows "Needs restart" with that cause only from this field, because the durable row alone may still look active. It does not rebuild Git history or the full review. + +**Start and resume (#91 part 2).** `POST /api/runner` with `action: "start"` or `"resume"` sends `expectedStateVersion` and an `actionId`, but no attempt ID. Both run the plan through `ItemExecutor`, item by item. Everything up to the first item's admission happens inside the user action's transaction, so a refusal rolls back the action's own writes. +- `start` runs a task that is `in review` or `queued`, has no execute attempt at its current plan revision, and has no pending requeue (that is `resume`'s). An `in review` task moves to `queued` in the same action. +- `resume` continues a task that is `running` or `queued` and has started running (it has an execute attempt at any plan revision, or recovery left it to requeue). It starts at the first item with no completed execute attempt at the current revision, whether the last item completed, failed or was stopped, and claims `requeue_pending` in the admitting transaction. This is also how a task stopped between items continues. +- Both refuse a closed task, a pending cancel, a runner that cannot run execute attempts, an active attempt, an earlier run still finishing, an active merge, a scope pause (#88), and a plan revised after an earlier revision's items made runner commits (#88: those commits must be reconciled with the new revision first; earlier attempts that committed nothing do not block). During shutdown both answer 503 before any of these, and record nothing. When none of these applies, a finding or scope pause owed from an earlier run is acted on first, and nothing is admitted (`outcome: "settled"`). Only then is the first item admitted: admission refuses a spent task budget, which also moves an idle `running` or `queued` task to `needs human` (a task still `in review` stays there: its move to `queued` rolls back with the refused action). +- `GET /api/runner` adds `startable` and `resumable`, computed by the same checks, so the view never offers an action that would be refused. The view also withholds them while the task budget has run out (the action may still act on owed work, or else lets admission refuse it with its move to `needs human`) and while the runner is shutting down or needs a restart. Retrying an execute attempt is refused: a plan item runs again only through `resume`. **User actions.** Cancel, retry and "run again" requests send `attemptId`, `expectedStateVersion` and an `actionId` idempotency key (see "Feedback-event contract"). A replayed `actionId` returns the saved outcome. The server takes the plan identity from its trusted configuration, never from the request, and every `Store` call is scoped by that identity. A mismatch returns HTTP 409 with the current state. The UI then shows that state and keeps any draft. @@ -357,7 +363,7 @@ Live planning invocation waits for D5. Until then, the server returns "Planning **What becomes an event.** Only feedback the user wrote or chose. Issue text, issue comments and agent output never become events. -`actionId` identifies the one user action (or confirmed external outcome) that caused the event. **For user actions it is an idempotency key the UI generates** (a UUID) when the user acts. The server validates it before opening any transaction: it must be a lowercase UUID v4 (the same regular expression as attempt IDs, 36 characters), or the request is refused with HTTP 400 and nothing is stored. The UI generates it and resends unchanged if it retries after a lost response. Every writing user action goes through one `user_actions` table, keyed by `(planKey, actionId)`. That covers notes, choice changes, finding acceptance, rejection, cancel, retry, "run again", **starting a merge**, and every planning endpoint that writes: plan import, suggestion start, suggestion cancel and suggestion apply. Each planning request carries an `actionId`; its `user_actions` row is inserted in the same transaction as the `Store` call it wraps, and the saved response includes the request ID or plan revision it produced. For a merge, the `user_actions` row is inserted in the same transaction as `beginMergeAttempt`, and it saves the merge attempt ID with the response "in progress". When the merge coordinator records the outcome, the same transaction updates that saved response. A replayed merge click therefore returns the same attempt's current outcome and never starts a second attempt. The row holds the action kind, a hash of the request body and the bounded response. **The first definite outcome is recorded, including a refusal.** If the action applies, its row is inserted in the same transaction. If a guard refuses it (for example HTTP 409 for a stale state version), the refusal and its response are recorded in their own transaction. Only outcomes that applied nothing for a passing reason are not recorded, so the UI may resend them: HTTP 503 during shutdown, and a storage error. A replay is handled before any other guard: +`actionId` identifies the one user action (or confirmed external outcome) that caused the event. **For user actions it is an idempotency key the UI generates** (a UUID) when the user acts. The server validates it before opening any transaction: it must be a lowercase UUID v4 (the same regular expression as attempt IDs, 36 characters), or the request is refused with HTTP 400 and nothing is stored. The UI generates it and resends unchanged if it retries after a lost response. Every writing user action goes through one `user_actions` table, keyed by `(planKey, actionId)`. That covers notes, choice changes, finding acceptance, rejection, cancel, retry, "run again", start and resume, **starting a merge**, and every planning endpoint that writes: plan import, suggestion start, suggestion cancel and suggestion apply. Each planning request carries an `actionId`; its `user_actions` row is inserted in the same transaction as the `Store` call it wraps, and the saved response includes the request ID or plan revision it produced. For a merge, the `user_actions` row is inserted in the same transaction as `beginMergeAttempt`, and it saves the merge attempt ID with the response "in progress". When the merge coordinator records the outcome, the same transaction updates that saved response. A replayed merge click therefore returns the same attempt's current outcome and never starts a second attempt. The row holds the action kind, a hash of the request body and the bounded response. **The first definite outcome is recorded, including a refusal.** If the action applies, its row is inserted in the same transaction. If a guard refuses it (for example HTTP 409 for a stale state version), the refusal and its response are recorded in their own transaction. Only outcomes that applied nothing for a passing reason are not recorded, so the UI may resend them: HTTP 503 during shutdown, and a storage error. A replay is handled before any other guard: - the same `actionId` with the same request hash returns the saved response and applies nothing, even though the state version has since moved on; - the same `actionId` with a different request hash (including a different action kind) is refused with HTTP 409 "Action ID already used" and applies nothing. diff --git a/github/issues.ts b/github/issues.ts index c143a9b9..de528b28 100644 --- a/github/issues.ts +++ b/github/issues.ts @@ -249,13 +249,19 @@ export class GhIssueGateway implements IssueGateway { if (issue.number !== number) throw new Error('GitHub returned a different issue.'); if (Object.hasOwn(issue, 'pull_request')) throw new Error(`#${number} is a pull request, not an issue.`); const title = boundedString(issue.title, 'title', 4096), body = boundedString(issue.body, 'body', MAX_BODY_LENGTH, true); - const collaborators = await this.#loadCollaborators(signal); - const comments: string[] = []; // The execute prompt carries the issue as one JSON data block of at most MAX_PROMPT_BYTES (dataJSON). A running byte - // count stops reading early; the exact check below uses the prompt's own serializer, so an issue accepted here is - // one the prompt can carry. + // count stops reading early (a title and body already over it read no collaborator or comment page); the exact check + // below uses the prompt's own serializer, so an issue accepted here is one the prompt can carry. const tooLong = () => new Error(`Issue #${number}'s title, body and collaborator comments are larger than the ${MAX_PROMPT_BYTES / 1024} KiB an execute prompt carries; codeboost does not cut an issue to fit.`); + // Only the size refusal is reworded; any other (text with a NUL, for one) keeps its own reason. + const carried = (text: IssueText) => { + try { dataJSON(text, 'Issue data'); } catch (error) { throw /exceeds/.test((error as Error).message) ? tooLong() : error; } + }; + // The title and body alone, as the prompt serializes them (escaping included): an issue that cannot fit reads nothing more. + carried({ number, title, body, comments: [] }); let total = Buffer.byteLength(title) + Buffer.byteLength(body); + const collaborators = await this.#loadCollaborators(signal); + const comments: string[] = []; for (let page = 1; ; page++) { const listed = await this.run(['api', '--method', 'GET', '-H', 'Accept: application/vnd.github+json', `repos/${this.repository}/issues/${number}/comments`, '-f', `per_page=${PAGE_SIZE}`, '-f', `page=${page}`], { signal }); @@ -280,8 +286,7 @@ export class GhIssueGateway implements IssueGateway { if (values.length < PAGE_SIZE) break; } const text = { number, title, body, comments }; - // Only the size refusal is reworded; any other (text with a NUL, for one) keeps its own reason. - try { dataJSON(text, 'Issue data'); } catch (error) { throw /exceeds/.test((error as Error).message) ? tooLong() : error; } + carried(text); return text; }); } diff --git a/runner/coordinator.ts b/runner/coordinator.ts index c9e52e84..5ed39532 100644 --- a/runner/coordinator.ts +++ b/runner/coordinator.ts @@ -74,6 +74,8 @@ export interface SlotLimits { readonly writable: number; readonly readOnly: numb export interface StartRequest { expectedStateVersion: number; kind: AttemptKind; item?: string | null; deadline: number; budgetMs?: number; retryOf?: string; expectedContext: AttemptRecord['context']; + /** Clears the task's requeue claim in the admitting transaction: the user's Resume of an interrupted task. */ + claimRequeue?: boolean; } export interface RunnerStatus { active: boolean; diff --git a/runner/execution.ts b/runner/execution.ts index 9c1435ad..9dc73205 100644 --- a/runner/execution.ts +++ b/runner/execution.ts @@ -259,6 +259,11 @@ export type ExecutionOutcome = | { kind: 'needs amendment'; item: string; outOfScope: string[]; checkpointId: string; completed: string[] } | { kind: 'needs human'; item: string; reason: string; completed: string[] } | { kind: 'stopped'; item: string; state: string; reason: string | null; completed: string[] }; +/** A run in progress: its plan, the item it is on and that item's attempt, and what it finished. */ +interface Run { + plan: ReturnType; index: number; attempt: AttemptRecord; done: string[]; unchanged: string[]; + stopped: (item: string, state: string, reason: string | null) => ExecutionOutcome; +} const refusal = (error: unknown) => error instanceof GuardRefusal || error instanceof ShuttingDownError; /** A write outside settlement: no shutdown capability. */ const direct = (fn: () => T): T => fn(); @@ -279,7 +284,7 @@ export class ItemExecutor { this.#deadlineMs = options.deadlineMs ?? 10 * 60_000; this.#write = settleWith(options.capability); } /** - * Tasks with a runTask in progress here, from its start to its return: a second one waits for none of its steps. + * Tasks with a run (runTask or begin) in progress here, from its start to its return: a second one waits for none of its steps. * The guard is per instance, so the server must keep one ItemExecutor per Store (as it keeps one coordinator). */ #inFlight = new Map>(); @@ -288,11 +293,41 @@ export class ItemExecutor { // Held for the whole run, so a second run can never pay this run's pause or finding, even in the microtasks between // the coordinator dropping the job and this run resuming. if (this.#inFlight.has(key)) return { kind: 'stopped', item: options.fromItem ?? this.#store.getPlan(identity).items[0]!.id, state: 'not started', - reason: 'An earlier run of this task is still finishing; start it again when that run has ended.', completed: [] }; - const run = this.#runTask(identity, options); + reason: 'An earlier run of this task is still finishing; try again when that run has ended.', completed: [] }; + const begun = this.#begin(identity, options, false); + return this.#track(key, 'outcome' in begun ? Promise.resolve(begun.outcome) : this.#continue(identity, begun)); + } + /** + * Start or resume a run from inside the caller's user action (#91 part 2). Everything up to the first item's admission + * is synchronous, so it all happens in the caller's transaction: anything that keeps the first item from being admitted + * throws (the refusal itself where admission refused), and the caller's own writes roll back with it. A finding or scope + * pause owed from an earlier run is acted on instead, as the run's whole outcome, and admits nothing (`attemptId` null). + */ + begin(identity: PlanIdentity, options: { fromItem?: string; claimRequeue?: boolean } = {}): { attemptId: string | null; outcome: Promise } { + const key = identityKey(identity); + if (this.#inFlight.has(key)) throw new GuardRefusal('An earlier run of this task is still finishing; try again when that run has ended.'); + const begun = this.#begin(identity, options, true); + if ('outcome' in begun) return { attemptId: null, outcome: this.#track(key, Promise.resolve(begun.outcome)) }; + return { attemptId: begun.attempt.id, outcome: this.#track(key, this.#continue(identity, begun)) }; + } + /** + * Where a run of the task stands (#91 part 2): whether any item was ever attempted (`started`, at any revision) or was + * attempted at the current revision (`begun`), whether an earlier revision left runner commits a revised plan must be + * reconciled with (#88), the items an execute attempt completed at the current revision, and the first item none did + * (null when every item has run). + */ + progress(identity: PlanIdentity): { started: boolean; begun: boolean; earlierCommits: boolean; completed: string[]; next: string | null } { + const plan = this.#store.getPlan(identity), { started, begun, earlierCommits, finished } = this.#store.executeProgress(identity, plan.revision); + const done = new Set(finished); + return { started, begun, earlierCommits, completed: plan.items.filter(item => done.has(item.id)).map(item => item.id), + next: plan.items.find(item => !done.has(item.id))?.id ?? null }; + } + /** Whether a run of this task is still in progress here (its last write may still be to come). */ + busy(identity: PlanIdentity): boolean { return this.#inFlight.has(identityKey(identity)); } + /** Held in #inFlight from its start to its return, so close() can await it. */ + #track(key: string, run: Promise): Promise { this.#inFlight.set(key, run.catch(() => undefined)); - try { return await run; } - finally { this.#inFlight.delete(key); } + return run.finally(() => this.#inFlight.delete(key)); } /** * Shutdown, after the coordinator's close: await every run in progress. A run's pause or escalation after its last @@ -302,53 +337,81 @@ export class ItemExecutor { async close(): Promise { while (this.#inFlight.size) await Promise.all([...this.#inFlight.values()]); } - async #runTask(identity: PlanIdentity, options: { fromItem?: string }): Promise { + /** + * The synchronous start of a run: owed work first, then the first item's admission. With `strict`, a run that cannot + * admit its first item throws instead of returning a stopped outcome. + */ + #begin(identity: PlanIdentity, options: { fromItem?: string; claimRequeue?: boolean }, strict: boolean): { outcome: ExecutionOutcome } | Run { const plan = this.#store.getPlan(identity); const start = options.fromItem ? plan.items.findIndex(item => item.id === options.fromItem) : 0; if (start < 0) throw new Error('Unknown plan item.'); - const done: string[] = [], unchanged: string[] = []; - const stopped = (item: string, state: string, reason: string | null): ExecutionOutcome => ({ kind: 'stopped', item, state, reason, completed: [...done] }); + const run: Run = { plan, index: start, attempt: undefined!, done: [], unchanged: [], stopped: undefined! }; + run.stopped = (item, state, reason) => ({ kind: 'stopped', item, state, reason, completed: [...run.done] }); + const notStarted = (reason: string, error?: unknown): { outcome: ExecutionOutcome } => { + if (strict) throw error ?? new GuardRefusal(reason); + return { outcome: run.stopped(options.fromItem ?? plan.items[start]!.id, 'not started', reason) }; + }; // Shutdown began (admission is closed): pay nothing owed and start nothing; the next run after restart does. - if (this.#runner.closing) return stopped(options.fromItem ?? plan.items[start]!.id, 'not started', 'The review server is shutting down.'); + if (this.#runner.closing) return notStarted('The review server is shutting down.', new ShuttingDownError()); // An earlier run of this task that is still finishing (its storage release) settles its own findings and pause. - if (this.#runner.isActive(identity)) - return stopped(options.fromItem ?? plan.items[start]!.id, 'not started', 'An earlier run of this task is still finishing; start it again when that run has ended.'); + if (this.#runner.isActive(identity)) return notStarted('An earlier run of this task is still finishing; try again when that run has ended.'); // A safety finding not yet acted on (a failed write, a human gate at the time) goes to needs human first. for (const earlier of this.#store.getAttempts(identity)) { const finding = this.#findings.get(earlier.id); - if (finding) return this.#escalate(identity, earlier, finding, stopped, [], true); + if (finding) return this.#settledOrRefused(this.#escalate(identity, earlier, finding, run.stopped, [], true), strict); } // A scope finding whose pause was never recorded (a failed write, the write gate, a crash) pauses now, before any item. const owed = this.#unpausedScopeFinding(identity); - if (owed) return this.#pause(identity, owed.row, owed.result, stopped, [], true); + if (owed) return this.#settledOrRefused(this.#pause(identity, owed.row, owed.result, run.stopped, [], true), strict); // Continuing after a scope pause (plan-format.md: reconcile the executed prefix with the audited head, validate the // remaining items from that checkpoint) is not built yet (#88), so a paused task runs no further items: fail closed. const checkpoint = this.#store.latestCheckpoint(identity); if (checkpoint) - return stopped(options.fromItem ?? plan.items[start]!.id, 'not started', - `${checkpoint.item} changed files outside its plan item. Continuing after a scope pause is not supported yet (#88), so this task runs no further items.`); - /** Where the next item must start: the context the previous item left, or the current one for the first item. */ - let expected: InvocationContext | null = null; - for (const item of plan.items.slice(start)) { - // Admission reads the context in this same turn, so it cannot notice a change saved during an earlier item. - const current = this.#store.currentContext(identity); - // After an item, the task is still running unless someone changed its status meanwhile: then the run stops. - if (expected && this.#store.getTask(identity).status !== 'running') - return stopped(item.id, 'not started', `The task's status changed to ${this.#store.getTask(identity).status} during the run; ${item.id} was not started.`); - if (this.#store.getPlan(identity).revision !== plan.revision) - return stopped(item.id, 'not started', `The plan changed to a new revision during the run; review it before running ${item.id}.`); - if (expected && !sameContext(current, expected)) - return stopped(item.id, 'not started', `The task's snapshot or assignment changed during the run; review it before running ${item.id}.`); - let attempt: AttemptRecord; - try { - attempt = this.#runner.start(identity, { - expectedStateVersion: this.#store.getTask(identity).stateVersion, kind: 'execute', item: item.id, - expectedContext: current, deadline: Date.now() + this.#deadlineMs, - }); - } catch (error) { - if (!refusal(error)) throw error; - return stopped(item.id, 'not started', error instanceof Error ? error.message : String(error)); - } + return notStarted(`${checkpoint.item} changed files outside its plan item. Continuing after a scope pause is not supported yet (#88), so this task runs no further items.`); + const admitted = this.#admit(identity, run, null, options.claimRequeue === true, strict); + if (!('id' in admitted)) return { outcome: admitted }; + run.attempt = admitted; + return run; + } + /** An owed escalation or pause: its outcome stands; one that could not be written refuses a strict start. */ + #settledOrRefused(outcome: ExecutionOutcome, strict: boolean): { outcome: ExecutionOutcome } { + if (strict && outcome.kind === 'stopped') throw new GuardRefusal(outcome.reason ?? 'The task cannot start.'); + return { outcome }; + } + /** + * Admit `run.plan.items[run.index]`. `expected` is the context the previous item left (null for the first). A refusal + * ends the run as not started, or, with `strict`, is thrown as it is, for the caller's user action to record. + */ + #admit(identity: PlanIdentity, run: Run, expected: InvocationContext | null, claimRequeue: boolean, strict: boolean): AttemptRecord | ExecutionOutcome { + const item = run.plan.items[run.index]!; + // Admission reads the context in this same turn, so it cannot notice a change saved during an earlier item. + const current = this.#store.currentContext(identity); + // After an item, the task is still running unless someone changed its status meanwhile: then the run stops. + if (expected && this.#store.getTask(identity).status !== 'running') + return run.stopped(item.id, 'not started', `The task's status changed to ${this.#store.getTask(identity).status} during the run; ${item.id} was not started.`); + if (this.#store.getPlan(identity).revision !== run.plan.revision) { + const reason = `The plan changed to a new revision during the run; review it before running ${item.id}.`; + // A strict begin reads the plan in this same turn, so this cannot fire there; if it ever did, it must refuse, not settle. + if (strict) throw new GuardRefusal(reason); + return run.stopped(item.id, 'not started', reason); + } + if (expected && !sameContext(current, expected)) + return run.stopped(item.id, 'not started', `The task's snapshot or assignment changed during the run; review it before running ${item.id}.`); + try { + return this.#runner.start(identity, { + expectedStateVersion: this.#store.getTask(identity).stateVersion, kind: 'execute', item: item.id, + expectedContext: current, deadline: Date.now() + this.#deadlineMs, ...(claimRequeue ? { claimRequeue } : {}), + }); + } catch (error) { + if (!refusal(error) || strict) throw error; + return run.stopped(item.id, 'not started', error instanceof Error ? error.message : String(error)); + } + } + /** The rest of a run once its first item is admitted: each item settles, then the next is admitted. */ + async #continue(identity: PlanIdentity, run: Run): Promise { + const { done, unchanged, stopped } = run; + for (;;) { + const item = run.plan.items[run.index]!, attempt = run.attempt; await this.#runner.settled(identity); const row = this.#store.getAttempt(identity, attempt.id); // Only the runner's own audit records a finding; it wins over any later stale or stop outcome. The terminal write @@ -367,10 +430,13 @@ export class ItemExecutor { if (result.outOfScope.length) return this.#pause(identity, row, result, stopped, done); const snapshotId = result.unchanged ? row.context.snapshotId : this.#store.snapshotWithHead(identity, result.head); if (!snapshotId) throw new Error(`The snapshot of ${item.id}'s commit is missing.`); + if (++run.index >= run.plan.items.length) return { kind: 'executed', items: done, unchanged }; // The item's own commit (recordHistory) raised the context generation by exactly one; any other change is not ours. - expected = { ...row.context, snapshotId, stateVersion: row.context.stateVersion + (result.unchanged ? 0 : 1) }; + const expected = { ...row.context, snapshotId, stateVersion: row.context.stateVersion + (result.unchanged ? 0 : 1) }; + const next = this.#admit(identity, run, expected, false, false); + if (!('id' in next)) return next; + run.attempt = next; } - return { kind: 'executed', items: done, unchanged }; } /** * A safety finding sends the task to needs human (plan-format.md, "After each run"). From running, queued or a review @@ -410,7 +476,8 @@ export class ItemExecutor { if (!(error instanceof GuardRefusal) && !(owed && error instanceof ShuttingDownError)) throw error; return stopped(item, owed ? 'not started' : row.state, `${violation} The task could not be moved to needs human yet: ${(error as Error).message}`); } - this.#findings.settle(row.id); + // Inside a caller's transaction (begin, #91 part 2) the move can still roll back; the finding stays owed until it commits. + this.#store.afterCommit(() => this.#findings.settle(row.id)); return { kind: 'needs human', item, reason: violation, completed: [...done] }; } /** The earliest completed execute attempt whose out-of-scope files have no checkpoint yet (its pause was lost). */ diff --git a/runner/lifecycle.ts b/runner/lifecycle.ts index bf9f7b17..e2029a17 100644 --- a/runner/lifecycle.ts +++ b/runner/lifecycle.ts @@ -60,6 +60,16 @@ export function settleWith(capability?: ShutdownCapability): (fn: () => T) => return capability ? fn => capability.run(fn) : fn => fn(); } +/** + * Quote text that is not codeboost's (a Docker label, a file name) for one line of terminal output. JSON quoting escapes + * C0 controls; this also escapes what it leaves literal and a terminal may still act on: C1 controls (U+0080-U+009F, NEL + * among them), the Unicode line and paragraph separators, and invisible format and bidi characters that reorder text. + */ +export function quoteForTerminal(text: string): string { + return JSON.stringify(text).replace(/[\u0080-\u009f\u200b-\u200f\u2028-\u202e\u2060-\u206f\ufeff]/gu, + char => `\\u${char.charCodeAt(0).toString(16).padStart(4, '0')}`); +} + 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; diff --git a/runner/production.ts b/runner/production.ts index e15e6b2c..7b2ecc4e 100644 --- a/runner/production.ts +++ b/runner/production.ts @@ -10,7 +10,7 @@ import { identityKey } from '../core/identity.ts'; import type { RunnerDeps } from './coordinator.ts'; import { DEFAULT_DIAGNOSTICS_CAP_BYTES } from './diagnostics.ts'; import { executionDeps, SafetyFindings, type AgentLauncher, type ExecutionSources } from './execution.ts'; -import { isUuidV4, type ShutdownCapability } from './lifecycle.ts'; +import { isUuidV4, quoteForTerminal, type ShutdownCapability } from './lifecycle.ts'; import { recoverStartup, removalCommand, type RecoveryDeps, type RecoveryReport, type RunnerLock } from './recovery.ts'; import type { ReviewService } from './review.ts'; import { openRunnerRepository, ownerOnlyDirectory } from './runner-repository.ts'; @@ -123,6 +123,15 @@ export function claudeLauncher(o: { imageId: string; runnerRoot: string; runnerO }; } +/** + * What startup recovery left for a person (runner-lifecycle.md: never removed automatically), one line each. Docker + * labels and file names are not codeboost's, so each is quoted: a newline or control character in one cannot forge a line. + */ +export function recoveryWarnings(report: Pick): string[] { + return [...report.unmatchedStorage.map(attemptId => `Task storage labelled with attempt ${quoteForTerminal(attemptId)} matches no attempt of this database; it was left in place.`), + ...report.unknownEntries.map(entry => `Unknown entry in the runner's attempt directory, left in place: ${quoteForTerminal(entry)}`)]; +} + export interface RunnerAssembly { readonly deps: RunnerDeps; readonly sources: ExecutionSources; diff --git a/runner/store.ts b/runner/store.ts index 4369d0ac..bc06ef66 100644 --- a/runner/store.ts +++ b/runner/store.ts @@ -788,6 +788,23 @@ export class Store { FROM attempts WHERE plan_key=? ORDER BY rowid DESC LIMIT ?) ORDER BY row_order`).all(identityKey(identity), limit) .map(row => { const { result: _result, ...attempt } = this.#attemptRecord(row); return { ...attempt, hasResult: row.has_result === 1 }; }); } + /** + * What `ItemExecutor.progress` needs about a task's execute attempts, without loading or decoding any attempt's result + * (up to 1 MiB each): a status poll asks for it (#91 part 2). `finished` lists each item completed at `revision` once. + */ + executeProgress(identity: PlanIdentity, revision: number): { started: boolean; begun: boolean; earlierCommits: boolean; finished: string[] } { + const key = identityKey(identity), rev = "json_extract(context,'$.planRevision')"; + const row = this.#get(`SELECT COUNT(*) > 0 AS started, COALESCE(SUM(${rev} = ?), 0) > 0 AS begun + FROM attempts WHERE plan_key=? AND kind='execute' AND item IS NOT NULL`, revision, key)!; + // Any writable kind's commit counts, as for hasRunnerCommit. Unlike it, a completed result that does not say it is unchanged + // (not valid JSON, or no `unchanged` field) counts as a commit: this check refuses a run, so it fails closed. + const earlier = this.#get(`SELECT 1 AS found FROM attempts WHERE plan_key=? AND kind IN (${WRITABLE_KINDS.map(() => '?').join(',')}) + AND state='completed' AND ${rev} != ? AND (NOT json_valid(result) OR COALESCE(json_extract(result,'$.unchanged'), 0) = 0) LIMIT 1`, + key, ...WRITABLE_KINDS, revision); + const finished = this.#db.prepare(`SELECT DISTINCT item FROM attempts WHERE plan_key=? AND kind='execute' AND item IS NOT NULL + AND state='completed' AND ${rev} = ?`).all(key, revision).map(entry => entry.item as string); + return { started: row.started === 1, begun: row.begun === 1, earlierCommits: !!earlier, finished }; + } 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)); } diff --git a/test/cli.test.ts b/test/cli.test.ts index 42c0a2ea..9a1832c9 100644 --- a/test/cli.test.ts +++ b/test/cli.test.ts @@ -2,7 +2,7 @@ import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { spawn, spawnSync } from 'node:child_process'; -import { chmodSync, existsSync, unlinkSync } from 'node:fs'; +import { chmodSync, existsSync, renameSync, unlinkSync } from 'node:fs'; import { BASE_IMAGE, CLAUDE_VERSION, CODEX_VERSION } from '../agents/container/image.ts'; import { randomUUID } from 'node:crypto'; import { fileURLToPath } from 'node:url'; @@ -52,7 +52,7 @@ function review() { const { CLAUDE_CODE_OAUTH_TOKEN: _token, ...env } = process.env; return spawnSync(process.execPath, [cli, '--config', config, ...args], { encoding: 'utf8', env: { ...env, HOME: join(root, 'home') }, timeout: 20_000 }); }; - return { run, fakeDocker, start }; + return { run, fakeDocker, start, database: demo.database }; } it('refuses a runner startup without a token with its message, exit 1 and the lock released', () => { @@ -125,3 +125,16 @@ it('ignores a runner block in a demo configuration opened with --config, and ser expect(await exited).toBe(0); } finally { child.kill('SIGKILL'); } }); + +it('refuses with a message, not a stack, when the database path changes during a runner startup', async () => { + const { fakeDocker, start, database } = review(), docker = fakeDocker(); + const cli = start(docker); + try { + await cli.until(() => cli.out().includes('recovering what an earlier run left'), 'startup recovery to begin'); + // The locked file moves away and another takes its name while startup is still running. + renameSync(database, `${database}.moved`); writeFileSync(database, ''); + docker.release(); + expect(await cli.exited).toBe(1); + expect(cli.err().trim()).toBe('The database path changed while opening. Refusing to start.'); + } finally { docker.release(); cli.child.kill('SIGKILL'); } +}); diff --git a/test/issues.test.ts b/test/issues.test.ts index 5c96f27b..986bdad4 100644 --- a/test/issues.test.ts +++ b/test/issues.test.ts @@ -279,6 +279,15 @@ describe('issue text for an execute prompt (#91)', () => { expect((await gateway(pages).gateway.issueText(7)).comments).toEqual([]); await expect(gateway([...pages, [comment('outsider', 'one more')]]).gateway.issueText(7)).rejects.toThrow(/more than 1000 comments/); }); + it('refuses a title and body already over the prompt budget before reading any collaborator or comment page', async () => { + const { gateway: g, calls } = gateway([[comment('member', 'never read')]], rawIssue({ body: 'x'.repeat(40_000) })); + await expect(g.issueText(7)).rejects.toThrow(/larger than the 32 KiB/); + expect(calls.map(args => args[5])).toEqual(['repos/owner/repo/issues/7']); + // Small in bytes, over the budget once the prompt escapes it ('<' becomes \u003c): refused just as early. + const escaped = gateway([[comment('member', 'never read')]], rawIssue({ body: '<'.repeat(6_000) })); + await expect(escaped.gateway.issueText(7)).rejects.toThrow(/larger than the 32 KiB/); + expect(escaped.calls.map(args => args[5])).toEqual(['repos/owner/repo/issues/7']); + }); it('stops on the caller\'s abort', async () => { const controller = new AbortController(); controller.abort(new Error('stopped')); diff --git a/test/runner-execution.test.ts b/test/runner-execution.test.ts index 12ca94ce..8f315fe0 100644 --- a/test/runner-execution.test.ts +++ b/test/runner-execution.test.ts @@ -1,3 +1,4 @@ +import { GuardRefusal } from '../runner/lifecycle.ts'; import { mkdtempSync, readFileSync, rmSync, statSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { DatabaseSync } from 'node:sqlite'; @@ -339,6 +340,78 @@ describe('item execution', () => { expect(h.log).toEqual([]); expect(h.store.getAttempts(identity)[0]!.diagnostic).toMatch(/Issue retrieval timed out/); }); + it('begin admits the first item before it returns, and the rest of the run follows (#91 part 2)', async () => { + const h = setup(); + const begun = h.executor.begin(identity); + // Admitted synchronously, inside whatever transaction the caller holds. + expect(h.store.getAttempts(identity).map(row => [row.id, row.item])).toEqual([[begun.attemptId, 'P1']]); + expect(await begun.outcome).toEqual({ kind: 'executed', items: ['P1', 'P2'], unchanged: [] }); + expect(h.executor.progress(identity)).toEqual({ started: true, begun: true, earlierCommits: false, completed: ['P1', 'P2'], next: null }); + }); + it('begin throws the admission refusal itself, where runTask reports it as not started', async () => { + const h = setup(); + h.store.cancelTask(identity, h.store.getTask(identity).stateVersion, randomUUID()); + expect(() => h.executor.begin(identity)).toThrow(GuardRefusal); + expect(await h.executor.runTask(identity)).toMatchObject({ kind: 'stopped', state: 'not started' }); + expect(h.store.getAttempts(identity)).toEqual([]); + }); + it('keeps a safety finding owed when the user action around begin rolls back after the escalation', () => { + const h = setup(); + const attempt = h.store.admitAttempt(identity, { expectedStateVersion: h.store.getTask(identity).stateVersion, kind: 'execute', item: 'P1', + expectedContext: h.store.currentContext(identity), deadline: Date.now() + 60_000 }); + h.store.markRunning(identity, attempt.id); + h.store.settleAttempt(identity, attempt.id, { firstReason: null, exitCode: 1, valid: false }); + const save = h.store.recordSafetyFinding; + h.store.recordSafetyFinding = () => { throw Object.assign(new Error('disk full'), { code: 'ERR_SQLITE_ERROR' }); }; + h.findings.record(attempt.id, 'Safety violation: test'); + h.store.recordSafetyFinding = save; + // The escalation's write rolls back with the action, so the finding must still be owed afterwards. + expect(() => h.store.userAction(identity, { actionId: randomUUID(), kind: 'resume', request: {} }, () => { h.executor.begin(identity); throw new Error('commit failed'); })).toThrow(/commit failed/); + expect(h.store.getTask(identity).status).toBe('running'); + expect(h.findings.get(attempt.id)).toBe('Safety violation: test'); + }); + it('begin throws ShuttingDownError once the runner stopped admission', () => { + const h = setup(); + h.runner.rejectAdmission(); + expect(() => h.executor.begin(identity)).toThrow(ShuttingDownError); + expect(h.store.getAttempts(identity)).toEqual([]); + }); + it('begin pays nothing owed once the runner stopped admission, and the finding stays owed', () => { + const h = setup(); + const attempt = h.store.admitAttempt(identity, { expectedStateVersion: h.store.getTask(identity).stateVersion, kind: 'execute', item: 'P1', + expectedContext: h.store.currentContext(identity), deadline: Date.now() + 60_000 }); + h.store.markRunning(identity, attempt.id); + h.store.settleAttempt(identity, attempt.id, { firstReason: null, exitCode: 1, valid: false }); + const save = h.store.recordSafetyFinding; + h.store.recordSafetyFinding = () => { throw Object.assign(new Error('disk full'), { code: 'ERR_SQLITE_ERROR' }); }; + h.findings.record(attempt.id, 'Safety violation: test'); + h.store.recordSafetyFinding = save; + h.runner.rejectAdmission(); + expect(() => h.executor.begin(identity)).toThrow(ShuttingDownError); + expect(h.store.getTask(identity).status).toBe('running'); + expect(h.findings.get(attempt.id)).toBe('Safety violation: test'); + }); + it('begin refuses a second run of a task while the first is still in progress', async () => { + const h = setup(); + const first = h.executor.begin(identity); + expect(() => h.executor.begin(identity)).toThrow(/still finishing/); + await first.outcome; + }); + it('begin refuses a second run between the first run\'s items, when no attempt is active', async () => { + const h = setup(); + const settled = h.runner.settled.bind(h.runner); + let between: unknown = null; + h.runner.settled = async id => { + await settled(id); + // P1's job is gone and P2 is not admitted yet: only the executor knows the run is still going. + if (between === null) { expect(h.runner.isActive(identity)).toBe(false); try { h.executor.begin(identity); between = 'admitted'; } catch (error) { between = error; } } + }; + const first = h.executor.begin(identity); + expect(await first.outcome).toEqual({ kind: 'executed', items: ['P1', 'P2'], unchanged: [] }); + expect(between).toBeInstanceOf(GuardRefusal); + expect((between as Error).message).toMatch(/still finishing/); + expect(h.store.getAttempts(identity).map(row => row.item)).toEqual(['P1', 'P2']); + }); it('commits a rename as the manifest audited it', async () => { const { executor, commits } = setup({ manifests: { P1: manifest([change('a.ts', { kind: 'rename', oldPath: 'old.ts' })], { digest: 'renamed' }) } }); await executor.runTask(identity); diff --git a/test/runner-production.test.ts b/test/runner-production.test.ts index e655fb0c..692b7ca2 100644 --- a/test/runner-production.test.ts +++ b/test/runner-production.test.ts @@ -6,7 +6,7 @@ import { afterEach, describe, expect, it, vi } from 'vitest'; import { createDemo } from '../scripts/demo.ts'; import { ReviewService } from '../runner/review.ts'; import { RUNNER_NOT_CONFIGURED, RUNNER_NOT_IN_DEMO, startServer } from '../web/server.ts'; -import { claudeLauncher, dRecoveryDeps, ISSUE_REUSE_MS, parseRunnerConfig, RUNNER_CREDENTIAL_MISSING, setUpRunner } from '../runner/production.ts'; +import { claudeLauncher, dRecoveryDeps, ISSUE_REUSE_MS, parseRunnerConfig, recoveryWarnings, RUNNER_CREDENTIAL_MISSING, setUpRunner } from '../runner/production.ts'; import type { RecoveryDeps } from '../runner/recovery.ts'; import type { AgentAdapterRequest } from '../agents/adapters/types.ts'; import { readCapturedFile } from '../agents/container/profile.ts'; @@ -194,7 +194,7 @@ describe('server with a runner setup', () => { expect(view.retryable).toBe(false); const response = await fetch(`${origin}/api/runner`, { method: 'POST', headers: { ...headers, 'content-type': 'application/json' }, body: JSON.stringify({ action: 'retry', attemptId: view.attempts[0]!.id, expectedStateVersion: view.stateVersion, actionId: randomUUID() }) }); - expect(await response.json()).toEqual({ error: expect.stringMatching(/Retrying a plan item on its own is not supported/) }); + expect(await response.json()).toEqual({ error: expect.stringMatching(/run again by resuming the task/) }); expect(app.service.store.getAttempts(demo.identity)).toHaveLength(1); }); it('tells a review without a runner block to add one, and a demo that it never runs the runner', async () => { @@ -211,6 +211,19 @@ describe('server with a runner setup', () => { }); }); +describe('recovery warnings', () => { + it('quotes names that are not codeboost\'s, so a newline cannot forge a line of output', () => { + const lines = recoveryWarnings({ unmatchedStorage: ['x\nRecovered 0 interrupted attempt(s).'], unknownEntries: ['/root/attempts/evil\nReview ready: http://x', '/root/attempts/\u001b[2Jx'] }); + expect(lines).toHaveLength(3); + for (const line of lines) expect(line).not.toMatch(/[\n\u001b]/); + expect(lines[1]).toContain('"/root/attempts/evil\\nReview ready: http://x"'); + // What JSON quoting leaves literal and a terminal still acts on: NEL (C1), the line and paragraph separators, bidi. + const unicode = recoveryWarnings({ unmatchedStorage: [], unknownEntries: ['a\u0085b\u2028c\u2029d\u202ee\u009bf'] }); + expect(unicode[0]).not.toMatch(/[\u0080-\u009f\u2028\u2029\u202e]/u); + expect(unicode[0]).toContain('"a\\u0085b\\u2028c\\u2029d\\u202ee\\u009bf"'); + }); +}); + describe('D adapters', () => { it('passes D\'s own handles through, and turns every unowned object into the command that removes it', async () => { const handle = Object.freeze({ runnerOwner: OWNER, attemptId: randomUUID(), allocationId: randomUUID() }); diff --git a/test/runner-start.test.ts b/test/runner-start.test.ts new file mode 100644 index 00000000..9c769b94 --- /dev/null +++ b/test/runner-start.test.ts @@ -0,0 +1,378 @@ +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { randomUUID } from 'node:crypto'; +import { request as httpRequest } from 'node:http'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { createDemo } from '../scripts/demo.ts'; +import { startServer } from '../web/server.ts'; +import { Store } from '../runner/store.ts'; +import type { ReviewService } from '../runner/review.ts'; +import type { RunnerDeps } from '../runner/coordinator.ts'; +import { SafetyFindings, type ExecutionSources } from '../runner/execution.ts'; +import type { AttemptKind } from '../runner/lifecycle.ts'; + +vi.setConfig({ testTimeout: 20_000 }); +const roots: string[] = [], cleanups: (() => Promise | void)[] = []; +afterEach(async () => { + for (const cleanup of cleanups.splice(0).reverse()) await cleanup(); + for (const root of roots.splice(0)) rmSync(root, { recursive: true, force: true }); +}); +const OWNER = 'c'.repeat(32); +type App = Awaited>; +/** + * A server with a runner whose preparation always fails: an admitted item ends `failed` without an agent, which is + * enough to see what start and resume admit. `before` shapes the Store before the runner exists, as recovery would. + */ +async function serve(options: { kinds?: AttemptKind[]; before?: (service: ReviewService) => void; findings?: (findings: SafetyFindings, service: ReviewService) => void } = {}) { + const root = mkdtempSync(join(tmpdir(), 'codeboost-start-')); roots.push(root); + const demo = createDemo(join(root, 'demo')); + const app = await startServer({ ...demo }, 0, undefined, undefined, 2_000, undefined, undefined, undefined, async service => { + options.before?.(service); + const deps: RunnerDeps = { runnerOwner: OWNER, kinds: options.kinds ?? ['execute'], prepare: async () => { throw new Error('no agent here'); }, + cleanupPreparation: async () => undefined, start: () => { throw new Error('no agent here'); }, validate: () => null }; + const sources: ExecutionSources = { planContext: () => service.planContext(), issue: () => ({ number: 1, title: '', body: '', comments: [] }), lessons: () => [], vendor: () => 'claude' }; + const findings = new SafetyFindings(service.store); + options.findings?.(findings, service); + return { deps, sources, findings, recovery: { finalized: [], requeue: [], removedDirectories: [], unknownEntries: [], unmatchedStorage: [], repairedMerges: [] } }; + }); + let closed = false; + const close = async () => { if (!closed) { closed = true; await app.close(); } }; + cleanups.push(close); + return { app, close, database: demo.database, identity: demo.identity, store: app.service.store, items: app.service.store.getPlan(demo.identity).items.map(item => item.id) }; +} +const view = async (app: App) => (await fetch(`${new URL(app.url).origin}/api/runner`, { headers: { 'x-codeboost-token': app.token } })).json() as Promise>; +async function act(app: App, action: string, request: { expectedStateVersion?: number; actionId?: string } = {}) { + const { stateVersion } = await view(app); + const response = await fetch(`${new URL(app.url).origin}/api/runner`, { method: 'POST', headers: { 'x-codeboost-token': app.token, 'content-type': 'application/json' }, + body: JSON.stringify({ action, expectedStateVersion: request.expectedStateVersion ?? stateVersion, actionId: request.actionId ?? randomUUID() }) }); + return { status: response.status, body: await response.json() as Record }; +} + +/** Before the runner exists: queue the task and leave one execute attempt of its first item, failed. */ +function failedFirstItem(service: ReviewService, budgetMs?: number) { + const s = service.store, id = service.config.identity; + s.transitionTask(id, s.getTask(id).stateVersion, 'queued'); + const attempt = s.admitAttempt(id, { expectedStateVersion: s.getTask(id).stateVersion, kind: 'execute', item: s.getPlan(id).items[0]!.id, + expectedContext: s.currentContext(id), deadline: Date.now() + 60_000, ...(budgetMs ? { budgetMs } : {}) }); + s.markRunning(id, attempt.id); + s.settleAttempt(id, attempt.id, { firstReason: null, exitCode: 1, valid: false }); + return attempt; +} + +/** Before the runner exists: as failedFirstItem, but the first item completed and made a runner commit. */ +function committedFirstItem(service: ReviewService, unchanged = false, outOfScope: string[] = [], kind: AttemptKind = 'execute') { + const s = service.store, id = service.config.identity; + s.transitionTask(id, s.getTask(id).stateVersion, 'queued'); + const item = s.getPlan(id).items[0]!.id, snapshot = s.getSnapshot(id), head = 'f'.repeat(40); + const attempt = s.admitAttempt(id, { expectedStateVersion: s.getTask(id).stateVersion, kind, item, + expectedContext: s.currentContext(id), deadline: Date.now() + 60_000 }); + s.markRunning(id, attempt.id); + // As finish() returns it: an unchanged item makes no commit, so it has no history. + if (unchanged) s.settleAttempt(id, attempt.id, { firstReason: null, exitCode: 0, valid: true, result: { head: snapshot.head, unchanged: true, inScope: [], outOfScope: [] } }); + else s.settleAttempt(id, attempt.id, { firstReason: null, exitCode: 0, valid: true, result: { head, unchanged: false, inScope: [], outOfScope }, + history: { base: snapshot.base, head, entries: [{ sha: head, owner: item, origin: 'owned', sourceSha: null }] } }); +} +/** Save a new plan revision, as a person editing the plan does. */ +function revise(service: ReviewService) { + const s = service.store, id = service.config.identity, plan = s.getPlan(id); + s.importRevision(JSON.stringify({ ...plan, revision: plan.revision + 1, summary: `${plan.summary} (revised)` }), 'json', service.planContext(), plan.revision); +} + +describe('start (#91 part 2)', () => { + it('moves a task in review to queued and admits its first item, in one action', async () => { + const { app, identity, store, items } = await serve(); + expect(store.getTask(identity).status).toBe('in review'); + expect(await view(app)).toMatchObject({ startable: true, resumable: false }); + const started = await act(app, 'start'); + expect(started.body.result).toMatchObject({ outcome: 'started', item: items[0] }); + const [attempt] = store.getAttempts(identity); + expect(attempt).toMatchObject({ id: started.body.result.attemptId, kind: 'execute', item: items[0] }); + await app.executor!.close(); + // The run ended with its first item (preparation fails here); nothing after it was started. + expect(store.getAttempts(identity)).toHaveLength(1); + expect(await view(app)).toMatchObject({ startable: false }); + }); + it('rolls the move to queued back when admission refuses the first item, and records the refusal', async () => { + // Only earlier-revision attempts, the budget spent, and back in review: runChoice passes, so the action moves the + // task to queued and admission itself refuses. + const { app, identity, store } = await serve({ before: service => { + failedFirstItem(service, 1); revise(service); + const s = service.store, id = service.config.identity; + s.transitionTask(id, s.getTask(id).stateVersion, 'in review'); + } }); + await new Promise(resolve => setTimeout(resolve, 5)); + const before = store.getTask(identity), attempts = store.getAttempts(identity).length, actionId = randomUUID(); + const queue = vi.spyOn(store, 'transitionTask'); + const refused = await act(app, 'start', { actionId }); + expect(refused).toMatchObject({ status: 409, body: { error: expect.stringMatching(/time budget has run out/) } }); + expect(queue).toHaveBeenCalledWith(identity, before.stateVersion, 'queued'); + // The move rolled back with the refusal, so the budget's effect (idle running or queued only) left it in review. + expect(store.getTask(identity)).toMatchObject({ status: 'in review', stateVersion: before.stateVersion }); + expect(store.getAttempts(identity)).toHaveLength(attempts); + // Recorded: the same action ID replays the refusal. + expect(await act(app, 'start', { actionId, expectedStateVersion: before.stateVersion })).toMatchObject({ status: 409, body: refused.body }); + }); + it('is refused once the plan has started, and resume runs a first item that failed again', async () => { + const { app, identity, store, items } = await serve(); + await act(app, 'start'); await app.executor!.close(); + expect(store.getAttempts(identity)[0]!.state).toBe('failed'); + expect(await view(app)).toMatchObject({ startable: false, resumable: true }); + expect((await act(app, 'start')).body.error).toMatch(/resume the task instead/); + expect((await act(app, 'resume')).body.result).toMatchObject({ outcome: 'started', item: items[0] }); + }); +}); + +describe('status polling (#91 part 2)', () => { + it('reads progress once per poll with a targeted query, never the full attempt history', async () => { + const { app, store } = await serve({ before: service => { failedFirstItem(service); } }); + const history = vi.spyOn(store, 'getAttempts'), progress = vi.spyOn(store, 'executeProgress'); + expect(await view(app)).toMatchObject({ resumable: true }); + expect(history).not.toHaveBeenCalled(); + expect(progress).toHaveBeenCalledTimes(1); + }); +}); + +describe('resume (#91 part 2)', () => { + it('claims the requeue recovery left and continues from the first unfinished item', async () => { + let interrupted = ''; + const { app, identity, store, items } = await serve({ before: service => { + const s = service.store, id = service.config.identity; + s.transitionTask(id, s.getTask(id).stateVersion, 'queued'); + const attempt = s.admitAttempt(id, { expectedStateVersion: s.getTask(id).stateVersion, kind: 'execute', item: s.getPlan(id).items[0]!.id, + expectedContext: s.currentContext(id), deadline: Date.now() + 60_000 }); + s.markRunning(id, attempt.id); + // What startup recovery does to an attempt a crash interrupted: failed "Interrupted", task requeued. + s.recoverInterrupted(Date.now()); + interrupted = attempt.id; + } }); + expect(store.getTask(identity).requeuePending).toBe(true); + expect(await view(app)).toMatchObject({ startable: false, resumable: true }); + const resumed = await act(app, 'resume'); + expect(resumed.body.result).toMatchObject({ outcome: 'started', item: items[0] }); + expect(store.getTask(identity).requeuePending).toBe(false); + expect(store.getAttempts(identity).map(row => row.id)).toEqual([interrupted, resumed.body.result.attemptId]); + }); + it('continues a task that stopped between items from the next item', async () => { + const { app, identity, store, items } = await serve({ before: service => { + const s = service.store, id = service.config.identity; + s.transitionTask(id, s.getTask(id).stateVersion, 'queued'); + const attempt = s.admitAttempt(id, { expectedStateVersion: s.getTask(id).stateVersion, kind: 'execute', item: s.getPlan(id).items[0]!.id, + expectedContext: s.currentContext(id), deadline: Date.now() + 60_000 }); + s.markRunning(id, attempt.id); + // What a completed, unchanged item leaves (no commit, so no history). + s.settleAttempt(id, attempt.id, { firstReason: null, exitCode: 0, valid: true, result: { head: s.getSnapshot(id).head, unchanged: true, inScope: [], outOfScope: [] } }); + } }); + expect(store.getTask(identity).status).toBe('running'); + expect(await view(app)).toMatchObject({ resumable: true, startable: false }); + expect((await act(app, 'resume')).body.result).toMatchObject({ outcome: 'started', item: items[1] }); + }); + it('is refused when no item has run, which start handles instead', async () => { + const { app } = await serve(); + expect((await act(app, 'resume')).body.error).toMatch(/start the task instead/); + }); +}); + +describe('start and resume refusals and races (#91 part 2)', () => { + it('resumes a revised plan from its first item when the earlier revision committed nothing', async () => { + const { app, identity, store, items } = await serve({ before: service => { failedFirstItem(service); revise(service); } }); + expect(await view(app)).toMatchObject({ startable: false, resumable: true }); + expect((await act(app, 'resume')).body.result).toMatchObject({ outcome: 'started', item: items[0] }); + expect(store.getAttempts(identity).at(-1)!.context.planRevision).toBe(store.getPlan(identity).revision); + }); + it('does not count an earlier item that completed unchanged as a commit, and reruns it at the new revision', async () => { + const { app, identity, store, items } = await serve({ before: service => { committedFirstItem(service, true); revise(service); } }); + expect(await view(app)).toMatchObject({ resumable: true, startable: false }); + // Completed at the earlier revision only: the new revision still starts at its first item. + expect((await act(app, 'resume')).body.result).toMatchObject({ outcome: 'started', item: items[0] }); + expect(store.getAttempts(identity).at(-1)!.context.planRevision).toBe(store.getPlan(identity).revision); + }); + it('lets a task whose only attempts were at an earlier, uncommitted revision start again from review', async () => { + const { app, identity, store, items } = await serve({ before: service => { + failedFirstItem(service); revise(service); + const s = service.store, id = service.config.identity; + s.transitionTask(id, s.getTask(id).stateVersion, 'in review'); + } }); + expect(await view(app)).toMatchObject({ startable: true, resumable: false }); + expect((await act(app, 'start')).body.result).toMatchObject({ outcome: 'started', item: items[0] }); + expect(store.getAttempts(identity).at(-1)!.context.planRevision).toBe(store.getPlan(identity).revision); + }); + it('offers a queued task with only earlier-revision attempts both actions, which run the same item', async () => { + for (const action of ['start', 'resume']) { + const { app, items } = await serve({ before: service => { + failedFirstItem(service); revise(service); + const s = service.store, id = service.config.identity; + s.transitionTask(id, s.getTask(id).stateVersion, 'queued'); + } }); + expect(await view(app)).toMatchObject({ startable: true, resumable: true }); + expect((await act(app, action)).body.result).toMatchObject({ outcome: 'started', item: items[0] }); + } + }); + it('refuses both, in the view and the action alike, once an earlier revision left commits (#88)', async () => { + const { app } = await serve({ before: service => { committedFirstItem(service); revise(service); } }); + expect(await view(app)).toMatchObject({ startable: false, resumable: false }); + expect((await act(app, 'resume')).body.error).toMatch(/revised after items of it were committed.*#88/); + expect((await act(app, 'start')).body.error).toMatch(/revised after items of it were committed.*#88/); + }); + it('counts a fix attempt\'s commit at an earlier revision too (#88)', async () => { + const { app } = await serve({ before: service => { committedFirstItem(service, false, [], 'fix'); revise(service); } }); + expect((await act(app, 'start')).body.error).toMatch(/revised after items of it were committed.*#88/); + }); + it('lets admission refuse a spent budget, which moves the idle task to needs human; the view offers nothing', async () => { + const { app, identity, store } = await serve({ before: service => { failedFirstItem(service, 1); } }); + await new Promise(resolve => setTimeout(resolve, 5)); + expect(await view(app)).toMatchObject({ resumable: false }); + expect((await act(app, 'resume')).body.error).toMatch(/time budget has run out/); + expect(store.getTask(identity).status).toBe('needs human'); + }); + it('points start to resume for a running task whose only attempts were at an earlier revision', async () => { + const { app } = await serve({ before: service => { failedFirstItem(service); revise(service); } }); + expect(await view(app)).toMatchObject({ startable: false, resumable: true }); + expect((await act(app, 'start')).body.error).toMatch(/resume the task instead/); + }); + it('names the status when a task that never ran is not running or queued, rather than pointing to start', async () => { + const { app } = await serve({ before: service => { + const s = service.store, id = service.config.identity; + s.transitionTask(id, s.getTask(id).stateVersion, 'needs human'); + } }); + expect((await act(app, 'resume')).body.error).toMatch(/The task is needs human; resume continues a task that is running or queued/); + expect((await act(app, 'start')).body.error).toMatch(/The task is needs human; start runs a task that is in review or queued/); + }); + it('does not point start to resume once every item has run', async () => { + const { app } = await serve({ before: service => { + const s = service.store, id = service.config.identity; + s.transitionTask(id, s.getTask(id).stateVersion, 'queued'); + for (const { id: item } of s.getPlan(id).items) { + const attempt = s.admitAttempt(id, { expectedStateVersion: s.getTask(id).stateVersion, kind: 'execute', item, + expectedContext: s.currentContext(id), deadline: Date.now() + 60_000 }); + s.markRunning(id, attempt.id); + s.settleAttempt(id, attempt.id, { firstReason: null, exitCode: 0, valid: true, result: { head: s.getSnapshot(id).head, unchanged: true, inScope: [], outOfScope: [] } }); + } + } }); + expect(await view(app)).toMatchObject({ startable: false, resumable: false }); + expect((await act(app, 'start')).body.error).toMatch(/The task is running; start runs a task that is in review or queued/); + expect((await act(app, 'resume')).body.error).toMatch(/Every item of this plan has run/); + }); + it('neither offers nor admits start or resume while a run of the task is still finishing between items', async () => { + const { app, identity, store } = await serve({ before: service => { failedFirstItem(service); } }); + expect(await view(app)).toMatchObject({ resumable: true }); + // Between two items no attempt is active; only the executor knows its run is still going. + vi.spyOn(app.executor!, 'busy').mockReturnValue(true); + expect(app.runner!.isActive(identity)).toBe(false); + expect(await view(app)).toMatchObject({ startable: false, resumable: false }); + expect((await act(app, 'resume')).body.error).toMatch(/still finishing; try again when that run has ended/); + expect(store.getAttempts(identity)).toHaveLength(1); + }); + it('points resume to start for a task in review whose only attempts were at an earlier, uncommitted revision', async () => { + const { app } = await serve({ before: service => { + failedFirstItem(service); revise(service); + const s = service.store, id = service.config.identity; + s.transitionTask(id, s.getTask(id).stateVersion, 'in review'); + } }); + expect((await act(app, 'resume')).body.error).toMatch(/start the task instead/); + }); + it('names the status, not resume, when start is refused for a task resume cannot run either', async () => { + const { app } = await serve({ before: service => { + failedFirstItem(service); + const s = service.store, id = service.config.identity; + s.transitionTask(id, s.getTask(id).stateVersion, 'needs human'); + } }); + expect((await act(app, 'start')).body.error).toMatch(/The task is needs human; start runs a task that is in review or queued/); + }); + it('does not offer start when the runner cannot run execute attempts', async () => { + const { app } = await serve({ kinds: ['review'] }); + expect(await view(app)).toMatchObject({ startable: false }); + }); + it('acts on a safety finding owed from an earlier run instead of admitting, and reports it settled', async () => { + let earlier = '', held!: SafetyFindings; + const { app, identity, store } = await serve({ before: service => { earlier = failedFirstItem(service).id; }, + findings: (findings, service) => { + held = findings; + // The durable save failed, so the finding is held only in memory: the executor owes the move to needs human. + const save = service.store.recordSafetyFinding; + service.store.recordSafetyFinding = () => { throw Object.assign(new Error('disk full'), { code: 'ERR_SQLITE_ERROR' }); }; + findings.record(earlier, 'Safety violation: test'); + service.store.recordSafetyFinding = save; + } }); + expect((await act(app, 'resume')).body.result).toEqual({ outcome: 'settled' }); + expect(store.getTask(identity).status).toBe('needs human'); + expect(store.getAttempts(identity)).toHaveLength(1); + // The escalation committed, so the finding is no longer owed. + expect(held.get(earlier)).toBeUndefined(); + }); + it('pauses for an amendment owed from an earlier run instead of admitting, and reports it settled', async () => { + const { app, identity, store } = await serve({ before: service => { committedFirstItem(service, false, ['other.ts']); } }); + expect((await act(app, 'resume')).body.result).toEqual({ outcome: 'settled' }); + expect(store.getTask(identity).status).toBe('needs amendment'); + expect(store.latestCheckpoint(identity)).toMatchObject({ item: store.getPlan(identity).items[0]!.id }); + expect(store.getAttempts(identity)).toHaveLength(1); + expect(await view(app)).toMatchObject({ startable: false, resumable: false }); + }); + it('moves a task in review to queued and then acts on an owed safety finding, in one action', async () => { + let earlier = ''; + const { app, identity, store } = await serve({ before: service => { + earlier = failedFirstItem(service).id; revise(service); + const s = service.store, id = service.config.identity; + s.transitionTask(id, s.getTask(id).stateVersion, 'in review'); + }, findings: (findings, service) => { + const save = service.store.recordSafetyFinding; + service.store.recordSafetyFinding = () => { throw Object.assign(new Error('disk full'), { code: 'ERR_SQLITE_ERROR' }); }; + findings.record(earlier, 'Safety violation: test'); + service.store.recordSafetyFinding = save; + } }); + const before = store.getTask(identity).stateVersion, moves = vi.spyOn(store, 'transitionTask'); + expect((await act(app, 'start')).body.result).toEqual({ outcome: 'settled' }); + // Queued first, then escalated from queued, both in the one action. + expect(moves.mock.calls.map(call => [call[1], call[2]])).toEqual([[before, 'queued'], [before + 1, 'needs human']]); + expect(store.getTask(identity).status).toBe('needs human'); + expect(store.getAttempts(identity)).toHaveLength(1); + }); + it('replays an action ID, and refuses a second resume made against the old state', async () => { + const { app, identity, store } = await serve({ before: service => { failedFirstItem(service); } }); + const { stateVersion } = await view(app), actionId = randomUUID(); + const first = await act(app, 'resume', { expectedStateVersion: stateVersion, actionId }); + // The saved outcome comes back; the runner view beside it is current, so it is not compared. + expect((await act(app, 'resume', { expectedStateVersion: stateVersion, actionId })).body.result).toEqual(first.body.result); + expect((await act(app, 'resume', { expectedStateVersion: stateVersion })).body.error).toMatch(/Stale task state/); + expect(store.getAttempts(identity)).toHaveLength(2); + }); + it('answers 503 to a start whose body finishes arriving after shutdown began, before any refusal, and records nothing', async () => { + // A task start would refuse: during shutdown the 503 still comes first, so no refusal is saved under the action ID. + const { app, close, database, identity } = await serve({ before: service => { + const s = service.store, id = service.config.identity; + s.transitionTask(id, s.getTask(id).stateVersion, 'needs human'); + } }); + const { stateVersion } = await view(app), actionId = randomUUID(), url = new URL(app.url); + const text = JSON.stringify({ action: 'start', expectedStateVersion: stateVersion, actionId }); + let finish!: () => void; + const response = new Promise((resolve, reject) => { + const req = httpRequest({ host: url.hostname, port: url.port, path: '/api/runner', method: 'POST', + headers: { 'x-codeboost-token': app.token, 'content-type': 'application/json', 'content-length': Buffer.byteLength(text) } }, res => { res.resume(); res.on('end', () => resolve(res.statusCode!)); }); + req.on('error', reject); + req.write(text.slice(0, 5)); + finish = () => req.end(text.slice(5)); + }); + await new Promise(resolve => setTimeout(resolve, 20)); + const closing = close(); + await new Promise(resolve => setTimeout(resolve, 20)); + finish(); + expect(await response).toBe(503); + await closing; + const store = new Store(database); cleanups.push(() => store.close()); + expect(store.getTask(identity).status).toBe('needs human'); + expect(store.getAttempts(identity)).toEqual([]); + expect(store.savedAction(identity, { actionId, kind: 'start', request: { attemptId: undefined, expectedStateVersion: stateVersion } })).toBeUndefined(); + }); + it('answers 503 to a start once the runner stopped admission, and leaves the task in review', async () => { + const { app, identity, store } = await serve(); + // Shutdown step 1 for the runner, with the HTTP server still open: the request reaches the action itself. + app.runner!.rejectAdmission(); + const actionId = randomUUID(); + const refused = await act(app, 'start', { actionId }); + expect(refused.status).toBe(503); + expect(store.getTask(identity).status).toBe('in review'); + expect(store.getAttempts(identity)).toEqual([]); + // Not recorded: the same action ID is refused afresh (503 again), never replayed as a saved refusal. + expect((await act(app, 'start', { actionId })).status).toBe(503); + }); +}); diff --git a/web/cli.ts b/web/cli.ts index 050d9615..135421f2 100644 --- a/web/cli.ts +++ b/web/cli.ts @@ -5,7 +5,7 @@ import { startServer, type RunnerSetup } from './server.ts'; import { createDemo } from '../scripts/demo.ts'; import { Store, requireSupportedNode } from '../runner/store.ts'; import { acquireRunnerLock, releasePreparation } from '../runner/recovery.ts'; -import { parseRunnerConfig, setUpRunner } from '../runner/production.ts'; +import { parseRunnerConfig, recoveryWarnings, setUpRunner } from '../runner/production.ts'; requireSupportedNode(); /** A refusal the person acts on (bad input, a held lock, a blocked recovery): its message and exit 1, not a stack. */ function refuse(error: unknown): never { console.error(error instanceof Error ? error.message : String(error)); process.exit(1); } @@ -46,8 +46,7 @@ if (values.help || (!values.demo && !values.config && values['release-preparatio if (recovery.finalized.length) console.log(`Recovered ${recovery.finalized.length} interrupted attempt(s).`); if (recovery.requeue.length) console.log(`Interrupted tasks waiting to resume: ${recovery.requeue.length}.`); // Left for a person (runner-lifecycle.md): never removed automatically. - for (const attemptId of recovery.unmatchedStorage) console.error(`Task storage labelled with attempt ${attemptId} matches no attempt of this database; it was left in place.`); - for (const entry of recovery.unknownEntries) console.error(`Unknown entry in the runner's attempt directory, left in place: ${entry}`); + for (const line of recoveryWarnings(recovery)) console.error(line); return assembly; } : undefined; // A first stop during startup waits for startup recovery to finish, so it is not cut off part way. A second one stops at @@ -69,7 +68,11 @@ if (values.help || (!values.demo && !values.config && values['release-preparatio } for (const signal of ['SIGINT', 'SIGTERM'] as const) process.removeListener(signal, duringStartup); try { lock.verify(); } - catch (error) { await app.close(); lock.release(); throw error; } + catch (error) { + await app.close(); lock.release(); + // The database path changed: a refusal the person acts on, so its message, not a stack. + refuse(error); + } const stop = () => void app.close().then(() => { lock.release(); process.exit(0); }, error => { lock.release(); console.error(error instanceof Error ? error.message : error); process.exit(1); }); // A second Ctrl+C does not skip shutdown (runner-lifecycle.md): agents are still being stopped and awaited. diff --git a/web/server.ts b/web/server.ts index 52157bf3..d52f907a 100644 --- a/web/server.ts +++ b/web/server.ts @@ -76,6 +76,60 @@ export async function startServer(config: ReviewConfig, port = 4318, questionAge .filter(note=>note.kind==='question') .map(note=>({id:note.id,answer:note.answer,answerActive:questions.isRunning(note.id)})); const identity = config.identity; + /** + * What `start` or `resume` would run (#91 part 2), or the refusal. It writes nothing, so the view asks it too and never + * offers what the action would refuse. `start` runs a task that is in review or queued and has attempted no item of its + * current plan revision; `resume` continues a running or queued task that attempted an item at any revision (or that + * recovery left to requeue), from the first item the current revision has not completed, whether its last item + * completed, failed or was stopped. A queued task with only earlier-revision attempts may take either; both run the + * same items. + */ + const runChoice = (action: 'start' | 'resume', forView = false, progress = executor?.progress(identity)) => { + if (!executor || !progress) throw new GuardRefusal(config.demo ? RUNNER_NOT_IN_DEMO : RUNNER_NOT_CONFIGURED); + // Shutdown began: answer 503 before any refusal, so nothing is recorded under the action ID and the UI may resend. + if (!forView && runner!.closing) throw new ShuttingDownError(); + const task = service.store.getTask(identity), { started, begun, earlierCommits, next } = progress; + // Checked in the order a person can act on: a closed task first, then what admission would refuse. + if (task.status === 'merged' || task.status === 'cancelled') throw new GuardRefusal(`The task is ${task.status}.`); + if (task.cancelRequested !== null) throw new GuardRefusal('The task is being cancelled.'); + if (!runner!.runs('execute')) throw new GuardRefusal('The runner cannot run execute attempts yet.'); + if (runner!.isActive(identity)) throw new GuardRefusal('An attempt is already active for this task.'); + // Between two items no attempt is active, but the run that admits the next one is still going. + if (executor.busy(identity)) throw new GuardRefusal('An earlier run of this task is still finishing; try again when that run has ended.'); + const merge = service.store.getMergeAttempt(identity); + if (merge && (merge.state === 'submitting' || merge.state === 'queued')) throw new GuardRefusal('A merge is in progress; wait for its outcome.'); + // Only the view stops here: the action lets admission refuse, because its refusal also moves the idle task to needs + // human (the time-limit mapping), and nothing else would. + if (forView && task.budgetDeadline !== null && task.budgetDeadline <= Date.now()) throw new GuardRefusal('The task time budget has run out; it needs a person.'); + if (service.store.latestCheckpoint(identity)) throw new GuardRefusal('The task paused for a scope amendment; continuing after one is not supported yet (#88).'); + // Commits an earlier revision's items made have not been reconciled with the revised plan (plan-format.md): rerunning + // the plan on top of them could redo or contradict that work, so it waits for #88 rather than guessing. Earlier + // attempts that committed nothing (failed, stopped, stale or unchanged) leave nothing to reconcile. + if (!begun && earlierCommits) throw new GuardRefusal('The plan was revised after items of it were committed; running a revised plan on top of those commits is not supported yet (#88).'); + if (action === 'start') { + // Point to resume wherever resume could run (a running task that ran any revision too); any other status is named as it is. + const ran = begun || task.requeuePending; + if (next !== null && ((ran && task.status === 'queued') || (task.status === 'running' && (started || task.requeuePending)))) + throw new GuardRefusal('This plan has already started running; resume the task instead.'); + if (task.status !== 'in review' && task.status !== 'queued') throw new GuardRefusal(`The task is ${task.status}; start runs a task that is in review or queued.`); + if (ran) throw new GuardRefusal(begun ? `This plan revision has already run; the task is ${task.status}.` : 'Recovery left this task to requeue; it cannot start until that is resolved.'); + return { fromItem: next!, claimRequeue: false, queue: task.status === 'in review' }; + } + // As for start: point to start only where start's own checks pass. + const startWould = (task.status === 'in review' || task.status === 'queued') && !begun && !task.requeuePending; + const toStart = 'This plan revision has not started running yet; start the task instead.'; + if (task.status !== 'running' && task.status !== 'queued') { + if (startWould) throw new GuardRefusal(toStart); + throw new GuardRefusal(`The task is ${task.status}; resume continues a task that is running or queued.`); + } + if (!started && !task.requeuePending) throw new GuardRefusal(startWould ? toStart : 'No item of this plan has run yet, so there is nothing to resume.'); + if (!next) throw new GuardRefusal('Every item of this plan has run.'); + return { fromItem: next, claimRequeue: task.requeuePending, queue: false }; + }; + const offered = (action: 'start' | 'resume', progress: ReturnType) => { + try { runChoice(action, true, progress); return true; } + catch (error) { if (error instanceof GuardRefusal) return false; throw error; } + }; /** Reads only task and attempt rows; never rebuilds history or the review. */ const runnerView = () => { const task = service.store.getTask(identity), attempts = service.store.recentAttempts(identity, 20); @@ -85,28 +139,41 @@ export async function startServer(config: ReviewConfig, port = 4318, questionAge && (task.status === 'running' || task.status === 'queued') && !task.requeuePending && task.cancelRequested === null && !status.active && !status.unresolved && !runner.unreleased && runner.runs(last.kind) && sameContext(last.context, service.store.currentContext(identity)) - // A plan item is retried only by resuming its task (#91 part 2); the retry action refuses it. + // A plan item runs again only by resuming its task; the retry action refuses it. && !(executor && last.kind === 'execute'); - return { available: !!runner, task, attempts, + const free = !!runner && !runner.closing && !status.unresolved && !runner.unreleased; + // One progress read per poll, shared by both flags. + const progress = free && executor ? executor.progress(identity) : undefined; + return { available: !!runner, task, attempts, startable: !!progress && offered('start', progress), resumable: !!progress && offered('resume', progress), stateVersion: task.stateVersion, retryable, stopRequested: status.stopRequested, unresolved: status.unresolved }; }; const runnerAction = (input: Record) => { const { action, attemptId, expectedStateVersion, actionId } = input; // Malformed requests are refused before userAction, so nothing is recorded under their action ID (HTTP 400). - if (!['cancel-attempt', 'retry', 'cancel-task'].includes(action as string)) throw new BadRequest('Unsupported runner action.'); + if (!['cancel-attempt', 'retry', 'cancel-task', 'start', 'resume'].includes(action as string)) throw new BadRequest('Unsupported runner action.'); if (!Number.isSafeInteger(expectedStateVersion)) throw new BadRequest('expectedStateVersion must be an integer.'); - if (action !== 'cancel-task') assertUuidV4(attemptId, 'Attempt ID'); + if (action === 'cancel-attempt' || action === 'retry') assertUuidV4(attemptId, 'Attempt ID'); return service.store.userAction(identity, { actionId: actionId as string, kind: action as string, request: { attemptId, expectedStateVersion } }, () => { if (service.store.getTask(identity).stateVersion !== expectedStateVersion) throw new GuardRefusal('Stale task state. Reload before writing.'); if (action === 'cancel-task') return { outcome: runner ? runner.cancelTask(identity, expectedStateVersion as number, actionId as string) : service.store.cancelTask(identity, expectedStateVersion as number, actionId as string) }; if (!runner) throw new GuardRefusal(config.demo ? RUNNER_NOT_IN_DEMO : RUNNER_NOT_CONFIGURED); + if (action === 'start' || action === 'resume') { + const choice = runChoice(action); + // In this transaction with the admission: a refused admission rolls the move to queued back with it. + if (choice.queue) service.store.transitionTask(identity, expectedStateVersion as number, 'queued'); + const begun = executor!.begin(identity, { fromItem: choice.fromItem, claimRequeue: choice.claimRequeue }); + // The run goes on after this request; its outcome is in the task and attempt rows. An error the run throws (storage, + // a missing snapshot) is only logged, and the next start or resume derives what is owed again. + void begun.outcome.catch(error => console.error(`Runner run failed: ${JSON.stringify(error instanceof Error ? error.message : String(error))}`)); + return begun.attemptId ? { outcome: 'started', attemptId: begun.attemptId, item: choice.fromItem } : { outcome: 'settled' }; + } if (action === 'cancel-attempt') { if (!runner.stop(identity, attemptId as string, 'cancelled')) throw new GuardRefusal('That attempt is not running.'); return { outcome: 'stopping' }; } const last = service.store.getAttempt(identity, attemptId as string); // A plan item runs only through the executor, which pauses or escalates after it; a bare retry would skip both. - if (executor && last.kind === 'execute') throw new GuardRefusal('Retrying a plan item on its own is not supported; it comes with resuming the task (#91 part 2).'); + if (executor && last.kind === 'execute') throw new GuardRefusal('A plan item is run again by resuming the task, not by retrying its attempt.'); const retry = runner.retry(identity, attemptId as string, { expectedStateVersion: expectedStateVersion as number, kind: last.kind, item: last.item, expectedContext: service.store.currentContext(identity), deadline: Date.now() + 10 * 60_000 }); return { outcome: 'started', attemptId: retry.id };