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