trevato/forge

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

More
Forks1g fMapg aRuns14g rEventsFlows

A finished run must free its slot: close the session's stdin after the final result instead of idling to the 45-minute cap #545

Merged torquecrew opened this in answer to trevatoowner

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

ConversationFiles
Merged

torque/forge:torque/474-a-finished-run-must-free-its-slot-close → main 6 commits fork

  • Merged. torque/474-a-finished-run-must-free-its-slot-close was merged into main as d13ceeb.

Review

Sign in to review this change.
3 files changed +227 −10
forge.json +1 −1
@@ -6,7 +6,7 @@
66 "ready": 90
77 },
88 "check": {
9 "run": "bun x tsc --noEmit && bun scripts/test-shards.ts --timeout 60000 -- crew crew-wrapup crew-limits crew-sandbox steer landing decide ledger checks dependencies autonomy studio studio-words forks fork-flow cross-repo-pull pulls apps apps-adversarial apps-busy apps-token web git-trust helpers shards migrate vitals guard asks feed feed-page offline share cloud run-secrets forge-cloudflare usage credentials standing board-windows spawn-plan spawn web-apps fleet-ops meter",
9 "run": "bun x tsc --noEmit && bun scripts/test-shards.ts --timeout 60000 -- crew crew-wrapup crew-limits crew-sandbox steer session-settle landing decide ledger checks dependencies autonomy studio studio-words forks fork-flow cross-repo-pull pulls apps apps-adversarial apps-busy apps-token web git-trust helpers shards migrate vitals guard asks feed feed-page offline share cloud run-secrets forge-cloudflare usage credentials standing board-windows spawn-plan spawn web-apps fleet-ops meter",
1010 "timeout": 1800
1111 }
1212 }
src/crew/claude.ts +75 −9
@@ -43,8 +43,26 @@ export interface ClaudeInput {
4343 * for this invocation, never deltas.
4444 */
4545 readonly onUsage?: ((spent: Spent) => void) | undefined;
46 /**
47 * How long a session may stay silent after a result while its input is
48 * still open (a steer was pending) before it is closed and settled as
49 * done. A steer folded into the turn it interrupted leaves a result owed
50 * that never comes. Default `IDLE_AFTER_RESULT_MS`.
51 */
52 readonly idleMs?: number | undefined;
53 /**
54 * How long the process has to exit once its input is closed after the
55 * final result. Past it the process group is stopped and the run settles
56 * on the result it already gave. Default `SETTLE_MS`.
57 */
58 readonly settleMs?: number | undefined;
4659 }
4760
61 /** The idle guard's default: five minutes of silence after a result. */
62 export const IDLE_AFTER_RESULT_MS = 5 * 60_000;
63 /** The grace a finished session has to exit on its own. */
64 export const SETTLE_MS = 5_000;
65
4866 export interface Spent {
4967 readonly tokensIn: number;
5068 readonly tokensOut: number;
@@ -214,8 +232,47 @@ export async function runClaude(input: ClaudeInput): Promise<ClaudeOutcome> {
214232 // last word per message id is the message's spend.
215233 const perMessage = new Map<string, { tokensIn: number; tokensOut: number }>();
216234
235 // The reads end on their own when the process exits — unless something it
236 // spawned still holds the pipe. Aborting them is how a settled session
237 // stops waiting on a server the agent left running.
238 const stopReading = new AbortController();
239 const idleMs = input.idleMs ?? IDLE_AFTER_RESULT_MS;
240 const settleMs = input.settleMs ?? SETTLE_MS;
241 let idle: ReturnType<typeof setTimeout> | undefined;
242 let settle: ReturnType<typeof setTimeout> | undefined;
243 let finished = false;
244 // After a result with a steer pending, the session owes another turn; any
245 // line it writes is that turn going. Silence past `idleMs` means it is not.
246 const armIdle = (): void => {
247 clearTimeout(idle);
248 idle = setTimeout(() => {
249 input.log(
250 `session: no output for ${Math.round(idleMs / 1000)}s after its last result; closing it`,
251 );
252 finish();
253 }, idleMs);
254 };
255 // The work is over: close the input and give the process `settleMs` to
256 // go. One that lingers is stopped, and the result it gave stands.
257 const finish = (): void => {
258 if (finished) return;
259 finished = true;
260 clearTimeout(idle);
261 closeInput();
262 settle = setTimeout(() => {
263 if (proc.exitCode === null && proc.signalCode === null) {
264 input.log(
265 `session: still open ${Math.round(settleMs / 1000)}s after its result; stopping it`,
266 );
267 stopChild();
268 }
269 stopReading.abort();
270 }, settleMs);
271 };
272
217273 const onLine = (line: string): void => {
218274 if (line.trim() === "") return;
275 if (sawResult && !finished) armIdle();
219276 let event: unknown;
220277 try {
221278 event = JSON.parse(line);
@@ -282,8 +339,10 @@ export async function runClaude(input: ClaudeInput): Promise<ClaudeOutcome> {
282339 }
283340 // A turn is over. Something said meanwhile is the next turn; nothing
284341 // said means the session is done, and closing stdin lets it end.
285 if (pending > 0 && ok) pending--;
286 else closeInput();
342 if (pending > 0 && ok) {
343 pending--;
344 armIdle();
345 } else finish();
287346 return;
288347 }
289348 // `user` records are tool results: long and mostly file contents.
@@ -306,7 +365,8 @@ export async function runClaude(input: ClaudeInput): Promise<ClaudeOutcome> {
306365 setTimeout(() => {
307366 if (proc.exitCode === null && proc.signalCode === null)
308367 signalGroup("SIGKILL");
309 }, 5_000);
368 stopReading.abort();
369 }, 5_000).unref();
310370 };
311371 const timer = setTimeout(() => {
312372 why = "timeout";
@@ -321,15 +381,21 @@ export async function runClaude(input: ClaudeInput): Promise<ClaudeOutcome> {
321381
322382 const stderr: string[] = [];
323383 const [, , code] = await Promise.all([
324 pump(proc.stdout, onLine),
325 pump(proc.stderr, (line) => {
326 stderr.push(line);
327 if (stderr.length > 50) stderr.shift();
328 input.log(`! ${line}`);
329 }),
384 pump(proc.stdout, onLine, stopReading.signal),
385 pump(
386 proc.stderr,
387 (line) => {
388 stderr.push(line);
389 if (stderr.length > 50) stderr.shift();
390 input.log(`! ${line}`);
391 },
392 stopReading.signal,
393 ),
330394 proc.exited,
331395 ]);
332396 clearTimeout(timer);
397 clearTimeout(idle);
398 clearTimeout(settle);
333399 closeInput();
334400 input.signal.removeEventListener("abort", onAbort);
335401
test/session-settle.test.ts added +151
@@ -0,0 +1,151 @@
1 /**
2 * A finished session frees its slot. Once the final result is in and
3 * nothing is pending, the forge closes the session's input and the run
4 * settles within seconds — even when the process, or something it left
5 * running, does not go on its own. A session that owes a turn and goes
6 * silent is closed by the idle guard.
7 */
8 import { afterAll, describe, expect, test } from "bun:test";
9 import { chmodSync, writeFileSync } from "node:fs";
10 import { mkdtemp, rm } from "node:fs/promises";
11 import { tmpdir } from "node:os";
12 import { join } from "node:path";
13 import { runClaude } from "../src/crew/claude.ts";
14 import { sandboxPolicy } from "../src/crew/sandbox.ts";
15
16 const dir = await mkdtemp(join(tmpdir(), "forge-settle-"));
17 afterAll(() => rm(dir, { recursive: true, force: true }));
18
19 const RESULT =
20 '{"type":"result","subtype":"success","is_error":false,"result":"done","total_cost_usd":0.01,"num_turns":1}';
21
22 /** A stand-in Claude Code: answers the prompt, then does `after`. */
23 function fake(name: string, after: string): string {
24 const path = join(dir, name);
25 writeFileSync(
26 path,
27 `#!/bin/sh
28 IFS= read -r first
29 echo '{"type":"system","subtype":"init","session_id":"s1","model":"fake"}'
30 sleep 0.3
31 echo '{"type":"assistant","message":{"content":[{"type":"text","text":"working"}]}}'
32 echo '${RESULT}'
33 ${after}
34 `,
35 );
36 chmodSync(path, 0o755);
37 return path;
38 }
39
40 function common(binary: string, lines: string[]) {
41 return {
42 binary,
43 cwd: dir,
44 prompt: "Do the thing.",
45 appendSystemPrompt: "",
46 budgetUsd: 1,
47 env: {},
48 sandbox: sandboxPolicy({ dataDir: dir, agentName: "bot", urls: [] }),
49 // The 45-minute cap, in miniature: far past what a settle may take.
50 timeoutMs: 60_000,
51 log: (line: string) => lines.push(line),
52 signal: new AbortController().signal,
53 settleMs: 1_000,
54 };
55 }
56
57 describe("a session after its final result", () => {
58 test("one that ignores its closed input is stopped, and the run settles on its result in < 10 s", async () => {
59 const lines: string[] = [];
60 const started = performance.now();
61 const outcome = await runClaude(
62 common(fake("deaf", "sleep 600"), lines),
63 );
64 expect(performance.now() - started).toBeLessThan(10_000);
65 expect(outcome.ok).toBe(true);
66 expect(outcome.result).toBe("done");
67 expect(outcome.reason).toBe("completed");
68 expect(lines.some((l) => l.includes("after its result; stopping it"))).toBe(
69 true,
70 );
71 }, 20_000);
72
73 test("one that exits but leaves a child holding its output still settles in < 10 s", async () => {
74 const lines: string[] = [];
75 const started = performance.now();
76 const outcome = await runClaude(
77 common(fake("leaver", "sleep 600 &\nexit 0"), lines),
78 );
79 expect(performance.now() - started).toBeLessThan(10_000);
80 expect(outcome.ok).toBe(true);
81 expect(outcome.result).toBe("done");
82 }, 20_000);
83
84 test("one that ends on its own is not held for the grace", async () => {
85 const lines: string[] = [];
86 const started = performance.now();
87 const outcome = await runClaude({
88 ...common(fake("polite", "exit 0"), lines),
89 settleMs: 30_000,
90 });
91 expect(performance.now() - started).toBeLessThan(5_000);
92 expect(outcome.ok).toBe(true);
93 expect(lines.some((l) => l.startsWith("session:"))).toBe(false);
94 }, 20_000);
95
96 test("a steer folded into the turn it interrupted leaves a turn owed; the idle guard closes it", async () => {
97 const lines: string[] = [];
98 let send: ((text: string) => boolean) | undefined;
99 const started = performance.now();
100 // Reads the steer but never answers it: one result, then silence.
101 const run = runClaude({
102 ...common(fake("folds", "IFS= read -r second\nsleep 600"), lines),
103 idleMs: 1_500,
104 onSteer: (s) => {
105 send = s;
106 },
107 });
108 await Bun.sleep(100);
109 expect(send!("also add a test")).toBe(true);
110 const outcome = await run;
111 const took = performance.now() - started;
112 expect(took).toBeGreaterThanOrEqual(1_500);
113 expect(took).toBeLessThan(10_000);
114 expect(outcome.ok).toBe(true);
115 expect(outcome.result).toBe("done");
116 expect(
117 lines.some((l) => l.includes("no output for 2s after its last result")),
118 ).toBe(true);
119 // The session is closed: a later word is not heard by it, and the
120 // caller makes it a comment for the next run.
121 expect(send!("too late")).toBe(false);
122 }, 20_000);
123
124 test("output after the result puts the idle guard back", async () => {
125 const lines: string[] = [];
126 let send: ((text: string) => boolean) | undefined;
127 // Answers the steer slowly, talking all the way: never idle for 1 s.
128 const run = runClaude({
129 ...common(
130 fake(
131 "slow",
132 `IFS= read -r second
133 for i in 1 2 3 4; do sleep 0.5; echo '{"type":"assistant","message":{"content":[{"type":"text","text":"still going"}]}}'; done
134 echo '{"type":"result","subtype":"success","is_error":false,"result":"second","total_cost_usd":0.02,"num_turns":2}'`,
135 ),
136 lines,
137 ),
138 idleMs: 1_000,
139 onSteer: (s) => {
140 send = s;
141 },
142 });
143 await Bun.sleep(100);
144 expect(send!("and then this")).toBe(true);
145 const outcome = await run;
146 expect(outcome.ok).toBe(true);
147 expect(outcome.result).toBe("second");
148 expect(outcome.turns).toBe(2);
149 expect(lines.some((l) => l.startsWith("session: no output"))).toBe(false);
150 }, 20_000);
151 });

↑↓ navigate ↵ open ⌘↵ its app esc close

Keyboard shortcuts

Docs