diff --git a/.github/workflows/agent-isolation.yml b/.github/workflows/agent-isolation.yml index 018b55ce..dbd22203 100644 --- a/.github/workflows/agent-isolation.yml +++ b/.github/workflows/agent-isolation.yml @@ -8,6 +8,7 @@ on: - 'test/agent-*.test.ts' - 'runner/question-*.ts' - 'runner/questions.ts' + - 'runner/planning*.ts' - 'runner/workspace.ts' - 'runner/runner-repository.ts' - 'runner/execution.ts' @@ -24,6 +25,7 @@ on: - 'test/agent-*.test.ts' - 'runner/question-*.ts' - 'runner/questions.ts' + - 'runner/planning*.ts' - 'runner/workspace.ts' - 'runner/runner-repository.ts' - 'runner/execution.ts' @@ -49,4 +51,4 @@ jobs: - run: npm ci --ignore-scripts - run: npm run typecheck # The Docker suites share one image tag and daemon, so run test files one at a time. - - run: npx vitest run --no-file-parallelism test/agent-contract.test.ts test/agent-clone.test.ts test/agent-container.test.ts test/agent-network.test.ts test/agent-policy.test.ts test/agent-proxy.test.ts test/agent-adapter.test.ts test/agent-supervisor.test.ts test/agent-output.test.ts test/agent-gate.test.ts test/agent-question.test.ts test/runner-workspace.test.ts + - run: npx vitest run --no-file-parallelism test/agent-contract.test.ts test/agent-clone.test.ts test/agent-container.test.ts test/agent-network.test.ts test/agent-policy.test.ts test/agent-proxy.test.ts test/agent-adapter.test.ts test/agent-supervisor.test.ts test/agent-output.test.ts test/agent-gate.test.ts test/agent-question.test.ts test/agent-planning.test.ts test/runner-workspace.test.ts diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d991ab1f..a7709445 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -18,6 +18,6 @@ jobs: - run: npm run typecheck # The Docker agent suites run one file at a time in the Agent isolation workflow; running them here # would put them in parallel against the same image tag and daemon. - - run: npm test -- --exclude test/agent-container.test.ts --exclude test/agent-network.test.ts --exclude test/agent-adapter.test.ts --exclude test/agent-supervisor.test.ts --exclude test/agent-gate.test.ts --exclude test/agent-question.test.ts --exclude test/runner-workspace.test.ts + - run: npm test -- --exclude test/agent-container.test.ts --exclude test/agent-network.test.ts --exclude test/agent-adapter.test.ts --exclude test/agent-supervisor.test.ts --exclude test/agent-gate.test.ts --exclude test/agent-question.test.ts --exclude test/agent-planning.test.ts --exclude test/runner-workspace.test.ts - run: npx playwright install --with-deps chromium - run: npm run test:browser diff --git a/docs/architecture.md b/docs/architecture.md index 1f6fb7ab..523a56a0 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -160,7 +160,7 @@ Dependencies point downward only. `core/` imports nothing that does input or out | Part | Main files | Owns | State | |---|---|---|---| | Command line | `web/cli.ts` | Parses `--demo` or `--config`, takes the single-runner lock, starts the server, releases the lock on Ctrl+C. | In use | -| HTTP server | `web/server.ts` | Binds `127.0.0.1` only. Checks the token, Host and Origin. Bounded JSON bodies. Routes `/api/review`, `/api/action`, `/api/questions`, `/api/merge`, `/api/issues`, `/api/settings`, plus `/api/runner` and `/api/plan/*` when a runner or planning provider is supplied. Ordered shutdown. | In use; `/api/runner` and `/api/plan/*` not connected | +| HTTP server | `web/server.ts` | Binds `127.0.0.1` only. Checks the token, Host and Origin. Bounded JSON bodies. Routes `/api/review`, `/api/action`, `/api/questions`, `/api/merge`, `/api/issues`, `/api/settings`, plus `/api/runner` and `/api/plan/*` when a runner or planning provider is supplied. Ordered shutdown. | In use; `/api/plan/suggestions` runs Claude through lane D for a review with a `github` block (#117) | | Browser screen | `web/public/*` | Review screen and Issues screen. Plain JavaScript with IBM Plex fonts served locally. Sends item and segment IDs and a state token, never fingerprints or ownership. | Yes | | Review service | `runner/review.ts` | Builds the review view: reads history, links it to the plan, computes approval states and merge blockers. Applies review commands through the store. | Yes | | Ask | `runner/questions.ts`, `runner/question-*.ts` | Answers a question about one plan item with a read-only agent. Runs lane D's synchronous setup in a worker thread so the server stays responsive. Tracks leftover containers. | Yes | diff --git a/docs/implementation/planning-acceptance.md b/docs/implementation/planning-acceptance.md index aca2f0fd..21dbc8dc 100644 --- a/docs/implementation/planning-acceptance.md +++ b/docs/implementation/planning-acceptance.md @@ -72,14 +72,15 @@ The provider and Ask share one container runner, `runReadOnlyAgent` in limits, credential rule, output check and wording as a `ReadOnlyFeature` (`ASK_FEATURE`, `PLANNING_FEATURE`). Ask's wording and cleanup error are unchanged (#117 part 1). -Left for #117 part 2: the provider's image build, clone and storage calls are -synchronous, so a server must run it off its request thread, as Ask does. That -worker should also set TMPDIR to a planning root, as Ask's does, so the input and -auth staging that lane D leaves when it gives up on cleanup is covered. E3's -default 120-second suggestion timeout is shorter than the provider's ten-minute -budget. Startup recovery does not yet remove planning's Docker resources after a -crash, because they are not attempt allocations in the Store. The recording -script prints its runner label if it leaves any. +#117 part 2 wires the provider into the server. Each request runs in planning's +own worker thread (`runner/planning.ts`, `AgentWorker` with `PLANNING_NAMING`), +whose TMPDIR is a planning root recorded in a ledger beside the review database, +so lane D's leftover input and auth staging is covered. Its Docker objects carry +the database's planning owner, apart from Ask's and the runner's, and the first +request of a process removes what earlier sessions left by that owner only. E3's +suggestion timer and the request share one ten-minute budget. The recording +script still runs the provider in its own process and prints its runner label if +it leaves anything. F/G still own live API and UI integration. The Store now guarantees revision/snapshot binding and pending-only settlement across processes, and E3 requires the caller's diff --git a/docs/implementation/runner-lifecycle.md b/docs/implementation/runner-lifecycle.md index 9813e299..ec673412 100644 --- a/docs/implementation/runner-lifecycle.md +++ b/docs/implementation/runner-lifecycle.md @@ -354,10 +354,10 @@ F1 owns the production planning endpoints that G4 needs. They wrap E3's coordina | `POST /api/plan/import` | `Store.importRevision` | expected revision | | `POST /api/plan/suggestions` | E3 coordinator start (`beginSuggestions`) | expected revision and snapshot | | `GET /api/plan/suggestions/:id` | `Store.getSuggestions` | reads only the request row | -| `POST /api/plan/suggestions/:id/cancel` | `pending`: `handle.cancel(reason)` on the `SuggestionHandle` that `SuggestionCoordinator.start()` returned. E3's coordinator has no cancel-by-ID method, so F keeps a map from request ID to handle, adds each handle when it starts the request, and removes it only after the handle's outcome settles. If the request is `pending` but F holds no handle (for example after a restart), F calls `Store.cancelSuggestions`. It is safe to do so because startup recovery has already stopped the crashed process's provider. A live planning provider runs only through D (G4 after D5), in a container labelled with this database's runner owner, and D's `recoverLeftovers` (startup step 2) stops it before admission opens. Until D5 there is no live planning provider at all ("Planning agent not available yet"). If a planning provider ever runs outside D, this endpoint must refuse such a request instead, until startup has reconciled it. `ready`: `Store.cancelSuggestions` (dismiss), because no invocation is running. | request is `pending` or `ready`, as `Store.cancelSuggestions` already allows | +| `POST /api/plan/suggestions/:id/cancel` | `pending`: `handle.cancel(reason)` on the `SuggestionHandle` that `SuggestionCoordinator.start()` returned. E3's coordinator has no cancel-by-ID method, so F keeps a map from request ID to handle, adds each handle when it starts the request, and removes it only after the handle's outcome settles. If the request is `pending` but F holds no handle (for example after a restart), F calls `Store.cancelSuggestions`. It is safe to do so because a crashed process's planning container has no route back to the Store: its output went to that process's worker, which ended with it. A live planning provider runs only through D (#117), in planning's own worker, in containers labelled with this database's planning owner (not the runner's or Ask's). Like Ask's, those containers are removed by D's `recoverLeftovers` for that owner before the first planning request of the next process, not at startup; until then a stray container can keep running read-only (its deadline is enforced from the host, which is gone) until Claude exits or that recovery removes it, and its output is discarded. If a planning provider ever runs outside D, or could report into the Store from another process, this endpoint must refuse such a request instead, until it has been reconciled. `ready`: `Store.cancelSuggestions` (dismiss), because no invocation is running. | request is `pending` or `ready`, as `Store.cancelSuggestions` already allows | | `POST /api/plan/suggestions/:id/apply` | `Store.applySuggestion` | request `ready` and bound to the current revision and snapshot | -Live planning invocation waits for D5. Until then, the server returns "Planning agent not available yet" instead of using a fake provider. +Live planning runs for a non-demo review with a `github` block (#117): `web/planning.ts` reads the issue from GitHub before each start, and `runner/planning.ts` runs the request in planning's worker. E3's suggestion timer and the request share one ten-minute budget (`PLANNING_BUDGET_MS`). At shutdown E3 aborts every request; one that has not settled within 20 seconds (`PLANNING_SHUTDOWN_GRACE_MS`, as Ask) is ended by abandoning planning's worker. A start replays its recorded outcome before reading GitHub; a GitHub failure is a 502 and is not recorded. Without a `github` block, or in a demo, the server returns "Planning agent not available yet". ## Feedback-event contract (for lane J) diff --git a/runner/planning-provider.ts b/runner/planning-provider.ts index d1267e52..8f9192bf 100644 --- a/runner/planning-provider.ts +++ b/runner/planning-provider.ts @@ -5,7 +5,7 @@ import type { InvocationInput, InvocationResult, StopReason } from '../agents/co import type { TaskStorageLimits } from '../agents/container/storage.ts'; import type { AuthorProvider, AuthorRequest } from '../core/planning-author.ts'; import { allocations, exceedsStorage, RetainedStorage, runReadOnlyAgent, sameContext, type ContainerDependencies, type LeftoverPolicy, type Provider, - type ReadOnlyFeature, type RepositorySize } from './question-container.ts'; + type ReadOnlyFeature, type ReadOnlyRun, type RepositorySize } from './question-container.ts'; /** Ask's lane D dependencies, less startup recovery, which the runner owns for planning. */ export type PlanningDependencies = Omit; @@ -63,7 +63,12 @@ export function removePlanningRoot(root: string): void { export const PLANNING_STORAGE: TaskStorageLimits = Object.freeze({ workBytes: 512 * 1024 * 1024, workInodes: 131_072, metadataBytes: 512 * 1024 * 1024, metadataInodes: 131_072, }); -const DEFAULT_TIMEOUT_MS = 10 * 60_000; +/** + * The whole budget of one planning request, setup included (#117). Lane D caps the invocation itself at ten minutes, + * and E3's suggestion timer uses this same budget, so the request settles inside it. + */ +export const PLANNING_BUDGET_MS = 10 * 60_000; +const DEFAULT_TIMEOUT_MS = PLANNING_BUDGET_MS; /** Codex reads files only through its shell, and planning runs no process, so Codex cannot plan (#75, #93). */ export const CODEX_PLANNING_REFUSED = 'Codex cannot write plans yet: it can read the code only by running commands, ' @@ -110,6 +115,22 @@ export const PLANNING_FEATURE: ReadOnlyFeature = Object.freeze({ : new AggregateError(failures, 'Planning container cleanup did not settle.'), }); +/** Where a planning request reads the code: the review's repository at the snapshot's head. */ +export interface PlanningSource { + readonly vendor: Provider; + readonly repository: string; + readonly head: string; + readonly snapshotId: string; +} +/** One planning request as a read-only run, without its owner, which the process running it supplies. */ +export function planningRun(request: AuthorRequest, source: PlanningSource, deadline: number): Omit { + if (request.phase !== 'planning' || request.access !== 'read-only') throw new Error('Planning requests must be read-only.'); + return { provider: source.vendor, repository: source.repository, head: source.head, attemptId: request.requestId, + taskId: `planning-${request.requestId}`, deadline, prompt: request.prompt, schemaText: request.schemaText, + context: { snapshotId: source.snapshotId, planId: request.identity.planId, planRevision: request.revision, + assignmentId: `${request.mode}-${request.issue}`, referencedCodeHash: source.head, stateVersion: 0 } }; +} + /** * E2's provider over lane D: each request runs in a fresh container on a read-only copy of `head`, in the "planning" * phase (read, list and search only; no commands) with vendor-only egress. The request's schema is the only file in @@ -129,11 +150,8 @@ export function createPlanningProvider(options: PlanningProviderOptions): Planni return { async invoke(request: AuthorRequest, signal: AbortSignal): Promise { if (request.phase !== 'planning' || request.access !== 'read-only') throw new Error('Planning requests must be read-only.'); signal.throwIfAborted(); - return runReadOnlyAgent(PLANNING_FEATURE, { provider: vendor, repository: options.repository, head: options.head, - runnerOwner: options.runnerOwner, attemptId: request.requestId, taskId: `planning-${request.requestId}`, - deadline: Date.now() + timeoutMs, prompt: request.prompt, schemaText: request.schemaText, - context: { snapshotId: options.snapshotId, planId: request.identity.planId, planRevision: request.revision, - assignmentId: `${request.mode}-${request.issue}`, referencedCodeHash: options.head, stateVersion: 0 } }, + return runReadOnlyAgent(PLANNING_FEATURE, { ...planningRun(request, { vendor, repository: options.repository, + head: options.head, snapshotId: options.snapshotId }, Date.now() + timeoutMs), runnerOwner: options.runnerOwner }, deps, signal, image, retained); } }; } diff --git a/runner/planning.ts b/runner/planning.ts new file mode 100644 index 00000000..d6fe00af --- /dev/null +++ b/runner/planning.ts @@ -0,0 +1,44 @@ +import type { AuthorProvider, AuthorRequest } from '../core/planning-author.ts'; +import { AgentWorker } from './question-agent.ts'; +import { LeftoverLedger, PLANNING_NAMING } from './question-leftovers.ts'; +import { PLANNING_BUDGET_MS, planningRun } from './planning-provider.ts'; +import type { ReviewService } from './review.ts'; + +// Leave the worker time to cancel the container and release storage before E3's own timer fires (as Ask does). +const SETTLE_MARGIN_MS = 5_000; +export const NO_PLANNING_AGENT = 'Choose an agent in Settings, then retry.'; + +/** + * The production planning provider (#117): every request runs in planning's own worker thread, so lane D's synchronous + * setup never blocks the server. The worker's TMPDIR is a planning root recorded in a ledger beside the review + * database, and every Docker object it creates carries the database's planning owner. The first request of a process + * removes what earlier sessions left, by that owner only, so Ask's and the runner's agents are never touched. + * + * The code is the review's repository at the current snapshot's head; E3 discards a reply whose snapshot or revision + * changed meanwhile. The vendor is the one chosen in Settings for Ask; Codex is refused (#93). + */ +export class PlanningAgent implements AuthorProvider { + readonly #service: ReviewService; + readonly #worker: AgentWorker; + constructor(service: ReviewService, options: { url?: URL; env?: Readonly>; + ledger?: LeftoverLedger; abandonAfterDeadlineMs?: number; terminateWaitMs?: number; releaseTimeoutMs?: number } = {}) { + this.#service = service; + // Beside the review database's canonical path, so a restart of the same review finds what an earlier session left. + const ledger = options.ledger ?? LeftoverLedger.forDatabase(service.config.database, PLANNING_NAMING); + this.#worker = new AgentWorker(options.url, ledger, { naming: PLANNING_NAMING, env: options.env, + abandonAfterDeadlineMs: options.abandonAfterDeadlineMs, terminateWaitMs: options.terminateWaitMs, + releaseTimeoutMs: options.releaseTimeoutMs, + runnerOwner: () => service.store.planningOwnerToken(ledger.identity!) }); + } + async invoke(request: AuthorRequest, signal: AbortSignal): Promise { + signal.throwIfAborted(); + const store = this.#service.store, vendor = store.questionProvider(); + if (!vendor) throw new Error(NO_PLANNING_AGENT); + const snapshot = store.getSnapshot(request.identity); + const run = planningRun(request, { vendor, repository: this.#service.reviewRepository().path, head: snapshot.head, + snapshotId: snapshot.id }, Date.now() + PLANNING_BUDGET_MS - SETTLE_MARGIN_MS); + return this.#worker.plan(run, signal); + } + /** Call only after every request has settled (E3's close). Releases what the worker still owns, then stops it. */ + close(): Promise { return this.#worker.close(); } +} diff --git a/runner/question-agent.ts b/runner/question-agent.ts index 2655d6b2..94b65140 100644 --- a/runner/question-agent.ts +++ b/runner/question-agent.ts @@ -1,9 +1,10 @@ import { randomBytes, randomUUID } from 'node:crypto'; import { Worker } from 'node:worker_threads'; import type { QuestionAgent } from './questions.ts'; -import { credentialEnvironment, questionCredential, stopOf, workerEnvironment, type Provider } from './question-container.ts'; +import { credentialEnvironment, questionCredential, stopOf, workerEnvironment, type Provider, type ReadOnlyRun } from './question-container.ts'; +import { planningCredential } from './planning-provider.ts'; import type { RecoverReply, ReleaseReply, WorkerReply, WorkerRequest } from './question-worker.ts'; -import { createAskRoot, removeAskRoot, type LeftoverLedger } from './question-leftovers.ts'; +import { ASK_NAMING, createWorkerRoot, removeWorkerRoot, type LeftoverLedger, type WorkerNaming } from './question-leftovers.ts'; export type { Provider } from './question-container.ts'; // Leave the worker time to cancel the container and release storage before the review's own timeout fires. @@ -16,8 +17,12 @@ const ABANDON_AFTER_DEADLINE_MS = 30_000; // Bounds the wait for an abandoned worker thread to stop (a synchronous Docker or Git call finishes first). const DEFAULT_TERMINATE_WAIT_MS = 15_000; -/** One worker owns every Ask container, so lane D's trusted image and allocations stay in one registry. */ -export class QuestionWorker { +/** + * One worker owns every container of one feature (Ask or planning), so lane D's trusted image and allocations stay in + * one registry. Each feature has its own worker, root, ledger, lock and owner (#117); `naming` says which. + */ +export class AgentWorker { + private readonly naming: WorkerNaming; private worker?: Worker; // The worker's TMPDIR. Recorded before the worker starts, deleted after it stops. private root?: string; @@ -44,8 +49,12 @@ export class QuestionWorker { private env: Readonly>; constructor(url = new URL('./question-worker.ts', import.meta.url), ledger?: LeftoverLedger, options: { abandonAfterDeadlineMs?: number; terminateWaitMs?: number; releaseTimeoutMs?: number; env?: Readonly>; - /** The review's Ask owner token. Without one, each worker uses a random owner, so it recovers nothing earlier. */ - runnerOwner?: () => string } = {}) { + /** The review's owner token for this feature. Without one, each worker uses a random owner, so it recovers nothing earlier. */ + runnerOwner?: () => string; + /** Which feature this worker serves. Defaults to the ledger's, then Ask's. */ + naming?: WorkerNaming } = {}) { + this.naming = options.naming ?? ledger?.naming ?? ASK_NAMING; + if (ledger && ledger.naming !== this.naming) throw new Error('A worker and its ledger must serve the same feature.'); const random = randomBytes(16).toString('hex'); this.owner = options.runnerOwner ?? (() => random); this.url = url; this.ledger = ledger; this.abandonAfterMs = options.abandonAfterDeadlineMs ?? ABANDON_AFTER_DEADLINE_MS; @@ -57,16 +66,16 @@ export class QuestionWorker { if (this.crashed) throw this.crashed; if (this.worker) return this.worker; // Stamped with this review's lock, so a later process can find it even if the record is renamed away or lost. - const root = createAskRoot(this.ledger?.lockPath ?? ''); + const root = createWorkerRoot(this.ledger?.lockPath ?? '', this.naming); // Durable before any setup: a process killed from here on still leaves a record of this root. - try { this.ledger?.recordRoot(root); } catch (error) { removeAskRoot(root); throw error; } + try { this.ledger?.recordRoot(root); } catch (error) { removeWorkerRoot(root, this.naming); throw error; } let worker: Worker; try { worker = new Worker(this.url, { env: workerEnvironment(process.env, root), - workerData: { credentials: credentialEnvironment(this.env) } }); + workerData: { credentials: credentialEnvironment(this.env), feature: this.naming.feature } }); } catch (error) { // Nothing ran in the root yet: delete it and drop the record, so close() can release the lock. - removeAskRoot(root); this.ledger?.forget(root); + removeWorkerRoot(root, this.naming); this.ledger?.forget(root); throw error; } this.root = root; @@ -81,7 +90,7 @@ export class QuestionWorker { if (!job) return; this.pending.delete(reply.id); clearTimeout(job.watchdog); - if (reply.attemptId !== job.attemptId) job.reject(new Error('The agent returned a result for a different question attempt.')); + if (reply.attemptId !== job.attemptId) job.reject(new Error(`The agent returned a result for a different ${this.naming.job} attempt.`)); else if (reply.ok) job.resolve(reply.text); else job.reject(new Error(reply.error)); }); const fail = (error: Error) => { if (this.worker === worker) void this.#abandon(`stopped (${error.message})`); }; @@ -104,7 +113,7 @@ export class QuestionWorker { async #abandonOnce(why: string) { const worker = this.worker; this.worker = undefined; - this.crashed ??= new Error(`The agent container worker ${why}. Its containers and storage may still exist, so Ask is off until codeboost restarts. The first question after the restart removes them.`); + this.crashed ??= new Error(`The agent container worker ${why}. Its containers and storage may still exist, so ${this.naming.label} is off until codeboost restarts. The first ${this.naming.job} after the restart removes them.`); for (const release of this.releases.values()) release(null); this.releases.clear(); this.#failRecoveries(this.crashed); @@ -133,7 +142,7 @@ export class QuestionWorker { if (!root) return; this.root = undefined; try { - removeAskRoot(root); this.ledger?.forget(root); + removeWorkerRoot(root, this.naming); this.ledger?.forget(root); // A thread that stopped after close() has no more files to write; the lock still stays if it was abandoned. if (this.closed && !this.#abandoning && !this.#recoveryStopped) this.ledger?.release(); } @@ -141,18 +150,31 @@ export class QuestionWorker { } agent(provider: Provider): QuestionAgent { return async (prompt, signal, scope, timeoutMs) => { - if (this.crashed) throw this.crashed; // Missing sign-in is reported before any Docker work, including recovery. - questionCredential(provider, this.env); - // Held until close, so no other process can recover, start a worker or write the record for this review. - this.ledger?.acquire(); - // The first question of a process removes what earlier sessions left. The check is single-flight: concurrent - // first questions share it, so recovery never runs once a question has started. - if (!this.scanned) await this.#startupScan(signal); - signal.throwIfAborted(); + await this.#admit(() => questionCredential(provider, this.env), signal); return this.#ask(provider, prompt, signal, scope, timeoutMs); }; } + /** + * Run one planning request in this worker (#117). `deadline` bounds the whole request, setup included; the caller's + * own timer must allow the worker its settle margin after it. + */ + async plan(run: Omit, signal: AbortSignal): Promise { + if (this.naming.feature !== 'planning') throw new Error('This worker does not run plans.'); + await this.#admit(() => planningCredential(run.provider, this.env), signal); + return this.#submit(run.attemptId, run.deadline, signal, id => ({ type: 'plan', id, run: { ...run, runnerOwner: this.runnerOwner! } })); + } + /** What every job waits for: no crash, a credential, the review's lock, and this process's startup check. */ + async #admit(credential: () => unknown, signal: AbortSignal) { + if (this.crashed) throw this.crashed; + credential(); + // Held until close, so no other process can recover, start a worker or write the record for this review. + this.ledger?.acquire(); + // The first job of a process removes what earlier sessions left. The check is single-flight: concurrent + // first jobs share it, so recovery never runs once a job has started. + if (!this.scanned) await this.#startupScan(signal); + signal.throwIfAborted(); + } #scanning?: Promise; /** * One startup check for all concurrent first questions: delete earlier Ask roots, then recover this review's Docker @@ -184,19 +206,25 @@ export class QuestionWorker { }); } #ask(provider: Provider, ...[prompt, signal, scope, timeoutMs]: Parameters) { + if (!scope) return Promise.reject(new Error('Ask needs the reviewed repository and head.')); + const deadline = Date.now() + Math.max(1_000, (timeoutMs ?? 120_000) - SETTLE_MARGIN_MS); + return this.#submit(scope.attemptId, deadline, signal, id => ({ type: 'ask', id, + question: { ...scope, provider, prompt, runnerOwner: this.runnerOwner!, deadline } })); + } + /** + * Post one job to the worker. The promise settles only when the worker reports that the container and its storage + * are gone, or when the worker is abandoned because it did not settle in time after `deadline`. + */ + #submit(attemptId: string, deadline: number, signal: AbortSignal, request: (id: string) => WorkerRequest) { return new Promise((resolve, reject) => { - if (!scope) { reject(new Error('Ask needs the reviewed repository and head.')); return; } let worker: Worker; try { worker = this.start(); } catch (error) { reject(error as Error); return; } const id = randomUUID(); - const question = { ...scope, provider, prompt, runnerOwner: this.runnerOwner!, - deadline: Date.now() + Math.max(1_000, (timeoutMs ?? 120_000) - SETTLE_MARGIN_MS) }; - const watchdog = setTimeout(() => { if (this.pending.has(id)) void this.#abandon('did not settle a question after its deadline'); }, - question.deadline - Date.now() + this.abandonAfterMs); + const watchdog = setTimeout(() => { if (this.pending.has(id)) void this.#abandon(`did not settle a ${this.naming.job} after its deadline`); }, + deadline - Date.now() + this.abandonAfterMs); watchdog.unref?.(); - this.pending.set(id, { attemptId: scope.attemptId, resolve, reject, watchdog }); - worker.postMessage({ type: 'ask', id, question } satisfies WorkerRequest); - // The promise settles only when the worker reports that the container and its storage are gone. + this.pending.set(id, { attemptId, resolve, reject, watchdog }); + worker.postMessage(request(id)); const cancel = () => worker.postMessage({ type: 'cancel', id, stop: stopOf(signal.reason), reason: signal.reason instanceof Error ? signal.reason.message : 'Agent cancelled.' } satisfies WorkerRequest); if (signal.aborted) cancel(); else signal.addEventListener('abort', cancel, { once: true }); @@ -225,7 +253,7 @@ export class QuestionWorker { // An abandonment already in progress (crash or watchdog) owns the worker: wait for its bounded settlement. if (!worker) { await this.#abandoning; return; } // Questions still waiting mean lane D has not settled; do not wait on it at shutdown. - if (this.pending.size) { await this.#abandon('was stopped at shutdown with questions still settling'); return; } + if (this.pending.size) { await this.#abandon(`was stopped at shutdown with ${this.naming.jobs} still settling`); return; } const id = randomUUID(); let timer: ReturnType | undefined; const released = await new Promise | null>(resolve => { @@ -238,10 +266,10 @@ export class QuestionWorker { // which keeps the root recorded until the thread has stopped. if (released === null) { await this.#abandon('did not report its storage before shutdown'); return; } this.worker = undefined; - if (released.remaining) console.error(`codeboost: ${released.remaining} Ask allocation(s) could not be removed at shutdown; the first question after the next start removes them.`); + if (released.remaining) console.error(`codeboost: ${released.remaining} ${this.naming.label} allocation(s) could not be removed at shutdown; the first ${this.naming.job} after the next start removes them.`); try { await worker.terminate(); } // A recovery still in flight stops with the thread; the next process's recovery finishes it. - finally { this.#failRecoveries(new Error('Server stopped. Retry the question.')); } + finally { this.#failRecoveries(new Error(`Server stopped. Retry the ${this.naming.job}.`)); } this.#removeRoot(); } /** Set when a recovery was stopped before it answered: its Docker clients may still be running. */ @@ -252,3 +280,6 @@ export class QuestionWorker { this.recoveries.clear(); } } +/** Ask's worker (see AgentWorker). */ +export const QuestionWorker = AgentWorker; +export type QuestionWorker = AgentWorker; diff --git a/runner/question-container.ts b/runner/question-container.ts index b1e0e0fe..bf04b0c7 100644 --- a/runner/question-container.ts +++ b/runner/question-container.ts @@ -8,7 +8,7 @@ import type { AgentAdapterRequest } from '../agents/adapters/types.ts'; import type { RecoveredTaskStorage, TaskFilesystems, TaskStorageLimits } from '../agents/container/storage.ts'; import { RUNNER_LABEL, type ResourceOwner } from '../agents/labels.ts'; import { recoverLeftovers, type RecoveryReport } from '../agents/recovery.ts'; -import { removeStaging } from './question-leftovers.ts'; +import { ASK_NAMING, removeStaging, type WorkerNaming } from './question-leftovers.ts'; import { removalCommand } from './recovery.ts'; export type Provider = 'claude' | 'codex'; @@ -350,14 +350,14 @@ const MAX_LISTED = 20; * * Recovery is asked not to look for objects without an owner label: they come from builds before runner labels and * may belong to any review on this daemon, so they neither block Ask nor are touched. Objects carrying this owner that - * recovery cannot identify keep Ask off. + * recovery cannot identify keep Ask off. Planning's worker runs it the same way with its own owner and `naming` (#117). */ export async function recoverQuestionStorage(runnerOwner: string, deps: Pick, - retained: RetainedStorage): Promise { + retained: RetainedStorage, naming: WorkerNaming = ASK_NAMING): Promise { let report: RecoveryReport; try { report = await deps.recover(runnerOwner); } catch (error) { - throw new Error(`Ask is off: codeboost could not remove what an earlier session of this review left in Docker (${error instanceof Error ? error.message.slice(0, 300) : 'unknown error'}). Check that Docker is running, then retry.`); + throw new Error(`${naming.label} is off: codeboost could not remove what an earlier session of this review left in Docker (${error instanceof Error ? error.message.slice(0, 300) : 'unknown error'}). Check that Docker is running, then retry.`); } for (const handle of report.storage) { try { deps.removeFilesystems(handle); } catch { retained.retain(handle); } @@ -366,5 +366,5 @@ export async function recoverQuestionStorage(runnerOwner: string, deps: Pick 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')}`); + throw new Error(`${naming.label} is off: Docker objects labelled with this review's ${naming.activity} 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/question-leftovers.ts b/runner/question-leftovers.ts index 0c9ce2d4..c3163789 100644 --- a/runner/question-leftovers.ts +++ b/runner/question-leftovers.ts @@ -21,10 +21,51 @@ const isTemporary = (path: unknown, prefix: string, parent: string): path is str typeof path === 'string' && path.length <= 4096 && isAbsolute(path) && dirname(path) === parent && new RegExp(`^${prefix}[A-Za-z0-9]{6}$`).test(basename(path)); /** - * The Ask root: one directory per question worker, set as the worker's TMPDIR, so every host copy it or lane D makes - * (reviewed clone, input, the Codex auth copy) lives inside it. Only this exact shape is accepted from the record. + * The names and wording of one feature that runs agents in its own worker (Ask, planning): its worker root, lock and + * record, and how its refusals name it. Ask's values reproduce Ask's names and messages exactly. */ -export const isAskRoot = (path: unknown): path is string => isTemporary(path, 'codeboost-ask-', tmpdir()); +export interface WorkerNaming { + /** The feature as a person reads it: "Ask is off: ...". */ + readonly label: string; + /** Prefix of the worker root, the worker's TMPDIR: `codeboost--XXXXXX`. */ + readonly root: string; + /** Prefix of the folder a root is prepared under before it is renamed into place. */ + readonly prep: string; + /** Prefix of the lock file, and of the private directory that holds the locks. */ + readonly lock: string; + /** Suffix of the record beside the review database. */ + readonly record: string; + /** What the host copies are, in refusals: "reviewed code" for Ask. */ + readonly code: string; + /** The worker's folders, in refusals: "Ask folders". */ + readonly folders: string; + /** The worker, in errors: "the Ask worker". */ + readonly worker: string; + /** One worker root, in refusals: "an Ask root". */ + readonly rootName: string; + /** What another process may be running, in refusals: "running Ask for this review". */ + readonly activity: string; + /** One unit of work, in messages: "the first question after the restart". */ + readonly job: string; + /** Units of work, in messages: "questions still settling". */ + readonly jobs: string; + /** Which jobs the feature's worker runs (see question-worker.ts). */ + readonly feature: 'questions' | 'planning'; +} +export const ASK_NAMING: WorkerNaming = Object.freeze({ label: 'Ask', root: 'ask', prep: 'askprep', lock: 'asklock', + record: 'ask-leftovers.json', code: 'reviewed code', folders: 'Ask folders', worker: 'the Ask worker', rootName: 'an Ask root', activity: 'Ask', + job: 'question', jobs: 'questions', feature: 'questions' }); +export const PLANNING_NAMING: WorkerNaming = Object.freeze({ label: 'Planning', root: 'plan', prep: 'planprep', lock: 'planlock', + record: 'planning-leftovers.json', code: 'planned code', folders: 'planning folders', worker: 'the planning worker', rootName: 'a planning root', activity: 'planning', + job: 'planning request', jobs: 'planning requests', feature: 'planning' }); + +/** + * A worker root: one directory per agent worker, set as the worker's TMPDIR, so every host copy it or lane D makes + * (the clone, input, the Codex auth copy) lives inside it. Only this exact shape is accepted from the record. + */ +export const isWorkerRoot = (path: unknown, naming: WorkerNaming): path is string => isTemporary(path, `codeboost-${naming.root}-`, tmpdir()); +/** The Ask root (see isWorkerRoot). */ +export const isAskRoot = (path: unknown): path is string => isWorkerRoot(path, ASK_NAMING); /** A question's staging directory, inside the worker's TMPDIR (the Ask root). */ export const isStagingPath = (path: unknown): path is string => isTemporary(path, 'codeboost-question-', tmpdir()); @@ -46,11 +87,13 @@ export function removeStaging(root: string): void { if (!isStagingPath(root)) throw new Error('Refusing to remove a path that is not an Ask staging directory.'); removeTree(root); } -/** Remove an Ask root and everything in it. Throws if it cannot be removed. */ -export function removeAskRoot(root: string): void { - if (!isAskRoot(root)) throw new Error('Refusing to remove a path that is not an Ask root.'); +/** Remove a worker root and everything in it. Throws if it cannot be removed. */ +export function removeWorkerRoot(root: string, naming: WorkerNaming): void { + if (!isWorkerRoot(root, naming)) throw new Error(`Refusing to remove a path that is not ${naming.rootName}.`); removeTree(root); } +/** Remove an Ask root and everything in it. Throws if it cannot be removed. */ +export const removeAskRoot = (root: string): void => removeWorkerRoot(root, ASK_NAMING); const MAX_LEFTOVERS = 100; const OWNER_FILE = '.owner'; @@ -75,47 +118,53 @@ function readNoFollow(path: string): string | undefined { return buffer.subarray(0, offset).toString('utf8'); } finally { closeSync(fd); } } -/** A codeboost Ask lock: a direct child of the temp directory with the lock name, so a stamp cannot aim elsewhere. */ -export const isAskLock = (path: unknown): path is string => typeof path === 'string' && path.length <= 4096 - && isAbsolute(path) && dirname(path) === lockDirectoryPath() && /^codeboost-asklock-[0-9a-f]+(?:-[0-9]+)?\.sqlite$/.test(basename(path)); +/** A codeboost worker lock: a direct child of the lock directory with the lock name, so a stamp cannot aim elsewhere. */ +export const isWorkerLock = (path: unknown, naming: WorkerNaming): path is string => typeof path === 'string' && path.length <= 4096 + && isAbsolute(path) && dirname(path) === lockDirectoryPath(naming) + && new RegExp(`^codeboost-${naming.lock}-[0-9a-f]+(?:-[0-9]+)?\\.sqlite$`).test(basename(path)); +/** A codeboost Ask lock (see isWorkerLock). */ +export const isAskLock = (path: unknown): path is string => isWorkerLock(path, ASK_NAMING); /** * Lock files live in a directory only this user can write, so no other local user can plant or swap one (for example * a symlink to an unrelated database) between the name check and SQLite opening it. Refused if it is not ours. */ -const lockDirectoryPath = () => join(tmpdir(), `codeboost-asklocks-${process.getuid?.() ?? 'user'}`); -function lockDirectory(): string { - const directory = lockDirectoryPath(); +const lockDirectoryPath = (naming: WorkerNaming) => join(tmpdir(), `codeboost-${naming.lock}s-${process.getuid?.() ?? 'user'}`); +function lockDirectory(naming: WorkerNaming): string { + const directory = lockDirectoryPath(naming); try { mkdirSync(directory, { mode: 0o700 }); } catch (error) { if ((error as NodeJS.ErrnoException).code !== 'EEXIST') throw error; } const stat = lstatSync(directory); if (!stat.isDirectory() || stat.isSymbolicLink() || (process.getuid && stat.uid !== process.getuid()) || (stat.mode & 0o077) !== 0) - throw new Error(`Ask is off: the lock directory ${directory} is not a private directory owned by you. Remove it, then retry.`); + throw new Error(`${naming.label} is off: the lock directory ${directory} is not a private directory owned by you. Remove it, then retry.`); return directory; } /** Open a lock file only if it is a regular file or absent; a symlink or other file type is refused, never followed. */ -function assertPlainLockFile(path: string): void { +function assertPlainLockFile(path: string, naming: WorkerNaming): void { const stat = lstatSync(path, { throwIfNoEntry: false }); - if (stat && (!stat.isFile() || stat.isSymbolicLink())) throw new Error(`Ask is off: ${path} is not a plain lock file. Remove it, then retry.`); + if (stat && (!stat.isFile() || stat.isSymbolicLink())) throw new Error(`${naming.label} is off: ${path} is not a plain lock file. Remove it, then retry.`); } /** - * Create an Ask root stamped with the lock of the process that owns it. The stamp is written under a preparation name - * and the folder is then renamed, so any folder visible under the Ask root name already carries its owner stamp. + * Create a worker root stamped with the lock of the process that owns it. The stamp is written under a preparation + * name and the folder is then renamed, so any folder visible under the root name already carries its owner stamp. */ -export function createAskRoot(lockPath: string): string { +export function createWorkerRoot(lockPath: string, naming: WorkerNaming): string { + const prepPrefix = `codeboost-${naming.prep}-`; for (let attempt = 0; attempt < 5; attempt++) { - const prep = mkdtempSync(join(tmpdir(), 'codeboost-askprep-')); + const prep = mkdtempSync(join(tmpdir(), prepPrefix)); writeFileSync(join(prep, OWNER_FILE), `${lockPath}\n`, { mode: 0o600, flag: 'wx' }); - const root = join(tmpdir(), `codeboost-ask-${basename(prep).slice('codeboost-askprep-'.length)}`); + const root = join(tmpdir(), `codeboost-${naming.root}-${basename(prep).slice(prepPrefix.length)}`); if (!existsSync(root)) try { renameSync(prep, root); return root; } catch { /* taken meanwhile; try another name */ } rmSync(prep, { recursive: true, force: true }); } - throw new Error('Could not create a folder for the Ask worker.'); + throw new Error(`Could not create a folder for ${naming.worker}.`); } -/** Whether another process holds an Ask lock file, tested without creating or keeping it. */ -function lockIsHeld(path: string): boolean { +/** Create an Ask root (see createWorkerRoot). */ +export const createAskRoot = (lockPath: string): string => createWorkerRoot(lockPath, ASK_NAMING); +/** Whether another process holds a worker lock file, tested without creating or keeping it. */ +function lockIsHeld(path: string, naming: WorkerNaming): boolean { const { DatabaseSync } = createRequire(import.meta.url)('node:sqlite') as typeof import('node:sqlite'); let probe: import('node:sqlite').DatabaseSync | undefined; try { - assertPlainLockFile(path); + assertPlainLockFile(path, naming); probe = new DatabaseSync(path, { timeout: 0 }); probe.exec('BEGIN EXCLUSIVE; ROLLBACK;'); return false; @@ -123,12 +172,12 @@ function lockIsHeld(path: string): boolean { finally { probe?.close(); } } -function parse(text: string): LedgerRecord { +function parse(text: string, naming: WorkerNaming): LedgerRecord { const value = JSON.parse(text) as { leftovers?: unknown; untracked?: unknown; roots?: unknown }; // `leftovers` and `untracked` appear only in records from builds before #65. const list = value?.leftovers ?? [], untracked = value?.untracked ?? 0, roots = value?.roots; if (!Array.isArray(list) || list.length > MAX_LEFTOVERS || !Number.isSafeInteger(untracked) || (untracked as number) < 0 - || !Array.isArray(roots) || roots.length > MAX_LEFTOVERS || !roots.every(isAskRoot)) + || !Array.isArray(roots) || roots.length > MAX_LEFTOVERS || !roots.every(root => isWorkerRoot(root, naming))) throw new Error('invalid record'); return { untracked: untracked as number, roots: roots as string[], leftovers: list.map(entry => { const { keeper, workVolume, metadataVolume } = (entry ?? {}) as Record; @@ -139,25 +188,27 @@ function parse(text: string): LedgerRecord { } /** - * Durable record of Ask's host copies (Ask roots) that outlived their worker, and the review's Ask lock. Docker - * leftovers need no record: Ask labels them with the review's own owner token, and the first question of each process - * removes them through lane D's scoped recovery (#51 item 4, #65). + * Durable record of a feature's host copies (worker roots) that outlived their worker, and the review's lock for that + * feature. Docker leftovers need no record: the feature labels them with the review's own owner token for it, and the + * first request of each process removes them through lane D's scoped recovery (#51 item 4, #65). Ask and planning each + * have their own ledger, lock and owner, so neither one's recovery or refusals affect the other (#117). */ export class LeftoverLedger { readonly path: string; + readonly naming: WorkerNaming; #lock?: import('node:sqlite').DatabaseSync; #refusal?: string; /** Where the exclusive lock lives; for a review database it is keyed by the file's identity (see forDatabase). */ lockPath: string; - /** The review database's device and inode, which key its Ask owner token (see forDatabase). */ + /** The review database's device and inode, which key the feature's owner token (see forDatabase). */ identity?: { readonly dev: bigint; readonly ino: bigint }; - constructor(path: string) { - this.path = path; - this.lockPath = join(lockDirectoryPath(), `codeboost-asklock-${createHash('sha256').update(path).digest('hex').slice(0, 32)}.sqlite`); + constructor(path: string, naming: WorkerNaming = ASK_NAMING) { + this.path = path; this.naming = naming; + this.lockPath = join(lockDirectoryPath(naming), `codeboost-${naming.lock}-${createHash('sha256').update(path).digest('hex').slice(0, 32)}.sqlite`); } /** - * Exclusive Ask lock for this review database, held for the question worker's lifetime. Only the holder checks, + * Exclusive lock on this feature for this review database, held for the worker's lifetime. Only the holder checks, * recovers, starts a worker or writes this record, so two processes on one review cannot both pass the startup check or * overwrite each other's record. It is an exclusive SQLite transaction on `.lock`: an OS file lock that the * operating system releases when its process ends, so no PID check or takeover is needed. @@ -166,14 +217,14 @@ export class LeftoverLedger { if (this.#lock) return; if (this.#refusal) throw new Error(this.#refusal); const { DatabaseSync } = createRequire(import.meta.url)('node:sqlite') as typeof import('node:sqlite'); - lockDirectory(); - assertPlainLockFile(this.lockPath); + lockDirectory(this.naming); + assertPlainLockFile(this.lockPath, this.naming); const lock = new DatabaseSync(this.lockPath, { timeout: 0 }); try { lock.exec('PRAGMA locking_mode=EXCLUSIVE; BEGIN EXCLUSIVE;'); } catch (error) { lock.close(); if (/locked|busy/i.test(String((error as Error).message))) - throw new Error('Ask is off: another codeboost process is running Ask for this review. Stop it, then retry.'); + throw new Error(`${this.naming.label} is off: another codeboost process is running ${this.naming.activity} for this review. Stop it, then retry.`); throw error; } this.#lock = lock; @@ -186,7 +237,7 @@ export class LeftoverLedger { } /** - * Delete unrecorded Ask roots whose owner is gone. A root outlives its record when the database is renamed or the + * Delete unrecorded worker roots whose owner is gone. A root outlives its record when the database is renamed or the * record is lost; its `.owner` stamp names the lock of the process that made it. A held lock means a live process * owns the root and it is left alone. A free lock, a missing lock file or a missing stamp means the owner is gone. */ @@ -194,36 +245,36 @@ export class LeftoverLedger { const stuck: string[] = []; for (const name of readdirSync(tmpdir())) { const root = join(tmpdir(), name); - if (!isAskRoot(root) || skip.has(root)) continue; - // Only a folder this user owns, carrying a valid stamp from createAskRoot, is ours to judge. createAskRoot stamps + if (!isWorkerRoot(root, this.naming) || skip.has(root)) continue; + // Only a folder this user owns, carrying a valid stamp from createWorkerRoot, is ours to judge. It stamps // every root before it becomes visible, so an unstamped, tampered or foreign folder is left in place. const stat = lstatSync(root, { throwIfNoEntry: false }); if (!stat || !stat.isDirectory() || stat.isSymbolicLink() || (process.getuid && stat.uid !== process.getuid())) continue; let owner = ''; try { owner = readNoFollow(join(root, OWNER_FILE))?.trim() ?? ''; } catch { continue; } - if (!isAskLock(owner)) continue; - if (owner !== this.lockPath && existsSync(owner) && lockIsHeld(owner)) continue; + if (!isWorkerLock(owner, this.naming)) continue; + if (owner !== this.lockPath && existsSync(owner) && lockIsHeld(owner, this.naming)) continue; // Our own lock is held by us, so our earlier-session roots (not the live one, which is skipped) are reclaimed. - try { removeAskRoot(root); } catch { stuck.push(root); } + try { removeWorkerRoot(root, this.naming); } catch { stuck.push(root); } } return stuck; } /** * The ledger for a review database, keyed by its canonical path so relative, absolute and symlinked spellings share - * one lock and record. A hard-linked database has no single canonical path, so Ask refuses to run on it. + * one lock and record. A hard-linked database has no single canonical path, so the feature refuses to run on it. */ - static forDatabase(database: string): LeftoverLedger { + static forDatabase(database: string, naming: WorkerNaming = ASK_NAMING): LeftoverLedger { const canonical = realpathSync(database); - const ledger = new LeftoverLedger(`${canonical}.ask-leftovers.json`); + const ledger = new LeftoverLedger(`${canonical}.${naming.record}`, naming); // One stat for the lock path, the hard-link check and the owner identity, so all three name the same file. const identity = statSync(canonical, { bigint: true }); ledger.identity = { dev: identity.dev, ino: identity.ino }; // The lock only excludes, so it may live in the temp directory; keyed by device and inode, every spelling and // every later name of this database file (including an atomic rename while a server runs) finds the same lock. - ledger.lockPath = join(lockDirectoryPath(), `codeboost-asklock-${identity.dev}-${identity.ino}.sqlite`); + ledger.lockPath = join(lockDirectoryPath(naming), `codeboost-${naming.lock}-${identity.dev}-${identity.ino}.sqlite`); if (identity.nlink > 1) - ledger.#refusal = `Ask is off: the review database ${canonical} has other hard links, so codeboost cannot tell whether another process is using it. Use a database file without hard links.`; + ledger.#refusal = `${naming.label} is off: the review database ${canonical} has other hard links, so codeboost cannot tell whether another process is using it. Use a database file without hard links.`; return ledger; } @@ -233,9 +284,9 @@ export class LeftoverLedger { // A planted link here could make this review act on another review's record: never follow one. text = readNoFollow(this.path); if (text === undefined) return { roots: [], leftovers: [], untracked: 0 }; - return parse(text); + return parse(text, this.naming); } - catch { throw new Error(`Ask is off: the record of leftover Ask folders (${this.path}) is unreadable. Delete the codeboost-ask-* folders in ${tmpdir()} that no running codeboost uses, then delete that file.`); } + catch { throw new Error(`${this.naming.label} is off: the record of leftover ${this.naming.folders} (${this.path}) is unreadable. Delete the codeboost-${this.naming.root}-* folders in ${tmpdir()} that no running codeboost uses, then delete that file.`); } } #write(record: LedgerRecord): void { @@ -251,25 +302,25 @@ export class LeftoverLedger { } catch (error) { rmSync(temporary, { force: true }); throw error; } } - /** Add an Ask root that may hold host copies, before anything is written into it. */ + /** Add a worker root that may hold host copies, before anything is written into it. */ recordRoot(root: string): void { - if (!isAskRoot(root)) throw new Error('Refusing to record a path that is not an Ask root.'); + if (!isWorkerRoot(root, this.naming)) throw new Error(`Refusing to record a path that is not ${this.naming.rootName}.`); const known = this.#read(); if (known.roots.includes(root)) return; // Roots hold host copies that only their path can find, so none is ever dropped: refuse to add one past the cap. if (known.roots.length >= MAX_LEFTOVERS) - throw new Error(`Ask is off: ${known.roots.length} Ask folders from earlier sessions could not be deleted. Delete the codeboost-ask-* folders in ${tmpdir()}, then retry.`); + throw new Error(`${this.naming.label} is off: ${known.roots.length} ${this.naming.folders} from earlier sessions could not be deleted. Delete the codeboost-${this.naming.root}-* folders in ${tmpdir()}, then retry.`); this.#write({ ...known, roots: [...known.roots, root] }); } - /** Drop an Ask root from the record after it has been deleted. */ + /** Drop a worker root from the record after it has been deleted. */ forget(root: string): void { const known = this.#read(); if (known.roots.includes(root)) this.#write({ ...known, roots: known.roots.filter(entry => entry !== root) }); } /** - * The host part of the first check of a process: delete Ask roots of earlier sessions, recorded or found by their + * The host part of the first check of a process: delete worker roots of earlier sessions, recorded or found by their * owner stamp, but never `active`. Throws, with the commands to finish by hand, while any remain. Then refuses while * the record still lists Docker storage from a build before #65, which recovery cannot find. */ @@ -277,19 +328,19 @@ export class LeftoverLedger { const known = this.#read(); // Roots this record does not list (a renamed database, a lost record) are found by their owner stamp. const orphans = this.#reclaimOrphanRoots(new Set([...known.roots, ...(options.active ? [options.active] : [])])); - if (orphans.length) throw new Error(`Ask is off: host copies of reviewed code or credentials from an earlier session could not be deleted. Delete them, then retry:\n${orphans.map(root => `rm -rf '${root}'`).join('\n')}`); + if (orphans.length) throw new Error(`${this.naming.label} is off: host copies of ${this.naming.code} or credentials from an earlier session could not be deleted. Delete them, then retry:\n${orphans.map(root => `rm -rf '${root}'`).join('\n')}`); // Host copies (reviewed code, Codex auth) need no Docker: delete earlier roots first, never the live one. const roots = known.roots.filter(root => { if (root === options.active) return true; - try { removeAskRoot(root); return false; } catch { return true; } + try { removeWorkerRoot(root, this.naming); return false; } catch { return true; } }); if (roots.length !== known.roots.length) this.#write({ ...known, roots }); const stuck = roots.filter(root => root !== options.active); - if (stuck.length) throw new Error(`Ask is off: host copies of reviewed code or credentials from an earlier session could not be deleted. Delete them, then retry:\n${stuck.map(root => `rm -rf '${root}'`).join('\n')}`); + if (stuck.length) throw new Error(`${this.naming.label} is off: host copies of ${this.naming.code} or credentials from an earlier session could not be deleted. Delete them, then retry:\n${stuck.map(root => `rm -rf '${root}'`).join('\n')}`); // Last, so host copies are deleted (and dropped from the record) before the user is told to delete the record. if (known.leftovers.length || known.untracked) { const commands = known.leftovers.flatMap(entry => [`docker rm -f ${entry.keeper}`, `docker volume rm ${entry.workVolume} ${entry.metadataVolume}`]); - throw new Error(`Ask is off: an earlier codeboost build recorded agent storage it could not remove, and codeboost cannot find it by owner. Remove ${commands.length ? 'what is left of it (a command for an object that is already gone fails harmlessly)' : 'it'}, and any other container, volume or network labelled io.codeboost.allocation, io.codeboost.invocation or io.codeboost.egress that no running codeboost uses. Then delete ${this.path} and retry.${commands.length ? `\n${commands.join('\n')}` : ''}`); + throw new Error(`${this.naming.label} is off: an earlier codeboost build recorded agent storage it could not remove, and codeboost cannot find it by owner. Remove ${commands.length ? 'what is left of it (a command for an object that is already gone fails harmlessly)' : 'it'}, and any other container, volume or network labelled io.codeboost.allocation, io.codeboost.invocation or io.codeboost.egress that no running codeboost uses. Then delete ${this.path} and retry.${commands.length ? `\n${commands.join('\n')}` : ''}`); } } } diff --git a/runner/question-worker.ts b/runner/question-worker.ts index 8fc1477c..da35ec80 100644 --- a/runner/question-worker.ts +++ b/runner/question-worker.ts @@ -4,11 +4,16 @@ import { captureInvocation } from '../agents/contract.ts'; import { buildAgentImage } from '../agents/container/image.ts'; import { prepareTaskFilesystems, removeTaskFilesystems } from '../agents/container/run.ts'; import { createTaskClone } from '../git/clone.ts'; -import { askInContainer, measureGitRepository, recoverAskOwner, recoverQuestionStorage, RetainedStorage, StopError, type ContainerDependencies, type ContainerQuestion } from './question-container.ts'; +import { askInContainer, measureGitRepository, recoverAskOwner, recoverQuestionStorage, RetainedStorage, runReadOnlyAgent, StopError, type ContainerDependencies, type ContainerQuestion, type ReadOnlyRun } from './question-container.ts'; +import { PLANNING_FEATURE, PlanningStorage } from './planning-provider.ts'; +import { ASK_NAMING, PLANNING_NAMING } from './question-leftovers.ts'; // Lane D's image build, clone and storage allocation are synchronous (Docker and Git calls), so they run here instead of blocking the review server. // Its trust registries (built image, clones, allocations, captured invocations) live in this worker's modules. -export type WorkerRequest = { type: 'ask'; id: string; question: ContainerQuestion } | { type: 'cancel'; id: string; reason: string; stop: StopError['stop'] } +// One worker serves one feature, named in workerData: Ask's worker runs only questions, planning's only plans (#117). +export type WorkerFeature = 'questions' | 'planning'; +export type WorkerRequest = { type: 'ask'; id: string; question: ContainerQuestion } | { type: 'plan'; id: string; run: ReadOnlyRun } + | { type: 'cancel'; id: string; reason: string; stop: StopError['stop'] } | { type: 'release'; id: string } | { type: 'recover'; id: string; runnerOwner: string }; export type WorkerReply = { id: string; attemptId: string; ok: true; text: string } | { id: string; attemptId: string; ok: false; error: string }; /** Reply to `release`: how many allocations and failed setups are still not removed after a final attempt. */ @@ -20,6 +25,7 @@ export type RecoverReply = { id: string; recovery: 'done' | 'failed'; error?: st // This worker's environment is an allowlist without credentials; the credential variables arrive as data and go // only to the adapters. const credentials: Readonly> = Object.freeze({ ...(workerData?.credentials ?? {}) }); +const feature: WorkerFeature = workerData?.feature === 'planning' ? 'planning' : 'questions'; const deps: ContainerDependencies = { buildImage: buildAgentImage, createClone: createTaskClone, @@ -32,7 +38,7 @@ const deps: ContainerDependencies = { env: credentials, }; const image: { id?: string } = {}; -const retained = new RetainedStorage(); +const retained = feature === 'planning' ? new PlanningStorage() : new RetainedStorage(); const active = new Map(); parentPort!.on('message', (message: WorkerRequest) => { @@ -44,19 +50,25 @@ parentPort!.on('message', (message: WorkerRequest) => { return; } if (message.type === 'recover') { - void recoverQuestionStorage(message.runnerOwner, deps, retained).then( + void recoverQuestionStorage(message.runnerOwner, deps, retained, feature === 'planning' ? PLANNING_NAMING : ASK_NAMING).then( () => parentPort!.postMessage({ id: message.id, recovery: 'done' } satisfies RecoverReply), (error: unknown) => parentPort!.postMessage({ id: message.id, recovery: 'failed', error: error instanceof Error ? error.message : 'Recovery failed.' } satisfies RecoverReply)); return; } - const controller = new AbortController(); - active.set(message.id, controller); + const controller = new AbortController(), id = message.id; + // A worker runs only its own feature's jobs, with that feature's leftovers and owner. + const attemptId = message.type === 'ask' ? message.question.attemptId : message.run.attemptId; + const refused = (message.type === 'ask') !== (feature === 'questions'); + active.set(id, controller); + const run = () => refused ? Promise.reject(new Error(`This worker does not run ${message.type === 'ask' ? 'questions' : 'plans'}.`)) + : message.type === 'ask' ? askInContainer(message.question, deps, controller.signal, image, retained) + : runReadOnlyAgent(PLANNING_FEATURE, message.run, deps, controller.signal, image, retained); // Defer so a cancel posted with the request is delivered before synchronous setup starts. - setImmediate(() => void askInContainer(message.question, deps, controller.signal, image, retained).then( - text => parentPort!.postMessage({ id: message.id, attemptId: message.question.attemptId, ok: true, text } satisfies WorkerReply), - (error: unknown) => parentPort!.postMessage({ id: message.id, attemptId: message.question.attemptId, ok: false, + setImmediate(() => void run().then( + text => parentPort!.postMessage({ id, attemptId, ok: true, text } satisfies WorkerReply), + (error: unknown) => parentPort!.postMessage({ id, attemptId, ok: false, error: controller.signal.aborted && controller.signal.reason instanceof Error ? controller.signal.reason.message : error instanceof Error ? error.message : 'Agent failed.' } satisfies WorkerReply), - ).finally(() => active.delete(message.id))); + ).finally(() => active.delete(id))); }); diff --git a/runner/store.ts b/runner/store.ts index bc06ef66..a9b4cf84 100644 --- a/runner/store.ts +++ b/runner/store.ts @@ -1343,6 +1343,14 @@ export class Store { askOwnerToken(file: { dev: number | bigint; ino: number | bigint }): string { return this.#ownerToken('ask_owner', "Ask is off: the stored Ask owner token is malformed. Remove the 'ask_owner' row from app_settings in the review database, then retry.", file); } + /** + * The per-database owner token planning writes as `io.codeboost.runner` (#117). It is separate from Ask's and the + * runner's, so planning's recovery never removes their agents and theirs never see planning's storage. Same rules as + * `runnerOwnerToken`. + */ + planningOwnerToken(file: { dev: number | bigint; ino: number | bigint }): string { + return this.#ownerToken('planning_owner', "Planning is off: the stored planning owner token is malformed. Remove the 'planning_owner' row from app_settings in the review database, then retry.", file); + } #ownerToken(key: string, malformed: string, file: { dev: number | bigint; ino: number | bigint }): string { const identity = { dev: String(file.dev), ino: String(file.ino) }; return this.#transaction(() => { diff --git a/test/agent-planning.test.ts b/test/agent-planning.test.ts new file mode 100644 index 00000000..6fb0fe1e --- /dev/null +++ b/test/agent-planning.test.ts @@ -0,0 +1,46 @@ +import { spawnSync } from 'node:child_process'; +import { randomUUID } from 'node:crypto'; +import { existsSync, mkdtempSync, realpathSync, rmSync, statSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { afterAll, describe, expect, it } from 'vitest'; +import { createDemo } from '../scripts/demo.ts'; +import { PlanningAgent } from '../runner/planning.ts'; +import { ReviewService } from '../runner/review.ts'; +import { prepareRecording } from './fixtures/planning/recording-inputs.ts'; + +// Real Docker (#117): the production planning agent starts planning's worker, which builds the image, clones the +// review's head, allocates bounded storage and runs Claude in the "planning" phase. Runs with the other Docker suites, +// one file at a time. +const roots: string[] = [], services: ReviewService[] = [], owners: string[] = []; +const docker = (...args: string[]) => spawnSync('docker', args, { encoding: 'utf8' }); +afterAll(() => { + for (const owner of owners) for (const kind of ['container', 'volume', 'network']) { + const names = docker(kind, 'ls', ...(kind === 'container' ? ['-a'] : []), '--quiet', '--filter', `label=io.codeboost.runner=${owner}`).stdout.split('\n').filter(Boolean); + if (names.length) docker(kind, 'rm', ...(kind === 'volume' ? [] : ['--force']), ...names); + } + services.forEach(service => service.close()); + for (const root of roots) rmSync(root, { recursive: true, force: true }); +}, 120_000); + +describe('planning in the agent container', () => { + it('reaches the vendor from inside the container (fake Claude token)', async () => { + const root = mkdtempSync(join(tmpdir(), 'agent-planning-')); roots.push(root); + const service = new ReviewService(createDemo(join(root, 'demo'))); services.push(service); + service.store.setQuestionProvider('claude'); + owners.push(service.store.planningOwnerToken(statSync(realpathSync(service.config.database), { bigint: true }))); + const snapshot = service.store.getSnapshot(service.config.identity); + // A real E2 planning request: its schema is the one lane D passes to Claude as --json-schema. + const request = { ...prepareRecording('draft', snapshot.head, randomUUID()).request, identity: service.config.identity }; + const agent = new PlanningAgent(service, { env: { CLAUDE_CODE_OAUTH_TOKEN: 'codeboost-invalid-test-token' } }); + try { + // Only a request that left the container through the vendor proxy can come back with Anthropic's 401. + await expect(agent.invoke(request, new AbortController().signal)).rejects.toThrow(/Claude could not write the plan.*(401|authenticate)/); + } finally { await agent.close(); } + // Closed cleanly: no Docker object carries planning's owner, and no planning root stays recorded. + const owner = owners.at(-1)!; + for (const kind of ['container', 'volume', 'network']) + expect(docker(kind, 'ls', ...(kind === 'container' ? ['-a'] : []), '--quiet', '--filter', `label=io.codeboost.runner=${owner}`).stdout.trim(), kind).toBe(''); + expect(existsSync(`${realpathSync(service.config.database)}.planning-leftovers.json`)).toBe(false); + }, 11 * 60_000); +}); diff --git a/test/fixtures/question-worker-stub.ts b/test/fixtures/question-worker-stub.ts index 286244df..386f27d6 100644 --- a/test/fixtures/question-worker-stub.ts +++ b/test/fixtures/question-worker-stub.ts @@ -38,7 +38,9 @@ parentPort!.on('message', (message: WorkerRequest) => { } return; } - const { prompt, provider, noteId, attemptId } = message.question; + // A planning job carries a ReadOnlyRun; the stub reads the same fields from either kind (#117). + const job = message.type === 'ask' ? message.question : { ...message.run, noteId: message.run.taskId }; + const { prompt, provider, noteId, attemptId } = job; if (prompt === 'crash') throw new Error('stub crashed'); if (prompt === 'leak') { remaining++; @@ -47,7 +49,13 @@ parentPort!.on('message', (message: WorkerRequest) => { } // Reports the owners this worker recovered, and the owner the question carries. if (prompt === 'recoveries') { - parentPort!.postMessage({ id: message.id, attemptId, ok: true, text: JSON.stringify({ recovered, owner: message.question.runnerOwner }) }); + parentPort!.postMessage({ id: message.id, attemptId, ok: true, text: JSON.stringify({ recovered, owner: job.runnerOwner }) }); + return; + } + // Reports which kind of job reached which feature's worker, and the job's owner and deadline (#117). + if (prompt === 'whoami') { + parentPort!.postMessage({ id: message.id, attemptId, ok: true, text: JSON.stringify({ kind: message.type, feature: workerData?.feature ?? null, + owner: job.runnerOwner, deadline: 'deadline' in job ? job.deadline : null, env: process.env.TMPDIR ?? null }) }); return; } // Never replies, like a question whose lane D cleanup does not settle. diff --git a/test/planning-production.test.ts b/test/planning-production.test.ts new file mode 100644 index 00000000..bf94dca4 --- /dev/null +++ b/test/planning-production.test.ts @@ -0,0 +1,179 @@ +import { randomUUID } from 'node:crypto'; +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 { AuthorProvider, AuthorRequest } from '../core/planning-author.ts'; +import type { IssueText } from '../github/issues.ts'; +import { createDemo } from '../scripts/demo.ts'; +import type { PlanningAgent } from '../runner/planning.ts'; +import { ReviewService, type ReviewConfig } from '../runner/review.ts'; +import { ISSUE_READ_TIMEOUT_MS, productionPlanning } from '../web/planning.ts'; +import { DatabaseSync } from 'node:sqlite'; +import { PLANNING_BUDGET_MS } from '../runner/planning-provider.ts'; +import { PLANNING_SHUTDOWN_GRACE_MS, startServer, type PlanningDeps } from '../web/server.ts'; + +// Production planning wiring (#117): when it is on, what each request is told, and how the server awaits the issue. +vi.setConfig({ testTimeout: 20_000 }); +const roots: string[] = [], closers: (() => Promise | void)[] = []; +afterEach(async () => { + for (const close of closers.splice(0).reverse()) await close(); + roots.splice(0).forEach(root => rmSync(root, { recursive: true, force: true })); +}); +function demo(): ReviewConfig { + const root = mkdtempSync(join(tmpdir(), 'codeboost-planning-prod-')); roots.push(root); + return createDemo(join(root, 'demo')); +} +/** The demo as a production review of a GitHub issue; the server never contacts GitHub in these tests. */ +function production(config = demo()): ReviewConfig { + const service = new ReviewService(config), issue = service.store.getPlan(config.identity).issue; service.close(); + return { ...config, demo: false, github: { repository: 'acme/retry-service', pullRequest: 7, issue } }; +} +const text = (number: number): IssueText => ({ number, title: 'Retries ignore the cap', body: 'Body with .', comments: ['Collaborator note.'] }); +const closable = (close = vi.fn(async () => undefined)) => ({ close, invoke: vi.fn() }) as unknown as PlanningAgent; + +it('is off in a demo and without a github block, so neither plans', () => { + const config = demo(); + expect(productionPlanning(config)).toBeUndefined(); + expect(productionPlanning({ ...config, demo: false })).toBeUndefined(); + expect(productionPlanning(production(config))).toBeTypeOf('function'); +}); + +it('tells each request the GitHub issue, the repository and the base commit, read with a bound', async () => { + const config = production(), service = new ReviewService(config); closers.push(() => service.close()); + const issueText = vi.fn(async (number: number) => text(number)), signal = new AbortController().signal; + const deps = productionPlanning(config, { issues: { issueText }, agent: () => closable() })!(service); + expect(await deps.describe(signal)).toEqual({ issue: text(config.github!.issue), approvedLessons: [], + repo: { name: 'acme/retry-service', baseRef: service.store.getSnapshot(config.identity).base } }); + expect(issueText).toHaveBeenCalledWith(config.github!.issue, { signal, timeoutMs: ISSUE_READ_TIMEOUT_MS }); +}); + +it('closes the planning agent when the server closes', async () => { + const config = production(), close = vi.fn(async () => undefined); + const setup = productionPlanning(config, { issues: { issueText: async number => text(number) }, agent: () => closable(close) })!; + const app = await startServer(config, 0, async () => 'answer', undefined, 2_000, undefined, undefined, setup); + await app.close(); + expect(close).toHaveBeenCalledTimes(1); +}); + +async function serve(planning: PlanningDeps) { + const config = demo(); + const app = await startServer(config, 0, async () => 'answer', undefined, 2_000, undefined, undefined, () => planning); + let closed = false; + const close = async () => { if (!closed) { closed = true; await app.close(); } }; + closers.push(close); + const api = async (method: string, path: string, body?: unknown) => { + const response = await fetch(`${new URL(app.url).origin}${path}`, { method, headers: { 'x-codeboost-token': app.token, + ...(body ? { 'content-type': 'application/json' } : {}) }, body: body ? JSON.stringify(body) : undefined }); + return { status: response.status, body: await response.json() as Record }; + }; + const view = (await api('GET', '/api/review')).body; + const start = (actionId = randomUUID()) => api('POST', '/api/plan/suggestions', { expectedRevision: view.plan.revision, + snapshotId: view.snapshot.id, feedback: '', actionId }); + /** Suggestion requests the Store recorded, read from the database itself. */ + const recorded = () => { + const db = new DatabaseSync(config.database); + try { return (db.prepare('SELECT COUNT(*) AS n FROM requests').get() as { n: number }).n; } finally { db.close(); } + }; + return { api, start, view, close, recorded }; +} +/** A provider that holds each request until it is aborted, as a running container would. */ +function holding() { + const requests: AuthorRequest[] = []; + const provider: AuthorProvider = { invoke: (request, signal) => { requests.push(request); + return new Promise((_, reject) => signal.addEventListener('abort', () => reject(signal.reason), { once: true })); } }; + return { provider, requests }; +} +const issueOf = (number: number) => ({ issue: text(number), approvedLessons: [], repo: { name: 'acme/retry-service', baseRef: 'abc' } }); + +it('awaits the issue before starting a suggestion, and sends it to the provider', async () => { + const requests: AuthorRequest[] = []; + // Holds each request open until shutdown aborts it, as a running container would. + const provider: AuthorProvider = { invoke: (request, signal) => { requests.push(request); + return new Promise((_, reject) => signal.addEventListener('abort', () => reject(signal.reason), { once: true })); } }; + let resolve!: (value: Awaited>) => void; + const described = new Promise>>(done => { resolve = done; }); + const { start, view } = await serve({ provider, describe: () => described }); + const started = start(); + await new Promise(done => setTimeout(done, 50)); + expect(requests).toHaveLength(0); + resolve({ issue: text(view.plan.issue), approvedLessons: [], repo: { name: 'acme/retry-service', baseRef: 'abc' } }); + const response = await started; + expect(response).toMatchObject({ status: 200, body: { result: { requestId: expect.any(String) } } }); + await vi.waitFor(() => expect(requests).toHaveLength(1)); + expect(requests[0]!.prompt).toContain('Retries ignore the cap'); + expect(requests[0]!.prompt).toContain('acme/retry-service'); +}); + +it('starts no suggestion when the issue cannot be read, and reports GitHub\'s failure as 502', async () => { + const invoke = vi.fn(); + const { start, recorded } = await serve({ provider: { invoke }, describe: async () => { throw new Error('GitHub is unreachable.'); } }); + const response = await start(); + expect(response).toMatchObject({ status: 502, body: { error: 'The issue could not be read from GitHub: GitHub is unreachable.' } }); + expect(invoke).not.toHaveBeenCalled(); + expect(recorded()).toBe(0); +}); + +it('replays a recorded start without reading GitHub again', async () => { + const { provider } = holding(), describe = vi.fn(); + const { start, view } = await serve({ provider, describe }); + describe.mockResolvedValueOnce(issueOf(view.plan.issue)); + const actionId = randomUUID(), first = await start(actionId); + expect(first.status).toBe(200); + describe.mockRejectedValue(new Error('GitHub is unreachable.')); + expect(await start(actionId)).toEqual(first); + expect(describe).toHaveBeenCalledTimes(1); +}); + +it('reads nothing from GitHub for a start against a stale revision', async () => { + const describe = vi.fn(); + const { api, view } = await serve({ provider: holding().provider, describe }); + const response = await api('POST', '/api/plan/suggestions', { expectedRevision: view.plan.revision + 1, snapshotId: view.snapshot.id, + feedback: '', actionId: randomUUID() }); + expect(response).toMatchObject({ status: 409, body: { error: 'Stale plan revision or snapshot. Reload before asking for suggestions.' } }); + expect(describe).not.toHaveBeenCalled(); +}); + +it('refuses a start with 503, recording nothing, when shutdown begins while the issue is read', async () => { + let resolve!: (value: ReturnType) => void; + const { start, view, close, recorded } = await serve({ provider: holding().provider, + describe: () => new Promise(done => { resolve = done; }) }); + const started = start(); + await vi.waitFor(() => expect(resolve).toBeTypeOf('function')); + const closing = close(); + resolve(issueOf(view.plan.issue)); + expect(await started).toMatchObject({ status: 503 }); + await closing; + expect(recorded()).toBe(0); +}); + +it('gives E3 the whole planning budget, not its 120-second default', async () => { + const { provider } = holding(); + const { start, view } = await serve({ provider, describe: async () => issueOf(view.plan.issue) }); + const delays: (number | undefined)[] = [], real = globalThis.setTimeout; + const spy = vi.spyOn(globalThis, 'setTimeout').mockImplementation(((fn: () => void, ms?: number, ...rest: unknown[]) => { + delays.push(ms); return real(fn, ms, ...rest); }) as typeof setTimeout); + try { expect((await start()).status).toBe(200); } finally { spy.mockRestore(); } + expect(delays).toContain(PLANNING_BUDGET_MS); + expect(delays).not.toContain(120_000); +}); + +it('ends a planning request that does not settle after shutdown aborts it by closing its worker', async () => { + // A request stuck in lane D's synchronous setup ignores its abort until its worker is abandoned. + let reject!: (error: Error) => void, issue = 0; + const order: string[] = []; + const served = await serve({ provider: { invoke: () => new Promise((_, fail) => { reject = fail; }) }, + describe: async () => issueOf(issue), close: async () => { order.push('planning closed'); reject(new Error('worker abandoned')); } }); + issue = served.view.plan.issue; + expect((await served.start()).status).toBe(200); + vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] }); + try { + const closing = served.close().then(() => order.push('server closed')); + await vi.advanceTimersByTimeAsync(PLANNING_SHUTDOWN_GRACE_MS - 1); + expect(order).toEqual([]); + await vi.advanceTimersByTimeAsync(1); + vi.useRealTimers(); + await closing; + expect(order).toEqual(['planning closed', 'server closed']); + } finally { vi.useRealTimers(); } +}); diff --git a/test/planning-worker.test.ts b/test/planning-worker.test.ts new file mode 100644 index 00000000..1cb6b9e0 --- /dev/null +++ b/test/planning-worker.test.ts @@ -0,0 +1,157 @@ +import { randomUUID } from 'node:crypto'; +import { existsSync, mkdtempSync, realpathSync, rmSync, statSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { basename, dirname, join } from 'node:path'; +import { Worker } from 'node:worker_threads'; +import { afterEach, expect, it, vi } from 'vitest'; +import type { AuthorRequest } from '../core/planning-author.ts'; +import { createDemo } from '../scripts/demo.ts'; +import { AgentWorker } from '../runner/question-agent.ts'; +import { ASK_NAMING, createWorkerRoot, LeftoverLedger, PLANNING_NAMING } from '../runner/question-leftovers.ts'; +import { CODEX_PLANNING_REFUSED, PLANNING_BUDGET_MS } from '../runner/planning-provider.ts'; +import { NO_PLANNING_AGENT, PlanningAgent } from '../runner/planning.ts'; +import { ReviewService } from '../runner/review.ts'; + +// Planning's worker, ledger and owner (#117), driven through the stub worker, which reports what reached it. +vi.setConfig({ testTimeout: 30_000 }); +const STUB = new URL('./fixtures/question-worker-stub.ts', import.meta.url); +const roots: string[] = [], services: ReviewService[] = [], closers: (() => Promise)[] = []; +afterEach(async () => { + for (const close of closers.splice(0).reverse()) await close(); + services.splice(0).forEach(service => service.close()); + roots.splice(0).forEach(root => rmSync(root, { recursive: true, force: true })); + vi.restoreAllMocks(); +}); +function review() { + const root = mkdtempSync(join(tmpdir(), 'codeboost-planning-worker-')); roots.push(root); + const service = new ReviewService(createDemo(join(root, 'demo'))); services.push(service); + return service; +} +const fileOf = (service: ReviewService) => statSync(realpathSync(service.config.database), { bigint: true }); +function request(service: ReviewService, prompt = 'whoami'): AuthorRequest { + const plan = service.store.getPlan(service.config.identity); + return { mode: 'suggest', phase: 'planning', access: 'read-only', identity: service.config.identity, requestId: randomUUID(), + issue: plan.issue, revision: plan.revision, prompt, schemaText: '{"type":"object"}' }; +} +function agent(service: ReviewService) { + const planning = new PlanningAgent(service, { url: STUB, env: { CLAUDE_CODE_OAUTH_TOKEN: 'test-token' } }); + closers.push(() => planning.close()); + return planning; +} +type WhoAmI = { kind: string; feature: string | null; owner: string; deadline: number | null; env: string | null }; + +it('gives planning its own owner token per database, apart from Ask\'s and the runner\'s', () => { + const service = review(), file = fileOf(service), owner = service.store.planningOwnerToken(file); + expect(owner).toMatch(/^[0-9a-f]{32}$/); + expect(service.store.planningOwnerToken(file)).toBe(owner); + expect(owner).not.toBe(service.store.askOwnerToken(file)); + expect(owner).not.toBe(service.store.runnerOwnerToken(file)); +}); + +it('keeps planning\'s record and lock apart from Ask\'s, so each feature locks only itself', () => { + const service = review(), database = realpathSync(service.config.database); + const ask = LeftoverLedger.forDatabase(database), planning = LeftoverLedger.forDatabase(database, PLANNING_NAMING); + expect(planning.path).toBe(`${database}.planning-leftovers.json`); + expect(basename(planning.lockPath)).toMatch(/^codeboost-planlock-\d+-\d+\.sqlite$/); + expect(basename(dirname(planning.lockPath))).toMatch(/^codeboost-planlocks-/); + expect(planning.lockPath).not.toBe(ask.lockPath); + ask.acquire(); planning.acquire(); + try { + expect(() => LeftoverLedger.forDatabase(database, PLANNING_NAMING).acquire()) + .toThrow(/^Planning is off: another codeboost process is running planning for this review\. Stop it, then retry\.$/); + } finally { planning.release(); ask.release(); } +}); + +it('never deletes Ask\'s folders when planning clears its own', () => { + const service = review(), database = realpathSync(service.config.database); + // An Ask root stamped with an Ask lock nobody holds is an orphan to Ask, and nothing to planning. + const askLock = join(tmpdir(), `codeboost-asklocks-${process.getuid?.() ?? 'user'}`, 'codeboost-asklock-999999-999999.sqlite'); + const askRoot = createWorkerRoot(askLock, ASK_NAMING); roots.push(askRoot); + const planningRoot = createWorkerRoot('', PLANNING_NAMING); roots.push(planningRoot); + writeFileSync(`${database}.planning-leftovers.json`, JSON.stringify({ roots: [planningRoot] })); + LeftoverLedger.forDatabase(database, PLANNING_NAMING).assertClear(); + expect(existsSync(planningRoot)).toBe(false); + expect(existsSync(askRoot)).toBe(true); + expect(existsSync(`${database}.planning-leftovers.json`)).toBe(false); + // Ask's own check does delete it, so the folder above was one a careless planning check could have deleted. + LeftoverLedger.forDatabase(database).assertClear(); + expect(existsSync(askRoot)).toBe(false); +}); + +it('never deletes planning\'s folders when Ask clears its own', () => { + const service = review(), database = realpathSync(service.config.database); + const planningLock = join(tmpdir(), `codeboost-planlocks-${process.getuid?.() ?? 'user'}`, 'codeboost-planlock-999999-999999.sqlite'); + const planningRoot = createWorkerRoot(planningLock, PLANNING_NAMING); roots.push(planningRoot); + LeftoverLedger.forDatabase(database).assertClear(); + expect(existsSync(planningRoot)).toBe(true); + LeftoverLedger.forDatabase(database, PLANNING_NAMING).assertClear(); + expect(existsSync(planningRoot)).toBe(false); +}); + +it.each([ + ['planning', 'ask', 'This worker does not run questions.'], + ['questions', 'plan', 'This worker does not run plans.'], +] as const)('refuses, in the worker itself, a job for another feature: a %s worker given %s', async (feature, type, error) => { + // The production worker script, with no credentials: the refusal comes before any Docker or Git work. + const worker = new Worker(new URL('../runner/question-worker.ts', import.meta.url), { workerData: { feature, credentials: {} } }); + try { + const id = randomUUID(), attemptId = randomUUID(); + const job = type === 'ask' + ? { type, id, question: { attemptId, provider: 'claude', prompt: 'p', repository: '/nonexistent', head: 'a'.repeat(40) } } + : { type, id, run: { attemptId, provider: 'claude', prompt: 'p', repository: '/nonexistent', head: 'a'.repeat(40) } }; + const reply = new Promise(resolve => worker.on('message', resolve)); + worker.postMessage(job); + expect(await reply).toEqual({ id, attemptId, ok: false, error }); + } finally { await worker.terminate(); } +}); + +it('refuses a worker whose ledger serves another feature, and plans only on planning\'s worker', async () => { + const service = review(); + expect(() => new AgentWorker(STUB, LeftoverLedger.forDatabase(service.config.database), { naming: PLANNING_NAMING })) + .toThrow('A worker and its ledger must serve the same feature.'); + const ask = new AgentWorker(STUB, undefined, { env: { CLAUDE_CODE_OAUTH_TOKEN: 'test-token' } }); + closers.push(() => ask.close()); + await expect(ask.plan({ provider: 'claude', repository: '/repo', head: 'a'.repeat(40), attemptId: randomUUID(), taskId: 't', + deadline: Date.now() + 60_000, prompt: 'whoami', schemaText: '{}', context: { snapshotId: 's', planId: 'p', planRevision: 1, + assignmentId: 'a', referencedCodeHash: 'a'.repeat(40), stateVersion: 0 } }, new AbortController().signal)) + .rejects.toThrow('This worker does not run plans.'); +}); + +it('runs a request in planning\'s worker, under a planning root, with the planning owner and a deadline inside the budget', async () => { + const service = review(); + service.store.setQuestionProvider('claude'); + const before = Date.now(); + const reply = JSON.parse(await agent(service).invoke(request(service), new AbortController().signal)) as WhoAmI; + expect(reply).toMatchObject({ kind: 'plan', feature: 'planning', owner: service.store.planningOwnerToken(fileOf(service)) }); + expect(basename(reply.env!)).toMatch(/^codeboost-plan-[A-Za-z0-9]{6}$/); + // The worker's deadline leaves its settle margin before E3's timer, which uses the whole budget. + expect(reply.deadline).toBeGreaterThanOrEqual(before + PLANNING_BUDGET_MS - 5_000); + expect(reply.deadline).toBeLessThanOrEqual(Date.now() + PLANNING_BUDGET_MS - 5_000); +}); + +it('recovers only planning\'s owner before its first request', async () => { + const service = review(); + service.store.setQuestionProvider('claude'); + const reply = JSON.parse(await agent(service).invoke(request(service, 'recoveries'), new AbortController().signal)) as + { recovered: string[]; owner: string }; + const file = fileOf(service); + expect(reply.recovered).toEqual([service.store.planningOwnerToken(file)]); + expect(reply.recovered).not.toContain(service.store.askOwnerToken(file)); +}); + +it.each([ + ['no agent is chosen', null, NO_PLANNING_AGENT], + ['Codex is chosen (#93)', 'codex', CODEX_PLANNING_REFUSED], +] as const)('refuses a request before any lock, worker or Docker work when %s', async (_, vendor, message) => { + const service = review(); + if (vendor) { + // The Store refuses Codex since #75; a database from before then still holds it. + const { DatabaseSync } = await import('node:sqlite'); + const db = new DatabaseSync(service.config.database); + try { db.prepare("INSERT INTO app_settings VALUES ('question_provider',?) ON CONFLICT(key) DO UPDATE SET value=excluded.value").run(vendor); } + finally { db.close(); } + } + const acquire = vi.spyOn(LeftoverLedger.prototype, 'acquire'); + await expect(agent(service).invoke(request(service), new AbortController().signal)).rejects.toThrow(message); + expect(acquire).not.toHaveBeenCalled(); +}); diff --git a/web/cli.ts b/web/cli.ts index 135421f2..faf113be 100644 --- a/web/cli.ts +++ b/web/cli.ts @@ -2,6 +2,7 @@ import { readFileSync } from 'node:fs'; import { resolve } from 'node:path'; import { parseArgs } from 'node:util'; import { startServer, type RunnerSetup } from './server.ts'; +import { productionPlanning } from './planning.ts'; import { createDemo } from '../scripts/demo.ts'; import { Store, requireSupportedNode } from '../runner/store.ts'; import { acquireRunnerLock, releasePreparation } from '../runner/recovery.ts'; @@ -12,7 +13,7 @@ function refuse(error: unknown): never { console.error(error instanceof Error ? 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).'); + 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 and plan suggestions (Claude Code in a Docker container, like Ask); demos never merge or plan. 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(() => { @@ -59,7 +60,8 @@ if (values.help || (!values.demo && !values.config && values['release-preparatio }; for (const signal of ['SIGINT', 'SIGTERM'] as const) process.on(signal, duringStartup); let app: Awaited>; - try { app = await startServer(config, port, undefined, undefined, undefined, undefined, undefined, undefined, runnerSetup); } + // Planning (#117) is on for any non-demo review with a github block; it needs no runner block. + try { app = await startServer(config, port, undefined, undefined, undefined, undefined, undefined, productionPlanning(config), 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. diff --git a/web/planning.ts b/web/planning.ts new file mode 100644 index 00000000..cfd9f334 --- /dev/null +++ b/web/planning.ts @@ -0,0 +1,35 @@ +import { GhIssueGateway, type IssueText } from '../github/issues.ts'; +import { PlanningAgent } from '../runner/planning.ts'; +import type { ReviewConfig, ReviewService } from '../runner/review.ts'; +import type { PlanningSetup } from './server.ts'; + +/** Bounds the GitHub read of the issue before a suggestion request starts. */ +export const ISSUE_READ_TIMEOUT_MS = 30_000; + +/** + * Production planning (#117): on for a review with a github block, never in a demo. It needs no runner block, since + * planning only reads code, as Ask does. The issue text is read from GitHub for each request (collaborators' comments + * only, bounded like an execute prompt's); approved lessons stay empty until lessons exist (L1 to L4). + */ +export function productionPlanning(config: ReviewConfig, options: { + issues?: { issueText(number: number, options: { signal?: AbortSignal; timeoutMs?: number }): Promise }; + agent?: (service: ReviewService) => PlanningAgent; +} = {}): PlanningSetup | undefined { + const github = config.github; + if (config.demo || !github) return undefined; + const issues = options.issues ?? new GhIssueGateway(github.repository); + return service => { + const provider = options.agent?.(service) ?? new PlanningAgent(service); + return { + provider, + async describe(signal) { + const text = await issues.issueText(github.issue, { signal, timeoutMs: ISSUE_READ_TIMEOUT_MS }); + // No base branch name is configured yet, so the base is named by its commit. + const base = service.store.getSnapshot(config.identity).base; + return { issue: { number: text.number, title: text.title, body: text.body, comments: [...text.comments] }, + approvedLessons: [], repo: { name: github.repository, baseRef: base } }; + }, + close: () => provider.close(), + }; + }; +} diff --git a/web/server.ts b/web/server.ts index d52f907a..41c39c4a 100644 --- a/web/server.ts +++ b/web/server.ts @@ -16,12 +16,25 @@ import { demoIssueGateway } from '../scripts/demo-issues.ts'; import { IssueBoard } from './issues.ts'; import { SuggestionCoordinator, type SuggestionHandle, type SuggestionInput, type SuggestionStore } from '../core/planning-suggestions.ts'; import type { AuthorProvider } from '../core/planning-author.ts'; -/** Live planning runs only through D (G4 after #51). Until a provider is injected, starting a suggestion is refused. */ +import { PLANNING_BUDGET_MS } from '../runner/planning-provider.ts'; +export type PlanningDescription = Pick & { repo: { name: string; baseRef: string } }; +/** Live planning runs only through D (G4 after #51, #117). Until a provider is injected, starting a suggestion is refused. */ export interface PlanningDeps { provider: AuthorProvider; - /** Trusted repository, issue and approved-lesson inputs for a suggestion request. */ - describe(): Pick & { repo: { name: string; baseRef: string } }; + /** + * Trusted repository, issue and approved-lesson inputs for a suggestion request. Production reads the issue from + * GitHub, so it may wait; `signal` ends with the HTTP request. + */ + describe(signal: AbortSignal): PlanningDescription | Promise; + /** Releases what the provider owns at shutdown, after every suggestion has settled. */ + close?(): Promise; } +/** A dependency the server reads from (GitHub) failed: 502, not recorded. */ +class UpstreamFailure extends Error {} +/** How long a planning request may take to settle after shutdown aborts it, before its worker is abandoned (Ask's grace). */ +export const PLANNING_SHUTDOWN_GRACE_MS = 20_000; +/** Production planning is built after the Store opens, from the review it serves (see web/cli.ts). */ +export type PlanningSetup = (service: ReviewService) => 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; @@ -33,10 +46,11 @@ export type RunnerSetup = (service: ReviewService, capability: ShutdownCapabilit 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) { +export async function startServer(config: ReviewConfig, port = 4318, questionAgent?: QuestionAgent, mergeGateway?: MergeGateway, shutdownDrainMs = MAX_SHUTDOWN_DRAIN_MS, issueGateway?: IssueGateway, runnerDeps?: RunnerDeps, planningInput?: PlanningDeps | PlanningSetup, 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; + let planning: PlanningDeps | undefined; /** 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. @@ -60,7 +74,9 @@ export async function startServer(config: ReviewConfig, port = 4318, questionAge settleSuggestion: (identity, id, expected, outcome) => capability.run(() => store.settleSuggestion(identity, id, expected, outcome)), getSuggestions: (identity, id) => store.getSuggestions(identity, id), }; - suggestions = planning ? new SuggestionCoordinator(suggestionStore, planning.provider) : null; + planning = typeof planningInput === 'function' ? planningInput(service) : planningInput; + // One budget for a suggestion, setup included: lane D's cap, which the provider's own deadline stays inside (#117). + suggestions = planning ? new SuggestionCoordinator(suggestionStore, planning.provider, PLANNING_BUDGET_MS) : null; } catch (error) { service.close(); throw error; } if (runnerSetup) { // Startup recovery runs here, before listen: nothing is admitted until it has finished. @@ -182,11 +198,27 @@ export async function startServer(config: ReviewConfig, port = 4318, questionAge /** Handles of suggestion requests started by this process, removed once their outcome settles. */ const suggestionHandles = new Map(); const requireAction = (input: Record) => { if (input.actionId === undefined) throw new BadRequest('actionId is required for this action.'); return input.actionId as string; }; - const planningAction = (path: string, input: Record) => { + const planningAction = async (path: string, input: Record, signal: AbortSignal) => { const actionId = requireAction(input), { actionId: _omit, ...request } = input; const imported = path === '/api/plan/import', started = path === '/api/plan/suggestions'; const match = /^\/api\/plan\/suggestions\/([0-9a-f-]{36})\/(cancel|apply)$/.exec(path); const kind = imported ? 'plan-import' : started ? 'suggestion-start' : `suggestion-${match![2]}`; + // The issue comes from GitHub (#117), so it is read before the recorded action, which runs synchronously. A replay + // returns its saved outcome without reading GitHub again, and a start the recorded action would refuse anyway (no + // planning, shutdown, stale revision or snapshot) does not read it at all. + let described: PlanningDescription | undefined; + if (started) { + const saved = service.store.savedAction(identity, { actionId, kind, request }); + if (saved) return saved.response; + const plan = service.store.getPlan(identity), snapshot = service.store.getSnapshot(identity); + if (planning && suggestions && !stopping && input.expectedRevision === plan.revision && input.snapshotId === snapshot.id) { + try { described = await planning.describe(signal); } + catch (error) { + if (stopping) throw new ShuttingDownError(); + throw new UpstreamFailure(`The issue could not be read from GitHub: ${error instanceof Error ? error.message : String(error)}`); + } + } + } return service.store.userAction(identity, { actionId, kind, request }, () => { if (imported) { if (typeof input.source !== 'string' || !['json', 'yaml'].includes(input.format as string) || !Number.isSafeInteger(input.expectedRevision)) throw new BadRequest('source, format and expectedRevision are required.'); @@ -199,7 +231,9 @@ export async function startServer(config: ReviewConfig, port = 4318, questionAge const plan = service.store.getPlan(identity), snapshot = service.store.getSnapshot(identity); if (input.expectedRevision !== plan.revision || input.snapshotId !== snapshot.id) throw new GuardRefusal('Stale plan revision or snapshot. Reload before asking for suggestions.'); if (typeof input.feedback !== 'string' || input.feedback.length > 4000) throw new BadRequest('feedback must be text of 4000 characters or fewer.'); - const context = service.planContext(), described = planning.describe(); + // Read above whenever the checks before this line pass; they cannot change across the synchronous action. + if (!described) throw new GuardRefusal('Planning agent not available yet.'); + const context = service.planContext(); const handle = suggestions.start({ context, revision: plan.revision, snapshotId: snapshot.id, issue: described.issue, approvedLessons: described.approvedLessons, feedback: input.feedback, repo: { ...described.repo, baseSha: snapshot.base, paths: context.baseEntries.map(entry => entry.path) } }); suggestionHandles.set(handle.id, handle); @@ -210,7 +244,8 @@ export async function startServer(config: ReviewConfig, port = 4318, questionAge if (match![2] === 'cancel') { const handle = suggestionHandles.get(id), current = service.store.getSuggestions(identity, id); if (current.state === 'pending' && handle) { handle.cancel('Cancelled by the user.'); return { state: 'cancelling' }; } - // No handle in this process: startup recovery has already stopped any provider (planning runs only through D). + // No handle in this process: the request belonged to a process that ended, and its container's output had no + // route back to the Store (planning runs only through D, in that process's worker; runner-lifecycle.md). service.store.cancelSuggestions(identity, id, 'Cancelled by the user.'); return { state: service.store.getSuggestions(identity, id).state }; } @@ -263,7 +298,7 @@ export async function startServer(config: ReviewConfig, port = 4318, questionAge return; } if(path==='/api/runner') { json(200, { result: runnerAction(input), runner: runnerView() }); return; } - if(planningPath) { json(200, { result: planningAction(path, input) }); return; } + if(planningPath) { json(200, { result: await planningAction(path, input, requestAbort.signal) }); return; } if(path==='/api/settings') {service.store.setQuestionProvider(input.questionProvider);json(200,{questionProvider:service.store.questionProvider()});return;} if(input.action==='retry-question') { const view=service.load();if(input.token!==view.token)throw new Error('Stale review state. Refresh and retry.'); @@ -320,6 +355,7 @@ export async function startServer(config: ReviewConfig, port = 4318, questionAge } catch (error) { if (error instanceof ShuttingDownError) { json(503, { error: error.message }); return; } if (error instanceof BadRequest) { json(400, { error: error.message }); return; } + if (error instanceof UpstreamFailure) { json(502, { error: error.message }); return; } json(409, { error: error instanceof Error ? error.message : 'Review failed.' }); } finally { @@ -365,7 +401,18 @@ export async function startServer(config: ReviewConfig, port = 4318, questionAge await step(() => executor?.close()); await step(() => closing); await step(() => questions.close()); - await step(() => suggestions?.close()); + // E3 aborts every suggestion at once and waits for each to settle. A planning request gets Ask's grace to settle + // after its abort; then the worker is closed, which abandons one stuck in a synchronous Docker call and so ends it. + await step(async () => { + const settled = suggestions?.close(); + if (planning?.close) { + let timer: ReturnType | undefined; + if (settled) await Promise.race([settled, new Promise(done => { timer = setTimeout(done, PLANNING_SHUTDOWN_GRACE_MS); })]); + clearTimeout(timer); + await planning.close(); + } + await settled; + }); 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)}`);