0af867f · d390161 · 9a8f63e
@@ -184,6 +184,9 @@
export interface CrewHandle {
 */
const DRAIN_MS = 3 * 60_000;
/** How long `stop()` waits for the tick's own sweeps (converge, autonomy) to settle. */
const SWEEPS_STOP_MS = 10_000;
const handles = new WeakMap<App, CrewHandle>();
/** The crew serving this forge, if one was started. */
@@ -257,20 +260,33 @@
export async function startCrew(
  // Repositories that run on their own take a turn once a minute, on the
  // same clock as the pump, so a pick becomes a run on the next tick.
  let lastAutonomy = 0;
  // The sweeps a tick started and has not seen settle: `stop()` waits for
  // them, so none reads `app.db` after its caller closed it (#9977).
  const sweeps = new Set<Promise<unknown>>();
  const track = (sweep: Promise<unknown>): void => {
    sweeps.add(sweep);
    void sweep.finally(() => sweeps.delete(sweep));
  };
  const timer = setInterval(() => {
    if (stopped) return;
    requeueInterrupted(app);
    recheckForDisk(app);
    if (Date.now() - lastAutonomy >= AUTONOMY_TICK_MS) {
      lastAutonomy = Date.now();
      settleAll(app).catch((error) =>
        log.error("converge sweep failed", { error: String(error) }),
      );
      autonomyTick(app).catch((error) =>
        log.error("autonomy tick failed", { error: String(error) }),
      );
    // A tick that throws is logged and the next one tries again: thrown
    // from a timer it is an uncaught error that takes the process down.
    try {
      requeueInterrupted(app);
      recheckForDisk(app);
      if (Date.now() - lastAutonomy >= AUTONOMY_TICK_MS) {
        lastAutonomy = Date.now();
        track(settleAll(app).catch((error) =>
          log.error("converge sweep failed", { error: String(error) }),
        ));
        track(autonomyTick(app).catch((error) =>
          log.error("autonomy tick failed", { error: String(error) }),
        ));
      }
      pump(app, ready, inFlight);
    } catch (error) {
      log.error("crew tick failed", { error: String(error) });
    }
    pump(app, ready, inFlight);
  }, options.tickMs ?? TICK_MS);
  // Clones no open issue, pull or run claims fill the volume (#5697).
  const sweepClones = (): void => {
@@ -299,6 +315,16 @@
export async function startCrew(
      unsubscribe();
      unfollow();
      unsettle();
      // A sweep still going would read the database its caller closes next;
      // bounded, for one stuck on the network must not hold a restart.
      if (sweeps.size > 0) {
        let bound: ReturnType<typeof setTimeout> | undefined;
        await Promise.race([
          Promise.all(sweeps),
          new Promise((resolve) => (bound = setTimeout(resolve, SWEEPS_STOP_MS))),
        ]);
        clearTimeout(bound);
      }
      if (options.drain === true && inFlight.size > 0) {
        // Let what is in flight finish: a run cut short is a person's
        // money spent twice. The new forge takes new work meanwhile.
@@ -0,0 +1,61 @@
/**
 * The crew's tick never takes the process down (#9977): a tick whose
 * database refuses it is logged, and the next tick does its work. Thrown
 * from the timer, the error would be uncaught: a forge would fall over,
 * and a test shard would exit without naming a test.
 */
import { afterAll, beforeAll, expect, test } from "bun:test";
import { mkdtemp, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { createApp, type App } from "../src/app.ts";
import { createRepo } from "../src/actions/repos.ts";
import { claimRun, createRun, getRun } from "../src/core/runs.ts";
import { createUser } from "../src/core/users.ts";
import { startCrew, type CrewHandle } from "../src/crew/dispatch.ts";
let dataDir: string;
let app: App;
let crew: CrewHandle;
beforeAll(async () => {
  dataDir = await mkdtemp(join(tmpdir(), "forge-crew-tick-"));
  app = createApp({ dataDir, port: 0 });
  crew = await startCrew(app, {
    forgeUrl: "http://127.0.0.1:9",
    localUrl: "http://127.0.0.1:9",
    binary: join(dataDir, "no-claude-here"),
    concurrency: 0,
    budgetUsd: 1,
    runTimeoutMs: 60_000,
    tickMs: 20,
  });
});
afterAll(async () => {
  await crew.stop();
  app.close();
  await rm(dataDir, { recursive: true, force: true });
});
test("a tick that throws is logged, and the next tick still does its work", async () => {
  const a = await createUser(app.db, { origin: "seed", trusted: true, username: "alice", email: "[email protected]", password: "a-good-passphrase", adminIfFirst: true });
  if (!a.ok) throw new Error(a.error);
  const made = await createRepo(app, a.value, { name: "ticks", visibility: "private" });
  if (!made.ok) throw new Error(String(made.error));
  // A run a dead forge left running: its lease lapsed a day ago, so every
  // tick's first step writes, to cancel it and queue its task again.
  const stale = createRun(app.db, { repoId: made.value.id, kind: "agent", reason: "push", actorId: a.value.id });
  claimRun(app.db, stale.id);
  const old = new Date(Date.now() - 86_400_000).toISOString();
  app.db.query("UPDATE run SET heartbeat_at = ?1, started_at = ?1 WHERE id = ?2").run(old, stale.id);
  // The database refuses every write: each tick throws on its first step.
  app.db.exec("PRAGMA query_only = ON");
  await Bun.sleep(200);
  app.db.exec("PRAGMA query_only = OFF");
  expect(getRun(app.db, stale.id)?.state).toBe("running");
  // The crew is still ticking: the lapsed run is put back on the queue.
  const deadline = Date.now() + 5_000;
  while (getRun(app.db, stale.id)?.state === "running" && Date.now() < deadline) await Bun.sleep(20);
  expect(getRun(app.db, stale.id)?.state).toBe("cancelled");
});
@@ -123,7 +123,9 @@
test("home says the whole crew's number", () => {
});
test("a repository that hit its ceiling yesterday restarts at its own time, spread over the morning", () => {
  const midnight = new Date(Date.UTC(2026, 10, 3, 0, 5, 0));
  // Midnight itself: repository ids are random, so an instant past 00:00 would
  // find a repository whose offset has already come and nothing held.
  const midnight = new Date(Date.UTC(2026, 10, 3, 0, 0, 0));
  for (const r of repos) {
    const run = createRun(app.db, { repoId: r.id, kind: "agent" });
    app.db.query("UPDATE run SET cost_usd = 1000, created_at = ?2, state = 'ok' WHERE id = ?1").run(run.id, new Date(Date.UTC(2026, 10, 2, 20, 0, 0)).toISOString());
@@ -147,13 +149,17 @@
test("a repository that hit its ceiling yesterday restarts at its own time, spre
});
test("a ceiling of 0 allows no spend and is held by the ceiling, so it is never also held until an offset", () => {
  const midnight = new Date(Date.UTC(2026, 10, 3, 0, 5, 0));
  // Midnight itself: repository ids are random, so an instant past 00:00 would
  // find a repository whose offset has already come and nothing held.
  const midnight = new Date(Date.UTC(2026, 10, 3, 0, 0, 0));
  forgetStagger();
  expect(staggeredUntil(app.db, { id: repos[0]!.id, autonomyCeilingUsd: 0 }, midnight)).toBeNull();
});
test("the stagger reads yesterday's spend once for one instant", () => {
  const midnight = new Date(Date.UTC(2026, 10, 3, 0, 5, 0));
  // Midnight itself: repository ids are random, so an instant past 00:00 would
  // find a repository whose offset has already come and nothing held.
  const midnight = new Date(Date.UTC(2026, 10, 3, 0, 0, 0));
  const run = createRun(app.db, { repoId: repos[3]!.id, kind: "agent" });
  app.db.query("UPDATE run SET cost_usd = 1000, created_at = ?2, state = 'ok' WHERE id = ?1").run(run.id, new Date(Date.UTC(2026, 10, 2, 20, 0, 0)).toISOString());
  forgetStagger();
@@ -812,6 +812,10 @@
describe("absorbing an app", () => {
    } finally {
      app.db.run("UPDATE repo SET visibility = 'private' WHERE id = ?1", [repo.id]);
      stub.stop(true);
      // The constructor makes it this forge's runtime: a deploy it was asked
      // for must settle before afterAll closes the database, or its
      // finishRun is an unhandled error in whichever file runs next.
      await runtime.stop();
    }
  });
@@ -15,8 +15,13 @@
import { startStandInCrew, stopStandInCrew, type StandInCrew } from "./helpers/c
const DAY = 86_400_000;
/** Each test's own timeout; every wait in it fits inside, with room to spare. */
const TEST_MS = 30_000;
/** One wait's deadline: a test waits at most three times, so a slow tick fails as this test, by name, before TEST_MS (#9977). */
const WAIT_MS = 8_000;
/** Wait until `done` holds, or fail at the deadline: the pump's tick is what we wait on, not a guessed pause. */
async function until(done: () => boolean, what: string, ms = 20_000): Promise<void> {
async function until(done: () => boolean, what: string, ms = WAIT_MS): Promise<void> {
  const deadline = Date.now() + ms;
  while (!done()) {
    if (Date.now() > deadline) throw new Error(`timed out waiting for ${what}`);
@@ -27,8 +32,8 @@
let standIn: StandInCrew;
beforeAll(async () => {
  standIn = await startStandInCrew({ tickMs: 50 });
}, 30_000);
afterAll(() => stopStandInCrew(standIn));
}, TEST_MS);
afterAll(() => stopStandInCrew(standIn), TEST_MS);
test("past the week's reserve an assigned run stays queued, with the reason in its log", async () => {
  const { app, alice, bot } = standIn;
@@ -60,7 +65,7 @@
test("past the week's reserve an assigned run stays queued, with the reason in i
  forgetWeek();
  await until(() => mine()?.state !== "queued", "the run to start");
  expect(mine()?.state).not.toBe("queued");
}, 30_000);
}, TEST_MS);
test("builds fill the pace slots, yet a review or assigned run still starts; a third of them waits (#8910)", async () => {
  const { app, alice, bot } = standIn;
@@ -110,4 +115,4 @@
test("builds fill the pace slots, yet a review or assigned run still starts; a t
  app.db.query("UPDATE run SET state = 'ok' WHERE id = ?1").run(gone);
  await until(() => stateOf(review) !== "queued", "the review to start");
  expect(stateOf(build)).toBe("queued");
}, 30_000);
}, TEST_MS);
		@@ -184,6 +184,9 @@ export interface CrewHandle {
184	184	*/
185	185	const DRAIN_MS = 3 * 60_000;
186	186
	187	/** How long `stop()` waits for the tick's own sweeps (converge, autonomy) to settle. */
	188	const SWEEPS_STOP_MS = 10_000;
	189
187	190	const handles = new WeakMap<App, CrewHandle>();
188	191
189	192	/** The crew serving this forge, if one was started. */
		@@ -257,20 +260,33 @@ export async function startCrew(
257	260	// Repositories that run on their own take a turn once a minute, on the
258	261	// same clock as the pump, so a pick becomes a run on the next tick.
259	262	let lastAutonomy = 0;
	263	// The sweeps a tick started and has not seen settle: `stop()` waits for
	264	// them, so none reads `app.db` after its caller closed it (#9977).
	265	const sweeps = new Set<Promise<unknown>>();
	266	const track = (sweep: Promise<unknown>): void => {
	267	sweeps.add(sweep);
	268	void sweep.finally(() => sweeps.delete(sweep));
	269	};
260	270	const timer = setInterval(() => {
261	271	if (stopped) return;
262		requeueInterrupted(app);
263		recheckForDisk(app);
264		if (Date.now() - lastAutonomy >= AUTONOMY_TICK_MS) {
265		lastAutonomy = Date.now();
266		settleAll(app).catch((error) =>
267		log.error("converge sweep failed", { error: String(error) }),
268		);
269		autonomyTick(app).catch((error) =>
270		log.error("autonomy tick failed", { error: String(error) }),
271		);
	272	// A tick that throws is logged and the next one tries again: thrown
	273	// from a timer it is an uncaught error that takes the process down.
	274	try {
	275	requeueInterrupted(app);
	276	recheckForDisk(app);
	277	if (Date.now() - lastAutonomy >= AUTONOMY_TICK_MS) {
	278	lastAutonomy = Date.now();
	279	track(settleAll(app).catch((error) =>
	280	log.error("converge sweep failed", { error: String(error) }),
	281	));
	282	track(autonomyTick(app).catch((error) =>
	283	log.error("autonomy tick failed", { error: String(error) }),
	284	));
	285	}
	286	pump(app, ready, inFlight);
	287	} catch (error) {
	288	log.error("crew tick failed", { error: String(error) });
272	289	}
273		pump(app, ready, inFlight);
274	290	}, options.tickMs ?? TICK_MS);
275	291	// Clones no open issue, pull or run claims fill the volume (#5697).
276	292	const sweepClones = (): void => {
		@@ -299,6 +315,16 @@ export async function startCrew(
299	315	unsubscribe();
300	316	unfollow();
301	317	unsettle();
	318	// A sweep still going would read the database its caller closes next;
	319	// bounded, for one stuck on the network must not hold a restart.
	320	if (sweeps.size > 0) {
	321	let bound: ReturnType<typeof setTimeout> \| undefined;
	322	await Promise.race([
	323	Promise.all(sweeps),
	324	new Promise((resolve) => (bound = setTimeout(resolve, SWEEPS_STOP_MS))),
	325	]);
	326	clearTimeout(bound);
	327	}
302	328	if (options.drain === true && inFlight.size > 0) {
303	329	// Let what is in flight finish: a run cut short is a person's
304	330	// money spent twice. The new forge takes new work meanwhile.
		@@ -0,0 +1,61 @@
	1	/**
	2	* The crew's tick never takes the process down (#9977): a tick whose
	3	* database refuses it is logged, and the next tick does its work. Thrown
	4	* from the timer, the error would be uncaught: a forge would fall over,
	5	* and a test shard would exit without naming a test.
	6	*/
	7	import { afterAll, beforeAll, expect, test } from "bun:test";
	8	import { mkdtemp, rm } from "node:fs/promises";
	9	import { tmpdir } from "node:os";
	10	import { join } from "node:path";
	11	import { createApp, type App } from "../src/app.ts";
	12	import { createRepo } from "../src/actions/repos.ts";
	13	import { claimRun, createRun, getRun } from "../src/core/runs.ts";
	14	import { createUser } from "../src/core/users.ts";
	15	import { startCrew, type CrewHandle } from "../src/crew/dispatch.ts";
	16
	17	let dataDir: string;
	18	let app: App;
	19	let crew: CrewHandle;
	20
	21	beforeAll(async () => {
	22	dataDir = await mkdtemp(join(tmpdir(), "forge-crew-tick-"));
	23	app = createApp({ dataDir, port: 0 });
	24	crew = await startCrew(app, {
	25	forgeUrl: "http://127.0.0.1:9",
	26	localUrl: "http://127.0.0.1:9",
	27	binary: join(dataDir, "no-claude-here"),
	28	concurrency: 0,
	29	budgetUsd: 1,
	30	runTimeoutMs: 60_000,
	31	tickMs: 20,
	32	});
	33	});
	34
	35	afterAll(async () => {
	36	await crew.stop();
	37	app.close();
	38	await rm(dataDir, { recursive: true, force: true });
	39	});
	40
	41	test("a tick that throws is logged, and the next tick still does its work", async () => {
	42	const a = await createUser(app.db, { origin: "seed", trusted: true, username: "alice", email: "[email protected]", password: "a-good-passphrase", adminIfFirst: true });
	43	if (!a.ok) throw new Error(a.error);
	44	const made = await createRepo(app, a.value, { name: "ticks", visibility: "private" });
	45	if (!made.ok) throw new Error(String(made.error));
	46	// A run a dead forge left running: its lease lapsed a day ago, so every
	47	// tick's first step writes, to cancel it and queue its task again.
	48	const stale = createRun(app.db, { repoId: made.value.id, kind: "agent", reason: "push", actorId: a.value.id });
	49	claimRun(app.db, stale.id);
	50	const old = new Date(Date.now() - 86_400_000).toISOString();
	51	app.db.query("UPDATE run SET heartbeat_at = ?1, started_at = ?1 WHERE id = ?2").run(old, stale.id);
	52	// The database refuses every write: each tick throws on its first step.
	53	app.db.exec("PRAGMA query_only = ON");
	54	await Bun.sleep(200);
	55	app.db.exec("PRAGMA query_only = OFF");
	56	expect(getRun(app.db, stale.id)?.state).toBe("running");
	57	// The crew is still ticking: the lapsed run is put back on the queue.
	58	const deadline = Date.now() + 5_000;
	59	while (getRun(app.db, stale.id)?.state === "running" && Date.now() < deadline) await Bun.sleep(20);
	60	expect(getRun(app.db, stale.id)?.state).toBe("cancelled");
	61	});
		@@ -123,7 +123,9 @@ test("home says the whole crew's number", () => {
123	123	});
124	124
125	125	test("a repository that hit its ceiling yesterday restarts at its own time, spread over the morning", () => {
126		const midnight = new Date(Date.UTC(2026, 10, 3, 0, 5, 0));
	126	// Midnight itself: repository ids are random, so an instant past 00:00 would
	127	// find a repository whose offset has already come and nothing held.
	128	const midnight = new Date(Date.UTC(2026, 10, 3, 0, 0, 0));
127	129	for (const r of repos) {
128	130	const run = createRun(app.db, { repoId: r.id, kind: "agent" });
129	131	app.db.query("UPDATE run SET cost_usd = 1000, created_at = ?2, state = 'ok' WHERE id = ?1").run(run.id, new Date(Date.UTC(2026, 10, 2, 20, 0, 0)).toISOString());
		@@ -147,13 +149,17 @@ test("a repository that hit its ceiling yesterday restarts at its own time, spre
147	149	});
148	150
149	151	test("a ceiling of 0 allows no spend and is held by the ceiling, so it is never also held until an offset", () => {
150		const midnight = new Date(Date.UTC(2026, 10, 3, 0, 5, 0));
	152	// Midnight itself: repository ids are random, so an instant past 00:00 would
	153	// find a repository whose offset has already come and nothing held.
	154	const midnight = new Date(Date.UTC(2026, 10, 3, 0, 0, 0));
151	155	forgetStagger();
152	156	expect(staggeredUntil(app.db, { id: repos[0]!.id, autonomyCeilingUsd: 0 }, midnight)).toBeNull();
153	157	});
154	158
155	159	test("the stagger reads yesterday's spend once for one instant", () => {
156		const midnight = new Date(Date.UTC(2026, 10, 3, 0, 5, 0));
	160	// Midnight itself: repository ids are random, so an instant past 00:00 would
	161	// find a repository whose offset has already come and nothing held.
	162	const midnight = new Date(Date.UTC(2026, 10, 3, 0, 0, 0));
157	163	const run = createRun(app.db, { repoId: repos[3]!.id, kind: "agent" });
158	164	app.db.query("UPDATE run SET cost_usd = 1000, created_at = ?2, state = 'ok' WHERE id = ?1").run(run.id, new Date(Date.UTC(2026, 10, 2, 20, 0, 0)).toISOString());
159	165	forgetStagger();
		@@ -812,6 +812,10 @@ describe("absorbing an app", () => {
812	812	} finally {
813	813	app.db.run("UPDATE repo SET visibility = 'private' WHERE id = ?1", [repo.id]);
814	814	stub.stop(true);
	815	// The constructor makes it this forge's runtime: a deploy it was asked
	816	// for must settle before afterAll closes the database, or its
	817	// finishRun is an unhandled error in whichever file runs next.
	818	await runtime.stop();
815	819	}
816	820	});
817	821
		@@ -15,8 +15,13 @@ import { startStandInCrew, stopStandInCrew, type StandInCrew } from "./helpers/c
15	15
16	16	const DAY = 86_400_000;
17	17
	18	/** Each test's own timeout; every wait in it fits inside, with room to spare. */
	19	const TEST_MS = 30_000;
	20	/** One wait's deadline: a test waits at most three times, so a slow tick fails as this test, by name, before TEST_MS (#9977). */
	21	const WAIT_MS = 8_000;
	22
18	23	/** Wait until `done` holds, or fail at the deadline: the pump's tick is what we wait on, not a guessed pause. */
19		async function until(done: () => boolean, what: string, ms = 20_000): Promise<void> {
	24	async function until(done: () => boolean, what: string, ms = WAIT_MS): Promise<void> {
20	25	const deadline = Date.now() + ms;
21	26	while (!done()) {
22	27	if (Date.now() > deadline) throw new Error(`timed out waiting for ${what}`);
		@@ -27,8 +32,8 @@ let standIn: StandInCrew;
27	32
28	33	beforeAll(async () => {
29	34	standIn = await startStandInCrew({ tickMs: 50 });
30		}, 30_000);
31		afterAll(() => stopStandInCrew(standIn));
	35	}, TEST_MS);
	36	afterAll(() => stopStandInCrew(standIn), TEST_MS);
32	37
33	38	test("past the week's reserve an assigned run stays queued, with the reason in its log", async () => {
34	39	const { app, alice, bot } = standIn;
		@@ -60,7 +65,7 @@ test("past the week's reserve an assigned run stays queued, with the reason in i
60	65	forgetWeek();
61	66	await until(() => mine()?.state !== "queued", "the run to start");
62	67	expect(mine()?.state).not.toBe("queued");
63		}, 30_000);
	68	}, TEST_MS);
64	69
65	70	test("builds fill the pace slots, yet a review or assigned run still starts; a third of them waits (#8910)", async () => {
66	71	const { app, alice, bot } = standIn;
		@@ -110,4 +115,4 @@ test("builds fill the pace slots, yet a review or assigned run still starts; a t
110	115	app.db.query("UPDATE run SET state = 'ok' WHERE id = ?1").run(gone);
111	116	await until(() => stateOf(review) !== "queued", "the review to start");
112	117	expect(stateOf(build)).toBe("queued");
113		}, 30_000);
	118	}, TEST_MS);