Skip to content
Sign up

flagon-io/g1t

public
0
356 linesCodeBlameRaw
1/**
2 * Metered model work: the one door every reply and every session step goes
3 * through, so g1t meters and bills an agent's model work one way
4 * (docs/WORKSPACE.md, "Budgets").
5 *
6 * 1. The paying agent's own monthly and daily caps (`budget.ts`), and the
7 * workspace's budget for all its agents together (`policy.ts`).
8 * 2. Whether the agent may use a model at all: g1t's hosted models are open
9 * as the runner decides (`hostedOpen`, `HOSTED_AGENT_WORKSPACES` and
10 * billing's status), or the workspace's own provider; the agent's
11 * `providers` narrows that.
12 * 3. A model session from integrations (`openModelSession`), routed by the
13 * workspace's model routes, as for a run.
14 * 4. The compute gate's reservation (`ComputeGate.admit`, kind `agent`):
15 * the workspace's spend limit, AI credit, pauses and g1t's breaker.
16 * 5. A billing run (`start_run`), so the work is charged as Agent tokens on
17 * the workspace's bill, under the paying agent.
18 * 6. The work itself, through the model proxy with the session's token.
19 * 7. `finish_run` with its cost and tokens, the reservation settled at
20 * cost, and the charge added to the paying agent's and the workspace's
21 * agent spend.
22 *
23 * Who does the work and who pays can differ: a colleague brought into a
24 * session, or a subagent, works on the budget of the agent at the root of
25 * the session's tree, so a chain never escapes the budget that started it.
26 */
27import { type ModelSession, type ModelTier, type RunTicket, ComputeGate, MODEL_ESTIMATE_MICROS, billingClient, integrationsClient } from "@g1t/contracts";
28
29import { hostedOpen } from "../../runner/src/hosted.ts";
30import { type AgentRouting as Policy, routingReader } from "../../runner/src/model-env.ts";
31import { type Spent, type Tokens, budgetBlock, chargedMicros, replyCapMicros, totalTokens } from "./budget.ts";
32import { BUILTIN_NO_MODEL } from "./orchestrator.ts";
33import { type PolicyRow, alertDue, markAlerted, policyBlock, readPolicy, workspaceSpendStatements } from "./policy.ts";
34import { dollars } from "./money.ts";
35import { type ReplyModel, allowedProviders, replyModel } from "./routing.ts";
36import { type Row, definitionOf, periods, spendStatements } from "./store.ts";
37import type { ModelAnswer, Send } from "./turn.ts";
38import type { ServiceBinding } from "@g1t/contracts";
39
40export type MeterEnv = {
41 DB: D1Database;
42 BILLING: ServiceBinding;
43 INTEGRATIONS: ServiceBinding;
44 MODELS?: ServiceBinding;
45 MODELS_URL?: string;
46 HOSTED_AGENT_WORKSPACES: string;
47 AGENT_ROUTING: string;
48 /** Notifications: the workspace's agent budget crossing 75, 90 or 100%. */
49 NOTIFY?: ServiceBinding;
50};
51
52/** The longest one model answer may take. */
53const MODEL_TIMEOUT_MS = 120_000;
54
55/** Staff's model defaults on top of `AGENT_ROUTING`, read at most once a minute, as in the runner. */
56const routingNow = routingReader();
57let gate: ComputeGate | null = null;
58
59/**
60 * Where an agent's spend shows in billing: the workspace, under the agent
61 * that pays. Billing keys runs and reservations by a repository; no
62 * repository is named with an `@`, so the agent's line never mixes with a
63 * project's.
64 */
65export function billingRepo(workspace: string, handle: string): { namespace: string; name: string } {
66 return { namespace: workspace.toLowerCase(), name: `@${handle}` };
67}
68
69async function rpc<T>(service: ServiceBinding, method: string, args: object): Promise<T> {
70 const response = await service.fetch(`https://service/rpc/${method}`, {
71 method: "POST",
72 headers: { "content-type": "application/json" },
73 body: JSON.stringify(args),
74 });
75 if (!response.ok) throw new Error(`${method} failed with status ${response.status}`);
76 return (await response.json()) as T;
77}
78
79type PriceTerms = { marginPercent: number; rate: number; rateOwn: number };
80let terms: { value: PriceTerms; until: number } | null = null;
81
82/** The model margin and the agent rates from billing's price book, kept ten minutes. */
83async function priceTerms(billing: ServiceBinding): Promise<PriceTerms> {
84 if (terms && terms.until > Date.now()) return terms.value;
85 type Book = { prices?: { meter: string; priceMicros?: number; price_micros?: number }[]; modelMarginPercent?: number; model_margin_percent?: number };
86 const book = await rpc<Book>(billing, "prices", {}).catch(() => null);
87 const price = (meter: string) => {
88 const found = book?.prices?.find((p) => p.meter === meter);
89 return found?.priceMicros ?? found?.price_micros ?? 0;
90 };
91 const value = {
92 marginPercent: book?.modelMarginPercent ?? book?.model_margin_percent ?? 0,
93 rate: price("agent_tokens"),
94 rateOwn: price("agent_tokens_own"),
95 };
96 // A failed read is tried again in a minute, not kept.
97 terms = { value, until: Date.now() + (book ? 10 * 60_000 : 60_000) };
98 return value;
99}
100
101export async function sha256Hex(text: string): Promise<string> {
102 const digest = await crypto.subtle.digest("SHA-256", new TextEncoder().encode(text));
103 return [...new Uint8Array(digest)].map((b) => b.toString(16).padStart(2, "0")).join("");
104}
105
106/**
107 * How work asks the model: one Messages API request through the model
108 * proxy, with the model session's token, non-streamed.
109 */
110function sendFor(env: MeterEnv, token: string): Send {
111 // The binding, not the proxy's public address: a Worker fetching another
112 // Worker's domain on the same zone can be refused or loop. The proxy reads
113 // only the path and the token, so the host does not matter.
114 const path = "/anthropic/v1/messages";
115 const fetcher = (url: string, init: RequestInit) => (env.MODELS ? env.MODELS.fetch(url, init) : fetch(url, init));
116 return async (body) => {
117 const response = await fetcher(env.MODELS ? `https://models${path}` : `${env.MODELS_URL!.replace(/\/+$/, "")}${path}`, {
118 method: "POST",
119 headers: { "content-type": "application/json", "x-api-key": token, "anthropic-version": "2023-06-01" },
120 body: JSON.stringify(body),
121 signal: AbortSignal.timeout(MODEL_TIMEOUT_MS),
122 });
123 const json = (await response.json().catch(() => null)) as (ModelAnswer & { error?: { message?: string } }) | null;
124 if (!response.ok || !json) throw new Error(`the model answered ${response.status}: ${json?.error?.message ?? "no answer"}`);
125 return json;
126 };
127}
128
129/** What the work is given: how to ask the model, and on what. */
130export type Model = {
131 send: Send;
132 model: ReplyModel;
133 /** The workspace's own provider pays for the model (g1t charges only the agent rate). */
134 ownModel: boolean;
135 policy: Policy;
136 /** The model the workspace's route names, on its own provider. */
137 sessionModel: string | null;
138 /** The workspace's own model connection, if any. */
139 own: string | null;
140};
141
142/** What the work used: its own tokens and cost, and any it spent for others (consults). */
143export type WorkUsage = { tokens: Tokens; cost: number; rounds: number };
144
145export type MeterInput = {
146 /** The agent doing the work. */
147 row: Row;
148 /** The agent whose budget pays; the same agent unless this is part of another's session. */
149 payer: Row;
150 /** The workspace's slug. */
151 slug: string;
152 task: "reply" | "session";
153 /** The tier the work starts on, before the agent's limits. */
154 start: ModelTier;
155 /** Who asked, by username, for billing's record. */
156 askerName: string | null;
157 /** What is left of a session's cap, so one step never overruns it. */
158 leftMicros?: number | null;
159 /** Agent tier limits narrower than the agent's own (a subagent's). */
160 limits?: { floor: ModelTier | null; ceiling: ModelTier | null } | null;
161};
162
163export type MeterBlock = { ok: false; reason: string; message: string };
164export type MeterDone<T> = { ok: true; value: T; model: string; tier: ModelTier | null; tokens: Tokens; cost: number; charged: number };
165
166/**
167 * Runs `work` as metered model work for `input.row`, paid by
168 * `input.payer`: checks every limit, opens and closes the model session,
169 * bills it, and records the spend. A limit that says no comes back as a
170 * block with what to tell people, before anything was spent. Whatever
171 * `work` throws is thrown on, after what was held is given back.
172 */
173export async function metered<T extends WorkUsage>(env: MeterEnv, input: MeterInput, work: (model: Model) => Promise<T>, now = new Date()): Promise<MeterBlock | MeterDone<T>> {
174 const db = env.DB;
175 const { row, payer, slug } = input;
176 if (!env.MODELS && !env.MODELS_URL) return { ok: false, reason: "no_models_url", message: "This installation has no model proxy set up." };
177 const billing = billingClient(env.BILLING);
178 const integrations = integrationsClient(env.INTEGRATIONS);
179 gate ??= new ComputeGate(env.BILLING);
180
181 // 1. The paying agent's caps, and the workspace's budget for every agent.
182 const payerDefinition = definitionOf(payer);
183 const [month, day] = periods(now);
184 const [spentRows, policy] = await Promise.all([
185 db
186 .prepare("SELECT period, micros FROM agent_spend WHERE agent_id = ? AND period IN (?, ?)")
187 .bind(payer.id, month, day)
188 .all<{ period: string; micros: number }>(),
189 readPolicy(db, row.workspace_id, month),
190 ]);
191 const spent: Spent = {
192 month: spentRows.results.find((r) => r.period === month)?.micros ?? 0,
193 day: spentRows.results.find((r) => r.period === day)?.micros ?? 0,
194 };
195 const blocked = budgetBlock(payerDefinition.budget, spent, now);
196 if (blocked) {
197 const message = payer.id === row.id ? blocked.message : `@${payer.handle}, who this work is for, is out of budget.`;
198 return { ok: false, reason: `budget_${blocked.cap}`, message };
199 }
200 const pool = policyBlock(policy);
201 if (pool) return { ok: false, reason: "workspace_agent_budget", message: pool };
202
203 // 2. Whether it may use a model at all.
204 const definition = definitionOf(row);
205 const [own, status] = await Promise.all([
206 integrations.modelProvider(slug).catch(() => null),
207 billing.status().catch(() => ({ enabled: false, live: false })),
208 ]);
209 const allowed = allowedProviders(definition.routing, own?.id ?? null);
210 const mayHosted = hostedOpen(slug, env.HOSTED_AGENT_WORKSPACES, status) && allowed.hosted;
211 if (!mayHosted && !allowed.own) {
212 const message = own
213 ? "My settings don't let me use any model this workspace has. An owner can change my providers on my profile."
214 : allowed.hosted
215 ? row.builtin
216 ? BUILTIN_NO_MODEL
217 : "g1t's hosted models aren't open to this workspace, and it has no model provider of its own. An owner can connect one under Integrations."
218 : "My settings let me use only this workspace's own model providers, and it has none. An owner can connect one under Integrations, or change my providers on my profile.";
219 return { ok: false, reason: "no_model", message };
220 }
221
222 // 3. A model session, routed by the workspace's model routes, billed under the payer.
223 const repo = billingRepo(slug, payer.handle);
224 const routing = await routingNow(env.AGENT_ROUTING, () => billing.modelDefaults());
225 const limits = { ...definition.routing, ...(input.limits ?? {}) };
226 const provisional = replyModel(routing, { ...limits, pinned: null }, { start: input.start });
227 const opened = await integrations.openModelSession({
228 workspace: slug,
229 repo,
230 number: 0,
231 task: input.task,
232 hostedOpen: mayHosted,
233 tier: provisional.tier,
234 requestedBy: input.askerName,
235 });
236 if (!opened.ok) return { ok: false, reason: "model_route", message: opened.error.message };
237 const session: ModelSession = opened.value;
238 let reservation: string | null = null;
239 let settled = false;
240 try {
241 const ownModel = session.billedTo === "workspace";
242 if (ownModel ? !allowed.own : !mayHosted) {
243 return {
244 ok: false,
245 reason: "provider_not_allowed",
246 message:
247 "This workspace routes agents to a model my settings don't allow. An owner can change my providers on my profile, or the workspace's model routes under Integrations.",
248 };
249 }
250 // A pinned model is for the workspace's own endpoints; on g1t's models the tier decides.
251 const model = replyModel(
252 routing,
253 { ...limits, pinned: ownModel ? definition.routing.pinned : null },
254 { chosen: session.tierChoice ?? null, named: ownModel ? session.model : null, start: input.start },
255 );
256
257 // 4. The workspace's own limits, through the compute gate.
258 const ent = await gate.entitlements(slug);
259 const estimate = ownModel ? 0 : input.task === "reply" ? MODEL_ESTIMATE_MICROS.reply : MODEL_ESTIMATE_MICROS.reply * 4;
260 const admission = await gate.admit({ workspace: slug, repo, public: false, kind: "agent", estimateMicros: estimate, hostedModel: !ownModel }, ent);
261 if (!admission.ok) return { ok: false, reason: `workspace_${admission.code}`, message: admission.message };
262 reservation = admission.reservation?.id ?? null;
263
264 // 5. The run it is billed as.
265 const named = ownModel && session.model ? session.model : null;
266 const started = await billing.startRun({
267 workspace: slug,
268 repo,
269 number: 0,
270 task: input.task,
271 model: ownModel ? `${model.modelName} (${session.providerName ?? "own provider"})` : model.modelName,
272 billedTo: ownModel ? "workspace" : "g1t",
273 session: session.id,
274 tier: named ? null : model.tier,
275 });
276 if (!started.ok) return { ok: false, reason: "billing", message: started.error.message };
277 const ticket: RunTicket | null = started.value;
278 const caps = [replyCapMicros(payerDefinition.budget, spent, ent && ent.runCapMicros > 0 ? ent.runCapMicros : null)];
279 if (input.leftMicros != null) caps.push(Math.max(1, Math.floor(input.leftMicros)));
280 const left = policy.monthly_micros ? policy.monthly_micros - policy.spent : null;
281 if (left != null) caps.push(Math.max(1, left));
282 const cap = caps.filter((c): c is number => c != null);
283 if (cap.length) await integrations.capModelSessions([await sha256Hex(session.token)], Math.min(...cap)).catch(() => 0);
284
285 // 6. The work.
286 const result = await work({ send: sendFor(env, session.token), model, ownModel, policy: routing, sessionModel: session.model, own: own?.id ?? null });
287 const priced = await priceTerms(env.BILLING);
288 const charged = chargedMicros({
289 costMicros: result.cost,
290 hosted: !ownModel,
291 marginPercent: priced.marginPercent,
292 ratePerMillionMicros: ownModel ? priced.rateOwn : priced.rate,
293 tokens: totalTokens(result.tokens),
294 });
295
296 // 7. Bill it, settle, and count it against the payer and the workspace.
297 if (ticket) {
298 await rpc(env.BILLING, "finish_run", { runId: ticket.runId, token: ticket.token, costUsd: result.cost / 1_000_000, turns: result.rounds, tokens: result.tokens }).catch(
299 (error: unknown) => console.error("agents: finish_run failed", ticket.runId, String(error)),
300 );
301 }
302 if (reservation) {
303 settled = true;
304 await gate.settle(reservation, result.cost);
305 }
306 if (charged > 0) {
307 await db.batch([...spendStatements(db, payer.id, charged, now, input.task), ...workspaceSpendStatements(db, row.workspace_id, charged, now)]);
308 await budgetAlert(env, slug, row.workspace_id, { ...policy, spent: policy.spent + charged }, month).catch((error: unknown) =>
309 console.error("agents: a budget alert was not sent", slug, String(error)),
310 );
311 }
312 return { ok: true, value: result, model: model.modelName, tier: named ? null : model.tier, tokens: result.tokens, cost: result.cost, charged };
313 } finally {
314 // What was held is given back however the work ended, and the model
315 // session's token stops working.
316 if (reservation && !settled) await gate.settle(reservation, 0);
317 await integrations.closeModelSessions([await sha256Hex(session.token)]).catch(() => 0);
318 }
319}
320
321/**
322 * Tells the owner who set the workspace's agent budget, once per level a
323 * month, when every agent's spend together crosses 75, 90 or 100% of it.
324 */
325async function budgetAlert(env: MeterEnv, slug: string, workspaceId: string, policy: PolicyRow, month: string): Promise<void> {
326 const level = alertDue(policy);
327 if (!level || !env.NOTIFY || !policy.monthly_micros) return;
328 if (!(await markAlerted(env.DB, workspaceId, month, level))) return;
329 const setBy = await env.DB.prepare("SELECT updated_by FROM agent_policies WHERE workspace_id = ?").bind(workspaceId).first<{ updated_by: string | null }>();
330 if (!setBy?.updated_by) return;
331 const title = level >= 100 ? "Your agents have used this month's budget" : `Your agents have used ${level}% of this month's budget`;
332 const body =
333 level >= 100
334 ? `${dollars(policy.spent)} of ${dollars(policy.monthly_micros)}. They won't start new work until it's raised or the month turns.`
335 : `${dollars(policy.spent)} of ${dollars(policy.monthly_micros)} so far this month.`;
336 await env.NOTIFY.fetch("https://service/rpc/notify", {
337 method: "POST",
338 headers: { "content-type": "application/json" },
339 body: JSON.stringify({
340 target: { username: setBy.updated_by },
341 notification: {
342 id: `agent-budget:${workspaceId}:${month}:${level}`,
343 kind: "approval",
344 workspace: slug,
345 title,
346 body,
347 href: `/${slug}/-/agents`,
348 actor: { kind: "system", id: "g1t", name: "g1t" },
349 channel_id: null,
350 created_at: new Date().toISOString(),
351 },
352 }),
353 });
354}
355
356export type { PolicyRow };