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 <noreply@anthropic.com>

Author: mchwang

docs/implementation/runner-lifecycle.md

@@ -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 {
@@ -58,10 +58,15 @@ export class RunnerCoordinator {
   #store: Store; #deps: RunnerDeps; #limits: SlotLimits;
   #jobs = new Map<string, Job>(); #markers = new Map<string, Marker>();
   #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: <T>(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 {
@@ -75,7 +80,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);
@@ -90,7 +95,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. */
@@ -136,7 +142,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;
@@ -163,6 +169,7 @@ export class RunnerCoordinator {
   }
   async #run(job: Job, attempt: AttemptRecord): Promise<void> {
     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)); }
@@ -182,7 +189,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');
@@ -219,7 +226,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' });

runner/coordinator.ts

@@ -29,11 +29,13 @@ 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 {}
 /**
@@ -44,6 +46,10 @@ export class RefusalWithEffect extends GuardRefusal {
   readonly effect: () => void;
   constructor(message: string, effect: () => void) { super(message); this.effect = effect; }
 }
+/** 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<T>(fn: () => T): T }
 
 export function bounded(reason: string): string {
   const text = reason.trim() || 'No reason given.';

runner/lifecycle.ts

@@ -1,6 +1,6 @@
 import type { ReviewService } from './review.ts';
 import { mergeActionResponse, type MergeAttempt } from './store.ts';
-import { ActionIdReused, GuardRefusal, MERGEABLE_STATUSES, assertUuidV4 } from './lifecycle.ts';
+import { ActionIdReused, GuardRefusal, MERGEABLE_STATUSES, ShuttingDownError, assertUuidV4, type ShutdownCapability } from './lifecycle.ts';
 import { MergeSubmissionError, type MergeGateway, type MergeQueueGateway, type MergeQueueObservation, type MergeResult, type RemoteMergeState } from '../github/merge.ts';
 
 type ReviewView = ReturnType<ReviewService['load']>;
@@ -42,9 +42,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: <T>(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 {
@@ -135,6 +138,7 @@ export class MergeCoordinator {
   async displayStatus(view = this.service.load(), signal?: AbortSignal): Promise<MergeStatus | MergeUnavailableStatus> {
     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() };
     }
@@ -282,9 +286,10 @@ export class MergeCoordinator {
         // report that action as failed; the durable submitting record is recoverable by polling.
         let applied = true;
         try {
-          applied = queueAttempt.kind === 'queue'
-            ? this.service.store.queueMergeAttempt(this.service.config.identity, queueAttempt.id, result.url)
-            : this.service.store.finishMergeAttempt(this.service.config.identity, queueAttempt.id, { state: 'merged', url: result.url });
+          const attempt = queueAttempt;
+          applied = this.#settle(() => attempt.kind === 'queue'
+            ? this.service.store.queueMergeAttempt(this.service.config.identity, attempt.id, result.url)
+            : this.service.store.finishMergeAttempt(this.service.config.identity, attempt.id, { state: 'merged', url: result.url }));
         } catch {}
         // Not applied: a concurrent poll changed the attempt first. A committed failure is the answer, not this success.
         if (!applied) {
@@ -302,15 +307,15 @@ export class MergeCoordinator {
       if (error instanceof CommittedFailure) throw error;
       // GitHub's refusal is definite only once the attempt's failed state (and the click's saved answer) is durable.
       let refusalSaved = false;
-      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') {
-          refusalSaved = this.service.store.finishMergeAttempt(this.service.config.identity, queueAttempt.id, {
+          refusalSaved = 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 {}
       // A definite refusal before admission is this click's outcome; a resend replays it. Aborts, the deadline and
       // shutdown applied nothing for a passing reason, so the same click may be sent again.
       if (!queueAttempt && !signal.aborted && action) {
@@ -360,6 +365,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)) {

runner/merge.ts

@@ -4,6 +4,7 @@ import { QuestionWorker } from './question-agent.ts';
 import { LeftoverLedger } from './question-leftovers.ts';
 import { StopError, type QuestionScope } from './question-container.ts';
 import type { ReviewNote } from './store.ts';
+import type { ShutdownCapability } from './lifecycle.ts';
 export type QuestionAgent = (prompt: string, signal: AbortSignal, scope?: QuestionScope, timeoutMs?: number) => Promise<string>;
 const QUESTION_TIMEOUT_MS = 120_000;
 const SHUTDOWN_SETTLE_MS = 20_000;
@@ -26,8 +27,10 @@ export class Questions {
   private service: ReviewService;
   private agent?: QuestionAgent;
   private worker: QuestionWorker;
-  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: <T>(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();
     // Beside the review database's canonical path, so a restart of the same review finds what an earlier session left.
     this.worker=new QuestionWorker(undefined,LeftoverLedger.forDatabase(service.config.database));
   }
@@ -65,9 +68,9 @@ export class Questions {
         invocation = agent(questionPrompt(view,note),controller.signal,scope,QUESTION_TIMEOUT_MS);
         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 () => {

runner/questions.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, MERGEABLE_STATUSES, DEFAULT_TASK_BUDGET_MS, FIRST_REASONS, GuardRefusal, ActionIdReused, RefusalWithEffect, MAX_RESULT_BYTES, TASK_STATUSES, TERMINAL_STATES,
+  ATTEMPT_PHASES, BadRequest, CLOSED_STATUSES, MERGEABLE_STATUSES, ShuttingDownError, type ShutdownCapability, DEFAULT_TASK_BUDGET_MS, FIRST_REASONS, GuardRefusal, ActionIdReused, RefusalWithEffect, 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';
@@ -121,13 +121,26 @@ 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: <T>(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;
   /** The user action whose transaction is open, so its events can prove they belong to it. */
   #action: { key: string; actionId: string } | null = null;
   /** Nested calls join the outer transaction, so a user action can wrap existing Store methods atomically. */
   #transaction<T>(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; }
@@ -816,8 +829,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 });