Skip to content
Sign up

flagon-io/g1t

public
0
315 linesCodeBlameRaw

Pick any line to see why it is the way it is: the commit, the pull request and issue it came from, and what the agent was thinking.

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