Skip to content

Commit 3767065

Browse files
mchwangclaude
andcommitted
F1b: runner coordinator for attempts, slots and shutdown
In-memory jobs with two-step admission and a slot reservation; abortable preparation with task-budget and deadline timers; a synchronous launch check before D's start; first-reason stops that keep the slot until D settles; Store-chosen terminal states; unresolved markers on storage failure; cancel task; and shutdown that keeps existing reasons and waits for every job. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1 parent 4c83e78 commit 3767065

2 files changed

Lines changed: 503 additions & 0 deletions

File tree

‎runner/coordinator.ts‎

Lines changed: 231 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,231 @@
1+
import { identityKey, type PlanIdentity } from '../core/identity.ts';
2+
import { captureInvocation, type InvocationHandle, type InvocationInput, type InvocationResult, type StopReason, type TaskClone } from '../agents/contract.ts';
3+
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';
5+
6+
/** What F's host-side preparation hands to D's start call. */
7+
export interface PreparedAttempt {
8+
readonly clone: TaskClone;
9+
readonly vendor: 'claude' | 'codex';
10+
readonly approvedArgv: readonly (readonly string[])[];
11+
}
12+
export interface RunnerDeps {
13+
/**
14+
* Host-side preparation (clone, prompt). On abort it must stop and await every subprocess it started, then reject.
15+
* It never leaves work running after it settles.
16+
*/
17+
prepare(attempt: AttemptRecord, signal: AbortSignal): Promise<PreparedAttempt>;
18+
/** Remove host-side preparation files only. Task storage waits for the terminal write. */
19+
cleanupPreparation(attempt: AttemptRecord): Promise<void>;
20+
/** D's start call: returns a handle at once, or throws with nothing left running. */
21+
start(input: InvocationInput, prepared: PreparedAttempt): InvocationHandle;
22+
/** Validate a clean result; throw with an actionable reason if it is invalid. Returns the value to persist. */
23+
validate(attempt: AttemptRecord, result: InvocationResult): unknown;
24+
now?(): number;
25+
}
26+
export interface SlotLimits { readonly writable: number; readonly readOnly: number }
27+
export interface StartRequest {
28+
expectedStateVersion: number; kind: AttemptKind; item?: string | null; deadline: number; budgetMs?: number; retryOf?: string;
29+
expectedContext: AttemptRecord['context'];
30+
}
31+
export interface RunnerStatus {
32+
active: boolean;
33+
stopRequested: { attemptId: string; reason: FirstReason; saved: boolean } | null;
34+
unresolved: { attemptId: string; reason: 'result-not-saved' | 'start-not-saved' } | null;
35+
}
36+
type Group = 'writable' | 'readOnly';
37+
interface Job {
38+
identity: PlanIdentity; key: string; group: Group; attemptId: string; attempt?: AttemptRecord;
39+
firstReason: FirstReason | null; reasonSaved: boolean; preparationTimedOut: boolean;
40+
controller: AbortController; handle?: InvocationHandle; timers: ReturnType<typeof setTimeout>[]; done?: Promise<void>;
41+
}
42+
interface Marker { group: Group; attemptId: string; reason: 'result-not-saved' | 'start-not-saved' }
43+
44+
const D_REASON: Record<FirstReason, StopReason> = { cancelled: 'cancelled', stale: 'cancelled', shutdown: 'shutdown', 'time-limit': 'timeout' };
45+
/** setTimeout accepts at most 2^31-1 ms; longer waits are re-armed. */
46+
const MAX_TIMER = 2_147_483_647;
47+
48+
/**
49+
* One runner coordinator per process and Store. Owns in-memory jobs, slots and unresolved markers.
50+
* See docs/implementation/runner-lifecycle.md ("Slots and concurrency", "Launch", "Rules for the running state").
51+
*/
52+
export class RunnerCoordinator {
53+
#store: Store; #deps: RunnerDeps; #limits: SlotLimits;
54+
#jobs = new Map<string, Job>(); #markers = new Map<string, Marker>();
55+
#closing = false;
56+
constructor(store: Store, deps: RunnerDeps, limits: SlotLimits = { writable: 1, readOnly: 1 }) {
57+
if (![limits.writable, limits.readOnly].every(n => Number.isSafeInteger(n) && n >= 1)) throw new Error('Slot limits must be positive integers.');
58+
this.#store = store; this.#deps = deps; this.#limits = limits;
59+
}
60+
get closing(): boolean { return this.#closing; }
61+
#now(): number { return this.#deps.now?.() ?? Date.now(); }
62+
#used(group: Group): number {
63+
let used = 0;
64+
for (const job of this.#jobs.values()) if (job.group === group) used++;
65+
for (const marker of this.#markers.values()) if (marker.group === group) used++;
66+
return used;
67+
}
68+
/**
69+
* Admission. In-memory checks and the slot reservation happen in one synchronous turn before the Store transaction;
70+
* a refused transaction releases the reservation in the same turn.
71+
*/
72+
start(identity: PlanIdentity, request: StartRequest): AttemptRecord {
73+
if (this.#closing) throw new GuardRefusal('The runner is shutting down.');
74+
const key = identityKey(identity);
75+
if (this.#jobs.has(key)) throw new GuardRefusal('An attempt is already active for this task.');
76+
const marker = this.#markers.get(key);
77+
if (marker) throw new GuardRefusal('Needs restart: the last result could not be saved.');
78+
if (!(request.kind in ATTEMPT_PHASES)) throw new GuardRefusal('Unknown attempt kind.');
79+
const group: Group = WRITABLE_KINDS.includes(request.kind) ? 'writable' : 'readOnly';
80+
if (this.#used(group) >= this.#limits[group]) throw new GuardRefusal('No free runner slot. Try again when the current attempt finishes.');
81+
const job: Job = { identity: { ...identity }, key, group, attemptId: '', firstReason: null, reasonSaved: true, preparationTimedOut: false, controller: new AbortController(), timers: [] };
82+
this.#jobs.set(key, job);
83+
let attempt: AttemptRecord;
84+
try { attempt = this.#store.admitAttempt(identity, { ...request, now: this.#now() }); }
85+
catch (error) { this.#jobs.delete(key); throw error; }
86+
job.attemptId = attempt.id; job.attempt = attempt;
87+
this.#arm(job, attempt);
88+
job.done = this.#run(job, attempt).catch(error => this.#unexpected(job, error));
89+
return attempt;
90+
}
91+
/** Retry is a new attempt bound to the latest failed or cancelled one. */
92+
retry(identity: PlanIdentity, attemptId: string, request: Omit<StartRequest, 'retryOf'>): AttemptRecord {
93+
return this.start(identity, { ...request, retryOf: attemptId });
94+
}
95+
/** User stop or detected staleness. The first reason wins; nothing is freed until settlement. */
96+
stop(identity: PlanIdentity, attemptId: string, reason: 'cancelled' | 'stale'): boolean {
97+
const job = this.#jobs.get(identityKey(identity));
98+
if (!job || job.attemptId !== attemptId) return false;
99+
return this.#requestStop(job, reason);
100+
}
101+
/** Cancel task: the Store records the reason and the pending close; the coordinator stops the running work. */
102+
cancelTask(identity: PlanIdentity, expectedStateVersion: number, actionId: string): 'closed' | 'stopping' {
103+
const outcome = this.#store.cancelTask(identity, expectedStateVersion, actionId);
104+
const job = this.#jobs.get(identityKey(identity));
105+
if (outcome === 'stopping' && job) this.#requestStop(job, 'cancelled');
106+
return outcome;
107+
}
108+
isActive(identity: PlanIdentity): boolean { return this.#jobs.has(identityKey(identity)); }
109+
status(identity: PlanIdentity): RunnerStatus {
110+
const key = identityKey(identity), job = this.#jobs.get(key), marker = this.#markers.get(key);
111+
return {
112+
active: !!job,
113+
stopRequested: job?.firstReason ? { attemptId: job.attemptId, reason: job.firstReason, saved: job.reasonSaved } : null,
114+
unresolved: marker ? { attemptId: marker.attemptId, reason: marker.reason } : null,
115+
};
116+
}
117+
/** Resolves when the task's current job has settled (or immediately if there is none). */
118+
async settled(identity: PlanIdentity): Promise<void> { await this.#jobs.get(identityKey(identity))?.done; }
119+
/**
120+
* Shutdown step 4: reject admission, record `shutdown` only where no reason is set, stop everything and await settlement.
121+
* No timer abandons a job (decision 4); the Store stays open for the caller to close afterwards.
122+
*/
123+
async close(): Promise<void> {
124+
this.#closing = true;
125+
const jobs = [...this.#jobs.values()];
126+
for (const job of jobs) this.#requestStop(job, 'shutdown');
127+
await Promise.all(jobs.map(job => job.done));
128+
}
129+
130+
#requestStop(job: Job, reason: FirstReason): boolean {
131+
if (job.firstReason) { job.handle?.cancel(D_REASON[job.firstReason]); return false; }
132+
job.firstReason = reason;
133+
try {
134+
if (job.attemptId && !this.#store.recordFirstReason(job.identity, job.attemptId, reason)) {
135+
// Another writer (for example cancel task) recorded a reason first; adopt the durable one.
136+
const durable = this.#store.getAttempt(job.identity, job.attemptId).firstReason;
137+
if (durable) job.firstReason = durable;
138+
}
139+
job.reasonSaved = true;
140+
} catch { job.reasonSaved = false; }
141+
job.controller.abort(new Error(`Stopped: ${job.firstReason}`));
142+
job.handle?.cancel(D_REASON[job.firstReason]);
143+
return true;
144+
}
145+
/** Task budget and, before launch, the attempt deadline. D enforces the deadline once it runs. */
146+
#arm(job: Job, attempt: AttemptRecord): void {
147+
const at = (when: number, fire: () => void) => {
148+
const wait = () => {
149+
const remaining = when - this.#now();
150+
if (remaining <= 0) return fire();
151+
job.timers.push(setTimeout(wait, Math.min(remaining, MAX_TIMER)));
152+
};
153+
wait();
154+
};
155+
const budget = this.#store.getTask(job.identity).budgetDeadline;
156+
if (budget !== null) at(budget, () => this.#requestStop(job, 'time-limit'));
157+
at(attempt.deadline, () => { if (!job.handle && !job.firstReason) { job.preparationTimedOut = true; job.controller.abort(new Error('Timed out while preparing.')); } });
158+
}
159+
async #run(job: Job, attempt: AttemptRecord): Promise<void> {
160+
try {
161+
let prepared: PreparedAttempt;
162+
try { prepared = await this.#deps.prepare(attempt, job.controller.signal); }
163+
catch (error) { return await this.#endBeforeLaunch(job, attempt, this.#preparationDetail(job, error)); }
164+
if (job.firstReason || job.preparationTimedOut) return await this.#endBeforeLaunch(job, attempt, this.#preparationDetail(job));
165+
// Launch check: one synchronous turn, no await between the checks and D's start call.
166+
const now = this.#now(), row = this.#store.getAttempt(job.identity, attempt.id), task = this.#store.getTask(job.identity);
167+
if (row.firstReason && !job.firstReason) job.firstReason = row.firstReason;
168+
if (row.state !== 'pending' || job.firstReason) return await this.#endBeforeLaunch(job, attempt, {});
169+
if (task.budgetDeadline !== null && now >= task.budgetDeadline) { this.#requestStop(job, 'time-limit'); return await this.#endBeforeLaunch(job, attempt, {}); }
170+
if (now >= attempt.deadline) return await this.#endBeforeLaunch(job, attempt, { stopReason: 'timeout', detail: 'Timed out while preparing.' });
171+
if (!sameContext(row.context, this.#store.currentContext(job.identity))) return await this.#endBeforeLaunch(job, attempt, {});
172+
let handle: InvocationHandle;
173+
try {
174+
const input = captureInvocation({ clone: prepared.clone, phase: ATTEMPT_PHASES[attempt.kind], vendor: prepared.vendor,
175+
approvedArgv: prepared.approvedArgv, deadline: attempt.deadline, attemptId: attempt.id, context: attempt.context }, now);
176+
handle = this.#deps.start(input, prepared);
177+
} catch (error) { return await this.#endBeforeLaunch(job, attempt, { detail: `Launch failed: ${message(error)}` }); }
178+
job.handle = handle;
179+
let running: boolean | undefined;
180+
try { running = this.#store.markRunning(job.identity, attempt.id); } catch { running = undefined; }
181+
if (running === undefined) {
182+
// A storage error, not a stop: keep ownership until D settles, then hold the slot under a marker.
183+
handle.cancel('capture-failure');
184+
await handle.settled.catch(() => undefined);
185+
this.#markers.set(job.key, { group: job.group, attemptId: attempt.id, reason: 'start-not-saved' });
186+
return;
187+
}
188+
if (!running) {
189+
// Refused because a first reason is now recorded: a normal stop.
190+
const durable = this.#store.getAttempt(job.identity, attempt.id).firstReason;
191+
if (durable && !job.firstReason) job.firstReason = durable;
192+
handle.cancel(D_REASON[job.firstReason ?? 'cancelled']);
193+
} else if (job.firstReason) handle.cancel(D_REASON[job.firstReason]);
194+
const result = await handle.settled;
195+
let valid = false, value: unknown, detail = result.stderr ? bounded(result.stderr) : undefined;
196+
if (!job.firstReason && result.exitCode === 0 && !result.stopReason) {
197+
try { value = this.#deps.validate(attempt, result); valid = true; }
198+
catch (error) { detail = `Invalid output: ${message(error)}`; }
199+
}
200+
this.#settle(job, { stopReason: result.stopReason, exitCode: result.exitCode, signal: result.signal, valid, result: value, detail });
201+
} finally {
202+
for (const timer of job.timers) clearTimeout(timer);
203+
if (this.#jobs.get(job.key) === job) this.#jobs.delete(job.key);
204+
}
205+
}
206+
#preparationDetail(job: Job, error?: unknown): { stopReason?: StopReason; detail?: string } {
207+
if (job.preparationTimedOut && !job.firstReason) return { stopReason: 'timeout', detail: 'Timed out while preparing.' };
208+
return error === undefined || job.firstReason ? {} : { detail: `Preparation failed: ${message(error)}` };
209+
}
210+
/** Ending without a handle: host-side cleanup, then the terminal write from the first reason. */
211+
async #endBeforeLaunch(job: Job, attempt: AttemptRecord, s: { stopReason?: StopReason; detail?: string }): Promise<void> {
212+
await this.#deps.cleanupPreparation(attempt).catch(() => undefined);
213+
this.#settle(job, { stopReason: s.stopReason, exitCode: null, signal: null, valid: false, detail: s.detail });
214+
}
215+
#settle(job: Job, s: { stopReason?: StopReason; exitCode: number | null; signal: string | null; valid: boolean; result?: unknown; detail?: string }): Classification | undefined {
216+
try {
217+
return this.#store.settleAttempt(job.identity, job.attemptId, { ...s, firstReason: job.firstReason });
218+
} catch {
219+
// The row's outcome is unknown: hold the slot until startup recovery reconciles it.
220+
this.#markers.set(job.key, { group: job.group, attemptId: job.attemptId, reason: 'result-not-saved' });
221+
return undefined;
222+
}
223+
}
224+
#unexpected(job: Job, error: unknown): void {
225+
// Fail closed: an unexpected error keeps the slot held until restart.
226+
this.#markers.set(job.key, { group: job.group, attemptId: job.attemptId, reason: 'result-not-saved' });
227+
if (this.#jobs.get(job.key) === job) this.#jobs.delete(job.key);
228+
console.error(`Runner job ${job.attemptId} failed unexpectedly: ${message(error)}`);
229+
}
230+
}
231+
const message = (error: unknown) => bounded(error instanceof Error ? error.message : String(error));

0 commit comments

Comments
 (0)