From bb554408fdd31694dccf305fd912c23413daac4a Mon Sep 17 00:00:00 2001 From: Zachary Lowden Date: Tue, 10 Nov 2026 19:03:48 -0600 Subject: [PATCH 1/3] ccusage: push each cycle's readings to an external dashboard An optional pusher for the Claude account usage poller: with MUSTER_CC_USAGE_PUSH_URL (https; http only to loopback) and MUSTER_CC_USAGE_PUSH_TOKEN (Bearer), after each poll cycle muster PUTs {source_instance, read_at, accounts} once, 10s timeout, expecting 204. accounts is built by the same ccUsageAccounts the GET /api/cc-accounts/usage handler uses, so the two are byte-for-byte equal. A failure is logged by class only (never the token, body or answer), counted in muster_cc_usage_push_failures_total{reason}, and not retried until the next cycle; redirects are not followed. Boot is refused when the URL is set with the poller off, without the token, or not https, and when the token or instance is set without the URL. MUSTER_CC_USAGE_PUSH_INSTANCE (<=64 chars) defaults to MUSTER_PUBLIC_URL's host, else "muster". Tests: golden (pushed accounts == API accounts, used_by included), refusal matrix, 204 vs 4xx/5xx/200/redirect/timeout/transport/build, no token or body in logs, one AfterPass per pass. 24 mutants, all killed by the specific assertion. Co-Authored-By: Claude Opus 5.5 (1M context) --- cmd/muster-server/ccusage_config.go | 48 +++++ cmd/muster-server/ccusage_config_test.go | 61 ++++++ cmd/muster-server/config.go | 6 +- cmd/muster-server/main.go | 18 ++ docs/configuration.md | 3 + docs/jobs.md | 35 +++ internal/api/cc_usage.go | 25 +++ internal/api/cc_usage_test.go | 52 +++++ internal/ccusage/poller.go | 9 + internal/ccusage/push.go | 191 ++++++++++++++++ internal/ccusage/push_test.go | 263 +++++++++++++++++++++++ internal/metrics/metrics.go | 15 ++ internal/metrics/metrics_test.go | 2 + tests/features_config_test.py | 3 + 14 files changed, 730 insertions(+), 1 deletion(-) create mode 100644 internal/ccusage/push.go create mode 100644 internal/ccusage/push_test.go diff --git a/cmd/muster-server/ccusage_config.go b/cmd/muster-server/ccusage_config.go index 5a7cc9117..075cca891 100644 --- a/cmd/muster-server/ccusage_config.go +++ b/cmd/muster-server/ccusage_config.go @@ -21,11 +21,23 @@ type ccUsageConfig struct { // Interval is MUSTER_CC_USAGE_POLL_INTERVAL (ccusage.DefaultInterval, 15m, // by default), from ccusage.MinInterval to ccusage.MaxInterval. Interval time.Duration + // Push is the optional pusher's; Push.URL "" is off. + Push ccUsagePushConfig +} + +// ccUsagePushConfig is the usage pusher's (MUSTER_CC_USAGE_PUSH_*): after each +// poll cycle it PUTs the readings to URL with Token as a bearer, labelled +// Instance. +type ccUsagePushConfig struct { + URL string + Token string + Instance string } // parseCCUsage reads the poller's configuration. The interval is refused // without the switch: a knob that does nothing reads as if it did something. func parseCCUsage(getenv func(string) string) (ccUsageConfig, error) { + var err error c := ccUsageConfig{On: envFlag(getenv, envCCUsagePoll), Interval: ccusage.DefaultInterval} if v := strings.TrimSpace(getenv(envCCUsagePollInterval)); v != "" { if !c.On { @@ -37,9 +49,45 @@ func parseCCUsage(getenv func(string) string) (ccUsageConfig, error) { } c.Interval = d } + if c.Push, err = parseCCUsagePush(getenv, c.On); err != nil { + return ccUsageConfig{}, err + } return c, nil } +// parseCCUsagePush reads the pusher's configuration. Boot is refused when the +// URL is set with the poller off (it would push nothing, ever), without a +// token, or not https (plain http only to a loopback host, for tests); and +// when the token or the instance is set without the URL (a knob that does +// nothing). The token never appears in a refusal. +func parseCCUsagePush(getenv func(string) string, pollerOn bool) (ccUsagePushConfig, error) { + u := strings.TrimSpace(getenv(envCCUsagePushURL)) + tok := strings.TrimSpace(getenv(envCCUsagePushToken)) + inst := strings.TrimSpace(getenv(envCCUsagePushInstance)) + if u == "" { + for _, v := range []struct{ name, val string }{{envCCUsagePushToken, tok}, {envCCUsagePushInstance, inst}} { + if v.val != "" { + return ccUsagePushConfig{}, fmt.Errorf("%s is set but %s is not: it does nothing without the push URL", v.name, envCCUsagePushURL) + } + } + return ccUsagePushConfig{}, nil + } + if !pollerOn { + return ccUsagePushConfig{}, fmt.Errorf("%s is set but %s is not: with the poller off there are no readings to push", envCCUsagePushURL, envCCUsagePoll) + } + if err := ccusage.CheckPushURL(u); err != nil { + return ccUsagePushConfig{}, fmt.Errorf("invalid %s: %v", envCCUsagePushURL, err) + } + if tok == "" { + return ccUsagePushConfig{}, fmt.Errorf("%s is set but %s is not: the push needs its bearer token", envCCUsagePushURL, envCCUsagePushToken) + } + instance, err := ccusage.PushInstance(inst, getenv(envPublicURL)) + if err != nil { + return ccUsagePushConfig{}, fmt.Errorf("invalid %s: %v", envCCUsagePushInstance, err) + } + return ccUsagePushConfig{URL: u, Token: tok, Instance: instance}, nil +} + // ccUsageRefusal is why this configuration cannot run the poller, or "": it // reads the pool's tokens (the claude-code kind) and keeps its readings in the // database. diff --git a/cmd/muster-server/ccusage_config_test.go b/cmd/muster-server/ccusage_config_test.go index f5a61dd44..e568f70d8 100644 --- a/cmd/muster-server/ccusage_config_test.go +++ b/cmd/muster-server/ccusage_config_test.go @@ -53,3 +53,64 @@ func TestCCUsagePollerNeedsThePoolAndADatabase(t *testing.T) { t.Fatalf("on with both: %q", r) } } + +// TestCCUsagePushRefusalMatrix: the pusher is off unless its URL is set; with +// it, boot is refused when the poller is off, the token is missing, or the URL +// is not https (http only to loopback) — and the token or instance alone is +// refused as a knob that does nothing. No refusal carries the token. +func TestCCUsagePushRefusalMatrix(t *testing.T) { + const tok = "push-token-SENTINEL-1234" + on := map[string]string{envCCUsagePoll: "1"} + with := func(kv map[string]string) map[string]string { + out := map[string]string{} + for k, v := range on { + out[k] = v + } + for k, v := range kv { + out[k] = v + } + return out + } + c, err := parseCCUsage(ccUsageEnv(on)) + if err != nil || c.Push != (ccUsagePushConfig{}) { + t.Fatalf("no push variables: %+v %v; want the pusher off", c.Push, err) + } + c, err = parseCCUsage(ccUsageEnv(with(map[string]string{envCCUsagePushURL: "https://dash.example.com/api/usage", + envCCUsagePushToken: tok, envPublicURL: "https://muster.example.com"}))) + if err != nil || c.Push != (ccUsagePushConfig{URL: "https://dash.example.com/api/usage", Token: tok, Instance: "muster.example.com"}) { + t.Fatalf("a valid push: %+v %v", c.Push, err) + } + c, err = parseCCUsage(ccUsageEnv(with(map[string]string{envCCUsagePushURL: "http://127.0.0.1:9/x", + envCCUsagePushToken: tok, envCCUsagePushInstance: "lab"}))) + if err != nil || c.Push.Instance != "lab" { + t.Fatalf("loopback http with an instance: %+v %v", c.Push, err) + } + c, err = parseCCUsage(ccUsageEnv(with(map[string]string{envCCUsagePushURL: "https://dash.example.com/x", envCCUsagePushToken: tok}))) + if err != nil || c.Push.Instance != "muster" { + t.Fatalf("no public URL: instance %q %v, want muster", c.Push.Instance, err) + } + for _, bad := range []struct { + name string + env map[string]string + names string // the variable the refusal must name + }{ + {"poller off", map[string]string{envCCUsagePushURL: "https://dash.example.com/x", envCCUsagePushToken: tok}, envCCUsagePoll}, + {"no token", with(map[string]string{envCCUsagePushURL: "https://dash.example.com/x"}), envCCUsagePushToken}, + {"plain http", with(map[string]string{envCCUsagePushURL: "http://dash.example.com/x", envCCUsagePushToken: tok}), envCCUsagePushURL}, + {"relative", with(map[string]string{envCCUsagePushURL: "/api/usage", envCCUsagePushToken: tok}), envCCUsagePushURL}, + {"userinfo", with(map[string]string{envCCUsagePushURL: "https://u:" + tok + "@dash.example.com/x", envCCUsagePushToken: tok}), envCCUsagePushURL}, + {"instance too long", with(map[string]string{envCCUsagePushURL: "https://dash.example.com/x", envCCUsagePushToken: tok, + envCCUsagePushInstance: strings.Repeat("i", 65)}), envCCUsagePushInstance}, + {"token without URL", with(map[string]string{envCCUsagePushToken: tok}), envCCUsagePushToken}, + {"instance without URL", with(map[string]string{envCCUsagePushInstance: "lab"}), envCCUsagePushInstance}, + } { + _, err := parseCCUsage(ccUsageEnv(bad.env)) + if err == nil || !strings.Contains(err.Error(), bad.names) { + t.Errorf("%s: %v; want a refusal naming %s", bad.name, err, bad.names) + continue + } + if strings.Contains(err.Error(), tok) { + t.Errorf("%s: the refusal carries the token: %v", bad.name, err) + } + } +} diff --git a/cmd/muster-server/config.go b/cmd/muster-server/config.go index b7fef542d..a9031ba29 100644 --- a/cmd/muster-server/config.go +++ b/cmd/muster-server/config.go @@ -284,9 +284,13 @@ const ( // THE CLAUDE ACCOUNT USAGE POLLER (internal/ccusage, docs/jobs.md "Claude // account usage"; ccusage_config.go parses them): its switch (off by - // default) and its interval. + // default) and its interval; and its optional pusher, which PUTs each + // cycle's readings to an external dashboard (off unless the URL is set). envCCUsagePoll = "MUSTER_CC_USAGE_POLL" envCCUsagePollInterval = "MUSTER_CC_USAGE_POLL_INTERVAL" + envCCUsagePushURL = "MUSTER_CC_USAGE_PUSH_URL" + envCCUsagePushToken = "MUSTER_CC_USAGE_PUSH_TOKEN" + envCCUsagePushInstance = "MUSTER_CC_USAGE_PUSH_INSTANCE" // Browser flows: the deployment's own infrastructure names (comma-separated) // that an AGENT's browser-flow text may not carry (internal/flows diff --git a/cmd/muster-server/main.go b/cmd/muster-server/main.go index 56f5c3c82..b9dea1540 100644 --- a/cmd/muster-server/main.go +++ b/cmd/muster-server/main.go @@ -752,6 +752,17 @@ func buildApp(ctx context.Context, cfg config, logger *log.Logger) (*app, error) srv := api.New(bus, cfg.authConfig(), logger) srv.UseExtensions(ext) + // The usage pusher (MUSTER_CC_USAGE_PUSH_URL): after each poll cycle the + // readings, exactly as GET /api/cc-accounts/usage answers the admin, are + // PUT to the dashboard. parseCCUsagePush refused it without the poller. + if a.ccUsage != nil && cfg.CCUsage.Push.URL != "" { + push := cfg.CCUsage.Push + a.ccUsage.AfterPass = (&ccusage.Pusher{URL: push.URL, Token: push.Token, Logger: logger, + Client: &http.Client{Timeout: ccusage.PushTimeout}, + Body: func(ctx context.Context, now time.Time) ([]byte, error) { + return srv.CCUsagePushBody(ctx, authz.OperatorUserID, push.Instance, now) + }}).PushAfterPass + } if cfg.goTrueConfigured() && a.pool != nil { srv.UseGoTrue(goTrueAuth(cfg, a.pool)) } @@ -1756,6 +1767,13 @@ func (a *app) logBanner(ext api.Extensions, port router.Port) { "token, from %s when the token may use it, else from a one-token probe of %s (one output token of quota per read; "+ "never an account the pool marks cooling); numbers only, in cc_account_usage, served at %s", envCCUsagePoll, ext.CCUsageInterval, ccusage.Endpoint, ccusage.ProbeURL, api.CCAccountUsagePath) + if a.cfg.CCUsage.Push.URL != "" { + l.Printf("claude account usage push: ON (%s) — after each poll cycle the readings are PUT to the configured URL "+ + "as source_instance %q (expects 204; a failure is logged and counted, and the next cycle pushes again)", + envCCUsagePushURL, a.cfg.CCUsage.Push.Instance) + } else { + l.Printf("claude account usage push: OFF — %s is unset", envCCUsagePushURL) + } } else { l.Printf("claude account usage: OFF — %s is unset; no account's utilization is read", envCCUsagePoll) } diff --git a/docs/configuration.md b/docs/configuration.md index 853088231..2e9657daf 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -98,6 +98,9 @@ See [auth.md](auth.md). | `MUSTER_ASK_LOG_CLASSIFY` | off | no | `1` runs the stall classifier on a plain-text question the [ask log](asks.md) kept, once per turn end (reusing the stall sweep's answer for the streak when it has one), on the owner's OpenRouter key; its scores are stored on the row. Needs `MUSTER_CHIEF_NUDGES=1` (the classifier's configuration). Off by default so the pre-filter's volume (`muster_ask_plain_text_prefilter_total`) is read before the classifier is spent on it. | | `MUSTER_CC_USAGE_POLL` | off | no | `1` reads each of the admin's pooled Claude accounts' subscription utilization once per `MUSTER_CC_USAGE_POLL_INTERVAL` and keeps the latest reading per account (numbers only): from the endpoint Claude Code's `/usage` reads when the account's token may use it (one discovery call per token; a `setup-token` is refused there), else from a one-token Messages probe's `anthropic-ratelimit-unified-*` headers, which spends one output token of the account's quota per read. An account the pool marks rate-limited is not probed until its cooldown ends. Served at `GET /api/cc-accounts/usage` and on Account → Credentials ([jobs](jobs.md#claude-account-usage)). muster-server itself calls `api.anthropic.com:443`. Needs the `claude-code` kind and a database, or boot is refused. Members' own accounts are not read. | | `MUSTER_CC_USAGE_POLL_INTERVAL` | `15m` | no | How often the usage poller reads every account (a Go duration, 5m–24h). A probe answered 429 waits longer (doubling, at least `Retry-After`, at most 6h — or until the window's reset when its status is `rejected`); a token the probe refused (401/403) is not read again until it changes. Refused without `MUSTER_CC_USAGE_POLL`. | +| `MUSTER_CC_USAGE_PUSH_URL` | none | no | Turns on the [usage pusher](jobs.md#pushing-the-readings): after each poll cycle muster `PUT`s the readings, exactly as `GET /api/cc-accounts/usage` answers the admin, to this URL, once, with a 10-second timeout, and expects `204`. A failure is logged (never the token or the body), counted in `muster_cc_usage_push_failures_total{reason}` and not retried until the next cycle; a redirect is not followed. An absolute `https` URL (plain `http` only to a loopback host); credentials in the URL are refused. Refused without `MUSTER_CC_USAGE_POLL` or without `MUSTER_CC_USAGE_PUSH_TOKEN`. | +| `MUSTER_CC_USAGE_PUSH_TOKEN` | none | with the push URL | The pusher's `Authorization: Bearer` token. Never logged. Refused without `MUSTER_CC_USAGE_PUSH_URL`. | +| `MUSTER_CC_USAGE_PUSH_INSTANCE` | `MUSTER_PUBLIC_URL`'s host, else `muster` | no | The pushed `source_instance` label, at most 64 characters (a public host longer than that falls back to `muster`). Refused without `MUSTER_CC_USAGE_PUSH_URL`. | | `MUSTER_SESSION_STATE` | off | no | `1` runs session state's pass (design: `docs/design/session-state.md`, #425): every 30 seconds each owner's sessions get a state, and every stop an episode in `session_episodes`. The stall classifier runs only where the chief's nudges would run it. It also replaces the waiting-window push and the chief's stall notice with **one push per stop** (an episode), naming the session and never quoting it, and draws the open stops as the Asks page's **Stalled / waiting** section ([attention.md](attention.md#stalled--waiting-session-state)). Off, every push is as before. Needs a database. | | `MUSTER_CHIEF_NUDGE_LIVE_OWNERS` | none | no | The user ids, comma-separated, whose stall nudges are **acted on**: a labelled nudge quoting the owner is sent, or a [notice](attention.md#notices) naming where the session is is put on the owner's Asks list (nothing to answer, nothing relayed). A value that is not a list of positive integers refuses to boot. | | `MUSTER_CHIEF_NUDGE_NEEDS_YOU_LIVE_OWNERS` | none | no | The user ids, comma-separated, whose **needs-you** stall nudges are sent too (with `MUSTER_SESSION_STATE=1`; with it off this does nothing): a nudge to an idle session the stall classifier says asked something, quoting the owner's words that answer it ([chief-nudges.md](chief-nudges.md#needs-you-nudges)). It is sent only for an owner listed here **and** in `MUSTER_CHIEF_NUDGE_LIVE_OWNERS`; for everyone else it is shadow (recorded on the Chief nudges tab, never pasted, and the session's stop pushed). Unset: every owner's needs-you nudges are shadow; their other nudges keep `MUSTER_CHIEF_NUDGE_LIVE_OWNERS` alone. Same format as that variable; a value that is not a list of positive integers refuses to boot. The boot line counts the owners on both lists. Taking an owner off it (and restarting) stops a needs-you nudge already queued at its paste. | diff --git a/docs/jobs.md b/docs/jobs.md index 7d791045b..8a5ccdb2f 100644 --- a/docs/jobs.md +++ b/docs/jobs.md @@ -735,6 +735,41 @@ call without an actor. account only. A job counts on the account its last turn ran on. - `poll_interval_seconds` is 0 with the poller off; the numbers then never move. +### Pushing the readings + +With `MUSTER_CC_USAGE_PUSH_URL` set (and its `MUSTER_CC_USAGE_PUSH_TOKEN`), +muster also **pushes** the readings to an external dashboard instead of +waiting to be read: after each poll cycle, once, it sends + +``` +PUT +Authorization: Bearer +Content-Type: application/json + +{"source_instance": "muster.example.com", + "read_at": "2026-11-10T18:21:00Z", + "accounts": [ … ]} +``` + +- `accounts` is the admin's accounts **byte for byte** as `GET + /api/cc-accounts/usage` answers them (above, `used_by` included): both are + built by the same code. +- `source_instance` is `MUSTER_CC_USAGE_PUSH_INSTANCE` (at most 64 + characters), else `MUSTER_PUBLIC_URL`'s host, else `muster`. `read_at` is + when the push was built. +- The push is made after **every** cycle, whether or not a reading changed, so + a dashboard can tell a quiet muster from a dead one. +- The dashboard answers **`204`**. Anything else — a 4xx, a 5xx, another 2xx, + a redirect (never followed: it would re-send the bearer), no answer in 10 + seconds, a transport error, or readings the store could not give — is logged + (the status or the error's class only: never the token, the body or the + answer) and counted in `muster_cc_usage_push_failures_total{reason}` + (`http_4xx`, `http_5xx`, `http_other`, `timeout`, `transport`, `build`). It + is not retried: the next cycle's push carries newer readings. +- Boot is refused when the URL is set with the poller off, without the token, + or not `https` (plain `http` is allowed only to a loopback host, for tests), + and when the token or the instance is set without the URL. + The same readings are drawn on **Account → Credentials** ("Claude account usage"), for the admin and for a member alike. diff --git a/internal/api/cc_usage.go b/internal/api/cc_usage.go index a59b2153f..5f6d2e0c7 100644 --- a/internal/api/cc_usage.go +++ b/internal/api/cc_usage.go @@ -2,6 +2,8 @@ package api import ( "context" + "encoding/json" + "errors" "fmt" "html" "net/http" @@ -185,6 +187,29 @@ func (s *Server) officerPoolOf(name string) (string, string, bool, bool) { return o.ID, o.Tier, proxied, true } +// ccUsagePush is the usage pusher's body (MUSTER_CC_USAGE_PUSH_URL): the +// accounts exactly as GET /api/cc-accounts/usage answers the admin (the same +// ccUsageAccounts, so the two cannot drift), under the instance's label. +type ccUsagePush struct { + SourceInstance string `json:"source_instance"` + ReadAt string `json:"read_at"` + Accounts []ccUsageAccount `json:"accounts"` +} + +// CCUsagePushBody is the usage pusher's request body at now for owner (the +// operator, whose pool the poller reads), labelled instance. ctx must act as +// owner. An error is the store's; the body is then not built. +func (s *Server) CCUsagePushBody(ctx context.Context, owner int64, instance string, now time.Time) ([]byte, error) { + if s.ext.CCUsage == nil { + return nil, errors.New("this server keeps no Claude account usage (no database)") + } + accts, err := s.ccUsageAccounts(ctx, owner, now) + if err != nil { + return nil, err + } + return json.Marshal(ccUsagePush{SourceInstance: instance, ReadAt: now.UTC().Format(time.RFC3339), Accounts: accts}) +} + func (s *Server) handleCCAccountUsage(w http.ResponseWriter, r *http.Request) { if s.ext.CCUsage == nil { s.writeJSON(w, http.StatusServiceUnavailable, map[string]any{"error": "this server keeps no Claude account usage (no database)"}) diff --git a/internal/api/cc_usage_test.go b/internal/api/cc_usage_test.go index 3279a48cc..be9dc2ef6 100644 --- a/internal/api/cc_usage_test.go +++ b/internal/api/cc_usage_test.go @@ -244,3 +244,55 @@ func TestAProxiedOfficerInstanceIsNotCreditedToItsAccount(t *testing.T) { t.Fatalf("alpha's officers = %+v, want only the routine instance %+v", o, want) } } + +// TestTheUsagePushCarriesTheAPIsAccountsByteForByte is the golden: for the +// same readings, the pushed body's `accounts` is byte-for-byte the `accounts` +// GET /api/cc-accounts/usage answers the admin (used_by included), beside +// exactly source_instance and read_at. +func TestTheUsagePushCarriesTheAPIsAccountsByteForByte(t *testing.T) { + h, _, store := newCCUsageFixture(t) + s := New(nil, AuthConfig{HookToken: tokTestHook, ServiceToken: ccUsageServiceToken}, log.New(io.Discard, "", 0)) + s.UseExtensions(Extensions{Users: stubUsers{}, CCUsage: store, CCUsageInterval: 5 * time.Minute}) + rec := ccUsageGet(h, CCAccountUsagePath, map[string]string{"Authorization": "Bearer " + ccUsageServiceToken, ActorHeader: "hq"}) + if rec.Code != http.StatusOK { + t.Fatalf("GET: %d %s", rec.Code, rec.Body) + } + // Now, as the GET's own clock (the attribution windows end at now), in a + // zone that is not UTC, so read_at's conversion is seen. + now := time.Now().In(time.FixedZone("UTC-5", -5*3600)) + body, err := s.CCUsagePushBody(context.Background(), 1, "example", now) + if err != nil { + t.Fatal(err) + } + var api, push map[string]json.RawMessage + if err := json.Unmarshal(rec.Body.Bytes(), &api); err != nil { + t.Fatal(err) + } + if err := json.Unmarshal(body, &push); err != nil { + t.Fatal(err) + } + if len(push) != 3 || string(push["source_instance"]) != `"example"` || string(push["read_at"]) != `"`+now.UTC().Format(time.RFC3339)+`"` { + t.Fatalf("push envelope: %s", body) + } + // Positive control: the fixture's attribution is in the answer, so the + // comparison below covers used_by and not only empty lists. + if !strings.Contains(string(api["accounts"]), `"requested_by":"pr-review"`) || !strings.Contains(string(api["accounts"]), `"seven_day_opus"`) { + t.Fatalf("the fixture's answer lacks its attribution or windows: %s", api["accounts"]) + } + if string(push["accounts"]) != string(api["accounts"]) { + t.Fatalf("the pushed accounts differ from the API's:\npush %s\napi %s", push["accounts"], api["accounts"]) + } + if store.owners[len(store.owners)-1] != 1 { + t.Fatalf("the push read owner %d, want the admin (1)", store.owners[len(store.owners)-1]) + } +} + +// TestTheUsagePushRefusesWithoutAStore: no database, no body (the push is +// counted as a build failure, never sent empty). +func TestTheUsagePushRefusesWithoutAStore(t *testing.T) { + s := New(nil, AuthConfig{HookToken: tokTestHook}, log.New(io.Discard, "", 0)) + s.UseExtensions(Extensions{Users: stubUsers{}}) + if b, err := s.CCUsagePushBody(context.Background(), 1, "example", time.Now()); err == nil { + t.Fatalf("built %s with no store", b) + } +} diff --git a/internal/ccusage/poller.go b/internal/ccusage/poller.go index 6b0f6d753..b20788b23 100644 --- a/internal/ccusage/poller.go +++ b/internal/ccusage/poller.go @@ -260,6 +260,12 @@ type Poller struct { Interval time.Duration Logger *log.Logger Now func() time.Time + // AfterPass, when set, runs once after each pass (Run), whether or not the + // pass read anything: the usage pusher (Pusher.PushAfterPass, + // MUSTER_CC_USAGE_PUSH_URL). It runs in the poll loop, so one cycle makes + // at most one call of it, and a slow one delays the next pass rather than + // overlapping it. + AfterPass func(ctx context.Context) } func (p *Poller) now() time.Time { @@ -290,6 +296,9 @@ func (p *Poller) Run(ctx context.Context) { if err := p.Pass(ctx); err != nil { p.logf("ccusage: pass: %v", err) } + if p.AfterPass != nil && ctx.Err() == nil { + p.AfterPass(ctx) + } select { case <-ctx.Done(): return diff --git a/internal/ccusage/push.go b/internal/ccusage/push.go new file mode 100644 index 000000000..2028cc000 --- /dev/null +++ b/internal/ccusage/push.go @@ -0,0 +1,191 @@ +package ccusage + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "log" + "net" + "net/http" + "net/url" + "strings" + "time" + + "github.com/ZacxDev/muster/internal/metrics" + "github.com/ZacxDev/muster/internal/noredirect" +) + +// PushTimeout bounds one push, connection to answer. +const PushTimeout = 10 * time.Second + +// MaxPushInstance bounds the pushed source_instance label. +const MaxPushInstance = 64 + +// Why a push failed: muster_cc_usage_push_failures_total's `reason`, a closed +// vocabulary (never a URL, a status code or an account). +const ( + PushReasonBuild = "build" + PushReasonTimeout = "timeout" + PushReasonTransport = "transport" + PushReason4xx = "http_4xx" + PushReason5xx = "http_5xx" + PushReasonOther = "http_other" +) + +// CheckPushURL is why u cannot be the push target, or nil. It must be an +// absolute https URL with a host and no userinfo; plain http is allowed only to +// a loopback host (a test's server), since the request carries the bearer. +func CheckPushURL(u string) error { + p, err := url.Parse(u) + if err != nil || p.Host == "" || p.Hostname() == "" { + return errors.New("want an absolute https URL") + } + if p.User != nil { + return errors.New("credentials in the URL are refused: the token goes in its own variable") + } + switch p.Scheme { + case "https": + return nil + case "http": + if isLoopback(p.Hostname()) { + return nil + } + return errors.New("plain http is refused (the request carries the bearer token): want https") + } + return errors.New("want an absolute https URL") +} + +func isLoopback(host string) bool { + if host == "localhost" { + return true + } + ip := net.ParseIP(host) + return ip != nil && ip.IsLoopback() +} + +// Pusher PUTs the readings to an external dashboard once per poll cycle +// (Poller.AfterPass). It holds no state across pushes: a failed push is logged +// and counted, and the next cycle's push carries the newer readings — there is +// no retry inside a cycle. +// +// 🔴 THE TOKEN IS IN ONE HEADER AND NOWHERE ELSE, and the body is never +// logged: a failure is logged as its class (a status, timeout, transport +// error), never the URL's query, the request or the answer's body. +type Pusher struct { + URL string + Token string + // Body is the request body for a push made at now: the readings exactly as + // GET /api/cc-accounts/usage answers them, with the instance label and + // read_at (api.Server.CCUsagePushBody). + Body func(ctx context.Context, now time.Time) ([]byte, error) + Client *http.Client + Logger *log.Logger + Now func() time.Time + // Timeout bounds one push; 0 is PushTimeout. + Timeout time.Duration +} + +func (p *Pusher) timeout() time.Duration { + if p.Timeout > 0 { + return p.Timeout + } + return PushTimeout +} + +func (p *Pusher) logf(format string, args ...any) { + if p.Logger != nil { + p.Logger.Printf(format, args...) + } +} + +func (p *Pusher) client() *http.Client { + c := p.Client + if c == nil { + c = &http.Client{} + } + // No redirect is followed: one would re-send the bearer elsewhere. + return noredirect.Client(c) +} + +// Push makes one push and reports its failure reason ("" for a 204). It is +// Poller.AfterPass's shape once wrapped (PushAfterPass). +func (p *Pusher) Push(ctx context.Context) string { + now := time.Now() + if p.Now != nil { + now = p.Now() + } + reason := p.push(ctx, now) + if reason != "" { + metrics.CCUsagePushFailures.WithLabelValues(reason).Inc() + } + return reason +} + +func (p *Pusher) push(ctx context.Context, now time.Time) string { + body, err := p.Body(ctx, now) + if err != nil { + p.logf("ccusage: push: the readings could not be assembled: %v", err) + return PushReasonBuild + } + ctx, cancel := context.WithTimeout(ctx, p.timeout()) + defer cancel() + req, err := http.NewRequestWithContext(ctx, http.MethodPut, p.URL, bytes.NewReader(body)) + if err != nil { + p.logf("ccusage: push: build the request") + return PushReasonBuild + } + req.Header.Set("Authorization", "Bearer "+p.Token) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("User-Agent", "muster-ccusage") + resp, err := p.client().Do(req) + if err != nil { + var nr *noredirect.Error + if errors.As(err, &nr) { + p.logf("ccusage: push: the dashboard answered a redirect, which is not followed") + return PushReasonOther + } + var ne interface{ Timeout() bool } + if errors.Is(ctx.Err(), context.DeadlineExceeded) || errors.Is(err, context.DeadlineExceeded) || (errors.As(err, &ne) && ne.Timeout()) { + p.logf("ccusage: push: failed (timeout after %s)", p.timeout()) + return PushReasonTimeout + } + p.logf("ccusage: push: failed (transport error)") + return PushReasonTransport + } + _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 64<<10)) + _ = resp.Body.Close() + switch { + case resp.StatusCode == http.StatusNoContent: + return "" + case resp.StatusCode >= 400 && resp.StatusCode < 500: + p.logf("ccusage: push: the dashboard answered HTTP %d", resp.StatusCode) + return PushReason4xx + case resp.StatusCode >= 500 && resp.StatusCode < 600: + p.logf("ccusage: push: the dashboard answered HTTP %d", resp.StatusCode) + return PushReason5xx + default: + p.logf("ccusage: push: the dashboard answered HTTP %d, want 204", resp.StatusCode) + return PushReasonOther + } +} + +// PushAfterPass is Push as a Poller.AfterPass. +func (p *Pusher) PushAfterPass(ctx context.Context) { p.Push(ctx) } + +// PushInstance is the source_instance label: explicit (≤ MaxPushInstance +// characters, refused otherwise), else the public URL's host when it fits, +// else "muster". +func PushInstance(explicit, publicURL string) (string, error) { + if v := strings.TrimSpace(explicit); v != "" { + if len(v) > MaxPushInstance { + return "", fmt.Errorf("over %d characters", MaxPushInstance) + } + return v, nil + } + if u, err := url.Parse(strings.TrimSpace(publicURL)); err == nil && u.Hostname() != "" && len(u.Hostname()) <= MaxPushInstance { + return u.Hostname(), nil + } + return "muster", nil +} diff --git a/internal/ccusage/push_test.go b/internal/ccusage/push_test.go new file mode 100644 index 000000000..f4b2a8794 --- /dev/null +++ b/internal/ccusage/push_test.go @@ -0,0 +1,263 @@ +package ccusage + +import ( + "bytes" + "context" + "errors" + "io" + "log" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "github.com/prometheus/client_golang/prometheus/testutil" + + "github.com/ZacxDev/muster/internal/metrics" +) + +const ( + pushTestToken = "push-test-token-SENTINEL-7f3a9c" + pushTestBody = `{"source_instance":"example","read_at":"2026-11-10T18:00:00Z","accounts":[{"account":"BODY-SENTINEL-alpha"}]}` +) + +var pushReasons = []string{PushReasonBuild, PushReasonTimeout, PushReasonTransport, PushReason4xx, PushReason5xx, PushReasonOther} + +func pushCounts() map[string]float64 { + out := map[string]float64{} + for _, r := range pushReasons { + out[r] = testutil.ToFloat64(metrics.CCUsagePushFailures.WithLabelValues(r)) + } + return out +} + +// pushTarget is a dashboard answering status (after delay), recording each +// request. Its answer body echoes the token, so a pusher that logged the +// answer would leak it. +type pushTarget struct { + mu sync.Mutex + status int + delay time.Duration + reqs []*http.Request + bodies [][]byte +} + +func (p *pushTarget) ServeHTTP(w http.ResponseWriter, r *http.Request) { + b, _ := io.ReadAll(r.Body) + p.mu.Lock() + p.reqs, p.bodies = append(p.reqs, r), append(p.bodies, b) + p.mu.Unlock() + if p.delay > 0 { + time.Sleep(p.delay) + } + if p.status == http.StatusFound { + http.Redirect(w, r, "http://127.0.0.1:1/elsewhere", http.StatusFound) + return + } + w.WriteHeader(p.status) + _, _ = w.Write([]byte("echo " + r.Header.Get("Authorization"))) +} + +func (p *pushTarget) count() int { + p.mu.Lock() + defer p.mu.Unlock() + return len(p.reqs) +} + +func newPusher(url string, logs io.Writer) *Pusher { + return &Pusher{URL: url, Token: pushTestToken, Logger: log.New(logs, "", 0), Timeout: 200 * time.Millisecond, + Now: func() time.Time { return time.Date(2026, 11, 10, 18, 0, 0, 0, time.UTC) }, + Body: func(context.Context, time.Time) ([]byte, error) { return []byte(pushTestBody), nil }} +} + +// TestAPushIsOnePUTWithTheBearerAndTheBody: a 204 is success (nothing +// counted); the request is one PUT carrying the bearer and the body as built. +func TestAPushIsOnePUTWithTheBearerAndTheBody(t *testing.T) { + tgt := &pushTarget{status: http.StatusNoContent} + srv := httptest.NewServer(tgt) + defer srv.Close() + before := pushCounts() + var logs bytes.Buffer + if r := newPusher(srv.URL+"/api/usage", &logs).Push(context.Background()); r != "" { + t.Fatalf("a 204 reported %q", r) + } + if tgt.count() != 1 { + t.Fatalf("%d requests, want 1", tgt.count()) + } + req := tgt.reqs[0] + if req.Method != http.MethodPut || req.URL.Path != "/api/usage" || req.Header.Get("Authorization") != "Bearer "+pushTestToken || + req.Header.Get("Content-Type") != "application/json" || string(tgt.bodies[0]) != pushTestBody { + t.Fatalf("request %s %s auth=%q ct=%q body=%s", req.Method, req.URL.Path, req.Header.Get("Authorization"), + req.Header.Get("Content-Type"), tgt.bodies[0]) + } + after := pushCounts() + for _, r := range pushReasons { + if after[r] != before[r] { + t.Fatalf("a 204 counted a %s failure", r) + } + } +} + +// TestAFailedPushIsCountedByReasonAndNotRetried: each failure is one request +// (no retry inside the cycle), one count under exactly its reason, and a log +// line that carries neither the token nor the body — not even when the +// dashboard's answer echoes the token. +func TestAFailedPushIsCountedByReasonAndNotRetried(t *testing.T) { + cases := []struct { + name string + status int + delay time.Duration + want string + reqs int + }{ + {"401", http.StatusUnauthorized, 0, PushReason4xx, 1}, + {"422", http.StatusUnprocessableEntity, 0, PushReason4xx, 1}, + {"500", http.StatusInternalServerError, 0, PushReason5xx, 1}, + {"503", http.StatusServiceUnavailable, 0, PushReason5xx, 1}, + {"200 is not 204", http.StatusOK, 0, PushReasonOther, 1}, + {"a redirect is not followed", http.StatusFound, 0, PushReasonOther, 1}, + {"no answer in time", http.StatusNoContent, time.Second, PushReasonTimeout, 1}, + } + var logs bytes.Buffer + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + tgt := &pushTarget{status: c.status, delay: c.delay} + srv := httptest.NewServer(tgt) + defer srv.Close() + before := pushCounts() + if r := newPusher(srv.URL, &logs).Push(context.Background()); r != c.want { + t.Fatalf("reason %q, want %q", r, c.want) + } + if n := tgt.count(); n != c.reqs { + t.Fatalf("%d requests, want %d", n, c.reqs) + } + after := pushCounts() + for _, r := range pushReasons { + want := before[r] + if r == c.want { + want++ + } + if after[r] != want { + t.Errorf("%s: count %v → %v, want %v", r, before[r], after[r], want) + } + } + }) + } + + t.Run("transport", func(t *testing.T) { + srv := httptest.NewServer(http.NotFoundHandler()) + u := srv.URL + srv.Close() // nothing listens there now + if r := newPusher(u, &logs).Push(context.Background()); r != PushReasonTransport { + t.Fatalf("reason %q, want %q", r, PushReasonTransport) + } + }) + t.Run("build", func(t *testing.T) { + tgt := &pushTarget{status: http.StatusNoContent} + srv := httptest.NewServer(tgt) + defer srv.Close() + p := newPusher(srv.URL, &logs) + p.Body = func(context.Context, time.Time) ([]byte, error) { return nil, errors.New("store down") } + if r := p.Push(context.Background()); r != PushReasonBuild || tgt.count() != 0 { + t.Fatalf("reason %q with %d requests, want %q and none", r, tgt.count(), PushReasonBuild) + } + }) + + out := logs.String() + if !strings.Contains(out, "HTTP 401") || !strings.Contains(out, "timeout") { + t.Fatalf("positive control: the failures were not logged at all: %s", out) + } + for _, secret := range []string{pushTestToken, "BODY-SENTINEL"} { + if strings.Contains(out, secret) { + t.Fatalf("the log carries %q:\n%s", secret, out) + } + } +} + +// TestCheckPushURL: https with a host; http only to loopback; no userinfo. +func TestCheckPushURL(t *testing.T) { + for _, ok := range []string{"https://dash.example.com/api/v1/usage", "https://dash.example.com:8443/x", + "http://127.0.0.1:8080/x", "http://localhost/x", "http://[::1]:9/x"} { + if err := CheckPushURL(ok); err != nil { + t.Errorf("%s: %v", ok, err) + } + } + for _, bad := range []string{"http://dash.example.com/x", "http://192.0.2.1/x", "dash.example.com/x", "/api/usage", + "https://", "ftp://dash.example.com/x", "https://user:pw@dash.example.com/x", "http://localhost.example.com/x", "::"} { + if err := CheckPushURL(bad); err == nil { + t.Errorf("%s: accepted", bad) + } + } +} + +// TestPushInstance: explicit wins (≤64 characters), else the public URL's +// host, else "muster". +func TestPushInstance(t *testing.T) { + for _, c := range []struct{ explicit, public, want string }{ + {"prod-east", "https://muster.example.com", "prod-east"}, + {strings.Repeat("a", 64), "", strings.Repeat("a", 64)}, + {"", "https://muster.example.com:8443/base", "muster.example.com"}, + {"", "", "muster"}, + {"", "https://" + strings.Repeat("h", 65) + ".example.com", "muster"}, + } { + if got, err := PushInstance(c.explicit, c.public); err != nil || got != c.want { + t.Errorf("PushInstance(%q, %q) = %q, %v; want %q", c.explicit, c.public, got, err, c.want) + } + } + if _, err := PushInstance(strings.Repeat("a", 65), ""); err == nil { + t.Fatal("a 65-character instance was accepted") + } +} + +// countingSource is a Source with no accounts that counts its reads: one per +// pass. +type countingSource struct { + mu sync.Mutex + n int +} + +func (c *countingSource) Accounts(context.Context, int64) (map[string]string, error) { + c.mu.Lock() + defer c.mu.Unlock() + c.n++ + return map[string]string{}, nil +} + +// TestRunCallsAfterPassOncePerPass: the poll loop makes exactly one AfterPass +// call per pass — never two for one pass, never one without a pass. +func TestRunCallsAfterPassOncePerPass(t *testing.T) { + src := &countingSource{} + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + var mu sync.Mutex + var seen []int // the pass count at each AfterPass + p := &Poller{Source: src, Store: &memStore{}, Owner: 1, Interval: 5 * time.Millisecond, + AfterPass: func(context.Context) { + src.mu.Lock() + n := src.n + src.mu.Unlock() + mu.Lock() + seen = append(seen, n) + if len(seen) == 4 { + cancel() + } + mu.Unlock() + }} + done := make(chan struct{}) + go func() { p.Run(ctx); close(done) }() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("Run did not stop") + } + if len(seen) != 4 || src.n != 4 { + t.Fatalf("%d AfterPass calls over %d passes, want 4 and 4", len(seen), src.n) + } + for i, n := range seen { + if n != i+1 { + t.Fatalf("AfterPass call %d came after pass %d (calls %v), want one per pass", i+1, n, seen) + } + } +} diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index 7a75a9276..833a7f88f 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -312,6 +312,21 @@ var ( Name: "muster_memo_relay_deliveries_total", Help: "Memo relay: delivery attempts to agents, by result ONLY — never an agent, scope or memo id.", }, []string{"result"}) + + // CCUsagePushFailures counts the Claude account usage pusher's failed + // pushes (MUSTER_CC_USAGE_PUSH_URL; internal/ccusage push.go), by reason: + // build — the readings could not be assembled (the store did not answer) + // timeout — no answer inside the push timeout + // transport — any other failure to send (DNS, refused connection, TLS) + // http_4xx — the dashboard answered 4xx (a bad token is 401/403) + // http_5xx — the dashboard answered 5xx + // http_other — any other status but 204 (a 200, a refused redirect) + // 🔴 `reason` is that closed vocabulary (ccusage.PushReason*): never a URL, + // a status code or an account name. + CCUsagePushFailures = promauto.NewCounterVec(prometheus.CounterOpts{ + Name: "muster_cc_usage_push_failures_total", + Help: "Claude account usage pushes that failed, by reason (build|timeout|transport|http_4xx|http_5xx|http_other).", + }, []string{"reason"}) ) // ---- opening a session (docs/telemetry.md) -------------------------------- diff --git a/internal/metrics/metrics_test.go b/internal/metrics/metrics_test.go index c9308c244..22e6a65f1 100644 --- a/internal/metrics/metrics_test.go +++ b/internal/metrics/metrics_test.go @@ -80,6 +80,7 @@ var wantMetrics = map[string][]string{ "muster_memo_relay_agent_checks_total": {"result"}, "muster_memo_relay_upstream_total": {"result"}, "muster_memo_relay_deliveries_total": {"result"}, + "muster_cc_usage_push_failures_total": {"reason"}, } // 🔴 LABELS THAT MAY NEVER APPEAR, BY NAME. This is the denylist half: the @@ -118,6 +119,7 @@ func gather(t *testing.T) map[string][]string { MemoRelayAgentChecks.WithLabelValues("eligible") MemoRelayUpstream.WithLabelValues("ok") MemoRelayDeliveries.WithLabelValues("delivered") + CCUsagePushFailures.WithLabelValues("timeout") OfficerTranscriptEvents.WithLabelValues("posted") OfficerLLMRequests.WithLabelValues("example:area", "1", "ok") OfficerLLMRefused.WithLabelValues("none", "path") diff --git a/tests/features_config_test.py b/tests/features_config_test.py index 98ef7aded..adcfb55f2 100644 --- a/tests/features_config_test.py +++ b/tests/features_config_test.py @@ -61,6 +61,9 @@ "MUSTER_CC_USAGE_POLL": "opt-in: the Claude account usage poller calls the provider's usage endpoint with every pooled " "account's token, so the Compose file leaves it off rather than forwarding a switch nobody set", "MUSTER_CC_USAGE_POLL_INTERVAL": "the usage poller's interval; refused without MUSTER_CC_USAGE_POLL, which the Compose file does not forward", + **{name: "the usage pusher; refused without MUSTER_CC_USAGE_POLL, which the Compose file does not forward" for name in [ + "MUSTER_CC_USAGE_PUSH_URL", "MUSTER_CC_USAGE_PUSH_TOKEN", "MUSTER_CC_USAGE_PUSH_INSTANCE", + ]}, "MUSTER_TEST_LLM_UPSTREAM": "test only: the Compose smoke test's fake model; only tests/compose-chief.override.yml sets it", "MUSTER_CHIEF_SPEC_DIR": "the Compose file mounts the chief's spec volume at the runtime's default, /run/muster-spec", "MUSTER_CHIEF_LLM": "the Kubernetes chief's switch onto the LLM proxy; muster refuses it beside MUSTER_CHIEF_RUNTIME=compose, " From e58c882fafa0f1ab6ccdcc73b497090fe8bf2498 Mon Sep 17 00:00:00 2001 From: Zachary Lowden Date: Tue, 10 Nov 2026 20:49:02 -0600 Subject: [PATCH 2/3] ccusage push: carry poll_interval_seconds; ledger the push variables The push body gains the API's own top-level poll_interval_seconds (the poller's configured interval in whole seconds), so the dashboard can mark a source stale at a multiple of it. accounts stays byte-for-byte the API's. The golden pins it equal to the API's and to 420 / 0 for a 7m / off interval (three mutants: hardcoded, omitted, minutes; all killed). Fixes CI's two ledger failures from the first push: the three MUSTER_CC_USAGE_PUSH_* variables join the Claude Code agents feature's settings and the banner exemptions (each reason read off the code). Co-Authored-By: Claude Opus 5.5 (1M context) --- cmd/muster-server/banner_test.go | 4 ++++ cmd/muster-server/features.go | 2 +- docs/jobs.md | 5 ++++- internal/api/cc_usage.go | 12 ++++++++---- internal/api/cc_usage_test.go | 19 +++++++++++++++++-- 5 files changed, 34 insertions(+), 8 deletions(-) diff --git a/cmd/muster-server/banner_test.go b/cmd/muster-server/banner_test.go index a1e6a32ac..9d736f4e1 100644 --- a/cmd/muster-server/banner_test.go +++ b/cmd/muster-server/banner_test.go @@ -474,6 +474,10 @@ var bannerLedger = []string{ // // ⚠ EACH REASON BELOW WAS READ OFF THE CODE, NOT INFERRED FROM THE GROUPING. var bannerExempt = map[string]string{ + envCCUsagePushURL: "printed on its own `claude account usage push: ON/OFF` line, which is drawn only under the poller's ON " + + "line; refused at boot with the poller off, whose OFF line names " + envCCUsagePoll, + envCCUsagePushToken: "a secret, never printed; refused at boot without " + envCCUsagePushURL + ", whose state the push line prints", + envCCUsagePushInstance: "printed with its value on the `claude account usage push: ON` line; refused at boot without " + envCCUsagePushURL, envCCUsagePollInterval: "printed with its value on the `claude account usage: ON` line; refused at boot with the poller off, " + "whose OFF line names " + envCCUsagePoll, envAskLogTextTTL: "printed with its value on the `ask log: ON` line; it means nothing with the log off, whose OFF line names " + envAskLog, diff --git a/cmd/muster-server/features.go b/cmd/muster-server/features.go index 924a636b8..43c04e680 100644 --- a/cmd/muster-server/features.go +++ b/cmd/muster-server/features.go @@ -102,7 +102,7 @@ func featureCatalog() []features.Feature { v(envAgentCCAccounts, "account names"), v(envAgentCCTokenPrefix+"", "claude setup-token")}, Settings: []string{envAgentCCStorage, envAgentCCCooldown, envAgentCCMaxJobs, envAgentCCFailover, envAgentCCBypassPools, envAgentCCPoolDeployKeys, envAgentCCEgressLockedPools, envAgentNetpolFromNS, envAgentNetpolFromLabels, envCheckoutWaitTimeout, - envChiefLLM, envCCUsagePoll, envCCUsagePollInterval}}, + envChiefLLM, envCCUsagePoll, envCCUsagePollInterval, envCCUsagePushURL, envCCUsagePushToken, envCCUsagePushInstance}}, // The documented install: the Compose stack's `chief` profile and the // Compose chief runtime (composechief.go). A Kubernetes deployment runs // it through MUSTER_AGENT_PROVISIONER instead (docs/chief.md). diff --git a/docs/jobs.md b/docs/jobs.md index 8a5ccdb2f..c54d2ea3f 100644 --- a/docs/jobs.md +++ b/docs/jobs.md @@ -748,6 +748,7 @@ Content-Type: application/json {"source_instance": "muster.example.com", "read_at": "2026-11-10T18:21:00Z", + "poll_interval_seconds": 900, "accounts": [ … ]} ``` @@ -756,7 +757,9 @@ Content-Type: application/json built by the same code. - `source_instance` is `MUSTER_CC_USAGE_PUSH_INSTANCE` (at most 64 characters), else `MUSTER_PUBLIC_URL`'s host, else `muster`. `read_at` is - when the push was built. + when the push was built. `poll_interval_seconds` is the poller's + configured interval in whole seconds, as the API's own field: a dashboard can + mark the source stale once a few intervals pass with no push. - The push is made after **every** cycle, whether or not a reading changed, so a dashboard can tell a quiet muster from a dead one. - The dashboard answers **`204`**. Anything else — a 4xx, a 5xx, another 2xx, diff --git a/internal/api/cc_usage.go b/internal/api/cc_usage.go index 5f6d2e0c7..0eff901f4 100644 --- a/internal/api/cc_usage.go +++ b/internal/api/cc_usage.go @@ -191,9 +191,12 @@ func (s *Server) officerPoolOf(name string) (string, string, bool, bool) { // accounts exactly as GET /api/cc-accounts/usage answers the admin (the same // ccUsageAccounts, so the two cannot drift), under the instance's label. type ccUsagePush struct { - SourceInstance string `json:"source_instance"` - ReadAt string `json:"read_at"` - Accounts []ccUsageAccount `json:"accounts"` + SourceInstance string `json:"source_instance"` + ReadAt string `json:"read_at"` + // PollIntervalSeconds is the poller's interval, as the API's own field: + // the dashboard marks the source stale at a multiple of it. + PollIntervalSeconds int `json:"poll_interval_seconds"` + Accounts []ccUsageAccount `json:"accounts"` } // CCUsagePushBody is the usage pusher's request body at now for owner (the @@ -207,7 +210,8 @@ func (s *Server) CCUsagePushBody(ctx context.Context, owner int64, instance stri if err != nil { return nil, err } - return json.Marshal(ccUsagePush{SourceInstance: instance, ReadAt: now.UTC().Format(time.RFC3339), Accounts: accts}) + return json.Marshal(ccUsagePush{SourceInstance: instance, ReadAt: now.UTC().Format(time.RFC3339), + PollIntervalSeconds: int(s.ext.CCUsageInterval / time.Second), Accounts: accts}) } func (s *Server) handleCCAccountUsage(w http.ResponseWriter, r *http.Request) { diff --git a/internal/api/cc_usage_test.go b/internal/api/cc_usage_test.go index be9dc2ef6..349779bbe 100644 --- a/internal/api/cc_usage_test.go +++ b/internal/api/cc_usage_test.go @@ -248,7 +248,7 @@ func TestAProxiedOfficerInstanceIsNotCreditedToItsAccount(t *testing.T) { // TestTheUsagePushCarriesTheAPIsAccountsByteForByte is the golden: for the // same readings, the pushed body's `accounts` is byte-for-byte the `accounts` // GET /api/cc-accounts/usage answers the admin (used_by included), beside -// exactly source_instance and read_at. +// exactly source_instance, read_at and the API's own poll_interval_seconds. func TestTheUsagePushCarriesTheAPIsAccountsByteForByte(t *testing.T) { h, _, store := newCCUsageFixture(t) s := New(nil, AuthConfig{HookToken: tokTestHook, ServiceToken: ccUsageServiceToken}, log.New(io.Discard, "", 0)) @@ -271,7 +271,8 @@ func TestTheUsagePushCarriesTheAPIsAccountsByteForByte(t *testing.T) { if err := json.Unmarshal(body, &push); err != nil { t.Fatal(err) } - if len(push) != 3 || string(push["source_instance"]) != `"example"` || string(push["read_at"]) != `"`+now.UTC().Format(time.RFC3339)+`"` { + if len(push) != 4 || string(push["poll_interval_seconds"]) != "300" || string(push["poll_interval_seconds"]) != string(api["poll_interval_seconds"]) || + string(push["source_instance"]) != `"example"` || string(push["read_at"]) != `"`+now.UTC().Format(time.RFC3339)+`"` { t.Fatalf("push envelope: %s", body) } // Positive control: the fixture's attribution is in the answer, so the @@ -282,6 +283,20 @@ func TestTheUsagePushCarriesTheAPIsAccountsByteForByte(t *testing.T) { if string(push["accounts"]) != string(api["accounts"]) { t.Fatalf("the pushed accounts differ from the API's:\npush %s\napi %s", push["accounts"], api["accounts"]) } + // A second interval (one no constant equals): the field follows the + // configured interval, and is 0 with the poller off, as the API's is. + for _, c := range []struct { + iv time.Duration + want string + }{{7 * time.Minute, "420"}, {0, "0"}} { + s2 := New(nil, AuthConfig{HookToken: tokTestHook}, log.New(io.Discard, "", 0)) + s2.UseExtensions(Extensions{Users: stubUsers{}, CCUsage: store, CCUsageInterval: c.iv}) + b, err := s2.CCUsagePushBody(context.Background(), 1, "example", now) + var m map[string]json.RawMessage + if err != nil || json.Unmarshal(b, &m) != nil || string(m["poll_interval_seconds"]) != c.want { + t.Fatalf("interval %s: poll_interval_seconds %s (%v), want %s", c.iv, m["poll_interval_seconds"], err, c.want) + } + } if store.owners[len(store.owners)-1] != 1 { t.Fatalf("the push read owner %d, want the admin (1)", store.owners[len(store.owners)-1]) } From d1d6b87fbe189f77a6a1ba4c9130271babe6d900 Mon Sep 17 00:00:00 2001 From: Zachary Lowden Date: Tue, 10 Nov 2026 21:44:19 -0600 Subject: [PATCH 3/3] ccusage push: log a 4xx answer's reason; count shutdown as canceled From the #587 audit: a dashboard refusing the push (a cap exceeded) said only "HTTP 400" every cycle. On a 4xx the first 512 bytes of its answer are now logged as one line: control characters made spaces, cut at 200 runes, and the token redacted should the dashboard echo it. The request body is still never logged. A push cut short by muster shutting down (the parent context ended) is now counted as reason "canceled", not "timeout" or "transport". Tests: the 400 answer appears as one sanitized line (with the token redacted, cut on a rune boundary), and a shutdown counts as canceled. Mutants: no redaction, no control-character strip, no cut, a cut by bytes, the answer not logged, the token logged, canceled reported as timeout, and no canceled branch. Every one was killed. Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/configuration.md | 2 +- docs/jobs.md | 10 ++++-- internal/ccusage/push.go | 50 +++++++++++++++++++++++++--- internal/ccusage/push_test.go | 62 ++++++++++++++++++++++++++++++++++- internal/metrics/metrics.go | 3 +- 5 files changed, 117 insertions(+), 10 deletions(-) diff --git a/docs/configuration.md b/docs/configuration.md index 2e9657daf..64ea349d5 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -98,7 +98,7 @@ See [auth.md](auth.md). | `MUSTER_ASK_LOG_CLASSIFY` | off | no | `1` runs the stall classifier on a plain-text question the [ask log](asks.md) kept, once per turn end (reusing the stall sweep's answer for the streak when it has one), on the owner's OpenRouter key; its scores are stored on the row. Needs `MUSTER_CHIEF_NUDGES=1` (the classifier's configuration). Off by default so the pre-filter's volume (`muster_ask_plain_text_prefilter_total`) is read before the classifier is spent on it. | | `MUSTER_CC_USAGE_POLL` | off | no | `1` reads each of the admin's pooled Claude accounts' subscription utilization once per `MUSTER_CC_USAGE_POLL_INTERVAL` and keeps the latest reading per account (numbers only): from the endpoint Claude Code's `/usage` reads when the account's token may use it (one discovery call per token; a `setup-token` is refused there), else from a one-token Messages probe's `anthropic-ratelimit-unified-*` headers, which spends one output token of the account's quota per read. An account the pool marks rate-limited is not probed until its cooldown ends. Served at `GET /api/cc-accounts/usage` and on Account → Credentials ([jobs](jobs.md#claude-account-usage)). muster-server itself calls `api.anthropic.com:443`. Needs the `claude-code` kind and a database, or boot is refused. Members' own accounts are not read. | | `MUSTER_CC_USAGE_POLL_INTERVAL` | `15m` | no | How often the usage poller reads every account (a Go duration, 5m–24h). A probe answered 429 waits longer (doubling, at least `Retry-After`, at most 6h — or until the window's reset when its status is `rejected`); a token the probe refused (401/403) is not read again until it changes. Refused without `MUSTER_CC_USAGE_POLL`. | -| `MUSTER_CC_USAGE_PUSH_URL` | none | no | Turns on the [usage pusher](jobs.md#pushing-the-readings): after each poll cycle muster `PUT`s the readings, exactly as `GET /api/cc-accounts/usage` answers the admin, to this URL, once, with a 10-second timeout, and expects `204`. A failure is logged (never the token or the body), counted in `muster_cc_usage_push_failures_total{reason}` and not retried until the next cycle; a redirect is not followed. An absolute `https` URL (plain `http` only to a loopback host); credentials in the URL are refused. Refused without `MUSTER_CC_USAGE_POLL` or without `MUSTER_CC_USAGE_PUSH_TOKEN`. | +| `MUSTER_CC_USAGE_PUSH_URL` | none | no | Turns on the [usage pusher](jobs.md#pushing-the-readings): after each poll cycle muster `PUT`s the readings, exactly as `GET /api/cc-accounts/usage` answers the admin, to this URL, once, with a 10-second timeout, and expects `204`. A failure is logged (never the token or the body; a 4xx's answer as one sanitized line), counted in `muster_cc_usage_push_failures_total{reason}` and not retried until the next cycle; a redirect is not followed. An absolute `https` URL (plain `http` only to a loopback host); credentials in the URL are refused. Refused without `MUSTER_CC_USAGE_POLL` or without `MUSTER_CC_USAGE_PUSH_TOKEN`. | | `MUSTER_CC_USAGE_PUSH_TOKEN` | none | with the push URL | The pusher's `Authorization: Bearer` token. Never logged. Refused without `MUSTER_CC_USAGE_PUSH_URL`. | | `MUSTER_CC_USAGE_PUSH_INSTANCE` | `MUSTER_PUBLIC_URL`'s host, else `muster` | no | The pushed `source_instance` label, at most 64 characters (a public host longer than that falls back to `muster`). Refused without `MUSTER_CC_USAGE_PUSH_URL`. | | `MUSTER_SESSION_STATE` | off | no | `1` runs session state's pass (design: `docs/design/session-state.md`, #425): every 30 seconds each owner's sessions get a state, and every stop an episode in `session_episodes`. The stall classifier runs only where the chief's nudges would run it. It also replaces the waiting-window push and the chief's stall notice with **one push per stop** (an episode), naming the session and never quoting it, and draws the open stops as the Asks page's **Stalled / waiting** section ([attention.md](attention.md#stalled--waiting-session-state)). Off, every push is as before. Needs a database. | diff --git a/docs/jobs.md b/docs/jobs.md index c54d2ea3f..5137d7723 100644 --- a/docs/jobs.md +++ b/docs/jobs.md @@ -765,9 +765,13 @@ Content-Type: application/json - The dashboard answers **`204`**. Anything else — a 4xx, a 5xx, another 2xx, a redirect (never followed: it would re-send the bearer), no answer in 10 seconds, a transport error, or readings the store could not give — is logged - (the status or the error's class only: never the token, the body or the - answer) and counted in `muster_cc_usage_push_failures_total{reason}` - (`http_4xx`, `http_5xx`, `http_other`, `timeout`, `transport`, `build`). It + (the status or the error's class only, never the token or the body; on a + 4xx, also the first 512 bytes of the dashboard's answer as one line, control + characters made spaces, at most 200 characters, the token redacted, so a + refusal such as a cap exceeded says why) and counted in + `muster_cc_usage_push_failures_total{reason}` (`http_4xx`, `http_5xx`, + `http_other`, `timeout`, `transport`, `build`, or `canceled` when muster + was shutting down mid-push). It is not retried: the next cycle's push carries newer readings. - Boot is refused when the URL is set with the poller off, without the token, or not `https` (plain `http` is allowed only to a loopback host, for tests), diff --git a/internal/ccusage/push.go b/internal/ccusage/push.go index 2028cc000..0fa91a9fc 100644 --- a/internal/ccusage/push.go +++ b/internal/ccusage/push.go @@ -12,6 +12,7 @@ import ( "net/url" "strings" "time" + "unicode" "github.com/ZacxDev/muster/internal/metrics" "github.com/ZacxDev/muster/internal/noredirect" @@ -32,6 +33,8 @@ const ( PushReason4xx = "http_4xx" PushReason5xx = "http_5xx" PushReasonOther = "http_other" + // PushReasonCanceled: muster was shutting down mid-push. + PushReasonCanceled = "canceled" ) // CheckPushURL is why u cannot be the push target, or nil. It must be an @@ -123,13 +126,13 @@ func (p *Pusher) Push(ctx context.Context) string { return reason } -func (p *Pusher) push(ctx context.Context, now time.Time) string { - body, err := p.Body(ctx, now) +func (p *Pusher) push(parent context.Context, now time.Time) string { + body, err := p.Body(parent, now) if err != nil { p.logf("ccusage: push: the readings could not be assembled: %v", err) return PushReasonBuild } - ctx, cancel := context.WithTimeout(ctx, p.timeout()) + ctx, cancel := context.WithTimeout(parent, p.timeout()) defer cancel() req, err := http.NewRequestWithContext(ctx, http.MethodPut, p.URL, bytes.NewReader(body)) if err != nil { @@ -146,6 +149,11 @@ func (p *Pusher) push(ctx context.Context, now time.Time) string { p.logf("ccusage: push: the dashboard answered a redirect, which is not followed") return PushReasonOther } + if parent.Err() != nil { + // muster is shutting down: not the dashboard's fault, nor a timeout. + p.logf("ccusage: push: canceled (shutting down)") + return PushReasonCanceled + } var ne interface{ Timeout() bool } if errors.Is(ctx.Err(), context.DeadlineExceeded) || errors.Is(err, context.DeadlineExceeded) || (errors.As(err, &ne) && ne.Timeout()) { p.logf("ccusage: push: failed (timeout after %s)", p.timeout()) @@ -154,13 +162,16 @@ func (p *Pusher) push(ctx context.Context, now time.Time) string { p.logf("ccusage: push: failed (transport error)") return PushReasonTransport } + answer, _ := io.ReadAll(io.LimitReader(resp.Body, maxPushAnswer)) _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 64<<10)) _ = resp.Body.Close() switch { case resp.StatusCode == http.StatusNoContent: return "" case resp.StatusCode >= 400 && resp.StatusCode < 500: - p.logf("ccusage: push: the dashboard answered HTTP %d", resp.StatusCode) + // A 4xx says why (a cap exceeded, a bad token): its answer is logged as + // one sanitized line, the token redacted should the dashboard echo it. + p.logf("ccusage: push: the dashboard answered HTTP %d: %s", resp.StatusCode, sanitizeAnswer(answer, p.Token)) return PushReason4xx case resp.StatusCode >= 500 && resp.StatusCode < 600: p.logf("ccusage: push: the dashboard answered HTTP %d", resp.StatusCode) @@ -171,6 +182,37 @@ func (p *Pusher) push(ctx context.Context, now time.Time) string { } } +// maxPushAnswer and maxPushAnswerRunes bound the logged 4xx answer: the +// bytes read, and the runes kept after sanitizing. +const ( + maxPushAnswer = 512 + maxPushAnswerRunes = 200 +) + +// sanitizeAnswer is a dashboard's answer as one log line: token redacted, +// control characters (newlines included) made spaces, at most +// maxPushAnswerRunes runes ("(empty)" for none). +func sanitizeAnswer(b []byte, token string) string { + s := strings.ToValidUTF8(string(b), "?") + if token != "" { + s = strings.ReplaceAll(s, token, "[redacted]") + } + s = strings.Map(func(r rune) rune { + if unicode.IsControl(r) { + return ' ' + } + return r + }, s) + s = strings.TrimSpace(s) + if r := []rune(s); len(r) > maxPushAnswerRunes { + s = string(r[:maxPushAnswerRunes]) + "…" + } + if s == "" { + return "(empty)" + } + return s +} + // PushAfterPass is Push as a Poller.AfterPass. func (p *Pusher) PushAfterPass(ctx context.Context) { p.Push(ctx) } diff --git a/internal/ccusage/push_test.go b/internal/ccusage/push_test.go index f4b2a8794..808f6e463 100644 --- a/internal/ccusage/push_test.go +++ b/internal/ccusage/push_test.go @@ -23,7 +23,7 @@ const ( pushTestBody = `{"source_instance":"example","read_at":"2026-11-10T18:00:00Z","accounts":[{"account":"BODY-SENTINEL-alpha"}]}` ) -var pushReasons = []string{PushReasonBuild, PushReasonTimeout, PushReasonTransport, PushReason4xx, PushReason5xx, PushReasonOther} +var pushReasons = []string{PushReasonBuild, PushReasonTimeout, PushReasonTransport, PushReason4xx, PushReason5xx, PushReasonOther, PushReasonCanceled} func pushCounts() map[string]float64 { out := map[string]float64{} @@ -40,6 +40,7 @@ type pushTarget struct { mu sync.Mutex status int delay time.Duration + answer string // the answer body; "" echoes the Authorization header reqs []*http.Request bodies [][]byte } @@ -57,6 +58,10 @@ func (p *pushTarget) ServeHTTP(w http.ResponseWriter, r *http.Request) { return } w.WriteHeader(p.status) + if p.answer != "" { + _, _ = w.Write([]byte(p.answer)) + return + } _, _ = w.Write([]byte("echo " + r.Header.Get("Authorization"))) } @@ -261,3 +266,58 @@ func TestRunCallsAfterPassOncePerPass(t *testing.T) { } } } + +// TestA4xxAnswerIsLoggedAsOneSanitizedLine: a dashboard's 4xx answer (a cap +// exceeded) reaches the log as one line: its sentence kept, control +// characters made spaces, the token redacted, cut at 200 runes. The request +// body never appears. +func TestA4xxAnswerIsLoggedAsOneSanitizedLine(t *testing.T) { + answer := "too many accounts: 21 > 20\nsecond line\x07 echo Bearer " + pushTestToken + " " + strings.Repeat("é", 400) + tgt := &pushTarget{status: http.StatusBadRequest, answer: answer} + srv := httptest.NewServer(tgt) + defer srv.Close() + var logs bytes.Buffer + if r := newPusher(srv.URL, &logs).Push(context.Background()); r != PushReason4xx { + t.Fatalf("reason %q, want %q", r, PushReason4xx) + } + out := logs.String() + lines := strings.Split(strings.TrimRight(out, "\n"), "\n") + if len(lines) != 1 { + t.Fatalf("%d log lines, want one: %q", len(lines), out) + } + line := lines[0] + if !strings.Contains(line, "HTTP 400: too many accounts: 21 > 20 second line echo Bearer [redacted] é") { + t.Fatalf("the answer is not in the log as one sanitized line: %q", line) + } + for _, secret := range []string{pushTestToken, "BODY-SENTINEL", "\x07"} { + if strings.Contains(line, secret) { + t.Fatalf("the log carries %q: %q", secret, line) + } + } + _, logged, _ := strings.Cut(line, "HTTP 400: ") + if n := len([]rune(logged)); n != maxPushAnswerRunes+1 { // the runes kept, then "…" + t.Fatalf("the logged answer is %d runes, want %d (cut, plus the ellipsis): %q", n, maxPushAnswerRunes+1, logged) + } + if !strings.HasSuffix(logged, "é…") { + t.Fatalf("the cut is not on a rune boundary: %q", logged) + } +} + +// TestAPushCutShortByShutdownIsCanceledNotTimeout: the parent context ending +// mid-push (muster shutting down) counts as canceled, never timeout. +func TestAPushCutShortByShutdownIsCanceledNotTimeout(t *testing.T) { + tgt := &pushTarget{status: http.StatusNoContent, delay: 300 * time.Millisecond} + srv := httptest.NewServer(tgt) + defer srv.Close() + ctx, cancel := context.WithCancel(context.Background()) + time.AfterFunc(20*time.Millisecond, cancel) + p := newPusher(srv.URL, io.Discard) + p.Timeout = 5 * time.Second + before := pushCounts() + if r := p.Push(ctx); r != PushReasonCanceled { + t.Fatalf("reason %q, want %q", r, PushReasonCanceled) + } + if after := pushCounts(); after[PushReasonCanceled] != before[PushReasonCanceled]+1 || after[PushReasonTimeout] != before[PushReasonTimeout] { + t.Fatalf("counts %v → %v, want one canceled and no timeout", before, after) + } +} diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index 833a7f88f..120e457a2 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -321,11 +321,12 @@ var ( // http_4xx — the dashboard answered 4xx (a bad token is 401/403) // http_5xx — the dashboard answered 5xx // http_other — any other status but 204 (a 200, a refused redirect) + // canceled — muster was shutting down mid-push // 🔴 `reason` is that closed vocabulary (ccusage.PushReason*): never a URL, // a status code or an account name. CCUsagePushFailures = promauto.NewCounterVec(prometheus.CounterOpts{ Name: "muster_cc_usage_push_failures_total", - Help: "Claude account usage pushes that failed, by reason (build|timeout|transport|http_4xx|http_5xx|http_other).", + Help: "Claude account usage pushes that failed, by reason (build|timeout|transport|http_4xx|http_5xx|http_other|canceled).", }, []string{"reason"}) )