Skip to content

Commit b091bbd

Browse files
mchwangclaude
andauthored
F1c: shutdown wiring and /api/runner (#57)
* 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> * Refuse malformed /api/runner requests with 400 before recording them An unknown action, a non-integer expectedStateVersion, or a missing or malformed attempt ID now throws BadRequest before userAction, so the request answers 400 and nothing is saved under its action ID. Before, a bad attempt ID was refused inside userAction and recorded, so a corrected resend under the same key got ActionIdReused. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * Run the shutdown test's Git commands through fixtureGit main (#82) runs every test Git command with the hardened options and an allowlisted environment. The review-load gate test now commits through test/fixtures/git.ts like the other tests. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * Test shutdown settlement after the gate; log unsaved answers; lighter runner view - A question running at shutdown now has a test that its "Server stopped" answer is written after the gate closes, and GitHub's merge refusal has one for the same case. Before, removing the capability from either write failed no test. - A failed answer that cannot be saved is logged instead of dropped. - /api/runner reads the last 20 attempts with a result flag, not every attempt's decoded result. - settleWith() in lifecycle.ts replaces three copies of the capability wrapper. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
1 parent d1fbd19 commit b091bbd

8 files changed

Lines changed: 419 additions & 41 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: 11 additions & 6 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, type UnreleasedResource } 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, settleWith } from './lifecycle.ts';
55

66
/** What F's host-side preparation hands to D's start call. */
77
export interface PreparedAttempt {
@@ -83,10 +83,15 @@ export class RunnerCoordinator {
8383
#closing = false;
8484
/** Set once D settles with `unreleased`: no new work until a restart's recovery confirms their removal. */
8585
#unreleased: UnreleasedResource[] | null = null;
86-
constructor(store: Store, deps: RunnerDeps, limits: SlotLimits = { writable: 1, readOnly: 1 }) {
86+
/** Settlement writes run with the shutdown capability, so they still land after the write gate closes. */
87+
#write: <T>(fn: () => T) => T;
88+
constructor(store: Store, deps: RunnerDeps, limits: SlotLimits = { writable: 1, readOnly: 1 }, capability?: ShutdownCapability) {
8789
if (![limits.writable, limits.readOnly].every(n => Number.isSafeInteger(n) && n >= 1)) throw new Error('Slot limits must be positive integers.');
8890
this.#store = store; this.#deps = deps; this.#limits = limits;
91+
this.#write = settleWith(capability);
8992
}
93+
/** Shutdown step 1: reject admission synchronously, in the same turn as the server flag and the Store gate. */
94+
rejectAdmission(): void { this.#closing = true; }
9095
get closing(): boolean { return this.#closing; }
9196
/** Resources D could not confirm removed; non-null keeps the runner closed to new work until restart. */
9297
get unreleased(): readonly UnreleasedResource[] | null { return this.#unreleased; }
@@ -102,7 +107,7 @@ export class RunnerCoordinator {
102107
* a refused transaction releases the reservation in the same turn.
103108
*/
104109
start(identity: PlanIdentity, request: StartRequest): AttemptRecord {
105-
if (this.#closing) throw new GuardRefusal('The runner is shutting down.');
110+
if (this.#closing) throw new ShuttingDownError();
106111
if (this.#unreleased) throw new GuardRefusal('Needs restart: an agent\'s cleanup could not be confirmed, so its containers or files may remain.');
107112
const key = identityKey(identity);
108113
if (this.#jobs.has(key)) throw new GuardRefusal('An attempt is already active for this task.');
@@ -191,7 +196,7 @@ export class RunnerCoordinator {
191196
if (job.firstReason) { job.handle?.cancel(D_REASON[job.firstReason]); return false; }
192197
job.firstReason = reason;
193198
try {
194-
if (job.attemptId && !this.#store.recordFirstReason(job.identity, job.attemptId, reason)) {
199+
if (job.attemptId && !this.#write(() => this.#store.recordFirstReason(job.identity, job.attemptId, reason))) {
195200
// Another writer (for example cancel task) recorded a reason first; adopt the durable one.
196201
const durable = this.#store.getAttempt(job.identity, job.attemptId).firstReason;
197202
if (durable) job.firstReason = durable;
@@ -264,7 +269,7 @@ export class RunnerCoordinator {
264269
} catch (error) { return await this.#endBeforeLaunch(job, attempt, { detail: `Launch failed: ${message(error)}` }); }
265270
job.handle = handle;
266271
let running: boolean | undefined;
267-
try { running = this.#store.markRunning(job.identity, attempt.id); } catch { running = undefined; }
272+
try { running = this.#write(() => this.#store.markRunning(job.identity, attempt.id)); } catch { running = undefined; }
268273
if (running === undefined) {
269274
// A storage error, not a stop: keep ownership until D settles, then hold the slot under a marker.
270275
handle.cancel('capture-failure');
@@ -346,7 +351,7 @@ export class RunnerCoordinator {
346351
}
347352
#settle(job: Job, s: { stopReason?: StopReason; exitCode: number | null; signal: string | null; valid: boolean; result?: unknown; detail?: string }): Classification | undefined {
348353
try {
349-
return this.#store.settleAttempt(job.identity, job.attemptId, { ...s, firstReason: job.firstReason });
354+
return this.#write(() => this.#store.settleAttempt(job.identity, job.attemptId, { ...s, firstReason: job.firstReason }));
350355
} catch {
351356
// The row's outcome is unknown: hold the slot until startup recovery reconciles it.
352357
this.#markers.set(job.key, { group: job.group, attemptId: job.attemptId, reason: 'result-not-saved' });

‎runner/lifecycle.ts‎

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

3535
/** A guard refused the action. Refusals are definite outcomes and are recorded for replay. */
3636
export class GuardRefusal extends Error {}
37+
/** A malformed request, refused before any transaction and never recorded. The server maps it to HTTP 400. */
38+
export class BadRequest extends GuardRefusal {}
3739
/** Reusing an action ID for a different request. */
3840
export class ActionIdReused extends GuardRefusal {}
3941
/**
@@ -44,6 +46,14 @@ export class RefusalWithEffect extends GuardRefusal {
4446
readonly effect: () => void;
4547
constructor(message: string, effect: () => void) { super(message); this.effect = effect; }
4648
}
49+
/** A Store write after shutdown began. The server maps it to HTTP 503, never to the 409 used for review errors. */
50+
export class ShuttingDownError extends Error { constructor() { super('The review server is shutting down.'); } }
51+
/** Lets settling coordinator code write after the gate closes. Only the server hands it out, and never to HTTP handlers. */
52+
export interface ShutdownCapability { run<T>(fn: () => T): T }
53+
/** How a coordinator runs its settlement writes: through the capability when it has one, directly otherwise (tests, demos). */
54+
export function settleWith(capability?: ShutdownCapability): <T>(fn: () => T) => T {
55+
return capability ? fn => capability.run(fn) : fn => fn();
56+
}
4757

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

‎runner/merge.ts‎

Lines changed: 15 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
import type { ReviewService } from './review.ts';
22
import { mergeActionResponse, type MergeAttempt } from './store.ts';
3-
import { ActionIdReused, GuardRefusal, MERGEABLE_STATUSES, assertUuidV4 } from './lifecycle.ts';
3+
import { ActionIdReused, GuardRefusal, MERGEABLE_STATUSES, ShuttingDownError, assertUuidV4, settleWith, type ShutdownCapability } from './lifecycle.ts';
44
import { MergeSubmissionError, type MergeGateway, type MergeQueueGateway, type MergeQueueObservation, type MergeResult, type RemoteMergeState } from '../github/merge.ts';
55

66
type ReviewView = ReturnType<ReviewService['load']>;
@@ -42,9 +42,12 @@ export class MergeCoordinator {
4242
readonly service: ReviewService;
4343
readonly gateway: MergeGateway;
4444
readonly operationTimeoutMs: number;
45-
constructor(service: ReviewService, gateway: MergeGateway, operationTimeoutMs = 14_000) {
45+
/** Settlement of an irreversible merge keeps its writes after the Store gate closes; request-path reconciliation does not. */
46+
#settle: <T>(fn: () => T) => T;
47+
constructor(service: ReviewService, gateway: MergeGateway, operationTimeoutMs = 14_000, capability?: ShutdownCapability) {
4648
if (!Number.isSafeInteger(operationTimeoutMs) || operationTimeoutMs < 1 || operationTimeoutMs > 14_000) throw new Error('Invalid merge operation deadline.');
4749
this.service = service; this.gateway = gateway; this.operationTimeoutMs = operationTimeoutMs;
50+
this.#settle = settleWith(capability);
4851
}
4952

5053
#attempt(): MergeAttempt | null {
@@ -135,6 +138,7 @@ export class MergeCoordinator {
135138
async displayStatus(view = this.service.load(), signal?: AbortSignal): Promise<MergeStatus | MergeUnavailableStatus> {
136139
try { return await this.status(view, false, signal); }
137140
catch (error) {
141+
if (error instanceof ShuttingDownError) throw error;
138142
if (signal?.aborted) throw signal.reason;
139143
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() };
140144
}
@@ -282,9 +286,10 @@ export class MergeCoordinator {
282286
// report that action as failed; the durable submitting record is recoverable by polling.
283287
let applied = true;
284288
try {
285-
applied = queueAttempt.kind === 'queue'
286-
? this.service.store.queueMergeAttempt(this.service.config.identity, queueAttempt.id, result.url)
287-
: this.service.store.finishMergeAttempt(this.service.config.identity, queueAttempt.id, { state: 'merged', url: result.url });
289+
const attempt = queueAttempt;
290+
applied = this.#settle(() => attempt.kind === 'queue'
291+
? this.service.store.queueMergeAttempt(this.service.config.identity, attempt.id, result.url)
292+
: this.service.store.finishMergeAttempt(this.service.config.identity, attempt.id, { state: 'merged', url: result.url }));
288293
} catch {}
289294
// Not applied: a concurrent poll changed the attempt first. A committed failure is the answer, not this success.
290295
if (!applied) {
@@ -302,15 +307,15 @@ export class MergeCoordinator {
302307
if (error instanceof CommittedFailure) throw error;
303308
// GitHub's refusal is definite only once the attempt's failed state (and the click's saved answer) is durable.
304309
let refusalSaved = false;
305-
if (queueAttempt) try {
310+
if (queueAttempt) try { const attempt = queueAttempt; this.#settle(() => {
306311
const message = error instanceof Error ? error.message : 'GitHub merge submission outcome is unknown.';
307312
if (error instanceof MergeSubmissionError && error.outcome === 'refused') {
308-
refusalSaved = this.service.store.finishMergeAttempt(this.service.config.identity, queueAttempt.id, {
313+
refusalSaved = this.service.store.finishMergeAttempt(this.service.config.identity, attempt.id, {
309314
state: 'failed', reason: message,
310315
requiresFreshReview: /head (?:branch |commit )?(?:was )?(?:modified|changed)|does not match.*head|stale review/i.test(message),
311316
});
312-
} else this.service.store.recordMergeAttemptDiagnostic(this.service.config.identity, queueAttempt.id, message);
313-
} catch {}
317+
} else this.service.store.recordMergeAttemptDiagnostic(this.service.config.identity, attempt.id, message);
318+
}); } catch {}
314319
// A definite refusal before admission is this click's outcome; a resend replays it. Aborts, the deadline and
315320
// shutdown applied nothing for a passing reason, so the same click may be sent again.
316321
if (!queueAttempt && !signal.aborted && action) {
@@ -360,6 +365,7 @@ export class MergeCoordinator {
360365
this.#publishQueueObservation(attempt, observation);
361366
return this.#queueStatus();
362367
} catch (error) {
368+
if (error instanceof ShuttingDownError) throw error;
363369
if (signal.aborted) throw signal.reason;
364370
const message = error instanceof Error ? error.message : 'Could not read the merge queue.';
365371
if (/head changed after review/i.test(message)) {

‎runner/questions.ts‎

Lines changed: 9 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ import { QuestionWorker } from './question-agent.ts';
44
import { LeftoverLedger } from './question-leftovers.ts';
55
import { StopError, type QuestionScope } from './question-container.ts';
66
import type { ReviewNote } from './store.ts';
7+
import { settleWith, type ShutdownCapability } from './lifecycle.ts';
78
export type QuestionAgent = (prompt: string, signal: AbortSignal, scope?: QuestionScope, timeoutMs?: number) => Promise<string>;
89
const QUESTION_TIMEOUT_MS = 120_000;
910
const SHUTDOWN_SETTLE_MS = 20_000;
@@ -26,8 +27,10 @@ export class Questions {
2627
private service: ReviewService;
2728
private agent?: QuestionAgent;
2829
private worker: QuestionWorker;
29-
constructor(service: ReviewService, agent?: QuestionAgent) {
30-
this.service=service; this.agent=agent;
30+
/** Settlement writes (finishAnswer after abort) keep working after the Store write gate closes. */
31+
private write: <T>(fn: () => T) => T;
32+
constructor(service: ReviewService, agent?: QuestionAgent, capability?: ShutdownCapability) {
33+
this.service=service; this.agent=agent; this.write = settleWith(capability);
3134
// Beside the review database's canonical path, so a restart of the same review finds what an earlier session left.
3235
this.worker=new QuestionWorker(undefined,LeftoverLedger.forDatabase(service.config.database));
3336
}
@@ -65,9 +68,11 @@ export class Questions {
6568
invocation = agent(questionPrompt(view,note),controller.signal,scope,QUESTION_TIMEOUT_MS);
6669
const text=await Promise.race([invocation,aborted]);
6770
if(typeof text!=='string'||!text.trim()||text.length>24000) throw new Error('Agent returned an empty or oversized answer.');
68-
this.service.store.finishAnswer(this.service.config.identity,id,attempt,{status:'complete',text:text.trim()});
71+
this.write(()=>this.service.store.finishAnswer(this.service.config.identity,id,attempt,{status:'complete',text:text.trim()}));
6972
} catch(error) {
70-
this.service.store.finishAnswer(this.service.config.identity,id,attempt,{status:'failed',error:(error instanceof Error?error.message:'Agent failed.').slice(0,1000)});
73+
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)})); }
74+
// The answer stays pending (the page shows it as interrupted); say why instead of dropping the error.
75+
catch(saveError) { console.error(`Question ${id} could not record its failed answer: ${saveError instanceof Error?saveError.message:String(saveError)}`); }
7176
} finally {clearTimeout(timeout);}
7277
})();
7378
const settled = done.finally(async () => {

‎runner/store.ts‎

Lines changed: 24 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, MERGEABLE_STATUSES, DEFAULT_TASK_BUDGET_MS, FIRST_REASONS, GuardRefusal, ActionIdReused, RefusalWithEffect, MAX_RESULT_BYTES, TASK_STATUSES, TERMINAL_STATES,
9+
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,
1010
assertUuidV4, bounded, classifySettlement, requestHash, sameContext,
1111
type AttemptKind, type AttemptState, type Classification, type FirstReason, type Settlement, type TaskStatus,
1212
} from './lifecycle.ts';
@@ -121,13 +121,26 @@ export class Store {
121121
}
122122
close(): void { this.#db.close(); }
123123
#get(sql: string, ...args: SQLInputValue[]) { return this.#db.prepare(sql).get(...args); }
124-
#run(sql: string, ...args: SQLInputValue[]) { return this.#db.prepare(sql).run(...args); }
124+
#run(sql: string, ...args: SQLInputValue[]) { this.#checkWrite(); return this.#db.prepare(sql).run(...args); }
125+
// Shutdown write gate (runner-lifecycle.md, "Shutdown" step 1).
126+
#gateClosed = false; #privileged = 0; #capabilityIssued = false;
127+
#checkWrite(): void { if (this.#gateClosed && this.#privileged === 0) throw new ShuttingDownError(); }
128+
/** Issued once, to the server, which hands it only to coordinators' settlement and close code. */
129+
shutdownCapability(): ShutdownCapability {
130+
if (this.#capabilityIssued) throw new Error('The shutdown capability was already issued.');
131+
this.#capabilityIssued = true;
132+
return Object.freeze({ run: <T>(fn: () => T): T => { this.#privileged++; try { return fn(); } finally { this.#privileged--; } } });
133+
}
134+
/** Shutdown step 1: from now on, every write without the capability throws ShuttingDownError. Reads still work. */
135+
closeWrites(): void { this.#gateClosed = true; }
136+
get writesClosed(): boolean { return this.#gateClosed; }
125137
#depth = 0;
126138
/** The user action whose transaction is open, so its events can prove they belong to it. */
127139
#action: { key: string; actionId: string } | null = null;
128140
/** Nested calls join the outer transaction, so a user action can wrap existing Store methods atomically. */
129141
#transaction<T>(fn: () => T): T {
130142
if (this.#depth > 0) { this.#depth++; try { return fn(); } finally { this.#depth--; } }
143+
this.#checkWrite();
131144
this.#db.exec('BEGIN IMMEDIATE'); this.#depth = 1;
132145
try { const result = fn(); this.#db.exec('COMMIT'); return result; }
133146
catch (error) { this.#db.exec('ROLLBACK'); throw error; }
@@ -621,6 +634,13 @@ export class Store {
621634
if (!row) throw new Error('Unknown attempt.');
622635
return this.#attemptRecord(row);
623636
}
637+
/** The latest attempts, oldest first, for status views: results are flagged, not read. */
638+
recentAttempts(identity: PlanIdentity, limit: number): (Omit<AttemptRecord, 'result'> & { hasResult: boolean })[] {
639+
return this.#db.prepare(`SELECT * FROM (SELECT id, kind, phase, item, state, context, deadline, first_reason, stop_reason, exit_code, signal,
640+
NULL AS result, result IS NOT NULL AS has_result, diagnostic, diagnostic_ref, created_at, started_at, settled_at, rowid AS row_order
641+
FROM attempts WHERE plan_key=? ORDER BY rowid DESC LIMIT ?) ORDER BY row_order`).all(identityKey(identity), limit)
642+
.map(row => { const { result: _result, ...attempt } = this.#attemptRecord(row); return { ...attempt, hasResult: row.has_result === 1 }; });
643+
}
624644
getAttempts(identity: PlanIdentity): AttemptRecord[] {
625645
return this.#db.prepare('SELECT * FROM attempts WHERE plan_key=? ORDER BY rowid').all(identityKey(identity)).map(row => this.#attemptRecord(row));
626646
}
@@ -816,8 +836,8 @@ export class Store {
816836
return { response: value, replayed: false };
817837
});
818838
} catch (error) {
819-
const storage = (error as { code?: string }).code === 'ERR_SQLITE_ERROR';
820-
if (!replaying && !storage && !(error instanceof ActionIdReused) && this.#depth === 0) {
839+
const storage = (error as { code?: string }).code === 'ERR_SQLITE_ERROR' || error instanceof ShuttingDownError;
840+
if (!replaying && !storage && !(error instanceof ActionIdReused) && !(error instanceof BadRequest) && this.#depth === 0) {
821841
const message = error instanceof Error ? bounded(error.message) : 'Refused.';
822842
this.#transaction(() => {
823843
if (!this.#get('SELECT 1 FROM user_actions WHERE plan_key=? AND action_id=?', key, action.actionId)) record({ ok: false, error: message });

0 commit comments

Comments
 (0)