trevato/forge

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

More
Forks1g fMapg aRuns11g rEventsFlows

The crew starts every run through one agent-runner seam, with today's CLI as its first runner #12132

Parked masoncrew opened this
ConversationFiles
Closed without merging

mason/forge:mason/12122-the-crew-starts-every-run-through-one-ag → main 2 commits fork

  • Closed without merging. Nothing was applied to main.

Review

Sign in to review this change.
11 files changed +194 −110
src/crew/agent_runner.ts added +138
@@ -0,0 +1,138 @@
1 /**
2 * The agent-runner seam (#12122): every crew run starts here. A runner takes
3 * a brief (prompt, system prompt, model, budget) and a sandbox, streams the
4 * run's steps (`log`, `onSaid`), reports spend (`onUsage`, `onMeasure`),
5 * pauses (`onCleared`, a rate-limited outcome) and its end (`AgentOutcome`),
6 * as plain data in and out. `crew/claude.ts` is the first runner; callers
7 * import this module, never a runner.
8 */
9
10 import type { Measure } from "../core/measure.ts";
11 import type { LimitWindow } from "../core/standing.ts";
12 import type { RunLimits } from "./run-cell.ts";
13 import type { SandboxPolicy } from "./sandbox.ts";
14 import { detectClaude, runClaude } from "./claude.ts";
15
16 export interface AgentRun {
17 readonly binary: string;
18 readonly cwd: string;
19 readonly prompt: string;
20 readonly appendSystemPrompt: string;
21 readonly model?: string | undefined;
22 readonly budgetUsd: number;
23 /** A session id to continue. */
24 readonly resume?: string | undefined;
25 readonly env: Record<string, string>;
26 readonly sandbox: SandboxPolicy;
27 /** The run's machine's memory and CPU ceilings (`crew/run-cell.ts`); none is today's cell. */
28 readonly limits?: RunLimits | undefined;
29 /** What `/system` calls this run's process tree (`crew:<name>`); unlabelled when absent. */
30 readonly label?: string | undefined;
31 readonly timeoutMs: number;
32 readonly log: (line: string) => void;
33 readonly signal: AbortSignal;
34 /** Images that go with the prompt, once: base64 data and its media type (#8820). */
35 readonly images?: readonly { readonly mediaType: string; readonly data: string }[] | undefined;
36 /**
37 * Hands the caller a way to say something to the agent while it runs.
38 * The words arrive as its next user turn, after the step it is on; a
39 * send after the session ended returns false.
40 */
41 readonly onSteer?: ((send: (text: string) => boolean) => void) | undefined;
42 /**
43 * The session's spend so far, each time it reports some: tokens as each
44 * assistant message's `usage` arrives, the cost once the result names it
45 * (Claude Code reports `total_cost_usd` only at the end). Running totals
46 * for this invocation, never deltas.
47 */
48 readonly onUsage?: ((spent: Spent) => void) | undefined;
49 /** The provider's measurement of a window, each time the session reports one (#8960). */
50 readonly onMeasure?: ((measure: Measure) => void) | undefined;
51 /** The provider said this window is under its thresholds: a stored reading of it is stale. */
52 readonly onCleared?: ((window: LimitWindow) => void) | undefined;
53 /**
54 * Talks to the forge through MCP alone (#7651): no file or shell tools, no
55 * clone, the one server named in `config` (`--mcp-config` JSON) allowed.
56 */
57 readonly mcp?: { readonly config: string; readonly allow: readonly string[] } | undefined;
58 /** The session id, once the init record names it (a restart resumes it). */
59 readonly onSession?: ((sessionId: string) => void) | undefined;
60 /** What the agent said or called, as each assistant message arrives. */
61 readonly onSaid?: ((said: { readonly kind: "text" | "tool" | "thinking"; readonly text: string }) => void) | undefined;
62 /**
63 * How long a session may stay silent after a result while its input is
64 * still open (a steer was pending) before it is closed and settled as
65 * done. A steer folded into the turn it interrupted leaves a result owed
66 * that never comes. Default the runner's own default (Claude: five minutes).
67 */
68 readonly idleMs?: number | undefined;
69 /**
70 * How long the process has to exit once its input is closed after the
71 * final result. Past it the process group is stopped and the run settles
72 * on the result it already gave. Default is the runner's own (Claude: five seconds).
73 */
74 readonly settleMs?: number | undefined;
75 }
76
77
78 export interface Spent {
79 readonly tokensIn: number;
80 readonly tokensOut: number;
81 readonly costUsd: number;
82 }
83
84 export interface AgentOutcome {
85 /** Whether the run ended with a result rather than an error. */
86 readonly ok: boolean;
87 readonly sessionId: string;
88 /** The agent's final message. */
89 readonly result: string;
90 readonly costUsd: number;
91 readonly turns: number;
92 /**
93 * Tokens the model read (fresh input, cache writes and cache reads) and
94 * wrote, from the result's `usage`; 0 when the run reported none.
95 */
96 readonly tokensIn: number;
97 readonly tokensOut: number;
98 /** What the last assistant message read: the session's context at its end; 0 if none. */
99 readonly contextTokens: number;
100 /** The model's context window in tokens, from the result's `modelUsage`; 0 when it named none. */
101 readonly contextWindow: number;
102 /** `completed`, `max_budget_reached`, `timeout`, `cancelled`, `error`, `exit <n>`. */
103 readonly reason: string;
104 }
105
106
107 /** The run spent its budget and the runner stopped it. */
108 export const BUDGET_REACHED = "max_budget_reached";
109
110 export type AgentDetection =
111 | { ok: true; version: string }
112 | { ok: false; error: string };
113
114 /** What a runner implements. `run` never throws; every ending is an outcome. */
115 export interface AgentRunner {
116 run(input: AgentRun): Promise<AgentOutcome>;
117 /** Whether `binary` is an agent this runner can drive. */
118 detect(binary: string): Promise<AgentDetection>;
119 }
120
121 // Wrapped, not bound, so the two modules may load in either order.
122 const claude: AgentRunner = {
123 run: (input) => runClaude(input),
124 detect: (binary) => detectClaude(binary),
125 };
126
127 /** The runner every run uses today. */
128 export function currentRunner(): AgentRunner {
129 return claude;
130 }
131
132 export function runAgent(input: AgentRun): Promise<AgentOutcome> {
133 return currentRunner().run(input);
134 }
135
136 export function detectAgent(binary: string): Promise<AgentDetection> {
137 return currentRunner().detect(binary);
138 }
src/crew/claude.ts +15 −93
@@ -12,11 +12,10 @@
1212 * settings files of the operator's, a spend budget, and a deadline.
1313 */
1414
15 import { clearedOf, measureOf, type Measure } from "../core/measure.ts";
16 import type { LimitWindow } from "../core/standing.ts";
15 import { clearedOf, measureOf } from "../core/measure.ts";
1716 import { backgroundArgv, crewCpus } from "../apps/background.ts";
1817 import { cgroupHost } from "../apps/cgroup.ts";
19 import { openRunCell, type RunLimits } from "./run-cell.ts";
18 import { openRunCell } from "./run-cell.ts";
2019 import { watchLabelled } from "../apps/contain.ts";
2120 import { childEnv, pump } from "../apps/process.ts";
2221 import { endedLine, liveTree } from "../runs/ended.ts";
@@ -24,102 +23,25 @@ import { CANCEL_GRACE_MS } from "../runs/grace.ts";
2423 import { spellings, stripForgeDirs } from "../runs/paths.ts";
2524 import type { SandboxPolicy } from "./sandbox.ts";
2625
27 export interface ClaudeInput {
28 readonly binary: string;
29 readonly cwd: string;
30 readonly prompt: string;
31 readonly appendSystemPrompt: string;
32 readonly model?: string | undefined;
33 readonly budgetUsd: number;
34 /** A session id to continue. */
35 readonly resume?: string | undefined;
36 readonly env: Record<string, string>;
37 readonly sandbox: SandboxPolicy;
38 /** The run's machine's memory and CPU ceilings (`crew/run-cell.ts`); none is today's cell. */
39 readonly limits?: RunLimits | undefined;
40 /** What `/system` calls this run's process tree (`crew:<name>`); unlabelled when absent. */
41 readonly label?: string | undefined;
42 readonly timeoutMs: number;
43 readonly log: (line: string) => void;
44 readonly signal: AbortSignal;
45 /** Images that go with the prompt, once: base64 data and its media type (#8820). */
46 readonly images?: readonly { readonly mediaType: string; readonly data: string }[] | undefined;
47 /**
48 * Hands the caller a way to say something to the agent while it runs.
49 * The words arrive as its next user turn, after the step it is on; a
50 * send after the session ended returns false.
51 */
52 readonly onSteer?: ((send: (text: string) => boolean) => void) | undefined;
53 /**
54 * The session's spend so far, each time it reports some: tokens as each
55 * assistant message's `usage` arrives, the cost once the result names it
56 * (Claude Code reports `total_cost_usd` only at the end). Running totals
57 * for this invocation, never deltas.
58 */
59 readonly onUsage?: ((spent: Spent) => void) | undefined;
60 /** The provider's measurement of a window, each time the session reports one (#8960). */
61 readonly onMeasure?: ((measure: Measure) => void) | undefined;
62 /** The provider said this window is under its thresholds: a stored reading of it is stale. */
63 readonly onCleared?: ((window: LimitWindow) => void) | undefined;
64 /**
65 * Talks to the forge through MCP alone (#7651): no file or shell tools, no
66 * clone, the one server named in `config` (`--mcp-config` JSON) allowed.
67 */
68 readonly mcp?: { readonly config: string; readonly allow: readonly string[] } | undefined;
69 /** The session id, once the init record names it (a restart resumes it). */
70 readonly onSession?: ((sessionId: string) => void) | undefined;
71 /** What the agent said or called, as each assistant message arrives. */
72 readonly onSaid?: ((said: { readonly kind: "text" | "tool" | "thinking"; readonly text: string }) => void) | undefined;
73 /**
74 * How long a session may stay silent after a result while its input is
75 * still open (a steer was pending) before it is closed and settled as
76 * done. A steer folded into the turn it interrupted leaves a result owed
77 * that never comes. Default `IDLE_AFTER_RESULT_MS`.
78 */
79 readonly idleMs?: number | undefined;
80 /**
81 * How long the process has to exit once its input is closed after the
82 * final result. Past it the process group is stopped and the run settles
83 * on the result it already gave. Default `SETTLE_MS`.
84 */
85 readonly settleMs?: number | undefined;
86 }
26 import type {
27 AgentOutcome,
28 AgentRun,
29 BUDGET_REACHED as SeamBudgetReached,
30 Spent,
31 } from "./agent_runner.ts";
32
33 /** Claude's names for the seam's types. */
34 export type ClaudeInput = AgentRun;
35 export type ClaudeOutcome = AgentOutcome;
36 export type { Spent };
8737
8838 /** The idle guard's default: five minutes of silence after a result. */
8939 export const IDLE_AFTER_RESULT_MS = 5 * 60_000;
9040 /** The grace a finished session has to exit on its own. */
9141 export const SETTLE_MS = 5_000;
9242
93 export interface Spent {
94 readonly tokensIn: number;
95 readonly tokensOut: number;
96 readonly costUsd: number;
97 }
98
99 export interface ClaudeOutcome {
100 /** Whether the run ended with a result rather than an error. */
101 readonly ok: boolean;
102 readonly sessionId: string;
103 /** The agent's final message. */
104 readonly result: string;
105 readonly costUsd: number;
106 readonly turns: number;
107 /**
108 * Tokens the model read (fresh input, cache writes and cache reads) and
109 * wrote, from the result's `usage`; 0 when the run reported none.
110 */
111 readonly tokensIn: number;
112 readonly tokensOut: number;
113 /** What the last assistant message read: the session's context at its end; 0 when none reported. */
114 readonly contextTokens: number;
115 /** The model's context window in tokens, from the result's `modelUsage`; 0 when it named none. */
116 readonly contextWindow: number;
117 /** `completed`, `max_budget_reached`, `timeout`, `cancelled`, `error`, `exit <n>`. */
118 readonly reason: string;
119 }
120
121 /** The run spent its budget and Claude Code stopped it. */
122 export const BUDGET_REACHED = "max_budget_reached";
43 /** The seam's value, spelled here so this module needs nothing from it at run time. */
44 export const BUDGET_REACHED: typeof SeamBudgetReached = "max_budget_reached";
12345
12446 /**
12547 * The tools an agent has. Enough to read, change, run and push code; no
src/crew/conversation.ts +4 −4
@@ -1,5 +1,5 @@
11 /**
2 * A person's message becomes one agent turn (#7651): one `runClaude`, resumed
2 * A person's message becomes one agent turn (#7651): one `runAgent`, resumed
33 * from the conversation's stored session, with no clone and no shell — only
44 * this forge's `/mcp`, reached with a `via` token of that person, so the
55 * agent can do what they can and no more (`core/via_token.ts`, the fence in
@@ -37,7 +37,7 @@ import { id } from "../kernel/id.ts";
3737 import { running } from "../kernel/version.ts";
3838 import { err, ok, type Result } from "../kernel/result.ts";
3939 import { appendRunLog } from "../runs/log.ts";
40 import { runClaude, type ClaudeOutcome } from "./claude.ts";
40 import { runAgent, type AgentOutcome } from "./agent_runner.ts";
4141 import { CONVERSATION_BRIEF } from "./conversation_brief.ts";
4242 import { isRateLimited, takeLimit, whenText } from "./limits.ts";
4343 import { credentialEnv, listenForSteer, steerRun } from "./runner.ts";
@@ -511,7 +511,7 @@ function begin(
511511 // nothing to enforce and an unenforced policy cannot fail the turn.
512512 const sandbox = sandboxPolicy({ dataDir: app.config.dataDir, agentName: "conversation", urls: [options.localUrl], enforced: false, runId });
513513 const turn = (async (): Promise<void> => {
514 let outcome: ClaudeOutcome | undefined;
514 let outcome: AgentOutcome | undefined;
515515 let unlisten = (): void => {};
516516 try {
517517 const cwd = turnDir(app.config.dataDir);
@@ -523,7 +523,7 @@ function begin(
523523 for (let session = resumeSession; ; session = undefined) {
524524 let named = false;
525525 if (session !== undefined) appendRunLog(app.config.dataDir, runId, `resuming session ${session}`);
526 outcome = await runClaude({
526 outcome = await runAgent({
527527 binary: options.binary,
528528 cwd,
529529 // Attachments go with the message itself, never with a resume.
src/crew/dispatch.ts +2 −2
@@ -85,7 +85,7 @@ import { isContributionFork } from "../core/contribute.ts";
8585 import { guestSql, lacks } from "../core/roles.ts";
8686 import { createLog } from "../kernel/log.ts";
8787 import { issueOfBranch, type RunReason } from "./brief.ts";
88 import { detectClaude } from "./claude.ts";
88 import { detectAgent } from "./agent_runner.ts";
8989 import { followDivergence, settleAll, settleOnPush } from "./converge.ts";
9090 import { breakerTripped } from "./breaker.ts";
9191 import { REPEAT_NOTE, repeatTripped } from "./repeat.ts";
@@ -209,7 +209,7 @@ export async function startCrew(
209209 app: App,
210210 options: CrewOptions,
211211 ): Promise<CrewHandle> {
212 const detected = await detectClaude(options.binary);
212 const detected = await detectAgent(options.binary);
213213 if (!detected.ok)
214214 log.warn("claude not usable; agent runs will fail", {
215215 error: detected.error,
src/crew/leftovers.ts +1 −1
@@ -1,7 +1,7 @@
11 /**
22 * Claude Code's sandbox bridges that outlived their session (#3708). Each
33 * session starts two `socat UNIX-LISTEN:<its temp dir>/…,fork TCP:…`
4 * relays in its own process group; `runClaude` kills that group once the
4 * relays in its own process group; the runner kills that group once the
55 * session ends (`reapGroup`), and this sweep is the net under it — for a
66 * session a forge before that fix, or a forge that died, left behind.
77 *
src/crew/limits.ts +2 −2
@@ -9,7 +9,7 @@ import type { Database } from "bun:sqlite";
99 import { noteLimit, type LimitWindow } from "../core/standing.ts";
1010 import { PLATFORM, restSubscription } from "../core/subscriptions.ts";
1111 import { isNetworkError } from "./network.ts";
12 import type { ClaudeOutcome } from "./claude.ts";
12 import type { AgentOutcome } from "./agent_runner.ts";
1313
1414 /** How many times one task may be paused before the forge gives up on it. */
1515 export const MAX_PAUSES = 6;
@@ -24,7 +24,7 @@ const SLACK_MS = 60_000;
2424 * failure. Claude Code ends such a run with `api_error`; the message names
2525 * the limit, and is checked too in case the ending is named differently.
2626 */
27 export function isRateLimited(outcome: ClaudeOutcome): boolean {
27 export function isRateLimited(outcome: AgentOutcome): boolean {
2828 if (outcome.ok || isNetworkError(outcome)) return false;
2929 return (
3030 outcome.reason === "api_error" ||
src/crew/network.ts +2 −2
@@ -5,7 +5,7 @@
55 * later goes on. The runner retries before it judges the run.
66 */
77
8 import type { ClaudeOutcome } from "./claude.ts";
8 import type { AgentOutcome } from "./agent_runner.ts";
99
1010 /** The one wording of a dead network; the interrupted-summary reader builds on it. */
1111 export const NETWORK =
@@ -18,7 +18,7 @@ export const NETWORK_RETRY_SECONDS: readonly number[] = [5, 20, 60];
1818 export const NETWORK_PAUSE_MS = 5 * 60_000;
1919
2020 /** Whether the run ended on the network rather than on the work. */
21 export function isNetworkError(outcome: ClaudeOutcome): boolean {
21 export function isNetworkError(outcome: AgentOutcome): boolean {
2222 if (outcome.ok) return false;
2323 return NETWORK.test(outcome.result) || NETWORK.test(outcome.reason);
2424 }
src/crew/one_shot.ts +2 −2
@@ -12,7 +12,7 @@ import { dailyLeft } from "../core/account_ceiling.ts";
1212 import { PLATFORM, pickSubscription } from "../core/subscriptions.ts";
1313 import { id } from "../kernel/id.ts";
1414 import { err, ok, type Result } from "../kernel/result.ts";
15 import { runClaude } from "./claude.ts";
15 import { runAgent } from "./agent_runner.ts";
1616 import { conversationOptions, sessionsDir, turnDir } from "./conversation.ts";
1717 import { isRateLimited } from "./limits.ts";
1818 import { credentialEnv } from "./runner.ts";
@@ -42,7 +42,7 @@ export async function oneShot(app: App, userId: string, system: string, prompt:
4242 .run(runId, userId, subscriptionId, budget, now.toISOString());
4343 let costUsd = budget;
4444 try {
45 const outcome = await runClaude({
45 const outcome = await runAgent({
4646 binary: options.binary,
4747 cwd: turnDir(app.config.dataDir),
4848 prompt,
src/crew/run-cell.ts +1 −1
@@ -1,5 +1,5 @@
11 /**
2 * A crew run's cell (#9469): the cgroup `runClaude` puts the agent in, held
2 * A crew run's cell (#9469): the cgroup the runner puts the agent in, held
33 * to its machine's ceilings (`crew/machines.ts`). A machine can only narrow
44 * it: memory is capped where today it is not, and cores are the lower of the
55 * machine's and the forge's own `FORGE_CREW_CPUS`. A machine that declares
src/crew/runner.ts +3 −3
@@ -140,7 +140,7 @@ import {
140140 NETWORK_RETRY_SECONDS,
141141 pause,
142142 } from "./network.ts";
143 import { BUDGET_REACHED, runClaude, type ClaudeOutcome } from "./claude.ts";
143 import { BUDGET_REACHED, runAgent, type AgentOutcome } from "./agent_runner.ts";
144144 import {
145145 isRateLimited,
146146 limitOf,
@@ -1539,9 +1539,9 @@ async function work(
15391539 prompt: string,
15401540 budgetUsd: number,
15411541 resume: string | undefined,
1542 ): Promise<ClaudeOutcome> => {
1542 ): Promise<AgentOutcome> => {
15431543 const spent = redactResult(
1544 await runClaude({
1544 await runAgent({
15451545 binary: options.binary,
15461546 cwd: dir,
15471547 prompt,
test/agent-runner-seam.test.ts added +24
@@ -0,0 +1,24 @@
1 import { expect, test } from "bun:test";
2 import { Glob } from "bun";
3
4 // Every crew run starts through src/crew/agent_runner.ts (#12122); a runner
5 // implementation is imported by the seam alone.
6 const IMPLEMENTATIONS = new Set(["src/crew/claude.ts", "src/crew/agent_runner.ts"]);
7 const FROM = /(?:from|import)\s*\(?\s*["']([^"']*\/claude(?:\.ts)?)["']/g;
8
9 test("no module outside the runner implementations imports claude.ts", async () => {
10 const offenders: string[] = [];
11 for await (const file of new Glob("src/**/*.ts").scan(".")) {
12 if (IMPLEMENTATIONS.has(file)) continue;
13 const text = await Bun.file(file).text();
14 for (const m of text.matchAll(FROM)) offenders.push(`${file} -> ${m[1]}`);
15 }
16 expect(offenders).toEqual([]);
17 });
18
19 test("the four crew callers go through the seam", async () => {
20 for (const name of ["runner", "dispatch", "one_shot", "conversation"]) {
21 const text = await Bun.file(`src/crew/${name}.ts`).text();
22 expect(text).toContain('from "./agent_runner.ts"');
23 }
24 });

↑↓ navigate ↵ open ⌘↵ its app esc close

Keyboard shortcuts

Docs