Skip to content

Commit 081750f

Browse files
mchwangclaude
andcommitted
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>
1 parent 3767065 commit 081750f

8 files changed

Lines changed: 374 additions & 38 deletions

File tree

‎docs/implementation/runner-lifecycle.md‎

Lines changed: 8 additions & 8 deletions
Large diffs are not rendered by default.

‎runner/coordinator.ts‎

Lines changed: 14 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
import { identityKey, type PlanIdentity } from '../core/identity.ts';
22
import { captureInvocation, type InvocationHandle, type InvocationInput, type InvocationResult, type StopReason, type TaskClone } from '../agents/contract.ts';
33
import type { AttemptRecord, Store } from './store.ts';
4-
import { ATTEMPT_PHASES, GuardRefusal, WRITABLE_KINDS, bounded, sameContext, type AttemptKind, type Classification, type FirstReason } from './lifecycle.ts';
4+
import { ATTEMPT_PHASES, GuardRefusal, ShuttingDownError, WRITABLE_KINDS, bounded, sameContext, type AttemptKind, type Classification, type FirstReason, type ShutdownCapability } from './lifecycle.ts';
55

66
/** What F's host-side preparation hands to D's start call. */
77
export interface PreparedAttempt {
@@ -53,10 +53,15 @@ export class RunnerCoordinator {
5353
#store: Store; #deps: RunnerDeps; #limits: SlotLimits;
5454
#jobs = new Map<string, Job>(); #markers = new Map<string, Marker>();
5555
#closing = false;
56-
constructor(store: Store, deps: RunnerDeps, limits: SlotLimits = { writable: 1, readOnly: 1 }) {
56+
/** Settlement writes run with the shutdown capability, so they still land after the write gate closes. */
57+
#write: <T>(fn: () => T) => T;
58+
constructor(store: Store, deps: RunnerDeps, limits: SlotLimits = { writable: 1, readOnly: 1 }, capability?: ShutdownCapability) {
5759
if (![limits.writable, limits.readOnly].every(n => Number.isSafeInteger(n) && n >= 1)) throw new Error('Slot limits must be positive integers.');
5860
this.#store = store; this.#deps = deps; this.#limits = limits;
61+
this.#write = capability ? fn => capability.run(fn) : fn => fn();
5962
}
63+
/** Shutdown step 1: reject admission synchronously, in the same turn as the server flag and the Store gate. */
64+
rejectAdmission(): void { this.#closing = true; }
6065
get closing(): boolean { return this.#closing; }
6166
#now(): number { return this.#deps.now?.() ?? Date.now(); }
6267
#used(group: Group): number {
@@ -70,7 +75,7 @@ export class RunnerCoordinator {
7075
* a refused transaction releases the reservation in the same turn.
7176
*/
7277
start(identity: PlanIdentity, request: StartRequest): AttemptRecord {
73-
if (this.#closing) throw new GuardRefusal('The runner is shutting down.');
78+
if (this.#closing) throw new ShuttingDownError();
7479
const key = identityKey(identity);
7580
if (this.#jobs.has(key)) throw new GuardRefusal('An attempt is already active for this task.');
7681
const marker = this.#markers.get(key);
@@ -85,7 +90,8 @@ export class RunnerCoordinator {
8590
catch (error) { this.#jobs.delete(key); throw error; }
8691
job.attemptId = attempt.id; job.attempt = attempt;
8792
this.#arm(job, attempt);
88-
job.done = this.#run(job, attempt).catch(error => this.#unexpected(job, error));
93+
// Start after the caller's transaction commits: a rolled-back admission must not leave a job running.
94+
job.done = Promise.resolve().then(() => this.#run(job, attempt)).catch(error => this.#unexpected(job, error));
8995
return attempt;
9096
}
9197
/** Retry is a new attempt bound to the latest failed or cancelled one. */
@@ -131,7 +137,7 @@ export class RunnerCoordinator {
131137
if (job.firstReason) { job.handle?.cancel(D_REASON[job.firstReason]); return false; }
132138
job.firstReason = reason;
133139
try {
134-
if (job.attemptId && !this.#store.recordFirstReason(job.identity, job.attemptId, reason)) {
140+
if (job.attemptId && !this.#write(() => this.#store.recordFirstReason(job.identity, job.attemptId, reason))) {
135141
// Another writer (for example cancel task) recorded a reason first; adopt the durable one.
136142
const durable = this.#store.getAttempt(job.identity, job.attemptId).firstReason;
137143
if (durable) job.firstReason = durable;
@@ -158,6 +164,7 @@ export class RunnerCoordinator {
158164
}
159165
async #run(job: Job, attempt: AttemptRecord): Promise<void> {
160166
try {
167+
if (!this.#store.getAttempts(job.identity).some(row => row.id === attempt.id)) return; // admission was rolled back
161168
let prepared: PreparedAttempt;
162169
try { prepared = await this.#deps.prepare(attempt, job.controller.signal); }
163170
catch (error) { return await this.#endBeforeLaunch(job, attempt, this.#preparationDetail(job, error)); }
@@ -177,7 +184,7 @@ export class RunnerCoordinator {
177184
} catch (error) { return await this.#endBeforeLaunch(job, attempt, { detail: `Launch failed: ${message(error)}` }); }
178185
job.handle = handle;
179186
let running: boolean | undefined;
180-
try { running = this.#store.markRunning(job.identity, attempt.id); } catch { running = undefined; }
187+
try { running = this.#write(() => this.#store.markRunning(job.identity, attempt.id)); } catch { running = undefined; }
181188
if (running === undefined) {
182189
// A storage error, not a stop: keep ownership until D settles, then hold the slot under a marker.
183190
handle.cancel('capture-failure');
@@ -214,7 +221,7 @@ export class RunnerCoordinator {
214221
}
215222
#settle(job: Job, s: { stopReason?: StopReason; exitCode: number | null; signal: string | null; valid: boolean; result?: unknown; detail?: string }): Classification | undefined {
216223
try {
217-
return this.#store.settleAttempt(job.identity, job.attemptId, { ...s, firstReason: job.firstReason });
224+
return this.#write(() => this.#store.settleAttempt(job.identity, job.attemptId, { ...s, firstReason: job.firstReason }));
218225
} catch {
219226
// The row's outcome is unknown: hold the slot until startup recovery reconciles it.
220227
this.#markers.set(job.key, { group: job.group, attemptId: job.attemptId, reason: 'result-not-saved' });

‎runner/lifecycle.ts‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -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
2727
/** Attempt, action and allocation IDs are lowercase UUID v4s; anything else is refused before use. */
2828
export function isUuidV4(value: unknown): value is string { return typeof value === 'string' && UUID_V4.test(value); }
2929
export function assertUuidV4(value: unknown, name: string): asserts value is string {
30-
if (!isUuidV4(value)) throw new GuardRefusal(`${name} must be a lowercase UUID v4.`);
30+
if (!isUuidV4(value)) throw new BadRequest(`${name} must be a lowercase UUID v4.`);
3131
}
3232

3333
/** A guard refused the action. Refusals are definite outcomes and are recorded for replay. */
3434
export class GuardRefusal extends Error {}
35+
/** A malformed request, refused before any transaction and never recorded. The server maps it to HTTP 400. */
36+
export class BadRequest extends GuardRefusal {}
3537
/** Reusing an action ID for a different request. */
3638
export class ActionIdReused extends GuardRefusal {}
39+
/** A Store write after shutdown began. The server maps it to HTTP 503, never to the 409 used for review errors. */
40+
export class ShuttingDownError extends Error { constructor() { super('The review server is shutting down.'); } }
41+
/** Lets settling coordinator code write after the gate closes. Only the server hands it out, and never to HTTP handlers. */
42+
export interface ShutdownCapability { run<T>(fn: () => T): T }
3743

3844
export function bounded(reason: string): string {
3945
const text = reason.trim() || 'No reason given.';

‎runner/merge.ts‎

Lines changed: 16 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
import type { ReviewService } from './review.ts';
2+
import { ShuttingDownError, type ShutdownCapability } from './lifecycle.ts';
23
import type { MergeAttempt } from './store.ts';
34
import { MergeSubmissionError, type MergeGateway, type MergeQueueGateway, type MergeQueueObservation, type MergeResult, type RemoteMergeState } from '../github/merge.ts';
45

@@ -30,9 +31,12 @@ export class MergeCoordinator {
3031
readonly service: ReviewService;
3132
readonly gateway: MergeGateway;
3233
readonly operationTimeoutMs: number;
33-
constructor(service: ReviewService, gateway: MergeGateway, operationTimeoutMs = 14_000) {
34+
/** Settlement of an irreversible merge keeps its writes after the Store gate closes; request-path reconciliation does not. */
35+
#settle: <T>(fn: () => T) => T;
36+
constructor(service: ReviewService, gateway: MergeGateway, operationTimeoutMs = 14_000, capability?: ShutdownCapability) {
3437
if (!Number.isSafeInteger(operationTimeoutMs) || operationTimeoutMs < 1 || operationTimeoutMs > 14_000) throw new Error('Invalid merge operation deadline.');
3538
this.service = service; this.gateway = gateway; this.operationTimeoutMs = operationTimeoutMs;
39+
this.#settle = capability ? fn => capability.run(fn) : fn => fn();
3640
}
3741

3842
#attempt(): MergeAttempt | null {
@@ -117,6 +121,7 @@ export class MergeCoordinator {
117121
async displayStatus(view = this.service.load(), signal?: AbortSignal): Promise<MergeStatus | MergeUnavailableStatus> {
118122
try { return await this.status(view, false, signal); }
119123
catch (error) {
124+
if (error instanceof ShuttingDownError) throw error;
120125
if (signal?.aborted) throw signal.reason;
121126
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() };
122127
}
@@ -171,21 +176,24 @@ export class MergeCoordinator {
171176
// The enqueue command has already committed externally. A local refresh failure must not
172177
// report that action as failed; the durable submitting record is recoverable by polling.
173178
try {
174-
if (queueAttempt.kind === 'queue') this.service.store.queueMergeAttempt(this.service.config.identity, queueAttempt.id, result.url);
175-
else this.service.store.finishMergeAttempt(this.service.config.identity, queueAttempt.id, { state: 'merged' });
179+
const attempt = queueAttempt;
180+
this.#settle(() => {
181+
if (attempt.kind === 'queue') this.service.store.queueMergeAttempt(this.service.config.identity, attempt.id, result.url);
182+
else this.service.store.finishMergeAttempt(this.service.config.identity, attempt.id, { state: 'merged' });
183+
});
176184
} catch {}
177185
}
178186
return { status: commandStatus, result };
179187
} catch (error) {
180-
if (queueAttempt) try {
188+
if (queueAttempt) try { const attempt = queueAttempt; this.#settle(() => {
181189
const message = error instanceof Error ? error.message : 'GitHub merge submission outcome is unknown.';
182190
if (error instanceof MergeSubmissionError && error.outcome === 'refused') {
183-
this.service.store.finishMergeAttempt(this.service.config.identity, queueAttempt.id, {
191+
this.service.store.finishMergeAttempt(this.service.config.identity, attempt.id, {
184192
state: 'failed', reason: message,
185193
requiresFreshReview: /head (?:branch |commit )?(?:was )?(?:modified|changed)|does not match.*head|stale review/i.test(message),
186194
});
187-
} else this.service.store.recordMergeAttemptDiagnostic(this.service.config.identity, queueAttempt.id, message);
188-
} catch {}
195+
} else this.service.store.recordMergeAttemptDiagnostic(this.service.config.identity, attempt.id, message);
196+
}); } catch {}
189197
if (signal.aborted && signal.reason instanceof Error) throw signal.reason;
190198
throw error;
191199
}
@@ -221,6 +229,7 @@ export class MergeCoordinator {
221229
this.#publishQueueObservation(attempt, observation);
222230
return this.#queueStatus();
223231
} catch (error) {
232+
if (error instanceof ShuttingDownError) throw error;
224233
if (signal.aborted) throw signal.reason;
225234
const message = error instanceof Error ? error.message : 'Could not read the merge queue.';
226235
if (/head changed after review/i.test(message)) {

‎runner/questions.ts‎

Lines changed: 8 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@ import { randomUUID } from 'node:crypto';
22
import type { ReviewService } from './review.ts';
33
import { cliQuestionAgent } from './question-agent.ts';
44
import type { ReviewNote } from './store.ts';
5+
import type { ShutdownCapability } from './lifecycle.ts';
56
export type QuestionAgent = (prompt: string, signal: AbortSignal) => Promise<string>;
67
export function questionPrompt(view: ReturnType<ReviewService['load']>, note: ReviewNote): string {
78
let remaining = 100_000;
@@ -21,7 +22,11 @@ export class Questions {
2122
private closing = false;
2223
private service: ReviewService;
2324
private agent?: QuestionAgent;
24-
constructor(service: ReviewService, agent?: QuestionAgent) { this.service=service; this.agent=agent; }
25+
/** Settlement writes (finishAnswer after abort) keep working after the Store write gate closes. */
26+
private write: <T>(fn: () => T) => T;
27+
constructor(service: ReviewService, agent?: QuestionAgent, capability?: ShutdownCapability) {
28+
this.service=service; this.agent=agent; this.write = capability ? fn => capability.run(fn) : fn => fn();
29+
}
2530
isRunning(id: string) { return this.running.has(id); }
2631
start(id: string, view: ReturnType<ReviewService['load']>) {
2732
if (this.closing) throw new Error('Server is stopping. Reconnect before asking again.');
@@ -43,9 +48,9 @@ export class Questions {
4348
invocation = agent(questionPrompt(view,note),controller.signal);
4449
const text=await Promise.race([invocation,aborted]);
4550
if(typeof text!=='string'||!text.trim()||text.length>24000) throw new Error('Agent returned an empty or oversized answer.');
46-
this.service.store.finishAnswer(this.service.config.identity,id,attempt,{status:'complete',text:text.trim()});
51+
this.write(()=>this.service.store.finishAnswer(this.service.config.identity,id,attempt,{status:'complete',text:text.trim()}));
4752
} catch(error) {
48-
this.service.store.finishAnswer(this.service.config.identity,id,attempt,{status:'failed',error:(error instanceof Error?error.message:'Agent failed.').slice(0,1000)});
53+
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 {}
4954
} finally {clearTimeout(timeout);}
5055
})();
5156
const settled = done.finally(async () => {

‎runner/store.ts‎

Lines changed: 17 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@ import { importPlan, applySuggestion, assertEditReply, type Plan, type PlanConte
66
import type { Approval, SegmentChoice } from '../core/approvals.ts';
77
import type { InvocationContext, StopReason } from '../agents/contract.ts';
88
import {
9-
ATTEMPT_PHASES, CLOSED_STATUSES, DEFAULT_TASK_BUDGET_MS, FIRST_REASONS, GuardRefusal, ActionIdReused, MAX_RESULT_BYTES, TASK_STATUSES, TERMINAL_STATES,
9+
ATTEMPT_PHASES, BadRequest, CLOSED_STATUSES, ShuttingDownError, type ShutdownCapability, DEFAULT_TASK_BUDGET_MS, FIRST_REASONS, GuardRefusal, ActionIdReused, MAX_RESULT_BYTES, TASK_STATUSES, TERMINAL_STATES,
1010
assertUuidV4, bounded, classifySettlement, requestHash, sameContext,
1111
type AttemptKind, type AttemptState, type Classification, type FirstReason, type Settlement, type TaskStatus,
1212
} from './lifecycle.ts';
@@ -115,11 +115,24 @@ export class Store {
115115
}
116116
close(): void { this.#db.close(); }
117117
#get(sql: string, ...args: SQLInputValue[]) { return this.#db.prepare(sql).get(...args); }
118-
#run(sql: string, ...args: SQLInputValue[]) { return this.#db.prepare(sql).run(...args); }
118+
#run(sql: string, ...args: SQLInputValue[]) { this.#checkWrite(); return this.#db.prepare(sql).run(...args); }
119+
// Shutdown write gate (runner-lifecycle.md, "Shutdown" step 1).
120+
#gateClosed = false; #privileged = 0; #capabilityIssued = false;
121+
#checkWrite(): void { if (this.#gateClosed && this.#privileged === 0) throw new ShuttingDownError(); }
122+
/** Issued once, to the server, which hands it only to coordinators' settlement and close code. */
123+
shutdownCapability(): ShutdownCapability {
124+
if (this.#capabilityIssued) throw new Error('The shutdown capability was already issued.');
125+
this.#capabilityIssued = true;
126+
return Object.freeze({ run: <T>(fn: () => T): T => { this.#privileged++; try { return fn(); } finally { this.#privileged--; } } });
127+
}
128+
/** Shutdown step 1: from now on, every write without the capability throws ShuttingDownError. Reads still work. */
129+
closeWrites(): void { this.#gateClosed = true; }
130+
get writesClosed(): boolean { return this.#gateClosed; }
119131
#depth = 0;
120132
/** Nested calls join the outer transaction, so a user action can wrap existing Store methods atomically. */
121133
#transaction<T>(fn: () => T): T {
122134
if (this.#depth > 0) { this.#depth++; try { return fn(); } finally { this.#depth--; } }
135+
this.#checkWrite();
123136
this.#db.exec('BEGIN IMMEDIATE'); this.#depth = 1;
124137
try { const result = fn(); this.#db.exec('COMMIT'); return result; }
125138
catch (error) { this.#db.exec('ROLLBACK'); throw error; }
@@ -736,8 +749,8 @@ export class Store {
736749
return { response: value, replayed: false };
737750
});
738751
} catch (error) {
739-
const storage = (error as { code?: string }).code === 'ERR_SQLITE_ERROR';
740-
if (!replaying && !storage && !(error instanceof ActionIdReused) && this.#depth === 0) {
752+
const storage = (error as { code?: string }).code === 'ERR_SQLITE_ERROR' || error instanceof ShuttingDownError;
753+
if (!replaying && !storage && !(error instanceof ActionIdReused) && !(error instanceof BadRequest) && this.#depth === 0) {
741754
const message = error instanceof Error ? bounded(error.message) : 'Refused.';
742755
this.#transaction(() => { if (!this.#get('SELECT 1 FROM user_actions WHERE plan_key=? AND action_id=?', key, action.actionId)) record({ ok: false, error: message }); });
743756
}

0 commit comments

Comments
 (0)