F1c: shutdown wiring and /api/runner

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<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 {
@@ -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<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)); }
@@ -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' });

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<T>(fn: () => T): T }
 
 export function bounded(reason: string): string {
   const text = reason.trim() || 'No reason given.';

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: <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 {
@@ -117,6 +121,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() };
     }
@@ -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)) {

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<string>;
 export function questionPrompt(view: ReturnType<ReviewService['load']>, 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: <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();
+  }
   isRunning(id: string) { return this.running.has(id); }
   start(id: string, view: ReturnType<ReviewService['load']>) {
     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 () => {

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: <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;
   /** 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; }
@@ -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 }); });
       }