flagon-io/g1t

public

The model proxy holds each run to its cost cap itself: a run's session carries its cap, a 402 at the cap, and a run token reaches only the message routes (integrations 0006)

syntaqxcommitted 2d agoParentd9ea97dBrowse files

159	159		tier: Option<String>,
160	160		#[serde(default)]
161	161		requested_by: Option<String>,
	162	+	#[serde(default)]
	163	+	cap_micros: Option<i64>,
	164	+	}
	165	+
	166	+	/// A run's model spend cap as a session keeps it: null for none (zero or
	167	+	/// less), and never more than $1,000, which no run is allowed. D1 takes
	168	+	/// numbers as doubles, which hold any such cap exactly.
	169	+	fn session_cap(cap_micros: i64) -> JsValue {
	170	+	if cap_micros > 0 { JsValue::from_f64(cap_micros.min(1_000_000_000) as f64) } else { JsValue::NULL }
162	171		}
163	172
164	173		#[derive(Deserialize)]
⋯
1324	1333		api_key: None,
1325	1334		auth_header: None,
1326	1335		gateway_token: None,
	1336	+	cap_micros: session.cap_micros.filter(\|cap\| *cap > 0),
1327	1337		};
1328	1338		let Some(connection_id) = session.connection_id else {
1329	1339		return Ok(Some(base));
⋯
1381	1391		api_key: secrets.secret.clone(),
1382	1392		auth_header: Some(models::auth_header(provider, &config)),
1383	1393		gateway_token: (provider == Provider::AnthropicEndpoint).then_some(secrets.signing_secret).flatten(),
	1394	+	cap_micros: None,
1384	1395		}))
1385	1396		}
1386	1397
⋯
1418	1429		Ok(result.meta()?.and_then(\|meta\| meta.changes).unwrap_or(0) as u32)
1419	1430		}
1420	1431
	1432	+	/// Sets the most a run may spend on models, which the model proxy holds
	1433	+	/// its token to. The sandbox calls it once it knows the run's caps.
	1434	+	async fn cap_model_sessions(&self, a: CapModelSessionsArgs) -> Result<u32> {
	1435	+	let hashes = closable_hashes(&a.token_hashes);
	1436	+	if hashes.is_empty() {
	1437	+	return Ok(0);
	1438	+	}
	1439	+	let marks = vec!["?"; hashes.len()].join(", ");
	1440	+	let mut values: Vec<JsValue> = vec![session_cap(a.cap_micros)];
	1441	+	values.extend(hashes.iter().map(\|hash\| JsValue::from(hash.as_str())));
	1442	+	values.push(rfc3339(now_ms()).into());
	1443	+	let result = self
	1444	+	.db
	1445	+	.prepare(format!("UPDATE model_sessions SET cap_micros = ? WHERE token_hash IN ({marks}) AND expires_at > ?"))
	1446	+	.bind(&values)?
	1447	+	.run()
	1448	+	.await?;
	1449	+	Ok(result.meta()?.and_then(\|meta\| meta.changes).unwrap_or(0) as u32)
	1450	+	}
	1451	+
1421	1452		// --- Writing back -----------------------------------------------------------
1422	1453
1423	1454		async fn on_event(&self, event: &Event) -> Result<()> {
⋯
1569	1600		"gateway_upstream" => reply(&service.gateway_upstream(args(body)?).await?),
1570	1601		"gateway_providers" => reply(&service.gateway_providers(args(body)?).await?),
1571	1602		"close_model_sessions" => reply(&service.close_model_sessions(args(body)?).await?),
	1603	+	"cap_model_sessions" => reply(&service.cap_model_sessions(args(body)?).await?),
1572	1604		"routes" => reply(&service.routes(args(body)?).await?),
1573	1605		"set_routes" => reply(&service.set_routes(args(body)?).await?),
1574	1606		_ => Response::error("Unknown method", 404),

15	15		* it passes and reported to billing afterwards, counted per run for usage
16	16		* views.
17	17		*
	18	+	* Each run is held to its cost cap here too, not only by the harness in the
	19	+	* sandbox: every answer's cost is added to the run's count (a Durable
	20	+	* Object per run, `run-spend.ts`), and once the run has spent its cap its
	21	+	* requests are refused with a 402 (`spend.ts`).
	22	+	*
18	23		* The same address is the AI Gateway for a workspace's own code: a request
19	24		* with one of the workspace's access tokens (`g1t_…`) instead of a run's,
20	25		* at `/anthropic` in Anthropic's format or `/openai/v1` in OpenAI's, goes
⋯
37	42
38	43		import { openaiError } from "./chat";
39	44		import { discover } from "./discover";
	45	+	import { anthropicErrorType } from "./gateway";
40	46		import { type AnthropicRequest, StreamTranslator, errorFromChat, estimateTokens, fromChat, toChat } from "./openai";
41		−	import { isAnswer, tokenReport } from "./report";
	47	+	import { isAnswer, runMayCall, tokenReport } from "./report";
42	48		import { type HostedRouting, presentedToken, upstreamRequest } from "./route";
	49	+	import type { RunSpend } from "./run-spend";
43	50		import { type GatewayDeps, isOpenAiPath, serveGateway } from "./serve";
	51	+	import { capOf, capReached, ceilingMicros, chargeFor, pricesFor, tooBusy } from "./spend";
44	52		import { measure } from "./usage";
45	53
	54	+	export { RunSpend } from "./run-spend";
	55	+
46	56		interface Env extends HostedRouting {
47	57		INTEGRATIONS: ServiceBinding;
48	58		BILLING: ServiceBinding;
49	59		IDENTITY: ServiceBinding;
	60	+	/** Each run's model spend, one object per model session. */
	61	+	RUN_SPEND: DurableObjectNamespace<RunSpend>;
50	62		}
51	63
52	64		/**
⋯
69	81
70	82		/** An error in the shape Anthropic's API uses, which the harness understands. */
71	83		function refuse(status: number, message: string): Response {
72		−	return Response.json(
73		−	{ type: "error", error: { type: status === 401 ? "authentication_error" : "not_found_error", message } },
74		−	{ status },
75		−	);
	84	+	return Response.json({ type: "error", error: { type: anthropicErrorType(status), message } }, { status });
	85	+	}
	86	+
	87	+	/** The run's spend count, by its session's id. */
	88	+	function runSpend(env: Env, upstream: ModelUpstream): DurableObjectStub<RunSpend> {
	89	+	return env.RUN_SPEND.get(env.RUN_SPEND.idFromName(upstream.session \|\| `${upstream.workspace}/${upstream.repo}#${upstream.number}`));
76	90		}
77	91
	92	+	/** One answer's place in its run's count, settled once the answer has gone by. */
	93	+	type Held = { spend: DurableObjectStub<RunSpend>; ticket: string; requested: Partial<AnthropicRequest> \| null; bodyLength: number };
	94	+
78	95		/**
79		−	* Passes an answer through and, once it has all gone by, tells billing what
80		−	* it used. Reporting happens after the answer, and a report that fails is
81		−	* dropped: the answer never waits on it or breaks for it.
	96	+	* Passes an answer through and, once it has all gone by, adds its cost to
	97	+	* the run's count and tells billing what it used. Both happen after the
	98	+	* answer, and a report that fails is dropped: the answer never waits on
	99	+	* it or breaks for it.
82	100		*/
83		−	function counted(answer: Response, upstream: ModelUpstream, env: Env, ctx: ExecutionContext): Response {
	101	+	function counted(answer: Response, upstream: ModelUpstream, env: Env, ctx: ExecutionContext, held: Held): Response {
84	102		const { response, tokens, model } = measure(answer);
85	103		ctx.waitUntil(
86	104		(async () => {
87		−	const report = tokenReport(upstream, await model, await tokens);
88		−	if (report) await billingClient(env.BILLING).recordTokens(report);
	105	+	const used = await tokens;
	106	+	const answeredBy = await model;
	107	+	const settle = (async () => {
	108	+	const prices = pricesFor(upstream.model ?? answeredBy ?? held.requested?.model, upstream.route, await offered(env).catch(() => []));
	109	+	const charge = chargeFor(prices, used, answer.ok, ceilingMicros(prices, held.bodyLength, held.requested?.max_tokens));
	110	+	await held.spend.settle(held.ticket, charge);
	111	+	})().catch((error: unknown) => console.error("models: a run's spend was not counted", upstream.session, String(error)));
	112	+	const report = tokenReport(upstream, answeredBy, used);
	113	+	const reported = report ? billingClient(env.BILLING).recordTokens(report).catch(() => undefined) : Promise.resolve();
	114	+	await Promise.all([settle, reported]);
89	115		})().catch(() => undefined),
90	116		);
91	117		return response;
92	118		}
93	119
	120	+	/** The fields of a request body the proxy reads, or null when it is not a JSON object. */
	121	+	function parsedBody(text: string): Record<string, unknown> \| null {
	122	+	try {
	123	+	const parsed = JSON.parse(text) as unknown;
	124	+	return parsed && typeof parsed === "object" && !Array.isArray(parsed) ? (parsed as Record<string, unknown>) : null;
	125	+	} catch {
	126	+	return null;
	127	+	}
	128	+	}
	129	+
94	130		/** Sends an Anthropic request to a provider that speaks OpenAI's API. */
95		−	async function viaChat(upstream: ModelUpstream, path: string, request: Request): Promise<Response> {
96		−	const body = (await request.json()) as AnthropicRequest;
	131	+	async function viaChat(upstream: ModelUpstream, path: string, body: AnthropicRequest \| null): Promise<Response> {
	132	+	if (!path.startsWith("/v1/messages")) return refuse(404, `${path} has no counterpart at this provider.`);
	133	+	if (!body) return refuse(400, "The request body is not a JSON object.");
97	134		const model = upstream.model ?? body.model ?? "";
98	135		if (path.startsWith("/v1/messages/count_tokens")) {
99	136		return Response.json({ input_tokens: estimateTokens(body) });
100	137		}
101		−	if (!path.startsWith("/v1/messages")) return refuse(404, `${path} has no counterpart at this provider.`);
102	138
103	139		const headers = new Headers({ "content-type": "application/json" });
104	140		if (upstream.gatewayToken) headers.set("cf-aig-authorization", `Bearer ${upstream.gatewayToken}`);
⋯
237	273		const upstream = await lookUp(env, token);
238	274		if (!upstream) return refuse(401, "This run's model token has expired, or its model connection was removed.");
239	275
240		−	const path = url.pathname.slice("/anthropic".length) + url.search;
	276	+	const route = url.pathname.slice("/anthropic".length);
	277	+	const path = route + url.search;
	278	+	if (!runMayCall(route, request.method)) return refuse(404, `A run's model token reaches only /anthropic/v1/messages and /anthropic/v1/models, not ${route}.`);
	279	+	const hasBody = request.method !== "GET" && request.method !== "HEAD";
	280	+	const text = hasBody ? await request.text() : null;
	281	+	const parsed = text === null ? null : parsedBody(text);
	282	+
	283	+	// A model's answer costs the run: it must be under its cap to start
	284	+	// one, and the answer's cost is added to its count once it has gone by.
	285	+	let held: Held \| null = null;
	286	+	if (isAnswer(route) && hasBody) {
	287	+	const spend = runSpend(env, upstream);
	288	+	const cap = capOf(upstream);
	289	+	let admitted;
	290	+	try {
	291	+	admitted = await spend.admit(cap);
	292	+	} catch (error) {
	293	+	console.error("models: a run's spend could not be checked", upstream.session, String(error));
	294	+	return refuse(503, "g1t could not check this run's spending just now. Try again.");
	295	+	}
	296	+	if (!admitted.ok) return admitted.reason === "cap" ? capReached(cap, admitted.spent) : tooBusy();
	297	+	held = { spend, ticket: admitted.ticket, requested: parsed as Partial<AnthropicRequest> \| null, bodyLength: text?.length ?? 0 };
	298	+	}
241	299		// Both routes answer in Anthropic's shape, so one reading counts either.
242		−	const answer = (response: Response) => (isAnswer(url.pathname.slice("/anthropic".length)) ? counted(response, upstream, env, ctx) : response);
243		−	if (upstream.api === "openai") return answer(await viaChat(upstream, path, request));
	300	+	const answer = (response: Response) => (held ? counted(response, upstream, env, ctx, held) : response);
	301	+	try {
	302	+	if (upstream.api === "openai") return answer(await viaChat(upstream, path, parsed as AnthropicRequest \| null));
244	303
245		−	const { url: target, headers } = upstreamRequest(upstream, env, path, request.headers);
246		−	// A route that names a model gets it for every request of the run,
247		−	// including the harness's small background ones.
248		−	let body: BodyInit \| null = request.method === "GET" \|\| request.method === "HEAD" ? null : request.body;
249		−	if (upstream.model && body && path.startsWith("/v1/messages")) {
250		−	const parsed = (await request.json()) as Record<string, unknown>;
251		−	body = JSON.stringify({ ...parsed, model: upstream.model });
252		−	headers.delete("content-length");
	304	+	const { url: target, headers } = upstreamRequest(upstream, env, path, request.headers);
	305	+	// A route that names a model gets it for every request of the run,
	306	+	// including the harness's small background ones.
	307	+	let body: string \| null = text;
	308	+	if (upstream.model && parsed && path.startsWith("/v1/messages")) {
	309	+	body = JSON.stringify({ ...parsed, model: upstream.model });
	310	+	headers.delete("content-length");
	311	+	}
	312	+	return answer(await fetch(target, { method: request.method, headers, body }));
	313	+	} catch (error) {
	314	+	// No answer: it cost nothing, and gives its place back.
	315	+	if (held) ctx.waitUntil(held.spend.settle(held.ticket, 0).catch(() => undefined));
	316	+	throw error;
253	317		}
254		−	return answer(await fetch(target, { method: request.method, headers, body }));
255	318		},
256	319		} satisfies ExportedHandler<Env>;

1	+	import assert from "node:assert/strict";
2	+	import { test } from "node:test";
3	+
4	+	import type { GatewayModel } from "@g1t/contracts";
5	+
6	+	import {
7	+	BACKSTOP_CAP_MICROS,
8	+	FRONTIER_PRICES,
9	+	HOLD_MS,
10	+	MAX_IN_FLIGHT,
11	+	STANDARD_PRICES,
12	+	SpendTally,
13	+	capOf,
14	+	capReached,
15	+	ceilingMicros,
16	+	chargeFor,
17	+	costMicros,
18	+	pricesFor,
19	+	tooBusy,
20	+	} from "./spend.ts";
21	+	import { NO_TOKENS } from "./usage.ts";
22	+
23	+	function model(fields: Partial<GatewayModel> & { model: string }): GatewayModel {
24	+	return { name: fields.model, provider: "anthropic", inputMicros: 0, outputMicros: 0, cacheReadMicros: 0, cacheWriteMicros: 0, ...fields };
25	+	}
26	+
27	+	const sonnet = model({ model: "claude-sonnet-5-5", inputMicros: 3_000_000, outputMicros: 15_000_000, cacheReadMicros: 300_000, cacheWriteMicros: 3_750_000, cacheWrite1hMicros: 6_000_000 });
28	+	const haiku = model({ model: "claude-haiku-5", inputMicros: 1_000_000, outputMicros: 5_000_000, cacheReadMicros: 100_000, cacheWriteMicros: 1_250_000 });
29	+	const embed = model({ model: "@cf/baai/bge-m3", provider: "workers-ai", kind: "embeddings", inputMicros: 12_000, outputMicros: 900_000_000 });
30	+	const offered = [haiku, sonnet, embed];
31	+
32	+	test("a run's cap is the one its sandbox set, else the most any run may cost", () => {
33	+	assert.equal(capOf({ capMicros: 2_000_000 }), 2_000_000);
34	+	assert.equal(capOf({ capMicros: null }), BACKSTOP_CAP_MICROS);
35	+	assert.equal(capOf({}), BACKSTOP_CAP_MICROS);
36	+	assert.equal(capOf({ capMicros: 0 }), BACKSTOP_CAP_MICROS);
37	+	});
38	+
39	+	test("an answer is priced by its model in the catalogue, dated ids and provider prefixes included", () => {
40	+	assert.equal(pricesFor("claude-sonnet-5-5", "g1t", offered), sonnet);
41	+	assert.equal(pricesFor("claude-sonnet-5-5-20260901", "g1t", offered), sonnet);
42	+	assert.equal(pricesFor("anthropic/claude-haiku-5", "anthropic", offered), haiku);
43	+	});
44	+
45	+	test("a model missing from the catalogue is priced high on g1t's models, at standard prices on the workspace's own", () => {
46	+	assert.equal(pricesFor("mystery", "g1t", offered), FRONTIER_PRICES);
47	+	assert.equal(pricesFor(null, "g1t", []), FRONTIER_PRICES);
48	+	const dearer = model({ model: "claude-opus-9", inputMicros: 20_000_000, outputMicros: 100_000_000 });
49	+	assert.equal(pricesFor("mystery", "g1t", [...offered, dearer]), dearer);
50	+	assert.equal(pricesFor("llama-local", "endpoint", offered), STANDARD_PRICES);
51	+	});
52	+
53	+	test("cost is billing's sum: each kind of token at its price, rounded up", () => {
54	+	// 1,000 in at $3, 500 out at $15, 10,000 cache reads at $0.30, 2,000 writes (500 of them hour-long).
55	+	const tokens = { input: 1_000, output: 500, cacheRead: 10_000, cacheWrite: 2_000, cacheWrite1h: 500 };
56	+	assert.equal(costMicros(sonnet, tokens), 3_000 + 7_500 + 3_000 + 5_625 + 3_000);
57	+	assert.equal(costMicros(sonnet, { ...NO_TOKENS, input: 1 }), 3);
58	+	assert.equal(costMicros(haiku, { ...NO_TOKENS, input: 1 }), 1);
59	+	// No hour-long price: those writes cost what five-minute ones do.
60	+	assert.equal(costMicros(haiku, { ...NO_TOKENS, cacheWrite: 1_000, cacheWrite1h: 1_000 }), 1_250);
61	+	});
62	+
63	+	test("a prompt past the model's threshold puts the whole request at the higher prices", () => {
64	+	const long = model({ model: "long", inputMicros: 1_000_000, outputMicros: 2_000_000, threshold: 100, overInputMicros: 2_000_000, overOutputMicros: 4_000_000 });
65	+	assert.equal(costMicros(long, { ...NO_TOKENS, input: 100, output: 1_000 }), 100 + 2_000);
66	+	assert.equal(costMicros(long, { ...NO_TOKENS, input: 101, output: 1_000 }), 202 + 4_000);
67	+	});
68	+
69	+	test("an answer is charged its cost, nothing when refused, and its ceiling when its usage could not be read", () => {
70	+	const ceiling = ceilingMicros(sonnet, 4_000, 1_000);
71	+	assert.equal(ceiling, 3_000 + 15_000);
72	+	assert.equal(ceilingMicros(sonnet, 0, undefined), 32_000 * 15);
73	+	assert.equal(chargeFor(sonnet, { ...NO_TOKENS, output: 100 }, true, ceiling), 1_500);
74	+	assert.equal(chargeFor(sonnet, NO_TOKENS, true, ceiling), ceiling);
75	+	assert.equal(chargeFor(sonnet, NO_TOKENS, false, ceiling), 0);
76	+	});
77	+
78	+	test("a run under its cap is admitted, and refused once it has spent it", () => {
79	+	const tally = new SpendTally();
80	+	const first = tally.admit(10_000, 0);
81	+	assert.ok(first.ok);
82	+	assert.equal(tally.settle(first.ticket, 9_999), 9_999);
83	+	const second = tally.admit(10_000, 1);
84	+	assert.ok(second.ok);
85	+	// The answer in flight takes it past; the next is refused.
86	+	tally.settle(second.ticket, 5_000);
87	+	assert.deepEqual(tally.admit(10_000, 2), { ok: false, reason: "cap", spent: 14_999 });
88	+	});
89	+
90	+	test("many requests at once are bounded, and an answer never settled gives its place back in time", () => {
91	+	const tally = new SpendTally();
92	+	const tickets: string[] = [];
93	+	for (let i = 0; i < MAX_IN_FLIGHT; i++) {
94	+	const admitted = tally.admit(1_000_000, 0);
95	+	assert.ok(admitted.ok);
96	+	tickets.push(admitted.ticket);
97	+	}
98	+	assert.deepEqual(tally.admit(1_000_000, 0), { ok: false, reason: "busy", spent: 0 });
99	+	tally.settle(tickets[0], 0);
100	+	assert.ok(tally.admit(1_000_000, 1).ok);
101	+	assert.equal(tally.admit(1_000_000, 1).ok, false);
102	+	assert.ok(tally.admit(1_000_000, HOLD_MS + 1).ok);
103	+	// Those from the start have lapsed; the one from a moment later has not.
104	+	assert.equal(tally.inFlight, 2);
105	+	});
106	+
107	+	test("a tally resumes from what was stored, and settling twice or with nonsense adds nothing", () => {
108	+	const tally = new SpendTally(500);
109	+	const admitted = tally.admit(1_000, 0);
110	+	assert.ok(admitted.ok);
111	+	tally.settle(admitted.ticket, Number.NaN);
112	+	tally.settle(admitted.ticket, -5);
113	+	assert.equal(tally.spent, 500);
114	+	assert.equal(new SpendTally(-1).spent, 0);
115	+	});
116	+
117	+	test("a run past its cap is told so in Anthropic's error shape, with a code", async () => {
118	+	const response = capReached(2_000_000, 2_031_000);
119	+	assert.equal(response.status, 402);
120	+	const body = (await response.json()) as { type: string; error: { type: string; code: string; message: string } };
121	+	assert.equal(body.type, "error");
122	+	assert.equal(body.error.type, "billing_error");
123	+	assert.equal(body.error.code, "run_cap_reached");
124	+	assert.match(body.error.message, /cost cap of \$2\.00 \(it has spent \$2\.03/);
125	+	const busy = tooBusy();
126	+	assert.equal(busy.status, 429);
127	+	assert.equal(busy.headers.get("retry-after"), "2");
128	+	});

1	+	/**
2	+	* A run's model spend, held to its cost cap by the proxy itself.
3	+	*
4	+	* The harness stops the agent at the run's cap (`--max-budget-usd`), but
5	+	* that is the sandbox's own word: an agent talked into calling the proxy
6	+	* directly would never be stopped by it. So the proxy counts what each of
7	+	* the run's answers cost and refuses its requests once the run has spent
8	+	* its cap, with a 402 the harness shows as the error it is.
9	+	*
10	+	* The count lives in one Durable Object per run (`run-spend.ts`): a run's
11	+	* requests land on many isolates, and an agent can send many at once, so a
12	+	* per-isolate count would let each isolate spend the cap again. This file
13	+	* is the counting and pricing alone, without the worker, for tests.
14	+	*/
15	+	import type { GatewayModel, ModelUpstream } from "@g1t/contracts";
16	+
17	+	import { anthropicError } from "./gateway.ts";
18	+	import { type Tokens, total } from "./usage.ts";
19	+
20	+	/** A model's prices per million tokens, in millionths of a dollar. */
21	+	export type Prices = Omit<GatewayModel, "model" \| "name" \| "provider" \| "kind">;
22	+
23	+	/**
24	+	* The cap of a run whose sandbox never set one (its guardrails and its plan
25	+	* both said none, or setting it failed): the most any run may be allowed
26	+	* to cost (`MAX_BUDGET_USD` in crates/contracts/src/guardrails.rs).
27	+	*/
28	+	export const BACKSTOP_CAP_MICROS = 100_000_000;
29	+
30	+	/**
31	+	* The most of a run's answers in flight at once. The cap is checked as a
32	+	* request starts and an answer's cost is known only when it ends, so the
33	+	* answers already in flight when the run reaches its cap can take it past
34	+	* by their cost; this bounds how many there can be. The harness runs a
35	+	* handful at a time, subagents included.
36	+	*/
37	+	export const MAX_IN_FLIGHT = 16;
38	+
39	+	/**
40	+	* How long an answer in flight is waited for before it no longer counts
41	+	* against `MAX_IN_FLIGHT`: one whose cost was never settled (the proxy's
42	+	* isolate went away mid-answer) must not hold a place for good.
43	+	*/
44	+	export const HOLD_MS = 15 * 60_000;
45	+
46	+	/** The output an answer is assumed to reach when its request names no `max_tokens`. */
47	+	const DEFAULT_MAX_TOKENS = 32_000;
48	+
49	+	/**
50	+	* Prices for a model g1t has none for. On g1t's models, a model missing
51	+	* from the catalogue is charged at a frontier model's list prices, so the
52	+	* cap errs on the side of stopping early. On the workspace's own provider
53	+	* (an endpoint whose model g1t does not list), at a standard model's, so
54	+	* an inexpensive model is not stopped far short of its cap; the harness,
55	+	* which knows no better, counts such a model much the same way.
56	+	*/
57	+	export const FRONTIER_PRICES: Prices = {
58	+	inputMicros: 15_000_000,
59	+	outputMicros: 75_000_000,
60	+	cacheReadMicros: 1_500_000,
61	+	cacheWriteMicros: 18_750_000,
62	+	cacheWrite1hMicros: 30_000_000,
63	+	};
64	+	export const STANDARD_PRICES: Prices = {
65	+	inputMicros: 3_000_000,
66	+	outputMicros: 15_000_000,
67	+	cacheReadMicros: 300_000,
68	+	cacheWriteMicros: 3_750_000,
69	+	cacheWrite1hMicros: 6_000_000,
70	+	};
71	+
72	+	/** The run's cap, in millionths of a dollar. */
73	+	export function capOf(upstream: Pick<ModelUpstream, "capMicros">): number {
74	+	const cap = upstream.capMicros;
75	+	return typeof cap === "number" && Number.isFinite(cap) && cap > 0 ? cap : BACKSTOP_CAP_MICROS;
76	+	}
77	+
78	+	/**
79	+	* The prices an answer is charged at: its model's in g1t's catalogue (by
80	+	* its id, or the catalogue id a dated id starts with), else the most
81	+	* expensive chat model's on g1t's models, else the fallbacks above.
82	+	*/
83	+	export function pricesFor(model: unknown, route: ModelUpstream["route"], offered: GatewayModel[]): Prices {
84	+	const id = (typeof model === "string" ? model : "").trim().toLowerCase().replace(/^[a-z0-9-]+\//, "");
85	+	if (id) {
86	+	const exact = offered.find((entry) => entry.model.toLowerCase() === id);
87	+	if (exact) return exact;
88	+	const prefixed = offered
89	+	.filter((entry) => id.startsWith(`${entry.model.toLowerCase()}-`))
90	+	.sort((a, b) => b.model.length - a.model.length)[0];
91	+	if (prefixed) return prefixed;
92	+	}
93	+	if (route !== "g1t") return STANDARD_PRICES;
94	+	const chat = offered.filter((entry) => (entry.kind ?? "chat") === "chat");
95	+	const dearest = chat.sort((a, b) => b.outputMicros - a.outputMicros)[0];
96	+	return dearest && dearest.outputMicros >= FRONTIER_PRICES.outputMicros ? dearest : FRONTIER_PRICES;
97	+	}
98	+
99	+	/**
100	+	* What tokens cost at a model's prices per million, rounded up to a whole
101	+	* millionth of a dollar: billing's own sum (`cost_micros` in
102	+	* services/billing/src/gateway.rs). A prompt longer than the model's
103	+	* threshold puts the whole request at the over-threshold prices.
104	+	*/
105	+	export function costMicros(prices: Prices, tokens: Tokens): number {
106	+	const prompt = tokens.input + tokens.cacheRead + tokens.cacheWrite;
107	+	const over = (prices.threshold ?? 0) > 0 && prompt > (prices.threshold ?? 0);
108	+	const pick = (base: number \| undefined, above: number \| undefined) => Math.max(0, (over ? above : base) ?? 0);
109	+	const hour = Math.min(tokens.cacheWrite1h ?? 0, tokens.cacheWrite);
110	+	const fiveMinutes = pick(prices.cacheWriteMicros, prices.overCacheWriteMicros);
111	+	const hourPrice = pick(prices.cacheWrite1hMicros, prices.overCacheWrite1hMicros) \|\| fiveMinutes;
112	+	const millionths =
113	+	tokens.input * pick(prices.inputMicros, prices.overInputMicros) +
114	+	tokens.output * pick(prices.outputMicros, prices.overOutputMicros) +
115	+	tokens.cacheRead * pick(prices.cacheReadMicros, prices.overCacheReadMicros) +
116	+	(tokens.cacheWrite - hour) * fiveMinutes +
117	+	hour * hourPrice;
118	+	return Math.ceil(millionths / 1_000_000);
119	+	}
120	+
121	+	/**
122	+	* The most a request could cost: its whole body as input (about four
123	+	* characters a token) and its `max_tokens` as output. What an answer that
124	+	* could not be read is charged.
125	+	*/
126	+	export function ceilingMicros(prices: Prices, bodyLength: number, maxTokens: unknown): number {
127	+	const output = typeof maxTokens === "number" && Number.isFinite(maxTokens) && maxTokens > 0 ? maxTokens : DEFAULT_MAX_TOKENS;
128	+	return costMicros(prices, { input: Math.ceil(bodyLength / 4), output, cacheRead: 0, cacheWrite: 0 });
129	+	}
130	+
131	+	/**
132	+	* What one answer is charged against the run's cap: nothing for a refused
133	+	* request; its tokens' cost; or, for an answer whose usage could not be
134	+	* read (cut off, or unreadable), the most it could have cost, so a reading
135	+	* that fails never makes an answer free.
136	+	*/
137	+	export function chargeFor(prices: Prices, tokens: Tokens, ok: boolean, ceiling: number): number {
138	+	if (!ok) return 0;
139	+	return total(tokens) === 0 ? ceiling : costMicros(prices, tokens);
140	+	}
141	+
142	+	export type Admission = { ok: true; ticket: string; spent: number } \| { ok: false; reason: "cap" \| "busy"; spent: number };
143	+
144	+	/**
145	+	* One run's count: what it has spent, and its answers in flight. Every
146	+	* method runs to its end without waiting, so a Durable Object's requests
147	+	* see each other's changes in order.
148	+	*/
149	+	export class SpendTally {
150	+	spent: number;
151	+	private readonly open = new Map<string, number>();
152	+	private issued = 0;
153	+
154	+	constructor(spent = 0) {
155	+	this.spent = Number.isFinite(spent) && spent > 0 ? spent : 0;
156	+	}
157	+
158	+	/** Whether another answer may start: the run is under its cap and not too busy. */
159	+	admit(cap: number, now: number): Admission {
160	+	for (const [ticket, since] of this.open) if (now - since > HOLD_MS) this.open.delete(ticket);
161	+	if (this.spent >= cap) return { ok: false, reason: "cap", spent: this.spent };
162	+	if (this.open.size >= MAX_IN_FLIGHT) return { ok: false, reason: "busy", spent: this.spent };
163	+	const ticket = `t${++this.issued}`;
164	+	this.open.set(ticket, now);
165	+	return { ok: true, ticket, spent: this.spent };
166	+	}
167	+
168	+	/** An answer has ended, costing `micros`. Returns what the run has spent. */
169	+	settle(ticket: string, micros: number): number {
170	+	this.open.delete(ticket);
171	+	if (Number.isFinite(micros) && micros > 0) this.spent += Math.ceil(micros);
172	+	return this.spent;
173	+	}
174	+
175	+	get inFlight(): number {
176	+	return this.open.size;
177	+	}
178	+	}
179	+
180	+	const dollars = (micros: number) => `$${(micros / 1_000_000).toFixed(2)}`;
181	+
182	+	/**
183	+	* The refusal of a run past its cap: Anthropic's error shape, which the
184	+	* harness shows as it is, with `code` for anything that reads it.
185	+	*/
186	+	export function capReached(cap: number, spent: number): Response {
187	+	const message = `This run reached its cost cap of ${dollars(cap)} (it has spent ${dollars(spent)} on models), so g1t refuses its model requests from here. Raise the cap in the project's guardrails or the workspace's billing, then start the work again.`;
188	+	return Response.json({ type: "error", error: { type: "billing_error", code: "run_cap_reached", message } }, { status: 402 });
189	+	}
190	+
191	+	/** The refusal of a request while the run has too many answers in flight; the harness retries it. */
192	+	export function tooBusy(): Response {
193	+	const response = anthropicError(429, "rate_limit_error", `This run has ${MAX_IN_FLIGHT} model requests in flight. Wait for one to finish.`);
194	+	response.headers.set("retry-after", "2");
195	+	return response;
196	+	}

464	464		/// gateway's own token, sent as `cf-aig-authorization`.
465	465		#[serde(default)]
466	466		pub gateway_token: Option<String>,
	467	+	/// The most the run may spend on models, in millionths of a dollar: the
	468	+	/// lower of its project's cost cap and its plan's. The proxy refuses
	469	+	/// the run's requests once it has spent this. `None` until the sandbox
	470	+	/// sets it (`cap_model_sessions`), and for the AI Gateway.
	471	+	#[serde(default)]
	472	+	pub cap_micros: Option<i64>,
467	473		}
468	474
469	475		// --- Methods -----------------------------------------------------------------
⋯
722	728		pub token_hashes: Vec<String>,
723	729		}
724	730
	731	+	/// `cap_model_sessions`: sets the most the runs whose model tokens hash to
	732	+	/// these (SHA-256, lowercase hex) may spend on models, in millionths of a
	733	+	/// dollar, which the model proxy holds them to. The sandbox sets it once
	734	+	/// it knows the run's guardrails and plan; zero or less clears it. Returns
	735	+	/// how many open sessions it set.
	736	+	#[derive(Debug, Serialize, Deserialize)]
	737	+	pub struct CapModelSessionsArgs {
	738	+	#[serde(alias = "tokenHashes")]
	739	+	pub token_hashes: Vec<String>,
	740	+	#[serde(alias = "capMicros")]
	741	+	pub cap_micros: i64,
	742	+	}
	743	+
725	744		/// `model_provider`: the workspace's own model connection, if it has one.
726	745		/// Returns `Option<Connection>`.
727	746		#[derive(Debug, Serialize, Deserialize)]

755	755		closeModelSessions: (tokenHashes) => call("close_model_sessions", { token_hashes: tokenHashes }),
756	756		routes: (workspace, viewer) => call("routes", { workspace, viewer }),
757	757		setRoutes: (actor, workspace, routes) => call("set_routes", { actor, workspace, routes }),
	758	+	capModelSessions: (tokenHashes, capMicros) => call("cap_model_sessions", { token_hashes: tokenHashes, cap_micros: capMicros }),
758	759		};
759	760		}
760	761

203	203		authHeader: string \| null;
204	204		/** For an endpoint behind an authenticated Cloudflare AI Gateway: its token. */
205	205		gatewayToken?: string \| null;
	206	+	/**
	207	+	* The most the run may spend on models, in millionths of a dollar: the
	208	+	* lower of its project's cost cap and its plan's. The proxy refuses the
	209	+	* run's requests once it has spent this. Null until the sandbox sets it.
	210	+	*/
	211	+	capMicros?: number \| null;
206	212		};
207	213
208	214		export type ConnectInput = {
⋯
265	271		* how many were open.
266	272		*/
267	273		closeModelSessions(tokenHashes: string[]): Promise<number>;
	274	+	/**
	275	+	* Sets the most the runs whose model tokens hash to these may spend on
	276	+	* models, in millionths of a dollar, which the model proxy holds them
	277	+	* to. Zero clears it. Returns how many open sessions it set.
	278	+	*/
	279	+	capModelSessions(tokenHashes: string[], capMicros: number): Promise<number>;
268	280		}
269	281
270	282		/** What each provider is for, as people choose between them. */

1	+	-- The most a run may spend on models, in millionths of a dollar: the lower
2	+	-- of its project's cost cap and its plan's, set by the sandbox once it
3	+	-- knows them. The model proxy refuses the run's requests past it. Null
4	+	-- until set; the proxy then holds the run to the most any run may cost.
5	+	ALTER TABLE model_sessions ADD COLUMN cap_micros INTEGER;

275	275		spend the same way it reports it for billing and stops the agent once
276	276		the spend reaches the cap. The step in flight when it does can take the
277	277		run a little past it.
	278	+	- g1t's model proxy holds the run to the same cap, whatever happens in the
	279	+	sandbox. It adds up what each of the run's model answers cost, and once
	280	+	the run has spent its cap it refuses the run's model requests with
	281	+	`402` and the error code `run_cap_reached`, which shows in the run's log.
	282	+	The proxy also takes at most 16 of a run's model requests at a time,
	283	+	and a run's model token reaches only `/v1/messages` (with
	284	+	`/v1/messages/count_tokens`) and `/v1/models`. The token stops working
	285	+	when the run ends.
278	286		- The time cap is enforced twice: the harness stops the agent when it
279	287		passes, and the sandbox itself is stopped three minutes after, whatever
280	288		is running in it.

3	3
4	4		import type { ModelUpstream } from "@g1t/contracts";
5	5
6		−	import { isAnswer, tokenReport } from "./report.ts";
	6	+	import { isAnswer, runMayCall, tokenReport } from "./report.ts";
7	7		import { NO_TOKENS, measure } from "./usage.ts";
8	8
9	9		const upstream: ModelUpstream = {
⋯
62	62		await whole.response.text();
63	63		assert.equal(await whole.model, "claude-y");
64	64		});
	65	+
	66	+	test("a run's token reaches answers, token counts and the model list, and nothing else", () => {
	67	+	assert.ok(runMayCall("/v1/messages", "POST"));
	68	+	assert.ok(runMayCall("/v1/messages?beta=true", "POST"));
	69	+	assert.ok(runMayCall("/v1/messages/count_tokens", "POST"));
	70	+	assert.ok(runMayCall("/v1/models", "GET"));
	71	+	assert.ok(runMayCall("/v1/models/claude-sonnet-5-5", "GET"));
	72	+	assert.equal(runMayCall("/v1/messages/batches", "POST"), false);
	73	+	assert.equal(runMayCall("/v1//messages", "POST"), false);
	74	+	assert.equal(runMayCall("/v1/complete", "POST"), false);
	75	+	assert.equal(runMayCall("/v1/models", "POST"), false);
	76	+	assert.equal(runMayCall("/v1/files", "GET"), false);
	77	+	});

28	28		}
29	29
30	30		/**
	31	+	* Whether a run's token may make a request: a model's answer, counting a
	32	+	* request's tokens, or listing models. Nothing else, so no other route
	33	+	* (a batch of messages, say) spends past the run's cap uncounted.
	34	+	*/
	35	+	export function runMayCall(path: string, method: string): boolean {
	36	+	const route = path.split("?")[0];
	37	+	if (isAnswer(route)) return true;
	38	+	if (/^\/v1\/messages\/count_tokens\/?$/.test(route)) return true;
	39	+	return (method === "GET" \|\| method === "HEAD") && /^\/v1\/models(\/[^/]+)?\/?$/.test(route);
	40	+	}
	41	+
	42	+	/**
31	43		* The report for one answer, or null when it used nothing or its session
32	44		* has no id to count it under. The model is the run's when its route names
33	45		* one, else the one that answered.

1	+	/**
2	+	* One run's model spend, as a Durable Object named by its model session.
3	+	*
4	+	* Why an object per run: the count must agree across every isolate the
5	+	* run's requests reach, at once, or an agent that sends many requests in
6	+	* parallel spends the cap once per isolate. A per-isolate count written
7	+	* back now and then has exactly that hole, and the proxy has no database
8	+	* of its own to count in (billing's token count is written after each
9	+	* answer, for usage views, and read nowhere near this fast). One object
10	+	* per run is the natural unit: a run is one model session, its object sees
11	+	* every one of its requests in order, and it costs two short requests per
12	+	* model answer (admit and settle) plus one storage write, far below what
13	+	* the answer itself costs. The object forgets the run a while after its
14	+	* session has lapsed.
15	+	*/
16	+	import { DurableObject } from "cloudflare:workers";
17	+
18	+	import { type Admission, SpendTally } from "./spend.ts";
19	+
20	+	/** How long after its first answer a run's count is kept: longer than a model session lives (3 hours). */
21	+	const FORGET_MS = 4 * 60 * 60_000;
22	+
23	+	export class RunSpend extends DurableObject<object> {
24	+	private tally = new SpendTally();
25	+
26	+	constructor(ctx: DurableObjectState, env: object) {
27	+	super(ctx, env);
28	+	void ctx.blockConcurrencyWhile(async () => {
29	+	this.tally = new SpendTally((await ctx.storage.get<number>("spent")) ?? 0);
30	+	});
31	+	}
32	+
33	+	/** Whether the run may start another answer under `cap`, in millionths of a dollar. */
34	+	async admit(cap: number): Promise<Admission> {
35	+	const admission = this.tally.admit(cap, Date.now());
36	+	if (admission.ok && admission.spent === 0 && (await this.ctx.storage.getAlarm()) === null) {
37	+	await this.ctx.storage.setAlarm(Date.now() + FORGET_MS);
38	+	}
39	+	return admission;
40	+	}
41	+
42	+	/** An answer has ended, costing `micros`. Returns what the run has spent. */
43	+	async settle(ticket: string, micros: number): Promise<number> {
44	+	const before = this.tally.spent;
45	+	const spent = this.tally.settle(ticket, micros);
46	+	if (spent !== before) await this.ctx.storage.put("spent", spent);
47	+	return spent;
48	+	}
49	+
50	+	override async alarm(): Promise<void> {
51	+	await this.ctx.storage.deleteAll();
52	+	this.tally = new SpendTally();
53	+	}
54	+	}

16	16		// Who an AI Gateway request's access token belongs to, and its scopes.
17	17		{ "binding": "IDENTITY", "service": "g1t-identity" }
18	18		],
	19	+	// Each run's model spend, one object per model session, so its cost cap
	20	+	// holds across every isolate its requests reach (src/run-spend.ts).
	21	+	"durable_objects": {
	22	+	"bindings": [{ "name": "RUN_SPEND", "class_name": "RunSpend" }]
	23	+	},
	24	+	"migrations": [{ "tag": "v1", "new_sqlite_classes": ["RunSpend"] }],
19	25		"vars": {
20	26		// g1t's own route, for runs g1t pays for: a Cloudflare AI Gateway,
21	27		// which holds g1t's key.

2	2		import { createHash } from "node:crypto";
3	3		import { test } from "node:test";
4	4
5		−	import { credentialHashes, holdCredentials, modelTokenHashes, pushGrant, remotePath, revokeCredentials, sha256Hex } from "./credentials.ts";
	5	+	import { capModelTokens, credentialHashes, holdCredentials, modelTokenHashes, pushGrant, remotePath, revokeCredentials, sha256Hex } from "./credentials.ts";
6	6
7	7		test("a remote names its repository", () => {
8	8		assert.deepEqual(remotePath("https://g1t.sh/acme/rocket.git"), { namespace: "acme", name: "rocket" });
⋯
98	98		assert.equal(integrations.calls[0]?.method, "close_model_sessions");
99	99		assert.equal(identity.calls.length, 0);
100	100		});
	101	+
	102	+	test("a run's model token is held to its cost cap, in millionths", async () => {
	103	+	const integrations = fakeIdentity();
	104	+	const storage = memoryStorage();
	105	+	const token = "g1tm_0123456789abcdef";
	106	+	await holdCredentials(fakeIdentity(), storage, { ANTHROPIC_API_KEY: token }, null);
	107	+	await capModelTokens(integrations, storage, 2.5);
	108	+	assert.equal(integrations.calls[0]?.method, "cap_model_sessions");
	109	+	assert.deepEqual(integrations.calls[0].body, { token_hashes: [await sha256Hex(token)], cap_micros: 2_500_000 });
	110	+	});
	111	+
	112	+	test("no cap, or no model token, asks integrations nothing", async () => {
	113	+	const integrations = fakeIdentity();
	114	+	const storage = memoryStorage();
	115	+	await capModelTokens(integrations, storage, 2);
	116	+	await holdCredentials(fakeIdentity(), storage, { ANTHROPIC_API_KEY: "g1tm_x" }, null);
	117	+	await capModelTokens(integrations, storage, null);
	118	+	await capModelTokens(integrations, storage, 0);
	119	+	assert.equal(integrations.calls.length, 0);
	120	+	});

100	100		}
101	101
102	102		/**
	103	+	* Tells integrations the most the run may spend on models, for the model
	104	+	* tokens `holdCredentials` kept, so the model proxy holds them to it
	105	+	* whatever the sandbox does (the harness's own cap is only the sandbox's
	106	+	* word). No cap, or no model token, leaves the proxy's backstop. Never
	107	+	* stops the sandbox from starting.
	108	+	*/
	109	+	export async function capModelTokens(integrations: ServiceBinding, storage: Storage, budgetUsd: number \| null \| undefined): Promise<void> {
	110	+	if (typeof budgetUsd !== "number" \|\| !Number.isFinite(budgetUsd) \|\| budgetUsd <= 0) return;
	111	+	const models = await storage.get<string[]>(MODEL_STORAGE_KEY);
	112	+	if (!models?.length) return;
	113	+	await call(integrations, "cap_model_sessions", { token_hashes: models, cap_micros: Math.round(budgetUsd * 1_000_000) }).catch((error: unknown) =>
	114	+	console.log("model sessions not capped", String(error)),
	115	+	);
	116	+	}
	117	+
	118	+	/**
103	119		* Ends a sandbox's credentials, once: its g1t tokens, and, when
104	120		* `integrations` is given, its model tokens, which the model proxy then
105	121		* refuses within seconds rather than when they would lapse.

90	90		import { BUMP_MINUTES, BUMP_TOKEN_TTL_SECONDS, bumpEnv, bumpProblem, bumpSandboxName, systemActor, registryHosts } from "./bump";
91	91		import { BACKUP_MINUTES, backupEnv, backupPace, backupSandboxName } from "./backup";
92	92		import { type ProjectSurroundings, readableSurroundings } from "./surroundings";
93		−	import { holdCredentials, pushGrant, remotePath, revokeCredentials, runCredential } from "./credentials";
	93	+	import { capModelTokens, holdCredentials, pushGrant, remotePath, revokeCredentials, runCredential } from "./credentials";
94	94		import { buildMentionPrompt, describeThread, handleMention, planMention } from "./mentions";
95	95		import { instructionsFor, repoInstructions, withBlock } from "./repo-instructions";
96	96		import { cancelTask, enqueueTask, handedOverStep, selfHostedRoute, taskEnv, taskRepo } from "./self-hosted";
⋯
450	450		const tracked = track ? await this.openRun(track, envVars, guard) : null;
451	451		// Its credentials are tied to the run, and revoked when it stops.
452	452		await holdCredentials(this.env.IDENTITY, this.ctx.storage, envVars, tracked?.runId ?? null);
	453	+	// Its model token is held to its cost cap by the model proxy too.
	454	+	await capModelTokens(this.env.INTEGRATIONS, this.ctx.storage, guard?.policy.budgetUsd ?? limits?.budgetUsd);
453	455		try {
454	456		const vars = tracked ? { ...envVars, AGENT_RUN: tracked.runId, AGENT_RUN_TOKEN: tracked.token } : envVars;
455	457		// The workspace's own runner, not a container: the same environment,