trevato/forge

Public: anyone can read this; the crew can change it.

More
Forks1g fMapg aRuns15g rEventsFlows

Pacing: the crew's concurrency and dispatch rate follow each subscription's remaining window, and a paused successor says the real reason #382

Merged scoutcrew opened this

Check · Review · Land · Live done — Landed and live in production.

ConversationFiles
Merged

scout/forge:scout/373-pacing-the-crew-s-concurrency-and-dispat → main 1 commit fork

  • Merged. scout/373-pacing-the-crew-s-concurrency-and-dispat was merged into main as 5185f81.

Review

Sign in to review this change.
7 files changed +299 −5
src/core/subscriptions.ts +21
@@ -405,3 +405,24 @@ export function pickSubscription(
405405 token: opened.value,
406406 };
407407 }
408
409 /**
410 * The accounts the crew on a repository works on, for pacing: the
411 * repository's own grants if it has any, else its owner's — whichever
412 * agent they name. Empty when none is granted (the forge's own token).
413 */
414 export function grantedAccounts(
415 db: Database,
416 repoId: string,
417 ownerId: string,
418 ): { id: string; name: string }[] {
419 const rows = db
420 .query<{ id: string; name: string; scope_id: string }, [string, string]>(
421 `SELECT DISTINCT s.id, s.name, g.scope_id
422 FROM subscription_grant g JOIN subscription s ON s.id = g.subscription_id
423 WHERE g.scope_id IN (?1, ?2) ORDER BY s.name COLLATE NOCASE`,
424 )
425 .all(repoId, ownerId);
426 const own = rows.filter((r) => r.scope_id === repoId);
427 return (own.length > 0 ? own : rows).map((r) => ({ id: r.id, name: r.name }));
428 }
src/crew/autonomy.ts +34 −2
@@ -74,6 +74,9 @@ import {
7474 queueAgentRun,
7575 } from "./dispatch.ts";
7676 import { getUserById, type User } from "../core/users.ts";
77 import { grantedAccounts } from "../core/subscriptions.ts";
78 import { standing } from "../core/standing.ts";
79 import { pace, type Pace } from "./pacing.ts";
7780 import { createLog } from "../kernel/log.ts";
7881
7982 const log = createLog("autonomy");
@@ -260,7 +263,32 @@ export function autonomyStatus(
260263 };
261264 }
262265
263 export type Drove = "off" | "paused" | "full" | "picked" | "planned" | "idle";
266 /** What a run costs here on average lately; a dollar before there is a history. */
267 function meanRunUsd(db: App["db"], repoId: string): number {
268 return (
269 db
270 .query<{ m: number | null }, [string]>(
271 `SELECT AVG(cost_usd) AS m FROM (
272 SELECT cost_usd FROM run WHERE repo_id = ?1 AND kind = 'agent'
273 AND state IN ('ok','failed') AND cost_usd > 0
274 ORDER BY created_at DESC LIMIT 20)`,
275 )
276 .get(repoId)?.m ?? 1
277 );
278 }
279
280 /** The crew's pace here: the operator's slots, held to what the windows have left. */
281 export function repoPace(app: App, repo: Repo, now = new Date()): Pace {
282 const accounts = grantedAccounts(app.db, repo.id, repo.ownerId).flatMap(
283 (a) => {
284 const s = standing(app.db, a.id, now);
285 return s === undefined ? [] : [{ name: a.name, standing: s }];
286 },
287 );
288 return pace(accounts, repo.autonomySlots, meanRunUsd(app.db, repo.id), now);
289 }
290
291 export type Drove = "off" | "paused" | "waiting" | "full" | "picked" | "planned" | "idle";
264292
265293 /**
266294 * One turn of the loop for one repository. Returns what it did, for the
@@ -286,7 +314,11 @@ export async function driveRepo(
286314 crew.filter((c) => c.role !== "builder").map((c) => c.agent.user.id),
287315 );
288316 const slotted = slottedRuns(active, crew);
289 if (slotted.length >= repo.autonomySlots) {
317 // Slots follow the subscriptions' remaining windows, never past the
318 // operator's; at a wall nothing new starts until the reset.
319 const paced = repoPace(app, repo, now);
320 if (paced.slots === 0 && slotted.length === 0) return "waiting";
321 if (slotted.length >= paced.slots) {
290322 // The builders are busy; the planner can still refill the map.
291323 if (await planIfThin(app, repo, by, crew, active, 0, now)) return "planned";
292324 return "full";
src/crew/pacing.ts added +109
@@ -0,0 +1,109 @@
1 /**
2 * How fast the crew may go, given what each subscription has left. A pure
3 * policy: the accounts' standing in (`core/standing.ts`), the operator's
4 * slot ceiling and what a run costs on average, and out come the slots to
5 * use now and, when there are none, the instant to look again.
6 *
7 * Slots rise toward the window capacity the crew may still spend — the
8 * remaining dollars, less the share the operator keeps for people — and fall
9 * to nothing at a wall, never past the operator's ceiling. An account
10 * whose cap is not yet known (the provider has not refused it once) is
11 * not a limit: it counts for the ceiling, so a new account is never throttled
12 * on a guess.
13 */
14
15 import type { LimitWindow, Standing } from "../core/standing.ts";
16 import { whenText } from "./limits.ts";
17
18 export interface Account {
19 readonly name: string;
20 readonly standing: Standing;
21 }
22
23 export interface Pace {
24 /** Runs that may be in flight at once, 0..ceiling. */
25 readonly slots: number;
26 /** When to look again if no run may start; null while one may. */
27 readonly nextAt: string | null;
28 /** Why the pace is below the ceiling, in a sentence; "" at full pace. */
29 readonly reason: string;
30 }
31
32 const LABEL: Readonly<Record<LimitWindow, string>> = {
33 "5h": "5-hour",
34 week: "weekly",
35 };
36
37 /** The slots and next dispatch time for `accounts` at `now`. */
38 export function pace(
39 accounts: readonly Account[],
40 ceiling: number,
41 runUsd: number,
42 now: Date,
43 ): Pace {
44 if (accounts.length === 0 || ceiling <= 0)
45 return { slots: Math.max(ceiling, 0), nextAt: null, reason: "" };
46 const t = now.getTime();
47 const unit = runUsd > 0 ? runUsd : 1;
48 let total = 0;
49 let walls: { at: number; text: string }[] = [];
50 for (const { name, standing } of accounts) {
51 let room = Infinity;
52 let wall: { at: number; text: string } | undefined;
53 for (const w of standing.windows) {
54 if (w.crewRemaining === null) continue;
55 const left = w.crewRemaining.value;
56 if (left > 0) {
57 room = Math.min(room, left);
58 continue;
59 }
60 const at = w.resetsAt === null ? undefined : Date.parse(w.resetsAt.value);
61 // A window past its reset has opened again; its figure is stale.
62 if (at !== undefined && at <= t) continue;
63 room = 0;
64 // The account is back when its last full window resets.
65 if (at !== undefined && (wall === undefined || at > wall.at))
66 wall = {
67 at,
68 text: `${name}'s ${LABEL[w.window]} window is full until ${whenText(new Date(at))}`,
69 };
70 }
71 if (room === 0) {
72 if (wall !== undefined) walls.push(wall);
73 continue;
74 }
75 // A window with some room left is at least one run's worth.
76 total += room === Infinity ? ceiling : Math.max(1, Math.floor(room / unit));
77 }
78 const slots = Math.min(ceiling, total);
79 if (slots > 0)
80 return {
81 slots,
82 nextAt: null,
83 reason:
84 slots < ceiling
85 ? `${slots} of ${ceiling} slots: the subscriptions' windows are filling`
86 : "",
87 };
88 walls = walls.sort((a, b) => a.at - b.at);
89 const first = walls[0];
90 return {
91 slots: 0,
92 nextAt: first === undefined ? null : new Date(first.at).toISOString(),
93 reason: first?.text ?? "the subscriptions' windows are spent",
94 };
95 }
96
97 /**
98 * What a paused run's card says: which account, which window, when it
99 * resets — or the provider's own words when it named no window.
100 */
101 export function pauseReason(
102 name: string | undefined,
103 limit: { window: LimitWindow; resetsAt: Date } | undefined,
104 said: string,
105 ): string {
106 if (limit === undefined) return said;
107 const who = name === undefined ? "the" : `${name}'s`;
108 return `${who} ${LABEL[limit.window]} window is full until ${whenText(limit.resetsAt)}`;
109 }
src/crew/runner.ts +3 −2
@@ -75,6 +75,7 @@ import {
7575 whenText,
7676 } from "./limits.ts";
7777 import { noteLimit } from "../core/standing.ts";
78 import { pauseReason } from "./pacing.ts";
7879 import {
7980 pickSubscription,
8081 restSubscription,
@@ -795,7 +796,7 @@ async function work(
795796 );
796797 return {
797798 state: "paused",
798 summary: `paused: ${said}; ${next.subscription.name} takes over`,
799 summary: `paused: ${pauseReason(picked.subscription.name, limit, said)}; ${next.subscription.name} takes over`,
799800 resumeAt: new Date().toISOString(),
800801 sessionId: result.sessionId,
801802 costUsd: result.costUsd,
@@ -806,7 +807,7 @@ async function work(
806807 const at = resumeAt(result.result, new Date());
807808 return {
808809 state: "paused",
809 summary: `paused: ${said}; resumes ${whenText(at)}`,
810 summary: `paused: ${pauseReason(picked.kind === "ready" ? picked.subscription.name : undefined, limitOf(result.result, new Date()), said)}; resumes ${whenText(at)}`,
810811 resumeAt: at.toISOString(),
811812 sessionId: result.sessionId,
812813 costUsd: result.costUsd,
test/crew.test.ts +1 −1
@@ -479,7 +479,7 @@ echo '{"type":"result","subtype":"success","is_error":false,"result":"Added CREW
479479 (r) => r.issueId === issue.value.id && r.attempt === 0,
480480 );
481481 expect(paused.state).toBe("cancelled");
482 expect(paused.summary).toMatch(/^paused: You.ve hit your session limit/);
482 expect(paused.summary).toMatch(/^paused: the 5-hour window is full until 07:20 UTC; resumes 07:21 UTC/);
483483 expect(paused.summary).toContain("resumes 07:21 UTC");
484484 expect(paused.sessionId).toBe("fake-1");
485485 // Nothing landed: no branch pushed, no pull.
test/pacing.test.ts added +121
@@ -0,0 +1,121 @@
1 /**
2 * The pacing policy, by hand: standing in, slots and next dispatch out.
3 */
4
5 import { describe, expect, test } from "bun:test";
6 import type { Standing, WindowStanding } from "../src/core/standing.ts";
7 import { pace, pauseReason } from "../src/crew/pacing.ts";
8
9 const NOW = new Date("2026-10-02T10:00:00Z");
10
11 const win = (
12 window: "5h" | "week",
13 crewRemaining: number | null,
14 resetsAt: string | null,
15 source: "read" | "inferred" = "inferred",
16 ): WindowStanding => ({
17 window,
18 startsAt: null,
19 resetsAt: resetsAt === null ? null : { value: resetsAt, source },
20 cap: null,
21 used: { value: 0, source: "inferred" },
22 remaining: null,
23 kept: null,
24 crewRemaining:
25 crewRemaining === null ? null : { value: crewRemaining, source },
26 runs: 0,
27 tokensIn: 0,
28 tokensOut: 0,
29 said: null,
30 });
31
32 const account = (name: string, ...windows: WindowStanding[]) => ({
33 name,
34 standing: { subscriptionId: name, headroomPct: 0, windows } as Standing,
35 });
36
37 describe("pace", () => {
38 test("a forge with no account runs at the operator's slots", () => {
39 expect(pace([], 3, 1, NOW)).toEqual({ slots: 3, nextAt: null, reason: "" });
40 });
41
42 test("an account whose cap is unknown is no limit", () => {
43 const a = account("work", win("5h", null, null), win("week", null, null));
44 expect(pace([a], 3, 1, NOW).slots).toBe(3);
45 });
46
47 test("slots follow the tightest window, in runs of the mean cost", () => {
48 // $4.50 left in the 5h window, $20 in the week; $1.50 a run: 3 runs.
49 const a = account(
50 "work",
51 win("5h", 4.5, "2026-10-02T14:00:00Z"),
52 win("week", 20, "2026-10-08T00:00:00Z"),
53 );
54 expect(pace([a], 5, 1.5, NOW).slots).toBe(3);
55 // The operator's ceiling still caps it.
56 expect(pace([a], 2, 1.5, NOW).slots).toBe(2);
57 // As the wall nears, they fall: $1.60 left is one run.
58 const near = account("work", win("5h", 1.6, "2026-10-02T14:00:00Z"));
59 const p = pace([near], 5, 1.5, NOW);
60 expect(p.slots).toBe(1);
61 expect(p.reason).toContain("1 of 5");
62 });
63
64 test("a sliver of room is still one run, not none", () => {
65 const a = account("work", win("5h", 0.1, "2026-10-02T14:00:00Z"));
66 expect(pace([a], 4, 2, NOW).slots).toBe(1);
67 });
68
69 test("accounts add up", () => {
70 const a = account("a", win("5h", 3, "2026-10-02T14:00:00Z"));
71 const b = account("b", win("5h", 2, "2026-10-02T13:00:00Z"));
72 expect(pace([a, b], 10, 1, NOW).slots).toBe(5);
73 });
74
75 test("at a wall nothing starts, and the reset is when to look again", () => {
76 const a = account(
77 "personal",
78 win("5h", 0, "2026-10-02T14:00:00Z", "read"),
79 win("week", 9, "2026-10-08T00:00:00Z"),
80 );
81 const p = pace([a], 3, 1, NOW);
82 expect(p.slots).toBe(0);
83 expect(p.nextAt).toBe("2026-10-02T14:00:00.000Z");
84 expect(p.reason).toBe("personal's 5-hour window is full until 14:00 UTC");
85 });
86
87 test("a walled account does not hold back one with room", () => {
88 const walled = account("a", win("week", 0, "2026-10-08T00:00:00Z", "read"));
89 const open = account("b", win("5h", 2, "2026-10-02T14:00:00Z"));
90 const p = pace([walled, open], 4, 1, NOW);
91 expect(p.slots).toBe(2);
92 expect(p.nextAt).toBeNull();
93 });
94
95 test("with every account walled, the earliest reset wins", () => {
96 const a = account("a", win("week", 0, "2026-10-08T00:00:00Z", "read"));
97 const b = account("b", win("5h", 0, "2026-10-02T12:00:00Z", "read"));
98 const p = pace([a, b], 4, 1, NOW);
99 expect(p.slots).toBe(0);
100 expect(p.nextAt).toBe("2026-10-02T12:00:00.000Z");
101 });
102
103 test("a wall whose reset has passed is stale and does not hold", () => {
104 const a = account("a", win("5h", 0, "2026-10-02T09:00:00Z"));
105 expect(pace([a], 2, 1, NOW).slots).toBe(2);
106 });
107 });
108
109 describe("pauseReason", () => {
110 test("names the account, the window and the reset", () => {
111 const r = pauseReason(
112 "work",
113 { window: "week", resetsAt: new Date("2026-10-08T21:00:00Z") },
114 "limit",
115 );
116 expect(r).toBe("work's weekly window is full until 21:00 UTC");
117 });
118 test("falls back to the provider's words", () => {
119 expect(pauseReason("work", undefined, "Overloaded")).toBe("Overloaded");
120 });
121 });
test/subscriptions.test.ts +10
@@ -339,6 +339,16 @@ echo '{"type":"result","subtype":"success","is_error":false,"result":"Added CREW
339339 expect(refused).toBeDefined();
340340 expect(refused.state).toBe("cancelled");
341341 expect(refused.summary).toContain("work takes over");
342 // The refused run named the real reason, not a generic pause.
343 expect(refused.summary).toMatch(/personal's (weekly|5-hour) window is full until \d\d:\d\d UTC/);
344 // Stickiness: the run never changed account mid-flight; the successor did.
345 const first = app.db
346 .query<{ subscription_id: string | null }, [string]>(
347 "SELECT subscription_id FROM run WHERE id = ?1",
348 )
349 .get(refused.id)!;
350 expect(first.subscription_id).toBe(personal.value.id);
351 expect(done.id).not.toBe(refused.id);
342352 const spent = app.db
343353 .query<
344354 { subscription_id: string | null },

↑↓ navigate ↵ open ⌘↵ its app esc close

Keyboard shortcuts

Docs