Sign in

flagon-io/g1t

public

g1t/services/billing/src/lib.rs

1,060 lines41,773 bytesCodeBlame

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 as a team: lifecycle, merge queue, billing and a new shell1//! The billing service: what agents cost, charged to the workspace they
2//! worked for.
3//!
4//! A workspace buys credit with a card. Before the runner starts an agent
5//! it asks here, and is refused if the workspace has none. When the
6//! agent's sandbox finishes it reports what the model cost, and that plus
7//! g1t's margin comes off the balance. Every change is a ledger entry, and
8//! a balance is always the sum of its ledger.
9//!
Paid features: a workspace turns on Deployments with a monthly plan10//! Paid features (deployments) are bought separately, as monthly plans;
11//! see `features`. They are never free.
12//!
Agents as a team: lifecycle, merge queue, billing and a new shell13//! Without a card processor configured the service says so and charges
14//! nothing, so that g1t still runs where billing has not been set up.
15//!
16//! Reached only through service bindings; see `g1t_contracts::billing` for
17//! the methods and their arguments.
18
Billing accounts, terms and enterprises; g1t is no longer free19mod accounts;
Paid features: a workspace turns on Deployments with a monthly plan20mod features;
Prices keep themselves current with what g1t pays21mod keeper;
Usage limits: unpaid usage can only go so far22mod limits;
Agents as a team: lifecycle, merge queue, billing and a new shell23mod stripe;
24
25use g1t_contracts::billing::*;
26use g1t_contracts::time::rfc3339;
27use g1t_contracts::{FailureCode, Outcome, Role, new_id};
Billing accounts, terms and enterprises; g1t is no longer free28use g1t_contracts::billing::TermsKind;
Agents as a team: lifecycle, merge queue, billing and a new shell29use g1t_kit::{args, now_ms, reply, rpc_method};
30use serde::Deserialize;
31use sha2::{Digest, Sha256};
32use worker::wasm_bindgen::JsValue;
Prices keep themselves current with what g1t pays33use worker::{Context, D1Database, Env, Request, Response, Result, ScheduleContext, ScheduledEvent, event};
Agents as a team: lifecycle, merge queue, billing and a new shell34
35use stripe::Stripe;
36
37const MIN_TOP_UP_CENTS: u32 = 500;
38const MAX_TOP_UP_CENTS: u32 = 50_000;
39const LEDGER_PAGE: u32 = 100;
40/// A run's reported cost is believed up to this much. A sandbox cannot
41/// spend more in the time it has, so anything above is a fault.
42const MAX_RUN_COST_USD: f64 = 100.0;
43
44/// What a run is charged: its cost plus the margin, rounded up to a whole
45/// millionth of a dollar.
46pub fn charge_micros(cost_usd: f64, margin_percent: u32) -> i64 {
47 let cost_micros = (cost_usd.clamp(0.0, MAX_RUN_COST_USD) * MICROS_PER_DOLLAR as f64).ceil();
48 (cost_micros * f64::from(100 + margin_percent) / 100.0).ceil() as i64
49}
50
51fn hash(token: &str) -> String {
52 hex::encode(Sha256::digest(token.as_bytes()))
53}
54
55fn optional(value: Option<&str>) -> JsValue {
56 value.map_or(JsValue::NULL, JsValue::from)
57}
58
59#[derive(Deserialize)]
60struct AccountRow {
61 balance_micros: i64,
62 customer_id: Option<String>,
63}
64
65#[derive(Deserialize)]
66struct LedgerRow {
67 id: String,
68 kind: EntryKind,
69 amount_micros: i64,
70 description: String,
71 repo: Option<String>,
72 number: Option<u32>,
73 task: Option<String>,
74 model: Option<String>,
75 created_by: Option<String>,
76 created_at: String,
Integrations: your own model provider, alerts that open issues, tickets agents read77 billed_to: Option<String>,
Billing on Stripe's pages, month-end charges, warnings; sudo by workspace78 #[serde(default)]
79 workspace: Option<String>,
Agents as a team: lifecycle, merge queue, billing and a new shell80}
81
82impl From<LedgerRow> for LedgerEntry {
83 fn from(row: LedgerRow) -> Self {
84 LedgerEntry {
85 id: row.id,
86 kind: row.kind,
87 amount_micros: row.amount_micros,
88 description: row.description,
89 repo: row.repo,
90 number: row.number,
91 task: row.task,
92 model: row.model,
Integrations: your own model provider, alerts that open issues, tickets agents read93 billed_to: row.billed_to.unwrap_or_else(|| "g1t".to_owned()),
Agents as a team: lifecycle, merge queue, billing and a new shell94 created_by: row.created_by,
95 created_at: row.created_at,
Billing on Stripe's pages, month-end charges, warnings; sudo by workspace96 workspace: row.workspace,
Agents as a team: lifecycle, merge queue, billing and a new shell97 }
98 }
99}
100
101#[derive(Deserialize)]
102struct RunRow {
103 workspace: String,
104 repo: String,
105 number: u32,
106 task: String,
107 model: String,
108 token_hash: String,
Integrations: your own model provider, alerts that open issues, tickets agents read109 billed_to: Option<String>,
110}
111
112impl RunRow {
113 fn own_provider(&self) -> bool {
114 self.billed_to.as_deref() == Some("workspace")
115 }
Agents as a team: lifecycle, merge queue, billing and a new shell116}
117
118#[derive(Deserialize)]
119struct CheckoutRow {
120 workspace: String,
121 created_by: String,
122}
123
124/// A row an `UPDATE … RETURNING` touched.
125#[derive(Deserialize)]
126struct Touched {
127 #[allow(dead_code)]
128 id: String,
129}
130
131struct Billing {
132 db: D1Database,
133 /// Absent when no card processor is configured.
134 stripe: Option<Stripe>,
135 margin_percent: u32,
Integrations: your own model provider, alerts that open issues, tickets agents read136 /// Charged for a run on the workspace's own model provider.
137 orchestration_fee_micros: i64,
Free while g1t is being built out; agents can check out their own forks138 /// While g1t is being built out, nothing is charged (`FREE_WHILE_BUILDING`).
139 free: bool,
A free allowance on g1t's models, so anyone can try its agents140 /// The free allowance on g1t's hosted models, when there is one.
141 trial: Option<TrialConfig>,
Paid features: a workspace turns on Deployments with a monthly plan142 /// The Deployments plan's monthly price (`DEPLOYMENTS_MONTHLY_CENTS`).
143 deployments_monthly_cents: u32,
Usage limits: unpaid usage can only go so far144 /// How far unpaid usage may go; see `limits`.
145 ceilings: limits::Ceilings,
Billing accounts, terms and enterprises; g1t is no longer free146 /// `PREPAID_ONLY`: the old rule, that agents need credit first.
147 prepaid_only: bool,
A free allowance on g1t's models, so anyone can try its agents148}
149
150/// `TRIAL_WORKSPACE_MICROS`, `TRIAL_TOTAL_MICROS` and `TRIAL_UNTIL`.
151struct TrialConfig {
152 per_workspace_micros: i64,
153 total_micros: i64,
154 /// RFC 3339, in UTC.
155 until: String,
156}
157
158#[derive(serde::Deserialize)]
159struct Sum {
160 micros: Option<i64>,
Agents as a team: lifecycle, merge queue, billing and a new shell161}
162
163impl Billing {
164 fn status(&self) -> Status {
165 Status {
166 enabled: self.stripe.is_some(),
167 live: self.stripe.as_ref().is_some_and(Stripe::live),
Free while g1t is being built out; agents can check out their own forks168 free: self.free,
Agents as a team: lifecycle, merge queue, billing and a new shell169 }
170 }
171
172 async fn row(&self, workspace: &str) -> Result<Option<AccountRow>> {
173 self.db
174 .prepare("SELECT balance_micros, customer_id FROM accounts WHERE workspace = ?")
175 .bind(&[workspace.into()])?
176 .first::<AccountRow>(None)
177 .await
178 }
179
180 async fn standing(&self, workspace: &str) -> Result<Account> {
Billing on Stripe's pages, month-end charges, warnings; sudo by workspace181 let row = self.row(workspace).await?;
182 let card = match (&self.stripe, row.as_ref().and_then(|row| row.customer_id.as_deref())) {
183 (Some(stripe), Some(customer)) => stripe.card(customer).await.ok().flatten().map(|card| Card {
184 brand: card.brand,
185 last4: card.last4,
186 exp_month: card.exp_month,
187 exp_year: card.exp_year,
188 }),
189 _ => None,
190 };
Agents as a team: lifecycle, merge queue, billing and a new shell191 Ok(Account {
192 workspace: workspace.to_owned(),
Billing on Stripe's pages, month-end charges, warnings; sudo by workspace193 balance_micros: row.map_or(0, |row| row.balance_micros),
Agents as a team: lifecycle, merge queue, billing and a new shell194 status: self.status(),
195 margin_percent: self.margin_percent,
Integrations: your own model provider, alerts that open issues, tickets agents read196 orchestration_fee_micros: self.orchestration_fee_micros,
Billing on Stripe's pages, month-end charges, warnings; sudo by workspace197 card,
Agents as a team: lifecycle, merge queue, billing and a new shell198 })
199 }
200
Billing on Stripe's pages, month-end charges, warnings; sudo by workspace201 /// The workspace's customer at Stripe, made the first time one is needed.
202 pub(crate) async fn customer_for(&self, workspace: &str) -> Result<String> {
203 let Some(stripe) = &self.stripe else {
204 return Err(worker::Error::RustError("payments are not set up".into()));
205 };
206 if let Some(customer) = self.row(workspace).await?.and_then(|row| row.customer_id) {
207 return Ok(customer);
208 }
209 let customer = stripe.create_customer(workspace).await?;
210 self.db
211 .prepare(
212 "INSERT INTO accounts (workspace, balance_micros, customer_id, created_at) VALUES (?1, 0, ?2, ?3)
213 ON CONFLICT (workspace) DO UPDATE SET customer_id = ?2",
214 )
215 .bind(&[workspace.into(), customer.as_str().into(), rfc3339(now_ms()).into()])?
216 .run()
217 .await?;
218 Ok(customer)
219 }
220
221 /// Stripe's hosted billing page for the workspace. Owners only.
222 async fn billing_portal(&self, a: BillingPortalArgs) -> Result<Outcome<Checkout>> {
223 let workspace = a.workspace.to_lowercase();
224 if a.actor.role_in(&workspace) != Some(Role::Owner) {
225 return Ok(Outcome::fail(FailureCode::Forbidden, "Only an owner can manage the workspace's billing."));
226 }
227 let Some(stripe) = &self.stripe else {
228 return Ok(Outcome::fail(FailureCode::Conflict, "Payments are not set up on this g1t."));
229 };
230 let customer = match self.customer_for(&workspace).await {
231 Ok(customer) => customer,
232 Err(error) => return Ok(Outcome::fail(FailureCode::Conflict, format!("Stripe could not be reached: {error}"))),
233 };
234 match stripe.portal_session(&customer, &a.return_url).await {
235 Ok(url) => Ok(Outcome::Ok(Checkout { url })),
236 Err(error) if stripe::is_missing(&error) => {
237 // The customer was removed at Stripe: a new one next time.
238 self.forget_customer(&workspace).await?;
239 Ok(Outcome::fail(FailureCode::Conflict, "Stripe no longer had this workspace's customer. Try again."))
240 }
241 Err(error) => Ok(Outcome::fail(FailureCode::Conflict, format!("Stripe's billing page could not be opened: {error}"))),
242 }
243 }
244
Agents as a team: lifecycle, merge queue, billing and a new shell245 /// Adds a ledger entry and moves the balance by the same amount, as
246 /// one write.
247 #[allow(clippy::too_many_arguments)]
248 async fn enter(
249 &self,
250 workspace: &str,
251 kind: EntryKind,
252 amount_micros: i64,
253 description: &str,
254 reference: &str,
255 run: Option<&RunRow>,
256 cost_micros: Option<i64>,
257 created_by: Option<&str>,
258 customer: Option<&str>,
259 ) -> Result<()> {
260 let now = now_ms();
261 let timestamp = rfc3339(now);
262 let kind = match kind {
263 EntryKind::TopUp => "top_up",
264 EntryKind::Usage => "usage",
265 };
266 self.db
267 .batch(vec![
268 self.db
269 .prepare(
270 "INSERT INTO ledger
271 (id, workspace, kind, amount_micros, description, repo, number, task,
Integrations: your own model provider, alerts that open issues, tickets agents read272 model, cost_micros, reference, created_by, created_at, billed_to)
273 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
Agents as a team: lifecycle, merge queue, billing and a new shell274 )
275 .bind(&[
276 new_id("led", now).into(),
277 workspace.into(),
278 kind.into(),
279 // D1 takes numbers as doubles, which hold every
280 // amount this service will see exactly.
281 (amount_micros as f64).into(),
282 description.into(),
283 optional(run.map(|run| run.repo.as_str())),
284 run.map_or(JsValue::NULL, |run| run.number.into()),
285 optional(run.map(|run| run.task.as_str())),
286 optional(run.map(|run| run.model.as_str())),
287 cost_micros.map_or(JsValue::NULL, |cost| (cost as f64).into()),
288 reference.into(),
289 optional(created_by),
290 timestamp.as_str().into(),
Integrations: your own model provider, alerts that open issues, tickets agents read291 run.map_or("g1t", |run| if run.own_provider() { "workspace" } else { "g1t" }).into(),
Agents as a team: lifecycle, merge queue, billing and a new shell292 ])?,
293 self.db
294 .prepare(
295 "INSERT INTO accounts (workspace, balance_micros, customer_id, created_at)
296 VALUES (?1, ?2, ?3, ?4)
297 ON CONFLICT (workspace) DO UPDATE SET
298 balance_micros = balance_micros + ?2,
299 customer_id = COALESCE(?3, customer_id)",
300 )
301 .bind(&[
302 workspace.into(),
303 (amount_micros as f64).into(),
304 optional(customer),
305 timestamp.as_str().into(),
306 ])?,
307 ])
308 .await?;
Billing accounts, terms and enterprises; g1t is no longer free309 // Money in clears a card declined at the limit.
310 if kind == "top_up" {
311 self.db
312 .prepare("UPDATE limits SET autopay_failed_at = NULL, autopay_error = NULL WHERE workspace = ?")
313 .bind(&[workspace.into()])?
314 .run()
315 .await?;
316 }
Agents as a team: lifecycle, merge queue, billing and a new shell317 Ok(())
318 }
319
320 async fn account(&self, a: AccountArgs) -> Result<Outcome<Account>> {
321 let workspace = a.workspace.to_lowercase();
322 if !a.viewer.is_some_and(|viewer| viewer.is_member(&workspace)) {
323 return Ok(members_only());
324 }
325 Ok(Outcome::Ok(self.standing(&workspace).await?))
326 }
327
328 async fn ledger(&self, a: AccountArgs) -> Result<Outcome<Vec<LedgerEntry>>> {
329 let workspace = a.workspace.to_lowercase();
330 if !a.viewer.is_some_and(|viewer| viewer.is_member(&workspace)) {
331 return Ok(members_only());
332 }
333 let rows = self
334 .db
335 .prepare("SELECT * FROM ledger WHERE workspace = ? ORDER BY id DESC LIMIT ?")
336 .bind(&[workspace.into(), LEDGER_PAGE.into()])?
337 .all()
338 .await?
339 .results::<LedgerRow>()?;
340 Ok(Outcome::Ok(
341 rows.into_iter().map(LedgerEntry::from).collect(),
342 ))
343 }
344
Usage, like a hosting provider's: what agents cost, per day, task, repository and pull request345 async fn usage(&self, a: UsageArgs) -> Result<Outcome<Usage>> {
346 let workspace = a.workspace.to_lowercase();
347 if !a.viewer.is_some_and(|viewer| viewer.is_member(&workspace)) {
348 return Ok(members_only());
349 }
350 #[derive(serde::Deserialize)]
351 struct SliceRow {
352 key: Option<String>,
353 micros: Option<i64>,
354 runs: Option<u32>,
355 }
Usage while free is shown at cost; agents get rustfmt and clippy356 // While nothing is charged, what was used is what there is to show.
357 let measure = if self.free { "COALESCE(cost_micros, 0)" } else { "-amount_micros" };
Usage, like a hosting provider's: what agents cost, per day, task, repository and pull request358 let slices = |key: &str, limit: u32| {
359 format!(
Usage while free is shown at cost; agents get rustfmt and clippy360 "SELECT {key} AS key, SUM({measure}) AS micros, COUNT(*) AS runs FROM ledger
Usage, like a hosting provider's: what agents cost, per day, task, repository and pull request361 WHERE workspace = ?1 AND kind = 'usage' AND created_at >= ?2
362 GROUP BY 1 ORDER BY micros DESC LIMIT {limit}"
363 )
364 };
365 let query = |sql: String| {
366 let db = &self.db;
367 let workspace = workspace.clone();
368 let since = a.since.clone();
369 async move {
370 let rows = db
371 .prepare(sql)
372 .bind(&[workspace.into(), since.into()])?
373 .all()
374 .await?
375 .results::<SliceRow>()?;
376 Ok::<Vec<UsageSlice>, worker::Error>(
377 rows.into_iter()
378 .map(|row| UsageSlice {
379 key: row.key.unwrap_or_else(|| "other".to_owned()),
380 micros: row.micros.unwrap_or_default(),
381 runs: row.runs.unwrap_or_default(),
382 })
383 .collect(),
384 )
385 }
386 };
387 #[derive(serde::Deserialize)]
388 struct Totals {
389 spent: Option<i64>,
390 cost: Option<i64>,
Integrations: your own model provider, alerts that open issues, tickets agents read391 provider: Option<i64>,
Usage, like a hosting provider's: what agents cost, per day, task, repository and pull request392 runs: Option<u32>,
393 added: Option<i64>,
394 }
395 let totals = self
396 .db
397 .prepare(
398 "SELECT
399 -SUM(CASE WHEN kind = 'usage' THEN amount_micros END) AS spent,
Integrations: your own model provider, alerts that open issues, tickets agents read400 SUM(CASE WHEN kind = 'usage' AND COALESCE(billed_to, 'g1t') = 'g1t' THEN cost_micros END) AS cost,
401 SUM(CASE WHEN kind = 'usage' AND billed_to = 'workspace' THEN cost_micros END) AS provider,
Usage, like a hosting provider's: what agents cost, per day, task, repository and pull request402 SUM(CASE WHEN kind = 'usage' THEN 1 ELSE 0 END) AS runs,
403 SUM(CASE WHEN kind = 'top_up' THEN amount_micros END) AS added
404 FROM ledger WHERE workspace = ?1 AND created_at >= ?2",
405 )
406 .bind(&[workspace.as_str().into(), a.since.as_str().into()])?
407 .first::<Totals>(None)
408 .await?;
409 let totals = totals.unwrap_or(Totals {
410 spent: None,
411 cost: None,
Integrations: your own model provider, alerts that open issues, tickets agents read412 provider: None,
Usage, like a hosting provider's: what agents cost, per day, task, repository and pull request413 runs: None,
414 added: None,
415 });
416 Ok(Outcome::Ok(Usage {
417 spent_micros: totals.spent.unwrap_or_default(),
418 cost_micros: totals.cost.unwrap_or_default(),
Integrations: your own model provider, alerts that open issues, tickets agents read419 provider_micros: totals.provider.unwrap_or_default(),
Usage while free is shown at cost; agents get rustfmt and clippy420 used_micros: totals.cost.unwrap_or_default() + totals.provider.unwrap_or_default(),
421 free: self.free,
Usage, like a hosting provider's: what agents cost, per day, task, repository and pull request422 runs: totals.runs.unwrap_or_default(),
423 added_micros: totals.added.unwrap_or_default(),
424 by_day: query(slices("substr(created_at, 1, 10) || '/' || COALESCE(task, 'other')", 400)).await?,
425 by_task: query(slices("task", 20)).await?,
426 by_repo: query(slices("repo", 20)).await?,
427 by_pull: query(slices("repo || '#' || number", 10)).await?,
428 by_model: query(slices("model", 10)).await?,
429 since: a.since,
430 }))
431 }
432
Agents as a team: lifecycle, merge queue, billing and a new shell433 async fn checkout(&self, a: CheckoutArgs) -> Result<Outcome<Checkout>> {
434 let workspace = a.workspace.to_lowercase();
435 if a.actor.role_in(&workspace) != Some(Role::Owner) {
436 return Ok(Outcome::fail(
437 FailureCode::Forbidden,
438 "Only an owner can add credit to a workspace.",
439 ));
440 }
441 let Some(stripe) = &self.stripe else {
442 return Ok(Outcome::fail(
443 FailureCode::Conflict,
444 "Payments are not set up on this g1t yet.",
445 ));
446 };
447 if !(MIN_TOP_UP_CENTS..=MAX_TOP_UP_CENTS).contains(&a.amount_cents) {
448 return Ok(Outcome::fail(
449 FailureCode::Invalid,
450 format!(
451 "Add between ${} and ${} at a time.",
452 MIN_TOP_UP_CENTS / 100,
453 MAX_TOP_UP_CENTS / 100
454 ),
455 ));
456 }
457 let customer = self.row(&workspace).await?.and_then(|row| row.customer_id);
Project dependencies: addresses, preview stacks, Affects, and agents who know458 let session = match stripe
459 .start_checkout(&workspace, a.amount_cents, customer.as_deref(), &a.return_url)
460 .await
461 {
462 Ok(session) => session,
463 // A customer saved under another Stripe account: start afresh.
464 Err(error) if customer.is_some() && stripe::is_missing(&error) => {
465 self.forget_customer(&workspace).await?;
466 stripe.start_checkout(&workspace, a.amount_cents, None, &a.return_url).await?
467 }
468 Err(error) => return Err(error),
469 };
Agents as a team: lifecycle, merge queue, billing and a new shell470 let Some(url) = session.url else {
471 return Err(worker::Error::RustError(
472 "the card processor returned no payment page".into(),
473 ));
474 };
475 self.db
476 .prepare(
477 "INSERT INTO checkouts (id, workspace, amount_cents, created_by, created_at)
478 VALUES (?, ?, ?, ?, ?)",
479 )
480 .bind(&[
481 session.id.into(),
482 workspace.into(),
483 a.amount_cents.into(),
484 a.actor.username.into(),
485 rfc3339(now_ms()).into(),
486 ])?
487 .run()
488 .await?;
489 Ok(Outcome::Ok(Checkout { url }))
490 }
491
492 /// Credits a payment if the processor says it was made and it has not
493 /// been credited before. The amount credited is what the processor
494 /// says was paid, not what anyone here remembers asking for.
495 async fn confirm(&self, a: ConfirmArgs) -> Result<Outcome<Account>> {
496 let workspace = a.workspace.to_lowercase();
497 if !a.viewer.is_some_and(|viewer| viewer.is_member(&workspace)) {
498 return Ok(members_only());
499 }
500 let (Some(stripe), Some(checkout)) = (
501 &self.stripe,
502 self.db
503 .prepare(
504 "SELECT workspace, created_by FROM checkouts
505 WHERE id = ? AND workspace = ? AND status = 'open'",
506 )
507 .bind(&[a.session.as_str().into(), workspace.as_str().into()])?
508 .first::<CheckoutRow>(None)
509 .await?,
510 ) else {
511 // Unknown, someone else's, or already credited: nothing to do.
512 return Ok(Outcome::Ok(self.standing(&workspace).await?));
513 };
514 let session = stripe.session(&a.session).await?;
515 let paid = session
516 .amount_total
517 .filter(|_| session.payment_status == "paid");
518 if let Some(cents) = paid {
519 // Only whoever flips it from open to paid enters the credit.
520 let claimed = self
521 .db
522 .prepare(
523 "UPDATE checkouts SET status = 'paid' WHERE id = ? AND status = 'open'
524 RETURNING id",
525 )
526 .bind(&[a.session.as_str().into()])?
527 .first::<Touched>(None)
528 .await?;
529 if claimed.is_some() {
530 self.enter(
531 &checkout.workspace,
532 EntryKind::TopUp,
533 i64::from(cents) * MICROS_PER_DOLLAR / 100,
534 "Credit added by card",
535 &session.id,
536 None,
537 None,
538 Some(&checkout.created_by),
539 session.customer.as_deref(),
540 )
541 .await?;
542 }
543 }
544 Ok(Outcome::Ok(self.standing(&workspace).await?))
545 }
546
Project dependencies: addresses, preview stacks, Affects, and agents who know547 /// Drops a saved customer the card processor no longer knows.
548 pub(crate) async fn forget_customer(&self, workspace: &str) -> Result<()> {
549 self.db
550 .prepare("UPDATE accounts SET customer_id = NULL WHERE workspace = ?")
551 .bind(&[workspace.into()])?
552 .run()
553 .await?;
554 Ok(())
555 }
556
Agents as a team: lifecycle, merge queue, billing and a new shell557 /// A refusal if the workspace has no credit to start an agent with.
558 async fn out_of_credit<T>(&self, workspace: &str) -> Result<Option<Outcome<T>>> {
Billing accounts, terms and enterprises; g1t is no longer free559 // Billing is postpaid: usage limits decide whether work starts
560 // (see `limits`), and credit is a prepayment that lowers what is
561 // owed. A balance no longer has to be positive to start.
562 if self.free || !self.prepaid_only {
Free while g1t is being built out; agents can check out their own forks563 return Ok(None);
564 }
Agents as a team: lifecycle, merge queue, billing and a new shell565 let balance = self
566 .row(workspace)
567 .await?
568 .map_or(0, |row| row.balance_micros);
569 Ok((balance <= 0).then(|| {
570 Outcome::fail(
571 FailureCode::PaymentRequired,
572 format!(
573 "The {workspace} workspace has no agent credit. An owner can add some under Billing on the workspace's page."
574 ),
575 )
576 }))
577 }
578
A free allowance on g1t's models, so anyone can try its agents579 /// A workspace's free allowance on g1t's hosted models: what its runs
580 /// there have cost against its share, and the pool everyone draws on.
Billing accounts, terms and enterprises; g1t is no longer free581 /// How much of a hosted model run's cost the workspace's free allowance
582 /// covers, if it is still open.
583 async fn trial_covers(&self, workspace: &str, cost_micros: i64) -> Result<i64> {
584 let trial = self.trial(TrialArgs { workspace: workspace.to_owned(), exempt: vec![] }).await?;
585 if !trial.open {
586 return Ok(0);
587 }
588 Ok(cost_micros.min((trial.limit_micros - trial.used_micros).max(0)))
589 }
590
A free allowance on g1t's models, so anyone can try its agents591 async fn trial(&self, a: TrialArgs) -> Result<Trial> {
592 let workspace = a.workspace.to_lowercase();
593 let Some(config) = &self.trial else {
594 return Ok(Trial {
595 open: false,
596 used_micros: 0,
597 limit_micros: 0,
598 ends_at: None,
599 reason: Some("off".to_owned()),
600 });
601 };
602 let used = self
603 .db
604 .prepare(
605 "SELECT SUM(cost_micros) AS micros FROM ledger
Billing accounts, terms and enterprises; g1t is no longer free606 WHERE kind = 'usage' AND COALESCE(billed_to, 'g1t') = 'g1t' AND COALESCE(task, '') NOT IN ('sandbox', 'deployments') AND workspace = ?",
A free allowance on g1t's models, so anyone can try its agents607 )
608 .bind(&[workspace.as_str().into()])?
609 .first::<Sum>(None)
610 .await?
611 .and_then(|sum| sum.micros)
612 .unwrap_or_default();
613 // Everyone's, but for the workspaces open to hosted models anyway.
614 let exempt: Vec<String> = a.exempt.iter().map(|name| name.trim().to_lowercase()).collect();
615 let marks = vec!["?"; exempt.len().max(1)].join(", ");
616 let mut values: Vec<JsValue> = exempt.iter().map(|name| JsValue::from(name.as_str())).collect();
617 if values.is_empty() {
618 values.push(JsValue::from(""));
619 }
620 let pooled = self
621 .db
622 .prepare(format!(
623 "SELECT SUM(cost_micros) AS micros FROM ledger
Billing accounts, terms and enterprises; g1t is no longer free624 WHERE kind = 'usage' AND COALESCE(billed_to, 'g1t') = 'g1t' AND COALESCE(task, '') NOT IN ('sandbox', 'deployments') AND workspace NOT IN ({marks})"
A free allowance on g1t's models, so anyone can try its agents625 ))
626 .bind(&values)?
627 .first::<Sum>(None)
628 .await?
629 .and_then(|sum| sum.micros)
630 .unwrap_or_default();
631 let reason = if rfc3339(now_ms()) >= config.until {
632 Some("ended")
633 } else if used >= config.per_workspace_micros {
634 Some("used")
635 } else if pooled >= config.total_micros {
636 Some("pool")
637 } else {
638 None
639 };
640 Ok(Trial {
641 open: reason.is_none(),
642 used_micros: used,
643 limit_micros: config.per_workspace_micros,
644 ends_at: Some(config.until.clone()),
645 reason: reason.map(str::to_owned),
646 })
647 }
648
Agents as a team: lifecycle, merge queue, billing and a new shell649 async fn can_start(&self, a: CanStartArgs) -> Result<Outcome<bool>> {
650 if self.stripe.is_none() {
651 return Ok(Outcome::Ok(true));
652 }
Usage limits: unpaid usage can only go so far653 if let Some(stopped) = self.stopped(&a.workspace).await? {
654 return Ok(stopped);
655 }
Agents as a team: lifecycle, merge queue, billing and a new shell656 Ok(self
657 .out_of_credit(&a.workspace.to_lowercase())
658 .await?
659 .unwrap_or(Outcome::Ok(true)))
660 }
661
662 async fn start_run(&self, a: StartRunArgs) -> Result<Outcome<Option<RunTicket>>> {
663 if self.stripe.is_none() {
664 return Ok(Outcome::Ok(None));
665 }
666 let workspace = a.workspace.to_lowercase();
Usage limits: unpaid usage can only go so far667 if let Some(stopped) = self.stopped(&workspace).await? {
668 return Ok(stopped);
669 }
Agents as a team: lifecycle, merge queue, billing and a new shell670 if let Some(refused) = self.out_of_credit(&workspace).await? {
671 return Ok(refused);
672 }
673 let now = now_ms();
674 let run_id = new_id("run", now);
675 let mut bytes = [0u8; 32];
676 getrandom::getrandom(&mut bytes).expect("no source of randomness");
677 let token = hex::encode(bytes);
678 self.db
679 .prepare(
Prices keep themselves current with what g1t pays680 "INSERT INTO runs (id, workspace, repo, number, task, model, token_hash, created_at, billed_to, session_id)
681 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
Agents as a team: lifecycle, merge queue, billing and a new shell682 )
683 .bind(&[
684 run_id.as_str().into(),
685 workspace.into(),
686 format!("{}/{}", a.repo.namespace, a.repo.name).into(),
687 a.number.into(),
688 a.task.into(),
689 a.model.into(),
690 hash(&token).into(),
691 rfc3339(now).into(),
Integrations: your own model provider, alerts that open issues, tickets agents read692 if a.billed_to == "workspace" { "workspace" } else { "g1t" }.into(),
Prices keep themselves current with what g1t pays693 optional(a.session.as_deref().filter(|_| a.billed_to != "workspace")),
Agents as a team: lifecycle, merge queue, billing and a new shell694 ])?
695 .run()
696 .await?;
697 Ok(Outcome::Ok(Some(RunTicket { run_id, token })))
698 }
699
700 async fn finish_run(&self, a: FinishRunArgs) -> Result<Outcome<bool>> {
701 let run = self
702 .db
703 .prepare(
Integrations: your own model provider, alerts that open issues, tickets agents read704 "SELECT workspace, repo, number, task, model, token_hash, billed_to FROM runs
Agents as a team: lifecycle, merge queue, billing and a new shell705 WHERE id = ? AND finished_at IS NULL",
706 )
707 .bind(&[a.run_id.as_str().into()])?
708 .first::<RunRow>(None)
709 .await?;
710 let Some(run) = run.filter(|run| run.token_hash == hash(&a.token)) else {
711 return Ok(Outcome::fail(FailureCode::NotFound, "Run not found."));
712 };
713 if !a.cost_usd.is_finite() || a.cost_usd < 0.0 {
714 return Ok(Outcome::fail(FailureCode::Invalid, "That is not a cost."));
715 }
716 // Only whoever closes the run charges for it.
717 let claimed = self
718 .db
719 .prepare(
720 "UPDATE runs SET finished_at = ? WHERE id = ? AND finished_at IS NULL RETURNING id",
721 )
722 .bind(&[rfc3339(now_ms()).into(), a.run_id.as_str().into()])?
723 .first::<Touched>(None)
724 .await?;
725 if claimed.is_none() {
726 return Ok(Outcome::Ok(false));
727 }
Integrations: your own model provider, alerts that open issues, tickets agents read728 // On the workspace's own provider, the model was paid for there:
729 // g1t charges its fee, and keeps the provider's cost to show.
Billing accounts, terms and enterprises; g1t is no longer free730 let base = if run.own_provider() {
Integrations: your own model provider, alerts that open issues, tickets agents read731 self.orchestration_fee_micros
732 } else {
Billing accounts, terms and enterprises; g1t is no longer free733 // The free allowance on g1t's models covers what it can.
734 let cost = charge_micros(a.cost_usd, 0);
735 let covered = self.trial_covers(&run.workspace, cost).await?;
736 charge_micros((cost - covered) as f64 / MICROS_PER_DOLLAR as f64, self.margin_percent)
Integrations: your own model provider, alerts that open issues, tickets agents read737 };
Billing accounts, terms and enterprises; g1t is no longer free738 let (charge, terms_note) = self.charged(&run.workspace, base).await?;
Integrations: your own model provider, alerts that open issues, tickets agents read739 let mut description = match run.task.as_str() {
Agents as a team: lifecycle, merge queue, billing and a new shell740 "plan" => format!("Planning for {}", run.repo),
741 "review" => format!("Review of {}#{}", run.repo, run.number),
742 "update" => format!("Catching up {}#{}", run.repo, run.number),
743 _ => format!("Work on {}#{}", run.repo, run.number),
744 };
Integrations: your own model provider, alerts that open issues, tickets agents read745 if run.own_provider() {
746 description.push_str(", on your own model provider");
747 }
Billing accounts, terms and enterprises; g1t is no longer free748 description.push_str(&terms_note);
Agents as a team: lifecycle, merge queue, billing and a new shell749 self.enter(
750 &run.workspace,
751 EntryKind::Usage,
752 -charge,
753 &description,
754 &a.run_id,
755 Some(&run),
756 Some(charge_micros(a.cost_usd, 0)),
757 None,
758 None,
759 )
760 .await?;
761 Ok(Outcome::Ok(true))
762 }
763}
764
Every sandbox is metered by the second765impl Billing {
766 /// Records how long a sandbox ran: its cost always, and a charge for
767 /// the seconds past the month's free minutes.
768 async fn record_sandbox(&self, a: RecordSandboxArgs) -> Result<Outcome<bool>> {
769 if self.stripe.is_none() || a.seconds == 0 {
770 return Ok(Outcome::Ok(false));
771 }
772 let workspace = a.workspace.to_lowercase();
773 let seen = self
774 .db
775 .prepare("SELECT id FROM ledger WHERE reference = ?")
776 .bind(&[a.reference.as_str().into()])?
777 .first::<Touched>(None)
778 .await?;
779 if seen.is_some() {
780 return Ok(Outcome::Ok(false));
781 }
782 let now = now_ms();
783 let timestamp = rfc3339(now);
784 let month = &timestamp[..7];
785 #[derive(Deserialize)]
786 struct Used {
787 seconds: i64,
788 }
789 let seconds = i64::from(a.seconds);
790 let after = self
791 .db
792 .prepare(
793 "INSERT INTO sandbox_months (workspace, month, seconds) VALUES (?1, ?2, ?3)
794 ON CONFLICT (workspace, month) DO UPDATE SET seconds = seconds + ?3
795 RETURNING seconds",
796 )
797 .bind(&[workspace.as_str().into(), month.into(), (seconds as f64).into()])?
798 .first::<Used>(None)
799 .await?
800 .map_or(seconds, |used| used.seconds);
801 let billable = sandbox_billable(after - seconds, seconds);
Prices keep themselves current with what g1t pays802 // From the price book, which follows what Cloudflare bills g1t.
803 let (cost_per_second, price_per_second) = self.price("sandbox_second").await?.unwrap_or((
804 sandbox_allowance::COST_MICROS_PER_SECOND as f64,
805 sandbox_allowance::MICROS_PER_SECOND as f64,
806 ));
Billing accounts, terms and enterprises; g1t is no longer free807 let (charge, terms_note) = self.charged(&workspace, (billable as f64 * price_per_second).ceil() as i64).await?;
Every sandbox is metered by the second808 let mut description = format!("{}: {} of sandbox time", a.description, duration(seconds));
809 if billable < seconds {
810 description.push_str(if billable == 0 {
811 ", within the month's free minutes"
812 } else {
813 ", partly within the month's free minutes"
814 });
815 }
Billing accounts, terms and enterprises; g1t is no longer free816 if billable > 0 {
817 description.push_str(&terms_note);
Every sandbox is metered by the second818 }
819 self.db
820 .batch(vec![
821 self.db
822 .prepare(
823 "INSERT INTO ledger
824 (id, workspace, kind, amount_micros, description, repo, task,
825 cost_micros, reference, created_at, billed_to)
826 VALUES (?, ?, 'usage', ?, ?, ?, 'sandbox', ?, ?, ?, 'g1t')",
827 )
828 .bind(&[
829 new_id("led", now).into(),
830 workspace.as_str().into(),
831 (-(charge as f64)).into(),
832 description.as_str().into(),
833 optional(a.repo.as_deref()),
Prices keep themselves current with what g1t pays834 (seconds as f64 * cost_per_second).ceil().into(),
Every sandbox is metered by the second835 a.reference.as_str().into(),
836 timestamp.as_str().into(),
837 ])?,
838 self.db
839 .prepare(
840 "INSERT INTO accounts (workspace, balance_micros, created_at)
841 VALUES (?1, ?2, ?3)
842 ON CONFLICT (workspace) DO UPDATE SET balance_micros = balance_micros + ?2",
843 )
844 .bind(&[
845 workspace.as_str().into(),
846 (-(charge as f64)).into(),
847 timestamp.as_str().into(),
848 ])?,
849 ])
850 .await?;
851 Ok(Outcome::Ok(true))
852 }
853}
854
855/// Of `seconds` used after `before` this month, how many are past the
856/// free minutes.
857fn sandbox_billable(before: i64, seconds: i64) -> i64 {
858 let free_left = (sandbox_allowance::FREE_SECONDS - before).max(0);
859 (seconds - free_left).max(0)
860}
861
862/// `1h 2m`, `3m 12s` or `40s`.
863fn duration(seconds: i64) -> String {
864 let (h, m, s) = (seconds / 3600, seconds % 3600 / 60, seconds % 60);
865 if h > 0 {
866 format!("{h}h {m}m")
867 } else if m > 0 {
868 format!("{m}m {s}s")
869 } else {
870 format!("{s}s")
871 }
872}
873
Billing accounts, terms and enterprises; g1t is no longer free874impl Billing {
875 /// What a workspace is charged for something that would be `base`:
876 /// nothing while g1t is free, or as its account's terms say. With a
877 /// note for the statement when it differs.
878 pub(crate) async fn charged(&self, workspace: &str, base: i64) -> Result<(i64, String)> {
879 if self.free {
880 return Ok((0, " (free while g1t is being built out)".to_owned()));
881 }
882 let terms = self.terms_of(workspace).await?;
883 let charge = terms.apply(base);
884 let note = match terms.kind {
885 TermsKind::Comped => " (comped)".to_owned(),
886 TermsKind::Custom if terms.discount_percent > 0 && base > 0 => format!(" ({}% off)", terms.discount_percent),
887 _ => String::new(),
888 };
889 Ok((charge, note))
890 }
891}
892
Agents as a team: lifecycle, merge queue, billing and a new shell893fn members_only<T>() -> Outcome<T> {
894 Outcome::fail(
895 FailureCode::Forbidden,
896 "Only members can see a workspace's billing.",
897 )
898}
899
Prices keep themselves current with what g1t pays900impl Billing {
901 fn from_env(env: &Env) -> Result<Self> {
902 Ok(Billing {
903 db: env.d1("DB")?,
904 stripe: env
905 .secret("STRIPE_SECRET_KEY")
906 .ok()
907 .map(|key| key.to_string())
908 .filter(|key| !key.is_empty())
909 .map(Stripe::new),
910 margin_percent: env
911 .var("MARGIN_PERCENT")
912 .ok()
913 .and_then(|percent| percent.to_string().parse().ok())
914 .unwrap_or(20),
915 orchestration_fee_micros: env
916 .var("ORCHESTRATION_FEE_MICROS")
917 .ok()
918 .and_then(|fee| fee.to_string().parse().ok())
919 .unwrap_or(100_000),
920 free: env.var("FREE_WHILE_BUILDING").is_ok_and(|v| v.to_string() == "true"),
921 ceilings: limits::Ceilings::from_env(&env),
Billing accounts, terms and enterprises; g1t is no longer free922 prepaid_only: env.var("PREPAID_ONLY").is_ok_and(|v| v.to_string() == "true"),
Prices keep themselves current with what g1t pays923 deployments_monthly_cents: env
924 .var("DEPLOYMENTS_MONTHLY_CENTS")
925 .ok()
926 .and_then(|cents| cents.to_string().parse().ok())
927 .unwrap_or(500),
928 trial: {
929 let number = |name: &str| env.var(name).ok().and_then(|v| v.to_string().parse::<i64>().ok());
930 match (
931 number("TRIAL_WORKSPACE_MICROS"),
932 number("TRIAL_TOTAL_MICROS"),
933 env.var("TRIAL_UNTIL").ok().map(|v| v.to_string()),
934 ) {
935 (Some(per_workspace_micros), Some(total_micros), Some(until))
936 if per_workspace_micros > 0 && !until.is_empty() =>
937 {
938 Some(TrialConfig {
939 per_workspace_micros,
940 total_micros,
941 until,
942 })
943 }
944 _ => None,
945 }
946 },
947 })
948 }
949}
950
951#[event(scheduled)]
952async fn scheduled(event: ScheduledEvent, env: Env, _ctx: ScheduleContext) {
953 let Ok(billing) = Billing::from_env(&env) else {
954 return;
955 };
956 let keeper = keeper::Keeper::from_env(&env);
957 if let Err(error) = billing.settle_runs(&keeper).await {
958 worker::console_error!("settling runs failed: {error}");
959 }
Billing accounts, terms and enterprises; g1t is no longer free960 if let Err(error) = billing.autopay().await {
961 worker::console_error!("paying at the limit failed: {error}");
962 }
Billing on Stripe's pages, month-end charges, warnings; sudo by workspace963 if let Err(error) = billing.close_months().await {
964 worker::console_error!("closing the month failed: {error}");
965 }
966 if let Ok(identity) = env.service("IDENTITY") {
967 if let Err(error) = billing.warn_limits(&identity).await {
968 worker::console_error!("warning owners failed: {error}");
969 }
970 }
The keeper reads Cloudflare as it really answers971 // Once a day, and at once if the costs were never checked: check every
972 // cost against what Cloudflare billed.
973 if event.cron() == keeper::DAILY || billing.never_checked().await.unwrap_or(false) {
Prices keep themselves current with what g1t pays974 if let Err(error) = billing.reconcile(&keeper).await {
975 worker::console_error!("checking costs against Cloudflare failed: {error}");
976 }
977 }
978}
979
Agents as a team: lifecycle, merge queue, billing and a new shell980#[event(fetch)]
981async fn fetch(mut request: Request, env: Env, _ctx: Context) -> Result<Response> {
982 let Some(method) = rpc_method(&request) else {
983 return Response::error("Not found", 404);
984 };
985 let body: serde_json::Value = request.json().await?;
Prices keep themselves current with what g1t pays986 let billing = Billing::from_env(&env)?;
Agents as a team: lifecycle, merge queue, billing and a new shell987 match method.as_str() {
988 "status" => reply(&billing.status()),
989 "account" => reply(&billing.account(args(body)?).await?),
990 "ledger" => reply(&billing.ledger(args(body)?).await?),
Usage, like a hosting provider's: what agents cost, per day, task, repository and pull request991 "usage" => reply(&billing.usage(args(body)?).await?),
Agents as a team: lifecycle, merge queue, billing and a new shell992 "checkout" => reply(&billing.checkout(args(body)?).await?),
993 "confirm" => reply(&billing.confirm(args(body)?).await?),
994 "can_start" => reply(&billing.can_start(args(body)?).await?),
A free allowance on g1t's models, so anyone can try its agents995 "trial" => reply(&billing.trial(args(body)?).await?),
Agents as a team: lifecycle, merge queue, billing and a new shell996 "start_run" => reply(&billing.start_run(args(body)?).await?),
997 "finish_run" => reply(&billing.finish_run(args(body)?).await?),
Paid features: a workspace turns on Deployments with a monthly plan998 "features" => reply(&billing.features(args(body)?).await?),
999 "subscribe" => reply(&billing.subscribe(args(body)?).await?),
1000 "confirm_subscription" => reply(&billing.confirm_subscription(args(body)?).await?),
1001 "cancel_subscription" => reply(&billing.cancel_subscription(args(body)?).await?),
1002 "has_feature" => reply(&billing.has_feature(args(body)?).await?),
1003 "charge_feature" => reply(&billing.charge_feature(args(body)?).await?),
Every sandbox is metered by the second1004 "record_sandbox" => reply(&billing.record_sandbox(args(body)?).await?),
Usage limits: unpaid usage can only go so far1005 "limit" => reply(&billing.limit(args(body)?).await?),
1006 "check_limit" => reply(&billing.check_limit(args(body)?).await?),
1007 "set_spend_limit" => reply(&billing.set_spend_limit(args(body)?).await?),
Prices keep themselves current with what g1t pays1008 "prices" => reply(&billing.prices().await?),
Billing on Stripe's pages, month-end charges, warnings; sudo by workspace1009 "billing_portal" => reply(&billing.billing_portal(args(body)?).await?),
1010 "admin_billing_link" => reply(&billing.admin_billing_link(args(body)?).await?),
Prices keep themselves current with what g1t pays1011 "note_pending" => reply(&billing.note_pending(args(body)?).await?),
Billing accounts, terms and enterprises; g1t is no longer free1012 "admin_accounts" => reply(&billing.admin_accounts(args(body)?).await?),
1013 "admin_account" => reply(&billing.admin_account(args(body)?).await?),
1014 "admin_set_terms" => reply(&billing.admin_set_terms(args(body)?).await?),
1015 "admin_create_enterprise" => reply(&billing.admin_create_enterprise(args(body)?).await?),
1016 "admin_attach" => reply(&billing.admin_attach(args(body)?).await?),
1017 "admin_credit" => reply(&billing.admin_credit(args(body)?).await?),
Agents as a team: lifecycle, merge queue, billing and a new shell1018 _ => Response::error("Unknown method", 404),
1019 }
1020}
1021
1022#[cfg(test)]
1023mod tests {
1024 use super::*;
1025
1026 #[test]
1027 fn a_run_is_charged_its_cost_plus_the_margin() {
1028 // $0.05 at 20% is six cents.
1029 assert_eq!(charge_micros(0.05, 20), 60_000);
1030 assert_eq!(charge_micros(1.0, 20), 1_200_000);
1031 assert_eq!(charge_micros(0.05, 0), 50_000);
1032 }
1033
1034 #[test]
1035 fn fractions_of_a_millionth_round_up_and_nothing_costs_less_than_nothing() {
1036 assert_eq!(charge_micros(0.000_000_4, 20), 2);
1037 assert_eq!(charge_micros(0.0, 20), 0);
1038 assert_eq!(charge_micros(-3.0, 20), 0);
1039 }
1040
1041 #[test]
Every sandbox is metered by the second1042 fn sandbox_seconds_are_charged_only_past_the_free_minutes() {
1043 let free = sandbox_allowance::FREE_SECONDS;
1044 assert_eq!(sandbox_billable(0, 600), 0);
1045 assert_eq!(sandbox_billable(free - 100, 600), 500);
1046 assert_eq!(sandbox_billable(free + 5, 600), 600);
1047 }
1048
1049 #[test]
1050 fn durations_read_plainly() {
1051 assert_eq!(duration(40), "40s");
1052 assert_eq!(duration(192), "3m 12s");
1053 assert_eq!(duration(3720), "1h 2m");
1054 }
1055
1056 #[test]
Agents as a team: lifecycle, merge queue, billing and a new shell1057 fn an_absurd_cost_is_capped() {
1058 assert_eq!(charge_micros(1e9, 20), 120 * MICROS_PER_DOLLAR);
1059 }
1060}