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/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/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/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..64ea349d5 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; 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. | | `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..5137d7723 100644 --- a/docs/jobs.md +++ b/docs/jobs.md @@ -735,6 +735,48 @@ 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", + "poll_interval_seconds": 900, + "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. `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, + 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 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), + 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..0eff901f4 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,33 @@ 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"` + // 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 +// 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), + PollIntervalSeconds: int(s.ext.CCUsageInterval / time.Second), 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..349779bbe 100644 --- a/internal/api/cc_usage_test.go +++ b/internal/api/cc_usage_test.go @@ -244,3 +244,70 @@ 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, 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)) + 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) != 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 + // 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"]) + } + // 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]) + } +} + +// 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..0fa91a9fc --- /dev/null +++ b/internal/ccusage/push.go @@ -0,0 +1,233 @@ +package ccusage + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "log" + "net" + "net/http" + "net/url" + "strings" + "time" + "unicode" + + "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" + // PushReasonCanceled: muster was shutting down mid-push. + PushReasonCanceled = "canceled" +) + +// 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(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(parent, 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 + } + 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()) + return PushReasonTimeout + } + 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: + // 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) + return PushReason5xx + default: + p.logf("ccusage: push: the dashboard answered HTTP %d, want 204", resp.StatusCode) + return PushReasonOther + } +} + +// 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) } + +// 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..808f6e463 --- /dev/null +++ b/internal/ccusage/push_test.go @@ -0,0 +1,323 @@ +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, PushReasonCanceled} + +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 + answer string // the answer body; "" echoes the Authorization header + 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) + if p.answer != "" { + _, _ = w.Write([]byte(p.answer)) + return + } + _, _ = 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) + } + } +} + +// 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 7a75a9276..120e457a2 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -312,6 +312,22 @@ 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) + // 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|canceled).", + }, []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, "