amadan
usage pipe: sidecar token accounting → instance relay → cloud ledger (#867)
* server+bot+cloud: the usage pipe — sidecar token accounting relayed into the cloud ledger (#857) Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * server+bot+cloud: review fixes — serialize instance-env pushes, idempotent usage reports Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> ---------
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
20 files changed,
+1298
−40
cloud.go+11 −0cloud_db.go+123 −8cloud_usage.go+97 −0cloud_usage_test.go+144 −0controld.go+52 −0docs/cloud-provisioner.md+25 −0internal/bot/anthropic.go+49 −2internal/bot/brain.go+1 −1internal/bot/chartersched.go+1 −0internal/bot/connect_v2_test.go+1 −1internal/bot/meter.go+192 −0internal/bot/meter_test.go+190 −0internal/bot/research_test.go+3 −3internal/bot/runtime.go+23 −9internal/bot/task_test.go+1 −1main.go+9 −1meter.go+61 −14meter_test.go+38 −0usage.go+151 −0usage_test.go+126 −0
diff --git a/cloud.go b/cloud.go| index 047e712..8ca6b1f 100644 |
| --- a/cloud.go |
| +++ b/cloud.go |
| @@ -87,6 +87,10 @@ type cloudServer struct { |
| // (cloudreplication.go); empty = classes never pushed, every space streams. |
| replicationClasses map[string]string |
| createBudget int // max creates fleet-wide per trailing 24h (0 = uncapped) |
| + // usageURL is this service's own public /api/usage URL, pushed into each |
| + // instance's env (LCHAT_METER_URL) alongside a minted meter token so |
| + // sidecar usage reports can flow (#857). Empty = the usage pipe is off. |
| + usageURL string |
| } |
| func (s *cloudServer) handler() http.Handler { |
| @@ -100,6 +104,9 @@ func (s *cloudServer) handler() http.Handler { |
| mux.HandleFunc("POST /api/instances/{host}/checkout", s.callerAuth(s.checkoutHandler)) |
| mux.HandleFunc("POST /api/instances/{host}/portal", s.callerAuth(s.portalHandler)) |
| mux.HandleFunc("POST /api/stripe/webhook", s.stripeWebhookHandler) |
| + // Usage ingest (#857): meter-token-gated, not caller-gated — an instance |
| + // can only ever report as itself. |
| + mux.HandleFunc("POST /api/usage", s.usageHandler) |
| // Admin surface: mint / list / revoke caller tokens at runtime (mirrors the |
| // push relay's admin API). |
| mux.HandleFunc("POST /api/admin/tokens", s.admin(s.mintTokenHandler)) |
| @@ -475,6 +482,10 @@ func runCloud(args []string) { |
| storageCaps: storageCaps, |
| replicationClasses: replicationClasses, |
| createBudget: *createBudget, |
| + usageURL: strings.TrimSpace(os.Getenv("LCHAT_CLOUD_USAGE_URL")), |
| + } |
| + if s.usageURL != "" { |
| + log.Printf("cloud: usage pipe on — meter tokens push %s", s.usageURL) |
| } |
| if *createBudget > 0 { |
| log.Printf("cloud: creation budget on (%d/day fleet-wide)", *createBudget) |
diff --git a/cloud_db.go b/cloud_db.go| index c8b04a4..bed2a39 100644 |
| --- a/cloud_db.go |
| +++ b/cloud_db.go |
| @@ -73,7 +73,121 @@ CREATE TABLE IF NOT EXISTS creates_log ( |
| if err := db.migrateBilling(); err != nil { |
| return err |
| } |
| - return db.migrateMeter() |
| + if err := db.migrateMeter(); err != nil { |
| + return err |
| + } |
| + return db.migrateUsage() |
| +} |
| + |
| +// migrateUsage (#857): the per-space usage ledger — quantities only (token |
| +// counts, request counts, run counts), never content — plus the meter token |
| +// each instance authenticates its reports with. The token itself is pushed |
| +// into the instance env (LCHAT_METER_URL/_TOKEN) by the sweep; only its hash |
| +// lives here, same posture as caller tokens. |
| +func (db *cloudDB) migrateUsage() error { |
| + if has, err := db.spacesHasColumn("meter_token_hash"); err != nil { |
| + return err |
| + } else if !has { |
| + if _, err := db.sql.Exec(`ALTER TABLE spaces ADD COLUMN meter_token_hash TEXT NOT NULL DEFAULT ''`); err != nil { |
| + return err |
| + } |
| + } |
| + _, err := db.sql.Exec(`CREATE TABLE IF NOT EXISTS usage ( |
| + space_id INTEGER NOT NULL, |
| + day TEXT NOT NULL, |
| + model TEXT NOT NULL DEFAULT '', |
| + dimension TEXT NOT NULL, |
| + quantity INTEGER NOT NULL DEFAULT 0, |
| + PRIMARY KEY (space_id, day, model, dimension) |
| +); |
| +-- Applied report ids — the dedupe ledger that makes ingest idempotent. A |
| +-- sidecar resends a report verbatim when an ack is lost; the id tells a |
| +-- replay from new usage. Pruned after a retention window (the sidecar |
| +-- retries on a minutes scale, so days of memory is generous). |
| +CREATE TABLE IF NOT EXISTS usage_reports ( |
| + space_id INTEGER NOT NULL, |
| + report_id TEXT NOT NULL, |
| + applied_at INTEGER NOT NULL, |
| + PRIMARY KEY (space_id, report_id) |
| +)`) |
| + return err |
| +} |
| + |
| +// setMeterTokenHash records the hash of the meter token last pushed to a |
| +// space's instance. Overwriting is the rotation path: last push wins. |
| +func (db *cloudDB) setMeterTokenHash(id int64, hash string) error { |
| + _, err := db.sql.Exec(`UPDATE spaces SET meter_token_hash = ? WHERE id = ?`, hash, id) |
| + return err |
| +} |
| + |
| +// spaceByMeterToken resolves a usage report's bearer token to its space. An |
| +// empty stored hash never matches (hashToken of any presented token is |
| +// non-empty), so unmetered spaces can't be reported against. |
| +func (db *cloudDB) spaceByMeterToken(token string) (int64, bool) { |
| + var id int64 |
| + err := db.sql.QueryRow( |
| + `SELECT id FROM spaces WHERE meter_token_hash = ? AND meter_token_hash != ''`, |
| + hashToken(token)).Scan(&id) |
| + return id, err == nil |
| +} |
| + |
| +// addUsage folds one report into the ledger — additive upserts, so repeated |
| +// partial-day reports accumulate instead of clobbering. Idempotent on |
| +// (space, report id): the report row and the quantity upserts share one |
| +// transaction, so a replay of an already-applied report (ack lost after |
| +// commit) is recognized and dropped whole. |
| +func (db *cloudDB) addUsage(spaceID int64, reportID string, rows []botUsageRow) error { |
| + tx, err := db.sql.Begin() |
| + if err != nil { |
| + return err |
| + } |
| + defer tx.Rollback() |
| + res, err := tx.Exec(`INSERT INTO usage_reports(space_id, report_id, applied_at) |
| +VALUES(?, ?, ?) ON CONFLICT(space_id, report_id) DO NOTHING`, |
| + spaceID, reportID, time.Now().Unix()) |
| + if err != nil { |
| + return err |
| + } |
| + if n, err := res.RowsAffected(); err != nil { |
| + return err |
| + } else if n == 0 { |
| + return nil // replay of an applied report — already counted |
| + } |
| + for _, r := range rows { |
| + if _, err := tx.Exec(`INSERT INTO usage(space_id, day, model, dimension, quantity) |
| +VALUES(?, ?, ?, ?, ?) |
| +ON CONFLICT(space_id, day, model, dimension) DO UPDATE SET quantity = quantity + excluded.quantity`, |
| + spaceID, r.Day, r.Model, r.Dimension, r.Quantity); err != nil { |
| + return err |
| + } |
| + } |
| + return tx.Commit() |
| +} |
| + |
| +// pruneUsageReports drops applied-report ids older than the cutoff. The |
| +// sidecar's retry horizon is minutes, so days of dedupe memory is generous. |
| +func (db *cloudDB) pruneUsageReports(cutoff int64) error { |
| + _, err := db.sql.Exec(`DELETE FROM usage_reports WHERE applied_at < ?`, cutoff) |
| + return err |
| +} |
| + |
| +// usageFor reads a space's ledger rows (tests + the future usage surface). |
| +func (db *cloudDB) usageFor(spaceID int64) ([]botUsageRow, error) { |
| + rows, err := db.sql.Query( |
| + `SELECT day, model, dimension, quantity FROM usage WHERE space_id = ? ORDER BY day, model, dimension`, spaceID) |
| + if err != nil { |
| + return nil, err |
| + } |
| + defer rows.Close() |
| + var out []botUsageRow |
| + for rows.Next() { |
| + var r botUsageRow |
| + if err := rows.Scan(&r.Day, &r.Model, &r.Dimension, &r.Quantity); err != nil { |
| + return nil, err |
| + } |
| + out = append(out, r) |
| + } |
| + return out, rows.Err() |
| } |
| // migrateSpacesID rebuilds a spaces table created under the original host-PK |
| @@ -522,17 +636,18 @@ func (db *cloudDB) setEntitlementTier(id int64, ent, tier string, paidUntil int6 |
| // meterRow is one space in the meter's sweep set — enough billing state to |
| // reconcile its quota alongside the byte reading. |
| type meterRow struct { |
| - ID int64 |
| - Host string |
| - Entitlement string |
| - Tier string |
| - QuotaBytes int64 |
| + ID int64 |
| + Host string |
| + Entitlement string |
| + Tier string |
| + QuotaBytes int64 |
| + MeterTokenHash string |
| } |
| // allFinalizedSpaces is the meter's sweep set: every space with a hostname. |
| func (db *cloudDB) allFinalizedSpaces() ([]meterRow, error) { |
| rows, err := db.sql.Query( |
| - `SELECT id, host, entitlement, tier, quota_bytes FROM spaces WHERE host IS NOT NULL ORDER BY id`) |
| + `SELECT id, host, entitlement, tier, quota_bytes, meter_token_hash FROM spaces WHERE host IS NOT NULL ORDER BY id`) |
| if err != nil { |
| return nil, err |
| } |
| @@ -540,7 +655,7 @@ func (db *cloudDB) allFinalizedSpaces() ([]meterRow, error) { |
| var out []meterRow |
| for rows.Next() { |
| var m meterRow |
| - if err := rows.Scan(&m.ID, &m.Host, &m.Entitlement, &m.Tier, &m.QuotaBytes); err != nil { |
| + if err := rows.Scan(&m.ID, &m.Host, &m.Entitlement, &m.Tier, &m.QuotaBytes, &m.MeterTokenHash); err != nil { |
| return nil, err |
| } |
| out = append(out, m) |
diff --git a/cloud_usage.go b/cloud_usage.go| new file mode 100644 |
| index 0000000..07d1d3d |
| --- /dev/null |
| +++ b/cloud_usage.go |
| @@ -0,0 +1,97 @@ |
| +// cloud_usage.go — the cloud half of the usage pipe (#857, |
| +// docs/ai-provisioning.md phase 1): ingest per-space usage reports into the |
| +// ledger, and mint + push the meter tokens that authenticate them. |
| +// |
| +// The pipe measures before any money exists: rows are quantities only (token |
| +// counts, request counts, run counts) keyed by (space, day, model, dimension) |
| +// — never content, never per-member attribution (#668's rule). Phase 3 turns |
| +// the ledger into an allowance/budget answer; until then every report is |
| +// answered "ok". |
| +package main |
| + |
| +import ( |
| + "bytes" |
| + "encoding/json" |
| + "fmt" |
| + "log" |
| + "net/http" |
| +) |
| + |
| +// usageHandler ingests one instance's report. Auth is the space's meter |
| +// token — minted by the sweep below, pushed into the instance env, and never |
| +// shown anywhere else — so an instance can only ever report as itself. |
| +func (s *cloudServer) usageHandler(w http.ResponseWriter, r *http.Request) { |
| + spaceID, ok := s.db.spaceByMeterToken(bearerTokenHeader(r)) |
| + if !ok { |
| + writeJSON(w, http.StatusForbidden, map[string]any{"error": "bad meter token"}) |
| + return |
| + } |
| + var body struct { |
| + ReportID string `json:"report_id"` |
| + Rows []botUsageRow `json:"rows"` |
| + } |
| + if json.NewDecoder(http.MaxBytesReader(w, r.Body, 64<<10)).Decode(&body) != nil { |
| + writeJSON(w, http.StatusBadRequest, map[string]any{"error": "bad request"}) |
| + return |
| + } |
| + // Same validator as the instance relay, so a report the relay accepted is |
| + // by construction one the ledger accepts. |
| + if err := validateUsageReport(body.ReportID, body.Rows); err != nil { |
| + writeJSON(w, http.StatusBadRequest, map[string]any{"error": err.Error()}) |
| + return |
| + } |
| + // addUsage is idempotent on (space, report id): a replay whose first ack |
| + // was lost lands here again and is dropped, never double-counted. |
| + if err := s.db.addUsage(spaceID, body.ReportID, body.Rows); err != nil { |
| + log.Printf("cloud: usage ingest space %d: %v", spaceID, err) |
| + writeJSON(w, http.StatusInternalServerError, map[string]any{"error": "ledger error"}) |
| + return |
| + } |
| + // Budget is phase 3 (docs/ai-provisioning.md): allowance and pool math |
| + // land here; until then the meter measures and everything is within |
| + // budget by definition. |
| + writeJSON(w, http.StatusOK, map[string]any{"ok": true, "budget": map[string]any{"state": "ok"}}) |
| +} |
| + |
| +// ---- meter-token mint + push (sweep-driven) -------------------------------------- |
| + |
| +// pushMeterToken mints a fresh meter token for a space and hands it to the |
| +// instance via controld (LCHAT_METER_URL/_TOKEN in the env). Push first, |
| +// record the hash after: if recording fails the next sweep mints and pushes |
| +// again, and the last push wins — the same self-healing shape as pushQuota. |
| +func (s *cloudServer) pushMeterToken(id int64, host string) { |
| + token, err := randToken() |
| + if err != nil { |
| + log.Printf("cloud: meter token %s: %v", host, err) |
| + return |
| + } |
| + if err := s.controlSetMeter(host, s.usageURL, token); err != nil { |
| + log.Printf("cloud: meter token push %s: %v (sweep will retry)", host, err) |
| + return |
| + } |
| + if err := s.db.setMeterTokenHash(id, hashToken(token)); err != nil { |
| + log.Printf("cloud: meter token record %s: %v (sweep re-mints)", host, err) |
| + return |
| + } |
| + log.Printf("cloud: meter token → %s", host) |
| +} |
| + |
| +// controlSetMeter pushes one instance's meter endpoint + token to controld. |
| +func (s *cloudServer) controlSetMeter(host, url, token string) error { |
| + payload, _ := json.Marshal(map[string]any{"url": url, "token": token}) |
| + req, err := http.NewRequest("POST", "http://control/instances/"+host+"/meter", bytes.NewReader(payload)) |
| + if err != nil { |
| + return err |
| + } |
| + req.Header.Set("Authorization", "Bearer "+s.controlToken) |
| + req.Header.Set("Content-Type", "application/json") |
| + resp, err := s.control.Do(req) |
| + if err != nil { |
| + return err |
| + } |
| + defer resp.Body.Close() |
| + if resp.StatusCode != http.StatusOK { |
| + return fmt.Errorf("control meter %s: HTTP %d", host, resp.StatusCode) |
| + } |
| + return nil |
| +} |
diff --git a/cloud_usage_test.go b/cloud_usage_test.go| new file mode 100644 |
| index 0000000..e879583 |
| --- /dev/null |
| +++ b/cloud_usage_test.go |
| @@ -0,0 +1,144 @@ |
| +package main |
| + |
| +import ( |
| + "bytes" |
| + "encoding/json" |
| + "net/http" |
| + "net/http/httptest" |
| + "testing" |
| + "time" |
| +) |
| + |
| +func ingestUsage(t *testing.T, s *cloudServer, token string, rows []botUsageRow) (int, map[string]any) { |
| + t.Helper() |
| + return ingestUsageReport(t, s, token, "bbbbbbbbbbbbbbbb", rows) |
| +} |
| + |
| +func ingestUsageReport(t *testing.T, s *cloudServer, token, reportID string, rows []botUsageRow) (int, map[string]any) { |
| + t.Helper() |
| + raw, _ := json.Marshal(map[string]any{"report_id": reportID, "rows": rows}) |
| + req := httptest.NewRequest("POST", "/api/usage", bytes.NewReader(raw)) |
| + if token != "" { |
| + req.Header.Set("Authorization", "Bearer "+token) |
| + } |
| + rec := httptest.NewRecorder() |
| + s.usageHandler(rec, req) |
| + var body map[string]any |
| + json.NewDecoder(rec.Body).Decode(&body) |
| + return rec.Code, body |
| +} |
| + |
| +// Ingest: the meter token resolves to exactly one space, rows upsert |
| +// additively (partial-day reports accumulate), and a stranger is refused. |
| +func TestUsageIngest(t *testing.T) { |
| + db := openTestCloudDB(t) |
| + id := insertSpaceRow(t, db, "space.11ch.at", 1000) |
| + if err := db.setMeterTokenHash(id, hashToken("meter-tok")); err != nil { |
| + t.Fatal(err) |
| + } |
| + s := &cloudServer{db: db} |
| + |
| + rows := []botUsageRow{ |
| + {Day: "2026-07-14", Model: "claude-sonnet-5", Dimension: "llm_input_tokens", Quantity: 100}, |
| + {Day: "2026-07-14", Dimension: "charter_runs", Quantity: 2}, |
| + } |
| + if code, body := ingestUsageReport(t, s, "meter-tok", "aa11aa11aa11aa11", rows); code != http.StatusOK { |
| + t.Fatalf("ingest: %d %v", code, body) |
| + } |
| + // A second partial-day report (new id) adds, never clobbers. |
| + if code, _ := ingestUsageReport(t, s, "meter-tok", "aa22aa22aa22aa22", rows[:1]); code != http.StatusOK { |
| + t.Fatal("second ingest failed") |
| + } |
| + // A REPLAY (same id — the ack-lost retry) is recognized and dropped whole. |
| + if code, _ := ingestUsageReport(t, s, "meter-tok", "aa11aa11aa11aa11", rows); code != http.StatusOK { |
| + t.Fatal("replay must still answer ok") |
| + } |
| + got, err := db.usageFor(id) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + byKey := map[string]int64{} |
| + for _, r := range got { |
| + byKey[r.Model+"|"+r.Dimension] = r.Quantity |
| + } |
| + if byKey["claude-sonnet-5|llm_input_tokens"] != 200 || byKey["|charter_runs"] != 2 { |
| + t.Fatalf("ledger = %v (replay must not double-count)", byKey) |
| + } |
| + // Prune drops old dedupe ids; a post-prune replay would re-count, which |
| + // is why the window is days against a minutes-scale retry horizon. |
| + if err := db.pruneUsageReports(time.Now().Unix() + 1); err != nil { |
| + t.Fatal(err) |
| + } |
| + |
| + if code, _ := ingestUsage(t, s, "wrong", rows); code != http.StatusForbidden { |
| + t.Fatal("bad token must be refused") |
| + } |
| + if code, _ := ingestUsage(t, s, "", rows); code != http.StatusForbidden { |
| + t.Fatal("missing token must be refused") |
| + } |
| + // An unmetered space's empty hash must never match any presented token. |
| + other := insertSpaceRow(t, db, "other.11ch.at", 1000) |
| + if sid, ok := db.spaceByMeterToken(""); ok { |
| + t.Fatalf("empty token resolved to space %d", sid) |
| + } |
| + _ = other |
| + if code, _ := ingestUsage(t, s, "meter-tok", []botUsageRow{{Day: "bad", Dimension: "llm_requests", Quantity: 1}}); code != http.StatusBadRequest { |
| + t.Fatal("invalid rows must be refused at the ledger too") |
| + } |
| +} |
| + |
| +// The sweep mints a meter token for any space without one, pushes it through |
| +// controld, and records only the hash. A space that already has one is left |
| +// alone — tokens rotate only when a push failed to record. |
| +func TestMeterSweepMintsTokens(t *testing.T) { |
| + db := openTestCloudDB(t) |
| + id := insertSpaceRow(t, db, "fresh.11ch.at", 1000) |
| + |
| + var pushes int |
| + var gotURL, gotToken string |
| + mux := http.NewServeMux() |
| + mux.HandleFunc("GET /instances/{host}/stats", func(w http.ResponseWriter, r *http.Request) { |
| + writeJSON(w, http.StatusOK, map[string]any{"storage_bytes": 1, "setup_complete": true, "members": 1}) |
| + }) |
| + mux.HandleFunc("POST /instances/{host}/meter", func(w http.ResponseWriter, r *http.Request) { |
| + var body struct{ URL, Token string } |
| + json.NewDecoder(r.Body).Decode(&body) |
| + pushes++ |
| + gotURL, gotToken = body.URL, body.Token |
| + writeJSON(w, http.StatusOK, map[string]any{"ok": true}) |
| + }) |
| + sock := fakeControl(t, mux.ServeHTTP) |
| + s := &cloudServer{db: db, control: controlClient(sock), controlToken: "t", |
| + usageURL: "https://cloud.example/api/usage"} |
| + |
| + s.meterOnce() |
| + if pushes != 1 || gotURL != "https://cloud.example/api/usage" || gotToken == "" { |
| + t.Fatalf("push: n=%d url=%q token-empty=%v", pushes, gotURL, gotToken == "") |
| + } |
| + // The pushed token authenticates; only its hash is stored. |
| + if sid, ok := db.spaceByMeterToken(gotToken); !ok || sid != id { |
| + t.Fatalf("pushed token does not resolve: ok=%v sid=%d", ok, sid) |
| + } |
| + // Converged: the next sweep must not re-mint. |
| + s.meterOnce() |
| + if pushes != 1 { |
| + t.Fatalf("second sweep re-minted (pushes=%d)", pushes) |
| + } |
| +} |
| + |
| +// With the usage pipe off (no LCHAT_CLOUD_USAGE_URL) the sweep never touches |
| +// meter tokens — self-hosted cloud deployments see no behavior change. |
| +func TestMeterSweepNoUsageURL(t *testing.T) { |
| + db := openTestCloudDB(t) |
| + insertSpaceRow(t, db, "fresh.11ch.at", 1000) |
| + mux := http.NewServeMux() |
| + mux.HandleFunc("GET /instances/{host}/stats", func(w http.ResponseWriter, r *http.Request) { |
| + writeJSON(w, http.StatusOK, map[string]any{"storage_bytes": 1}) |
| + }) |
| + mux.HandleFunc("POST /instances/{host}/meter", func(w http.ResponseWriter, r *http.Request) { |
| + t.Error("meter push must not happen with the pipe off") |
| + }) |
| + sock := fakeControl(t, mux.ServeHTTP) |
| + s := &cloudServer{db: db, control: controlClient(sock), controlToken: "t"} |
| + s.meterOnce() |
| +} |
diff --git a/controld.go b/controld.go| index 3fe772f..d5c0f67 100644 |
| --- a/controld.go |
| +++ b/controld.go |
| @@ -106,6 +106,7 @@ func (c *controld) handler() http.Handler { |
| mux.HandleFunc("DELETE /instances/{host}", c.auth(c.deleteHandler)) |
| mux.HandleFunc("GET /instances/{host}/stats", c.auth(c.statsHandler)) |
| mux.HandleFunc("POST /instances/{host}/quota", c.auth(c.quotaHandler)) |
| + mux.HandleFunc("POST /instances/{host}/meter", c.auth(c.meterHandler)) |
| mux.HandleFunc("POST /instances/{host}/replication", c.auth(c.replicationHandler)) |
| mux.HandleFunc("POST /instances/{host}/materialize", c.auth(c.materializeHandler)) |
| return mux |
| @@ -272,6 +273,57 @@ func (c *controld) quotaHandler(w http.ResponseWriter, r *http.Request) { |
| writeJSON(w, http.StatusOK, map[string]any{"ok": true, "restarted": restarted}) |
| } |
| +// meterHandler hands one instance its usage-pipe credentials (#857): upsert |
| +// LCHAT_METER_URL + LCHAT_METER_TOKEN in the env and bounce the unit so they |
| +// take effect — the same shape as quotaHandler, minus the registry-quota |
| +// bookkeeping. Both empty clears the pair (metering off). |
| +func (c *controld) meterHandler(w http.ResponseWriter, r *http.Request) { |
| + host := strings.ToLower(strings.TrimSpace(r.PathValue("host"))) |
| + if !validHost(host) { |
| + writeControlErr(w, http.StatusBadRequest, "invalid_host", "not a valid hostname") |
| + return |
| + } |
| + var req struct { |
| + URL string `json:"url"` |
| + Token string `json:"token"` |
| + } |
| + if err := json.NewDecoder(io.LimitReader(r.Body, 4<<10)).Decode(&req); err != nil || |
| + (req.URL == "") != (req.Token == "") { |
| + writeControlErr(w, http.StatusBadRequest, "bad_request", "url and token required together (both empty clears)") |
| + return |
| + } |
| + envPath := filepath.Join(instanceDataDir, host+".env") |
| + if err := setInstanceMeter(envPath, req.URL, req.Token); err != nil { |
| + if os.IsNotExist(err) { |
| + writeControlErr(w, http.StatusNotFound, "unknown_host", "no such instance") |
| + return |
| + } |
| + log.Printf("controld: meter %s: %v", host, err) |
| + writeControlErr(w, http.StatusInternalServerError, "meter_failed", err.Error()) |
| + return |
| + } |
| + // Root just rewrote the env; the router (service user) reads it at wake. |
| + if err := matchOwner(envPath, instanceDataDir); err != nil { |
| + log.Printf("controld: meter %s: env ownership: %v", host, err) |
| + } |
| + restarted := false |
| + hibernating := false |
| + if c.reg != nil { |
| + if rt, err := c.reg.get(host); err == nil && rt != nil && rt.Hibernate { |
| + hibernating = true // picks the env up on its next wake |
| + } |
| + } |
| + if !hibernating { |
| + if err := systemctl("try-restart", "lchat@"+host); err == nil { |
| + restarted = true |
| + } else { |
| + log.Printf("controld: meter %s: restart: %v (applies on next start)", host, err) |
| + } |
| + } |
| + log.Printf("controld: meter %s → %v (restarted=%v)", host, req.URL != "", restarted) |
| + writeJSON(w, http.StatusOK, map[string]any{"ok": true, "restarted": restarted}) |
| +} |
| + |
| // statsHandler reports the reaper's guards for a space — setup_complete, member |
| // count, and has_content (any message or thread). Read-only against the live |
| // instance DB. validHost guards the DB path against traversal before it's joined |
diff --git a/docs/cloud-provisioner.md b/docs/cloud-provisioner.md| index 8d67468..c4905e2 100644 |
| --- a/docs/cloud-provisioner.md |
| +++ b/docs/cloud-provisioner.md |
| @@ -359,6 +359,31 @@ it. Unset = no enforcement, and operator-run instances (not in this ledger) |
| are never touched. Caps are sized to the box today (see |
| `tier-economics.md`); they rise when attachments move off-box (#550). |
| +## The usage pipe (#857, #668 slices 2–3) — AI usage into the ledger |
| + |
| +With `LCHAT_CLOUD_USAGE_URL` set (the cloud's own public `/api/usage` URL), |
| +the same meter sweep mints each ledger space a **meter token** and pushes it |
| +into the instance env (`LCHAT_METER_URL` + `LCHAT_METER_TOKEN`, via controld |
| +`POST /instances/{host}/meter`, hash-only in the ledger, same self-healing |
| +best-effort shape as the quota push). The flow is sidecar → instance → cloud: |
| + |
| +- The **sidecar** counts model spend — it's the only place spend is visible — |
| + as (day, model, dimension) quantities: `llm_requests`, the four token |
| + counters (input / output / cache-write / cache-read, priced differently), |
| + and `charter_runs`. It flushes every few minutes to its instance |
| + (`POST /api/bot-usage`, bot-token-gated), keeping its egress surface at the |
| + two destinations `docs/bots.md` promises. |
| +- The **instance** is a stateless relay: validate, forward with the meter |
| + token, hand the response's budget state back to the sidecar. Unconfigured |
| + (self-hosted, BYO key) it accepts and drops — answer `"unmetered"`, nothing |
| + recorded anywhere. |
| +- The **cloud** upserts into `usage(space_id, day, model, dimension, |
| + quantity)` — quantities only, never content, never per-member attribution |
| + (#668's rule) — and answers a budget state, `"ok"` unconditionally until |
| + the phase-3 allowance/budget math lands (docs/ai-provisioning.md). |
| + |
| +Unset `LCHAT_CLOUD_USAGE_URL` = the pipe is off fleet-wide: no tokens minted, |
| +no env pushes, instances answer sidecars `"unmetered"`. |
| **Replication classes (the durability ladder).** With |
| `LCHAT_CLOUD_REPLICATION_CLASSES` set |
| (`free=daily,family=daily,community=batched,business=stream` — same key |
diff --git a/internal/bot/anthropic.go b/internal/bot/anthropic.go| index d5902d4..9945dfe 100644 |
| --- a/internal/bot/anthropic.go |
| +++ b/internal/bot/anthropic.go |
| @@ -25,16 +25,21 @@ type anthropicClient struct { |
| key string |
| base string |
| http *http.Client |
| + // record, when set, receives every successful call's usage block — the |
| + // usage meter's tap (#857). It sits here, not at the call sites, so every |
| + // model path (reply, classifier, task, research, app build) is counted |
| + // without each one remembering to. |
| + record func(model string, u anthUsage) |
| } |
| -func newAnthropicClient(key string) *anthropicClient { |
| +func newAnthropicClient(key string, record func(model string, u anthUsage)) *anthropicClient { |
| base := os.Getenv("ANTHROPIC_BASE_URL") |
| if base == "" { |
| base = "https://api.anthropic.com" |
| } |
| // Long timeout: a thinking run over a large cached system prompt can |
| // legitimately take a couple of minutes. |
| - return &anthropicClient{key: key, base: base, http: &http.Client{Timeout: 5 * time.Minute}} |
| + return &anthropicClient{key: key, base: base, http: &http.Client{Timeout: 5 * time.Minute}, record: record} |
| } |
| type anthBlock struct { |
| @@ -109,6 +114,18 @@ type anthRequest struct { |
| type anthResponse struct { |
| Content []anthBlock `json:"content"` |
| StopReason string `json:"stop_reason"` |
| + Usage anthUsage `json:"usage"` |
| +} |
| + |
| +// anthUsage is the token accounting block every Messages response carries. |
| +// All four counters are kept distinct — cache reads and writes price |
| +// differently from fresh input, so the usage meter (#857) records each as its |
| +// own dimension. |
| +type anthUsage struct { |
| + InputTokens int64 `json:"input_tokens"` |
| + OutputTokens int64 `json:"output_tokens"` |
| + CacheCreationInputTokens int64 `json:"cache_creation_input_tokens"` |
| + CacheReadInputTokens int64 `json:"cache_read_input_tokens"` |
| } |
| // anthHTTPError is a non-200 response — carried as an error so create's retry |
| @@ -158,6 +175,9 @@ func (c *anthropicClient) do(req anthRequest, stream bool) (*anthResponse, error |
| resp, err = c.doRequest(body) |
| } |
| if err == nil { |
| + if c.record != nil { |
| + c.record(req.Model, resp.Usage) |
| + } |
| return resp, nil |
| } |
| lastErr = err |
| @@ -256,6 +276,13 @@ func (c *anthropicClient) doStream(body []byte) (*anthResponse, error) { |
| PartialJSON string `json:"partial_json"` |
| StopReason string `json:"stop_reason"` |
| } `json:"delta"` |
| + // Usage rides message_start (input + cache counters, inside the |
| + // message envelope) and message_delta (cumulative output tokens, |
| + // as a sibling of delta) — both are folded into out.Usage. |
| + Message *struct { |
| + Usage anthUsage `json:"usage"` |
| + } `json:"message"` |
| + Usage *anthUsage `json:"usage"` |
| Error *struct { |
| Type string `json:"type"` |
| Message string `json:"message"` |
| @@ -265,6 +292,10 @@ func (c *anthropicClient) doStream(body []byte) (*anthResponse, error) { |
| continue |
| } |
| switch ev.Type { |
| + case "message_start": |
| + if ev.Message != nil { |
| + out.Usage = ev.Message.Usage |
| + } |
| case "content_block_start": |
| for len(blocks) <= ev.Index { |
| blocks = append(blocks, nil) |
| @@ -289,6 +320,22 @@ func (c *anthropicClient) doStream(body []byte) (*anthResponse, error) { |
| if ev.Delta != nil && ev.Delta.StopReason != "" { |
| out.StopReason = ev.Delta.StopReason |
| } |
| + // message_delta usage counters are cumulative totals — take the |
| + // latest non-zero values rather than summing. |
| + if ev.Usage != nil { |
| + if ev.Usage.OutputTokens > 0 { |
| + out.Usage.OutputTokens = ev.Usage.OutputTokens |
| + } |
| + if ev.Usage.InputTokens > 0 { |
| + out.Usage.InputTokens = ev.Usage.InputTokens |
| + } |
| + if ev.Usage.CacheCreationInputTokens > 0 { |
| + out.Usage.CacheCreationInputTokens = ev.Usage.CacheCreationInputTokens |
| + } |
| + if ev.Usage.CacheReadInputTokens > 0 { |
| + out.Usage.CacheReadInputTokens = ev.Usage.CacheReadInputTokens |
| + } |
| + } |
| case "error": |
| if ev.Error != nil { |
| return nil, fmt.Errorf("anthropic: stream error: %s: %s", ev.Error.Type, ev.Error.Message) |
diff --git a/internal/bot/brain.go b/internal/bot/brain.go| index 6294dc5..84478ff 100644 |
| --- a/internal/bot/brain.go |
| +++ b/internal/bot/brain.go |
| @@ -87,7 +87,7 @@ func (r *Runtime) loadConfig() { |
| // silent, and declareKind (which runs right after this) re-badges honestly. |
| if r.AIKey != "" && (c.AIEnabled == nil || *c.AIEnabled) { |
| if r.ai == nil { |
| - r.ai = newAnthropicClient(r.AIKey) |
| + r.ai = newAnthropicClient(r.AIKey, r.recordUsage) |
| } |
| } else { |
| r.ai = nil |
diff --git a/internal/bot/chartersched.go b/internal/bot/chartersched.go| index 8d17870..9b023db 100644 |
| --- a/internal/bot/chartersched.go |
| +++ b/internal/bot/chartersched.go |
| @@ -260,6 +260,7 @@ func (r *Runtime) charterTick(selfID string, selfKey []byte, tp *tickPayload) { |
| default: |
| r.clearCharterFails(tp.Bot) |
| r.postTaskResult(c.Thread, key, c.Title, &c.Task, res.text, res.card) |
| + r.meterInc("charter_runs", 1) // completed runs only — failures cost tokens but deliver nothing |
| } |
| } |
diff --git a/internal/bot/connect_v2_test.go b/internal/bot/connect_v2_test.go| index b32b89f..e15522f 100644 |
| --- a/internal/bot/connect_v2_test.go |
| +++ b/internal/bot/connect_v2_test.go |
| @@ -487,7 +487,7 @@ func TestConnectChipFailureGetsSpokenExplanation(t *testing.T) { |
| t.Setenv("ANTHROPIC_BASE_URL", claude.URL) |
| h := newConnectHarness(t) |
| - h.rt.ai = newAnthropicClient("test-key") |
| + h.rt.ai = newAnthropicClient("test-key", nil) |
| h.rt.model = defaultModel |
| grant := h.seal(t, map[string]any{"service": "github", "token": "ghp_x", "exp": 0, "access": "read"}) |
diff --git a/internal/bot/meter.go b/internal/bot/meter.go| new file mode 100644 |
| index 0000000..a6610e7 |
| --- /dev/null |
| +++ b/internal/bot/meter.go |
| @@ -0,0 +1,192 @@ |
| +// The usage meter (#857, docs/ai-provisioning.md phase 1). The sidecar is the |
| +// only place model spend is visible, so it counts here and reports upward: |
| +// token counters per (day, model) plus activity counters, flushed to the |
| +// *instance* (POST /api/bot-usage, bot-token-gated like charter-next-due) |
| +// rather than straight to the cloud. Via-instance keeps the sidecar's egress |
| +// surface at its two destinations (the instance + the model API — docs/ |
| +// bots.md) and keeps the meter token in exactly one env file, the instance's; |
| +// the instance relays to the cloud ledger and relays the budget state back. |
| +// |
| +// Counters are quantities only — token counts, request counts, run counts — |
| +// never content, per the meter-only-what-ciphertext-exposes rule (#668). They |
| +// live in memory: a sidecar restart loses at most one flush interval, which |
| +// is acceptable for metering (billing-grade reconciliation is cloud-side). |
| +package bot |
| + |
| +import ( |
| + "context" |
| + "crypto/rand" |
| + "encoding/hex" |
| + "sync" |
| + "time" |
| +) |
| + |
| +// UsageRow is one meter cell: a (day, model, dimension) quantity. Model is |
| +// empty for activity dimensions (charter_runs). Day is UTC — the ledger's |
| +// grain, deliberately coarse. |
| +type UsageRow struct { |
| + Day string `json:"day"` |
| + Model string `json:"model,omitempty"` |
| + Dimension string `json:"dimension"` |
| + Quantity int64 `json:"quantity"` |
| +} |
| + |
| +type meterKey struct{ day, model, dim string } |
| + |
| +type usageMeter struct { |
| + mu sync.Mutex |
| + rows map[meterKey]int64 |
| +} |
| + |
| +func newUsageMeter() *usageMeter { return &usageMeter{rows: map[meterKey]int64{}} } |
| + |
| +func meterDay() string { return time.Now().UTC().Format("2006-01-02") } |
| + |
| +// addLLM records one Messages call's usage block, all four token counters |
| +// kept distinct (cache reads/writes price differently from fresh input), plus |
| +// the request count. |
| +func (m *usageMeter) addLLM(model string, u anthUsage) { |
| + m.mu.Lock() |
| + defer m.mu.Unlock() |
| + day := meterDay() |
| + for dim, n := range map[string]int64{ |
| + "llm_requests": 1, |
| + "llm_input_tokens": u.InputTokens, |
| + "llm_output_tokens": u.OutputTokens, |
| + "llm_cache_write_tokens": u.CacheCreationInputTokens, |
| + "llm_cache_read_tokens": u.CacheReadInputTokens, |
| + } { |
| + if n > 0 { |
| + m.rows[meterKey{day, model, dim}] += n |
| + } |
| + } |
| +} |
| + |
| +// inc records a model-less activity dimension (e.g. charter_runs). |
| +func (m *usageMeter) inc(dim string, n int64) { |
| + if n <= 0 { |
| + return |
| + } |
| + m.mu.Lock() |
| + defer m.mu.Unlock() |
| + m.rows[meterKey{meterDay(), "", dim}] += n |
| +} |
| + |
| +// drain snapshots and clears the counters; the caller restores the snapshot |
| +// if the flush fails, so a transient error loses nothing. |
| +func (m *usageMeter) drain() []UsageRow { |
| + m.mu.Lock() |
| + defer m.mu.Unlock() |
| + if len(m.rows) == 0 { |
| + return nil |
| + } |
| + out := make([]UsageRow, 0, len(m.rows)) |
| + for k, q := range m.rows { |
| + out = append(out, UsageRow{Day: k.day, Model: k.model, Dimension: k.dim, Quantity: q}) |
| + } |
| + m.rows = map[meterKey]int64{} |
| + return out |
| +} |
| + |
| +// restore folds a failed flush's rows back into the live counters. |
| +func (m *usageMeter) restore(rows []UsageRow) { |
| + m.mu.Lock() |
| + defer m.mu.Unlock() |
| + for _, r := range rows { |
| + m.rows[meterKey{r.Day, r.Model, r.Dimension}] += r.Quantity |
| + } |
| +} |
| + |
| +// recordUsage is the model client's accounting tap (its record hook). |
| +func (r *Runtime) recordUsage(model string, u anthUsage) { |
| + if r.meter != nil { |
| + r.meter.addLLM(model, u) |
| + } |
| +} |
| + |
| +// meterInc bumps an activity dimension from anywhere in the runtime. |
| +func (r *Runtime) meterInc(dim string, n int64) { |
| + if r.meter != nil { |
| + r.meter.inc(dim, n) |
| + } |
| +} |
| + |
| +// meterFlushEvery paces reporting. Coarse on purpose: the ledger's grain is a |
| +// day, so nothing downstream needs freshness — this only bounds how much a |
| +// crash can lose. |
| +const meterFlushEvery = 5 * time.Minute |
| + |
| +// usageReport is one flush in flight: drained rows plus the id the ledger |
| +// dedupes replays on. A failed flush keeps the report VERBATIM and resends |
| +// it — never re-drains into it — because "failed" includes "the cloud |
| +// committed but the ack got lost", and only a byte-identical replay under |
| +// the same id lets the ledger recognize and drop the duplicate. |
| +type usageReport struct { |
| + ID string `json:"report_id"` |
| + Rows []UsageRow `json:"rows"` |
| +} |
| + |
| +func newReportID() string { |
| + raw := make([]byte, 16) |
| + if _, err := rand.Read(raw); err != nil { |
| + // Out of entropy is unrecoverable weirdness; a constant would break |
| + // dedupe silently, so fail the report instead (retried next tick). |
| + return "" |
| + } |
| + return hex.EncodeToString(raw) |
| +} |
| + |
| +func (r *Runtime) meterLoop(ctx context.Context) { |
| + t := time.NewTicker(meterFlushEvery) |
| + defer t.Stop() |
| + for { |
| + select { |
| + case <-t.C: |
| + r.flushUsage() |
| + case <-ctx.Done(): |
| + // Best-effort final flush — the graceful paths (sleep co-spawn |
| + // SIGTERM, deploy restart) shouldn't drop their last interval. |
| + r.flushUsage() |
| + return |
| + } |
| + } |
| +} |
| + |
| +// flushUsage reports counters to the instance and remembers the budget state |
| +// the response carries (inert today; phase 3 gates the model path on it). |
| +// Runs only on the meterLoop goroutine, so pendingUsage needs no lock: a |
| +// pending report is retried as-is before anything new is drained, and fresh |
| +// counts keep accumulating in the meter for the report after it. |
| +func (r *Runtime) flushUsage() { |
| + if r.meter == nil { |
| + return |
| + } |
| + if r.pendingUsage == nil { |
| + rows := r.meter.drain() |
| + if len(rows) == 0 { |
| + return |
| + } |
| + id := newReportID() |
| + if id == "" { |
| + r.meter.restore(rows) |
| + return |
| + } |
| + r.pendingUsage = &usageReport{ID: id, Rows: rows} |
| + } |
| + var resp struct { |
| + OK bool `json:"ok"` |
| + Budget struct { |
| + State string `json:"state"` |
| + } `json:"budget"` |
| + } |
| + if err := r.post("/api/bot-usage", map[string]any{ |
| + "token": r.Session.Token, "report_id": r.pendingUsage.ID, "rows": r.pendingUsage.Rows, |
| + }, &resp); err != nil { |
| + r.logf("usage: flush failed (%v) — report retried next tick", err) |
| + return |
| + } |
| + r.pendingUsage = nil |
| + if resp.Budget.State != "" { |
| + r.budgetState.Store(resp.Budget.State) |
| + } |
| +} |
diff --git a/internal/bot/meter_test.go b/internal/bot/meter_test.go| new file mode 100644 |
| index 0000000..515e312 |
| --- /dev/null |
| +++ b/internal/bot/meter_test.go |
| @@ -0,0 +1,190 @@ |
| +package bot |
| + |
| +import ( |
| + "encoding/json" |
| + "net/http" |
| + "net/http/httptest" |
| + "testing" |
| +) |
| + |
| +func rowsByKey(rows []UsageRow) map[string]int64 { |
| + out := map[string]int64{} |
| + for _, r := range rows { |
| + out[r.Day+"|"+r.Model+"|"+r.Dimension] = r.Quantity |
| + } |
| + return out |
| +} |
| + |
| +// The accumulator's contract: calls merge into (day, model, dimension) |
| +// cells, drain empties, and restore folds a failed flush back in losslessly. |
| +func TestUsageMeterAccumulateDrainRestore(t *testing.T) { |
| + m := newUsageMeter() |
| + m.addLLM("claude-sonnet-5", anthUsage{InputTokens: 100, OutputTokens: 20, CacheReadInputTokens: 300}) |
| + m.addLLM("claude-sonnet-5", anthUsage{InputTokens: 50, OutputTokens: 5, CacheCreationInputTokens: 7}) |
| + m.addLLM("claude-haiku-4-5", anthUsage{InputTokens: 9}) |
| + m.inc("charter_runs", 1) |
| + |
| + day := meterDay() |
| + rows := m.drain() |
| + got := rowsByKey(rows) |
| + for key, want := range map[string]int64{ |
| + day + "|claude-sonnet-5|llm_requests": 2, |
| + day + "|claude-sonnet-5|llm_input_tokens": 150, |
| + day + "|claude-sonnet-5|llm_output_tokens": 25, |
| + day + "|claude-sonnet-5|llm_cache_read_tokens": 300, |
| + day + "|claude-sonnet-5|llm_cache_write_tokens": 7, |
| + day + "|claude-haiku-4-5|llm_requests": 1, |
| + day + "|claude-haiku-4-5|llm_input_tokens": 9, |
| + day + "||charter_runs": 1, |
| + } { |
| + if got[key] != want { |
| + t.Errorf("%s = %d, want %d", key, got[key], want) |
| + } |
| + } |
| + if m.drain() != nil { |
| + t.Fatal("second drain should be empty") |
| + } |
| + |
| + // A failed flush restores; new counts merge on top. |
| + m.restore(rows) |
| + m.addLLM("claude-sonnet-5", anthUsage{InputTokens: 1}) |
| + again := rowsByKey(m.drain()) |
| + if again[day+"|claude-sonnet-5|llm_input_tokens"] != 151 { |
| + t.Fatalf("restore+add did not merge: %v", again) |
| + } |
| +} |
| + |
| +// A zero-usage block (a failed parse, an empty response) records only the |
| +// request count — never zero-quantity rows. |
| +func TestUsageMeterSkipsZeroes(t *testing.T) { |
| + m := newUsageMeter() |
| + m.addLLM("m", anthUsage{}) |
| + rows := m.drain() |
| + if len(rows) != 1 || rows[0].Dimension != "llm_requests" || rows[0].Quantity != 1 { |
| + t.Fatalf("want just llm_requests=1, got %+v", rows) |
| + } |
| +} |
| + |
| +// The non-streamed path: the response's usage block reaches both the returned |
| +// response and the record hook (the meter's tap). |
| +func TestAnthropicUsageParsed(t *testing.T) { |
| + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| + json.NewEncoder(w).Encode(map[string]any{ |
| + "content": []map[string]any{{"type": "text", "text": "hi"}}, |
| + "stop_reason": "end_turn", |
| + "usage": map[string]any{ |
| + "input_tokens": 120, "output_tokens": 34, |
| + "cache_creation_input_tokens": 8, "cache_read_input_tokens": 4000, |
| + }, |
| + }) |
| + })) |
| + defer srv.Close() |
| + |
| + var gotModel string |
| + var got anthUsage |
| + c := &anthropicClient{key: "k", base: srv.URL, http: srv.Client(), |
| + record: func(model string, u anthUsage) { gotModel, got = model, u }} |
| + resp, err := c.create(anthRequest{Model: "claude-sonnet-5", MaxTokens: 10}) |
| + if err != nil { |
| + t.Fatalf("create: %v", err) |
| + } |
| + want := anthUsage{InputTokens: 120, OutputTokens: 34, CacheCreationInputTokens: 8, CacheReadInputTokens: 4000} |
| + if resp.Usage != want { |
| + t.Fatalf("response usage = %+v, want %+v", resp.Usage, want) |
| + } |
| + if gotModel != "claude-sonnet-5" || got != want { |
| + t.Fatalf("record hook got (%q, %+v)", gotModel, got) |
| + } |
| +} |
| + |
| +// The streamed path: input + cache counters ride message_start, the final |
| +// (cumulative) output count rides message_delta. |
| +func TestAnthropicUsageParsedStream(t *testing.T) { |
| + stream := "event: message_start\n" + |
| + `data: {"type":"message_start","message":{"usage":{"input_tokens":200,"output_tokens":1,"cache_read_input_tokens":900}}}` + "\n\n" + |
| + "event: content_block_start\n" + |
| + `data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}` + "\n\n" + |
| + "event: content_block_delta\n" + |
| + `data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"ok"}}` + "\n\n" + |
| + "event: message_delta\n" + |
| + `data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":57}}` + "\n\n" + |
| + "event: message_stop\n" + |
| + `data: {"type":"message_stop"}` + "\n\n" |
| + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| + w.Header().Set("Content-Type", "text/event-stream") |
| + w.Write([]byte(stream)) |
| + })) |
| + defer srv.Close() |
| + |
| + var got anthUsage |
| + c := &anthropicClient{key: "k", base: srv.URL, http: srv.Client(), |
| + record: func(_ string, u anthUsage) { got = u }} |
| + resp, err := c.createStream(anthRequest{Model: "m", MaxTokens: 10}) |
| + if err != nil { |
| + t.Fatalf("createStream: %v", err) |
| + } |
| + want := anthUsage{InputTokens: 200, OutputTokens: 57, CacheReadInputTokens: 900} |
| + if resp.Usage != want { |
| + t.Fatalf("stream usage = %+v, want %+v", resp.Usage, want) |
| + } |
| + if got != want { |
| + t.Fatalf("record hook got %+v", got) |
| + } |
| +} |
| + |
| +// The outbox contract (the ack-loss fix): a failed flush keeps the report |
| +// VERBATIM — same id, same rows — so the ledger can dedupe the replay, and |
| +// usage counted during the outage rides the NEXT report, never the pending one. |
| +func TestFlushUsageOutboxRetriesVerbatim(t *testing.T) { |
| + type report struct { |
| + ReportID string `json:"report_id"` |
| + Rows []UsageRow `json:"rows"` |
| + } |
| + var got []report |
| + fail := true |
| + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| + var rep report |
| + json.NewDecoder(r.Body).Decode(&rep) |
| + got = append(got, rep) |
| + if fail { |
| + http.Error(w, "cloud blip", http.StatusBadGateway) |
| + return |
| + } |
| + json.NewEncoder(w).Encode(map[string]any{"ok": true, "budget": map[string]any{"state": "ok"}}) |
| + })) |
| + defer srv.Close() |
| + |
| + r := &Runtime{Base: srv.URL, Session: &Session{Token: "tok", Name: "Eleven"}, Logf: t.Logf} |
| + r.client = srv.Client() |
| + r.meter = newUsageMeter() |
| + |
| + r.meter.inc("charter_runs", 1) |
| + r.flushUsage() // fails; report stays pending |
| + r.meter.inc("charter_runs", 5) |
| + r.flushUsage() // still failing; the SAME report is resent, new counts stay in the meter |
| + fail = false |
| + r.flushUsage() // succeeds: pending report lands |
| + r.flushUsage() // then the accumulated counts land as a NEW report |
| + |
| + if len(got) != 4 { |
| + t.Fatalf("attempts = %d, want 4", len(got)) |
| + } |
| + if got[0].ReportID == "" || got[0].ReportID != got[1].ReportID || got[1].ReportID != got[2].ReportID { |
| + t.Fatalf("retries must reuse the report id: %q %q %q", got[0].ReportID, got[1].ReportID, got[2].ReportID) |
| + } |
| + if got[3].ReportID == got[2].ReportID { |
| + t.Fatal("post-success flush must mint a fresh report id") |
| + } |
| + if len(got[2].Rows) != 1 || got[2].Rows[0].Quantity != 1 { |
| + t.Fatalf("pending report mutated across retries: %+v", got[2].Rows) |
| + } |
| + if len(got[3].Rows) != 1 || got[3].Rows[0].Quantity != 5 { |
| + t.Fatalf("outage-time counts must ride the next report: %+v", got[3].Rows) |
| + } |
| + if r.pendingUsage != nil { |
| + t.Fatal("nothing should be pending after success") |
| + } |
| + if r.budgetState.Load() != "ok" { |
| + t.Fatalf("budget state not stored: %v", r.budgetState.Load()) |
| + } |
| +} |
diff --git a/internal/bot/research_test.go b/internal/bot/research_test.go| index 0b52d44..86acbc2 100644 |
| --- a/internal/bot/research_test.go |
| +++ b/internal/bot/research_test.go |
| @@ -83,7 +83,7 @@ func TestResearchEndToEnd(t *testing.T) { |
| `{"content":[{"type":"text","text":"You have one repo: paul/lchat."}],"stop_reason":"end_turn"}`, |
| ) |
| h := newConnectHarness(t) |
| - h.rt.ai = newAnthropicClient("test-key") |
| + h.rt.ai = newAnthropicClient("test-key", nil) |
| grant := h.seal(t, map[string]any{"service": "github", "token": "ghp_research", "exp": time.Now().Add(time.Minute).UnixMilli(), "access": "read"}) |
| h.rt.handleResearch("t1", h.key, &researchPayload{ |
| @@ -138,7 +138,7 @@ func TestResearchEndToEnd(t *testing.T) { |
| func TestResearchForeignGrantStaysSilent(t *testing.T) { |
| newFakeGitHub(t, `[]`) |
| h := newConnectHarness(t) |
| - h.rt.ai = newAnthropicClient("test-key") |
| + h.rt.ai = newAnthropicClient("test-key", nil) |
| other, err := e2ee.GenerateUserKeypair() |
| if err != nil { |
| t.Fatal(err) |
| @@ -160,7 +160,7 @@ func TestResearchExpiredGrantNeverReachesProvider(t *testing.T) { |
| `{"content":[{"type":"text","text":"That look expired."}],"stop_reason":"end_turn"}`, |
| ) |
| h := newConnectHarness(t) |
| - h.rt.ai = newAnthropicClient("test-key") |
| + h.rt.ai = newAnthropicClient("test-key", nil) |
| grant := h.seal(t, map[string]any{"service": "github", "token": "ghp_x", |
| "exp": time.Now().Add(-time.Minute).UnixMilli(), "access": "read"}) |
| h.rt.handleResearch("t1", h.key, &researchPayload{ |
diff --git a/internal/bot/runtime.go b/internal/bot/runtime.go| index 2621b51..7030cfc 100644 |
| --- a/internal/bot/runtime.go |
| +++ b/internal/bot/runtime.go |
| @@ -10,6 +10,7 @@ import ( |
| "regexp" |
| "strings" |
| "sync" |
| + "sync/atomic" |
| "time" |
| "github.com/coder/websocket" |
| @@ -51,14 +52,23 @@ type Runtime struct { |
| // charter would run once per host (botcmd elects it). |
| Charters bool |
| - client *http.Client |
| - ai *anthropicClient |
| - connect *connectConfig // connected-services (docs/connected-services.md) |
| - keys map[string]threadEntry |
| - cursors map[string]int64 // threadId -> last message id we've processed |
| - agents map[string]bool // roster labels of ALL agents — never reply to one |
| - mention *regexp.Regexp |
| - bare *regexp.Regexp // bare-name mention, for listen mode "name" |
| + client *http.Client |
| + ai *anthropicClient |
| + meter *usageMeter // model-spend accounting, flushed to the instance (#857, meter.go) |
| + // pendingUsage is the one usage report in flight — kept verbatim across |
| + // failed flushes so the ledger can dedupe a replay whose first ack was |
| + // lost. Touched only on the meterLoop goroutine. |
| + pendingUsage *usageReport |
| + // budgetState is the space's AI budget as of the last usage flush ("ok", |
| + // "unmetered", …). Inert today — phase 3 (docs/ai-provisioning.md) gates |
| + // the model path on it. |
| + budgetState atomic.Value |
| + connect *connectConfig // connected-services (docs/connected-services.md) |
| + keys map[string]threadEntry |
| + cursors map[string]int64 // threadId -> last message id we've processed |
| + agents map[string]bool // roster labels of ALL agents — never reply to one |
| + mention *regexp.Regexp |
| + bare *regexp.Regexp // bare-name mention, for listen mode "name" |
| // Console-managed config (loadConfig, re-fetched each connect). |
| persona string |
| @@ -254,10 +264,14 @@ func (r *Runtime) Run(ctx context.Context) error { |
| r.charterHomes = map[string]charterHome{} |
| r.mention = regexp.MustCompile(`(?i)(^|[^a-z0-9])@` + regexp.QuoteMeta(r.Session.Name) + `\b`) |
| r.bare = regexp.MustCompile(`(?i)(^|[^a-z0-9])` + regexp.QuoteMeta(r.Session.Name) + `\b`) |
| + r.meter = newUsageMeter() |
| if r.AIKey != "" { |
| - r.ai = newAnthropicClient(r.AIKey) |
| + r.ai = newAnthropicClient(r.AIKey, r.recordUsage) |
| } |
| r.connect = loadConnectConfig() |
| + // The usage meter flushes on its own clock, independent of connection |
| + // state — a failed flush restores the counters and the next tick retries. |
| + go r.meterLoop(ctx) |
| if r.Charters { |
| // The slow reconcile loop: heals lost ticks, drops alarms for deleted |
| // charters, and arms charters that appeared while we were connected. |
diff --git a/internal/bot/task_test.go b/internal/bot/task_test.go| index 90a52a2..bf36797 100644 |
| --- a/internal/bot/task_test.go |
| +++ b/internal/bot/task_test.go |
| @@ -39,7 +39,7 @@ func TestExecTaskRetriesWithoutThinking(t *testing.T) { |
| defer srv.Close() |
| t.Setenv("ANTHROPIC_BASE_URL", srv.URL) |
| - r := &Runtime{Session: &Session{Name: "Eleven"}, ai: newAnthropicClient("test-key"), Logf: t.Logf} |
| + r := &Runtime{Session: &Session{Name: "Eleven"}, ai: newAnthropicClient("test-key", nil), Logf: t.Logf} |
| res, err := r.execTask("Test Bot", &TaskSpec{Prompt: "write the edition"}) |
| if err != nil { |
| t.Fatal(err) |
diff --git a/main.go b/main.go| index d8a7076..c20c7b4 100644 |
| --- a/main.go |
| +++ b/main.go |
| @@ -92,6 +92,11 @@ type Server struct { |
| apns apnsSender // nil unless APNs (direct p8 or relay) is configured |
| maxUpload int64 // per-upload attachment cap (LCHAT_MAX_UPLOAD_BYTES) |
| blobQuota int64 // per-instance total blob-bytes ceiling (LCHAT_BLOB_QUOTA_BYTES) |
| + // The usage-pipe relay (usage.go, #857): where sidecar usage reports go |
| + // and the per-space meter token that authenticates them. Pushed into the |
| + // env by the provisioner; empty = unmetered (self-hosted / BYO key). |
| + meterURL string // LCHAT_METER_URL |
| + meterToken string // LCHAT_METER_TOKEN |
| // Directory of sideloaded mini-app modules served at /static/apps-custom/ |
| // (LCHAT_APPS_DIR); empty disables the feature. See customapps.go. |
| appsDir string |
| @@ -274,7 +279,9 @@ func main() { |
| entitlement: strings.TrimSpace(os.Getenv("LCHAT_ENTITLEMENT")), |
| maxUpload: envBytes("LCHAT_MAX_UPLOAD_BYTES", defaultMaxUploadBytes), |
| blobQuota: envBytes("LCHAT_BLOB_QUOTA_BYTES", defaultBlobQuotaBytes), |
| - appsDir: strings.TrimSpace(os.Getenv("LCHAT_APPS_DIR"))} |
| + meterURL: strings.TrimSpace(os.Getenv("LCHAT_METER_URL")), |
| + meterToken: strings.TrimSpace(os.Getenv("LCHAT_METER_TOKEN")), |
| + appsDir: strings.TrimSpace(os.Getenv("LCHAT_APPS_DIR"))} |
| s.brakes = newModBrakes() |
| // Tamper evidence (checkpoint.go): by default the signing seed lives in |
| @@ -415,6 +422,7 @@ func (s *Server) handler() http.Handler { |
| // The charter host's wake-time report (bot-token-gated; folded into |
| // GET /api/next-due so hibernated charters still fire on time). |
| mux.HandleFunc("POST /api/charter-next-due", s.postCharterNextDue) |
| + mux.HandleFunc("POST /api/bot-usage", s.postBotUsage) |
| // The sandbox frame document: an opaque-origin cage for member-app code. |
| mux.HandleFunc("GET /app-sandbox", s.serveAppSandbox) |
diff --git a/meter.go b/meter.go| index f869f76..b56a654 100644 |
| --- a/meter.go |
| +++ b/meter.go |
| @@ -24,6 +24,7 @@ import ( |
| "path/filepath" |
| "strconv" |
| "strings" |
| + "sync" |
| "time" |
| ) |
| @@ -92,6 +93,13 @@ func (s *cloudServer) meterOnce() (metered int) { |
| return 0 |
| } |
| now := time.Now().Unix() |
| + if s.usageURL != "" { |
| + // Applied-report dedupe ids age out with the sweep (usage pipe, #857); |
| + // the sidecar retry horizon is minutes, so a week is generous. |
| + if err := s.db.pruneUsageReports(now - 7*24*3600); err != nil { |
| + log.Printf("cloud: meter: usage-report prune: %v", err) |
| + } |
| + } |
| for _, sp := range spaces { |
| st, err := s.controlStats(sp.Host) |
| if err != nil { |
| @@ -106,6 +114,11 @@ func (s *cloudServer) meterOnce() (metered int) { |
| if want := s.capFor(sp.Entitlement, sp.Tier); want > 0 && want != sp.QuotaBytes { |
| s.pushQuota(sp.ID, sp.Host, sp.Entitlement, sp.Tier) |
| } |
| + // The usage pipe (#857): any space without a meter token gets one — |
| + // how the fleet converges without a per-space backfill step. |
| + if s.usageURL != "" && sp.MeterTokenHash == "" { |
| + s.pushMeterToken(sp.ID, sp.Host) |
| + } |
| // Replication class reconcile (cloudreplication.go): no ledger record |
| // to compare, so push each pass — controld no-ops on a matching class. |
| s.pushReplication(sp.Host, sp.Entitlement, sp.Tier) |
| @@ -227,27 +240,61 @@ func entitlementLabel(entitlement, tier string) string { |
| // ---- quota apply (controld-side, root) ------------------------------------------- |
| -// setInstanceQuota upserts LCHAT_BLOB_QUOTA_BYTES in an instance's env file |
| -// (0 removes it — back to the compiled default) and, when entitlement is |
| -// non-empty, mirrors LCHAT_ENTITLEMENT ("free"/"paid" — the UI's billing- |
| -// affordance gate) in the same write, preserving every other var. Pure but |
| -// for the filesystem, so it's directly testable. |
| -func setInstanceQuota(envPath string, quota int64, entitlement string) error { |
| +// instanceEnvMu serializes every read-modify-write of instance env files. |
| +// controld's quota and meter pushes arrive on separate request goroutines |
| +// (a Stripe webhook and the sweep can land together) and write DIFFERENT |
| +// keys to the SAME file — unserialized, the loser's read-before-write |
| +// snapshot would silently drop the winner's keys, and because both callers |
| +// record success in the ledger, nothing would ever retry. One process-wide |
| +// mutex is plenty at this scale; per-host granularity would buy nothing. |
| +var instanceEnvMu sync.Mutex |
| + |
| +// updateInstanceEnv applies one mutation to an env file under the lock — |
| +// the only correct way to touch an instance env after the daemon is up. |
| +func updateInstanceEnv(envPath string, mutate func(env map[string]string)) error { |
| + instanceEnvMu.Lock() |
| + defer instanceEnvMu.Unlock() |
| env, err := readEnvFile(envPath) |
| if err != nil { |
| return err |
| } |
| - if quota <= 0 { |
| - delete(env, "LCHAT_BLOB_QUOTA_BYTES") |
| - } else { |
| - env["LCHAT_BLOB_QUOTA_BYTES"] = strconv.FormatInt(quota, 10) |
| - } |
| - if entitlement != "" { |
| - env["LCHAT_ENTITLEMENT"] = entitlement |
| - } |
| + mutate(env) |
| return writeEnvFile(envPath, env) |
| } |
| +// setInstanceQuota upserts LCHAT_BLOB_QUOTA_BYTES in an instance's env file |
| +// (0 removes it — back to the compiled default) and, when entitlement is |
| +// non-empty, mirrors LCHAT_ENTITLEMENT ("free"/"paid" — the UI's billing- |
| +// affordance gate) in the same write, preserving every other var. |
| +func setInstanceQuota(envPath string, quota int64, entitlement string) error { |
| + return updateInstanceEnv(envPath, func(env map[string]string) { |
| + if quota <= 0 { |
| + delete(env, "LCHAT_BLOB_QUOTA_BYTES") |
| + } else { |
| + env["LCHAT_BLOB_QUOTA_BYTES"] = strconv.FormatInt(quota, 10) |
| + } |
| + if entitlement != "" { |
| + env["LCHAT_ENTITLEMENT"] = entitlement |
| + } |
| + }) |
| +} |
| + |
| +// setInstanceMeter upserts LCHAT_METER_URL + LCHAT_METER_TOKEN in an |
| +// instance's env file (both empty removes the pair — metering off), |
| +// preserving every other var. The controld half of the meter-token push, |
| +// next to its quota sibling. |
| +func setInstanceMeter(envPath, url, token string) error { |
| + return updateInstanceEnv(envPath, func(env map[string]string) { |
| + if url == "" { |
| + delete(env, "LCHAT_METER_URL") |
| + delete(env, "LCHAT_METER_TOKEN") |
| + } else { |
| + env["LCHAT_METER_URL"] = url |
| + env["LCHAT_METER_TOKEN"] = token |
| + } |
| + }) |
| +} |
| + |
| // pushQuotaByID looks a space's billing state up and applies its cap — |
| // the webhook-side entry point (events carry ids, not rows). |
| func (s *cloudServer) pushQuotaByID(id int64) { |
diff --git a/meter_test.go b/meter_test.go| index f0fb416..9dd2136 100644 |
| --- a/meter_test.go |
| +++ b/meter_test.go |
| @@ -5,6 +5,7 @@ import ( |
| "net/http" |
| "os" |
| "path/filepath" |
| + "sync" |
| "testing" |
| "time" |
| ) |
| @@ -231,3 +232,40 @@ func TestQuotaPushOnUpgradeAndSweep(t *testing.T) { |
| t.Errorf("unconfigured caps pushed: %v", pushed2) |
| } |
| } |
| + |
| +// The quota and meter pushes write DIFFERENT keys to the SAME env file from |
| +// separate request goroutines; updateInstanceEnv serializes them so neither |
| +// side's read-before-write snapshot can drop the other's keys. Hammered |
| +// concurrently, every key family must survive. |
| +func TestInstanceEnvWritesSerialized(t *testing.T) { |
| + envPath := filepath.Join(t.TempDir(), "space.env") |
| + if err := writeEnvFile(envPath, map[string]string{"LCHAT_ADMIN_TOKEN": "keepme"}); err != nil { |
| + t.Fatal(err) |
| + } |
| + var wg sync.WaitGroup |
| + for i := 0; i < 50; i++ { |
| + wg.Add(2) |
| + go func() { |
| + defer wg.Done() |
| + if err := setInstanceQuota(envPath, 12345, "free"); err != nil { |
| + t.Error(err) |
| + } |
| + }() |
| + go func() { |
| + defer wg.Done() |
| + if err := setInstanceMeter(envPath, "https://cloud.example/api/usage", "tok"); err != nil { |
| + t.Error(err) |
| + } |
| + }() |
| + } |
| + wg.Wait() |
| + env, err := readEnvFile(envPath) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + for _, key := range []string{"LCHAT_ADMIN_TOKEN", "LCHAT_BLOB_QUOTA_BYTES", "LCHAT_METER_URL", "LCHAT_METER_TOKEN"} { |
| + if env[key] == "" { |
| + t.Errorf("%s lost in concurrent pushes: %v", key, env) |
| + } |
| + } |
| +} |
diff --git a/usage.go b/usage.go| new file mode 100644 |
| index 0000000..6436919 |
| --- /dev/null |
| +++ b/usage.go |
| @@ -0,0 +1,151 @@ |
| +// usage.go — the instance half of the usage pipe (#857, |
| +// docs/ai-provisioning.md phase 1). The sidecar counts model spend (it's the |
| +// only place spend is visible) and reports here over its bot token; this |
| +// handler validates and relays to the cloud ledger, authenticating with the |
| +// per-space meter token the provisioner pushed into the env |
| +// (LCHAT_METER_URL / LCHAT_METER_TOKEN). The instance persists nothing — it |
| +// stays a stateless relay, and rows are quantities only (token counts, |
| +// request counts, run counts), never content, per the |
| +// meter-only-what-ciphertext-exposes rule (#668). |
| +// |
| +// Unconfigured — a self-hosted instance, or a BYO-key space — the handler |
| +// accepts and drops: the sidecar keeps working and nothing is recorded |
| +// anywhere. The relayed response carries the space's AI budget state back to |
| +// the sidecar; today the cloud always answers "ok" (phase 3 computes it). |
| +package main |
| + |
| +import ( |
| + "bytes" |
| + "encoding/json" |
| + "fmt" |
| + "net/http" |
| + "regexp" |
| + "time" |
| +) |
| + |
| +// botUsageRow is the wire shape shared by both hops (sidecar → instance → |
| +// cloud) and the cloud ledger's cell: a (day, model, dimension) quantity. |
| +type botUsageRow struct { |
| + Day string `json:"day"` |
| + Model string `json:"model,omitempty"` |
| + Dimension string `json:"dimension"` |
| + Quantity int64 `json:"quantity"` |
| +} |
| + |
| +var usageDayRe = regexp.MustCompile(`^\d{4}-\d{2}-\d{2}$`) |
| + |
| +// usageDimensions is the closed set a report may carry — additions are a |
| +// deliberate schema decision, not a sidecar deploy detail. |
| +var usageDimensions = map[string]bool{ |
| + "llm_requests": true, |
| + "llm_input_tokens": true, |
| + "llm_output_tokens": true, |
| + "llm_cache_write_tokens": true, |
| + "llm_cache_read_tokens": true, |
| + "charter_runs": true, |
| +} |
| + |
| +const maxUsageRows = 64 |
| + |
| +// usageReportIDRe: the dedupe key a sidecar mints per flush — hex, bounded. |
| +// The ledger records it and drops replays, so an ack lost after a commit |
| +// can't double-count (the sidecar resends the same report verbatim). |
| +var usageReportIDRe = regexp.MustCompile(`^[0-9a-f]{16,64}$`) |
| + |
| +// validateUsageReport is shared by the instance relay and the cloud |
| +// ingester, so a report the relay accepts is by construction one the ledger |
| +// accepts. |
| +func validateUsageReport(reportID string, rows []botUsageRow) error { |
| + if !usageReportIDRe.MatchString(reportID) { |
| + return fmt.Errorf("bad report id") |
| + } |
| + if len(rows) == 0 || len(rows) > maxUsageRows { |
| + return fmt.Errorf("1..%d rows required", maxUsageRows) |
| + } |
| + for _, row := range rows { |
| + switch { |
| + case !usageDayRe.MatchString(row.Day): |
| + return fmt.Errorf("bad day %q", row.Day) |
| + case !usageDimensions[row.Dimension]: |
| + return fmt.Errorf("unknown dimension %q", row.Dimension) |
| + case len(row.Model) > 128: |
| + return fmt.Errorf("model too long") |
| + case row.Quantity <= 0 || row.Quantity > 1e12: |
| + return fmt.Errorf("bad quantity %d", row.Quantity) |
| + } |
| + } |
| + return nil |
| +} |
| + |
| +// postBotUsage ingests a sidecar's usage report. The right principal is a |
| +// bot's own device token, same gate as the charter wake-time report. |
| +func (s *Server) postBotUsage(w http.ResponseWriter, r *http.Request) { |
| + var body struct { |
| + Token string `json:"token"` |
| + ReportID string `json:"report_id"` |
| + Rows []botUsageRow `json:"rows"` |
| + } |
| + if json.NewDecoder(http.MaxBytesReader(w, r.Body, 64<<10)).Decode(&body) != nil { |
| + httpError(w, http.StatusBadRequest, "bad request") |
| + return |
| + } |
| + uid, ok := s.db.deviceTokenUser(body.Token) |
| + if !ok || !s.db.isAgent(uid) { |
| + httpError(w, http.StatusForbidden, "bot token required") |
| + return |
| + } |
| + if err := validateUsageReport(body.ReportID, body.Rows); err != nil { |
| + httpError(w, http.StatusBadRequest, err.Error()) |
| + return |
| + } |
| + if s.meterURL == "" || s.meterToken == "" { |
| + // No meter configured: accept and drop, and say so — "unmetered" is |
| + // what lets a BYO-key or self-hosted sidecar tell "nothing counts |
| + // here" from "the meter is down". |
| + writeJSON(w, http.StatusOK, map[string]any{"ok": true, "budget": map[string]any{"state": "unmetered"}}) |
| + return |
| + } |
| + budget, err := relayUsage(s.meterURL, s.meterToken, body.ReportID, body.Rows) |
| + if err != nil { |
| + // The sidecar restores its counters and retries next tick — nothing |
| + // is lost to a cloud blip, so a plain 502 is the whole story. |
| + httpError(w, http.StatusBadGateway, "usage relay failed") |
| + return |
| + } |
| + writeJSON(w, http.StatusOK, map[string]any{"ok": true, "budget": budget}) |
| +} |
| + |
| +// relayUsage forwards a validated report to the cloud ledger and returns the |
| +// budget object from its response. |
| +func relayUsage(meterURL, meterToken, reportID string, rows []botUsageRow) (map[string]any, error) { |
| + raw, err := json.Marshal(map[string]any{"report_id": reportID, "rows": rows}) |
| + if err != nil { |
| + return nil, err |
| + } |
| + req, err := http.NewRequest("POST", meterURL, bytes.NewReader(raw)) |
| + if err != nil { |
| + return nil, err |
| + } |
| + req.Header.Set("Content-Type", "application/json") |
| + req.Header.Set("Authorization", "Bearer "+meterToken) |
| + client := &http.Client{Timeout: 15 * time.Second} |
| + resp, err := client.Do(req) |
| + if err != nil { |
| + return nil, err |
| + } |
| + defer resp.Body.Close() |
| + if resp.StatusCode != http.StatusOK { |
| + return nil, fmt.Errorf("meter: HTTP %d", resp.StatusCode) |
| + } |
| + var out struct { |
| + OK bool `json:"ok"` |
| + Budget map[string]any `json:"budget"` |
| + } |
| + if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { |
| + return nil, err |
| + } |
| + if out.Budget == nil { |
| + out.Budget = map[string]any{"state": "ok"} |
| + } |
| + return out.Budget, nil |
| +} |
diff --git a/usage_test.go b/usage_test.go| new file mode 100644 |
| index 0000000..41be643 |
| --- /dev/null |
| +++ b/usage_test.go |
| @@ -0,0 +1,126 @@ |
| +package main |
| + |
| +import ( |
| + "bytes" |
| + "encoding/json" |
| + "net/http" |
| + "net/http/httptest" |
| + "testing" |
| +) |
| + |
| +func postUsage(t *testing.T, base, token string, rows []botUsageRow) (int, map[string]any) { |
| + t.Helper() |
| + return postUsageReport(t, base, token, "aaaaaaaaaaaaaaaa", rows) |
| +} |
| + |
| +func postUsageReport(t *testing.T, base, token, reportID string, rows []botUsageRow) (int, map[string]any) { |
| + t.Helper() |
| + raw, _ := json.Marshal(map[string]any{"token": token, "report_id": reportID, "rows": rows}) |
| + resp, err := http.Post(base+"/api/bot-usage", "application/json", bytes.NewReader(raw)) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer resp.Body.Close() |
| + var body map[string]any |
| + json.NewDecoder(resp.Body).Decode(&body) |
| + return resp.StatusCode, body |
| +} |
| + |
| +// The relay end to end: a bot's report is validated, forwarded to the cloud |
| +// meter with the instance's bearer token, and the cloud's budget state comes |
| +// back to the sidecar untouched. |
| +func TestBotUsageRelay(t *testing.T) { |
| + ts, s := newTestServer(t) |
| + agent := provisionNamed(t, ts.URL, "Eleven", true) |
| + |
| + var gotAuth string |
| + var gotRows []botUsageRow |
| + cloud := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| + gotAuth = r.Header.Get("Authorization") |
| + var body struct { |
| + Rows []botUsageRow `json:"rows"` |
| + } |
| + json.NewDecoder(r.Body).Decode(&body) |
| + gotRows = body.Rows |
| + writeJSON(w, http.StatusOK, map[string]any{"ok": true, "budget": map[string]any{"state": "low"}}) |
| + })) |
| + defer cloud.Close() |
| + s.meterURL, s.meterToken = cloud.URL, "meter-tok" |
| + |
| + rows := []botUsageRow{ |
| + {Day: "2026-07-14", Model: "claude-sonnet-5", Dimension: "llm_input_tokens", Quantity: 150}, |
| + {Day: "2026-07-14", Dimension: "charter_runs", Quantity: 1}, |
| + } |
| + code, body := postUsage(t, ts.URL, agent.Token, rows) |
| + if code != http.StatusOK { |
| + t.Fatalf("relay: %d %v", code, body) |
| + } |
| + if gotAuth != "Bearer meter-tok" { |
| + t.Fatalf("cloud auth = %q", gotAuth) |
| + } |
| + if len(gotRows) != 2 || gotRows[0].Quantity != 150 { |
| + t.Fatalf("cloud rows = %+v", gotRows) |
| + } |
| + if budget, _ := body["budget"].(map[string]any); budget["state"] != "low" { |
| + t.Fatalf("budget not passed through: %v", body) |
| + } |
| +} |
| + |
| +// Unconfigured (self-hosted / BYO key): accept, drop, and say "unmetered". |
| +func TestBotUsageUnmetered(t *testing.T) { |
| + ts, _ := newTestServer(t) |
| + agent := provisionNamed(t, ts.URL, "Eleven", true) |
| + code, body := postUsage(t, ts.URL, agent.Token, []botUsageRow{ |
| + {Day: "2026-07-14", Dimension: "llm_requests", Quantity: 1}}) |
| + if code != http.StatusOK { |
| + t.Fatalf("unmetered: %d %v", code, body) |
| + } |
| + if budget, _ := body["budget"].(map[string]any); budget["state"] != "unmetered" { |
| + t.Fatalf("want unmetered, got %v", body) |
| + } |
| +} |
| + |
| +// The gate: bots only, and malformed rows never leave the instance. |
| +func TestBotUsageGateAndValidation(t *testing.T) { |
| + ts, s := newTestServer(t) |
| + alice := provisionNamed(t, ts.URL, "Alice", false) |
| + agent := provisionNamed(t, ts.URL, "Eleven", true) |
| + s.meterURL, s.meterToken = "http://cloud.invalid", "tok" // must never be dialed below |
| + |
| + good := []botUsageRow{{Day: "2026-07-14", Dimension: "llm_requests", Quantity: 1}} |
| + if code, _ := postUsage(t, ts.URL, alice.Token, good); code != http.StatusForbidden { |
| + t.Fatalf("member token: want 403, got %d", code) |
| + } |
| + if code, _ := postUsage(t, ts.URL, "nonsense", good); code != http.StatusForbidden { |
| + t.Fatalf("stranger: want 403, got %d", code) |
| + } |
| + for name, rows := range map[string][]botUsageRow{ |
| + "empty": {}, |
| + "bad day": {{Day: "yesterday", Dimension: "llm_requests", Quantity: 1}}, |
| + "bad dimension": {{Day: "2026-07-14", Dimension: "plaintext_words", Quantity: 1}}, |
| + "bad quantity": {{Day: "2026-07-14", Dimension: "llm_requests", Quantity: 0}}, |
| + } { |
| + if code, _ := postUsage(t, ts.URL, agent.Token, rows); code != http.StatusBadRequest { |
| + t.Fatalf("%s: want 400, got %d", name, code) |
| + } |
| + } |
| + // A report without a well-formed dedupe id is refused too. |
| + if code, _ := postUsageReport(t, ts.URL, agent.Token, "not hex!", good); code != http.StatusBadRequest { |
| + t.Fatal("bad report id: want 400") |
| + } |
| +} |
| + |
| +// A cloud blip is a 502 — the sidecar keeps its counters and retries. |
| +func TestBotUsageCloudDown(t *testing.T) { |
| + ts, s := newTestServer(t) |
| + agent := provisionNamed(t, ts.URL, "Eleven", true) |
| + dead := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {})) |
| + dead.Close() // connection refused from here on |
| + s.meterURL, s.meterToken = dead.URL, "tok" |
| + |
| + code, _ := postUsage(t, ts.URL, agent.Token, []botUsageRow{ |
| + {Day: "2026-07-14", Dimension: "llm_requests", Quantity: 1}}) |
| + if code != http.StatusBadGateway { |
| + t.Fatalf("cloud down: want 502, got %d", code) |
| + } |
| +} |