From 081750fbbf493a960cc23c073342631ac7e6b860 Mon Sep 17 00:00:00 2001 From: mchwang Date: Sat, 26 Sep 2026 02:01:04 -0700 Subject: [PATCH] F1c: shutdown wiring and /api/runner Store write gate with a one-time shutdown capability for settling coordinators; ShuttingDownError maps to 503 and BadRequest to 400; the merge coordinator rethrows ShuttingDownError instead of swallowing it and settles irreversible merges through the capability; the runner rejects admission at step 1; admitted requests drain (AGENTS.md) and the gate closes after the drain; idle keep-alive sockets are closed during shutdown; /api/runner status and replayable cancel-attempt, retry and cancel-task. Corrects the contract's shutdown steps 1 and 3 to follow the AGENTS.md drain rule. Co-Authored-By: Claude Opus 5.5 --- docs/implementation/runner-lifecycle.md | 16 +- runner/coordinator.ts | 21 ++- runner/lifecycle.ts | 8 +- runner/merge.ts | 23 ++- runner/questions.ts | 11 +- runner/store.ts | 21 ++- test/runner-shutdown.test.ts | 239 ++++++++++++++++++++++++ web/server.ts | 73 +++++++- 8 files changed, 374 insertions(+), 38 deletions(-) create mode 100644 test/runner-shutdown.test.ts diff --git a/docs/implementation/runner-lifecycle.md b/docs/implementation/runner-lifecycle.md index 841574e..3dad348 100644 --- a/docs/implementation/runner-lifecycle.md +++ b/docs/implementation/runner-lifecycle.md @@ -25,7 +25,7 @@ 3. For attempts that can write to the task folder, the durable record becomes terminal only after the container has stopped. "Cancel" first records the reason and shows "Stopping". It does not free anything. 4. A result is saved only if a compare-and-swap succeeds: the attempt ID must still be the task's current attempt, and the captured context (including the context generation) must still be current. Late results are thrown away, but their container must still stop before its slot is freed. 5. Retry is allowed only when the last attempt is terminal on disk, nothing for that task is still running in this process, and its captured context still matches the current code, plan and assignment. Otherwise the user must start a new request. -6. Shutdown order: reject new work → drain HTTP requests (with a time limit) → abort and await request-owned work → cancel and await runner jobs → write terminal states → close storage → release the runner lock. +6. Shutdown order: reject new requests and new runner work → drain admitted HTTP requests (with a time limit) → abort what remains, close the Store write gate, and await request-owned work → cancel and await runner jobs → write terminal states → close storage → release the runner lock. 7. Each piece of user feedback becomes one append-only **feedback event**, written in the same transaction as the user action, or, for a merge, in the same transaction as the confirmed outcome. Lane J reads these events after a task closes. ## Terms used @@ -224,9 +224,9 @@ The server computes `retryable` and sends it to the UI. The UI never works it ou | Step | Action | Existing? | |---|---|---| -| 1 | Set `stopping` on the server and `closing` on every coordinator (runner, questions, suggestions, merge), in the same synchronous turn, before any active-work list is copied. New API requests get HTTP 503. Every coordinator's start method checks `closing` synchronously and throws, so a request admitted before shutdown cannot start new work after it. | server flag: yes. Coordinator barrier: **new**. Today `close()` sets only `stopping`, and `questions.close()` runs later, so a request that was still reading its body can call `questions.start()` after shutdown began. The server also rechecks `stopping` after reading a body only for `merge`, so `service.act`, `setQuestionProvider` and future planning writers can still change the `Store`. F1 adds a `stopping` check after the body is read and before **every** mutating dispatch, returning HTTP 503, and adds regressions for each writer. Two GET handlers also write: `/api/review` (`ReviewService.load()` calls `recordHistory` when HEAD moved) and `/api/merge` (queue polling records merge observations and outcomes). `/api/review` can also write after an await: `merges.displayStatus()` checks GitHub and can then record a direct merge (`finishMergeAttempt`, `runner/merge.ts`). Per-handler checks would miss paths like this, so F1 adds a **write gate on the `Store`**. Step 1 closes the gate in the same synchronous turn. After that, every `Store` write method throws "Shutting down" unless the caller passes the shutdown capability. The server hands that capability at construction to each coordinator's own settlement and `close()` code: the runner, questions (`finishAnswer` after abort), suggestions (`settleSuggestion` in E3's `close()`) and merge. Startup recovery holds it too. HTTP handlers and `ReviewService` never receive it. So after step 1, only work that is settling can write, and it can still record its terminal rows before step 7 closes the `Store`. The gate throws a distinct `ShuttingDownError`, which the server maps to HTTP 503 "The review server is shutting down.", never to the 409 used for review errors. So an admitted `/api/review` whose `load()` reaches `recordHistory` after step 1 returns 503 with no view, and the UI treats it like any other 503 during shutdown. There is no partial or read-only view. Merge reconciliation and queue polling hit the same gate. **Today both catch every error and turn it into a status** (`displayStatus()` and `#pollQueue()` in `runner/merge.ts`), which would swallow the gate error and answer HTTP 200. F1 changes each of these catch blocks to rethrow `ShuttingDownError` first, before any other handling, so these paths also end with 503. The next startup checks GitHub again, so a skipped merge observation is recovered, not lost. Regressions cover `/api/review` history, `/api/review` direct-merge reconciliation during the GitHub await, and `/api/merge` polling. | +| 1 | Set `stopping` on the server and make the runner coordinator reject admission, in the same synchronous turn, before any active-work list is copied. New API requests get HTTP 503, and new runner work throws `ShuttingDownError`. **Admitted requests are not rejected:** AGENTS.md says to drain already-admitted HTTP requests, and to recheck the flag only at irreversible boundaries. So after reading a body, the server rechecks `stopping` only for `merge`. A question admitted before shutdown still saves its note and starts its agent; `questions.close()` (step 6) then aborts it with "Server stopped". **Changed while implementing F1c:** an earlier revision of this row rejected every admitted write and closed the Store write gate here. That broke the drain rule and the existing browser regression "drains an in-flight question request before closing its agent manager", so the gate now closes in step 3. | server flag and merge recheck: yes. Runner barrier: new. | | 2 | Stop accepting connections, and wait for admitted requests up to the drain limit (at most 14.5 s, below the 15 s request timeout). | yes | -| 3 | After the drain limit, abort the signals of the remaining requests, destroy requests that are still reading a body, then await request-owned work (the merge coordinator). | yes | +| 3 | After the drain limit, abort the signals of the remaining requests, destroy requests that are still reading a body, then **close the `Store` write gate**, then await request-owned work (the merge coordinator). From here on, every `Store` write throws `ShuttingDownError` unless the caller passes the shutdown capability. The server hands that capability at construction only to coordinators' settlement and close code (the runner, questions' `finishAnswer`, merge settlement after an irreversible command, and E3's `settleSuggestion` when it is wired), and to startup recovery; HTTP handlers and `ReviewService` never receive it. So a request-path write still pending after the abort (an aborted `/api/review` whose `load()` reaches `recordHistory`, or merge reconciliation and queue polling after a GitHub await) fails. The server maps `ShuttingDownError` to HTTP 503 "The review server is shutting down.", never to the 409 used for review errors, and there is no partial view. **Today the merge coordinator catches every error in `displayStatus()` and `#pollQueue()`**, which would swallow the gate error and answer 200; F1 makes each of those catch blocks rethrow `ShuttingDownError` first. The next startup checks GitHub again, so a skipped merge observation is recovered. The server also closes idle keep-alive connections once shutdown starts and after each request finishes, so a finished request's socket cannot hold `server.close()` open until its keep-alive timeout. | abort: yes. Gate, capability, rethrow and idle-connection close: new. | | 4 | For every `pending` and `running` attempt that has an in-memory job, record the first reason `shutdown` **only if no first reason is set yet**. A job already marked `cancelled`, `stale` or `time-limit` keeps its reason. For `running`, call `cancel('shutdown')` and await `settled`. For `pending`, await its preparation promise (and the D start call if it is in progress), remove only its host-side preparation files (task storage waits for the terminal write, as in step 5), and cancel any handle that start returned, then await its `settled` (the "Launch" table). F does not abandon a job after a timer (decision 4). **This is not yet guaranteed to end:** D4's supervisor escalates to a forced kill, but it retries unfinished container, network or setup cleanup every second with no limit (`agents/adapters/supervisor.ts`). If Docker is unreachable, `settled` never resolves and shutdown waits with the `Store` open. See the prerequisite below. | new | | 5 | For each attempt from step 4, in the order set by the task-storage holder row: the partial-output export (writable attempts), then the terminal state from its first reason (a `shutdown` reason gives `cancelled` "Stopped by shutdown") with its `diagnostic_ref`, then `removeTaskFilesystems`, and only then free its slot. A failed write leaves the row non-terminal, and a failed removal keeps the slot held. Startup recovery handles both. | new | | 6 | Await server closure; await the question and suggestion coordinators' `close()`. | yes (questions); suggestions: new wiring | @@ -425,7 +425,7 @@ Each case needs a test that fails before the fix and passes after it. Each test | Submit, then the user edits, then the response returns | Feedback submitted → user types more → response arrives → draft kept | | Code reassigned or snapshot changed, then retry or render | Snapshot advances during a run → result discarded → row `stale` with the cause → retry disabled; "run again" allowed | | Terminal write fails after settlement (review finding) | `settled` resolves → the terminal write throws → slot and unresolved marker remain → retry and admission refused → restart recovery makes the row terminal | -| Request admitted, then shutdown, then the request tries to start work (review finding) | Question body half-sent → shutdown sets `closing` → body completes → `questions.start()` throws → no answer attempt is saved | +| Request admitted, then shutdown, then the request completes (AGENTS.md drain rule) | Question body half-sent → shutdown → body completes during the drain → the note is saved and its agent starts → `questions.close()` aborts it → answer `failed` "Server stopped" (existing browser regression). A change note admitted the same way is saved | | Merge confirmed, then the event write is lost (review finding) | `merged` recorded without `task-closed` → restart → reconciliation inserts exactly one event → a second restart inserts none | | Attempt's own transitions, then publish (review round 2) | Admit → `pending` → `running` (the state version rises each time) → with no context change, a valid result publishes `completed`; the context generation is unchanged | | Two admissions race for one slot (review round 2) | Two admissions in the same tick for different tasks with one free slot → exactly one reserves and saves `pending`; a refused `Store` transaction releases its reservation | @@ -446,11 +446,11 @@ Each case needs a test that fails before the fix and passes after it. Each test | Stop during launch, then the start throws (review round 9) | User cancels while preparation finishes → D start throws → row `cancelled`, not `failed` | | Reason write fails (review round 9) | The first-reason write throws → settlement → the terminal write stores `cancelled` with the reason from memory | | Time limit, then crash (review round 9) | `time-limit` saved → process killed → restart → row `cancelled`, task `needs human`, retry refused | -| Admitted write after shutdown (review round 9) | For each of `act`, `setQuestionProvider` and the planning writers: body half-sent → shutdown → body completes → HTTP 503 and no `Store` change | +| Request still reading after the drain limit (review round 9, revised) | Body half-sent → shutdown → the drain limit passes → the request is destroyed → no `Store` change. A request arriving after shutdown began gets 503 | | Stale, then shutdown (review round 9) | Attempt marked `stale` → shutdown → row `stale`, not requeued | | Copied database (review round 9) | Copy the database file → start the copy → new token; recovery for the copy leaves the original's resources alone | | Review edit during the final merge check (review round 10) | Merge passes its GitHub checks → approval changed before the final re-read → merge refused | -| Writing GET after shutdown (review round 10) | `/api/review` with a moved HEAD, and `/api/merge` with a queue result, both admitted → shutdown → neither writes to the `Store` | +| Writing GET after the gate closes (review round 10, revised) | `/api/review` with a moved HEAD, and `/api/merge` with a queue result, whose writes arrive after step 3 → `ShuttingDownError` → 503, and neither writes to the `Store` | | Legacy resource present (review round 10) | An agent container without a runner label exists → startup refuses admission and lists it | | Lost response, then replay (review round 10) | Add a note → the transaction commits → the response is lost → resend with the same `actionId` → one note, one event, same response | | v5 upgrade (review round 10) | Migrate a v5 database with an open and a merged plan → two task rows with the backfill values; the merged one has one `task-closed` | @@ -464,14 +464,14 @@ Each case needs a test that fails before the fix and passes after it. Each test | Refused action, then replay (review round 12) | Retry refused with 409 → the response is lost → the state changes → replay with the same `actionId` → the same 409 comes back and nothing is applied | | Reject reopens the task (review round 12) | Reject with feedback → next revision, affected items marked, task `queued`, one `reject` event, no `task-closed` | | Admit work on a closed task (review round 13) | Task `cancelled` → a fresh (not retry) admission → refused, no attempt row | -| Question settling after the gate closes (review round 13) | Question running → shutdown → the abort path's `finishAnswer` writes through the capability → row `failed` with the shutdown reason; `questions.close()` resolves | +| Question settling after the gate closes (review round 13) | Question running → shutdown → the gate closes in step 3 → the abort path's `finishAnswer` writes through the capability → row `failed` with the shutdown reason; `questions.close()` resolves | | Hard-link alias (review round 13) | Hard-link the database to a second name → start through either name → refused | | First start with no database (review round 13) | Start with a missing database path → lock created → database created → identity written into the lock → a concurrent second start exits at step 1 | | Preparation child ignores SIGTERM (review round 14) | Clone subprocess ignores `SIGTERM` → cancel → `SIGKILL` after the grace period → `close` awaited → row `cancelled` → slot freed; shutdown completes | | Cancel task during a merge (review round 14) | Merge attempt `queued` → cancel task → refused; after the merge is recorded as `merged`, the task is `merged` with one `task-closed` | | Retry after queue removal (review round 14) | Merge attempt `removed` with `requiresFreshReview` → F's retry endpoint does not offer or accept a merge retry | | Cancel task while an attempt runs (review round 15) | Attempt running → cancel task → the task stays open with `cancel_requested`, "Stopping, then cancelling" → the provider returns a valid result → not published → one transaction: row `cancelled`, task `cancelled`, one `task-closed` → then `removeTaskFilesystems` → then the slot is freed | -| Review load hits the gate (review round 15) | `/api/review` admitted with a moved HEAD → shutdown → `recordHistory` throws `ShuttingDownError` → HTTP 503, not 409 | +| Review load hits the gate (review round 15) | `/api/review` whose `recordHistory` runs after the gate closed → `ShuttingDownError` → HTTP 503, not 409 | | Crash during preparation (review round 15) | Clone half-written in the attempt directory → process killed → restart → diagnostic saved, directory removed, an unknown sibling directory left and reported | | Cancel task recorded after the final await (review round 16) | Result validated → cancel task commits `cancel_requested` → publication transaction refuses `completed` → row `cancelled`, task `cancelled` | | Crash with a live clone child (review round 16) | Clone child running → runner killed → restart → recovery kills and awaits the saved process group before capturing and removing the directory | diff --git a/runner/coordinator.ts b/runner/coordinator.ts index ee46c7b..787641e 100644 --- a/runner/coordinator.ts +++ b/runner/coordinator.ts @@ -1,7 +1,7 @@ import { identityKey, type PlanIdentity } from '../core/identity.ts'; import { captureInvocation, type InvocationHandle, type InvocationInput, type InvocationResult, type StopReason, type TaskClone } from '../agents/contract.ts'; import type { AttemptRecord, Store } from './store.ts'; -import { ATTEMPT_PHASES, GuardRefusal, WRITABLE_KINDS, bounded, sameContext, type AttemptKind, type Classification, type FirstReason } from './lifecycle.ts'; +import { ATTEMPT_PHASES, GuardRefusal, ShuttingDownError, WRITABLE_KINDS, bounded, sameContext, type AttemptKind, type Classification, type FirstReason, type ShutdownCapability } from './lifecycle.ts'; /** What F's host-side preparation hands to D's start call. */ export interface PreparedAttempt { @@ -53,10 +53,15 @@ export class RunnerCoordinator { #store: Store; #deps: RunnerDeps; #limits: SlotLimits; #jobs = new Map(); #markers = new Map(); #closing = false; - constructor(store: Store, deps: RunnerDeps, limits: SlotLimits = { writable: 1, readOnly: 1 }) { + /** Settlement writes run with the shutdown capability, so they still land after the write gate closes. */ + #write: (fn: () => T) => T; + constructor(store: Store, deps: RunnerDeps, limits: SlotLimits = { writable: 1, readOnly: 1 }, capability?: ShutdownCapability) { if (![limits.writable, limits.readOnly].every(n => Number.isSafeInteger(n) && n >= 1)) throw new Error('Slot limits must be positive integers.'); this.#store = store; this.#deps = deps; this.#limits = limits; + this.#write = capability ? fn => capability.run(fn) : fn => fn(); } + /** Shutdown step 1: reject admission synchronously, in the same turn as the server flag and the Store gate. */ + rejectAdmission(): void { this.#closing = true; } get closing(): boolean { return this.#closing; } #now(): number { return this.#deps.now?.() ?? Date.now(); } #used(group: Group): number { @@ -70,7 +75,7 @@ export class RunnerCoordinator { * a refused transaction releases the reservation in the same turn. */ start(identity: PlanIdentity, request: StartRequest): AttemptRecord { - if (this.#closing) throw new GuardRefusal('The runner is shutting down.'); + if (this.#closing) throw new ShuttingDownError(); const key = identityKey(identity); if (this.#jobs.has(key)) throw new GuardRefusal('An attempt is already active for this task.'); const marker = this.#markers.get(key); @@ -85,7 +90,8 @@ export class RunnerCoordinator { catch (error) { this.#jobs.delete(key); throw error; } job.attemptId = attempt.id; job.attempt = attempt; this.#arm(job, attempt); - job.done = this.#run(job, attempt).catch(error => this.#unexpected(job, error)); + // Start after the caller's transaction commits: a rolled-back admission must not leave a job running. + job.done = Promise.resolve().then(() => this.#run(job, attempt)).catch(error => this.#unexpected(job, error)); return attempt; } /** Retry is a new attempt bound to the latest failed or cancelled one. */ @@ -131,7 +137,7 @@ export class RunnerCoordinator { if (job.firstReason) { job.handle?.cancel(D_REASON[job.firstReason]); return false; } job.firstReason = reason; try { - if (job.attemptId && !this.#store.recordFirstReason(job.identity, job.attemptId, reason)) { + if (job.attemptId && !this.#write(() => this.#store.recordFirstReason(job.identity, job.attemptId, reason))) { // Another writer (for example cancel task) recorded a reason first; adopt the durable one. const durable = this.#store.getAttempt(job.identity, job.attemptId).firstReason; if (durable) job.firstReason = durable; @@ -158,6 +164,7 @@ export class RunnerCoordinator { } async #run(job: Job, attempt: AttemptRecord): Promise { try { + if (!this.#store.getAttempts(job.identity).some(row => row.id === attempt.id)) return; // admission was rolled back let prepared: PreparedAttempt; try { prepared = await this.#deps.prepare(attempt, job.controller.signal); } catch (error) { return await this.#endBeforeLaunch(job, attempt, this.#preparationDetail(job, error)); } @@ -177,7 +184,7 @@ export class RunnerCoordinator { } catch (error) { return await this.#endBeforeLaunch(job, attempt, { detail: `Launch failed: ${message(error)}` }); } job.handle = handle; let running: boolean | undefined; - try { running = this.#store.markRunning(job.identity, attempt.id); } catch { running = undefined; } + try { running = this.#write(() => this.#store.markRunning(job.identity, attempt.id)); } catch { running = undefined; } if (running === undefined) { // A storage error, not a stop: keep ownership until D settles, then hold the slot under a marker. handle.cancel('capture-failure'); @@ -214,7 +221,7 @@ export class RunnerCoordinator { } #settle(job: Job, s: { stopReason?: StopReason; exitCode: number | null; signal: string | null; valid: boolean; result?: unknown; detail?: string }): Classification | undefined { try { - return this.#store.settleAttempt(job.identity, job.attemptId, { ...s, firstReason: job.firstReason }); + return this.#write(() => this.#store.settleAttempt(job.identity, job.attemptId, { ...s, firstReason: job.firstReason })); } catch { // The row's outcome is unknown: hold the slot until startup recovery reconciles it. this.#markers.set(job.key, { group: job.group, attemptId: job.attemptId, reason: 'result-not-saved' }); diff --git a/runner/lifecycle.ts b/runner/lifecycle.ts index 91e7265..451885f 100644 --- a/runner/lifecycle.ts +++ b/runner/lifecycle.ts @@ -27,13 +27,19 @@ const UUID_V4 = /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f /** Attempt, action and allocation IDs are lowercase UUID v4s; anything else is refused before use. */ export function isUuidV4(value: unknown): value is string { return typeof value === 'string' && UUID_V4.test(value); } export function assertUuidV4(value: unknown, name: string): asserts value is string { - if (!isUuidV4(value)) throw new GuardRefusal(`${name} must be a lowercase UUID v4.`); + if (!isUuidV4(value)) throw new BadRequest(`${name} must be a lowercase UUID v4.`); } /** A guard refused the action. Refusals are definite outcomes and are recorded for replay. */ export class GuardRefusal extends Error {} +/** A malformed request, refused before any transaction and never recorded. The server maps it to HTTP 400. */ +export class BadRequest extends GuardRefusal {} /** Reusing an action ID for a different request. */ export class ActionIdReused extends GuardRefusal {} +/** A Store write after shutdown began. The server maps it to HTTP 503, never to the 409 used for review errors. */ +export class ShuttingDownError extends Error { constructor() { super('The review server is shutting down.'); } } +/** Lets settling coordinator code write after the gate closes. Only the server hands it out, and never to HTTP handlers. */ +export interface ShutdownCapability { run(fn: () => T): T } export function bounded(reason: string): string { const text = reason.trim() || 'No reason given.'; diff --git a/runner/merge.ts b/runner/merge.ts index 2d675f9..08b715c 100644 --- a/runner/merge.ts +++ b/runner/merge.ts @@ -1,4 +1,5 @@ import type { ReviewService } from './review.ts'; +import { ShuttingDownError, type ShutdownCapability } from './lifecycle.ts'; import type { MergeAttempt } from './store.ts'; import { MergeSubmissionError, type MergeGateway, type MergeQueueGateway, type MergeQueueObservation, type MergeResult, type RemoteMergeState } from '../github/merge.ts'; @@ -30,9 +31,12 @@ export class MergeCoordinator { readonly service: ReviewService; readonly gateway: MergeGateway; readonly operationTimeoutMs: number; - constructor(service: ReviewService, gateway: MergeGateway, operationTimeoutMs = 14_000) { + /** Settlement of an irreversible merge keeps its writes after the Store gate closes; request-path reconciliation does not. */ + #settle: (fn: () => T) => T; + constructor(service: ReviewService, gateway: MergeGateway, operationTimeoutMs = 14_000, capability?: ShutdownCapability) { if (!Number.isSafeInteger(operationTimeoutMs) || operationTimeoutMs < 1 || operationTimeoutMs > 14_000) throw new Error('Invalid merge operation deadline.'); this.service = service; this.gateway = gateway; this.operationTimeoutMs = operationTimeoutMs; + this.#settle = capability ? fn => capability.run(fn) : fn => fn(); } #attempt(): MergeAttempt | null { @@ -117,6 +121,7 @@ export class MergeCoordinator { async displayStatus(view = this.service.load(), signal?: AbortSignal): Promise { try { return await this.status(view, false, signal); } catch (error) { + if (error instanceof ShuttingDownError) throw error; if (signal?.aborted) throw signal.reason; return { available: true, ready: false, action: null, blockers: [{ code: 'github', message: `Could not read GitHub merge state. ${error instanceof Error ? error.message : 'Unknown error.'}` }], remote: null, queue: this.#queueStatus() }; } @@ -171,21 +176,24 @@ export class MergeCoordinator { // The enqueue command has already committed externally. A local refresh failure must not // report that action as failed; the durable submitting record is recoverable by polling. try { - if (queueAttempt.kind === 'queue') this.service.store.queueMergeAttempt(this.service.config.identity, queueAttempt.id, result.url); - else this.service.store.finishMergeAttempt(this.service.config.identity, queueAttempt.id, { state: 'merged' }); + const attempt = queueAttempt; + this.#settle(() => { + if (attempt.kind === 'queue') this.service.store.queueMergeAttempt(this.service.config.identity, attempt.id, result.url); + else this.service.store.finishMergeAttempt(this.service.config.identity, attempt.id, { state: 'merged' }); + }); } catch {} } return { status: commandStatus, result }; } catch (error) { - if (queueAttempt) try { + if (queueAttempt) try { const attempt = queueAttempt; this.#settle(() => { const message = error instanceof Error ? error.message : 'GitHub merge submission outcome is unknown.'; if (error instanceof MergeSubmissionError && error.outcome === 'refused') { - this.service.store.finishMergeAttempt(this.service.config.identity, queueAttempt.id, { + this.service.store.finishMergeAttempt(this.service.config.identity, attempt.id, { state: 'failed', reason: message, requiresFreshReview: /head (?:branch |commit )?(?:was )?(?:modified|changed)|does not match.*head|stale review/i.test(message), }); - } else this.service.store.recordMergeAttemptDiagnostic(this.service.config.identity, queueAttempt.id, message); - } catch {} + } else this.service.store.recordMergeAttemptDiagnostic(this.service.config.identity, attempt.id, message); + }); } catch {} if (signal.aborted && signal.reason instanceof Error) throw signal.reason; throw error; } @@ -221,6 +229,7 @@ export class MergeCoordinator { this.#publishQueueObservation(attempt, observation); return this.#queueStatus(); } catch (error) { + if (error instanceof ShuttingDownError) throw error; if (signal.aborted) throw signal.reason; const message = error instanceof Error ? error.message : 'Could not read the merge queue.'; if (/head changed after review/i.test(message)) { diff --git a/runner/questions.ts b/runner/questions.ts index 748df3a..91d4d1c 100644 --- a/runner/questions.ts +++ b/runner/questions.ts @@ -2,6 +2,7 @@ import { randomUUID } from 'node:crypto'; import type { ReviewService } from './review.ts'; import { cliQuestionAgent } from './question-agent.ts'; import type { ReviewNote } from './store.ts'; +import type { ShutdownCapability } from './lifecycle.ts'; export type QuestionAgent = (prompt: string, signal: AbortSignal) => Promise; export function questionPrompt(view: ReturnType, note: ReviewNote): string { let remaining = 100_000; @@ -21,7 +22,11 @@ export class Questions { private closing = false; private service: ReviewService; private agent?: QuestionAgent; - constructor(service: ReviewService, agent?: QuestionAgent) { this.service=service; this.agent=agent; } + /** Settlement writes (finishAnswer after abort) keep working after the Store write gate closes. */ + private write: (fn: () => T) => T; + constructor(service: ReviewService, agent?: QuestionAgent, capability?: ShutdownCapability) { + this.service=service; this.agent=agent; this.write = capability ? fn => capability.run(fn) : fn => fn(); + } isRunning(id: string) { return this.running.has(id); } start(id: string, view: ReturnType) { if (this.closing) throw new Error('Server is stopping. Reconnect before asking again.'); @@ -43,9 +48,9 @@ export class Questions { invocation = agent(questionPrompt(view,note),controller.signal); const text=await Promise.race([invocation,aborted]); if(typeof text!=='string'||!text.trim()||text.length>24000) throw new Error('Agent returned an empty or oversized answer.'); - this.service.store.finishAnswer(this.service.config.identity,id,attempt,{status:'complete',text:text.trim()}); + this.write(()=>this.service.store.finishAnswer(this.service.config.identity,id,attempt,{status:'complete',text:text.trim()})); } catch(error) { - this.service.store.finishAnswer(this.service.config.identity,id,attempt,{status:'failed',error:(error instanceof Error?error.message:'Agent failed.').slice(0,1000)}); + try { this.write(()=>this.service.store.finishAnswer(this.service.config.identity,id,attempt,{status:'failed',error:(error instanceof Error?error.message:'Agent failed.').slice(0,1000)})); } catch {} } finally {clearTimeout(timeout);} })(); const settled = done.finally(async () => { diff --git a/runner/store.ts b/runner/store.ts index ad8eabd..fb20574 100644 --- a/runner/store.ts +++ b/runner/store.ts @@ -6,7 +6,7 @@ import { importPlan, applySuggestion, assertEditReply, type Plan, type PlanConte import type { Approval, SegmentChoice } from '../core/approvals.ts'; import type { InvocationContext, StopReason } from '../agents/contract.ts'; import { - ATTEMPT_PHASES, CLOSED_STATUSES, DEFAULT_TASK_BUDGET_MS, FIRST_REASONS, GuardRefusal, ActionIdReused, MAX_RESULT_BYTES, TASK_STATUSES, TERMINAL_STATES, + ATTEMPT_PHASES, BadRequest, CLOSED_STATUSES, ShuttingDownError, type ShutdownCapability, DEFAULT_TASK_BUDGET_MS, FIRST_REASONS, GuardRefusal, ActionIdReused, MAX_RESULT_BYTES, TASK_STATUSES, TERMINAL_STATES, assertUuidV4, bounded, classifySettlement, requestHash, sameContext, type AttemptKind, type AttemptState, type Classification, type FirstReason, type Settlement, type TaskStatus, } from './lifecycle.ts'; @@ -115,11 +115,24 @@ export class Store { } close(): void { this.#db.close(); } #get(sql: string, ...args: SQLInputValue[]) { return this.#db.prepare(sql).get(...args); } - #run(sql: string, ...args: SQLInputValue[]) { return this.#db.prepare(sql).run(...args); } + #run(sql: string, ...args: SQLInputValue[]) { this.#checkWrite(); return this.#db.prepare(sql).run(...args); } + // Shutdown write gate (runner-lifecycle.md, "Shutdown" step 1). + #gateClosed = false; #privileged = 0; #capabilityIssued = false; + #checkWrite(): void { if (this.#gateClosed && this.#privileged === 0) throw new ShuttingDownError(); } + /** Issued once, to the server, which hands it only to coordinators' settlement and close code. */ + shutdownCapability(): ShutdownCapability { + if (this.#capabilityIssued) throw new Error('The shutdown capability was already issued.'); + this.#capabilityIssued = true; + return Object.freeze({ run: (fn: () => T): T => { this.#privileged++; try { return fn(); } finally { this.#privileged--; } } }); + } + /** Shutdown step 1: from now on, every write without the capability throws ShuttingDownError. Reads still work. */ + closeWrites(): void { this.#gateClosed = true; } + get writesClosed(): boolean { return this.#gateClosed; } #depth = 0; /** Nested calls join the outer transaction, so a user action can wrap existing Store methods atomically. */ #transaction(fn: () => T): T { if (this.#depth > 0) { this.#depth++; try { return fn(); } finally { this.#depth--; } } + this.#checkWrite(); this.#db.exec('BEGIN IMMEDIATE'); this.#depth = 1; try { const result = fn(); this.#db.exec('COMMIT'); return result; } catch (error) { this.#db.exec('ROLLBACK'); throw error; } @@ -736,8 +749,8 @@ export class Store { return { response: value, replayed: false }; }); } catch (error) { - const storage = (error as { code?: string }).code === 'ERR_SQLITE_ERROR'; - if (!replaying && !storage && !(error instanceof ActionIdReused) && this.#depth === 0) { + const storage = (error as { code?: string }).code === 'ERR_SQLITE_ERROR' || error instanceof ShuttingDownError; + if (!replaying && !storage && !(error instanceof ActionIdReused) && !(error instanceof BadRequest) && this.#depth === 0) { const message = error instanceof Error ? bounded(error.message) : 'Refused.'; this.#transaction(() => { if (!this.#get('SELECT 1 FROM user_actions WHERE plan_key=? AND action_id=?', key, action.actionId)) record({ ok: false, error: message }); }); } diff --git a/test/runner-shutdown.test.ts b/test/runner-shutdown.test.ts new file mode 100644 index 0000000..ac3db14 --- /dev/null +++ b/test/runner-shutdown.test.ts @@ -0,0 +1,239 @@ +import { mkdtempSync, rmSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { execFileSync } from 'node:child_process'; +import { request as httpRequest } from 'node:http'; +import { randomUUID } from 'node:crypto'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { createDemo } from '../scripts/demo.ts'; +import { startServer } from '../web/server.ts'; +import { Store } from '../runner/store.ts'; +import { ReviewService } from '../runner/review.ts'; +import { MergeCoordinator } from '../runner/merge.ts'; +import { ShuttingDownError } from '../runner/lifecycle.ts'; +import type { RunnerDeps } from '../runner/coordinator.ts'; +import type { InvocationResult } from '../agents/contract.ts'; +import type { MergeGateway, MergeQueueGateway, RemoteMergeState } from '../github/merge.ts'; + +const roots: string[] = []; +const cleanups: (() => Promise | void)[] = []; +afterEach(async () => { + for (const cleanup of cleanups.splice(0).reverse()) await cleanup(); + for (const root of roots.splice(0)) rmSync(root, { recursive: true, force: true }); + vi.restoreAllMocks(); +}); +const oid = (n: number) => n.toString(16).padStart(40, '0'); +function demo() { const root = mkdtempSync(join(tmpdir(), 'codeboost-shutdown-')); roots.push(root); return createDemo(join(root, 'demo')); } +type App = Awaited>; +async function serve(deps?: RunnerDeps) { + const config = demo(); + const app = await startServer(config, 0, async () => 'answer', undefined, 2_000, deps); + let closed = false; + const close = async () => { if (!closed) { closed = true; await app.close(); } }; + cleanups.push(close); + return { app, config, close }; +} +const origin = (app: App) => new URL(app.url).origin; +async function api(app: App, method: string, path: string, body?: unknown) { + const response = await fetch(`${origin(app)}${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 }; +} +/** Sends half a JSON body, lets the test start shutdown, then sends the rest. */ +function partialPost(app: App, path: string, body: unknown) { + const text = JSON.stringify(body), url = new URL(`${origin(app)}${path}`); + let finish!: () => void; + const response = new Promise<{ status: number; body: Record }>((resolve, reject) => { + const req = httpRequest({ host: url.hostname, port: url.port, path, method: 'POST', headers: { 'x-codeboost-token': app.token, 'content-type': 'application/json', 'content-length': Buffer.byteLength(text) } }, res => { + let data = ''; res.on('data', chunk => { data += chunk; }); res.on('end', () => resolve({ status: res.statusCode!, body: JSON.parse(data) })); + }); + req.on('error', reject); + req.write(text.slice(0, 5)); + finish = () => req.end(text.slice(5)); + }); + return { response, finish: () => finish() }; +} +const tick = () => new Promise(resolve => setTimeout(resolve, 20)); + +describe('Store write gate', () => { + it('refuses writes after closeWrites unless the capability runs them, and issues the capability once', () => { + const config = demo(), store = new Store(config.database); cleanups.push(() => store.close()); + const capability = store.shutdownCapability(); + expect(() => store.shutdownCapability()).toThrow(/already issued/); + store.closeWrites(); + expect(() => store.setQuestionProvider('claude')).toThrow(ShuttingDownError); + expect(() => store.transitionTask(config.identity, store.getTask(config.identity).stateVersion, 'queued')).toThrow(ShuttingDownError); + expect(store.getTask(config.identity).status).toBe('in review'); + capability.run(() => store.setQuestionProvider('claude')); + expect(store.questionProvider()).toBe('claude'); + }); + it('does not record a user action refused by the gate, so the UI may resend it', () => { + const config = demo(), store = new Store(config.database); cleanups.push(() => store.close()); + const actionId = randomUUID(); + store.closeWrites(); + expect(() => store.userAction(config.identity, { actionId, kind: 'note', request: {} }, () => 1)).toThrow(ShuttingDownError); + store.close(); cleanups.pop(); + const reopened = new Store(config.database); cleanups.push(() => reopened.close()); + expect(reopened.userAction(config.identity, { actionId, kind: 'note', request: {} }, () => 2)).toEqual({ response: 2, replayed: false }); + }); +}); + +describe('merge coordinator after the gate closes', () => { + function harness(remote: Partial) { + const identity = { repositoryId: 'repo', taskId: 'task', planId: 'plan' }; + const store = new Store(':memory:'); cleanups.push(() => store.close()); + const plan = { schema_version: 1 as const, revision: 1, issue: 1, summary: 'M', questions: [], items: [{ id: 'P1', title: 'M', intent: 'M', files: [{ path: 'a', kind: 'edit' as const, renamed_from: null, change: 'M' }], acceptance: [{ type: 'check' as const, text: 'M' }], depends_on: [] }] }; + store.createPlan(JSON.stringify(plan), 'json', { identity, issue: 1, baseEntries: [{ path: 'a', kind: 'file' }], pathKey: p => p, allowedCommands: [] }, oid(1), oid(2)); + const view = { items: [{ id: 'P1', state: 'approved', outside: [], acceptance: [], checks: {} }], plan: { revision: 1 }, segments: [], notes: [], snapshot: { id: store.getSnapshot(identity).id, base: oid(1), head: oid(2) }, token: 't', expected: { revision: 1, snapshotId: store.getSnapshot(identity).id, reviewVersion: store.reviewVersion(identity) } } as unknown as ReturnType; + const service = { store, config: { identity }, load: vi.fn(() => view) } as unknown as ReviewService; + const state: RemoteMergeState = { base: oid(1), head: oid(2), pullRequestState: 'OPEN', mergeable: 'MERGEABLE', rulesKnown: true, atomicBaseGuard: true, mergeQueue: false, requiredChecks: [], alreadyFixed: 'clear', ...remote }; + const gateway: MergeGateway & MergeQueueGateway = { + inspect: vi.fn(async () => state), merge: vi.fn(async () => ({ url: 'https://github.example/pr/1' })), + queueWatermark: vi.fn(async () => 'c'), inspectQueue: vi.fn(async () => ({ state: 'merged', reviewedHead: oid(2), mergedAt: '2026-09-26T00:00:00Z' }) as never), + }; + return { store, identity, view, service, gateway }; + } + it('rethrows ShuttingDownError from direct-merge reconciliation in displayStatus instead of returning a blocker', async () => { + const h = harness({ pullRequestState: 'MERGED' }); + h.store.beginMergeAttempt(h.identity, h.view.expected as never, oid(2), null, 'direct'); + const coordinator = new MergeCoordinator(h.service, h.gateway); + h.store.closeWrites(); + await expect(coordinator.displayStatus(h.view)).rejects.toBeInstanceOf(ShuttingDownError); + expect(h.store.getMergeAttempt(h.identity)!.state).toBe('submitting'); + }); + it('rethrows ShuttingDownError from queue polling instead of reporting a queue status', async () => { + const h = harness({ mergeQueue: true }); + const attempt = h.store.beginMergeAttempt(h.identity, h.view.expected as never, oid(2), 'c', 'queue'); + h.store.queueMergeAttempt(h.identity, attempt.id, 'https://github.example/pr/1'); + const coordinator = new MergeCoordinator(h.service, h.gateway); + h.store.closeWrites(); + await expect(coordinator.pollQueue()).rejects.toBeInstanceOf(ShuttingDownError); + expect(h.store.getMergeAttempt(h.identity)!.state).toBe('queued'); + }); + it('still records a merge that GitHub completed after the gate closed, through the capability', async () => { + const h = harness({}); + const capability = h.store.shutdownCapability(); + (h.gateway.merge as ReturnType).mockImplementation(async () => { h.store.closeWrites(); return { url: 'https://github.example/pr/1' }; }); + await new MergeCoordinator(h.service, h.gateway, 14_000, capability).merge('t'); + expect(h.store.getMergeAttempt(h.identity)!.state).toBe('merged'); + expect(h.store.getTask(h.identity).status).toBe('merged'); + }); +}); + +describe('server shutdown', () => { + it('lets an admitted request finish during the drain (AGENTS.md: drain admitted requests)', async () => { + const { app, config, close } = await serve(); + const view = (await api(app, 'GET', '/api/review')).body; + const sent = partialPost(app, '/api/action', { action: 'note', kind: 'change', item: 'P1', text: 'admitted note', token: view.token }); + await tick(); + const closing = close(); + await tick(); + sent.finish(); + expect((await sent.response).status).toBe(200); + await closing; + const store = new Store(config.database); cleanups.push(() => store.close()); + expect(store.getReviewNotes(config.identity).filter(note => note.text === 'admitted note')).toHaveLength(1); + }); + it('destroys a request still reading its body after the drain limit, and writes nothing', async () => { + const { app, config, close } = await serve(); + const view = (await api(app, 'GET', '/api/review')).body; + const sent = partialPost(app, '/api/action', { action: 'note', kind: 'change', item: 'P1', text: 'never finished', token: view.token }); + sent.response.catch(() => undefined); + await tick(); + await close(); + await expect(sent.response).rejects.toThrow(); + const store = new Store(config.database); cleanups.push(() => store.close()); + expect(store.getReviewNotes(config.identity).filter(note => note.text === 'never finished')).toHaveLength(0); + }); + it('answers 503 to a request that arrives after shutdown began', async () => { + const { app, close } = await serve(); + const view = (await api(app, 'GET', '/api/review')).body; + const closing = close(); + const late = await fetch(`${origin(app)}/api/action`, { method: 'POST', headers: { 'x-codeboost-token': app.token, 'content-type': 'application/json' }, body: JSON.stringify({ action: 'note', kind: 'change', item: 'P1', text: 'late', token: view.token }) }).then(r => r.status, () => 'refused'); + expect([503, 'refused']).toContain(late); + await closing; + }); + it('rejects runner admission at step 1 and closes the write gate after the drain', async () => { + const { deps } = { deps: { prepare: async () => { throw new Error('unused'); }, cleanupPreparation: async () => undefined, start: () => { throw new Error('unused'); }, validate: () => null } as RunnerDeps }; + const { app, close } = await serve(deps); + const closing = close(); + // Step 1 is synchronous; the Store gate closes only after the drain (step 3). + expect(app.runner!.closing).toBe(true); + expect(app.service.store.writesClosed).toBe(false); + await closing; + expect(app.service.store.writesClosed).toBe(true); + }); + it('answers 503, not 409, when a review load hits the write gate', async () => { + const { app, config } = await serve(); + writeFileSync(join(config.repository, 'moved.txt'), 'x'); + execFileSync('git', ['-c', 'core.hooksPath=/dev/null', 'add', '-A'], { cwd: config.repository }); + execFileSync('git', ['-c', 'core.hooksPath=/dev/null', '-c', 'user.name=t', '-c', 'user.email=t@t', 'commit', '-qm', 'move head'], { cwd: config.repository }); + app.service.store.closeWrites(); + const response = await api(app, 'GET', '/api/review'); + expect(response).toEqual({ status: 503, body: { error: 'The review server is shutting down.' } }); + }); +}); + +describe('/api/runner', () => { + function fakeRunner() { + const settles: ((over?: Partial) => void)[] = []; + const deps: RunnerDeps = { + prepare: async attempt => ({ clone: { id: `c-${attempt.id}`, taskId: 't', directory: '/tmp/x', head: oid(2) }, vendor: 'claude', approvedArgv: [] }), + cleanupPreparation: async () => undefined, + start: input => { + let settle!: (r: InvocationResult) => void; + const settled = new Promise(resolve => { settle = resolve; }); + settles.push(over => settle({ attemptId: input.attemptId, context: input.context, exitCode: 0, signal: null, stdout: 'ok', stderr: '', ...over })); + return { attemptId: input.attemptId, settled, cancel: () => undefined }; + }, + validate: (_a, r) => r.stdout, + }; + return { deps, settles }; + } + it('reports the task without a runner and refuses runner-only actions', async () => { + const { app } = await serve(); + const view = (await api(app, 'GET', '/api/runner')).body; + expect(view).toMatchObject({ available: false, retryable: false, stopRequested: null, unresolved: null, task: { status: 'in review' } }); + const refused = await api(app, 'POST', '/api/runner', { action: 'retry', attemptId: randomUUID(), expectedStateVersion: view.stateVersion, actionId: randomUUID() }); + expect(refused).toMatchObject({ status: 409, body: { error: 'The runner is not available yet.' } }); + }); + it('replays cancel task by action ID, refuses a reused ID, and answers 400 for a malformed ID', async () => { + const { app } = await serve(); + const version = (await api(app, 'GET', '/api/runner')).body.stateVersion; + const actionId = randomUUID(), cancel = { action: 'cancel-task', expectedStateVersion: version, actionId }; + const first = await api(app, 'POST', '/api/runner', cancel); + expect(first).toMatchObject({ status: 200, body: { result: { outcome: 'closed' }, runner: { task: { status: 'cancelled' } } } }); + expect((await api(app, 'POST', '/api/runner', cancel)).body.result).toEqual(first.body.result); + expect((await api(app, 'POST', '/api/runner', { ...cancel, action: 'retry' })).status).toBe(409); + expect((await api(app, 'POST', '/api/runner', { ...cancel, actionId: 'not-a-uuid' })).status).toBe(400); + }); + it('cancels a running attempt, then retries it, and shutdown waits for the running retry', async () => { + const { deps, settles } = fakeRunner(); + const { app, config, close } = await serve(deps); + const store = app.service.store, identity = config.identity; + store.transitionTask(identity, store.getTask(identity).stateVersion, 'queued'); + const attempt = app.runner!.start(identity, { expectedStateVersion: store.getTask(identity).stateVersion, kind: 'review', expectedContext: store.currentContext(identity), deadline: Date.now() + 60_000 }); + for (let i = 0; i < 50 && settles.length === 0; i++) await tick(); + let view = (await api(app, 'GET', '/api/runner')).body; + expect(view).toMatchObject({ available: true, retryable: false, attempts: [{ id: attempt.id, state: 'running' }] }); + const cancel = await api(app, 'POST', '/api/runner', { action: 'cancel-attempt', attemptId: attempt.id, expectedStateVersion: view.stateVersion, actionId: randomUUID() }); + expect(cancel.body.result).toEqual({ outcome: 'stopping' }); + expect((await api(app, 'GET', '/api/runner')).body.stopRequested).toEqual({ attemptId: attempt.id, reason: 'cancelled', saved: true }); + settles[0]!({ exitCode: null, stopReason: 'cancelled' }); + await app.runner!.settled(identity); + view = (await api(app, 'GET', '/api/runner')).body; + expect(view).toMatchObject({ retryable: true, attempts: [{ state: 'cancelled' }] }); + const retryId = randomUUID(), retry = { action: 'retry', attemptId: attempt.id, expectedStateVersion: view.stateVersion, actionId: retryId }; + const started = await api(app, 'POST', '/api/runner', retry); + expect(started.body.result).toMatchObject({ outcome: 'started' }); + expect((await api(app, 'POST', '/api/runner', retry)).body.result).toEqual(started.body.result); + for (let i = 0; i < 50 && settles.length < 2; i++) await tick(); + let closed = false; + const closing = close().then(() => { closed = true; }); + await tick(); + expect(closed).toBe(false); + settles[1]!({ exitCode: null, stopReason: 'shutdown' }); + await closing; + const reopened = new Store(config.database); cleanups.push(() => reopened.close()); + expect(reopened.getAttempt(identity, started.body.result.attemptId)).toMatchObject({ state: 'cancelled', diagnostic: 'Stopped by shutdown' }); + }); +}); diff --git a/web/server.ts b/web/server.ts index 348d53b..e1c8cb0 100644 --- a/web/server.ts +++ b/web/server.ts @@ -6,21 +6,58 @@ import { ReviewService, type ReviewConfig } from '../runner/review.ts'; import { Questions, type QuestionAgent } from '../runner/questions.ts'; import { GhMergeGateway, type MergeGateway } from '../github/merge.ts'; import { MergeCoordinator } from '../runner/merge.ts'; +import { RunnerCoordinator, type RunnerDeps } from '../runner/coordinator.ts'; +import { BadRequest, GuardRefusal, ShuttingDownError, sameContext } from '../runner/lifecycle.ts'; const publicRoot = new URL('./public/', import.meta.url); -export async function startServer(config: ReviewConfig, port = 4318, questionAgent?: QuestionAgent, mergeGateway?: MergeGateway, shutdownDrainMs = 14_500) { +export async function startServer(config: ReviewConfig, port = 4318, questionAgent?: QuestionAgent, mergeGateway?: MergeGateway, shutdownDrainMs = 14_500, runnerDeps?: RunnerDeps) { if (!Number.isSafeInteger(shutdownDrainMs) || shutdownDrainMs < 1 || shutdownDrainMs > 14_500) throw new Error('Invalid shutdown drain deadline.'); const service = new ReviewService(config), token = randomBytes(32).toString('hex'); - let questions: Questions, merges: MergeCoordinator | null; + let questions: Questions, merges: MergeCoordinator | null, runner: RunnerCoordinator | null; + // Only coordinators' settlement and close code receive this; HTTP handlers never do. + const capability = service.store.shutdownCapability(); try { if (!config.demo && config.github && config.github.issue !== service.store.getPlan(config.identity).issue) throw new Error('The GitHub merge issue must match the stored plan issue.'); - questions=new Questions(service,questionAgent); - merges = !config.demo && (mergeGateway || config.github) ? new MergeCoordinator(service, mergeGateway ?? new GhMergeGateway(config.github!)) : null; + questions=new Questions(service,questionAgent,capability); + merges = !config.demo && (mergeGateway || config.github) ? new MergeCoordinator(service, mergeGateway ?? new GhMergeGateway(config.github!), 14_000, capability) : null; + // The runner starts only with an injected D; until #51 lands, runner actions report that it is unavailable. + runner = runnerDeps ? new RunnerCoordinator(service.store, runnerDeps, undefined, capability) : null; } catch (error) { service.close(); throw error; } const loadReview=()=>{const view=service.load();return {...view,notes:view.notes.map(note=>({...note,answerActive:questions.isRunning(note.id)}))};}; const load=async(signal?:AbortSignal)=>{const view=loadReview();return {...view,merge:merges?await merges.displayStatus(view,signal):{available:false}};}; const answerStatuses=()=>service.store.getReviewNotes(config.identity) .filter(note=>note.kind==='question') .map(note=>({id:note.id,answer:note.answer,answerActive:questions.isRunning(note.id)})); + const identity = config.identity; + /** Reads only task and attempt rows; never rebuilds history or the review. */ + const runnerView = () => { + const task = service.store.getTask(identity), attempts = service.store.getAttempts(identity).slice(-20); + const status = runner?.status(identity) ?? { active: false, stopRequested: null, unresolved: null }; + const last = attempts.find(attempt => attempt.id === task.currentAttemptId); + const retryable = !!runner && !!last && (last.state === 'failed' || last.state === 'cancelled') + && (task.status === 'running' || task.status === 'queued') && !task.requeuePending && task.cancelRequested === null + && !status.active && !status.unresolved && sameContext(last.context, service.store.currentContext(identity)); + return { available: !!runner, task, attempts: attempts.map(({ result, ...attempt }) => ({ ...attempt, hasResult: result !== null })), + stateVersion: task.stateVersion, retryable, stopRequested: status.stopRequested, unresolved: status.unresolved }; + }; + const runnerAction = (input: Record) => { + const { action, attemptId, expectedStateVersion, actionId } = input; + if (!['cancel-attempt', 'retry', 'cancel-task'].includes(action as string)) throw new GuardRefusal('Unsupported runner action.'); + if (!Number.isSafeInteger(expectedStateVersion)) throw new GuardRefusal('expectedStateVersion is required.'); + return service.store.userAction(identity, { actionId: actionId as string, kind: action as string, request: { attemptId, expectedStateVersion } }, () => { + if (service.store.getTask(identity).stateVersion !== expectedStateVersion) throw new GuardRefusal('Stale task state. Reload before writing.'); + if (action === 'cancel-task') return { outcome: runner ? runner.cancelTask(identity, expectedStateVersion as number, actionId as string) : service.store.cancelTask(identity, expectedStateVersion as number, actionId as string) }; + if (!runner) throw new GuardRefusal('The runner is not available yet.'); + if (typeof attemptId !== 'string') throw new GuardRefusal('attemptId is required.'); + if (action === 'cancel-attempt') { + if (!runner.stop(identity, attemptId, 'cancelled')) throw new GuardRefusal('That attempt is not running.'); + return { outcome: 'stopping' }; + } + const last = service.store.getAttempt(identity, attemptId); + const retry = runner.retry(identity, attemptId, { expectedStateVersion: expectedStateVersion as number, kind: last.kind, item: last.item, + expectedContext: service.store.currentContext(identity), deadline: Date.now() + 10 * 60_000 }); + return { outcome: 'started', attemptId: retry.id }; + }).response; + }; let stopping = false; const activeRequests=new Set<{abort:AbortController;request:IncomingMessage;readingBody:boolean}>(); const server = createServer(async (req, res) => { @@ -41,14 +78,17 @@ export async function startServer(config: ReviewConfig, port = 4318, questionAge if (req.method === 'GET' && path === '/api/questions') { json(200,{notes:answerStatuses()});return; } if (req.method === 'GET' && path === '/api/merge') { if(!merges)throw new Error('Merging is not configured for this review.');json(200,{queue:await merges.pollQueue()});return; } if (req.method === 'GET' && path === '/api/review') { json(200, await load(requestAbort.signal)); return; } - if (req.method !== 'POST' || !['/api/action','/api/settings'].includes(path) || req.headers['content-type'] !== 'application/json') { json(405, { error: 'Unsupported request.' }); return; } + if (req.method === 'GET' && path === '/api/runner') { json(200, runnerView()); return; } + if (req.method !== 'POST' || !['/api/action','/api/settings','/api/runner'].includes(path) || req.headers['content-type'] !== 'application/json') { json(405, { error: 'Unsupported request.' }); return; } const chunks: Buffer[] = []; let size = 0; activeRequest.readingBody=true; try { for await (const chunk of req) { size += chunk.length; if (size > 16384) { json(413, { error: 'Request too large.' }); return; } chunks.push(chunk); } } finally { activeRequest.readingBody=false; } const body = new TextDecoder('utf-8', { fatal: true }).decode(Buffer.concat(chunks)); const input=JSON.parse(body); + // Admitted requests drain normally (AGENTS.md); only the irreversible merge boundary rechecks the flag. if (stopping && input.action === 'merge') { json(503, { error: 'The review server is shutting down.' }); return; } + if(path==='/api/runner') { json(200, { result: runnerAction(input), runner: runnerView() }); 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.'); @@ -79,14 +119,25 @@ export async function startServer(config: ReviewConfig, port = 4318, questionAge res.writeHead(200, { 'Content-Type': 'font/woff2' }); res.end(readFileSync(fileURLToPath(new URL(`../node_modules/@fontsource/${family}/files/${font[1]}`, import.meta.url)))); return; } json(404, { error: 'Not found.' }); - } catch (error) { json(409, { error: error instanceof Error ? error.message : 'Review failed.' }); } - finally {activeRequests.delete(activeRequest);} + } catch (error) { + if (error instanceof ShuttingDownError) { json(503, { error: error.message }); return; } + if (error instanceof BadRequest) { json(400, { error: error.message }); return; } + json(409, { error: error instanceof Error ? error.message : 'Review failed.' }); + } + finally { + activeRequests.delete(activeRequest); + // During shutdown a finished request's keep-alive socket would hold server.close() open until its timeout. + if (stopping) setImmediate(() => server.closeIdleConnections()); + } }); server.requestTimeout = 15000; await new Promise((resolve, reject) => { server.once('error', reject); server.listen(port, '127.0.0.1', () => { server.removeListener('error', reject); resolve(); }); }).catch(error => { service.close(); throw error; }); const address = server.address(); if (!address || typeof address === 'string') throw new Error('Cannot determine local address.'); - return { server, service, token, url: `http://127.0.0.1:${address.port}/#${token}`, close: async () => { + return { server, service, token, runner, url: `http://127.0.0.1:${address.port}/#${token}`, close: async () => { + // Step 1, one synchronous turn: reject new API requests and new runner work. Admitted requests drain (step 2). stopping = true; + runner?.rejectAdmission(); + server.closeIdleConnections(); const closing = new Promise((resolve, reject) => server.close(error => error ? reject(error) : resolve())); let timer: ReturnType | undefined; await Promise.race([closing, new Promise(resolve => { timer=setTimeout(resolve,shutdownDrainMs); })]); @@ -96,7 +147,13 @@ export async function startServer(config: ReviewConfig, port = 4318, questionAge active.abort.abort(reason); if(active.readingBody)active.request.destroy(reason); } + // Step 3: after the drain, close the Store write gate. A request-path write still pending after the abort + // (for example merge reconciliation after a GitHub await) now fails with 503; settling coordinators keep the capability. + service.store.closeWrites(); + server.closeIdleConnections(); await merges?.close(); + // Step 4: stop runner jobs (shutdown reason only where none is set) and await settlement; no timer abandons a job. + await runner?.close(); await closing; await questions.close(); service.close();