diff --git a/README.md b/README.md index 53bc4ecc..214fe230 100644 --- a/README.md +++ b/README.md @@ -33,6 +33,26 @@ An existing-store configuration may add a trusted GitHub binding: The issue must match the stored plan. The authenticated `gh` account must be able to read the pull request, issue timeline, applicable rulesets, and classic branch protection, and to merge the PR. Codeboost unions required checks from both rule sources, requires strict server-enforced current-base checks, rechecks the base and head immediately before merging, and passes the reviewed head to `gh pr merge --match-head-commit`. Missing permissions and ambiguous rule responses block the merge. When the branch uses a merge queue, codeboost adds the exact reviewed head to the queue and treats the merge as done only when GitHub confirms it merged; a queued pull request is not merged. If GitHub removes it from the queue while the plan, snapshot, review and head are unchanged, Merge offers a retry after the status refresh; if the head changed, it goes back to review first. A moved base and any unexecuted `cmd:` acceptance check remain blocked until [#22](https://github.com/codeabovelab/codeboost/issues/22) adds the runner path. +## Runner (opt-in) + +A configuration with a `github` block may also add a `runner` block. The runner carries out plan items with Claude Code in the agent container and commits each item in its own repository, never in yours: + +```json +{ + "runner": { + "root": "/absolute/path/owned/by/you", + "committer": { "name": "codeboost", "email": "runner@example.invalid" } + } +} +``` + +`root` must be a directory only you can write. Optional fields: +- `diagnosticsDir`: where partial output of stopped attempts is kept, in a folder per database (`//diagnostics`), so databases sharing it never delete each other's files. The default is the runner root. +- `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). + ## Library - `core/plan.ts`: schema validation; YAML/JSON import; projected file-state and dependency checks; literal command parsing; individual suggestion validation and application. diff --git a/docs/implementation/runner-lifecycle.md b/docs/implementation/runner-lifecycle.md index f02a9509..79107500 100644 --- a/docs/implementation/runner-lifecycle.md +++ b/docs/implementation/runner-lifecycle.md @@ -256,6 +256,7 @@ codeboost never writes the user's repository. The commits it makes live in a **r | 3 | After the drain limit, abort the signals of the remaining requests, destroy requests that are still reading a body, then **close the `Store` write gate**, then await request-owned work (the merge coordinator). From here on, every `Store` write throws `ShuttingDownError` unless the caller passes the shutdown capability. The server hands that capability at construction only to coordinators' settlement and close code (the runner, questions' `finishAnswer`, merge settlement after an irreversible command, and E3's `settleSuggestion` when it is wired), and to startup recovery; HTTP handlers and `ReviewService` never receive it. So a request-path write still pending after the abort (an aborted `/api/review` whose `load()` reaches `recordHistory`, or merge reconciliation and queue polling after a GitHub await) fails. The server maps `ShuttingDownError` to HTTP 503 "The review server is shutting down.", never to the 409 used for review errors, and there is no partial view. **Today the merge coordinator catches every error in `displayStatus()` and `#pollQueue()`**, which would swallow the gate error and answer 200; F1 makes each of those catch blocks rethrow `ShuttingDownError` first. The next startup checks GitHub again, so a skipped merge observation is recovered. The server also closes idle keep-alive connections once shutdown starts and after each request finishes, so a finished request's socket cannot hold `server.close()` open until its keep-alive timeout. | abort: yes. Gate, capability, rethrow and idle-connection close: new. | | 4 | For every `pending` and `running` attempt that has an in-memory job, record the first reason `shutdown` **only if no first reason is set yet**. A job already marked `cancelled`, `stale` or `time-limit` keeps its reason. For `running`, call `cancel('shutdown')` and await `settled`. For `pending`, await its preparation promise (the D start call itself returns at once, so it is never in progress), remove only its host-side preparation files (task storage waits for the terminal write, as in step 5), and cancel any handle that start returned, then await its `settled` (the "Launch" table). F does not abandon a job after a timer (decision 4). **This ends:** D's supervisor escalates to a forced kill, and retries unfinished container, network or setup cleanup every second for at most 60 seconds after the first failure (`CLEANUP_RETRY_WINDOW_MS`, `agents/adapters/supervisor.ts`). Retries after the first get only what is left of the window, and every cleanup subprocess is killed at its deadline, so settlement ends about 90 seconds after cleanup starts at worst (up to 30 seconds for the first, failed attempt, then the 60-second window). Then `settled` resolves with a `stopReason` and an `unreleased` list, even if Docker is unreachable or hangs. F records `unreleased` durably before it frees the slot. See the prerequisite below. | new | | 5 | For each attempt from step 4, in the order set by the task-storage holder row: the partial-output export (writable attempts), then the terminal state from its first reason (a `shutdown` reason gives `cancelled` "Stopped by shutdown") with its `diagnostic_ref`, then `removeTaskFilesystems`, and only then free its slot. A failed write leaves the row non-terminal, and a failed removal keeps the slot held. Startup recovery handles both. | new | +| 5a | Await every plan run in progress (`ItemExecutor.close()`, #91). A run resumes after its attempt settles and may then pause for amendment or move the task to needs human: settlement writes, made with the capability, that must land before step 7 closes the `Store`. Admission is closed, so no run starts another item. | new (#91) | | 6 | Await server closure; await the question and suggestion coordinators' `close()`. | yes (questions); suggestions: new wiring | | 7 | Close the `Store`. | yes | | 8 | Release the single-runner lock (decision 1). Release it on every exit path after it was taken, including a startup failure. | new | @@ -278,9 +279,9 @@ codeboost never writes the user's repository. The commits it makes live in a **r |---|---|---| | Runner owner token | F | A random ID of exactly 32 lowercase hexadecimal characters (`^[0-9a-f]{32}$`), stored in the database (`app_settings`). A stored value that does not match is refused before it is used as a label or in any path, and startup exits. It is stored together with the database file's device and inode numbers. It is read, or created, in startup recovery step 2, after the lock is taken and the `Store` is opened. If the stored device and inode don't match the open file (the database was copied), F creates a new token. So a copy never shares a token with its original. | | Token in every request | D contract | Done in #51 item 3. `InvocationInput` has a required `runnerOwner` (32 lowercase hex characters), and so does every allocator that runs before an invocation exists: `prepareTaskFilesystems` takes an owner (`runnerOwner`, `attemptId`, and a caller-chosen `allocationId`, a lowercase UUID v4), and the adapters take a caller-chosen `networkAllocationId` for the vendor network. Every resource D creates carries `io.codeboost.runner`, `io.codeboost.attempt` and `io.codeboost.allocation` (`agents/labels.ts`): task-storage volumes, the keeper and the seeder with the storage owner; the network and proxy with the invocation's runner and attempt and the network allocation; the agent container with its own attempt and the storage allocation it mounts. D checks these labels whenever it validates a resource, and refuses task storage that belongs to another runner. Each allocation ID names one allocation: before creating anything, D refuses an ID held by another allocation in the process or still carried by any Docker object (so a reuse after a restart is caught). That check and the create are not atomic across processes, so right after the first create D checks again that its object is the only one with the ID, and backs out if not. Cleanup proves ownership in one of two ways. A Docker ID captured when D created the object is proof on its own (Docker never reuses IDs), so the vendor network, proxy and agent container are removed by that ID even if their labels are wrong. An object found only by name (a create whose ID never came back, or task storage, which has no captured IDs) is removed only when its runner, attempt and allocation labels all match. The adapters check `networkAllocationId` before returning a handle, and `captureInvocation` refuses an attempt ID that cannot be written as a label. | -| Recovered storage handles | D | Done in #51 item 4: the report's `storage` lists one `RecoveredTaskStorage` handle per allocation, with its runner, attempt and allocation IDs and whichever parts exist, and both `removeTaskFilesystems` and `exportTaskDiff` accept it. D tracks task-storage ownership in a process-local `WeakMap`, which is empty after a restart. So `recoverLeftovers` returns, for each kept volume set, an **authenticated recovery handle** that D registers in its ownership map after inspecting every part (volumes and keeper) and checking its task-storage kind and all three owner labels. It refuses storage this process allocated and has not removed. **Binding to an attempt:** task-storage volumes and the keeper also carry `io.codeboost.attempt=`, from the owner passed to `prepareTaskFilesystems`. **F chooses the allocation ID, not D.** F generates it (a lowercase UUID v4) and saves it as `attempts.allocation_id` **before** it starts the asynchronous allocation, then passes it to `prepareTaskFilesystems`, which labels the volumes and keeper with it (`io.codeboost.allocation`). So a crash in the middle of allocation still leaves a row that matches whatever D created. The D prerequisite includes accepting a caller-provided allocation ID. The handle carries both IDs. F uses a handle only if its attempt ID and allocation ID both match one attempt row of this database. A handle that matches no row, or disagrees with the row, is reported and listed by `--list-unowned-agent-resources` for the user to remove by hand; it is never exported or removed automatically. `exportTaskDiff` and `removeTaskFilesystems` accept these handles. F never builds a `TaskFilesystems` value from report fields. | +| Recovered storage handles | D | Done in #51 item 4: the report's `storage` lists one `RecoveredTaskStorage` handle per allocation, with its runner, attempt and allocation IDs and whichever parts exist, and both `removeTaskFilesystems` and `exportTaskDiff` accept it. D tracks task-storage ownership in a process-local `WeakMap`, which is empty after a restart. So `recoverLeftovers` returns, for each kept volume set, an **authenticated recovery handle** that D registers in its ownership map after inspecting every part (volumes and keeper) and checking its task-storage kind and all three owner labels. It refuses storage this process allocated and has not removed. **Binding to an attempt:** task-storage volumes and the keeper also carry `io.codeboost.attempt=`, from the owner passed to `prepareTaskFilesystems`. **F chooses the allocation ID, not D.** F generates it (a lowercase UUID v4) and saves it as `attempts.allocation_id` **before** it starts the asynchronous allocation, then passes it to `prepareTaskFilesystems`, which labels the volumes and keeper with it (`io.codeboost.allocation`). So a crash in the middle of allocation still leaves a row that matches whatever D created. The D prerequisite includes accepting a caller-provided allocation ID. The handle carries both IDs. F uses a handle only if its attempt ID and allocation ID both match one attempt row of this database. A handle that matches no row, or disagrees with the row, is reported at startup (the CLI prints its attempt ID) for the user to remove by hand; it is never exported or removed automatically. `exportTaskDiff` and `removeTaskFilesystems` accept these handles. F never builds a `TaskFilesystems` value from report fields. | | Scoped recovery | D | Done in #51 item 4 (`agents/recovery.ts`). `recoverLeftovers(runnerOwner): Promise` acts only on resources whose `io.codeboost.runner` label equals the token. It stops and removes agent containers, egress proxies, unfinished seeders, export and inspection containers and networks, by ID and networks last, and resolves only when they are gone. It removes an object only when it carries exactly one kind label with the value D writes and complete owner labels. **It keeps task storage whole: the volumes and their keeper container**, which it recognises by their task-storage labels, and lists one handle per allocation so F can export partial output first. A handle names whichever parts exist, since a crash during allocation can leave any subset; export needs the keeper. F then calls `removeTaskFilesystems` for each one (startup recovery step 4). It refuses to run while this process holds task storage of the runner. It rejects with a `RecoveryError` (bounded message, plus the objects it did remove) if anything cannot be stopped or removed, or if a storage check cannot reach Docker, and then issues no handles. | -| Unowned resources | D and F | Done in D by #51 item 4: the report's `unowned` lists them with the reason `no-runner-label`, alongside objects of this runner that D does not create (`unknown-kind`) and storage whose parts disagree (`inconsistent-storage`); none is removed. Resources with codeboost labels but no `io.codeboost.runner` label (from builds before this change) fail closed. The report lists them, and F refuses to open admission while any exist. **codeboost never removes them itself**, because it cannot prove that no older-build runner is alive: older builds take no lock this build can see, and an idle older runner may have no container running at the moment of a check. So `--list-unowned-agent-resources` prints each resource and the exact `docker rm` / `docker network rm` / `docker volume rm` commands. The user stops every older codeboost process, runs those commands, and starts again. There is no machine-wide lock. | +| Unowned resources | D and F | Done in D by #51 item 4: the report's `unowned` lists them with the reason `no-runner-label`, alongside objects of this runner that D does not create (`unknown-kind`) and storage whose parts disagree (`inconsistent-storage`); none is removed. Resources with codeboost labels but no `io.codeboost.runner` label (from builds before this change) fail closed. The report lists them, and F refuses to open admission while any exist. **codeboost never removes them itself**, because it cannot prove that no older-build runner is alive: older builds take no lock this build can see, and an idle older runner may have no container running at the moment of a check. So startup stops with a message that names each resource as the exact `docker container rm -f` / `docker network rm` / `docker volume rm` command that removes it (`runner/production.ts`, #91). The user stops every older codeboost process, runs those commands, and starts again. There is no machine-wide lock. | | When it runs | F | Only while holding this database's single-runner lock (startup recovery step 1 comes first). | | Ask's owner | Ask | Ask labels its objects with a second per-database token, `ask_owner` (`Store.askOwnerToken`, same rules as the runner token), and recovers it under the review's Ask lock (#65, `agent-isolation.md`). The runner's recovery therefore never sees Ask's storage, and Ask's recovery never removes the runner's agents. Unlike the runner, Ask does not refuse because of `no-runner-label` objects: it calls `recoverLeftovers(askOwner, timeoutMs, { unowned: false })`, which skips the daemon-wide search for them. | @@ -318,6 +319,8 @@ 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. + ## 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. diff --git a/github/issues.ts b/github/issues.ts index b04d6059..c143a9b9 100644 --- a/github/issues.ts +++ b/github/issues.ts @@ -1,4 +1,5 @@ import { ghEnvironment } from './gh-env.ts'; +import { dataJSON, MAX_PROMPT_BYTES } from '../core/planning-author.ts'; import { runWithInput } from './run-with-input.ts'; const PAGE_SIZE = 100; @@ -41,6 +42,9 @@ export interface IssueSnapshot { readonly issues: readonly RepositoryIssue[]; } +/** The issue text an execute prompt carries, as untrusted data. */ +export interface IssueText { readonly number: number; readonly title: string; readonly body: string; readonly comments: readonly string[] } + export interface IssueGateway { readonly repository: string; fetch(options?: { signal?: AbortSignal; timeoutMs?: number }): Promise; @@ -229,7 +233,61 @@ export class GhIssueGateway implements IssueGateway { })); } - async fetch(options: { signal?: AbortSignal; timeoutMs?: number } = {}): Promise { + /** + * One issue's text for an execute prompt (#91): its title, body and the comments of repository collaborators only + * (design, "Which comments reach the agent"), oldest first. Everything stays untrusted data inside the prompt. + * Fails closed on anything malformed, on a pull request, and past the comment page limit. + */ + async issueText(number: number, options: { signal?: AbortSignal; timeoutMs?: number } = {}): Promise { + if (!Number.isSafeInteger(number) || number < 1) throw new Error('Invalid issue number.'); + return this.#bounded(options, async signal => { + let decoded: unknown; + const output = await this.run(['api', '--method', 'GET', '-H', 'Accept: application/vnd.github+json', `repos/${this.repository}/issues/${number}`], { signal }); + try { decoded = JSON.parse(output); } + catch { throw new Error('GitHub returned invalid issue JSON.'); } + const issue = object(decoded, 'GitHub returned a malformed issue.'); + 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. + 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.`); + let total = Buffer.byteLength(title) + Buffer.byteLength(body); + 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 }); + let values: unknown; + try { values = JSON.parse(listed); } + catch { throw new Error('GitHub returned invalid comment JSON.'); } + if (!Array.isArray(values) || values.length > PAGE_SIZE) throw new Error('GitHub returned an invalid comment page.'); + // Only an empty page past the limit ends the read cleanly; anything on it is more than codeboost reads. + if (page > MAX_PAGES && values.length) throw new Error(`Issue #${number} has more than ${MAX_ISSUES} comments; codeboost does not read past that limit.`); + for (const value of values) { + const comment = object(value, 'GitHub returned a malformed comment.'); + // A deleted ("ghost") author is nobody's collaborator. Other people's comments are dropped before their body is + // checked, so none of theirs can make the issue unreadable. + if (comment.user === null) continue; + const author = login(object(comment.user, 'GitHub returned an invalid comment author.').login, 'issue author'); + if (!collaborators.has(author.toLocaleLowerCase('en-US'))) continue; + const text = boundedString(comment.body, 'comment', MAX_BODY_LENGTH, true); + total += Buffer.byteLength(text); + if (total > MAX_PROMPT_BYTES) throw tooLong(); + comments.push(text); + } + 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; } + return text; + }); + } + + /** Runs `load` under the caller's signal and an overall deadline, as `fetch` does. */ + async #bounded(options: { signal?: AbortSignal; timeoutMs?: number }, load: (signal: AbortSignal) => Promise): Promise { const timeoutMs = options.timeoutMs ?? 12_000; if (!Number.isSafeInteger(timeoutMs) || timeoutMs < 1 || timeoutMs > 30_000) throw new Error('Invalid issue retrieval timeout.'); options.signal?.throwIfAborted(); @@ -238,11 +296,9 @@ export class GhIssueGateway implements IssueGateway { options.signal?.addEventListener('abort', relay, { once: true }); const timer = setTimeout(() => controller.abort(new Error('Issue retrieval timed out.')), timeoutMs); try { - const issues = await this.#load(controller.signal); + const value = await load(controller.signal); controller.signal.throwIfAborted(); - const retrievedAt = this.now(); - if (!Number.isFinite(retrievedAt.getTime())) throw new Error('Issue retrieval clock is invalid.'); - return { repository: this.repository, retrievedAt: retrievedAt.toISOString(), issues }; + return value; } catch (error) { if (options.signal?.aborted) throw options.signal.reason; if (controller.signal.aborted) throw new Error('Issue retrieval timed out.'); @@ -252,4 +308,14 @@ export class GhIssueGateway implements IssueGateway { options.signal?.removeEventListener('abort', relay); } } + + async fetch(options: { signal?: AbortSignal; timeoutMs?: number } = {}): Promise { + return this.#bounded(options, async signal => { + const issues = await this.#load(signal); + signal.throwIfAborted(); + const retrievedAt = this.now(); + if (!Number.isFinite(retrievedAt.getTime())) throw new Error('Issue retrieval clock is invalid.'); + return { repository: this.repository, retrievedAt: retrievedAt.toISOString(), issues }; + }); + } } diff --git a/runner/coordinator.ts b/runner/coordinator.ts index f1f048e9..c9e52e84 100644 --- a/runner/coordinator.ts +++ b/runner/coordinator.ts @@ -29,6 +29,11 @@ export interface RunnerDeps { * under the same token, because D refuses storage owned by another runner. */ readonly runnerOwner: string; + /** + * The attempt kinds these deps can run. Admission refuses any other kind before it writes anything, so a retry of, + * say, a review attempt never reaches deps that would run it as an execute. Omitted: every kind. + */ + readonly kinds?: readonly AttemptKind[]; /** * Host-side preparation (clone, prompt). On abort it must stop and await every subprocess it started, then reject. * It never leaves work running after it settles. @@ -135,6 +140,8 @@ export class RunnerCoordinator { get closing(): boolean { return this.#closing; } /** Resources D could not confirm removed; non-null keeps the runner closed to new work until restart. */ get unreleased(): readonly UnreleasedResource[] | null { return this.#unreleased; } + /** Whether admission would take this kind (`RunnerDeps.kinds`); a view must not offer what admission refuses. */ + runs(kind: AttemptKind): boolean { return !this.#deps.kinds || this.#deps.kinds.includes(kind); } #now(): number { return this.#deps.now?.() ?? Date.now(); } #used(group: Group): number { let used = 0; @@ -154,6 +161,7 @@ export class RunnerCoordinator { const marker = this.#markers.get(key); if (marker) throw new GuardRefusal(NEEDS_RESTART[marker.reason]); if (!(request.kind in ATTEMPT_PHASES)) throw new GuardRefusal('Unknown attempt kind.'); + if (this.#deps.kinds && !this.#deps.kinds.includes(request.kind)) throw new GuardRefusal(`The runner cannot run ${request.kind} attempts yet.`); const group: Group = WRITABLE_KINDS.includes(request.kind) ? 'writable' : 'readOnly'; if (this.#used(group) >= this.#limits[group]) throw new GuardRefusal('No free runner slot. Try again when the current attempt finishes.'); const job: Job = { identity: { ...identity }, key, group, attemptId: '', firstReason: null, reasonSaved: true, preparationTimedOut: false, controller: new AbortController(), timers: [] }; diff --git a/runner/diagnostics.ts b/runner/diagnostics.ts index bf48e3d5..8579aa5c 100644 --- a/runner/diagnostics.ts +++ b/runner/diagnostics.ts @@ -8,13 +8,20 @@ import type { Store } from './store.ts'; export const DEFAULT_DIAGNOSTICS_CAP_BYTES = 256 * 1024 * 1024; const NAME = /^([0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12})\.diff$/; +/** + * D's export bytes as saved: a diff D cut at its byte limit ends with a `codeboost:` notice line (like D's own notices), + * so a reader never takes a cut diff for everything the agent changed. + */ +export function partialOutput(diff: Buffer, truncated: boolean): Buffer { + return truncated ? Buffer.concat([diff, Buffer.from(`\ncodeboost: the partial output was cut at ${diff.length} bytes; later changes are not shown.\n`)]) : diff; +} /** * Save a stopped attempt's partial output as `/.diff` (runner-lifecycle.md, the task storage * row), then apply retention. The file is written through a unique temporary file opened exclusively, owner-only, and * renamed into place; a failed write leaves nothing. Returns the path the attempt row references as `diagnostic_ref`. */ export function saveDiagnostic(store: Store, directory: string, attemptId: string, bytes: Buffer, - capBytes = DEFAULT_DIAGNOSTICS_CAP_BYTES): string { + capBytes = DEFAULT_DIAGNOSTICS_CAP_BYTES, alsoKeep: readonly string[] = []): string { if (!isUuidV4(attemptId)) throw new Error('Attempt ID must be a UUID v4.'); if (!Number.isSafeInteger(capBytes) || capBytes < 1) throw new Error('The diagnostics cap must be a positive integer.'); const path = join(directory, `${attemptId}.diff`), temporary = join(directory, `.${attemptId}.${randomUUID()}.tmp`); @@ -23,7 +30,7 @@ export function saveDiagnostic(store: Store, directory: string, attemptId: strin renameSync(temporary, path); } catch (error) { rmSync(temporary, { force: true }); throw error; } // The saved file stands even if retention cannot run now (the write gate closed): the next save retains again. - try { retain(store, directory, capBytes, path); } + try { retain(store, directory, capBytes, [path, ...alsoKeep]); } catch (error) { console.error(`Diagnostics retention did not run: ${JSON.stringify(error instanceof Error ? error.message : String(error))}`); } return path; } @@ -32,9 +39,10 @@ export function saveDiagnostic(store: Store, directory: string, attemptId: strin * Retention never deletes a file an attempt row still references. Over the cap, it deletes unreferenced files oldest * first; if that is not enough, it takes the oldest referenced file and, in one transaction, clears that row's * `diagnostic_ref` and notes the removal in its diagnostic, and only after that commit deletes the file. The file just - * saved (`keep`) is never removed: it is not referenced until the terminal write. + * saved (`keep`) is never removed: it is not referenced until the terminal write. Startup recovery saves several before + * its one finalization transaction references them, so it keeps every one it saved. */ -export function retain(store: Store, directory: string, capBytes: number, keep?: string): void { +export function retain(store: Store, directory: string, capBytes: number, keep: readonly string[] = []): void { const files = readdirSync(directory).flatMap(name => { const match = NAME.exec(name), path = join(directory, name); if (!match) return []; @@ -46,7 +54,7 @@ export function retain(store: Store, directory: string, capBytes: number, keep?: for (const unreferenced of [true, false]) { for (const file of files) { if (total <= capBytes) return; - if (file.path === keep || referenced.has(file.path) === unreferenced) continue; + if (keep.includes(file.path) || referenced.has(file.path) === unreferenced) continue; if (!unreferenced && !store.forgetDiagnostic(file.path)) continue; rmSync(file.path, { force: true }); total -= file.size; diff --git a/runner/execution.ts b/runner/execution.ts index 07a52c25..9c1435ad 100644 --- a/runner/execution.ts +++ b/runner/execution.ts @@ -5,7 +5,8 @@ import { prepareExecution } from '../core/execution-prompt.ts'; import { neutralizeMentions, neutralizeReferences } from '../core/pull-request-body.ts'; import { auditRun, type ChangeManifest } from '../core/run-audit.ts'; import { TaskTreeRefused, type DeclaredLinkSnapshot, type DeclaredOperation, type TaskTreeCheck } from '../agents/container/changes.ts'; -import { saveDiagnostic } from './diagnostics.ts'; +import type { IssueText } from '../github/issues.ts'; +import { partialOutput, saveDiagnostic } from './diagnostics.ts'; import { ownerOnlyDirectory } from './runner-repository.ts'; import { FinishFailure, NEEDS_RESTART, PreparationFailure, type PreparedAttempt, type RunnerCoordinator, type RunnerDeps } from './coordinator.ts'; import type { AttemptRecord, Store } from './store.ts'; @@ -55,7 +56,8 @@ export type AgentLauncher = (input: InvocationInput, prompt: string, workspace: /** Trusted runner-side sources for a task. Issue text and lessons are untrusted data inside the prompt. */ export interface ExecutionSources { planContext(identity: PlanIdentity): PlanContext; - issue(identity: PlanIdentity): { number: number; title: string; body: string; comments: readonly string[] }; + /** The issue text the prompt carries; fetched per attempt, so it may await (and must stop on abort). */ + issue(identity: PlanIdentity, signal: AbortSignal): IssueText | Promise; lessons(identity: PlanIdentity): readonly string[]; vendor(identity: PlanIdentity): 'claude' | 'codex'; } @@ -134,14 +136,18 @@ export function executionDeps(store: Store, workspace: TaskWorkspace, launch: Ag }; return { runnerOwner, + // Only execute attempts have deps yet; admission refuses the rest before any row is written (#91). + kinds: ['execute'], async prepare(attempt, signal) { if (attempt.kind !== 'execute' || !attempt.item) throw new Error('Execution deps run execute attempts for one plan item.'); const identity = identityOf(attempt), plan = store.getPlan(identity, attempt.context.planRevision); const item = plan.items.find(entry => entry.id === attempt.item)!; const context = sources.planContext(identity); const baseHead = store.getSnapshot(identity, attempt.context.snapshotId).head; + const issue = await sources.issue(identity, signal); + signal.throwIfAborted(); const request = prepareExecution({ identity, attemptId: attempt.id, mode: 'execute', plan, itemId: item.id, - issue: sources.issue(identity), approvedLessons: sources.lessons(identity), allowedCommands: context.allowedCommands }); + issue, approvedLessons: sources.lessons(identity), allowedCommands: context.allowedCommands }); const vendor = sources.vendor(identity); const declaredPaths = [...new Set(item.files.flatMap(file => [file.path, ...(file.renamed_from ? [file.renamed_from] : [])]))]; const ws = await workspace.materialize(attempt, baseHead, signal); @@ -215,10 +221,10 @@ export function executionDeps(store: Store, workspace: TaskWorkspace, launch: Ag if (!workspace.exportPartial || !options.diagnostics) return {}; const signal = AbortSignal.timeout(options.exportDeadlineMs ?? 60_000); try { - const { diff } = await workspace.exportPartial(data.workspace, { baseHead: data.baseHead }, signal); + const { diff, truncated } = await workspace.exportPartial(data.workspace, { baseHead: data.baseHead }, signal); // Only the current user may write where diagnostics are kept. ownerOnlyDirectory(options.diagnostics.directory); - return { diagnosticRef: saveDiagnostic(store, options.diagnostics.directory, attempt.id, diff, options.diagnostics.capBytes) }; + return { diagnosticRef: saveDiagnostic(store, options.diagnostics.directory, attempt.id, partialOutput(diff, truncated), options.diagnostics.capBytes) }; } catch (error) { return { failure: signal.aborted ? 'the export did not finish within its deadline' : (error instanceof Error ? error.message : String(error)) }; } @@ -276,17 +282,26 @@ export class ItemExecutor { * Tasks with a runTask 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 Set(); + #inFlight = new Map>(); async runTask(identity: PlanIdentity, options: { fromItem?: string } = {}): Promise { const key = identityKey(identity); // 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: [] }; - this.#inFlight.add(key); - try { return await this.#runTask(identity, options); } + const run = this.#runTask(identity, options); + this.#inFlight.set(key, run.catch(() => undefined)); + try { return await 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 + * attempt settles is a settlement write, so the Store must stay open until each run has returned (#91). Admission is + * already closed, so no run starts another item. + */ + async close(): Promise { + while (this.#inFlight.size) await Promise.all([...this.#inFlight.values()]); + } async #runTask(identity: PlanIdentity, options: { fromItem?: string }): Promise { const plan = this.#store.getPlan(identity); const start = options.fromItem ? plan.items.findIndex(item => item.id === options.fromItem) : 0; diff --git a/runner/production.ts b/runner/production.ts new file mode 100644 index 00000000..e15e6b2c --- /dev/null +++ b/runner/production.ts @@ -0,0 +1,196 @@ +import { randomUUID } from 'node:crypto'; +import { chmodSync, mkdirSync, writeFileSync } from 'node:fs'; +import { isAbsolute, join } from 'node:path'; +import { buildAgentImage } from '../agents/container/image.ts'; +import { exportTaskDiff, removeTaskFilesystemsAsync, type TaskStorageLimits } from '../agents/container/storage.ts'; +import { recoverLeftovers } from '../agents/recovery.ts'; +import { startClaudeInvocation } from '../agents/adapters/claude.ts'; +import { GhIssueGateway, type IssueText } from '../github/issues.ts'; +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 { recoverStartup, removalCommand, type RecoveryDeps, type RecoveryReport, type RunnerLock } from './recovery.ts'; +import type { ReviewService } from './review.ts'; +import { openRunnerRepository, ownerOnlyDirectory } from './runner-repository.ts'; +import { createTaskWorkspace, workspaceFilesystems } from './workspace.ts'; + +/** + * The production runner (#91): D's real workspace, launcher, recovery, export and removal, under the database's runner + * token. Opt-in through the review configuration's `runner` block; demos never run it. + */ +export interface RunnerConfig { + /** Absolute; owner-only. Attempt directories and the runner-owned repository live under `//`. */ + readonly root: string; + /** + * Absolute; owner-only. Partial output of stopped attempts is kept in `/`, one folder per + * database, so runners of several databases sharing it never retain (delete) each other's files. Default ``. + */ + readonly diagnosticsDir?: string; + readonly diagnosticsCapBytes?: number; + /** Who runner commits are by. */ + readonly committer: { readonly name: string; readonly email: string }; + readonly limits?: Partial; +} +/** Execute attempts write code: more room than Ask's read-only questions. tmpfs uses memory only for bytes stored. */ +export const EXECUTE_STORAGE: TaskStorageLimits = Object.freeze({ + workBytes: 2 * 1024 * 1024 * 1024, workInodes: 262_144, metadataBytes: 1024 * 1024 * 1024, metadataInodes: 131_072, +}); +/** The profile requires exactly one schema.json in the input mount; an execute answer is a plain-text summary. */ +const EXECUTE_SCHEMA = '{"$schema":"https://json-schema.org/draft/2020-12/schema","title":"codeboost execute summary","type":"string"}\n'; +/** How long one read of the issue serves later plan items: the length of a short multi-item run. */ +export const ISSUE_REUSE_MS = 5 * 60_000; +export const RUNNER_CREDENTIAL_MISSING = 'The runner runs Claude Code, which needs CLAUDE_CODE_OAUTH_TOKEN. Create one with `claude setup-token`, set it, and restart codeboost.'; + +/** Check the `runner` block of a review configuration before anything is created. */ +export function parseRunnerConfig(value: unknown): RunnerConfig { + if (!value || typeof value !== 'object' || Array.isArray(value)) throw new Error('The runner block must be an object.'); + const v = value as Record; + const path = (field: string, text: unknown) => { + if (typeof text !== 'string' || !isAbsolute(text)) throw new Error(`runner.${field} must be an absolute path.`); + // D mounts paths under the root into containers, and a Docker mount cannot hold a comma or a line break: refused here, + // before anything is created, instead of at every launch. + if (/[\0\n\r,]/.test(text)) throw new Error(`runner.${field} cannot contain a comma or a line break: Docker cannot mount such a path.`); + return text; + }; + const committer = v.committer as Record | undefined; + // Git refuses angle brackets and newlines in an identity; a commit with them would fail only after the agent ran. + const ident = (text: unknown) => typeof text === 'string' && text.trim() !== '' && text.length <= 200 && !/[<>\n\r\0]/.test(text); + if (!committer || !ident(committer.name) || !ident(committer.email)) throw new Error('runner.committer needs a name and an email without <, > or line breaks.'); + const limits: Partial = {}; + if (v.limits !== undefined) { + if (!v.limits || typeof v.limits !== 'object' || Array.isArray(v.limits)) throw new Error('runner.limits must be an object.'); + for (const [key, n] of Object.entries(v.limits)) { + if (!Object.hasOwn(EXECUTE_STORAGE, key) || !Number.isSafeInteger(n) || (n as number) < 1) throw new Error(`runner.limits.${key} is not a positive storage limit.`); + (limits as Record)[key] = n as number; + } + } + const cap = v.diagnosticsCapBytes; + if (cap !== undefined && (!Number.isSafeInteger(cap) || (cap as number) < 1)) throw new Error('runner.diagnosticsCapBytes must be a positive integer.'); + return Object.freeze({ root: path('root', v.root), ...(v.diagnosticsDir !== undefined ? { diagnosticsDir: path('diagnosticsDir', v.diagnosticsDir) } : {}), + ...(cap !== undefined ? { diagnosticsCapBytes: cap as number } : {}), + committer: Object.freeze({ name: committer.name as string, email: committer.email as string }), limits: Object.freeze(limits) }); +} + +/** + * Lane D's recovery, export and removal for `recoverStartup`. Unlike Ask (#65), the runner looks for codeboost objects + * without a runner label too, and every unowned object blocks startup (runner-lifecycle.md, "Unowned resources"): an + * older build takes no lock this build can see, so one may still be running. Each is reported as a shell line that + * removes it, with its reason as a comment, for a person to run once every older codeboost process has stopped. + * `image` builds the agent image once D's recovery has stopped every leftover agent, so none keeps writing to its + * storage during a long build, and before any export's deadline starts: the build blocks, and a timer armed before it + * would fire as soon as it returned. + */ +export function dRecoveryDeps(image: () => string): RecoveryDeps { + return { + async recoverLeftovers(runnerOwner) { + const report = await recoverLeftovers(runnerOwner, 120_000); + // Any recovered storage may be exported, and every export needs the image. + if (report.storage.length) image(); + return { + // D's handle object itself: its trust is keyed by identity, so a copy would be refused. + storage: report.storage.map(handle => ({ attemptId: handle.attemptId, allocationId: handle.allocationId, handle })), + unowned: report.unowned.map(resource => + `${removalCommand(resource)} # ${resource.reason}`), + }; + }, + exportTaskDiff: (handle, input, maxBytes, signal) => + exportTaskDiff(handle as Parameters[0], { base: input.base, metadataBaseline: input.metadataBaseline, imageId: image(), maxBytes, signal }), + removeTaskFilesystems: handle => removeTaskFilesystemsAsync(handle as Parameters[0]), + }; +} + +/** + * D's Claude start call for an execute attempt. The input mount holds only the schema, in the attempt's own directory, + * which the coordinator removes with the rest of its preparation once D settled. + */ +export function claudeLauncher(o: { imageId: string; runnerRoot: string; runnerOwner: string; token: string; + start?: typeof startClaudeInvocation }): AgentLauncher { + return (input, prompt, workspace, treeCheck) => { + if (!isUuidV4(input.attemptId)) throw new Error('Attempt ID must be a UUID v4.'); + const directory = join(o.runnerRoot, o.runnerOwner, 'attempts', input.attemptId, 'input'); + mkdirSync(directory, { mode: 0o755 }); + // The container user reads it; the directory stays writable by its owner, so cleanup can remove it. + chmodSync(directory, 0o755); + const schema = join(directory, 'schema.json'); + writeFileSync(schema, EXECUTE_SCHEMA, { mode: 0o444, flag: 'wx' }); + // The umask narrows a create mode (077 gives 0400), and D refuses a schema the container user cannot read. + chmodSync(schema, 0o444); + // D requires the pre-launch tree check made for this storage (#81): it mounts the head's gitlinks from it. + return (o.start ?? startClaudeInvocation)({ invocation: input, filesystems: workspaceFilesystems(workspace), inputDirectory: directory, + imageId: o.imageId, prompt, networkAllocationId: randomUUID(), treeCheck }, o.token); + }; +} + +export interface RunnerAssembly { + readonly deps: RunnerDeps; + readonly sources: ExecutionSources; + readonly findings: SafetyFindings; + readonly recovery: RecoveryReport; +} +/** + * Startup with a runner, after the Store opened and before the server admits anything (runner-lifecycle.md, "Startup + * recovery"): verify the lock still names the database, run recovery under the database's runner token, build the + * agent image (once D's recovery returns, if it found any storage; otherwise after recovery), then assemble the + * execution deps. Any failure stops startup; the caller closes the Store. + */ +export async function setUpRunner(o: { service: ReviewService; capability: ShutdownCapability; config: RunnerConfig; lock: Pick; + env: Readonly>; buildImage?: () => string; recovery?: (image: () => string) => RecoveryDeps; + /** Told before the slow part (recovery and the image build) starts. */ + onSlowStart?: () => void }): Promise { + const { service, config } = o, review = service.config; + if (review.demo) throw new Error('Demos never run the runner.'); + if (!review.github) throw new Error('The runner reads the plan\'s issue from GitHub: add a github block to the review configuration.'); + const token = o.env.CLAUDE_CODE_OAUTH_TOKEN; + if (!token) throw new Error(RUNNER_CREDENTIAL_MISSING); + // Decision 1, step 4: the database path still names the locked file, now that the Store has opened it. + o.lock.verify(); + const runnerOwner = service.store.runnerOwnerToken(o.lock.file); + ownerOnlyDirectory(config.root); + // Per database (its runner token): retention reads references from this Store only, so it must never see another + // database's files. The lock is per database, so two runners can share a root or a configured directory. + const diagnosticsDir = ownerOnlyDirectory(config.diagnosticsDir ?? config.root, runnerOwner, 'diagnostics'); + o.onSlowStart?.(); + let built: string | undefined; + const image = () => built ??= (o.buildImage ?? buildAgentImage)(); + const recovery = await recoverStartup({ store: service.store, runnerOwner, runnerRoot: config.root, diagnosticsDir, + diagnosticsCapBytes: config.diagnosticsCapBytes, deps: (o.recovery ?? dRecoveryDeps)(image) }); + const imageId = image(); + const identity = review.identity; + const repository = await openRunnerRepository({ runnerRoot: config.root, runnerOwner, repositoryId: identity.repositoryId, source: review.repository }); + // The review reads runner commits from the same repository the runner writes; a configured path must agree. + if (review.runnerRepository !== undefined && review.runnerRepository !== repository.path) + throw new Error(`runnerRepository (${review.runnerRepository}) is not the runner's repository (${repository.path}); remove it from the configuration.`); + review.runnerRepository = repository.path; + const workspace = createTaskWorkspace({ store: service.store, runnerRoot: config.root, runnerOwner, repository, imageId, + limits: { ...EXECUTE_STORAGE, ...config.limits }, committer: config.committer }); + const issues = new GhIssueGateway(review.github.repository); + /** + * The last issue text read, reused for ISSUE_REUSE_MS: a plan's items run one after another and would otherwise each + * read the issue, every collaborator page and every comment page again. Only a completed read is kept. The window is + * measured on the monotonic clock, so a wall-clock step back cannot stretch it. + */ + let lastRead: { number: number; at: number; text: IssueText } | null = null; + const issueText = async (number: number, signal: AbortSignal): Promise => { + if (lastRead && lastRead.number === number && performance.now() - lastRead.at < ISSUE_REUSE_MS) return lastRead.text; + const at = performance.now(), text = await issues.issueText(number, { signal, timeoutMs: 30_000 }); + lastRead = { number, at, text }; + return text; + }; + const only = (requested: typeof identity) => { + if (identityKey(requested) !== identityKey(identity)) throw new Error('This server runs only its configured plan.'); + }; + const sources: ExecutionSources = { + planContext: requested => { only(requested); return service.planContext(); }, + issue: (requested, signal) => { only(requested); return issueText(service.store.getPlan(identity).issue, signal); }, + // Learning (L1–L4) is not built: no lesson is approved yet. + lessons: () => [], + // Codex is refused in every phase (#93). + vendor: () => 'claude', + }; + const findings = new SafetyFindings(service.store, o.capability); + const deps = executionDeps(service.store, workspace, claudeLauncher({ imageId, runnerRoot: config.root, runnerOwner, token }), sources, runnerOwner, findings, + { diagnostics: { directory: diagnosticsDir, capBytes: config.diagnosticsCapBytes ?? DEFAULT_DIAGNOSTICS_CAP_BYTES } }); + return { deps, sources, findings, recovery }; +} diff --git a/runner/question-container.ts b/runner/question-container.ts index 84ad9a19..53282f25 100644 --- a/runner/question-container.ts +++ b/runner/question-container.ts @@ -9,6 +9,7 @@ import type { RecoveredTaskStorage, TaskFilesystems, TaskStorageLimits } from '. import { RUNNER_LABEL, type ResourceOwner } from '../agents/labels.ts'; import { recoverLeftovers, type RecoveryReport } from '../agents/recovery.ts'; import { removeStaging } from './question-leftovers.ts'; +import { removalCommand } from './recovery.ts'; export type Provider = 'claude' | 'codex'; /** What the review knows about a question when it asks the agent. */ @@ -276,8 +277,7 @@ export async function recoverQuestionStorage(runnerOwner: string, deps: Pick resource.labels[RUNNER_LABEL] === runnerOwner); if (!ours.length) return; - const commands = ours.slice(0, MAX_LISTED).map(resource => - `docker ${resource.kind} rm${resource.kind === 'container' ? ' -f' : ''} ${resource.id ?? resource.name}`); + const commands = ours.slice(0, MAX_LISTED).map(removalCommand); if (ours.length > MAX_LISTED) commands.push(`… and ${ours.length - MAX_LISTED} more labelled ${RUNNER_LABEL}=${runnerOwner}`); throw new Error(`Ask is off: Docker objects labelled with this review's Ask owner are not ones codeboost can identify, so it did not remove them. Remove them, then retry:\n${commands.join('\n')}`); } diff --git a/runner/recovery.ts b/runner/recovery.ts index a3d796e4..b6b77b7d 100644 --- a/runner/recovery.ts +++ b/runner/recovery.ts @@ -1,4 +1,4 @@ -import { closeSync, constants, fstatSync, lstatSync, mkdirSync, openSync, readdirSync, realpathSync, rmSync, statSync, statfsSync, writeFileSync } from 'node:fs'; +import { closeSync, constants, fstatSync, lstatSync, mkdirSync, openSync, readdirSync, realpathSync, rmSync, statSync, statfsSync } from 'node:fs'; import { homedir } from 'node:os'; import { dirname, join, resolve } from 'node:path'; import { execFileSync } from 'node:child_process'; @@ -6,15 +6,23 @@ import { createRequire } from 'node:module'; import type { DatabaseSync } from 'node:sqlite'; import type { Store } from './store.ts'; import { WRITABLE_KINDS, isUuidV4 } from './lifecycle.ts'; +import { partialOutput, saveDiagnostic } from './diagnostics.ts'; +import { ownerOnlyDirectory } from './runner-repository.ts'; /** * Startup recovery and the single-runner lock. See docs/implementation/runner-lifecycle.md, - * "Startup recovery" and decision 1. D's recovery, export and removal are injected until #51 provides them. + * "Startup recovery" and decision 1. D's recovery, export and removal are injected: `runner/production.ts` passes D's own. */ export class LockHeld extends Error { constructor() { super('Another codeboost runner is using this database.'); } } export class RecoveryBlocked extends Error { readonly items: string[]; - constructor(message: string, items: string[]) { super(`${message}: ${items.join(', ')}`); this.items = items; } + /** One item per line, so a list of commands can be copied as it is. */ + constructor(message: string, items: string[]) { super(`${message}:\n${items.join('\n')}`); this.items = items; } +} +/** The shell line that removes a Docker object D reported, its name or ID quoted as one shell word. */ +export function removalCommand(resource: { kind: 'container' | 'network' | 'volume'; id?: string; name: string }): string { + const word = `'${(resource.id ?? resource.name).replaceAll("'", "'\\''")}'`; + return `docker ${resource.kind} rm${resource.kind === 'container' ? ' -f' : ''} ${word}`; } /** Linux statfs magic numbers for network filesystems, where POSIX locks are unreliable. */ @@ -118,8 +126,11 @@ export interface RecoveredStorage { readonly attemptId: string; readonly allocat export interface RecoveryDeps { /** D (#51): stop leftover agent containers, proxies and networks for this owner; keep task storage and return authenticated handles. */ recoverLeftovers(runnerOwner: string): Promise<{ storage: RecoveredStorage[]; unowned: string[] }>; - /** D (#51, before F2): bounded diff of a recovered task volume; must stop its own work when the signal aborts. */ - exportTaskDiff(handle: unknown, maxBytes: number, signal: AbortSignal): Promise; + /** + * D's bounded diff of a recovered task storage against `base`, the commit it was seeded from, checked against the + * `metadataBaseline` F saved at allocation. It must stop its own work when the signal aborts. + */ + exportTaskDiff(handle: unknown, input: { base: string; metadataBaseline: string }, maxBytes: number, signal: AbortSignal): Promise<{ diff: Buffer; truncated: boolean }>; removeTaskFilesystems(handle: unknown): Promise; /** F3: abort an interrupted rebase. Until F3 exists no rebase is ever recorded. */ abortRebase?(planKey: string, marker: unknown): Promise; @@ -128,6 +139,8 @@ export interface RecoveryDeps { export interface RecoveryOptions { store: Store; runnerOwner: string; runnerRoot: string; diagnosticsDir: string; deps: RecoveryDeps; now?: () => number; exportDeadlineMs?: number; graceMs?: number; + /** The diagnostics directory's total byte cap; retention as on the live path (`saveDiagnostic`). */ + diagnosticsCapBytes?: number; } export interface RecoveryReport { finalized: { attemptId: string; planKey: string; state: string; requeued: boolean }[]; @@ -150,23 +163,31 @@ export async function recoverStartup(o: RecoveryOptions): Promise isUuidV4(s.attemptId) && o.store.attemptOwner(s.attemptId) !== null); + if (recovered.unowned.length) throw new RecoveryBlocked('Docker holds codeboost objects this runner will not remove itself: objects without a runner label may belong to an older build that is still running, and objects of this runner it cannot identify are not ones it made. Stop every older codeboost process, check and run these commands, then start again', recovered.unowned); + // A handle is used only if both its attempt ID and its allocation ID match one attempt row of this database + // (runner-lifecycle.md, "Recovered storage handles"); any other is reported and left for a person. + const matched = recovered.storage.filter(s => isUuidV4(s.attemptId) && isUuidV4(s.allocationId) && o.store.attemptAllocation(s.attemptId) === s.allocationId); const unmatchedStorage = recovered.storage.filter(s => !matched.includes(s)).map(s => s.attemptId); // 3. Export phase: stopped writable attempts only, outside any transaction, fixed names, bounded deadline. const exports: Record = {}; - mkdirSync(o.diagnosticsDir, { recursive: true, mode: 0o700 }); + ownerOnlyDirectory(o.diagnosticsDir); for (const storage of matched) { const attempt = interrupted.find(a => a.id === storage.attemptId); if (!attempt || !WRITABLE_KINDS.includes(attempt.kind)) continue; + // Saved after D's allocation returned; a crash before that leaves nothing D's export would accept (fail closed). + if (!attempt.metadataBaseline || !attempt.storageBase) { + exports[storage.attemptId] = { failure: 'its storage baseline was never saved' }; + continue; + } const controller = new AbortController(), timer = setTimeout(() => controller.abort(new Error('Export timed out.')), o.exportDeadlineMs ?? 60_000); - const exporting = o.deps.exportTaskDiff(storage.handle, EXPORT_LIMIT, controller.signal); + const exporting = o.deps.exportTaskDiff(storage.handle, { base: attempt.storageBase, metadataBaseline: attempt.metadataBaseline }, EXPORT_LIMIT, controller.signal); try { - const diff = await Promise.race([exporting, + const { diff, truncated } = await Promise.race([exporting, new Promise((_, reject) => controller.signal.addEventListener('abort', () => reject(controller.signal.reason), { once: true }))]); - const file = join(o.diagnosticsDir, `${storage.attemptId}.diff`); - writeFileSync(file, diff.subarray(0, EXPORT_LIMIT), { mode: 0o600 }); - exports[storage.attemptId] = { diagnosticRef: file }; + // Earlier exports of this pass are not referenced until finalization, so retention must keep them too. + const saved = Object.values(exports).flatMap(entry => entry.diagnosticRef ? [entry.diagnosticRef] : []); + exports[storage.attemptId] = { diagnosticRef: saveDiagnostic(o.store, o.diagnosticsDir, storage.attemptId, + partialOutput(diff.subarray(0, EXPORT_LIMIT), truncated || diff.length > EXPORT_LIMIT), o.diagnosticsCapBytes, saved) }; } catch (error) { exports[storage.attemptId] = { failure: error instanceof Error ? error.message : String(error) }; } finally { clearTimeout(timer); } // A timed-out export must stop before step 4 removes its storage. D stops on abort; if it does not, fail closed. diff --git a/runner/review.ts b/runner/review.ts index a61ebcaf..c05c224c 100644 --- a/runner/review.ts +++ b/runner/review.ts @@ -19,6 +19,11 @@ export interface ReviewConfig { database: string; repository: string; export class ReviewService { store: Store; config: ReviewConfig; + /** + * The base tree listing of the last base commit read. A commit's tree never changes, so a later call for the same + * base reuses it instead of running Git again on the server's thread (each runner item asks for it, #91). + */ + #baseEntries: { base: string; entries: readonly BaseEntry[] } | null = null; constructor(config: ReviewConfig) { if (typeof config.pathIdentity?.caseSensitive !== 'boolean' || !['none', 'NFC'].includes(config.pathIdentity.unicodeNormalization)) throw new Error('Known checkout path identity is required.'); this.config = config; this.store = new Store(config.database); @@ -141,13 +146,18 @@ export class ReviewService { }; // Hardened like every repository Git call (#82, #83): no replace objects, hooks, network or inherited environment. const git = (args: string[], maxBuffer?: number) => execFileSync('git', [...HARDENED_GIT_OPTIONS, ...args], { cwd: repository, env: hardenedGitEnvironment(), encoding: 'utf8', maxBuffer, stdio: ['ignore', 'pipe', 'pipe'] }); - const listing = git(['ls-tree', '-rz', snapshot.base], 64 * 1024 * 1024); - const baseEntries: BaseEntry[] = listing.split('\0').filter(Boolean).map(record => { - const split = record.indexOf('\t'), [mode, , oid] = record.slice(0, split).split(' '), path = record.slice(split + 1); - if (mode === '160000') return { path, kind: 'gitlink' }; - if (mode === '120000') return { path, kind: 'symlink', target: git(['cat-file', 'blob', oid!]) }; - return { path, kind: 'file' }; - }); + if (this.#baseEntries?.base !== snapshot.base) { + const listing = git(['ls-tree', '-rz', snapshot.base], 64 * 1024 * 1024); + const entries: BaseEntry[] = listing.split('\0').filter(Boolean).map(record => { + const split = record.indexOf('\t'), [mode, , oid] = record.slice(0, split).split(' '), path = record.slice(split + 1); + if (mode === '160000') return { path, kind: 'gitlink' }; + if (mode === '120000') return { path, kind: 'symlink', target: git(['cat-file', 'blob', oid!]) }; + return { path, kind: 'file' }; + }); + this.#baseEntries = { base: snapshot.base, entries }; + } + // Copies: callers own what they are given, and the cached listing stays as Git reported it. + const baseEntries = this.#baseEntries.entries.map(entry => ({ ...entry })); return { identity, issue: plan.issue, baseEntries, pathKey, allowedCommands: [] }; } /** With an actionId (inside Store.userAction), feedback-producing actions record their event in the same transaction. */ diff --git a/runner/store.ts b/runner/store.ts index 461231cc..4369d0ac 100644 --- a/runner/store.ts +++ b/runner/store.ts @@ -65,6 +65,12 @@ export interface AttemptRecord { */ safetyFinding: string | null; } +/** A non-terminal attempt at startup, with what recovery needs to stop its preparation and export its storage. */ +export interface InterruptedAttempt extends AttemptRecord { + planKey: string; preparationPgid: number | null; preparationStartedAt: number | null; allocationId: string | null; + /** Saved after D's allocation returned (#91); null when it never did. */ + metadataBaseline: string | null; storageBase: 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 { @@ -96,8 +102,8 @@ export class Store { 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 as number; - if (![0, 1, 2, 3, 4, 5, 6, 7, 8].includes(version)) throw new Error('Unsupported store schema version.'); - if (version === 8) return; + if (![0, 1, 2, 3, 4, 5, 6, 7, 8, 9].includes(version)) throw new Error('Unsupported store schema version.'); + if (version === 9) 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)); @@ -134,6 +140,14 @@ export class Store { this.#db.exec('ALTER TABLE attempts ADD COLUMN safety_finding TEXT'); this.#db.exec('PRAGMA user_version=8'); } + // What startup recovery needs to export a recovered task storage (#91): D's metadata baseline and the commit the + // storage was seeded from. Idempotent, like v8. + if (version < 9) { + const columns = this.#db.prepare('PRAGMA table_info(attempts)').all().map(column => column.name); + if (!columns.includes('metadata_baseline')) this.#db.exec('ALTER TABLE attempts ADD COLUMN metadata_baseline TEXT'); + if (!columns.includes('storage_base')) this.#db.exec('ALTER TABLE attempts ADD COLUMN storage_base TEXT'); + this.#db.exec('PRAGMA user_version=9'); + } }); } catch (error) { this.#db.close(); throw error; } } @@ -1381,11 +1395,24 @@ export class Store { if (this.#run(`UPDATE attempts SET allocation_id=? WHERE plan_key=? AND id=? AND state='pending' AND allocation_id IS NULL`, allocationId, identityKey(identity), id).changes !== 1) throw new GuardRefusal('Allocation can be recorded once, for a pending attempt.'); } + /** + * Saved once D's allocation returns, for the allocation recorded before it: startup recovery passes both to D's export + * of a recovered storage, which refuses without the baseline. A crash before this write leaves them null, and that + * export then fails closed. + */ + recordAllocationBaseline(identity: PlanIdentity, id: string, allocationId: string, metadataBaseline: string, base: string): void { + if (!/^[0-9a-f]{64}$/.test(metadataBaseline)) throw new GuardRefusal('Invalid metadata baseline.'); + if (!/^(?:[0-9a-f]{40}|[0-9a-f]{64})$/.test(base)) throw new GuardRefusal('Invalid storage base commit.'); + if (this.#run(`UPDATE attempts SET metadata_baseline=?, storage_base=? WHERE plan_key=? AND id=? AND state='pending' AND allocation_id=? AND metadata_baseline IS NULL`, + metadataBaseline, base, identityKey(identity), id, allocationId).changes !== 1) + throw new GuardRefusal('The allocation baseline can be recorded once, for the pending attempt\'s own allocation.'); + } /** Every non-terminal attempt, across all plans, with the fields recovery needs. */ - interruptedAttempts(): (AttemptRecord & { planKey: string; preparationPgid: number | null; preparationStartedAt: number | null; allocationId: string | null })[] { + interruptedAttempts(): InterruptedAttempt[] { return this.#db.prepare("SELECT * FROM attempts WHERE state IN ('pending','running') ORDER BY rowid").all().map(row => ({ ...this.#attemptRecord(row), planKey: row.plan_key as string, preparationPgid: row.preparation_pgid as number | null, preparationStartedAt: row.preparation_started_at as number | null, allocationId: row.allocation_id as string | null, + metadataBaseline: row.metadata_baseline as string | null, storageBase: row.storage_base as string | null, })); } /** @@ -1447,6 +1474,10 @@ export class Store { attemptOwner(attemptId: string): string | null { return (this.#get('SELECT plan_key FROM attempts WHERE id=?', attemptId)?.plan_key as string | undefined) ?? null; } + /** The allocation ID F saved for an attempt, across all plans; null if none (or no such attempt). */ + attemptAllocation(attemptId: string): string | null { + return (this.#get('SELECT allocation_id FROM attempts WHERE id=?', attemptId)?.allocation_id as string | null | undefined) ?? null; + } /** Interrupted rebases recorded by F3 (none exist before F3). */ rebasesInProgress(): { planKey: string; marker: unknown }[] { return this.#db.prepare('SELECT plan_key, rebase_in_progress FROM tasks WHERE rebase_in_progress IS NOT NULL').all() diff --git a/runner/workspace.ts b/runner/workspace.ts index f6bce7fb..1b16f2f8 100644 --- a/runner/workspace.ts +++ b/runner/workspace.ts @@ -28,6 +28,8 @@ export interface WorkspaceOptions { /** What `materialize` keeps for the later steps, behind `WorkspaceRef.storage`. */ interface Held { readonly filesystems: TaskFilesystems; readonly attemptId: string; readonly identity: PlanIdentity } const held = (workspace: WorkspaceRef) => workspace.storage as Held; +/** The task storage D allocated for a workspace `materialize` returned: what D's start call mounts. */ +export const workspaceFilesystems = (workspace: WorkspaceRef): TaskFilesystems => held(workspace).filesystems; /** * The real task workspace over lane D (#87): a fresh clone of the runner-owned repository at the recorded head, copied @@ -60,6 +62,10 @@ export function createTaskWorkspace(options: WorkspaceOptions): TaskWorkspace { store.recordAllocation(identity, attempt.id, allocationId); const filesystems = await prepareTaskFilesystemsAsync(clone, options.limits, imageId, { runnerOwner: options.runnerOwner, attemptId: attempt.id, allocationId }, { signal, onProcessGroup: record, timeoutMs: 120_000 }); + // Startup recovery can export a recovered storage only with these (#91). Without them that export fails closed and + // records why, so a failed write costs only the partial output of a crash; it does not stop this attempt. + try { store.recordAllocationBaseline(identity, attempt.id, allocationId, filesystems.metadataBaseline, head); } + catch (error) { console.error(`Runner job ${attempt.id} could not save its storage baseline: ${JSON.stringify(error instanceof Error ? error.message : String(error))}`); } return { clone, storage: { filesystems, attemptId: attempt.id, identity } satisfies Held }; } catch (error) { // No subprocess ever started (its spawn failed at once): the "starting" marker must not block the next startup. diff --git a/test/cli.test.ts b/test/cli.test.ts new file mode 100644 index 00000000..42c0a2ea --- /dev/null +++ b/test/cli.test.ts @@ -0,0 +1,127 @@ +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 { BASE_IMAGE, CLAUDE_VERSION, CODEX_VERSION } from '../agents/container/image.ts'; +import { randomUUID } from 'node:crypto'; +import { fileURLToPath } from 'node:url'; +import { afterEach, expect, it, vi } from 'vitest'; +import { createDemo } from '../scripts/demo.ts'; +import { Store } from '../runner/store.ts'; +import { RUNNER_CREDENTIAL_MISSING } from '../runner/production.ts'; + +vi.setConfig({ testTimeout: 30_000 }); +const cli = fileURLToPath(new URL('../web/cli.ts', import.meta.url)); +const roots: string[] = []; +afterEach(() => { for (const root of roots.splice(0)) rmSync(root, { recursive: true, force: true }); }); +/** A non-demo review with a runner block, and a HOME of its own so the lock directory is the test's. */ +function review() { + const root = mkdtempSync(join(tmpdir(), 'codeboost-cli-')); roots.push(root); + const demo = createDemo(join(root, 'demo')), store = new Store(demo.database), issue = store.getPlan(demo.identity).issue; store.close(); + mkdirSync(join(root, 'runner'), { mode: 0o700 }); mkdirSync(join(root, 'home'), { mode: 0o700 }); + const config = join(root, 'review.json'); + writeFileSync(config, JSON.stringify({ ...demo, demo: false, github: { repository: 'owner/repo', pullRequest: 1, issue }, + runner: { root: join(root, 'runner'), committer: { name: 'codeboost', email: 'runner@codeboost.invalid' } } })); + /** + * A `docker` that answers every call with nothing (no leftovers, an image that matches the pinned profile) once the hold + * file is gone: startup recovery waits on it, as on a slow daemon, for as long as the test keeps the file. + */ + const fakeDocker = () => { + const bin = join(root, 'bin'), hold = join(bin, 'hold'), image = JSON.stringify([{ Id: `sha256:${'a'.repeat(64)}`, Config: { User: '10001:10001', + Labels: { 'org.opencontainers.image.base.name': BASE_IMAGE, 'io.codeboost.codex.version': CODEX_VERSION, 'io.codeboost.claude.version': CLAUDE_VERSION, 'io.codeboost.profile.version': '1' } } }]); + mkdirSync(bin, { mode: 0o700 }); writeFileSync(hold, ''); + writeFileSync(join(bin, 'docker'), `#!/bin/sh\nwhile [ -e '${hold}' ]; do sleep 0.05; done\nif [ "$1 $2" = "image inspect" ]; then printf '%s' '${image}'; fi\nexit 0\n`); + chmodSync(join(bin, 'docker'), 0o755); + return { path: `${bin}:${process.env.PATH}`, release: () => { if (existsSync(hold)) unlinkSync(hold); } }; + }; + /** The CLI as a child process with a token and the fake docker, its output collected. */ + const start = (docker: { path: string }) => { + const child = spawn(process.execPath, [cli, '--config', config, '--port', '0'], + { env: { ...process.env, HOME: join(root, 'home'), PATH: docker.path, CLAUDE_CODE_OAUTH_TOKEN: 'test-token' }, stdio: ['ignore', 'pipe', 'pipe'] }); + let stdout = '', stderr = ''; + child.stdout!.on('data', chunk => { stdout += chunk; }); child.stderr!.on('data', chunk => { stderr += chunk; }); + const until = async (check: () => boolean, what: string) => { + for (let n = 0; n < 400 && !check(); n++) await new Promise(resolve => setTimeout(resolve, 50)); + if (!check()) throw new Error(`Timed out waiting for ${what}. stdout: ${stdout} stderr: ${stderr}`); + }; + const exited = new Promise(resolve => child.once('exit', code => resolve(code))); + return { child, until, exited, out: () => stdout, err: () => stderr }; + }; + const run = (...args: string[]) => { + 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 }; +} + +it('refuses a runner startup without a token with its message, exit 1 and the lock released', () => { + const { run } = review(); + for (let n = 0; n < 2; n++) { + const result = run('--port', '0'); + expect(result.status).toBe(1); + // The second start meets the same refusal, not "Another codeboost runner": the first released the lock. + expect(result.stderr.trim()).toBe(RUNNER_CREDENTIAL_MISSING); + } +}); +it('refuses a bad port and an unknown preparation with a message, never a stack', () => { + const { run } = review(); + const port = run('--port', 'x'); + expect([port.status, port.stderr.trim()]).toEqual([1, 'Invalid port.']); + for (let n = 0; n < 2; n++) { + const release = run('--release-preparation', randomUUID()); + expect([release.status, release.stderr.trim()]).toEqual([1, 'Unknown attempt.']); + } +}); + +it('waits for startup recovery on a first Ctrl+C, then stops cleanly once startup has finished', async () => { + const { fakeDocker, start } = review(), docker = fakeDocker(); + const cli = start(docker); + try { + await cli.until(() => cli.out().includes('recovering what an earlier run left'), 'startup recovery to begin'); + cli.child.kill('SIGINT'); + await cli.until(() => cli.out().includes('Stopping once startup has finished'), 'the first-signal message'); + // Recovery is still waiting on Docker: the process has not stopped. + await new Promise(resolve => setTimeout(resolve, 300)); + expect(cli.child.exitCode).toBeNull(); + docker.release(); + expect(await cli.exited).toBe(0); + expect(cli.out()).not.toContain('Review ready'); + } finally { docker.release(); cli.child.kill('SIGKILL'); } +}); +it('stops at once with exit 130 on a second Ctrl+C during startup', async () => { + const { fakeDocker, start } = review(), docker = fakeDocker(); + const cli = start(docker); + try { + await cli.until(() => cli.out().includes('recovering what an earlier run left'), 'startup recovery to begin'); + cli.child.kill('SIGINT'); + await cli.until(() => cli.out().includes('Stopping once startup has finished'), 'the first-signal message'); + cli.child.kill('SIGINT'); + expect(await cli.exited).toBe(130); + expect(cli.err()).toContain('Stopped during startup. The next start recovers what this one left.'); + } finally { docker.release(); cli.child.kill('SIGKILL'); } +}); + +it('reports that --release-preparation needs --config, instead of printing help and exiting 0', () => { + const result = spawnSync(process.execPath, [cli, '--release-preparation', randomUUID()], { encoding: 'utf8', timeout: 20_000 }); + expect([result.status, result.stderr.trim()]).toEqual([1, '--release-preparation needs --config.']); +}); +it('ignores a runner block in a demo configuration opened with --config, and serves the demo', async () => { + const root = mkdtempSync(join(tmpdir(), 'codeboost-cli-')); roots.push(root); + const demo = createDemo(join(root, 'demo')); + mkdirSync(join(root, 'home'), { mode: 0o700 }); + const config = join(root, 'demo-with-runner.json'); + writeFileSync(config, JSON.stringify({ ...demo, runner: { root: join(root, 'runner'), committer: { name: 'codeboost', email: 'runner@codeboost.invalid' } } })); + const child = spawn(process.execPath, [cli, '--config', config, '--port', '0'], { env: { ...process.env, HOME: join(root, 'home') }, stdio: ['ignore', 'pipe', 'pipe'] }); + let stdout = '', stderr = ''; + child.stdout!.on('data', chunk => { stdout += chunk; }); child.stderr!.on('data', chunk => { stderr += chunk; }); + const exited = new Promise(resolve => child.once('exit', code => resolve(code))); + try { + for (let n = 0; n < 400 && !stdout.includes('Review ready') && child.exitCode === null; n++) await new Promise(resolve => setTimeout(resolve, 50)); + expect(stderr).toBe(''); + expect(stdout).toContain('Review ready'); + expect(stdout).not.toContain('recovering what an earlier run left'); + child.kill('SIGINT'); + expect(await exited).toBe(0); + } finally { child.kill('SIGKILL'); } +}); diff --git a/test/issues.test.ts b/test/issues.test.ts index 78991af3..5c96f27b 100644 --- a/test/issues.test.ts +++ b/test/issues.test.ts @@ -1,5 +1,6 @@ import { describe, expect, it, vi } from 'vitest'; -import { GhIssueGateway, ISSUE_PAGE_MAX_BYTES } from '../github/issues.ts'; +import { GhIssueGateway, ISSUE_PAGE_MAX_BYTES, type IssueText } from '../github/issues.ts'; +import { prepareExecution } from '../core/execution-prompt.ts'; const rawIssue = (overrides: Record = {}) => ({ number: 7, @@ -231,3 +232,56 @@ describe('GitHub issue retrieval', () => { await expect(gateway.fetch({ signal: controller.signal })).rejects.toBe(cancelled); }); }); + +describe('issue text for an execute prompt (#91)', () => { + const comment = (login: string | null, body: string) => ({ body, user: login === null ? null : { login } }); + const gateway = (pages: unknown[][], issue: Record = rawIssue(), collaborators = ['member']) => { + const calls: string[][] = []; + return { calls, gateway: new GhIssueGateway('owner/repo', async args => { + calls.push([...args]); + if (isCollaboratorRequest(args)) return JSON.stringify(collaborators.map(login => ({ login }))); + const path = args[5]!; + if (path.endsWith('/comments')) return JSON.stringify(pages[Number(args.at(-1)!.split('=')[1]) - 1] ?? []); + return JSON.stringify(issue); + }) }; + }; + it('keeps only collaborators\' comments, oldest first, across pages', async () => { + const full = Array.from({ length: 100 }, (_, n) => comment(n % 2 ? 'Member' : 'outsider', `c${n}`)); + const { gateway: g, calls } = gateway([full, [comment('member', 'last'), comment(null, 'ghost')]]); + const text = await g.issueText(7); + expect(text).toMatchObject({ number: 7, title: 'Fix retries', body: 'Keep issue text as data.' }); + expect(text.comments).toEqual([...full.filter((_, n) => n % 2).map(c => c.body), 'last']); + expect(calls.filter(args => args[5]!.endsWith('/comments')).map(args => args.at(-1))).toEqual(['page=1', 'page=2']); + }); + it('refuses a pull request, a different issue, and text too long for a prompt', async () => { + await expect(gateway([], rawIssue({ pull_request: { url: 'x' } })).gateway.issueText(7)).rejects.toThrow(/is a pull request/); + await expect(gateway([], rawIssue({ number: 8 })).gateway.issueText(7)).rejects.toThrow(/different issue/); + const long = Array.from({ length: 9 }, () => comment('member', 'x'.repeat(65_000))); + await expect(gateway([long]).gateway.issueText(7)).rejects.toThrow(/larger than the 32 KiB an execute prompt carries/); + }); + it('accepts exactly what an execute prompt can carry, end to end, and refuses the rest at the fetch (#91)', async () => { + const promptOf = (issue: IssueText) => prepareExecution({ identity: { repositoryId: 'repo', taskId: 'task', planId: 'plan' }, attemptId: 'attempt-1', + mode: 'execute', itemId: 'P1', approvedLessons: [], allowedCommands: [], issue, + plan: { schema_version: 1, issue: 7, revision: 1, summary: 'S', questions: [], items: [{ id: 'P1', title: 'T', intent: 'I', + files: [{ path: 'a', kind: 'edit', renamed_from: null, change: 'x' }], acceptance: [], depends_on: [] }] } }); + // Under the budget: what the fetch returns, the prompt carries. + const fits = await gateway([[comment('member', 'x'.repeat(30_000))]]).gateway.issueText(7); + expect(promptOf(fits).prompt).toContain('x'.repeat(30_000)); + // Small in characters, over the budget once escaped as the prompt escapes it ('<' becomes \u003c): refused at the + // fetch, with the reason, instead of at every attempt's preparation. + await expect(gateway([[comment('member', '<'.repeat(6_000))]]).gateway.issueText(7)).rejects.toThrow(/larger than the 32 KiB/); + expect(() => promptOf({ number: 7, title: 'Fix retries', body: 'Keep issue text as data.', comments: ['<'.repeat(6_000)] })).toThrow(/exceeds 32 KiB/); + }); + it('ignores an oversized comment from someone else, and reads exactly the page limit', async () => { + const huge = comment('outsider', 'x'.repeat(70_000)); + expect((await gateway([[huge, comment('member', 'kept')]]).gateway.issueText(7)).comments).toEqual(['kept']); + const pages = Array.from({ length: 10 }, () => Array.from({ length: 100 }, () => comment('outsider', 'spam'))); + 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('stops on the caller\'s abort', async () => { + const controller = new AbortController(); + controller.abort(new Error('stopped')); + await expect(gateway([]).gateway.issueText(7, { signal: controller.signal })).rejects.toThrow(/stopped/); + }); +}); diff --git a/test/publish.test.ts b/test/publish.test.ts index 5b9d74b4..9e32ea54 100644 --- a/test/publish.test.ts +++ b/test/publish.test.ts @@ -324,7 +324,7 @@ describe('schema v7', () => { expect(upgraded.taskPullRequests(identity)).toEqual([]); expect(upgraded.latestAlreadyFixed(identity)).toBeNull(); const db = new DatabaseSync(path, { readOnly: true }); - expect(db.prepare('PRAGMA user_version').get()).toEqual({ user_version: 8 }); + expect(db.prepare('PRAGMA user_version').get()).toEqual({ user_version: 9 }); const names = db.prepare("SELECT name FROM sqlite_master WHERE type IN ('table','index') AND (name LIKE '%pull_requests%' OR name LIKE 'already_fixed_checks%') AND name NOT LIKE 'sqlite_autoindex%' ORDER BY name").all().map(row => row.name); expect(names).toEqual(['already_fixed_checks', 'already_fixed_checks_task', 'task_pull_requests', 'task_pull_requests_number', 'task_pull_requests_opening', 'task_pull_requests_task']); db.close(); diff --git a/test/question-agent.test.ts b/test/question-agent.test.ts index 6ec675d4..66f21339 100644 --- a/test/question-agent.test.ts +++ b/test/question-agent.test.ts @@ -421,7 +421,7 @@ it('keeps Ask off, with removal commands, for objects of its own owner that reco const handle = recovered(1); const error = await recoverQuestionStorage(OWNER, recovery({ storage: [handle], unowned: [mine, theirs] }, value => { removed.push(value); }).deps, new RetainedStorage()).catch((caught: Error) => caught); - expect((error as Error).message.split('\n')).toEqual([expect.stringContaining("labelled with this review's Ask owner"), 'docker container rm -f id-codeboost-odd']); + expect((error as Error).message.split('\n')).toEqual([expect.stringContaining("labelled with this review's Ask owner"), "docker container rm -f 'id-codeboost-odd'"]); // Storage it could identify is still removed. expect(removed).toEqual([handle]); }); diff --git a/test/review.test.ts b/test/review.test.ts index cc923d01..8a8b1503 100644 --- a/test/review.test.ts +++ b/test/review.test.ts @@ -175,3 +175,14 @@ it('keeps observing the user\'s HEAD for a task the runner has not committed to, fixtureGit(config.repository,'commit','--allow-empty','-m','User work'); expect(service.load().snapshot.head).toBe(fixtureGit(config.repository,'rev-parse','HEAD')); }); +it('reads a base commit\'s tree once and hands each caller its own copy (#91)',()=>{ + const {service,config}=fixture(); + const first=service.planContext(); + expect(first.baseEntries.length).toBeGreaterThan(0); + first.baseEntries[0]!.path='changed by a caller'; + // Git is not asked again for the same base: the listing survives the repository becoming unreadable. + service.config={...config,repository:join(config.repository,'missing')}; + const second=service.planContext(); + expect(second.baseEntries[0]!.path).not.toBe('changed by a caller'); + expect(second.baseEntries.slice(1)).toEqual(first.baseEntries.slice(1)); +}); diff --git a/test/runner-coordinator.test.ts b/test/runner-coordinator.test.ts index e8e6f3de..f9bf6881 100644 --- a/test/runner-coordinator.test.ts +++ b/test/runner-coordinator.test.ts @@ -72,6 +72,17 @@ async function until(check: () => boolean, label: string) { } describe('admission and slots', () => { + it('refuses a kind its deps cannot run before writing anything (#91)', () => { + const { store } = setup(); + const limited = new RunnerCoordinator(store, { ...fakeD().deps, kinds: ['execute'] }); coordinators.push(limited); + const before = store.getTask(A).stateVersion; + expect(() => limited.start(A, request(store, A, { kind: 'review' }))).toThrow(/cannot run review attempts/); + expect(store.getTask(A).stateVersion).toBe(before); + expect(store.getAttempts(A)).toEqual([]); + expect(limited.isActive(A)).toBe(false); + // What the view uses to offer retry: the same answer admission gives. + expect([limited.runs('review'), limited.runs('execute')]).toEqual([false, true]); + }); it('runs an attempt to completion, removes its preparation files and frees its slot', async () => { const { store, runner, launches, preparations, cleaned } = setup(); const attempt = runner.start(A, request(store, A)); diff --git a/test/runner-execution.test.ts b/test/runner-execution.test.ts index e2369f43..12ca94ce 100644 --- a/test/runner-execution.test.ts +++ b/test/runner-execution.test.ts @@ -38,7 +38,7 @@ function setup(options: { manifests?: Record void; /** The fake agent's result names this attempt instead. */ - foreignResult?: boolean } = {}) { + foreignResult?: boolean; issue?: ExecutionSources['issue'] } = {}) { const dir = mkdtempSync(join(tmpdir(), 'codeboost-exec-')); dirs.push(dir); const path = join(dir, 'state.sqlite'), store = new Store(path); store.createPlan(JSON.stringify(options.plan ?? plan), 'json', context, oid(1), oid(2)); @@ -79,7 +79,7 @@ function setup(options: { manifests?: Record { throw options.pathKeyError; } } : context; - const sources: ExecutionSources = { planContext: () => auditContext, issue: () => ({ number: 1, title: 'Issue', body: 'Please fix', comments: [] }), lessons: () => [], vendor: () => 'claude' }; + const sources: ExecutionSources = { planContext: () => auditContext, issue: options.issue ?? (() => ({ number: 1, title: 'Issue', body: 'Please fix', comments: [] })), lessons: () => [], vendor: () => 'claude' }; const prompts: string[] = [], argv: (readonly (readonly string[])[])[] = [], owners: string[] = [], checks: unknown[] = []; const capability = options.capability?.(store), findings = new SafetyFindings(store, capability); if (options.settleError) store.settleAttempt = () => { throw Object.assign(new Error('disk full'), { code: 'ERR_SQLITE_ERROR' }); }; @@ -308,6 +308,37 @@ describe('item execution', () => { expect(await h.executor.runTask(identity)).toMatchObject({ kind: 'needs amendment', item: 'P1' }); expect(store.getTask(identity).status).toBe('needs amendment'); }); + it('keeps the Store open at shutdown until a run\'s pause after its last attempt is written (#91)', async () => { + let releaseStarted!: () => void, finishRelease!: () => void; + const started = new Promise(resolve => { releaseStarted = resolve; }); + const gate = new Promise(resolve => { finishRelease = resolve; }); + const h = setup({ manifests: { P1: manifest([change('a.ts'), change('extra.ts', { kind: 'add', oldType: undefined })]) }, + capability: s => s.shutdownCapability(), release: async () => { releaseStarted(); await gate; } }); + // The run resumes some time after its job settles; nothing orders that before the coordinator's close returns. + const settled = h.runner.settled.bind(h.runner); + h.runner.settled = async target => { await settled(target); await new Promise(resolve => setTimeout(resolve, 30)); }; + const run = h.executor.runTask(identity); + await started; + // Shutdown, server order: close the write gate, stop the coordinator, then await the executor. + h.store.closeWrites(); + let closed = false; + const closing = h.runner.close().then(() => h.executor.close()).then(() => { closed = true; }); + await new Promise(resolve => setTimeout(resolve, 20)); + expect(closed).toBe(false); + finishRelease(); + await closing; + // The pause landed through the capability before close returned, so closing the Store now loses nothing. + expect(h.store.getTask(identity).status).toBe('needs amendment'); + await expect(run).resolves.toMatchObject({ kind: 'needs amendment', item: 'P1' }); + }); + it('fetches the issue for each attempt, and fails it before any storage when the fetch fails', async () => { + const signals: AbortSignal[] = []; + const h = setup({ issue: (_identity, signal) => { signals.push(signal); return Promise.reject(new Error('Issue retrieval timed out.')); } }); + expect(await h.executor.runTask(identity)).toMatchObject({ kind: 'stopped', item: 'P1', state: 'failed' }); + expect(signals).toHaveLength(1); + expect(h.log).toEqual([]); + expect(h.store.getAttempts(identity)[0]!.diagnostic).toMatch(/Issue retrieval timed out/); + }); 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-lifecycle-store.test.ts b/test/runner-lifecycle-store.test.ts index 840d03e1..7e2cf2d3 100644 --- a/test/runner-lifecycle-store.test.ts +++ b/test/runner-lifecycle-store.test.ts @@ -662,6 +662,30 @@ describe('durable safety findings (#87 item 3)', () => { db.exec('ALTER TABLE attempts DROP COLUMN safety_finding; PRAGMA user_version=7;'); db.close(); const reopened = open(path); expect(reopened.getAttempt(identity, attempt.id).safetyFinding).toBeNull(); - expect(new DatabaseSync(path).prepare('PRAGMA user_version').get()).toEqual({ user_version: 8 }); + expect(new DatabaseSync(path).prepare('PRAGMA user_version').get()).toEqual({ user_version: 9 }); + }); +}); + +describe('allocation baseline (#91)', () => { + it('saves the baseline once, for the pending attempt\'s own allocation, and hands it to recovery', () => { + const { store } = queued(); const attempt = admit(store), allocationId = randomUUID(); + expect(() => store.recordAllocationBaseline(identity, attempt.id, allocationId, 'b'.repeat(64), oid(2))).toThrow(GuardRefusal); + store.recordAllocation(identity, attempt.id, allocationId); + expect(() => store.recordAllocationBaseline(identity, attempt.id, randomUUID(), 'b'.repeat(64), oid(2))).toThrow(GuardRefusal); + expect(() => store.recordAllocationBaseline(identity, attempt.id, allocationId, 'short', oid(2))).toThrow(/baseline/); + expect(() => store.recordAllocationBaseline(identity, attempt.id, allocationId, 'b'.repeat(64), 'HEAD')).toThrow(/base commit/); + store.recordAllocationBaseline(identity, attempt.id, allocationId, 'b'.repeat(64), oid(2)); + expect(() => store.recordAllocationBaseline(identity, attempt.id, allocationId, 'c'.repeat(64), oid(2))).toThrow(GuardRefusal); + expect(store.attemptAllocation(attempt.id)).toBe(allocationId); + expect(store.interruptedAttempts()).toEqual([expect.objectContaining({ id: attempt.id, allocationId, metadataBaseline: 'b'.repeat(64), storageBase: oid(2) })]); + }); + it('adds the baseline columns to a version 8 database', () => { + const { path, store } = queued(); const attempt = admit(store); + store.close(); stores.splice(stores.indexOf(store), 1); + const db = new DatabaseSync(path); + db.exec('ALTER TABLE attempts DROP COLUMN metadata_baseline; ALTER TABLE attempts DROP COLUMN storage_base; PRAGMA user_version=8;'); db.close(); + const reopened = open(path); + expect(reopened.interruptedAttempts()).toEqual([expect.objectContaining({ id: attempt.id, metadataBaseline: null, storageBase: null })]); + expect(new DatabaseSync(path).prepare('PRAGMA user_version').get()).toEqual({ user_version: 9 }); }); }); diff --git a/test/runner-production.test.ts b/test/runner-production.test.ts new file mode 100644 index 00000000..e655fb0c --- /dev/null +++ b/test/runner-production.test.ts @@ -0,0 +1,273 @@ +import { existsSync, lstatSync, mkdirSync, mkdtempSync, readFileSync, readdirSync, rmSync, statSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { randomUUID } from 'node:crypto'; +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 type { RecoveryDeps } from '../runner/recovery.ts'; +import type { AgentAdapterRequest } from '../agents/adapters/types.ts'; +import { readCapturedFile } from '../agents/container/profile.ts'; +import type { InvocationInput } from '../agents/contract.ts'; +import { recoverLeftovers } from '../agents/recovery.ts'; +import { exportTaskDiff, removeTaskFilesystemsAsync } from '../agents/container/storage.ts'; +import { RunnerCoordinator, type RunnerDeps } from '../runner/coordinator.ts'; +import { SafetyFindings, type ExecutionSources } from '../runner/execution.ts'; + +vi.mock('../agents/recovery.ts', async original => ({ ...await original(), recoverLeftovers: vi.fn() })); +const issueReads: number[] = []; +vi.mock('../github/issues.ts', async original => { + const actual = await original(); + return { ...actual, GhIssueGateway: class extends actual.GhIssueGateway { + override async issueText(number: number) { issueReads.push(number); return { number, title: 'T', body: 'B', comments: [] }; } + } }; +}); +vi.mock('../agents/container/storage.ts', async original => ({ ...await original(), + exportTaskDiff: vi.fn(async () => ({ diff: Buffer.from('d'), truncated: false })), removeTaskFilesystemsAsync: vi.fn(async () => undefined) })); + +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 }); + vi.restoreAllMocks(); +}); +const OWNER = 'c'.repeat(32), committer = { name: 'codeboost', email: 'runner@codeboost.invalid' }; +function fixture() { + const root = mkdtempSync(join(tmpdir(), 'codeboost-production-')); roots.push(root); + const demo = createDemo(join(root, 'demo')); + const service = new ReviewService(demo); + const issue = service.store.getPlan(demo.identity).issue; + // A real review, not a demo: the runner needs a github block for the issue text. + const config = { ...demo, demo: false, github: { repository: 'owner/repo', pullRequest: 1, issue } }; + service.config = config; + cleanups.push(() => { try { service.close(); } catch { /* closed by the test */ } }); + return { root, service, config, demo }; +} +const recovery = (calls: string[], over: Partial = {}): RecoveryDeps => ({ + recoverLeftovers: async () => { calls.push('recover'); return { storage: [], unowned: [] }; }, + exportTaskDiff: async () => ({ diff: Buffer.alloc(0), truncated: false }), + removeTaskFilesystems: async () => undefined, + ...over, +}); +const lock = (calls: string[], fail = false) => ({ file: { dev: 1n, ino: 2n }, verify: () => { calls.push('verify'); if (fail) throw new Error('The database path changed while opening. Refusing to start.'); } }); +const capability = { run: (fn: () => T) => fn() }; + +describe('runner configuration', () => { + it('accepts a complete block and refuses relative paths, bad committers and unknown limits', () => { + expect(parseRunnerConfig({ root: '/r', committer, limits: { workBytes: 1024 } })).toMatchObject({ root: '/r', committer, limits: { workBytes: 1024 } }); + expect(() => parseRunnerConfig({ root: 'r', committer })).toThrow(/runner.root must be an absolute path/); + expect(() => parseRunnerConfig({ root: '/work,old/runner', committer })).toThrow(/runner.root cannot contain a comma/); + expect(() => parseRunnerConfig({ root: '/r', diagnosticsDir: '/d\nx', committer })).toThrow(/runner.diagnosticsDir cannot contain/); + expect(() => parseRunnerConfig({ root: '/r', diagnosticsDir: 'd', committer })).toThrow(/diagnosticsDir/); + expect(() => parseRunnerConfig({ root: '/r', committer: { name: 'x', email: 'a' } })).toThrow(/committer/); + expect(() => parseRunnerConfig({ root: '/r', committer, limits: { bytes: 1 } })).toThrow(/runner.limits.bytes/); + // Inherited names are not limits either. + for (const key of ['toString', 'constructor', '__proto__']) expect(() => parseRunnerConfig({ root: '/r', committer, limits: JSON.parse(`{"${key}": 1}`) })).toThrow(/runner.limits/); + expect(() => parseRunnerConfig({ root: '/r', committer, limits: { workBytes: 0 } })).toThrow(/runner.limits.workBytes/); + expect(() => parseRunnerConfig([])).toThrow(/must be an object/); + }); +}); + +describe('runner startup', () => { + it('verifies the lock, recovers, then builds the image, before it assembles anything', async () => { + const { root, service } = fixture(), calls: string[] = []; + const assembly = await setUpRunner({ service, capability, config: { root: join(root, 'runner'), committer }, lock: lock(calls), env: { CLAUDE_CODE_OAUTH_TOKEN: 't' }, + buildImage: () => { calls.push('image'); return 'sha256:x'; }, recovery: () => { calls.push('deps'); return recovery(calls); } }); + // D's recovery stops every leftover agent before the (possibly long) image build. + expect(calls).toEqual(['verify', 'deps', 'recover', 'image']); + expect(assembly.deps.kinds).toEqual(['execute']); + // The token is the database's own, kept for this file identity. + expect(assembly.deps.runnerOwner).toBe(service.store.runnerOwnerToken({ dev: 1n, ino: 2n })); + // The review now reads runner commits from the repository the runner writes. + const repositories = join(root, 'runner', assembly.deps.runnerOwner, 'repositories'); + expect(service.config.runnerRepository).toBe(join(repositories, readdirSync(repositories)[0]!)); + // Per database, under its runner token: another database sharing the root keeps its own folder. + expect(statSync(join(root, 'runner', assembly.deps.runnerOwner, 'diagnostics')).mode & 0o777).toBe(0o700); + expect(existsSync(join(root, 'runner', 'diagnostics'))).toBe(false); + expect(await assembly.sources.vendor(service.config.identity)).toBe('claude'); + expect(() => assembly.sources.planContext({ ...service.config.identity, planId: 'other' })).toThrow(/only its configured plan/); + }); + it('reads the issue once for a run of items, and again once the reuse window has passed', async () => { + const { root, service } = fixture(); + const { sources } = await setUpRunner({ service, capability, config: { root: join(root, 'runner'), committer }, lock: lock([]), env: { CLAUDE_CODE_OAUTH_TOKEN: 't' }, + buildImage: () => 'x', recovery: () => recovery([]) }); + issueReads.length = 0; + const signal = new AbortController().signal, identity = service.config.identity; + await sources.issue(identity, signal); await sources.issue(identity, signal); + expect(issueReads).toHaveLength(1); + // A wall-clock step back does not keep the old text. + vi.spyOn(Date, 'now').mockReturnValue(0); + await sources.issue(identity, signal); + expect(issueReads).toHaveLength(1); + const now = performance.now(); + vi.spyOn(performance, 'now').mockReturnValue(now + ISSUE_REUSE_MS + 1); + await sources.issue(identity, signal); + expect(issueReads).toHaveLength(2); + }); + it('refuses before touching anything without a github block, a token, or with a demo', async () => { + for (const [patch, env, message] of [[{ github: undefined }, { CLAUDE_CODE_OAUTH_TOKEN: 't' }, /github block/], [{}, {}, RUNNER_CREDENTIAL_MISSING], [{ demo: true }, { CLAUDE_CODE_OAUTH_TOKEN: 't' }, /Demos never/]] as const) { + const { root, service, config } = fixture(), calls: string[] = []; + service.config = { ...config, ...patch }; + await expect(setUpRunner({ service, capability, config: { root: join(root, 'runner'), committer }, lock: lock(calls), env, + buildImage: () => { calls.push('image'); return 'x'; }, recovery: () => recovery(calls) })).rejects.toThrow(message); + expect(calls).toEqual([]); + expect(existsSync(join(root, 'runner'))).toBe(false); + } + }); + it('stops startup when the lock no longer names the database, or recovery fails, and admits nothing', async () => { + const { root, service } = fixture(), calls: string[] = []; + await expect(setUpRunner({ service, capability, config: { root: join(root, 'runner'), committer }, lock: lock(calls, true), env: { CLAUDE_CODE_OAUTH_TOKEN: 't' }, + buildImage: () => { calls.push('image'); return 'x'; }, recovery: () => recovery(calls) })).rejects.toThrow(/path changed/); + expect(calls).toEqual(['verify']); + await expect(setUpRunner({ service, capability, config: { root: join(root, 'runner'), committer }, lock: lock([]), env: { CLAUDE_CODE_OAUTH_TOKEN: 't' }, + buildImage: () => 'x', recovery: () => recovery([], { recoverLeftovers: async () => { throw new Error('docker down'); } }) })).rejects.toThrow(/docker down/); + }); + it('refuses a configured runnerRepository that is not the runner\'s own', async () => { + const { root, service, config } = fixture(); + service.config = { ...config, runnerRepository: join(root, 'elsewhere.git') }; + await expect(setUpRunner({ service, capability, config: { root: join(root, 'runner'), committer }, lock: lock([]), env: { CLAUDE_CODE_OAUTH_TOKEN: 't' }, + buildImage: () => 'x', recovery: () => recovery([]) })).rejects.toThrow(/is not the runner's repository/); + }); +}); + +describe('server with a runner setup', () => { + it('runs the setup before it listens, and closes the Store without listening when the setup fails', async () => { + const { demo } = fixture(); + let closed = false; + const setup = vi.fn(async (service: ReviewService) => { + const close = service.store.close.bind(service.store); + service.store.close = () => { closed = true; close(); }; + throw new Error('recovery blocked'); + }); + await expect(startServer({ ...demo }, 0, undefined, undefined, 2_000, undefined, undefined, undefined, setup)).rejects.toThrow(/recovery blocked/); + expect(setup).toHaveBeenCalledOnce(); + expect(closed).toBe(true); + }); + /** A runner whose agent never runs: enough to drive the server's own wiring. */ + const assembly = (service: ReviewService) => { + const deps: RunnerDeps = { runnerOwner: OWNER, 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' }; + return { deps, sources, findings: new SafetyFindings(service.store), recovery: { finalized: [], requeue: [], removedDirectories: [], unknownEntries: [], unmatchedStorage: [], repairedMerges: [] } }; + }; + it('closes the Store only after the plan runs in progress, and after a failing step', async () => { + const { demo } = fixture(); + const app = await startServer({ ...demo }, 0, undefined, undefined, 2_000, undefined, undefined, undefined, async service => assembly(service)); + const order: string[] = []; + const executorClose = app.executor!.close.bind(app.executor); + // A later failure must not hide the first one. + app.executor!.close = async () => { await new Promise(resolve => setTimeout(resolve, 20)); order.push('executor'); await executorClose(); throw new Error('executor close failed'); }; + const storeClose = app.service.store.close.bind(app.service.store); + app.service.store.close = () => { order.push('store'); storeClose(); }; + vi.spyOn(app.runner as RunnerCoordinator, 'close').mockRejectedValueOnce(new Error('runner close failed')); + await expect(app.close()).rejects.toThrow(/runner close failed/); + // A failing step skips nothing after it: plan runs are still awaited before the Store closes. + expect(order).toEqual(['executor', 'store']); + // Without the failure: executor first, then the Store. + const second = await startServer({ ...demo }, 0, undefined, undefined, 2_000, undefined, undefined, undefined, async service => assembly(service)); + const order2: string[] = []; + const close2 = second.executor!.close.bind(second.executor); + second.executor!.close = async () => { await new Promise(resolve => setTimeout(resolve, 20)); order2.push('executor'); return close2(); }; + const store2 = second.service.store.close.bind(second.service.store); + second.service.store.close = () => { order2.push('store'); store2(); }; + await second.close(); + expect(order2).toEqual(['executor', 'store']); + }); + it('refuses to retry a plan item outside the executor', async () => { + const { demo } = fixture(); + const app = await startServer({ ...demo }, 0, undefined, undefined, 2_000, undefined, undefined, undefined, async service => { + const store = service.store, identity = demo.identity; + if (store.getTask(identity).status !== 'queued') store.transitionTask(identity, store.getTask(identity).stateVersion, 'queued'); + const attempt = store.admitAttempt(identity, { expectedStateVersion: store.getTask(identity).stateVersion, kind: 'execute', item: store.getPlan(identity).items[0]!.id, + expectedContext: store.currentContext(identity), deadline: Date.now() + 60_000 }); + store.markRunning(identity, attempt.id); + store.settleAttempt(identity, attempt.id, { firstReason: null, exitCode: 1, valid: false }); + return assembly(service); + }); + cleanups.push(() => app.close()); + const origin = new URL(app.url).origin, headers = { 'x-codeboost-token': app.token }; + const view = await (await fetch(`${origin}/api/runner`, { headers })).json() as { stateVersion: number; attempts: { id: string }[]; retryable: boolean }; + // The view must not offer what the action refuses. + 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(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 () => { + for (const [demoMode, message] of [[false, RUNNER_NOT_CONFIGURED], [true, RUNNER_NOT_IN_DEMO]] as const) { + const { demo } = fixture(); + const app = await startServer({ ...demo, demo: demoMode }, 0, undefined, undefined, 2_000); + cleanups.push(() => app.close()); + const task = await (await fetch(`${new URL(app.url).origin}/api/runner`, { headers: { 'x-codeboost-token': app.token } })).json() as { stateVersion: number; available: boolean }; + expect(task.available).toBe(false); + 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: 'retry', attemptId: randomUUID(), expectedStateVersion: task.stateVersion, actionId: randomUUID() }) }); + expect(await response.json()).toEqual({ error: message }); + } + }); +}); + +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() }); + const spy = vi.mocked(recoverLeftovers).mockResolvedValue({ removed: [], storage: [handle], unowned: [ + { kind: 'volume', name: 'legacy', labels: {}, reason: 'no-runner-label' }, + { kind: 'container', name: 'ours', id: 'abc', labels: { 'io.codeboost.runner': OWNER }, reason: 'unknown-kind' }, + ] }); + const report = await dRecoveryDeps(() => 'sha256:x').recoverLeftovers(OWNER); + // The daemon-wide search for unlabelled objects stays on (runner-lifecycle.md, "Unowned resources"). + expect(spy).toHaveBeenCalledWith(OWNER, 120_000); + expect(report.storage[0]!.handle).toBe(handle); + // Each line runs as it is in a shell; the reason is a comment. + expect(report.unowned).toEqual(["docker volume rm 'legacy' # no-runner-label", "docker container rm -f 'abc' # unknown-kind"]); + }); + it('builds the image before exporting, exports with the row\'s base and baseline, and removes D\'s own handle', async () => { + let builds = 0; + // Memoized, as setUpRunner passes it. + let built: string | undefined; + const deps = dRecoveryDeps(() => built ??= (builds++, 'sha256:x')), handle = { recovered: true }, signal = new AbortController().signal; + vi.mocked(recoverLeftovers).mockResolvedValue({ removed: [], storage: [], unowned: [] }); + await deps.recoverLeftovers(OWNER); + // Nothing to export: the build waits until recovery ends. + expect(builds).toBe(0); + // Storage to export: built before recoverStartup arms any export deadline. + vi.mocked(recoverLeftovers).mockResolvedValue({ removed: [], storage: [Object.freeze({ runnerOwner: OWNER, attemptId: randomUUID(), allocationId: randomUUID() })], unowned: [] }); + await deps.recoverLeftovers(OWNER); + expect(builds).toBe(1); + await deps.exportTaskDiff(handle, { base: 'a'.repeat(40), metadataBaseline: 'b'.repeat(64) }, 1024, signal); + expect(vi.mocked(exportTaskDiff)).toHaveBeenCalledWith(handle, { base: 'a'.repeat(40), metadataBaseline: 'b'.repeat(64), imageId: 'sha256:x', maxBytes: 1024, signal }); + expect(builds).toBe(1); + await deps.removeTaskFilesystems(handle); + expect(vi.mocked(removeTaskFilesystemsAsync).mock.calls[0]![0]).toBe(handle); + }); + it('launches Claude with a schema-only input mount, the workspace\'s storage and its pre-launch tree check', () => { + const root = mkdtempSync(join(tmpdir(), 'codeboost-launch-')); roots.push(root); + const attemptId = randomUUID(), attemptDir = join(root, OWNER, 'attempts', attemptId); + mkdirSync(attemptDir, { recursive: true, mode: 0o700 }); + let seen: AgentAdapterRequest | undefined, token: string | undefined; + const filesystems = { keeper: 'k' }; + // The schema must stay readable by the container user under any umask. + const umask = process.umask(0o077); + cleanups.push(() => { process.umask(umask); }); + const launch = claudeLauncher({ imageId: 'sha256:x', runnerRoot: root, runnerOwner: OWNER, token: 'secret', + start: (request, t) => { seen = request; token = t; return { attemptId, settled: new Promise(() => undefined), cancel: () => undefined }; } }); + const treeCheck = { checked: true } as never; + launch({ attemptId } as InvocationInput, 'the prompt', { clone: {} as never, storage: { filesystems } }, treeCheck); + expect(seen).toMatchObject({ prompt: 'the prompt', imageId: 'sha256:x', inputDirectory: join(attemptDir, 'input'), filesystems }); + // D refuses an execute start without the pre-launch tree check made for this storage (#81): it is passed on as it is. + expect(seen!.treeCheck).toBe(treeCheck); + expect(token).toBe('secret'); + expect(readdirSync(join(attemptDir, 'input'))).toEqual(['schema.json']); + expect(JSON.parse(readFileSync(join(attemptDir, 'input', 'schema.json'), 'utf8'))).toMatchObject({ type: 'string' }); + expect(lstatSync(join(attemptDir, 'input')).mode & 0o005).toBe(0o005); + expect(lstatSync(join(attemptDir, 'input', 'schema.json')).mode & 0o777).toBe(0o444); + // D's own reader accepts it (a regular, unlinked, bounded file), not only these copied rules. + expect(new TextDecoder().decode(readCapturedFile(join(attemptDir, 'input', 'schema.json'), 'Schema input').content)).toContain('"type":"string"'); + // The coordinator's preparation cleanup removes the attempt directory with it. + rmSync(attemptDir, { recursive: true }); + }); +}); diff --git a/test/runner-recovery.test.ts b/test/runner-recovery.test.ts index a4ceae7b..c5fba3c8 100644 --- a/test/runner-recovery.test.ts +++ b/test/runner-recovery.test.ts @@ -102,7 +102,14 @@ function fixture(count = 1) { const admit = (identity: PlanIdentity, extra: Record = {}) => store.admitAttempt(identity, { expectedStateVersion: store.getTask(identity).stateVersion, kind: 'execute', expectedContext: store.currentContext(identity), deadline: Date.now() + 60_000, ...extra, }); - return { d, path, store, raw, admit }; + /** What F saves at allocation (#91): the allocation ID first, then D's metadata baseline and the seeded commit. */ + const allocate = (identity: PlanIdentity, attemptId: string, handle: unknown, saved: { baseline?: boolean } = {}) => { + const allocationId = randomUUID(); + store.recordAllocation(identity, attemptId, allocationId); + if (saved.baseline !== false) store.recordAllocationBaseline(identity, attemptId, allocationId, 'b'.repeat(64), oid(2)); + return { attemptId, allocationId, handle }; + }; + return { d, path, store, raw, admit, allocate }; } describe('runner owner token', () => { @@ -195,7 +202,7 @@ describe('startup recovery sequence', () => { const calls: string[] = []; const d: RecoveryDeps = { recoverLeftovers: async () => { calls.push('recover'); return { storage: [], unowned: [] }; }, - exportTaskDiff: async () => { calls.push('export'); return Buffer.from('diff'); }, + exportTaskDiff: async () => { calls.push('export'); return { diff: Buffer.from('diff'), truncated: false }; }, removeTaskFilesystems: async handle => { calls.push(`remove:${String(handle)}`); }, processes: { isAlive: () => true, terminate: async pgid => { calls.push(`terminate:${pgid}`); } }, ...over, @@ -203,36 +210,41 @@ describe('startup recovery sequence', () => { return { d, calls }; } it('stops preparation first, then D recovery, export, finalization and removal, in that order', async () => { - const { d: root, store, admit } = fixture(2); - const writable = admit(id(1)); store.markRunning(id(1), writable.id); + const { d: root, store, admit, allocate } = fixture(2); + const writable = admit(id(1)); const w = allocate(id(1), writable.id, 'w'); store.markRunning(id(1), writable.id); const readOnly = admit(id(2), { kind: 'review' }); store.markPreparationStarting(id(2), readOnly.id, Date.now()); store.recordPreparationGroup(id(2), readOnly.id, 4242); + const r = allocate(id(2), readOnly.id, 'r'); + const inputs: unknown[] = []; const { d, calls } = deps({ - recoverLeftovers: async () => { calls.push('recover'); return { storage: [{ attemptId: writable.id, allocationId: randomUUID(), handle: 'w' }, { attemptId: readOnly.id, allocationId: randomUUID(), handle: 'r' }], unowned: [] }; }, - exportTaskDiff: async () => { calls.push(`export:${store.getAttempt(id(1), writable.id).state}`); return Buffer.from('partial'); }, + recoverLeftovers: async () => { calls.push('recover'); return { storage: [w, r], unowned: [] }; }, + exportTaskDiff: async (_h, input) => { inputs.push(input); calls.push(`export:${store.getAttempt(id(1), writable.id).state}`); return { diff: Buffer.from('partial'), truncated: false }; }, }); const report = await recoverStartup({ store, runnerOwner: token, runnerRoot: join(root, 'runner'), diagnosticsDir: join(root, 'diag'), deps: d }); expect(calls).toEqual(['terminate:4242', 'recover', 'export:running', 'remove:w', 'remove:r']); const finalized = store.getAttempt(id(1), writable.id); expect(finalized).toMatchObject({ state: 'failed', diagnosticRef: join(root, 'diag', `${writable.id}.diff`) }); expect(readFileSync(finalized.diagnosticRef!, 'utf8')).toBe('partial'); + // D's export of a recovered handle gets the baseline and seeded commit saved at allocation. + expect(inputs).toEqual([{ base: oid(2), metadataBaseline: 'b'.repeat(64) }]); + expect(statSync(finalized.diagnosticRef!).mode & 0o777).toBe(0o600); expect(report.requeue).toContain(store.getTask(id(1)).planKey); }); it('stops before finalizing anything when D recovery rejects or reports unowned resources', async () => { for (const recoverLeftovers of [async () => { throw new Error('docker down'); }, async () => ({ storage: [], unowned: ['container legacy'] })]) { const { d: root, store, admit } = fixture(); const attempt = admit(id(1)); store.markRunning(id(1), attempt.id); - await expect(recoverStartup({ store, runnerOwner: token, runnerRoot: join(root, 'r'), diagnosticsDir: join(root, 'd'), deps: deps({ recoverLeftovers }).d })).rejects.toThrow(/docker down|older build/); + await expect(recoverStartup({ store, runnerOwner: token, runnerRoot: join(root, 'r'), diagnosticsDir: join(root, 'd'), deps: deps({ recoverLeftovers }).d })).rejects.toThrow(/docker down|cannot identify/); expect(store.getAttempt(id(1), attempt.id).state).toBe('running'); } }); it('records an export timeout as a diagnostic and still finalizes and removes the storage', async () => { - const { d: root, store, admit } = fixture(); - const attempt = admit(id(1)); store.markRunning(id(1), attempt.id); + const { d: root, store, admit, allocate } = fixture(); + const attempt = admit(id(1)); const h = allocate(id(1), attempt.id, 'h'); store.markRunning(id(1), attempt.id); const { d, calls } = deps({ - recoverLeftovers: async () => ({ storage: [{ attemptId: attempt.id, allocationId: randomUUID(), handle: 'h' }], unowned: [] }), + recoverLeftovers: async () => ({ storage: [h], unowned: [] }), // D's export stops a moment after its signal aborts. - exportTaskDiff: (_h, _m, signal) => new Promise((_, reject) => signal.addEventListener('abort', () => setTimeout(() => { calls.push('export-stopped'); reject(signal.reason); }, 40), { once: true })), + exportTaskDiff: (_h, _i, _m, signal) => new Promise((_, reject) => signal.addEventListener('abort', () => setTimeout(() => { calls.push('export-stopped'); reject(signal.reason); }, 40), { once: true })), }); await recoverStartup({ store, runnerOwner: token, runnerRoot: join(root, 'r'), diagnosticsDir: join(root, 'd'), deps: d, exportDeadlineMs: 30 }); expect(store.getAttempt(id(1), attempt.id)).toMatchObject({ state: 'failed', diagnosticRef: null }); @@ -242,10 +254,10 @@ describe('startup recovery sequence', () => { expect(calls.indexOf('remove:h')).toBeGreaterThan(calls.indexOf('export-stopped')); }); it('stops startup, leaving storage and the attempt untouched, when a timed-out export does not stop', async () => { - const { d: root, store, admit } = fixture(); - const attempt = admit(id(1)); store.markRunning(id(1), attempt.id); + const { d: root, store, admit, allocate } = fixture(); + const attempt = admit(id(1)); const h = allocate(id(1), attempt.id, 'h'); store.markRunning(id(1), attempt.id); const { d, calls } = deps({ - recoverLeftovers: async () => ({ storage: [{ attemptId: attempt.id, allocationId: randomUUID(), handle: 'h' }], unowned: [] }), + recoverLeftovers: async () => ({ storage: [h], unowned: [] }), exportTaskDiff: () => new Promise(() => undefined), }); await expect(recoverStartup({ store, runnerOwner: token, runnerRoot: join(root, 'r'), diagnosticsDir: join(root, 'd'), deps: d, exportDeadlineMs: 30, graceMs: 30 })) @@ -254,10 +266,10 @@ describe('startup recovery sequence', () => { expect(store.getAttempt(id(1), attempt.id).state).toBe('running'); }); it('removes nothing when the finalization transaction fails', async () => { - const { d: root, store, admit } = fixture(); - const attempt = admit(id(1)); store.markRunning(id(1), attempt.id); + const { d: root, store, admit, allocate } = fixture(); + const attempt = admit(id(1)); const h = allocate(id(1), attempt.id, 'h'); store.markRunning(id(1), attempt.id); vi.spyOn(store, 'recoverInterrupted').mockImplementation(() => { throw Object.assign(new Error('disk'), { code: 'ERR_SQLITE_ERROR' }); }); - const { d, calls } = deps({ recoverLeftovers: async () => ({ storage: [{ attemptId: attempt.id, allocationId: randomUUID(), handle: 'h' }], unowned: [] }) }); + const { d, calls } = deps({ recoverLeftovers: async () => ({ storage: [h], unowned: [] }) }); await expect(recoverStartup({ store, runnerOwner: token, runnerRoot: join(root, 'r'), diagnosticsDir: join(root, 'd'), deps: d })).rejects.toThrow(/disk/); expect(calls.some(c => c.startsWith('remove'))).toBe(false); }); @@ -274,6 +286,50 @@ describe('startup recovery sequence', () => { expect(report.unknownEntries).toHaveLength(2); expect(existsSync(root)).toBe(true); }); + it('uses a handle only when its allocation ID matches the attempt row, and leaves any other for a person', async () => { + const { d: root, store, admit, allocate } = fixture(); + const attempt = admit(id(1)); const saved = allocate(id(1), attempt.id, 'saved'); store.markRunning(id(1), attempt.id); + const other = { attemptId: attempt.id, allocationId: randomUUID(), handle: 'other' }; + const { d, calls } = deps({ recoverLeftovers: async () => ({ storage: [other, saved], unowned: [] }) }); + const report = await recoverStartup({ store, runnerOwner: token, runnerRoot: join(root, 'r'), diagnosticsDir: join(root, 'd'), deps: d }); + expect(calls).toEqual(['export', 'remove:saved']); + expect(report.unmatchedStorage).toEqual([attempt.id]); + }); + it('keeps every diff this recovery saved when retention runs, so no finalized row points at a deleted file', async () => { + const { d: root, store, admit, allocate } = fixture(2); + const first = admit(id(1)); const a = allocate(id(1), first.id, 'a'); store.markRunning(id(1), first.id); + const second = admit(id(2)); const b = allocate(id(2), second.id, 'b'); store.markRunning(id(2), second.id); + const { d } = deps({ recoverLeftovers: async () => ({ storage: [a, b], unowned: [] }), + exportTaskDiff: async () => ({ diff: Buffer.alloc(40, 1), truncated: false }) }); + // A cap that holds one diff: retention runs on the second save, before finalization references the first. + await recoverStartup({ store, runnerOwner: token, runnerRoot: join(root, 'r'), diagnosticsDir: join(root, 'd'), deps: d, diagnosticsCapBytes: 50 }); + for (const [identity, attempt] of [[id(1), first], [id(2), second]] as const) { + const ref = store.getAttempt(identity, attempt.id).diagnosticRef; + expect(ref && existsSync(ref)).toBe(true); + } + }); + it('marks a diff D cut at its limit, so it is never read as all the agent changed', async () => { + const { d: root, store, admit, allocate } = fixture(); + const attempt = admit(id(1)); const h = allocate(id(1), attempt.id, 'h'); store.markRunning(id(1), attempt.id); + const { d } = deps({ recoverLeftovers: async () => ({ storage: [h], unowned: [] }), + exportTaskDiff: async () => ({ diff: Buffer.from('diff --git a/x b/x'), truncated: true }) }); + await recoverStartup({ store, runnerOwner: token, runnerRoot: join(root, 'r'), diagnosticsDir: join(root, 'd'), deps: d }); + expect(readFileSync(store.getAttempt(id(1), attempt.id).diagnosticRef!, 'utf8')).toMatch(/^diff --git a\/x b\/x\ncodeboost: the partial output was cut at 18 bytes; later changes are not shown\.\n$/); + }); + it('lists each unowned object on its own line', async () => { + const { d: root, store } = fixture(); + const run = recoverStartup({ store, runnerOwner: token, runnerRoot: join(root, 'r'), diagnosticsDir: join(root, 'd'), + deps: deps({ recoverLeftovers: async () => ({ storage: [], unowned: ['docker volume rm a # no-runner-label', 'docker network rm b # unknown-kind'] }) }).d }); + await expect(run).rejects.toThrow(/then start again:\ndocker volume rm a {2}# no-runner-label\ndocker network rm b {2}# unknown-kind$/); + }); + it('records an export failure, without calling D, when the allocation baseline was never saved', async () => { + const { d: root, store, admit, allocate } = fixture(); + const attempt = admit(id(1)); const h = allocate(id(1), attempt.id, 'h', { baseline: false }); store.markRunning(id(1), attempt.id); + const { d, calls } = deps({ recoverLeftovers: async () => ({ storage: [h], unowned: [] }) }); + await recoverStartup({ store, runnerOwner: token, runnerRoot: join(root, 'r'), diagnosticsDir: join(root, 'd'), deps: d }); + expect(calls).toEqual(['remove:h']); + expect(store.getAttempt(id(1), attempt.id).diagnostic).toMatch(/Partial output could not be exported: its storage baseline was never saved/); + }); it('fails closed on an unrecorded preparation until it is released, and refuses release while a process uses it', async () => { const { d: root, store, admit } = fixture(); const attempt = admit(id(1)); store.markPreparationStarting(id(1), attempt.id, Date.now()); diff --git a/test/runner-shutdown.test.ts b/test/runner-shutdown.test.ts index bf3986bc..25765855 100644 --- a/test/runner-shutdown.test.ts +++ b/test/runner-shutdown.test.ts @@ -5,7 +5,7 @@ import { request as httpRequest } from 'node:http'; import { randomUUID } from 'node:crypto'; import { afterEach, describe, expect, it, vi } from 'vitest'; import { createDemo } from '../scripts/demo.ts'; -import { startServer } from '../web/server.ts'; +import { RUNNER_NOT_IN_DEMO, startServer } from '../web/server.ts'; import { Store } from '../runner/store.ts'; import { ReviewService } from '../runner/review.ts'; import { MergeCoordinator } from '../runner/merge.ts'; @@ -220,7 +220,7 @@ describe('/api/runner', () => { const view = (await api(app, 'GET', '/api/runner')).body; expect(view).toMatchObject({ available: false, retryable: false, stopRequested: null, unresolved: null, task: { status: 'in review' } }); const refused = await api(app, 'POST', '/api/runner', { action: 'retry', attemptId: randomUUID(), expectedStateVersion: view.stateVersion, actionId: randomUUID() }); - expect(refused).toMatchObject({ status: 409, body: { error: 'The runner is not available yet.' } }); + expect(refused).toMatchObject({ status: 409, body: { error: RUNNER_NOT_IN_DEMO } }); }); it('replays cancel task by action ID, refuses a reused ID, and answers 400 for a malformed ID', async () => { const { app } = await serve(); diff --git a/test/runner-workspace-baseline.test.ts b/test/runner-workspace-baseline.test.ts new file mode 100644 index 00000000..e32e75e7 --- /dev/null +++ b/test/runner-workspace-baseline.test.ts @@ -0,0 +1,40 @@ +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { afterEach, expect, it, vi } from 'vitest'; +import type { Plan, PlanContext } from '../core/plan.ts'; + +// D's preparation calls succeed without Docker or Git; what is under test is what materialize saves around them. +vi.mock('../runner/runner-repository.ts', async original => ({ ...await original(), ensureCommit: async () => undefined })); +vi.mock('../git/clone.ts', async original => ({ ...await original(), + createTaskCloneAsync: async (o: { parent: string; head: string }) => ({ id: 'clone', taskId: 'task', directory: o.parent, head: o.head }) })); +vi.mock('../agents/container/storage.ts', async original => ({ ...await original(), + prepareTaskFilesystemsAsync: async () => ({ keeper: 'k', workVolume: 'w', metadataVolume: 'm', workBytes: 1, workInodes: 1, metadataBytes: 1, metadataInodes: 1, metadataBaseline: 'e'.repeat(64) }) })); +const { Store } = await import('../runner/store.ts'); +const { createTaskWorkspace } = await import('../runner/workspace.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: Plan = { schema_version: 1, revision: 1, issue: 1, summary: 'S', questions: [], items: [{ id: 'P1', title: 'T', intent: 'I', + files: [{ path: 'a', kind: 'edit', renamed_from: null, change: 'x' }], acceptance: [{ type: 'check', text: 'ok' }], depends_on: [] }] }; +const oid = (n: number) => n.toString(16).padStart(40, '0'); +const dirs: string[] = []; +afterEach(() => { for (const dir of dirs.splice(0)) rmSync(dir, { recursive: true, force: true }); }); + +it('saves the storage baseline and the head it was seeded from as soon as D\'s allocation returns (#91)', async () => { + const dir = mkdtempSync(join(tmpdir(), 'workspace-baseline-')); dirs.push(dir); + const store = new Store(join(dir, 'state.sqlite')); + try { + store.createPlan(JSON.stringify(plan), 'json', context, oid(1), oid(2)); + store.transitionTask(identity, store.getTask(identity).stateVersion, 'queued'); + const attempt = store.admitAttempt(identity, { expectedStateVersion: store.getTask(identity).stateVersion, kind: 'execute', item: 'P1', + expectedContext: store.currentContext(identity), deadline: Date.now() + 60_000 }); + const workspace = createTaskWorkspace({ store, runnerRoot: join(dir, 'runner'), runnerOwner: '0123456789abcdef0123456789abcdef', + repository: { path: join(dir, 'repo.git'), source: join(dir, 'source') }, imageId: `sha256:${'a'.repeat(64)}`, + limits: { workBytes: 1, workInodes: 1, metadataBytes: 1, metadataInodes: 1 }, committer: { name: 'c', email: 'c@e' } }); + await workspace.materialize(attempt, oid(7), new AbortController().signal); + const row = store.interruptedAttempts()[0]!; + expect(row).toMatchObject({ metadataBaseline: 'e'.repeat(64), storageBase: oid(7) }); + expect(row.allocationId).toBe(store.attemptAllocation(attempt.id)); + } finally { store.close(); } +}); diff --git a/web/cli.ts b/web/cli.ts index 1400c2d9..050d9615 100644 --- a/web/cli.ts +++ b/web/cli.ts @@ -1,28 +1,79 @@ import { readFileSync } from 'node:fs'; import { resolve } from 'node:path'; import { parseArgs } from 'node:util'; -import { startServer } from './server.ts'; +import { startServer, type RunnerSetup } from './server.ts'; import { createDemo } from '../scripts/demo.ts'; -import { requireSupportedNode } from '../runner/store.ts'; -import { acquireRunnerLock } from '../runner/recovery.ts'; +import { Store, requireSupportedNode } from '../runner/store.ts'; +import { acquireRunnerLock, releasePreparation } from '../runner/recovery.ts'; +import { parseRunnerConfig, setUpRunner } from '../runner/production.ts'; requireSupportedNode(); -const { values } = parseArgs({ options: { demo: { type:'boolean' }, directory:{type:'string'}, config:{type:'string'}, port:{type:'string'}, help:{type:'boolean'} } }); -if (values.help || (!values.demo && !values.config)) { - console.log('codeboost local review\n\nDemo: npm run demo\nExisting store: npm start -- --config /absolute/path/review.json\nOptions: --port 4318 --directory /path/to/demo\n\nThe configuration binds a trusted repository, database, plan identity, and known path identity. Configure the question agent in Settings; Ask runs it in a Docker container (Claude Code needs CLAUDE_CODE_OAUTH_TOKEN; Codex cannot answer questions yet). A github block enables the guarded merge gate; demos never merge.'); +/** 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); } +const checked = (fn: () => T): T => { try { return fn(); } catch (error) { return refuse(error); } }; +const { values } = parseArgs({ options: { demo: { type:'boolean' }, directory:{type:'string'}, config:{type:'string'}, port:{type:'string'}, help:{type:'boolean'}, 'release-preparation':{type:'string'} } }); +if (values.help || (!values.demo && !values.config && values['release-preparation'] === undefined)) { + console.log('codeboost local review\n\nDemo: npm run demo\nExisting store: npm start -- --config /absolute/path/review.json\nOptions: --port 4318 --directory /path/to/demo\n\nThe configuration binds a trusted repository, database, plan identity, and known path identity. Configure the question agent in Settings; Ask runs it in a Docker container (Claude Code needs CLAUDE_CODE_OAUTH_TOKEN; Codex cannot answer questions yet). A github block enables the guarded merge gate; demos never merge. A runner block (root, committer, optional diagnosticsDir and limits) turns on the runner: it needs Docker, the github block and CLAUDE_CODE_OAUTH_TOKEN, and runs startup recovery before the server opens.\n\n--release-preparation : after startup recovery reports a preparation whose process was never recorded, and you have stopped that process, remove its attempt directory (refused while any process still uses it).'); +} else if (values['release-preparation'] !== undefined) { + const attemptId = values['release-preparation']; + checked(() => { + if (!values.config) throw new Error('--release-preparation needs --config.'); + const config = JSON.parse(readFileSync(resolve(values.config), 'utf8')); + if (config.runner === undefined) throw new Error('This review has no runner block, so it has no preparations to release.'); + const runnerConfig = parseRunnerConfig(config.runner); + // Under the runner lock, so no runner of this database is starting or running while the directory is checked. + const lock = acquireRunnerLock(config.database); + try { + const store = new Store(config.database); + try { + lock.verify(); + releasePreparation({ store, runnerRoot: runnerConfig.root, runnerOwner: store.runnerOwnerToken(lock.file), attemptId }); + console.log('Released. Start codeboost again.'); + } finally { store.close(); } + } finally { lock.release(); } + }); } else { const port = Number(values.port ?? '4318'); - if (!Number.isInteger(port) || port < 0 || port > 65535) throw new Error('Invalid port.'); - const config = values.demo ? createDemo(values.directory ?? '.codeboost-local/demo') : JSON.parse(readFileSync(resolve(values.config!), 'utf8')); + if (!Number.isInteger(port) || port < 0 || port > 65535) refuse(new Error('Invalid port.')); + const config = checked(() => values.demo ? createDemo(values.directory ?? '.codeboost-local/demo') : JSON.parse(readFileSync(resolve(values.config!), 'utf8'))); + // Checked before the lock, so a malformed block changes nothing. Demos never run the runner, however they were opened. + const runnerConfig = checked(() => config.demo !== true && config.runner !== undefined ? parseRunnerConfig(config.runner) : null); // Decision 1: one runner per database, held as an OS lock keyed by the database file's device and inode. - let lock: ReturnType; - try { lock = acquireRunnerLock(config.database); } - catch (error) { console.error(error instanceof Error ? error.message : String(error)); process.exit(1); } + const lock = checked(() => acquireRunnerLock(config.database)); + const runnerSetup: RunnerSetup | undefined = runnerConfig ? async (service, capability) => { + const assembly = await setUpRunner({ service, capability, config: runnerConfig, lock, env: process.env, + onSlowStart: () => console.log('Building the agent image and recovering what an earlier run left. This can take a few minutes the first time.') }); + const { recovery } = assembly; + 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}`); + 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 + // once: that is a crash to recovery, which the next start runs again. A terminal Ctrl+C also reaches an agent image + // build in progress (it shares the terminal), which then fails and ends startup. + let stopping = false; + const duringStartup = () => { + if (stopping) { console.error('Stopped during startup. The next start recovers what this one left.'); process.exit(130); } + stopping = true; console.log('Stopping once startup has finished. Press Ctrl+C again to stop now.'); + }; + for (const signal of ['SIGINT', 'SIGTERM'] as const) process.on(signal, duringStartup); let app: Awaited>; - try { app = await startServer(config, port); } - catch (error) { lock.release(); throw error; } + try { app = await startServer(config, port, undefined, undefined, undefined, undefined, undefined, undefined, runnerSetup); } + catch (error) { + lock.release(); + // A refused runner startup (no token, a blocked recovery) is the person's to act on: its message, not a stack. + if (!runnerSetup) throw error; + console.error(error instanceof Error ? error.message : String(error)); process.exit(1); + } + for (const signal of ['SIGINT', 'SIGTERM'] as const) process.removeListener(signal, duringStartup); try { lock.verify(); } catch (error) { await app.close(); lock.release(); throw error; } - console.log(`Review ready: ${app.url}\nRepository: ${config.repository}\nDatabase: ${config.database}\nSource files are read-only. Press Ctrl+C to stop.`); - let stopping=false; - for(const signal of ['SIGINT','SIGTERM'] as const) process.on(signal,()=>{if(!stopping){stopping=true;void app.close().then(()=>{lock.release();process.exit(0);},error=>{lock.release();console.error(error instanceof Error?error.message:error);process.exit(1);});}}); + 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. + for (const signal of ['SIGINT', 'SIGTERM'] as const) process.on(signal, () => { if (stopping) { console.log('Still stopping agents…'); return; } stopping = true; stop(); }); + if (stopping) stop(); + else console.log(`Review ready: ${app.url}\nRepository: ${config.repository}\nDatabase: ${config.database}\nSource files are read-only. Press Ctrl+C to stop.`); } diff --git a/web/server.ts b/web/server.ts index feedccd9..52157bf3 100644 --- a/web/server.ts +++ b/web/server.ts @@ -7,6 +7,9 @@ import { Questions, type QuestionAgent } from '../runner/questions.ts'; import { GhMergeGateway, type MergeGateway } from '../github/merge.ts'; import { MERGE_OPERATION_TIMEOUT_MS, MergeCoordinator, MergeNotApplied, MergeOutcomeUnknown } from '../runner/merge.ts'; import { RunnerCoordinator, type RunnerDeps } from '../runner/coordinator.ts'; +import { ItemExecutor } from '../runner/execution.ts'; +import type { RunnerAssembly } from '../runner/production.ts'; +import type { ShutdownCapability } from '../runner/lifecycle.ts'; import { BadRequest, GuardRefusal, ShuttingDownError, assertUuidV4, isUuidV4, sameContext } from '../runner/lifecycle.ts'; import { GhIssueGateway, type IssueGateway } from '../github/issues.ts'; import { demoIssueGateway } from '../scripts/demo-issues.ts'; @@ -22,10 +25,20 @@ export interface PlanningDeps { const publicRoot = new URL('./public/', import.meta.url); /** The longest shutdown waits for admitted requests to finish before aborting them; below the 15 s request timeout. */ export const MAX_SHUTDOWN_DRAIN_MS = 14_500; -export async function startServer(config: ReviewConfig, port = 4318, questionAgent?: QuestionAgent, mergeGateway?: MergeGateway, shutdownDrainMs = MAX_SHUTDOWN_DRAIN_MS, issueGateway?: IssueGateway, runnerDeps?: RunnerDeps, planning?: PlanningDeps) { +/** + * The production runner's startup (#91, `setUpRunner`): run after the Store opens and before the server listens, so + * startup recovery finishes before anything is admitted. A rejection stops startup. + */ +export type RunnerSetup = (service: ReviewService, capability: ShutdownCapability) => Promise; +export const RUNNER_NOT_CONFIGURED = 'The runner is not configured. Add a runner block to the review configuration and restart codeboost.'; +/** Demos never run the runner, whatever their configuration says. */ +export const RUNNER_NOT_IN_DEMO = 'Demos do not run the runner. Use a review configuration with a runner block.'; +export async function startServer(config: ReviewConfig, port = 4318, questionAgent?: QuestionAgent, mergeGateway?: MergeGateway, shutdownDrainMs = MAX_SHUTDOWN_DRAIN_MS, issueGateway?: IssueGateway, runnerDeps?: RunnerDeps, planning?: PlanningDeps, runnerSetup?: RunnerSetup) { if (!Number.isSafeInteger(shutdownDrainMs) || shutdownDrainMs < 1 || shutdownDrainMs > MAX_SHUTDOWN_DRAIN_MS) throw new Error('Invalid shutdown drain deadline.'); const service = new ReviewService(config), token = randomBytes(32).toString('hex'); let questions: Questions, merges: MergeCoordinator | null, issues: IssueBoard, runner: RunnerCoordinator | null, suggestions: SuggestionCoordinator | null; + /** Runs a task's plan items; one per Store, like the coordinator. Only the production runner has one. */ + let executor: ItemExecutor | null = null; // Only coordinators' settlement and close code receive this; HTTP handlers never do. const capability = service.store.shutdownCapability(); try { @@ -35,7 +48,8 @@ export async function startServer(config: ReviewConfig, port = 4318, questionAge issues = new IssueBoard(issueGateway ?? (config.demo ? demoIssueGateway() : config.github ? new GhIssueGateway(config.github.repository) : null), 'Issue ranking needs a GitHub repository. Add a github block with a repository to the review configuration.'); merges = !config.demo && (mergeGateway || config.github) ? new MergeCoordinator(service, mergeGateway ?? new GhMergeGateway(config.github!), MERGE_OPERATION_TIMEOUT_MS, capability) : null; - // The runner starts only with an injected D; until #51 lands, runner actions report that it is unavailable. + if (runnerDeps && runnerSetup) throw new Error('Pass runner deps or a runner setup, not both.'); + // Without a runner (no runner block, or a demo), runner actions report that it is not configured. runner = runnerDeps ? new RunnerCoordinator(service.store, runnerDeps, undefined, capability) : null; // E3's settlement writes (completeSuggestions, settleSuggestion in close()) run with the shutdown capability. const store = service.store; @@ -48,6 +62,14 @@ export async function startServer(config: ReviewConfig, port = 4318, questionAge }; suggestions = planning ? new SuggestionCoordinator(suggestionStore, planning.provider) : null; } catch (error) { service.close(); throw error; } + if (runnerSetup) { + // Startup recovery runs here, before listen: nothing is admitted until it has finished. + try { + const assembly = await runnerSetup(service, capability); + runner = new RunnerCoordinator(service.store, assembly.deps, undefined, capability); + executor = new ItemExecutor(service.store, runner, assembly.sources, assembly.findings, { capability }); + } catch (error) { service.close(); throw error; } + } const loadReview=()=>{const view=service.load();return {...view,notes:view.notes.map(note=>({...note,answerActive:questions.isRunning(note.id)}))};}; const load=async(signal?:AbortSignal)=>{const view=loadReview();return {...view,merge:merges?await merges.displayStatus(view,signal):{available:false}};}; const answerStatuses=()=>service.store.getReviewNotes(config.identity) @@ -61,7 +83,10 @@ export async function startServer(config: ReviewConfig, port = 4318, questionAge const last = attempts.find(attempt => attempt.id === task.currentAttemptId); const retryable = !!runner && !!last && (last.state === 'failed' || last.state === 'cancelled') && (task.status === 'running' || task.status === 'queued') && !task.requeuePending && task.cancelRequested === null - && !status.active && !status.unresolved && sameContext(last.context, service.store.currentContext(identity)); + && !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. + && !(executor && last.kind === 'execute'); return { available: !!runner, task, attempts, stateVersion: task.stateVersion, retryable, stopRequested: status.stopRequested, unresolved: status.unresolved }; }; @@ -74,12 +99,14 @@ export async function startServer(config: ReviewConfig, port = 4318, questionAge 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('The runner is not available yet.'); + if (!runner) throw new GuardRefusal(config.demo ? RUNNER_NOT_IN_DEMO : RUNNER_NOT_CONFIGURED); 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).'); 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 }; @@ -237,7 +264,7 @@ export async function startServer(config: ReviewConfig, port = 4318, questionAge server.requestTimeout = 15000; await new Promise((resolve, reject) => { server.once('error', reject); server.listen(port, '127.0.0.1', () => { server.removeListener('error', reject); resolve(); }); }).catch(error => { service.close(); throw error; }); const address = server.address(); if (!address || typeof address === 'string') throw new Error('Cannot determine local address.'); - return { server, service, token, runner, url: `http://127.0.0.1:${address.port}/#${token}`, close: async () => { + return { server, service, token, runner, executor, url: `http://127.0.0.1:${address.port}/#${token}`, close: async () => { // Step 1, one synchronous turn: reject new API requests and new runner work. Admitted requests drain (step 2). stopping = true; // Same turn as the admission flag: a request already reading its body must not start a new Ask worker. @@ -259,11 +286,22 @@ export async function startServer(config: ReviewConfig, port = 4318, questionAge server.closeIdleConnections(); // Abort the issue refresh now; its close settles without rejecting, so it is always awaited. const issuesClosed=issues.close(); - try { await merges?.close(); } finally { await issuesClosed; } + // Every step runs even when an earlier one fails: agents are still stopped, plan runs awaited and the Store closed + // last. The first failure is reported; a later one never hides it. + const failures: unknown[] = []; + const step = async (run: () => Promise | unknown) => { try { await run(); } catch (error) { failures.push(error); } }; + await step(() => merges?.close()); + await step(() => issuesClosed); // Step 4: stop runner jobs (shutdown reason only where none is set) and await settlement; no timer abandons a job. - await runner?.close(); - await closing; - // Close the store even if Ask's or planning's cleanup fails, then report that failure. - try { await questions.close(); } finally { try { await suggestions?.close(); } finally { service.close(); } } + await step(() => runner?.close()); + // Then every plan run in progress: its pause or escalation after the last attempt is a settlement write (#91). + await step(() => executor?.close()); + await step(() => closing); + await step(() => questions.close()); + await step(() => suggestions?.close()); + await step(() => service.close()); + // Each later failure is still reported, so none is lost behind the first. + for (const later of failures.slice(1)) console.error(`Shutdown step also failed: ${later instanceof Error ? later.message : String(later)}`); + if (failures.length) throw failures[0]; } }; }