amadan

carlos / platform Public

Add reviewed sponsored usage observation API and CLI

claude 9409c76ffdd159d4238e54598506eb35acec29ef parent 945616c
9 files changed, +1865 −8
  • billing_sponsored_cmd.go +15 −6
  • billing_sponsored_observation.go +379 −0
  • billing_sponsored_observation_test.go +351 −0
  • docs/site/cli.md +11 −1
  • internal/console/server.go +1 −0
  • internal/console/sponsoredobservation.go +515 −0
  • internal/console/sponsoredobservation_test.go +591 −0
  • internal/console/sponsoredoperator.go +1 −1
  • internal/console/testdata/sponsored-observation-statement.json +1 −0

CI passed — run details

CI log 
=== gate ===
cd launchctx && go vet ./... && go test ./... -count=1
ok  	amadan.net/carlos/platform.git/launchctx	0.011s
?   	amadan.net/carlos/platform.git/launchctx/controlname	[no test files]
?   	amadan.net/carlos/platform.git/launchctx/wireid	[no test files]
go vet ./...
gofmt -l .: clean
go test ./... github.com/benbjohnson/litestream/s3 -count=1 -timeout 30m
ok  	github.com/carlosframework/platform	161.070s
ok  	github.com/carlosframework/platform/cmd/console	2.950s
?   	github.com/carlosframework/platform/cmd/pricebook-export	[no test files]
ok  	github.com/carlosframework/platform/internal/activator	55.985s
ok  	github.com/carlosframework/platform/internal/agentsock	0.015s
ok  	github.com/carlosframework/platform/internal/analytics	0.017s
ok  	github.com/carlosframework/platform/internal/appconfig	0.732s
ok  	github.com/carlosframework/platform/internal/apperr	0.067s
ok  	github.com/carlosframework/platform/internal/applog	0.008s
ok  	github.com/carlosframework/platform/internal/authority	0.004s
ok  	github.com/carlosframework/platform/internal/awsbilling	0.007s
ok  	github.com/carlosframework/platform/internal/awsinv	0.003s
ok  	github.com/carlosframework/platform/internal/billingstripe	0.199s
ok  	github.com/carlosframework/platform/internal/caps	0.011s
ok  	github.com/carlosframework/platform/internal/certwarm	0.002s
ok  	github.com/carlosframework/platform/internal/cgroup	0.038s
ok  	github.com/carlosframework/platform/internal/clamd	0.054s
?   	github.com/carlosframework/platform/internal/cliflag	[no test files]
ok  	github.com/carlosframework/platform/internal/console	224.517s
?   	github.com/carlosframework/platform/internal/controlname	[no test files]
?   	github.com/carlosframework/platform/internal/controlsock	[no test files]
ok  	github.com/carlosframework/platform/internal/dbverify	0.017s
ok  	github.com/carlosframework/platform/internal/dnsrec	0.017s
ok  	github.com/carlosframework/platform/internal/docsite	0.017s
ok  	github.com/carlosframework/platform/internal/domains	1.078s
ok  	github.com/carlosframework/platform/internal/economics	0.017s
ok  	github.com/carlosframework/platform/internal/edge	20.387s
ok  	github.com/carlosframework/platform/internal/edgesfile	0.017s
ok  	github.com/carlosframework/platform/internal/entitlement	0.017s
ok  	github.com/carlosframework/platform/internal/fleetapi	0.095s
ok  	github.com/carlosframework/platform/internal/gate	1.655s
ok  	github.com/carlosframework/platform/internal/host	7.113s
ok  	github.com/carlosframework/platform/internal/infraview	0.017s
ok  	github.com/carlosframework/platform/internal/instancectl	0.415s
ok  	github.com/carlosframework/platform/internal/instanceexec	0.017s
ok  	github.com/carlosframework/platform/internal/instancefs	0.017s
ok  	github.com/carlosframework/platform/internal/lease	5.549s
ok  	github.com/carlosframework/platform/internal/ledger	0.017s
ok  	github.com/carlosframework/platform/internal/mail	0.056s
ok  	github.com/carlosframework/platform/internal/maintenance	0.091s
ok  	github.com/carlosframework/platform/internal/maintenanceexec	5.535s
ok  	github.com/carlosframework/platform/internal/maintenanceguard	0.056s
ok  	github.com/carlosframework/platform/internal/metrics	0.021s
ok  	github.com/carlosframework/platform/internal/objstore	8.443s
ok  	github.com/carlosframework/platform/internal/ops	15.445s
ok  	github.com/carlosframework/platform/internal/ops/admissionmodel	0.101s
ok  	github.com/carlosframework/platform/internal/plans	0.070s
ok  	github.com/carlosframework/platform/internal/privatefile	0.020s
ok  	github.com/carlosframework/platform/internal/provision	0.849s
ok  	github.com/carlosframework/platform/internal/publicrelease	0.007s
ok  	github.com/carlosframework/platform/internal/registry	1.391s
ok  	github.com/carlosframework/platform/internal/relay	0.189s
ok  	github.com/carlosframework/platform/internal/release	11.501s
ok  	github.com/carlosframework/platform/internal/resourcebudget	0.315s
ok  	github.com/carlosframework/platform/internal/s3compat	0.009s
ok  	github.com/carlosframework/platform/internal/schedule	1.037s
ok  	github.com/carlosframework/platform/internal/sigfence	0.446s
ok  	github.com/carlosframework/platform/internal/status	0.033s
ok  	github.com/carlosframework/platform/internal/steering	0.009s
ok  	github.com/carlosframework/platform/internal/svcrole	0.004s
ok  	github.com/carlosframework/platform/internal/telemetry	0.006s
ok  	github.com/carlosframework/platform/internal/tunnelclient	0.006s
ok  	github.com/carlosframework/platform/internal/unixdial	0.004s
ok  	github.com/carlosframework/platform/internal/usage	3.486s
ok  	github.com/carlosframework/platform/internal/version	0.002s
ok  	github.com/carlosframework/platform/internal/vet	1.559s
ok  	github.com/benbjohnson/litestream/s3	4.044s
=== cross ===
--- shard 0 of 5 ---
cross-compiling darwin/arm64
--- shard 1 of 5 ---
cross-compiling darwin/amd64
--- shard 2 of 5 ---
cross-compiling linux/arm64
--- shard 3 of 5 ---
cross-compiling windows/amd64
--- shard 4 of 5 ---
cross-compiling windows/arm64
=== minio-contract ===
minio contract: arm64, minio RELEASE.2025-09-07T16-13-09Z, mc RELEASE.2025-08-13T08-35-41Z
  minio: downloading https://github.com/minio/minio/releases/download/RELEASE.2025-09-07T16-13-09Z/minio.linux-arm64.RELEASE.2025-09-07T16-13-09Z
  minio: downloaded, digest verified
  mc: downloading https://github.com/minio/mc/releases/download/RELEASE.2025-08-13T08-35-41Z/mc.linux-arm64.RELEASE.2025-08-13T08-35-41Z
  mc: downloaded, digest verified
  minio healthy on 127.0.0.1:20999
Added `local` successfully.
Bucket created successfully `local/contract-bucket`.
Bucket created successfully `local/denied-bucket`.
Created policy `contract-only` successfully.
Added user `scopeduser` successfully.
Attached Policies: [contract-only]
To User: scopeduser
  buckets and scoped user ready
=== RUN   TestDeliveredContractMinio
=== RUN   TestDeliveredContractMinio/put
=== RUN   TestDeliveredContractMinio/get_round-trips_the_body
=== RUN   TestDeliveredContractMinio/list_shows_the_key
=== RUN   TestDeliveredContractMinio/head_and_list_agree_on_the_version
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_its_own_object
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_its_own_object/not_by_PutObjectTagging
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_its_own_object/and_not_by_x-amz-tagging_on_upload
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/not_by_CreateMultipartUpload
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/not_by_CopyObject_with_a_REPLACE_tagging_directive
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/but_multipart_itself_is_allowed
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/and_copy_itself_is_allowed
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/a_POST-policy_forged_tag_with_a_matching_version_is_still_not_a_verdict
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/and_an_ordinary_tagged-with-something-else_upload_is_allowed
=== RUN   TestDeliveredContractMinio/delete_removes_it
=== RUN   TestDeliveredContractMinio/cross-bucket_denial
--- PASS: TestDeliveredContractMinio (0.02s)
    --- PASS: TestDeliveredContractMinio/put (0.00s)
    --- PASS: TestDeliveredContractMinio/get_round-trips_the_body (0.00s)
    --- PASS: TestDeliveredContractMinio/list_shows_the_key (0.00s)
    --- PASS: TestDeliveredContractMinio/head_and_list_agree_on_the_version (0.00s)
    --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_its_own_object (0.00s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_its_own_object/not_by_PutObjectTagging (0.00s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_its_own_object/and_not_by_x-amz-tagging_on_upload (0.00s)
    --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route (0.01s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/not_by_CreateMultipartUpload (0.00s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/not_by_CopyObject_with_a_REPLACE_tagging_directive (0.00s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/but_multipart_itself_is_allowed (0.00s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/and_copy_itself_is_allowed (0.00s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/a_POST-policy_forged_tag_with_a_matching_version_is_still_not_a_verdict (0.00s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/and_an_ordinary_tagged-with-something-else_upload_is_allowed (0.00s)
    --- PASS: TestDeliveredContractMinio/delete_removes_it (0.00s)
    --- PASS: TestDeliveredContractMinio/cross-bucket_denial (0.00s)
PASS
ok  	github.com/carlosframework/platform/internal/objstore	0.024s
minio contract: PASS
diff --git a/billing_sponsored_cmd.go b/billing_sponsored_cmd.go
index ab437f02b..7b92b3826 100644
--- a/billing_sponsored_cmd.go
+++ b/billing_sponsored_cmd.go
@@ -21,7 +21,7 @@ import (
const (
sponsoredCopyUsageHelpHeading = "Sponsored account enrollment"
sponsoredCopyUsageDocsHeading = "`carlos billing sponsored`"
- sponsoredCopyUsageHelpSummary = "Use `preview` to read an account snapshot, `enroll` to submit a saved review, and `status` to read the historical result. Enrollment needs an explicit console from `--console` or `CARLOS_CONSOLE`. The CLI does not create operation IDs, fill choices, rewrite the roster, or choose a target."
+ sponsoredCopyUsageHelpSummary = "Use `preview`, `observe`, and `status` to read sponsored billing data. Use `enroll` to submit a saved review."
sponsoredCopyUsageUsagePreview = "`carlos billing sponsored preview [--console URL] [--json]`"
sponsoredCopyUsageUsageEnroll = "`carlos billing sponsored enroll --roster FILE [--console URL] [--json]`"
sponsoredCopyUsageUsageStatus = "`carlos billing sponsored status --operation 64hex [--console URL] [--json]`"
@@ -458,8 +458,8 @@ func (v sponsoredWireRequest) validate() error {
}
type sponsoredInvocation struct {
- verb, console, roster, operation string
- json, help bool
+ verb, console, roster, operation, account, month string
+ json, help bool
}
func parseSponsored(args []string) (sponsoredInvocation, error) {
@@ -476,7 +476,7 @@ func parseSponsored(args []string) (sponsoredInvocation, error) {
return v, nil
}
v.verb = args[0]
- if v.verb != "preview" && v.verb != "enroll" && v.verb != "status" {
+ if v.verb != "preview" && v.verb != "enroll" && v.verb != "status" && v.verb != "observe" {
return v, bad
}
if billingArgsAreUnambiguous(args[1:]) != nil {
@@ -487,6 +487,8 @@ func parseSponsored(args []string) (sponsoredInvocation, error) {
fs.StringVar(&v.console, "console", "", "")
fs.StringVar(&v.roster, "roster", "", "")
fs.StringVar(&v.operation, "operation", "", "")
+ fs.StringVar(&v.account, "account", "", "")
+ fs.StringVar(&v.month, "month", "", "")
fs.BoolVar(&v.json, "json", false, "")
if err := fs.Parse(args[1:]); err != nil {
if errors.Is(err, flag.ErrHelp) {
@@ -505,11 +507,15 @@ func parseSponsored(args []string) (sponsoredInvocation, error) {
invalid = invalid || strings.TrimSpace(v.console) == ""
case "roster":
invalid = invalid || v.verb != "enroll" || strings.TrimSpace(v.roster) == ""
+ case "account":
+ invalid = invalid || v.verb != "observe" || !domains.ValidSqid(v.account)
+ case "month":
+ invalid = invalid || v.verb != "observe" || !sponsoredObservationMonth(v.month)
case "operation":
invalid = invalid || v.verb != "status" || !sponsoredWireHex(v.operation)
}
})
- if invalid || (v.verb == "enroll" && v.roster == "") || (v.verb == "status" && !sponsoredWireHex(v.operation)) {
+ if invalid || (v.verb == "observe" && (!domains.ValidSqid(v.account) || !sponsoredObservationMonth(v.month))) || (v.verb == "enroll" && v.roster == "") || (v.verb == "status" && !sponsoredWireHex(v.operation)) {
return v, bad
}
if v.verb == "enroll" && v.console == "" && strings.TrimSpace(os.Getenv("CARLOS_CONSOLE")) == "" {
@@ -518,7 +524,7 @@ func parseSponsored(args []string) (sponsoredInvocation, error) {
return v, nil
}
func sponsoredHelp() string {
- return strings.Join([]string{sponsoredCopyUsageHelpHeading, sponsoredCopyUsageHelpSummary, sponsoredCopyUsageUsagePreview, sponsoredCopyUsageUsageEnroll, sponsoredCopyUsageUsageStatus}, "\n\n") + "\n"
+ return strings.Join([]string{sponsoredCopyUsageHelpHeading, sponsoredCopyUsageHelpSummary, sponsoredCopyUsageUsagePreview, observationCopyUsage, sponsoredCopyUsageUsageEnroll, sponsoredCopyUsageUsageStatus}, "\n\n") + "\n"
}
func sponsoredReadBounded(r io.Reader, limit int64) ([]byte, error) {
raw, err := io.ReadAll(io.LimitReader(r, limit+1))
@@ -691,6 +697,9 @@ func runBillingSponsoredWith(args []string, session func(string) (*apiSession, e
return runSponsored(context.Background(), sess, inv, body, request, w)
}
func runSponsored(ctx context.Context, sess *apiSession, inv sponsoredInvocation, body []byte, request sponsoredWireRequest, w io.Writer) error {
+ if inv.verb == "observe" {
+ return runSponsoredObservation(ctx, sess, inv, w)
+ }
writing := inv.verb == "enroll"
method, path := http.MethodGet, "/api/billing/sponsored/preview"
if writing {
diff --git a/billing_sponsored_observation.go b/billing_sponsored_observation.go
new file mode 100644
index 000000000..f75abfe3a
--- /dev/null
+++ b/billing_sponsored_observation.go
@@ -0,0 +1,379 @@
+package main
+
+import (
+ "bytes"
+ "context"
+ "crypto/sha256"
+ "encoding/base64"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "github.com/carlosframework/platform/internal/domains"
+ usagedata "github.com/carlosframework/platform/internal/usage"
+ "io"
+ "net/http"
+ "reflect"
+ "sort"
+ "time"
+ "unicode/utf8"
+)
+
+const (
+ observationCopyAccountNotFound = "The selected account was not found or is not enrolled."
+ observationCopyStatementNotFound = "No stored statement exists for the selected account and month. This does not mean zero usage or change approved funding."
+ observationCopyUnsupportedStore = "The configured store cannot provide this observation."
+ observationCopyUnsupported = "The stored statement uses an unsupported version."
+ observationCopyInvalidStored = "The stored statement cannot be read safely."
+ observationCopyTooLarge = "The complete observation is too large to return. No partial result was returned."
+ observationCopyFailed = "The sponsored billing result could not be read."
+ observationCopyProtocol = "The sponsored billing response was invalid or incomplete."
+ observationCopyOutput = "The sponsored billing result was received, but the complete output could not be written. Run the read command again."
+ observationCopyUsage = "`carlos billing sponsored observe --account SQID --month YYYY-MM [--console URL] [--json]`"
+ observationCopyHeading = "Stored usage observation"
+ observationCopyConsole = "Selected console"
+ observationCopyAccount = "Account"
+ observationCopyMonth = "Requested month"
+ observationCopyObserved = "Observed"
+ observationCopyFirst = "First enrollment operation"
+ observationCopyReceipt = "Receipt digest"
+ observationCopyRetirement = "Retirement history"
+ observationCopyKey = "Source key"
+ observationCopyETag = "ETag"
+ observationCopySHA = "Statement SHA-256"
+ observationCopyUpdated = "Statement updated"
+ observationCopyClosed = "Statement status"
+ observationCopyCoverage = "Coverage"
+ observationCopyAnomalies = "Anomalies"
+)
+
+type sponsoredObservation struct {
+ V int `json:"v"`
+ AccountID int64 `json:"account_id"`
+ Sqid string `json:"sqid"`
+ Month string `json:"month"`
+ FirstOperationID string `json:"first_operation_id"`
+ ReceiptDigest string `json:"receipt_digest"`
+ ObservedAt time.Time `json:"observed_at"`
+ DeletedAt int64 `json:"deleted_at"`
+ Retirement string `json:"retirement"`
+ SourceKey string `json:"source_key"`
+ ETag string `json:"etag,omitempty"`
+ StatementSHA256 string `json:"statement_sha256"`
+ StatementBase64 string `json:"statement_base64"`
+ Flags []string `json:"flags"`
+}
+
+func (out *sponsoredObservation) UnmarshalJSON(raw []byte) error {
+ var v sponsoredObservation
+ var etag *string
+ var observedAt json.RawMessage
+ if err := sponsoredWireObject(raw, false, map[string]sponsoredWireField{
+ "v": {target: &v.V}, "account_id": {target: &v.AccountID}, "sqid": {target: &v.Sqid}, "month": {target: &v.Month},
+ "first_operation_id": {target: &v.FirstOperationID}, "receipt_digest": {target: &v.ReceiptDigest},
+ "observed_at": {target: &observedAt}, "deleted_at": {target: &v.DeletedAt}, "retirement": {target: &v.Retirement},
+ "source_key": {target: &v.SourceKey}, "etag": {target: &etag, optional: true}, "statement_sha256": {target: &v.StatementSHA256},
+ "statement_base64": {target: &v.StatementBase64}, "flags": {target: &v.Flags},
+ }); err != nil {
+ return err
+ }
+ if json.Unmarshal(observedAt, &v.ObservedAt) != nil {
+ return errSponsoredWire
+ }
+ canonical, err := json.Marshal(v.ObservedAt.UTC().Format(time.RFC3339Nano))
+ if err != nil || !bytes.Equal(bytes.TrimSpace(observedAt), canonical) {
+ return errSponsoredWire
+ }
+ if etag != nil {
+ if *etag == "" {
+ return errSponsoredWire
+ }
+ v.ETag = *etag
+ }
+ *out = v
+ return nil
+}
+
+func sponsoredObservationMonth(s string) bool {
+ if len(s) != 7 || s[4] != '-' {
+ return false
+ }
+ for i, c := range s {
+ if i != 4 && (c < '0' || c > '9') {
+ return false
+ }
+ }
+ return s[5:] >= "01" && s[5:] <= "12"
+}
+
+// Walk every object, including additive fields, before decoding known fields.
+func sponsoredObservationJSON(raw []byte) bool {
+ if !utf8.Valid(raw) {
+ return false
+ }
+ d := json.NewDecoder(bytes.NewReader(raw))
+ d.UseNumber()
+ var walk func(int) bool
+ walk = func(depth int) bool {
+ tok, err := d.Token()
+ if err != nil {
+ return false
+ }
+ delim, ok := tok.(json.Delim)
+ if !ok {
+ return true
+ }
+ if depth >= 10000 {
+ return false
+ }
+ switch delim {
+ case '{':
+ seen := map[string]bool{}
+ for d.More() {
+ k, e := d.Token()
+ if e != nil {
+ return false
+ }
+ s, ok := k.(string)
+ if !ok || seen[s] {
+ return false
+ }
+ seen[s] = true
+ if !walk(depth + 1) {
+ return false
+ }
+ }
+ end, e := d.Token()
+ return e == nil && end == json.Delim('}')
+ case '[':
+ for d.More() {
+ if !walk(depth + 1) {
+ return false
+ }
+ }
+ end, e := d.Token()
+ return e == nil && end == json.Delim(']')
+ }
+ return false
+ }
+ return walk(0) && d.Decode(new(json.RawMessage)) == io.EOF
+}
+
+func sponsoredObservationNumberMap(raw json.RawMessage, days bool) bool {
+ if bytes.Equal(bytes.TrimSpace(raw), []byte("null")) {
+ return false
+ }
+ var m map[string]json.RawMessage
+ if json.Unmarshal(raw, &m) != nil || m == nil {
+ return false
+ }
+ for _, v := range m {
+ if days {
+ if !sponsoredObservationNumberMap(v, false) {
+ return false
+ }
+ } else {
+ if bytes.Equal(bytes.TrimSpace(v), []byte("null")) {
+ return false
+ }
+ var number float64
+ if json.Unmarshal(v, &number) != nil {
+ return false
+ }
+ }
+ }
+ return true
+}
+
+func sponsoredObservationStatement(raw []byte, account, month string, at time.Time) (usagedata.Statement, []string, string) {
+ var s usagedata.Statement
+ invalid := func() (usagedata.Statement, []string, string) { return usagedata.Statement{}, nil, "invalid_statement" }
+ if len(raw) > sponsoredWireRequestLimit || !sponsoredObservationJSON(raw) {
+ return invalid()
+ }
+ var version int
+ if sponsoredWireObject(raw, false, map[string]sponsoredWireField{"v": {target: &version}}) != nil {
+ return invalid()
+ }
+ if version != 1 {
+ return usagedata.Statement{}, nil, "unsupported_statement"
+ }
+ fields := map[string]sponsoredWireField{
+ "v": {target: &s.V}, "account": {target: &s.Account}, "month": {target: &s.Month}, "meters": {target: &s.Meters},
+ "covered_days": {target: &s.CoveredDays}, "days_in_month": {target: &s.DaysInMonth}, "updated_at": {target: &s.UpdatedAt},
+ "source_days": {target: &s.SourceDays, optional: true, nullable: true}, "days": {target: &s.Days, optional: true, nullable: true},
+ "closed": {target: &s.Closed, optional: true}, "evidence": {target: &s.Evidence, optional: true, nullable: true},
+ }
+ if sponsoredWireObject(raw, false, fields) != nil {
+ return invalid()
+ }
+ if s.Account != account || s.Month != month || !sponsoredObservationMonth(month) {
+ return invalid()
+ }
+ var root map[string]json.RawMessage
+ _ = json.Unmarshal(raw, &root)
+ if !sponsoredObservationNumberMap(root["meters"], false) {
+ return invalid()
+ }
+ for _, k := range []string{"source_days", "days"} {
+ if b, ok := root[k]; ok && !bytes.Equal(bytes.TrimSpace(b), []byte("null")) && !sponsoredObservationNumberMap(b, k == "days") {
+ return invalid()
+ }
+ }
+ if b, ok := root["evidence"]; ok && !bytes.Equal(bytes.TrimSpace(b), []byte("null")) {
+ var key, sha string
+ if sponsoredWireObject(b, false, map[string]sponsoredWireField{"key": {target: &key, optional: true}, "sha256": {target: &sha, optional: true}}) != nil {
+ return invalid()
+ }
+ }
+ flags := []string{}
+ negative := false
+ for _, v := range s.Meters {
+ negative = negative || v < 0
+ }
+ for _, day := range s.Days {
+ for _, v := range day {
+ negative = negative || v < 0
+ }
+ }
+ if negative {
+ flags = append(flags, "negative_meter_values")
+ }
+ date, _ := time.Parse("2006-01", month)
+ n := time.Date(date.Year(), date.Month()+1, 0, 0, 0, 0, 0, time.UTC).Day()
+ badDays := s.DaysInMonth != n || s.CoveredDays < 0 || s.CoveredDays > n
+ for _, v := range s.SourceDays {
+ badDays = badDays || v < 0 || v > n
+ }
+ if badDays {
+ flags = append(flags, "inconsistent_day_counts")
+ }
+ if s.UpdatedAt.After(at) {
+ flags = append(flags, "future_updated_at")
+ }
+ sort.Strings(flags)
+ return s, flags, ""
+}
+
+func sponsoredObservationErrorSpec(code string) (int, string) {
+ switch code {
+ case "invalid_request":
+ return 400, sponsoredCopyInvalidRequestError
+ case "account_not_found":
+ return 404, observationCopyAccountNotFound
+ case "statement_not_found":
+ return 404, observationCopyStatementNotFound
+ case "unsupported_store":
+ return 422, observationCopyUnsupportedStore
+ case "unsupported_statement":
+ return 422, observationCopyUnsupported
+ case "invalid_statement":
+ return 422, observationCopyInvalidStored
+ case "observation_too_large":
+ return 422, observationCopyTooLarge
+ default:
+ return 500, observationCopyFailed
+ }
+}
+
+func sponsoredObservationValidate(raw []byte, account, month string) (sponsoredObservation, []byte, error) {
+ var v sponsoredObservation
+ bad := errors.New(observationCopyProtocol)
+ if len(raw) > sponsoredWireResponseLimit || json.Unmarshal(raw, &v) != nil || v.V != 1 || v.AccountID <= 0 || v.Sqid != domains.Sqid(v.AccountID) || v.Sqid != account || v.Month != month || !sponsoredObservationMonth(month) || !sponsoredWireHex(v.FirstOperationID) || !sponsoredWireHex(v.ReceiptDigest) || !sponsoredWireHex(v.StatementSHA256) || v.ObservedAt.IsZero() || v.DeletedAt < 0 || v.SourceKey != usagedata.StatementKey(account, month) || v.Flags == nil {
+ return v, nil, bad
+ }
+ if v.Retirement != "none" && v.Retirement != "pending" && v.Retirement != "committed" {
+ return v, nil, bad
+ }
+ if len(v.StatementBase64) > base64.StdEncoding.EncodedLen(sponsoredWireRequestLimit) {
+ return v, nil, bad
+ }
+ statement, err := base64.StdEncoding.Strict().DecodeString(v.StatementBase64)
+ if err != nil || len(statement) > sponsoredWireRequestLimit || base64.StdEncoding.EncodeToString(statement) != v.StatementBase64 {
+ return v, nil, bad
+ }
+ digest := sha256.Sum256(statement)
+ if hex.EncodeToString(digest[:]) != v.StatementSHA256 {
+ return v, nil, bad
+ }
+ _, flags, code := sponsoredObservationStatement(statement, account, month, v.ObservedAt)
+ if code != "" || !reflect.DeepEqual(v.Flags, flags) {
+ return v, nil, bad
+ }
+ return v, statement, nil
+}
+
+func sponsoredObservationRefusal(status int, raw []byte, console string) error {
+ if status == 401 {
+ return sponsoredRefusal(status, raw, "observe", console)
+ }
+ var code, message string
+ if sponsoredWireObject(raw, true, map[string]sponsoredWireField{"code": {target: &code}, "error": {target: &message}}) == nil {
+ switch code {
+ case "invalid_request", "account_not_found", "statement_not_found", "unsupported_store", "unsupported_statement", "invalid_statement", "observation_too_large", "observation_failed":
+ want, text := sponsoredObservationErrorSpec(code)
+ if status == want && message == text {
+ return errors.New(text)
+ }
+ }
+ }
+ return errors.New(observationCopyFailed)
+}
+
+func runSponsoredObservation(ctx context.Context, sess *apiSession, inv sponsoredInvocation, w io.Writer) error {
+ status, raw, err := sess.sponsoredRequest(ctx, http.MethodGet, "/api/billing/sponsored/observations/"+inv.account+"/"+inv.month, nil)
+ if err != nil {
+ var moved *consoleMovedError
+ if errors.As(err, &moved) {
+ return err
+ }
+ return errors.New(observationCopyFailed)
+ }
+ if status != 200 {
+ return sponsoredObservationRefusal(status, raw, sess.console)
+ }
+ v, statement, err := sponsoredObservationValidate(raw, inv.account, inv.month)
+ if err != nil {
+ return err
+ }
+ var out bytes.Buffer
+ if inv.json {
+ out.Write(raw)
+ out.WriteByte('\n')
+ } else {
+ fmt.Fprintln(&out, observationCopyHeading)
+ row := func(label string, value any) { fmt.Fprintf(&out, "%s: %s\n", label, sponsoredHumanRow(value)) }
+ row(observationCopyConsole, sess.console)
+ row(observationCopyAccount, v.Sqid)
+ row(observationCopyMonth, v.Month)
+ row(observationCopyObserved, v.ObservedAt)
+ row(observationCopyFirst, v.FirstOperationID)
+ row(observationCopyReceipt, v.ReceiptDigest)
+ row(observationCopyRetirement, struct {
+ State string `json:"retirement"`
+ DeletedAt int64 `json:"deleted_at"`
+ }{v.Retirement, v.DeletedAt})
+ row(observationCopyKey, v.SourceKey)
+ var etag any
+ if v.ETag != "" {
+ etag = v.ETag
+ }
+ row(observationCopyETag, etag)
+ row(observationCopySHA, v.StatementSHA256)
+ var data map[string]json.RawMessage
+ _ = json.Unmarshal(statement, &data)
+ row(observationCopyUpdated, data["updated_at"])
+ row(observationCopyClosed, data["closed"])
+ coverage := map[string]json.RawMessage{}
+ for _, k := range []string{"meters", "covered_days", "source_days", "days", "days_in_month", "evidence"} {
+ coverage[k] = data[k]
+ }
+ row(observationCopyCoverage, coverage)
+ row(observationCopyAnomalies, v.Flags)
+ }
+ n, err := w.Write(out.Bytes())
+ if err != nil || n != out.Len() {
+ return errors.New(observationCopyOutput)
+ }
+ return nil
+}
diff --git a/billing_sponsored_observation_test.go b/billing_sponsored_observation_test.go
new file mode 100644
index 000000000..d318ca2a6
--- /dev/null
+++ b/billing_sponsored_observation_test.go
@@ -0,0 +1,351 @@
+package main
+
+import (
+ "bytes"
+ "context"
+ "crypto/sha256"
+ "encoding/base64"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "io"
+ "net/http"
+ "net/http/httptest"
+ "os"
+ "path/filepath"
+ "strings"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/console"
+ "github.com/carlosframework/platform/internal/release"
+ usagedata "github.com/carlosframework/platform/internal/usage"
+)
+
+func observationClientFixture(t *testing.T) (sponsoredObservation, []byte, []byte) {
+ t.Helper()
+ raw, e := os.ReadFile("internal/console/testdata/sponsored-observation-statement.json")
+ if e != nil {
+ t.Fatal(e)
+ }
+ at := time.Date(2026, 10, 15, 0, 0, 0, 0, time.UTC)
+ _, flags, c := sponsoredObservationStatement(raw, "bab", "2026-10", at)
+ if c != "" {
+ t.Fatal(c)
+ }
+ d := sha256.Sum256(raw)
+ v := sponsoredObservation{V: 1, AccountID: 1, Sqid: "bab", Month: "2026-10", FirstOperationID: strings.Repeat("a", 64), ReceiptDigest: strings.Repeat("b", 64), ObservedAt: at, Retirement: "none", SourceKey: usagedata.StatementKey("bab", "2026-10"), StatementSHA256: hex.EncodeToString(d[:]), StatementBase64: base64.StdEncoding.EncodeToString(raw), Flags: flags}
+ b, e := json.Marshal(v)
+ if e != nil {
+ t.Fatal(e)
+ }
+ return v, raw, b
+}
+
+func TestSponsoredObservationCLIValidation(t *testing.T) {
+ v, raw, wire := observationClientFixture(t)
+ got, b, e := sponsoredObservationValidate(wire, "bab", "2026-10")
+ if e != nil || !bytes.Equal(b, raw) || got.StatementSHA256 != v.StatementSHA256 {
+ t.Fatal(e)
+ }
+ changes := []func(*sponsoredObservation){func(v *sponsoredObservation) { v.StatementSHA256 = strings.Repeat("0", 64) }, func(v *sponsoredObservation) { v.StatementBase64 = "!!!!" }, func(v *sponsoredObservation) { v.StatementBase64 += "\n" }, func(v *sponsoredObservation) { v.Sqid = "bad" }, func(v *sponsoredObservation) { v.AccountID = 2 }, func(v *sponsoredObservation) { v.Month = "2026-11" }, func(v *sponsoredObservation) { v.SourceKey = "other" }, func(v *sponsoredObservation) { v.FirstOperationID = "bad" }, func(v *sponsoredObservation) { v.ReceiptDigest = "bad" }, func(v *sponsoredObservation) { v.Flags = []string{} }, func(v *sponsoredObservation) { v.Flags = nil }, func(v *sponsoredObservation) { v.Retirement = "active" }, func(v *sponsoredObservation) { v.V = 2 }}
+ for i, change := range changes {
+ copy := v
+ change(&copy)
+ bad, _ := json.Marshal(copy)
+ if _, _, e := sponsoredObservationValidate(bad, "bab", "2026-10"); e == nil {
+ t.Fatal("accepted tamper", i)
+ }
+ }
+ for _, bad := range [][]byte{[]byte(`null`), append(append([]byte{}, wire...), []byte(`{}`)...), bytes.Replace(wire, []byte(`"v":1`), []byte(`"v":1,"v":1`), 1), bytes.Replace(wire, []byte(`"v":1`), []byte(`"V":1`), 1)} {
+ if _, _, e := sponsoredObservationValidate(bad, "bab", "2026-10"); e == nil {
+ t.Fatal("accepted framing")
+ }
+ }
+ additive := bytes.Replace(wire, []byte(`"v":1`), []byte(`"v":1,"future":null`), 1)
+ if _, _, e := sponsoredObservationValidate(additive, "bab", "2026-10"); e != nil {
+ t.Fatal("additive", e)
+ }
+}
+
+func TestSponsoredObservationCLIParsing(t *testing.T) {
+ for _, month := range []string{"0000-01", "2026-10", "9999-12"} {
+ v, e := parseSponsored([]string{"observe", "--account", "bab", "--month", month})
+ if e != nil || v.verb != "observe" || v.month != month {
+ t.Fatal(v, e)
+ }
+ }
+ for _, args := range [][]string{{"observe"}, {"observe", "--account", "bab"}, {"observe", "--account", "iii", "--month", "2026-10"}, {"observe", "--account", "bab", "--month", "2026-13"}, {"observe", "--account", "bab", "--month", "2026-10", "--month", "2026-11"}, {"observe", "--account", "bab", "--month", "2026-10", "--all"}, {"preview", "--account", "bab"}, {"observe", "--account", "bab", "--month", "2026-10", "--roster", "x"}} {
+ if _, e := parseSponsored(args); e == nil || e.Error() != sponsoredCopyUsageArgumentError {
+ t.Fatal(args, e)
+ }
+ }
+ help := sponsoredHelp()
+ if !strings.Contains(help, observationCopyUsage) || strings.Index(help, observationCopyUsage) > strings.Index(help, sponsoredCopyUsageUsageEnroll) {
+ t.Fatal("help placement")
+ }
+}
+
+func TestSponsoredObservationCLIDispatchAndOutput(t *testing.T) {
+ _, raw, wire := observationClientFixture(t)
+ var calls atomic.Int32
+ sess := sponsoredClientSession(t, func(w http.ResponseWriter, r *http.Request) {
+ calls.Add(1)
+ if r.Method != "GET" || r.URL.Path != "/api/billing/sponsored/observations/bab/2026-10" || r.URL.RawQuery != "" || r.Header.Get("Authorization") != "Bearer test-token" {
+ t.Error("request", r.Method, r.URL)
+ }
+ b, _ := io.ReadAll(r.Body)
+ if len(b) != 0 {
+ t.Error("body")
+ }
+ w.Write(wire)
+ })
+ var out bytes.Buffer
+ e := runBillingWith([]string{"sponsored", "observe", "--account", "bab", "--month", "2026-10", "--json"}, func(string) (*apiSession, error) { return sess, nil }, &out)
+ if e != nil || calls.Load() != 1 || !bytes.Equal(out.Bytes(), append(append([]byte{}, wire...), '\n')) {
+ t.Fatal("json output", e)
+ }
+ var parsed sponsoredObservation
+ json.Unmarshal(out.Bytes(), &parsed)
+ b, _ := base64.StdEncoding.DecodeString(parsed.StatementBase64)
+ if !bytes.Equal(raw, b) {
+ t.Fatal("raw bytes changed")
+ }
+ out.Reset()
+ e = runSponsored(context.Background(), sess, sponsoredInvocation{verb: "observe", account: "bab", month: "2026-10"}, nil, sponsoredWireRequest{}, &out)
+ if e != nil || !strings.HasPrefix(out.String(), observationCopyHeading+"\n") || !strings.Contains(out.String(), observationCopyAnomalies+":") {
+ t.Fatal(e, out.String())
+ }
+ for _, b := range out.Bytes() {
+ if b >= 0x7f {
+ t.Fatal("nonASCII human data")
+ }
+ }
+}
+
+type observationFailWriter struct{}
+
+func (observationFailWriter) Write([]byte) (int, error) { return 0, errors.New("private raw writer") }
+func TestSponsoredObservationCLIRefusalsAndFailures(t *testing.T) {
+ for _, code := range []string{"invalid_request", "account_not_found", "statement_not_found", "unsupported_store", "unsupported_statement", "invalid_statement", "observation_too_large", "observation_failed"} {
+ status, text := sponsoredObservationErrorSpec(code)
+ raw, _ := json.Marshal(map[string]string{"code": code, "error": text})
+ if e := sponsoredObservationRefusal(status, raw, "https://fixture.test"); e.Error() != text {
+ t.Fatal(code, e)
+ }
+ if e := sponsoredObservationRefusal(status+1, raw, "https://fixture.test"); e.Error() != observationCopyFailed {
+ t.Fatal("wrong status", code, e)
+ }
+ corrupt, _ := json.Marshal(map[string]string{"code": code, "error": "private provider raw body"})
+ if e := sponsoredObservationRefusal(status, corrupt, "https://fixture.test"); e.Error() != observationCopyFailed {
+ t.Fatal("reflection", e)
+ }
+ }
+ for _, raw := range [][]byte{[]byte(`{"code":"other","error":"private"}`), []byte(`{"error":"private"}`), []byte("404 page not found\n")} {
+ if e := sponsoredObservationRefusal(404, raw, "https://fixture.test"); e.Error() != observationCopyFailed {
+ t.Fatal(e)
+ }
+ }
+ auth := []byte(`{"error":"not signed in — run: carlos auth login --console https://fixture.test"}`)
+ if e := sponsoredObservationRefusal(401, auth, "https://fixture.test"); e.Error() != "not signed in — run: carlos auth login --console https://fixture.test" {
+ t.Fatal(e)
+ }
+ _, _, wire := observationClientFixture(t)
+ sess := sponsoredClientSession(t, func(w http.ResponseWriter, r *http.Request) { w.Write(wire) })
+ if e := runSponsoredObservation(context.Background(), sess, sponsoredInvocation{account: "bab", month: "2026-10"}, observationFailWriter{}); e == nil || e.Error() != observationCopyOutput {
+ t.Fatal(e)
+ }
+ for _, body := range [][]byte{[]byte("private malformed"), bytes.Repeat([]byte("x"), sponsoredWireResponseLimit+1)} {
+ bad := sponsoredClientSession(t, func(w http.ResponseWriter, r *http.Request) { w.Write(body) })
+ var out bytes.Buffer
+ e := runSponsoredObservation(context.Background(), bad, sponsoredInvocation{account: "bab", month: "2026-10"}, &out)
+ if e == nil || out.Len() != 0 || strings.Contains(e.Error(), "private") {
+ t.Fatal(e)
+ }
+ }
+}
+
+func TestSponsoredObservationCLIRedirect(t *testing.T) {
+ old := quietConsoleOffers
+ quietConsoleOffers = true
+ t.Cleanup(func() { quietConsoleOffers = old })
+ var sourceCalls, targetCalls atomic.Int32
+ target := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { targetCalls.Add(1) }))
+ defer target.Close()
+ source := sponsoredClientSession(t, func(w http.ResponseWriter, r *http.Request) {
+ sourceCalls.Add(1)
+ w.Header().Set("Location", target.URL+r.URL.Path)
+ w.WriteHeader(308)
+ })
+ var out bytes.Buffer
+ e := runSponsoredObservation(context.Background(), source, sponsoredInvocation{account: "bab", month: "2026-10"}, &out)
+ var moved *consoleMovedError
+ if !errors.As(e, &moved) || sourceCalls.Load() != 1 || targetCalls.Load() != 0 || out.Len() != 0 {
+ t.Fatal(e, sourceCalls.Load(), targetCalls.Load())
+ }
+}
+
+func TestSponsoredObservationCLIRealConsole(t *testing.T) {
+ tmp := t.TempDir()
+ st, e := console.Open(filepath.Join(tmp, "console.db"))
+ if e != nil {
+ t.Fatal(e)
+ }
+ defer st.Close()
+ seed, pub64, e := release.GenerateKey()
+ if e != nil {
+ t.Fatal(e)
+ }
+ key, _ := release.ParseSigningKey(seed, "test")
+ pub, _ := release.ParsePublicKey(pub64, "test")
+ keymail := httptest.NewServer(http.NotFoundHandler())
+ defer keymail.Close()
+ var srv *console.Server
+ ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { srv.ServeHTTP(w, r) }))
+ defer ts.Close()
+ store := release.NewDirStore(filepath.Join(tmp, "objects"))
+ srv, e = console.NewServer(console.Config{Origin: ts.URL, Keymail: keymail.URL, Registry: filepath.Join(tmp, "registry.db"), Releases: store, ReleasePub: pub, Key: key, AppScheme: "http"}, st)
+ if e != nil {
+ t.Fatal(e)
+ }
+ addr := "operator@example.test"
+ if e = st.AddMember(1, addr, console.RoleOwner, time.Now()); e != nil {
+ t.Fatal(e)
+ }
+ token, _ := e2eDeviceLogin(t, ts.URL, st, addr, "observation")
+ if e = st.SetOperator(addr, true); e != nil {
+ t.Fatal(e)
+ }
+ sess := &apiSession{console: ts.URL, address: addr, c: newConsoleClient(ts.URL, token)}
+ f := sponsoredClientFixture(t)
+ var req sponsoredWireRequest
+ json.Unmarshal(f["request"], &req)
+ req.Items[0].AccountID = 1
+ req.Items[0].Sqid = "bab"
+ req.Items[0].ConfirmSystemAccount = true
+ b, _ := json.Marshal(req)
+ var enrolled bytes.Buffer
+ if e = runSponsored(context.Background(), sess, sponsoredInvocation{verb: "enroll", json: true}, b, req, &enrolled); e != nil {
+ t.Fatal(e)
+ }
+ _, raw, _ := observationClientFixture(t)
+ if _, e = store.Put(context.Background(), usagedata.StatementKey("bab", "2026-10"), raw, ""); e != nil {
+ t.Fatal(e)
+ }
+ var out bytes.Buffer
+ e = runBillingWith([]string{"sponsored", "observe", "--account", "bab", "--month", "2026-10", "--json"}, func(string) (*apiSession, error) { return sess, nil }, &out)
+ if e != nil {
+ t.Fatal(e)
+ }
+ _, got, e := sponsoredObservationValidate(bytes.TrimSpace(out.Bytes()), "bab", "2026-10")
+ if e != nil || !bytes.Equal(raw, got) {
+ t.Fatal(e)
+ }
+ out.Reset()
+ e = runSponsoredObservation(context.Background(), sess, sponsoredInvocation{account: "bab", month: "2026-11"}, &out)
+ if e == nil || e.Error() != observationCopyStatementNotFound || out.Len() != 0 {
+ t.Fatal(e)
+ }
+}
+
+func TestSponsoredObservationCLIEmptyETag(t *testing.T) {
+ _, _, wire := observationClientFixture(t)
+ wire = bytes.Replace(wire, []byte(`"v":1`), []byte(`"v":1,"etag":""`), 1)
+ if _, _, err := sponsoredObservationValidate(wire, "bab", "2026-10"); err == nil {
+ t.Fatal("explicit empty ETag accepted; absence must be represented by omission")
+ }
+}
+
+func TestSponsoredObservationVersionDispatch(t *testing.T) {
+ for _, tc := range []struct{ raw, code string }{
+ {`{"v":2,"next_shape":[1,2,3]}`, "unsupported_statement"},
+ {`{"v":2,"meters":"different future shape"}`, "unsupported_statement"},
+ {`{"v":null}`, "invalid_statement"}, {`{"v":"2"}`, "invalid_statement"},
+ {`{"v":2,"next":{"x":1,"x":2}}`, "invalid_statement"},
+ } {
+ if _, _, code := sponsoredObservationStatement([]byte(tc.raw), "bab", "2026-10", time.Now()); code != tc.code {
+ t.Fatalf("%s: got %s want %s", tc.raw, code, tc.code)
+ }
+ }
+}
+
+func TestSponsoredObservationCLICanonicalObservedAt(t *testing.T) {
+ _, _, wire := observationClientFixture(t)
+ for _, tc := range []struct {
+ value string
+ valid bool
+ }{
+ {"2026-10-15T00:00:00Z", true},
+ {"2026-10-15T00:00:00.123456789Z", true},
+ {"2026-10-15T00:00:00.000000001Z", true},
+ {"2026-10-15T00:00:00.1Z", true},
+ {"2026-10-15T01:00:00+01:00", false},
+ {"2026-10-15T00:00:00+00:00", false},
+ {"2026-10-15T00:00:00.100Z", false},
+ {"2026-10-15T00:00:00.000Z", false},
+ } {
+ t.Run(tc.value, func(t *testing.T) {
+ value, _ := json.Marshal(tc.value)
+ raw := bytes.Replace(wire, []byte(`"2026-10-15T00:00:00Z"`), value, 1)
+ sess := sponsoredClientSession(t, func(w http.ResponseWriter, r *http.Request) { w.Write(raw) })
+ var out bytes.Buffer
+ err := runSponsoredObservation(context.Background(), sess, sponsoredInvocation{account: "bab", month: "2026-10", json: true}, &out)
+ if tc.valid {
+ if err != nil || !bytes.Equal(out.Bytes(), append(append([]byte{}, raw...), '\n')) {
+ t.Fatalf("canonical timestamp refused or altered: %v", err)
+ }
+ } else if err == nil || err.Error() != observationCopyProtocol || out.Len() != 0 {
+ t.Fatalf("noncanonical observed_at accepted: error=%v output_bytes=%d", err, out.Len())
+ }
+ })
+ }
+}
+
+func TestSponsoredObservationDepthBoundary(t *testing.T) {
+ for _, depth := range []int{9999, 10000, 10001} {
+ raw := []byte(strings.Repeat("[", depth) + "0" + strings.Repeat("]", depth))
+ if got := sponsoredObservationJSON(raw); got != (depth <= 10000) {
+ t.Fatalf("array depth %d accepted=%v", depth, got)
+ }
+ if depth > 10000 {
+ if _, _, code := sponsoredObservationStatement(raw, "bab", "2026-10", time.Time{}); code != "invalid_statement" {
+ t.Fatal(code)
+ }
+ }
+ }
+ for _, arrays := range []int{9999, 10000, 10001} {
+ raw := []byte(`{"x":` + strings.Repeat("[", arrays) + "0" + strings.Repeat("]", arrays) + "}")
+ if got := sponsoredObservationJSON(raw); got != (arrays+1 <= 10000) {
+ t.Fatalf("object plus %d arrays accepted=%v", arrays, got)
+ }
+ if _, _, code := sponsoredObservationStatement(raw, "bab", "2026-10", time.Time{}); code != "invalid_statement" {
+ t.Fatal(code)
+ }
+ }
+}
+
+func TestSponsoredObservationCLIDeepStatement(t *testing.T) {
+ v, _, _ := observationClientFixture(t)
+ for _, raw := range [][]byte{
+ []byte(strings.Repeat("[", 10001) + "0" + strings.Repeat("]", 10001)),
+ []byte(`{"x":` + strings.Repeat("[", 10001)),
+ } {
+ digest := sha256.Sum256(raw)
+ v.StatementSHA256 = hex.EncodeToString(digest[:])
+ v.StatementBase64 = base64.StdEncoding.EncodeToString(raw)
+ wire, err := json.Marshal(v)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, _, err := sponsoredObservationValidate(wire, "bab", "2026-10"); err == nil {
+ t.Fatal("deep decoded statement accepted")
+ }
+ sess := sponsoredClientSession(t, func(w http.ResponseWriter, r *http.Request) { w.Write(wire) })
+ var out bytes.Buffer
+ err = runSponsoredObservation(context.Background(), sess, sponsoredInvocation{account: "bab", month: "2026-10", json: true}, &out)
+ if err == nil || err.Error() != observationCopyProtocol || out.Len() != 0 {
+ t.Fatalf("deep CLI statement: %v output=%d", err, out.Len())
+ }
+ }
+}
diff --git a/docs/site/cli.md b/docs/site/cli.md
index 67013c679..887c0fd9c 100644
--- a/docs/site/cli.md
+++ b/docs/site/cli.md
@@ -1296,7 +1296,7 @@ to log in.
#### `carlos billing sponsored`
-Use `preview` to read an account snapshot, `enroll` to submit a saved review, and `status` to read the historical result. Enrollment needs an explicit console from `--console` or `CARLOS_CONSOLE`. The CLI does not create operation IDs, fill choices, rewrite the roster, or choose a target.
+Use `preview`, `observe`, and `status` to read sponsored billing data. Use `enroll` to submit a saved review.
`carlos billing sponsored preview [--console URL] [--json]`
@@ -1308,6 +1308,16 @@ Account 1, `bab`, is marked as a system account. To enroll it, use `choice: enro
Preview output contains owner email addresses and account facts. You control where command output and manually saved files are shared. The CLI does not automatically save, cache, or upload this data. Diagnostics do not include raw rosters, owner lists, notes, tokens, database details, or corrupt receipt bodies.
+`carlos billing sponsored observe --account SQID --month YYYY-MM [--console URL] [--json]`
+
+An observation reads one stored statement for one selected enrolled account and calendar month. It is reported evidence only. It does not establish a complete baseline, funding level, allowance, growth boundary, freshness, publisher identity or authority, or verified billing.
+
+If no statement exists for the selected account and month, there is no observation for that selection. This does not mean zero usage or narrow approved funding. Open and closed are states reported by the source. Neither state is independent billing verification.
+
+Observed is when the console completed the read; Statement updated is the time reported by the source. Missing optional coverage stays unknown and does not mean zero or complete coverage. An optional ETag is an opaque store value and stays absent when unavailable. The statement SHA-256 identifies the exact stored bytes returned; it does not prove who published them. The command reports negative meter values, inconsistent day counts, and a source update time later than the observation time as anomalies without hiding or changing the stored evidence.
+
+The raw statement and account observation may contain sensitive usage data. You control where the output is shared. Diagnostics and logs do not contain statement bytes or provider errors.
+
`carlos billing sponsored enroll --roster FILE [--console URL] [--json]`
##### Roster file
diff --git a/internal/console/server.go b/internal/console/server.go
index 659c31886..ecba19703 100644
--- a/internal/console/server.go
+++ b/internal/console/server.go
@@ -1157,6 +1157,7 @@ func NewServer(cfg Config, st *Store) (*Server, error) {
s.mux.HandleFunc("GET /api/billing/house", s.handleAPIBillingHouseList)
s.mux.HandleFunc("POST /api/billing/house", s.handleAPIBillingHouseSet)
s.mux.HandleFunc("GET /api/billing/sponsored/preview", s.handleAPIBillingSponsoredPreview)
+ s.mux.HandleFunc("GET /api/billing/sponsored/observations/{account}/{month}", s.handleAPIBillingSponsoredObservation)
s.mux.HandleFunc("POST /api/billing/sponsored/enrollments", s.handleAPIBillingSponsoredEnroll)
s.mux.HandleFunc("GET /api/billing/sponsored/enrollments/{operation}", s.handleAPIBillingSponsoredStatus)
s.mux.HandleFunc("POST /a/{sqid}/errors/level", s.handleAccountErrorLevel)
diff --git a/internal/console/sponsoredobservation.go b/internal/console/sponsoredobservation.go
new file mode 100644
index 000000000..f9db636e9
--- /dev/null
+++ b/internal/console/sponsoredobservation.go
@@ -0,0 +1,515 @@
+package console
+
+import (
+ "bytes"
+ "context"
+ "crypto/sha256"
+ "database/sql"
+ "encoding/base64"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "github.com/carlosframework/platform/internal/domains"
+ "github.com/carlosframework/platform/internal/release"
+ "github.com/carlosframework/platform/internal/usage"
+ "io"
+ "net/http"
+ "os"
+ "sort"
+ "time"
+ "unicode/utf8"
+)
+
+const (
+ observationCopyAccountNotFound = "The selected account was not found or is not enrolled."
+ observationCopyStatementNotFound = "No stored statement exists for the selected account and month. This does not mean zero usage or change approved funding."
+ observationCopyUnsupportedStore = "The configured store cannot provide this observation."
+ observationCopyUnsupported = "The stored statement uses an unsupported version."
+ observationCopyInvalidStored = "The stored statement cannot be read safely."
+ observationCopyTooLarge = "The complete observation is too large to return. No partial result was returned."
+ observationCopyFailed = "The sponsored billing result could not be read."
+)
+
+type sponsoredObservation struct {
+ V int `json:"v"`
+ AccountID int64 `json:"account_id"`
+ Sqid string `json:"sqid"`
+ Month string `json:"month"`
+ FirstOperationID string `json:"first_operation_id"`
+ ReceiptDigest string `json:"receipt_digest"`
+ ObservedAt time.Time `json:"observed_at"`
+ DeletedAt int64 `json:"deleted_at"`
+ Retirement string `json:"retirement"`
+ SourceKey string `json:"source_key"`
+ ETag string `json:"etag,omitempty"`
+ StatementSHA256 string `json:"statement_sha256"`
+ StatementBase64 string `json:"statement_base64"`
+ Flags []string `json:"flags"`
+}
+
+func (out *sponsoredObservation) UnmarshalJSON(raw []byte) error {
+ var v sponsoredObservation
+ var etag *string
+ var observedAt json.RawMessage
+ if err := sponsoredWireObject(raw, false, map[string]sponsoredWireField{
+ "v": {target: &v.V}, "account_id": {target: &v.AccountID}, "sqid": {target: &v.Sqid}, "month": {target: &v.Month},
+ "first_operation_id": {target: &v.FirstOperationID}, "receipt_digest": {target: &v.ReceiptDigest},
+ "observed_at": {target: &observedAt}, "deleted_at": {target: &v.DeletedAt}, "retirement": {target: &v.Retirement},
+ "source_key": {target: &v.SourceKey}, "etag": {target: &etag, optional: true}, "statement_sha256": {target: &v.StatementSHA256},
+ "statement_base64": {target: &v.StatementBase64}, "flags": {target: &v.Flags},
+ }); err != nil {
+ return err
+ }
+ if json.Unmarshal(observedAt, &v.ObservedAt) != nil {
+ return errSponsoredWire
+ }
+ canonical, err := json.Marshal(v.ObservedAt.UTC().Format(time.RFC3339Nano))
+ if err != nil || !bytes.Equal(bytes.TrimSpace(observedAt), canonical) {
+ return errSponsoredWire
+ }
+ if etag != nil {
+ if *etag == "" {
+ return errSponsoredWire
+ }
+ v.ETag = *etag
+ }
+ *out = v
+ return nil
+}
+
+func sponsoredObservationMonth(s string) bool {
+ if len(s) != 7 || s[4] != '-' {
+ return false
+ }
+ for i, c := range s {
+ if i != 4 && (c < '0' || c > '9') {
+ return false
+ }
+ }
+ return s[5:] >= "01" && s[5:] <= "12"
+}
+
+// Walk every object, including additive fields, before decoding known fields.
+func sponsoredObservationJSON(raw []byte) bool {
+ if !utf8.Valid(raw) {
+ return false
+ }
+ d := json.NewDecoder(bytes.NewReader(raw))
+ d.UseNumber()
+ var walk func(int) bool
+ walk = func(depth int) bool {
+ tok, err := d.Token()
+ if err != nil {
+ return false
+ }
+ delim, ok := tok.(json.Delim)
+ if !ok {
+ return true
+ }
+ if depth >= 10000 {
+ return false
+ }
+ switch delim {
+ case '{':
+ seen := map[string]bool{}
+ for d.More() {
+ k, e := d.Token()
+ if e != nil {
+ return false
+ }
+ s, ok := k.(string)
+ if !ok || seen[s] {
+ return false
+ }
+ seen[s] = true
+ if !walk(depth + 1) {
+ return false
+ }
+ }
+ end, e := d.Token()
+ return e == nil && end == json.Delim('}')
+ case '[':
+ for d.More() {
+ if !walk(depth + 1) {
+ return false
+ }
+ }
+ end, e := d.Token()
+ return e == nil && end == json.Delim(']')
+ }
+ return false
+ }
+ return walk(0) && d.Decode(new(json.RawMessage)) == io.EOF
+}
+
+func sponsoredObservationNumberMap(raw json.RawMessage, days bool) bool {
+ if bytes.Equal(bytes.TrimSpace(raw), []byte("null")) {
+ return false
+ }
+ var m map[string]json.RawMessage
+ if json.Unmarshal(raw, &m) != nil || m == nil {
+ return false
+ }
+ for _, v := range m {
+ if days {
+ if !sponsoredObservationNumberMap(v, false) {
+ return false
+ }
+ } else {
+ if bytes.Equal(bytes.TrimSpace(v), []byte("null")) {
+ return false
+ }
+ var number float64
+ if json.Unmarshal(v, &number) != nil {
+ return false
+ }
+ }
+ }
+ return true
+}
+
+func sponsoredObservationStatement(raw []byte, account, month string, at time.Time) (usage.Statement, []string, string) {
+ var s usage.Statement
+ invalid := func() (usage.Statement, []string, string) { return usage.Statement{}, nil, "invalid_statement" }
+ if len(raw) > sponsoredWireRequestLimit || !sponsoredObservationJSON(raw) {
+ return invalid()
+ }
+ var version int
+ if sponsoredWireObject(raw, false, map[string]sponsoredWireField{"v": {target: &version}}) != nil {
+ return invalid()
+ }
+ if version != 1 {
+ return usage.Statement{}, nil, "unsupported_statement"
+ }
+ fields := map[string]sponsoredWireField{
+ "v": {target: &s.V}, "account": {target: &s.Account}, "month": {target: &s.Month}, "meters": {target: &s.Meters},
+ "covered_days": {target: &s.CoveredDays}, "days_in_month": {target: &s.DaysInMonth}, "updated_at": {target: &s.UpdatedAt},
+ "source_days": {target: &s.SourceDays, optional: true, nullable: true}, "days": {target: &s.Days, optional: true, nullable: true},
+ "closed": {target: &s.Closed, optional: true}, "evidence": {target: &s.Evidence, optional: true, nullable: true},
+ }
+ if sponsoredWireObject(raw, false, fields) != nil {
+ return invalid()
+ }
+ if s.Account != account || s.Month != month || !sponsoredObservationMonth(month) {
+ return invalid()
+ }
+ var root map[string]json.RawMessage
+ _ = json.Unmarshal(raw, &root)
+ if !sponsoredObservationNumberMap(root["meters"], false) {
+ return invalid()
+ }
+ for _, k := range []string{"source_days", "days"} {
+ if b, ok := root[k]; ok && !bytes.Equal(bytes.TrimSpace(b), []byte("null")) && !sponsoredObservationNumberMap(b, k == "days") {
+ return invalid()
+ }
+ }
+ if b, ok := root["evidence"]; ok && !bytes.Equal(bytes.TrimSpace(b), []byte("null")) {
+ var key, sha string
+ if sponsoredWireObject(b, false, map[string]sponsoredWireField{"key": {target: &key, optional: true}, "sha256": {target: &sha, optional: true}}) != nil {
+ return invalid()
+ }
+ }
+ flags := []string{}
+ negative := false
+ for _, v := range s.Meters {
+ negative = negative || v < 0
+ }
+ for _, day := range s.Days {
+ for _, v := range day {
+ negative = negative || v < 0
+ }
+ }
+ if negative {
+ flags = append(flags, "negative_meter_values")
+ }
+ date, _ := time.Parse("2006-01", month)
+ n := time.Date(date.Year(), date.Month()+1, 0, 0, 0, 0, 0, time.UTC).Day()
+ badDays := s.DaysInMonth != n || s.CoveredDays < 0 || s.CoveredDays > n
+ for _, v := range s.SourceDays {
+ badDays = badDays || v < 0 || v > n
+ }
+ if badDays {
+ flags = append(flags, "inconsistent_day_counts")
+ }
+ if s.UpdatedAt.After(at) {
+ flags = append(flags, "future_updated_at")
+ }
+ sort.Strings(flags)
+ return s, flags, ""
+}
+
+func sponsoredObservationErrorSpec(code string) (int, string) {
+ switch code {
+ case "invalid_request":
+ return 400, sponsoredCopyInvalidRequestError
+ case "account_not_found":
+ return 404, observationCopyAccountNotFound
+ case "statement_not_found":
+ return 404, observationCopyStatementNotFound
+ case "unsupported_store":
+ return 422, observationCopyUnsupportedStore
+ case "unsupported_statement":
+ return 422, observationCopyUnsupported
+ case "invalid_statement":
+ return 422, observationCopyInvalidStored
+ case "observation_too_large":
+ return 422, observationCopyTooLarge
+ default:
+ return 500, observationCopyFailed
+ }
+}
+
+func sponsoredObservationIdentity(ctx context.Context, s *Store, actor, sqid string) (sponsoredObservation, error) {
+ var out sponsoredObservation
+ tx, err := s.db.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
+ if err != nil {
+ return out, err
+ }
+ defer tx.Rollback()
+ if err = authorizeSponsoredEnrollment(ctx, tx, actor); err != nil {
+ return out, err
+ }
+ err = tx.QueryRowContext(ctx, `SELECT id,deleted_at FROM accounts WHERE sqid=?`, sqid).Scan(&out.AccountID, &out.DeletedAt)
+ if errors.Is(err, sql.ErrNoRows) {
+ return out, errSponsoredEnrollmentNotFound
+ }
+ if err != nil {
+ return out, err
+ }
+ if out.AccountID <= 0 || domains.Sqid(out.AccountID) != sqid {
+ return out, errSponsoredEnrollmentCorrupt
+ }
+ out.Sqid = sqid
+ first, err := sponsoredEnrollmentMember(ctx, tx, out.AccountID, sqid)
+ if err != nil {
+ return out, err
+ }
+ if first == "" {
+ return out, errSponsoredEnrollmentNotFound
+ }
+ receipt, err := readSponsoredEnrollmentTx(ctx, tx, first)
+ if err != nil {
+ return out, err
+ }
+ found := false
+ for _, row := range receipt.Rows {
+ if row.AccountID == out.AccountID && row.Sqid == sqid && row.Outcome == sponsoredAdded && row.FirstEnrollmentOperationID == first {
+ found = true
+ }
+ }
+ if !found {
+ return out, errSponsoredEnrollmentCorrupt
+ }
+ out.FirstOperationID = first
+ if err = tx.QueryRowContext(ctx, `SELECT body_digest FROM sponsored_enrollment_receipts WHERE operation_id=?`, first).Scan(&out.ReceiptDigest); err != nil {
+ return out, err
+ }
+ var committed int64
+ err = tx.QueryRowContext(ctx, `SELECT committed_at FROM billing_retirements WHERE account_id=?`, out.AccountID).Scan(&committed)
+ switch {
+ case errors.Is(err, sql.ErrNoRows):
+ out.Retirement = "none"
+ case err != nil:
+ return out, err
+ case committed == 0:
+ out.Retirement = "pending"
+ case committed > 0:
+ out.Retirement = "committed"
+ default:
+ return out, errSponsoredEnrollmentCorrupt
+ }
+ if err = tx.Commit(); err != nil {
+ return out, err
+ }
+ return out, nil
+}
+
+func sponsoredObservationRead(ctx context.Context, store release.Store, key string) ([]byte, string, string) {
+ var open func(context.Context, string) (io.ReadCloser, string, error)
+ switch s := store.(type) {
+ case *release.DirStore:
+ if s != nil {
+ open = s.Open
+ }
+ case *release.S3Store:
+ if s != nil && s.Client != nil {
+ open = s.Open
+ }
+ }
+ if open == nil {
+ return nil, "", "unsupported_store"
+ }
+ if ctx.Err() != nil {
+ return nil, "", "observation_failed"
+ }
+ body, etag, err := open(ctx, key)
+ if err != nil {
+ if body != nil {
+ _ = body.Close()
+ }
+ if errors.Is(err, os.ErrNotExist) {
+ return nil, "", "statement_not_found"
+ }
+ return nil, "", "observation_failed"
+ }
+ if body == nil {
+ return nil, "", "observation_failed"
+ }
+ raw, code := sponsoredObservationReadBody(ctx, body)
+ if code != "" {
+ return nil, "", code
+ }
+ return raw, etag, ""
+}
+
+func sponsoredObservationReadBody(ctx context.Context, body io.ReadCloser) ([]byte, string) {
+ var b bytes.Buffer
+ chunk := make([]byte, 32<<10)
+ code := ""
+ for {
+ if ctx.Err() != nil {
+ code = "observation_failed"
+ break
+ }
+ remaining := sponsoredWireRequestLimit + 1 - b.Len()
+ if remaining < len(chunk) {
+ chunk = chunk[:remaining]
+ }
+ n, err := body.Read(chunk)
+ if ctx.Err() != nil {
+ code = "observation_failed"
+ break
+ }
+ if n > 0 {
+ b.Write(chunk[:n])
+ }
+ if b.Len() > sponsoredWireRequestLimit {
+ code = "observation_too_large"
+ break
+ }
+ if err == io.EOF {
+ break
+ }
+ if err != nil {
+ code = "observation_failed"
+ break
+ }
+ }
+ if err := body.Close(); err != nil {
+ code = "observation_failed"
+ }
+ if ctx.Err() != nil {
+ code = "observation_failed"
+ }
+ if code != "" {
+ return nil, code
+ }
+ return b.Bytes(), ""
+}
+
+func sponsoredObservationError(w http.ResponseWriter, code string) {
+ status, message := sponsoredObservationErrorSpec(code)
+ w.Header().Set("Content-Type", "application/json")
+ w.Header().Set("Cache-Control", "no-store")
+ w.WriteHeader(status)
+ _ = json.NewEncoder(w).Encode(struct {
+ Code string `json:"code"`
+ Error string `json:"error"`
+ }{code, message})
+}
+
+func (s *Server) handleAPIBillingSponsoredObservation(w http.ResponseWriter, r *http.Request) {
+ actor, ok := s.billingOperator(w, r)
+ if !ok {
+ return
+ }
+ if r.Method != http.MethodGet {
+ w.Header().Set("Allow", http.MethodGet)
+ w.WriteHeader(http.StatusMethodNotAllowed)
+ return
+ }
+ // Query/body validation retains the existing helper, buffering its refusal
+ // so this door can add its finite machine code without changing the helper.
+ q := &sponsoredObservationQueryWriter{header: make(http.Header)}
+ if !sponsoredOperatorQuery(q, r) {
+ sponsoredObservationError(w, "invalid_request")
+ return
+ }
+ sqid, month := r.PathValue("account"), r.PathValue("month")
+ if !domains.ValidSqid(sqid) || !billingMonthPattern.MatchString(month) || !sponsoredObservationMonth(month) {
+ sponsoredObservationError(w, "invalid_request")
+ return
+ }
+ ctx, cancel := context.WithTimeout(r.Context(), 30*time.Second)
+ defer cancel()
+ first, err := sponsoredObservationIdentity(ctx, s.store, actor, sqid)
+ if err != nil {
+ if errors.Is(err, errSponsoredEnrollmentDenied) {
+ http.NotFound(w, r)
+ } else if errors.Is(err, errSponsoredEnrollmentNotFound) {
+ sponsoredObservationError(w, "account_not_found")
+ } else {
+ sponsoredObservationError(w, "observation_failed")
+ }
+ return
+ }
+ key := usage.StatementKey(sqid, month)
+ raw, etag, code := sponsoredObservationRead(ctx, s.cfg.Releases, key)
+ if code == "" && !utf8.ValidString(etag) {
+ code = "observation_failed"
+ }
+ observed := time.Now().UTC()
+ var flags []string
+ if code == "" {
+ _, flags, code = sponsoredObservationStatement(raw, sqid, month, observed)
+ }
+ last, err := sponsoredObservationIdentity(ctx, s.store, actor, sqid)
+ if err != nil {
+ if errors.Is(err, errSponsoredEnrollmentDenied) {
+ http.NotFound(w, r)
+ } else {
+ sponsoredObservationError(w, "observation_failed")
+ }
+ return
+ }
+ if last.AccountID != first.AccountID || last.Sqid != first.Sqid || last.FirstOperationID != first.FirstOperationID || last.ReceiptDigest != first.ReceiptDigest {
+ sponsoredObservationError(w, "observation_failed")
+ return
+ }
+ if code != "" {
+ sponsoredObservationError(w, code)
+ return
+ }
+ last.V = 1
+ last.Month = month
+ last.ObservedAt = observed
+ last.SourceKey = key
+ last.ETag = etag
+ last.Flags = flags
+ digest := sha256.Sum256(raw)
+ last.StatementSHA256 = hex.EncodeToString(digest[:])
+ last.StatementBase64 = base64.StdEncoding.EncodeToString(raw)
+ // Marshal/size checks happen before any success header; deadline covers it.
+ if _, err = sponsoredOperatorEncode(last); err != nil {
+ if errors.Is(err, errSponsoredEnrollmentTooLarge) {
+ sponsoredObservationError(w, "observation_too_large")
+ } else {
+ sponsoredObservationError(w, "observation_failed")
+ }
+ return
+ }
+ if ctx.Err() != nil {
+ sponsoredObservationError(w, "observation_failed")
+ return
+ }
+ started, err := sponsoredOperatorSuccess(w, last)
+ if !started && err != nil {
+ sponsoredObservationError(w, "observation_failed")
+ }
+}
+
+type sponsoredObservationQueryWriter struct{ header http.Header }
+
+func (w *sponsoredObservationQueryWriter) Header() http.Header { return w.header }
+func (w *sponsoredObservationQueryWriter) WriteHeader(int) {}
+func (w *sponsoredObservationQueryWriter) Write(b []byte) (int, error) { return len(b), nil }
diff --git a/internal/console/sponsoredobservation_test.go b/internal/console/sponsoredobservation_test.go
new file mode 100644
index 000000000..dfd9880e2
--- /dev/null
+++ b/internal/console/sponsoredobservation_test.go
@@ -0,0 +1,591 @@
+package console
+
+import (
+ "bytes"
+ "context"
+ "encoding/base64"
+ "encoding/json"
+ "errors"
+ "io"
+ "net/http"
+ "net/http/httptest"
+ "os"
+ "os/exec"
+ "path/filepath"
+ "reflect"
+ "runtime/debug"
+ "strings"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/aws/aws-sdk-go-v2/aws"
+ "github.com/aws/aws-sdk-go-v2/credentials"
+ awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
+ "github.com/carlosframework/platform/internal/domains"
+ "github.com/carlosframework/platform/internal/release"
+ "github.com/carlosframework/platform/internal/usage"
+)
+
+func observationFixture(t *testing.T) (*consoleFixture, string, []byte) {
+ t.Helper()
+ fx := newConsoleFixture(t)
+ token := billingOperatorToken(t, fx, "operator@example.test", "observation")
+ req := operatorRequest(t)
+ req.Items[0].AccountID = 1
+ req.Items[0].Sqid = "bab"
+ req.Items[0].ConfirmSystemAccount = true
+ status, _, raw := operatorHTTP(t, fx, "POST", "/api/billing/sponsored/enrollments", token, operatorJSON(t, req))
+ if status != 200 {
+ t.Fatalf("enrollment: %d %s", status, raw)
+ }
+ var sync int
+ if e := fx.st.db.QueryRow(`PRAGMA synchronous`).Scan(&sync); e != nil || sync != 2 {
+ t.Fatalf("FULL: %d %v", sync, e)
+ }
+ b, e := os.ReadFile("testdata/sponsored-observation-statement.json")
+ if e != nil {
+ t.Fatal(e)
+ }
+ return fx, token, b
+}
+func observationStore(t *testing.T, fx *consoleFixture, raw []byte) {
+ t.Helper()
+ d := release.NewDirStore(t.TempDir())
+ if _, e := d.Put(context.Background(), usage.StatementKey("bab", "2026-10"), raw, ""); e != nil {
+ t.Fatal(e)
+ }
+ fx.srv.cfg.Releases = d
+}
+
+const observationPath = "/api/billing/sponsored/observations/bab/2026-10"
+
+func observationCode(t *testing.T, b []byte) string {
+ t.Helper()
+ var v struct {
+ Code string `json:"code"`
+ }
+ if json.Unmarshal(b, &v) != nil {
+ t.Fatalf("bad error %q", b)
+ }
+ return v.Code
+}
+
+func TestSponsoredObservationDoor(t *testing.T) {
+ fx, token, raw := observationFixture(t)
+ observationStore(t, fx, raw)
+ before := operatorSnapshot(t, fx.st)
+ status, h, b := operatorHTTP(t, fx, "GET", observationPath, token, nil)
+ if status != 200 || h.Get("Cache-Control") != "no-store" {
+ t.Fatalf("%d %s", status, b)
+ }
+ var v sponsoredObservation
+ if json.Unmarshal(b, &v) != nil {
+ t.Fatal("decode")
+ }
+ decoded, e := base64.StdEncoding.DecodeString(v.StatementBase64)
+ if e != nil || !bytes.Equal(decoded, raw) || v.ETag != "" || v.Sqid != "bab" || v.Retirement != "none" || v.SourceKey != usage.StatementKey("bab", "2026-10") || !reflect.DeepEqual(v.Flags, []string{"future_updated_at", "inconsistent_day_counts", "negative_meter_values"}) {
+ t.Fatalf("wrong observation %+v %v", v, e)
+ }
+ var fields map[string]json.RawMessage
+ json.Unmarshal(b, &fields)
+ if _, ok := fields["etag"]; ok {
+ t.Fatal("Dir ETag synthesized")
+ }
+ if !reflect.DeepEqual(before, operatorSnapshot(t, fx.st)) {
+ t.Fatal("read mutated SQL")
+ }
+ sponsoredExec(t, fx.st.db, `UPDATE accounts SET deleted_at=99 WHERE id=1`)
+ status, _, b = operatorHTTP(t, fx, "GET", observationPath, token, nil)
+ if status != 200 || json.Unmarshal(b, &v) != nil || v.DeletedAt != 99 {
+ t.Fatalf("deleted history %d %s", status, b)
+ }
+ for _, committed := range []int64{0, 20} {
+ sponsoredExec(t, fx.st.db, `INSERT OR REPLACE INTO billing_retirements(account_id,started_at,committed_at) VALUES(1,10,?)`, committed)
+ status, _, b = operatorHTTP(t, fx, "GET", observationPath, token, nil)
+ if status != 200 || json.Unmarshal(b, &v) != nil {
+ t.Fatalf("retirement history %d %s", status, b)
+ }
+ want := "pending"
+ if committed > 0 {
+ want = "committed"
+ }
+ if v.Retirement != want {
+ t.Fatal(v.Retirement)
+ }
+ }
+}
+
+func TestSponsoredObservationAuthAndRequests(t *testing.T) {
+ fx, token, _ := observationFixture(t)
+ trip := &observationTripwire{}
+ member := cliToken(t, fx, "member@example.test", "observation-member")
+ fx.srv.cfg.Releases = trip
+ _, wantH, wantB := operatorHTTP(t, fx, "GET", "/unknown", member, nil)
+ for _, target := range []string{observationPath, observationPath + "?x=1", "/api/billing/sponsored/observations/bab/2026-13"} {
+ code, h, b := operatorHTTP(t, fx, "GET", target, member, []byte("x"))
+ if code != 404 || !reflect.DeepEqual(h, wantH) || !bytes.Equal(b, wantB) {
+ t.Fatal("hidden door", code, h, string(b))
+ }
+ }
+ for _, credential := range []string{"", "fake"} {
+ code, _, _ := operatorHTTP(t, fx, "GET", observationPath, credential, nil)
+ if code != 401 {
+ t.Fatal(code)
+ }
+ }
+ for _, tc := range []struct {
+ path string
+ body []byte
+ }{{observationPath + "?x", nil}, {observationPath, []byte("x")}, {"/api/billing/sponsored/observations/iii/2026-10", nil}, {"/api/billing/sponsored/observations/bab/2026-13", nil}} {
+ code, _, b := operatorHTTP(t, fx, "GET", tc.path, token, tc.body)
+ if code != 400 || observationCode(t, b) != "invalid_request" {
+ t.Fatal(code, string(b))
+ }
+ }
+ status, _, b := operatorHTTP(t, fx, "GET", observationPath, token, nil)
+ if status != 422 || observationCode(t, b) != "unsupported_store" {
+ t.Fatal(status, string(b))
+ }
+ status, _, b = operatorHTTP(t, fx, "GET", "/api/billing/sponsored/observations/"+domains.Sqid(100)+"/2026-10", token, nil)
+ if status != 404 || observationCode(t, b) != "account_not_found" {
+ t.Fatal(status, string(b))
+ }
+ if trip.calls.Load() != 0 {
+ t.Fatal("unsupported store called")
+ }
+}
+
+type observationTripwire struct {
+ release.Store
+ calls atomic.Int32
+}
+
+func (s *observationTripwire) Get(context.Context, string) ([]byte, string, error) {
+ s.calls.Add(1)
+ return nil, "", errors.New("private provider")
+}
+func (s *observationTripwire) Open(context.Context, string) (io.ReadCloser, string, error) {
+ s.calls.Add(1)
+ return nil, "", errors.New("private provider")
+}
+
+type observationGetOnly struct {
+ release.Store
+ calls atomic.Int32
+}
+
+func (s *observationGetOnly) Get(context.Context, string) ([]byte, string, error) {
+ s.calls.Add(1)
+ return nil, "", nil
+}
+func observationS3(t *testing.T, h http.HandlerFunc) *release.S3Store {
+ t.Helper()
+ srv := httptest.NewServer(h)
+ t.Cleanup(srv.Close)
+ client := awss3.New(awss3.Options{Region: "us-east-1", Credentials: aws.NewCredentialsCache(credentials.NewStaticCredentialsProvider("dummy", "dummy", "")), BaseEndpoint: aws.String(srv.URL), UsePathStyle: true, RetryMaxAttempts: 1, HTTPClient: srv.Client()})
+ return &release.S3Store{Client: client, Bucket: "fixture"}
+}
+
+func TestSponsoredObservationStoresAndBoundaries(t *testing.T) {
+ trip := &observationTripwire{}
+ get := &observationGetOnly{}
+ var dir *release.DirStore
+ var s3 *release.S3Store
+ for _, s := range []release.Store{nil, dir, s3, &release.S3Store{}, trip, get, release.NewPrefixStore(trip, "p"), release.NewPrefixStore(release.NewDirStore(t.TempDir()), "p")} {
+ if _, _, code := sponsoredObservationRead(context.Background(), s, "a"); code != "unsupported_store" {
+ t.Fatal(code)
+ }
+ }
+ if trip.calls.Load() != 0 || get.calls.Load() != 0 {
+ t.Fatal("unsupported adapter called")
+ }
+ for _, size := range []int{sponsoredWireRequestLimit - 1, sponsoredWireRequestLimit, sponsoredWireRequestLimit + 1} {
+ payload := bytes.Repeat([]byte("x"), size)
+ for _, kind := range []string{"dir", "s3"} {
+ var store release.Store
+ if kind == "dir" {
+ d := release.NewDirStore(t.TempDir())
+ d.Put(context.Background(), "a", payload, "")
+ store = d
+ } else {
+ store = observationS3(t, func(w http.ResponseWriter, r *http.Request) { w.Header().Set("ETag", `"opaque"`); w.Write(payload) })
+ }
+ raw, etag, code := sponsoredObservationRead(context.Background(), store, "a")
+ if size > sponsoredWireRequestLimit {
+ if code != "observation_too_large" || raw != nil {
+ t.Fatal(kind, code)
+ }
+ } else if code != "" || !bytes.Equal(raw, payload) || (kind == "dir" && etag != "") || (kind == "s3" && etag != `"opaque"`) {
+ t.Fatal(kind, code, etag)
+ }
+ }
+ }
+}
+
+type observationReadCloser struct {
+ r io.Reader
+ closed int
+ closeErr error
+ after func()
+}
+
+func (r *observationReadCloser) Read(b []byte) (int, error) {
+ n, e := r.r.Read(b)
+ if r.after != nil {
+ r.after()
+ }
+ return n, e
+}
+func (r *observationReadCloser) Close() error { r.closed++; return r.closeErr }
+func TestSponsoredObservationCancellationAndClose(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+ body := &observationReadCloser{r: strings.NewReader(strings.Repeat("x", 65536)), after: cancel}
+ if b, c := sponsoredObservationReadBody(ctx, body); b != nil || c != "observation_failed" || body.closed != 1 {
+ t.Fatal(c, body.closed)
+ }
+ body = &observationReadCloser{r: strings.NewReader("valid"), closeErr: errors.New("private close")}
+ if b, c := sponsoredObservationReadBody(context.Background(), body); b != nil || c != "observation_failed" || body.closed != 1 {
+ t.Fatal(c, body.closed)
+ }
+ entered := make(chan struct{})
+ store := observationS3(t, func(w http.ResponseWriter, r *http.Request) {
+ w.WriteHeader(200)
+ w.(http.Flusher).Flush()
+ close(entered)
+ <-r.Context().Done()
+ })
+ ctx, cancel = context.WithCancel(context.Background())
+ done := make(chan string, 1)
+ go func() { _, _, c := sponsoredObservationRead(ctx, store, "a"); done <- c }()
+ <-entered
+ cancel()
+ select {
+ case c := <-done:
+ if c != "observation_failed" {
+ t.Fatal(c)
+ }
+ case <-time.After(5 * time.Second):
+ t.Fatal("cancel blocked")
+ }
+}
+
+func TestSponsoredObservationRechecksAfterEveryReadOutcome(t *testing.T) {
+ for _, outcome := range []string{"success", "absent", "invalid", "version", "oversize", "read_error", "deep"} {
+ t.Run(outcome, func(t *testing.T) {
+ fx, token, raw := observationFixture(t)
+ entered, releaseRead := make(chan struct{}), make(chan struct{})
+ fx.srv.cfg.Releases = observationS3(t, func(w http.ResponseWriter, r *http.Request) {
+ close(entered)
+ <-releaseRead
+ if outcome == "absent" {
+ w.WriteHeader(404)
+ io.WriteString(w, `<Error><Code>NoSuchKey</Code></Error>`)
+ } else if outcome == "deep" {
+ w.Write(bytes.Repeat([]byte("["), sponsoredWireRequestLimit))
+ } else if outcome == "version" {
+ w.Write(bytes.Replace(raw, []byte(`"v":1`), []byte(`"v":2`), 1))
+ } else if outcome == "oversize" {
+ w.Write(bytes.Repeat([]byte("x"), sponsoredWireRequestLimit+1))
+ } else if outcome == "read_error" {
+ w.WriteHeader(500)
+ io.WriteString(w, "private provider error")
+ } else if outcome == "invalid" {
+ io.WriteString(w, "private corrupt statement")
+ } else {
+ w.Write(raw)
+ }
+ })
+ type result struct {
+ code int
+ headers http.Header
+ body []byte
+ }
+ done := make(chan result, 1)
+ go func() {
+ code, h, b := operatorHTTP(t, fx, "GET", observationPath, token, nil)
+ done <- result{code, h, b}
+ }()
+ <-entered
+ ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+ defer cancel()
+ if _, e := fx.st.db.ExecContext(ctx, `UPDATE users SET operator=0 WHERE address=?`, "operator@example.test"); e != nil {
+ close(releaseRead)
+ t.Fatal("SQL held during I/O", e)
+ }
+ close(releaseRead)
+ v := <-done
+ _, h, b := operatorHTTP(t, fx, "GET", "/unknown", token, nil)
+ if v.code != 404 || !reflect.DeepEqual(v.headers, h) || !bytes.Equal(v.body, b) {
+ t.Fatal("post-read auth", v.code, string(v.body))
+ }
+ })
+ }
+}
+
+func TestSponsoredObservationStatementSemantics(t *testing.T) {
+ raw, e := os.ReadFile("testdata/sponsored-observation-statement.json")
+ if e != nil {
+ t.Fatal(e)
+ }
+ at := time.Date(2026, 10, 15, 0, 0, 0, 0, time.UTC)
+ _, flags, c := sponsoredObservationStatement(raw, "bab", "2026-10", at)
+ if c != "" || len(flags) != 3 {
+ t.Fatal(c, flags)
+ }
+ for _, tc := range []struct{ from, to, code string }{
+ {`"v":1`, `"v":2`, "unsupported_statement"}, {`"account":"bab"`, `"account":"bad"`, "invalid_statement"},
+ {`"meters":{"instance_hours":-1}`, `"meters":null`, "invalid_statement"},
+ {`"v":1`, `"v":1,"v":1`, "invalid_statement"}, {`"v":1`, `"V":1`, "invalid_statement"},
+ {`"unknown":{"html":"<&>"}`, `"unknown":{"x":1,"x":2}`, "invalid_statement"},
+ {`"updated_at":"2100-01-01T00:00:00Z"`, `"updated_at":null`, "invalid_statement"},
+ } {
+ b := bytes.Replace(raw, []byte(tc.from), []byte(tc.to), 1)
+ if bytes.Equal(b, raw) {
+ t.Fatal("fixture replacement", tc.from)
+ }
+ _, _, c := sponsoredObservationStatement(b, "bab", "2026-10", at)
+ if c != tc.code {
+ t.Fatal(tc.from, c)
+ }
+ }
+ for _, b := range [][]byte{append(append([]byte{}, raw...), []byte(`{}`)...), []byte("\xff"), []byte(`null`)} {
+ if _, _, c := sponsoredObservationStatement(b, "bab", "2026-10", at); c != "invalid_statement" {
+ t.Fatal(c)
+ }
+ }
+}
+
+func TestSponsoredObservationWholeReceipt(t *testing.T) {
+ fx, token, raw := observationFixture(t)
+ observationStore(t, fx, raw)
+ // The named membership still exists, but its parent no longer accounts for all children.
+ sponsoredInsertAccount(t, fx.st.db, 100, "sibling", 10, 0)
+ var first string
+ fx.st.db.QueryRow(`SELECT first_operation_id FROM sponsored_cohort_memberships WHERE account_id=1`).Scan(&first)
+ sponsoredExec(t, fx.st.db, `INSERT INTO sponsored_cohort_memberships(account_id,sqid,first_operation_id) VALUES(100,?,?)`, domains.Sqid(100), first)
+ code, _, b := operatorHTTP(t, fx, "GET", observationPath, token, nil)
+ if code != 500 || observationCode(t, b) != "observation_failed" {
+ t.Fatal(code, string(b))
+ }
+}
+
+func TestSponsoredObservationResponseBound(t *testing.T) {
+ fx, token, raw := observationFixture(t)
+ // Enforce the final envelope cap even when provider metadata alone is huge.
+ fx.srv.cfg.Releases = observationS3(t, func(w http.ResponseWriter, r *http.Request) {
+ w.Header().Set("ETag", strings.Repeat("a", sponsoredWireResponseLimit))
+ w.Write(raw)
+ })
+ code, _, _ := operatorHTTP(t, fx, "GET", observationPath, token, nil)
+ // net/http's header limit may refuse first; either safe refusal is valid here.
+ if code != 422 && code != 500 {
+ t.Fatal(code)
+ }
+ for _, extra := range []int{-1, 0, 1} {
+ v := sponsoredObservation{Flags: []string{}, ETag: ""}
+ b, _ := json.Marshal(v)
+ v.ETag = strings.Repeat("x", sponsoredWireResponseLimit-len(b)-len(`,"etag":""`)+extra)
+ b, e := sponsoredOperatorEncode(v)
+ if extra <= 0 {
+ if e != nil || len(b) != sponsoredWireResponseLimit+extra {
+ t.Fatal(extra, len(b), e)
+ }
+ } else if !errors.Is(e, errSponsoredEnrollmentTooLarge) {
+ t.Fatal(e)
+ }
+ }
+}
+
+func TestSponsoredObservationSemanticSiblingCorruption(t *testing.T) {
+ fx := newConsoleFixture(t)
+ token := billingOperatorToken(t, fx, "operator@example.test", "siblings")
+ sponsoredInsertAccount(t, fx.st.db, 100, "sibling", 1, 0)
+ req := operatorRequest(t)
+ req.Items[0].AccountID = 1
+ req.Items[0].Sqid = "bab"
+ req.Items[0].ConfirmSystemAccount = true
+ sibling := req.Items[0]
+ sibling.AccountID = 100
+ sibling.Sqid = domains.Sqid(100)
+ sibling.ConfirmSystemAccount = false
+ req.Items = append(req.Items, sibling)
+ status, _, b := operatorHTTP(t, fx, "POST", "/api/billing/sponsored/enrollments", token, operatorJSON(t, req))
+ if status != 200 {
+ t.Fatal(status, string(b))
+ }
+ raw, e := os.ReadFile("testdata/sponsored-observation-statement.json")
+ if e != nil {
+ t.Fatal(e)
+ }
+ observationStore(t, fx, raw)
+ var body []byte
+ fx.st.db.QueryRow(`SELECT body FROM sponsored_enrollment_receipts WHERE operation_id=?`, req.OperationID).Scan(&body)
+ var receipt sponsoredEnrollmentReceipt
+ if json.Unmarshal(body, &receipt) != nil {
+ t.Fatal("receipt")
+ }
+ selected := receipt.Rows[0]
+ receipt.Rows[1].Outcome = sponsoredExcludedByOperator
+ body = operatorJSON(t, receipt)
+ sponsoredExec(t, fx.st.db, `DROP TRIGGER sponsored_receipts_no_update`)
+ sponsoredExec(t, fx.st.db, `UPDATE sponsored_enrollment_receipts SET body=?,body_digest=? WHERE operation_id=?`, body, sponsoredEnrollmentDigest("receipt", body), req.OperationID)
+ if !reflect.DeepEqual(selected, receipt.Rows[0]) {
+ t.Fatal("selected row changed")
+ }
+ status, _, b = operatorHTTP(t, fx, "GET", observationPath, token, nil)
+ if status != 500 || observationCode(t, b) != "observation_failed" {
+ t.Fatal("sibling corruption bypassed", status, string(b))
+ }
+}
+
+func TestSponsoredObservationBindingDriftAfterRead(t *testing.T) {
+ fx, token, raw := observationFixture(t)
+ entered, resume := make(chan struct{}), make(chan struct{})
+ fx.srv.cfg.Releases = observationS3(t, func(w http.ResponseWriter, r *http.Request) { close(entered); <-resume; w.Write(raw) })
+ done := make(chan int, 1)
+ go func() { code, _, _ := operatorHTTP(t, fx, "GET", observationPath, token, nil); done <- code }()
+ <-entered
+ // Fixture-only corruption while I/O is blocked; selected actor remains authorized.
+ sponsoredExec(t, fx.st.db, `DROP TRIGGER sponsored_members_no_delete`)
+ sponsoredExec(t, fx.st.db, `DELETE FROM sponsored_cohort_memberships WHERE account_id=1`)
+ close(resume)
+ if code := <-done; code != 500 {
+ t.Fatal("binding drift", code)
+ }
+}
+
+func TestSponsoredObservationReportedFacts(t *testing.T) {
+ raw, e := os.ReadFile("testdata/sponsored-observation-statement.json")
+ if e != nil {
+ t.Fatal(e)
+ }
+ var fields map[string]json.RawMessage
+ json.Unmarshal(raw, &fields)
+ fields["meters"] = json.RawMessage(`{}`)
+ fields["covered_days"] = json.RawMessage(`0`)
+ fields["days_in_month"] = json.RawMessage(`31`)
+ fields["updated_at"] = json.RawMessage(`"2026-10-01T00:00:00Z"`)
+ for _, k := range []string{"source_days", "days", "evidence"} {
+ delete(fields, k)
+ }
+ for _, closed := range []string{"false", "true"} {
+ fields["closed"] = json.RawMessage(closed)
+ b, _ := json.Marshal(fields)
+ s, flags, code := sponsoredObservationStatement(b, "bab", "2026-10", time.Date(2026, 10, 15, 0, 0, 0, 0, time.UTC))
+ if code != "" || len(flags) != 0 || s.SourceDays != nil {
+ t.Fatal(code, flags)
+ }
+ }
+ for _, k := range []string{"source_days", "days"} {
+ bad := map[string]json.RawMessage{}
+ for key, v := range fields {
+ bad[key] = v
+ }
+ bad[k] = json.RawMessage(`{"x":null}`)
+ b, _ := json.Marshal(bad)
+ if _, _, code := sponsoredObservationStatement(b, "bab", "2026-10", time.Now()); code != "invalid_statement" {
+ t.Fatal(k, code)
+ }
+ }
+}
+
+func TestSponsoredObservationHEADDoesNotObserve(t *testing.T) {
+ fx, token, raw := observationFixture(t)
+ member := cliToken(t, fx, "member@example.test", "head-member")
+ var calls atomic.Int32
+ fx.srv.cfg.Releases = observationS3(t, func(w http.ResponseWriter, r *http.Request) { calls.Add(1); w.Write(raw) })
+ for _, path := range []string{observationPath, observationPath + "?bad"} {
+ code, h, b := operatorHTTP(t, fx, "HEAD", path, token, nil)
+ if code != 405 || h.Get("Allow") != "GET" || len(b) != 0 || calls.Load() != 0 {
+ t.Fatalf("HEAD performed observation: status=%d allow=%q bytes=%d storage_calls=%d", code, h.Get("Allow"), len(b), calls.Load())
+ }
+ }
+ _, wantH, wantB := operatorHTTP(t, fx, "HEAD", "/unknown", member, nil)
+ code, h, b := operatorHTTP(t, fx, "HEAD", observationPath, member, nil)
+ if code != 404 || !reflect.DeepEqual(h, wantH) || !bytes.Equal(b, wantB) || calls.Load() != 0 {
+ t.Fatal("HEAD hidden gate", code)
+ }
+ code, _, _ = operatorHTTP(t, fx, "HEAD", observationPath, "", nil)
+ if code != 401 || calls.Load() != 0 {
+ t.Fatal("HEAD authentication", code)
+ }
+}
+
+func TestSponsoredObservationVersionDispatch(t *testing.T) {
+ for _, tc := range []struct{ raw, code string }{
+ {`{"v":2,"next_shape":[1,2,3]}`, "unsupported_statement"},
+ {`{"v":2,"meters":"different future shape"}`, "unsupported_statement"},
+ {`{"v":null}`, "invalid_statement"}, {`{"v":"2"}`, "invalid_statement"},
+ {`{"v":2,"next":{"x":1,"x":2}}`, "invalid_statement"},
+ } {
+ if _, _, code := sponsoredObservationStatement([]byte(tc.raw), "bab", "2026-10", time.Now()); code != tc.code {
+ t.Fatalf("%s: got %s want %s", tc.raw, code, tc.code)
+ }
+ }
+}
+
+func TestSponsoredObservationDepthIsolated(t *testing.T) {
+ if os.Getenv("SPONSORED_OBSERVATION_DEPTH_CHILD") == "1" {
+ debug.SetMaxStack(32 << 20)
+ raw := bytes.Repeat([]byte("["), sponsoredWireRequestLimit)
+ if _, _, code := sponsoredObservationStatement(raw, "bab", "2026-10", time.Time{}); code != "invalid_statement" {
+ t.Fatalf("depth refusal=%s", code)
+ }
+ return
+ }
+ executable, err := os.Executable()
+ if err != nil {
+ t.Fatal(err)
+ }
+ dir := t.TempDir()
+ ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
+ defer cancel()
+ cmd := exec.CommandContext(ctx, executable, "-test.run=^TestSponsoredObservationDepthIsolated$", "-test.count=1")
+ cmd.Dir = dir
+ // This is the test binary, not carlos; inherit no login or provider environment.
+ cmd.Env = []string{"SPONSORED_OBSERVATION_DEPTH_CHILD=1", "HOME=" + dir, "TMPDIR=" + dir, "PATH=/usr/bin:/bin", "TZ=UTC", "CARLOS_CONFIG=", "CARLOS_CREDENTIALS=", "CARLOS_CONSOLE=", "CARLOS_DESTINATION=", "AWS_EC2_METADATA_DISABLED=true", "AWS_CONFIG_FILE=" + filepath.Join(dir, "absent-config"), "AWS_SHARED_CREDENTIALS_FILE=" + filepath.Join(dir, "absent-credentials")}
+ output, err := cmd.CombinedOutput()
+ if ctx.Err() != nil {
+ t.Fatalf("isolated depth child timed out: %v", ctx.Err())
+ }
+ if err != nil {
+ t.Fatalf("isolated depth child failed: %v\n%s", err, output)
+ }
+}
+
+func TestSponsoredObservationDepthBoundary(t *testing.T) {
+ for _, depth := range []int{9999, 10000, 10001} {
+ raw := []byte(strings.Repeat("[", depth) + "0" + strings.Repeat("]", depth))
+ if got := sponsoredObservationJSON(raw); got != (depth <= 10000) {
+ t.Fatalf("array depth %d accepted=%v", depth, got)
+ }
+ if depth > 10000 {
+ if _, _, code := sponsoredObservationStatement(raw, "bab", "2026-10", time.Time{}); code != "invalid_statement" {
+ t.Fatal(code)
+ }
+ }
+ }
+ for _, arrays := range []int{9999, 10000, 10001} {
+ raw := []byte(`{"x":` + strings.Repeat("[", arrays) + "0" + strings.Repeat("]", arrays) + "}")
+ if got := sponsoredObservationJSON(raw); got != (arrays+1 <= 10000) {
+ t.Fatalf("object plus %d arrays accepted=%v", arrays, got)
+ }
+ if _, _, code := sponsoredObservationStatement(raw, "bab", "2026-10", time.Time{}); code != "invalid_statement" {
+ t.Fatal(code)
+ }
+ }
+}
+
+func TestSponsoredObservationDepthDirRoute(t *testing.T) {
+ fx, token, _ := observationFixture(t)
+ observationStore(t, fx, bytes.Repeat([]byte("["), sponsoredWireRequestLimit))
+ before := operatorSnapshot(t, fx.st)
+ status, _, body := operatorHTTP(t, fx, "GET", observationPath, token, nil)
+ if status != 422 || observationCode(t, body) != "invalid_statement" {
+ t.Fatalf("depth route: %d %s", status, body)
+ }
+ if !reflect.DeepEqual(before, operatorSnapshot(t, fx.st)) {
+ t.Fatal("malformed statement read mutated SQL")
+ }
+}
diff --git a/internal/console/sponsoredoperator.go b/internal/console/sponsoredoperator.go
index cfbe0efa2..3a7a37b48 100644
--- a/internal/console/sponsoredoperator.go
+++ b/internal/console/sponsoredoperator.go
@@ -18,7 +18,7 @@ import (
const (
sponsoredCopyUsageHelpHeading = "Sponsored account enrollment"
sponsoredCopyUsageDocsHeading = "`carlos billing sponsored`"
- sponsoredCopyUsageHelpSummary = "Use `preview` to read an account snapshot, `enroll` to submit a saved review, and `status` to read the historical result. Enrollment needs an explicit console from `--console` or `CARLOS_CONSOLE`. The CLI does not create operation IDs, fill choices, rewrite the roster, or choose a target."
+ sponsoredCopyUsageHelpSummary = "Use `preview`, `observe`, and `status` to read sponsored billing data. Use `enroll` to submit a saved review."
sponsoredCopyUsageUsagePreview = "`carlos billing sponsored preview [--console URL] [--json]`"
sponsoredCopyUsageUsageEnroll = "`carlos billing sponsored enroll --roster FILE [--console URL] [--json]`"
sponsoredCopyUsageUsageStatus = "`carlos billing sponsored status --operation 64hex [--console URL] [--json]`"
diff --git a/internal/console/testdata/sponsored-observation-statement.json b/internal/console/testdata/sponsored-observation-statement.json
new file mode 100644
index 000000000..5c4140e9b
--- /dev/null
+++ b/internal/console/testdata/sponsored-observation-statement.json
@@ -0,0 +1 @@
+{ "v":1,"account":"bab","month":"2026-10","meters":{"instance_hours":-1},"covered_days":32,"days_in_month":30,"updated_at":"2100-01-01T00:00:00Z","source_days":null,"days":null,"evidence":null,"closed":false,"unknown":{"html":"<&>"}}