From d69da42122077166bbb0083632a1ed2d530963da Mon Sep 17 00:00:00 2001 From: Wes McKinney Date: Thu, 10 Sep 2026 16:50:24 -0500 Subject: [PATCH 1/5] feat(export): preserve joint activity dimensions in reporting Independent project, model and agent totals cannot answer combined filters. Add opt-in reporting v3 with session-free five-minute cells, cost provenance and project-key selection. Use the shared Activity interval engine so these facts follow the same activity rules as the report. Select usage survivors and allocate authoritative costs before filtering projects. Bind complete cell sets and scope into each hour digest so late corrections replace prior usage instead of adding another charge. Retain the device bucket peak to keep aggregate concurrency bounds useful. Keep existing v1/v2 contracts and the v2 default unchanged. Document the replacement and precision limits, and measure the extra aggregation and payload cost with a synthetic SQLite benchmark. --- cmd/agentsview/export_reporting.go | 27 ++- cmd/agentsview/export_reporting_joint_test.go | 93 ++++++++ cmd/agentsview/export_reporting_test.go | 2 +- docs/internal/performance-gates.md | 8 + docs/reporting-export.md | 93 +++++++- internal/activity/activity.go | 3 + internal/activity/joint.go | 142 ++++++++++++ internal/activity/joint_test.go | 83 +++++++ internal/activity/streaming.go | 8 +- internal/db/reporting_export.go | 31 ++- internal/db/reporting_export_test.go | 11 + internal/db/reporting_joint.go | 147 +++++++++++++ internal/db/reporting_joint_bench_test.go | 65 ++++++ internal/db/reporting_joint_test.go | 207 ++++++++++++++++++ internal/export/reporting.go | 11 +- internal/export/reporting_joint.go | 107 +++++++++ internal/export/reporting_joint_test.go | 34 +++ 17 files changed, 1059 insertions(+), 13 deletions(-) create mode 100644 cmd/agentsview/export_reporting_joint_test.go create mode 100644 internal/activity/joint.go create mode 100644 internal/activity/joint_test.go create mode 100644 internal/db/reporting_joint.go create mode 100644 internal/db/reporting_joint_bench_test.go create mode 100644 internal/db/reporting_joint_test.go create mode 100644 internal/export/reporting_joint.go create mode 100644 internal/export/reporting_joint_test.go diff --git a/cmd/agentsview/export_reporting.go b/cmd/agentsview/export_reporting.go index 135f285ece..237b47ea65 100644 --- a/cmd/agentsview/export_reporting.go +++ b/cmd/agentsview/export_reporting.go @@ -27,6 +27,7 @@ func defaultExportReportingDeps() exportReportingDeps { func newExportHourCommand(deps exportReportingDeps) *cobra.Command { var schemaVersion *int + var projectKeys *[]string command := &cobra.Command{ Use: "hour YYYY-MM-DD-HH", Short: "Export one closed UTC reporting hour", @@ -36,6 +37,9 @@ func newExportHourCommand(deps exportReportingDeps) *cobra.Command { if err := validateReportingSchemaVersion(*schemaVersion); err != nil { return err } + if err := export.ValidateReportingProjectScope(*schemaVersion, *projectKeys); err != nil { + return err + } now := deps.now() hourStart, err := export.ParseReportingHourKey(args[0], now) if err != nil { @@ -52,6 +56,7 @@ func newExportHourCommand(deps exportReportingDeps) *cobra.Command { Date: hourStart.Truncate(24 * time.Hour), Now: now, SchemaVersion: *schemaVersion, + ProjectKeys: *projectKeys, }, ) if err != nil { @@ -68,11 +73,13 @@ func newExportHourCommand(deps exportReportingDeps) *cobra.Command { }, } schemaVersion = bindReportingSchemaVersion(command) + projectKeys = bindReportingProjectKeys(command) return command } func newExportDayCommand(deps exportReportingDeps) *cobra.Command { var schemaVersion *int + var projectKeys *[]string command := &cobra.Command{ Use: "day YYYY-MM-DD", Short: "Export all closed UTC reporting hours for a date", @@ -82,6 +89,9 @@ func newExportDayCommand(deps exportReportingDeps) *cobra.Command { if err := validateReportingSchemaVersion(*schemaVersion); err != nil { return err } + if err := export.ValidateReportingProjectScope(*schemaVersion, *projectKeys); err != nil { + return err + } date, err := export.ParseReportingDate(args[0]) if err != nil { return err @@ -95,6 +105,7 @@ func newExportDayCommand(deps exportReportingDeps) *cobra.Command { cmd.Context(), db.ReportingExportOptions{ Date: date, Now: deps.now(), SchemaVersion: *schemaVersion, + ProjectKeys: *projectKeys, }, ) if err != nil { @@ -104,6 +115,7 @@ func newExportDayCommand(deps exportReportingDeps) *cobra.Command { }, } schemaVersion = bindReportingSchemaVersion(command) + projectKeys = bindReportingProjectKeys(command) return command } @@ -111,6 +123,7 @@ func newExportDigestCommand(deps exportReportingDeps) *cobra.Command { var fromValue string var toValue string var schemaVersion *int + var projectKeys *[]string command := &cobra.Command{ Use: "digest --from YYYY-MM-DD --to YYYY-MM-DD", Short: "Export reporting digests for a UTC date range", @@ -120,6 +133,9 @@ func newExportDigestCommand(deps exportReportingDeps) *cobra.Command { if err := validateReportingSchemaVersion(*schemaVersion); err != nil { return err } + if err := export.ValidateReportingProjectScope(*schemaVersion, *projectKeys); err != nil { + return err + } if fromValue == "" || toValue == "" { return fmt.Errorf("--from and --to are required") } @@ -155,6 +171,7 @@ func newExportDigestCommand(deps exportReportingDeps) *cobra.Command { cmd.Context(), db.ReportingExportOptions{ Date: date, Now: now, SchemaVersion: *schemaVersion, + ProjectKeys: *projectKeys, }, ) if err != nil { @@ -181,6 +198,7 @@ func newExportDigestCommand(deps exportReportingDeps) *cobra.Command { }, } schemaVersion = bindReportingSchemaVersion(command) + projectKeys = bindReportingProjectKeys(command) command.Flags().StringVar( &fromValue, "from", "", "First UTC date (YYYY-MM-DD)", ) @@ -196,11 +214,18 @@ func bindReportingSchemaVersion(command *cobra.Command) *int { version, "schema-version", export.ReportingSchemaVersion, - fmt.Sprintf("Reporting export schema version (only %d is supported)", export.ReportingSchemaVersion), + fmt.Sprintf("Reporting export schema version (%d or %d)", export.ReportingSchemaVersion, export.ReportingJointSchemaVersion), ) return version } +func bindReportingProjectKeys(command *cobra.Command) *[]string { + keys := new([]string) + command.Flags().StringArrayVar(keys, "project-key", nil, + "Limit the complete export to an archive project key (repeatable; schema 4)") + return keys +} + func validateReportingSchemaVersion(version int) error { if !export.IsSupportedReportingSchemaVersion(version) { return fmt.Errorf("unsupported reporting schema version %d", version) diff --git a/cmd/agentsview/export_reporting_joint_test.go b/cmd/agentsview/export_reporting_joint_test.go new file mode 100644 index 0000000000..9d7de459f0 --- /dev/null +++ b/cmd/agentsview/export_reporting_joint_test.go @@ -0,0 +1,93 @@ +package main + +import ( + "encoding/json/v2" + "errors" + "testing" + "time" + + "github.com/spf13/cobra" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "go.kenn.io/agentsview/internal/db" + "go.kenn.io/agentsview/internal/export" +) + +func TestExportJointProjectScopeAgreesAcrossHourDayAndDigest(t *testing.T) { + seedExportReportingGoldenArchive(t) + now := time.Date(2026, 7, 29, 0, 0, 0, 0, time.UTC) + out, stderr, err := executeExportSessionsCommand(newExportReportingTestRoot(now), + "export", "day", "--schema-version", "4", "2026-07-28") + require.NoError(t, err) + assert.Empty(t, stderr) + var all export.ReportingDay + require.NoError(t, json.Unmarshal([]byte(out), &all)) + var key string + foundStandalone := false + for _, cell := range all.Hours[11].Joint.Cells { + if cell.Project == reportingGoldenProject { + key = cell.ProjectKey + } + if cell.Agent == "cursor" { + foundStandalone = true + assert.Empty(t, cell.ProjectKey) + assert.Equal(t, "unknown", cell.Automation) + assert.Equal(t, int64(4_000), cell.Pricing.ReportedCost.Microdollars) + assert.Zero(t, cell.AgentMinutes) + } + } + require.True(t, foundStandalone, "standalone cost must remain visible without invented attribution") + require.NotEmpty(t, key) + out, stderr, err = executeExportSessionsCommand(newExportReportingTestRoot(now), + "export", "day", "--schema-version", "4", "--project-key", key, "2026-07-28") + require.NoError(t, err) + assert.Empty(t, stderr) + var day export.ReportingDay + require.NoError(t, json.Unmarshal([]byte(out), &day)) + assert.Equal(t, int64(200), day.Hours[11].Usage.Totals.OutputTokens) + assert.Equal(t, 3.0, day.Hours[11].Activity.Totals.AgentMinutes) + assert.NotContains(t, out, "fixture-cross") + assert.NotContains(t, out, `"content"`) + for _, cell := range day.Hours[11].Joint.Cells { + assert.Equal(t, key, cell.ProjectKey) + } + + hourOut, _, err := executeExportSessionsCommand(newExportReportingTestRoot(now), + "export", "hour", "--schema-version", "4", "--project-key", key, "2026-07-28-11") + require.NoError(t, err) + var hour export.ReportingHour + require.NoError(t, json.Unmarshal([]byte(hourOut), &hour)) + assert.Equal(t, day.Hours[11], hour) + digestOut, _, err := executeExportSessionsCommand(newExportReportingTestRoot(now), + "export", "digest", "--schema-version", "4", "--project-key", key, + "--from", "2026-07-28", "--to", "2026-07-28") + require.NoError(t, err) + var digest export.ReportingDigest + require.NoError(t, json.Unmarshal([]byte(digestOut), &digest)) + require.Len(t, digest.Days, 1) + assert.Equal(t, day.Digest, digest.Days[0].DayDigest) + assert.Equal(t, hour.Digest, digest.Days[0].HourDigests[11]) +} + +func TestExportJointRejectsScopeOnOldVersionBeforeOpening(t *testing.T) { + for _, args := range [][]string{ + {"export", "hour", "2026-07-28-12"}, + {"export", "day", "2026-07-28"}, + {"export", "digest", "--from", "2026-07-28", "--to", "2026-07-28"}, + } { + t.Run(args[1], func(t *testing.T) { + opened := false + deps := exportReportingDeps{now: time.Now, + openDatabase: func(*cobra.Command) (*db.DB, func(), error) { + opened = true + return nil, nil, errors.New("archive must not be opened") + }} + out, _, err := executeExportSessionsCommand(newExportReportingTestRootWithDeps(deps), + append(args, "--schema-version", "3", "--project-key", "synthetic-key")...) + assert.ErrorContains(t, err, "project scope requires reporting schema 4") + assert.False(t, opened) + assert.Empty(t, out) + }) + } +} diff --git a/cmd/agentsview/export_reporting_test.go b/cmd/agentsview/export_reporting_test.go index 3e93047e48..2fe652c423 100644 --- a/cmd/agentsview/export_reporting_test.go +++ b/cmd/agentsview/export_reporting_test.go @@ -202,7 +202,7 @@ func TestExportReportingSchemaVersionRejectsBeforeOpen(t *testing.T) { {"export", "day", "2026-07-28"}, {"export", "digest", "--from", "2026-07-28", "--to", "2026-07-28"}, } { - for _, version := range []int{1, 2, 4} { + for _, version := range []int{1, 2, 99} { t.Run(args[1]+"/"+strconv.Itoa(version), func(t *testing.T) { opened := false deps := exportReportingDeps{ diff --git a/docs/internal/performance-gates.md b/docs/internal/performance-gates.md index 6fc3135070..31bd1343ad 100644 --- a/docs/internal/performance-gates.md +++ b/docs/internal/performance-gates.md @@ -283,6 +283,14 @@ shape (call + late output, full inline signal/secret maintenance). ## Adding a benchmark to local comparisons +`BenchmarkReportingJointDay` compares reporting v3 and v4 on the same synthetic +SQLite archive: 200 sessions, eight models, three agents, model switches and +overlap, with four or 100 projects. It measures snapshot reads, aggregation, +canonical digests and serialization, and reports allocations, bytes and cell +count. Setup is outside the timer. It does not measure CLI startup, full-history +digest screening or a production-sized archive. Run it directly with +`go test -tags fts5 ./internal/db -run '^$' -bench '^BenchmarkReportingJointDay$' -benchmem`. + The local comparison includes every benchmark in the selected packages. A benchmark present in only one revision has no baseline and is reported without a threshold check. diff --git a/docs/reporting-export.md b/docs/reporting-export.md index 913aa1bd8f..fb05dd9ae3 100644 --- a/docs/reporting-export.md +++ b/docs/reporting-export.md @@ -22,8 +22,8 @@ hours; the current UTC date contains only closed hours and has no day digest. Digest ranges are inclusive, require both bounds, and may contain at most 31 dates. -Version 3 is the default and the only supported version. Hour, day, and digest -commands accept `--schema-version 3`; versions 1 and 2 are no longer available. +Version 3 is the default. Hour, day, and digest commands also accept +`--schema-version 4` for joint cells; versions 1 and 2 are no longer available. Any other value is rejected before the archive is opened or output is written. Integrations pinned to an older version must update to version 3 and refresh their saved digests. @@ -122,6 +122,91 @@ ordered closed-hour documents, and a `digest` only when all 24 hours are present. `agentsview export hour H` is constructed by the same day reader and emits byte-for-byte the canonical hour element contained by `export day D`. +## Joint bucket cells (version 3) + +Version 4 keeps version 3's accounting rules and adds a `joint` object to every +hour. Independent project, model and agent breakdowns cannot answer a combined +filter such as "model A on project B". Joint cells retain those relationships +without exporting session identifiers, titles, messages or tool content. + +```sh +agentsview export day --schema-version 4 2026-07-28 +agentsview export hour --schema-version 4 --project-key 2026-07-28-13 +agentsview export digest --schema-version 4 --project-key \ + --from 2026-07-01 --to 2026-07-28 +``` + +Repeat `--project-key` to include more projects. Get the archive-scoped keys +from the project breakdowns or a session export's project map. No keys means the +whole archive, including unattributed usage. An explicit key selects only that +project; an unknown key produces an empty replacement, not an error. An empty +key is invalid. Version 3 rejects project selection. + +The selected scope applies to the **whole hour**, including existing totals, +breakdowns and per-device bucket maxima. The exporter chooses canonical usage +survivors and allocates authoritative costs before applying scope. A duplicate +observation excluded by that selection cannot become a new charge just because +its winning observation belongs to another project. This is export selection, +not an authorization check; callers still own permission and destination rules. + +`joint.project_keys` contains the sorted, unique requested keys (`[]` for the +whole archive). `joint.cells` is a sparse array with these fields: + +| Field | Meaning | +| ----------------------------- | ----------------------------------------------------------------------------------------------- | +| `bucket_start` | UTC start of a half-open five-minute bucket | +| `project`, `project_key` | Safe display label and canonical archive-scoped key; an empty key is unattributed | +| `agent`, `model` | Producer agent and model; `unknown` when absent | +| `automation` | `interactive`, `automated`, or `unknown` for observations without session classification | +| `agent_minutes`, `max_agents` | Sum of inferred activity durations and simultaneous peak within this cell | +| `usage` | Input, output, cache-creation and cache-read tokens, plus cost in integer microdollars | +| `pricing` | `computed_cost`, `reported_cost`, `allocated_cost` in integer microdollars, and `unpriced_rows` | + +Known cost is partitioned across the three pricing fields; their sum is the +cell's usage cost. `allocated_cost` identifies an authoritative total +apportioned by the existing accounting rules, not separately measured +per-message spend. `unpriced_rows` counts canonical usage observations without +complete pricing. Known fees still contribute to cost when token prices are +unknown; that cost is incomplete, not free usage. Activity-only cells have zero +usage. Usage without an activity interval still contributes tokens and cost, but +not invented minutes. Usage is assigned by observation timestamp, not spread +over an activity interval. + +Activity uses the same gap cap, model attribution, clipping and overlap removal +as the Activity report. Agent-minutes are not measured human working time. A +report edge inside a cell has five-minute precision; consumers must not prorate +that cell and claim an exact instant-level result. + +### Concurrency and corrections + +A model switch can create two cells for one session in the same bucket. Adding +their maxima can overstate even a single device's peak. For selected cells in a +bucket, use this upper bound across devices: + +```text +sum over devices of min(device bucket max_agents, + sum of that device's selected cell max_agents) +``` + +Take the maximum bucket bound for a window-level bound. The companion device +maximum is `activity.buckets[].max_agents` from the same permitted export scope. +This is not exact selected concurrency or exact cross-device concurrency. + +Cells are sorted by bucket, project key, agent, model and automation. Their +entire contents and requested project scope participate in the hour digest, +including for quiet hours. A changed hour replaces its entire previous cell set +in the same scope; removed cells are retractions. An empty set retracts all +previous cells. Never append a replacement as additional usage, or combine +overlapping export scopes as independent sources. + +Old parser corrections, project changes, pricing changes and deletions can +change a closed hour. Re-export and replace it when its digest changes; closing +an hour does not freeze its meaning. Version 4 does not add a history checkpoint +or a deletion journal. Digest screening still computes full day exports before +returning identities, and a single-hour command still reads the corresponding +day. Joint export adds aggregation and output proportional to the populated +cells; it is not a source-side incremental optimization. + ## Quiet hours A quiet hour means that the archive has no activity or usage observation for @@ -235,7 +320,9 @@ independent concurrency peaks for each category. It retains the complete Claude snapshot selection and web-search charging introduced in version 2. Versions 1 and 2 are no longer emitted. -Integrations should request and require `schema_version: 3`, reject unknown +Version 4 adds scoped joint cells while version 3 remains the default. + +Integrations should request and require their intended `schema_version`, reject unknown fields, and verify the canonical content digest before accepting an hour. The new fields change hour and day digests, including quiet hours; refresh previously saved digests when updating. Adding, renaming, or removing a field, diff --git a/internal/activity/activity.go b/internal/activity/activity.go index 707f0dcb32..5495dd912c 100644 --- a/internal/activity/activity.go +++ b/internal/activity/activity.go @@ -35,6 +35,7 @@ type SessionMeta struct { SessionID string Title string Project string + ProjectKey string // optional canonical key for joint export grouping Agent string Machine string StartedAt string // RFC3339 or "" @@ -77,6 +78,7 @@ type UsageRow struct { Cost money.Money CostSource export.CostSource SessionCost *money.Money + CostAllocated bool // an authoritative session total was apportioned Priced bool Contributes bool Agent string @@ -246,6 +248,7 @@ type Report struct { SessionsNextCursor string `json:"sessions_next_cursor,omitempty"` SessionsTotal int `json:"sessions_total"` Intervals []ReportInterval `json:"-"` + JointActivity []JointActivityCell `json:"-"` } func SanitizeProjectLabels( diff --git a/internal/activity/joint.go b/internal/activity/joint.go new file mode 100644 index 0000000000..977e941674 --- /dev/null +++ b/internal/activity/joint.go @@ -0,0 +1,142 @@ +package activity + +import ( + "cmp" + "context" + "slices" + "time" +) + +// JointActivityCell retains the dimensions of an activity contribution without +// retaining session identity. Project is an internal label; exporters resolve +// it through their canonical project map before serialization. +type JointActivityCell struct { + BucketStart time.Time + Project string + ProjectKey string + Agent string + Model string + IsAutomated bool + AgentMinutes float64 + MaxAgents int +} + +// AggregateCandidatesWithJointActivity also collects sparse joint cells from +// the same effective intervals as the ordinary report. Ordinary report reads +// do not pay for this extra aggregation. Usage cells remain the export owner's +// responsibility, after its complete-scope survivor and cost allocation pass. +func AggregateCandidatesWithJointActivity( + ctx context.Context, p Params, sessions []SessionMeta, + candidates []IntervalCandidate, usage []UsageRow, +) (Report, error) { + joint := &jointActivityAccumulator{ + windows: rangeWindows(p), sessions: make(map[string]SessionMeta, len(sessions)), + cells: make(map[jointActivityKey]*jointActivityState), + } + for _, session := range sessions { + joint.sessions[session.SessionID] = session + } + artifacts, err := buildCandidateArtifactsFromSource(ctx, p, sessions, + func(ctx context.Context, yield func(IntervalCandidate) error) error { + for _, candidate := range candidates { + if err := ctx.Err(); err != nil { + return err + } + if err := yield(candidate); err != nil { + return err + } + } + return nil + }, usage, false, joint) + if err != nil { + return Report{}, err + } + artifacts.Report.BySession = artifacts.Sessions + artifacts.Report.JointActivity, err = joint.finish(ctx) + return artifacts.Report, err +} + +type jointActivityKey struct { + bucket int + project, agent, model string + automated bool +} + +type jointActivityState struct { + cell JointActivityCell + deltas map[time.Time]int +} + +type jointActivityAccumulator struct { + windows []BucketWindow + sessions map[string]SessionMeta + cells map[jointActivityKey]*jointActivityState +} + +func (a *jointActivityAccumulator) add(iv interval) { + session := a.sessions[iv.sessionID] + if session.Agent == "" { + session.Agent = "unknown" + } + project := session.ProjectKey + if project == "" { + project = session.Project + } + for i := max(0, windowIndex(a.windows, iv.start)); i < len(a.windows) && a.windows[i].Start.Before(iv.end); i++ { + window := a.windows[i] + start, end := maxTime(iv.start, window.Start), minTime(iv.end, window.End) + if !end.After(start) { + continue + } + key := jointActivityKey{i, project, session.Agent, iv.model, session.IsAutomated} + state := a.cells[key] + if state == nil { + state = &jointActivityState{ + cell: JointActivityCell{BucketStart: window.Start, Project: session.Project, ProjectKey: session.ProjectKey, + Agent: session.Agent, Model: iv.model, IsAutomated: session.IsAutomated}, + deltas: make(map[time.Time]int), + } + a.cells[key] = state + } + state.cell.AgentMinutes += end.Sub(start).Minutes() + state.deltas[start]++ + state.deltas[end]-- + } +} + +func (a *jointActivityAccumulator) finish(ctx context.Context) ([]JointActivityCell, error) { + cells := make([]JointActivityCell, 0, len(a.cells)) + for _, state := range a.cells { + if err := ctx.Err(); err != nil { + return nil, err + } + points := make([]time.Time, 0, len(state.deltas)) + for at := range state.deltas { + points = append(points, at) + } + slices.SortFunc(points, time.Time.Compare) + live := 0 + for _, at := range points { + // Merge same-instant exits and entries: intervals are half-open. + live += state.deltas[at] + state.cell.MaxAgents = max(state.cell.MaxAgents, live) + } + cells = append(cells, state.cell) + } + slices.SortFunc(cells, func(a, b JointActivityCell) int { + for _, order := range []int{a.BucketStart.Compare(b.BucketStart), + cmp.Compare(a.Project, b.Project), cmp.Compare(a.Agent, b.Agent), cmp.Compare(a.Model, b.Model)} { + if order != 0 { + return order + } + } + if a.IsAutomated == b.IsAutomated { + return 0 + } + if a.IsAutomated { + return 1 + } + return -1 + }) + return cells, nil +} diff --git a/internal/activity/joint_test.go b/internal/activity/joint_test.go new file mode 100644 index 0000000000..202a571212 --- /dev/null +++ b/internal/activity/joint_test.go @@ -0,0 +1,83 @@ +package activity + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestJointActivityModelSwitchDoesNotDoubleCountSession(t *testing.T) { + start := mustStart(t, "2026-07-28T12:00:00Z") + p := Params{RangeStart: start, RangeEnd: start.Add(10 * time.Minute), + EffectiveEnd: start.Add(10 * time.Minute), Loc: time.UTC, + GapCapSeconds: 300, Bucket: BucketSpec{Unit: BucketMinute, NominalSeconds: 300}} + sessions := []SessionMeta{{SessionID: "a", Project: "project-a", Agent: "agent-a"}} + candidates := []IntervalCandidate{ + {SessionID: "a", Start: start, End: start.Add(2 * time.Minute), ClosingRole: "assistant", ClosingModel: "model-a"}, + // An overlapping observation extends this session; it is not another agent. + {SessionID: "a", Start: start.Add(time.Minute), End: start.Add(5 * time.Minute), ClosingRole: "assistant", ClosingModel: "model-b"}, + {SessionID: "a", Start: start.Add(2 * time.Minute), End: start.Add(4 * time.Minute), ClosingRole: "assistant", ClosingModel: "duplicate"}, + } + report, err := AggregateCandidatesWithJointActivity(t.Context(), p, sessions, candidates, nil) + require.NoError(t, err) + assert.Equal(t, 1, report.Buckets[0].MaxAgents) + assert.Equal(t, 5.0, report.Totals.AgentMinutes) + assert.Equal(t, []JointActivityCell{ + {BucketStart: start, Project: "project-a", Agent: "agent-a", Model: "model-a", AgentMinutes: 2, MaxAgents: 1}, + {BucketStart: start, Project: "project-a", Agent: "agent-a", Model: "model-b", AgentMinutes: 3, MaxAgents: 1}, + }, report.JointActivity) +} + +func TestJointActivityKeepsDimensionsAndClipsAtBuckets(t *testing.T) { + start := mustStart(t, "2026-07-28T12:00:00Z") + p := Params{RangeStart: start, RangeEnd: start.Add(10 * time.Minute), + EffectiveEnd: start.Add(10 * time.Minute), Loc: time.UTC, + GapCapSeconds: 300, Bucket: BucketSpec{Unit: BucketMinute, NominalSeconds: 300}} + sessions := []SessionMeta{ + {SessionID: "a", Project: "project-a", Agent: "agent-a"}, + {SessionID: "b", Project: "project-a", Agent: "agent-a"}, + {SessionID: "c", Project: "project-b", Agent: "agent-b", IsAutomated: true}, + } + candidates := []IntervalCandidate{ + {SessionID: "a", Start: start.Add(4 * time.Minute), End: start.Add(7 * time.Minute), PriorModel: "model-a"}, + {SessionID: "b", Start: start.Add(5 * time.Minute), End: start.Add(6 * time.Minute), ClosingRole: "assistant", ClosingModel: "model-a"}, + {SessionID: "c", Start: start.Add(5 * time.Minute), End: start.Add(6 * time.Minute)}, + } + report, err := AggregateCandidatesWithJointActivity(t.Context(), p, sessions, candidates, nil) + require.NoError(t, err) + assert.Equal(t, []JointActivityCell{ + {BucketStart: start, Project: "project-a", Agent: "agent-a", Model: "model-a", AgentMinutes: 1, MaxAgents: 1}, + {BucketStart: start.Add(5 * time.Minute), Project: "project-a", Agent: "agent-a", Model: "model-a", AgentMinutes: 3, MaxAgents: 2}, + {BucketStart: start.Add(5 * time.Minute), Project: "project-b", Agent: "agent-b", Model: "unknown", IsAutomated: true, AgentMinutes: 1, MaxAgents: 1}, + }, report.JointActivity) + assert.Equal(t, 3, report.Buckets[1].MaxAgents) + + ctx, cancel := context.WithCancel(t.Context()) + cancel() + _, err = AggregateCandidatesWithJointActivity(ctx, p, sessions, candidates, nil) + assert.ErrorIs(t, err, context.Canceled) +} + +func TestJointActivityCanonicalProjectAliasesSharePeak(t *testing.T) { + start := mustStart(t, "2026-07-28T12:00:00Z") + p := Params{RangeStart: start, RangeEnd: start.Add(5 * time.Minute), + EffectiveEnd: start.Add(5 * time.Minute), Loc: time.UTC, + GapCapSeconds: 300, Bucket: BucketSpec{Unit: BucketMinute, NominalSeconds: 300}} + sessions := []SessionMeta{ + {SessionID: "a", Project: "alias-a", ProjectKey: "canonical-project", Agent: "unknown"}, + {SessionID: "b", Project: "alias-b", ProjectKey: "canonical-project"}, + } + candidates := []IntervalCandidate{ + {SessionID: "a", Start: start, End: start.Add(time.Minute)}, + {SessionID: "b", Start: start.Add(time.Minute), End: start.Add(2 * time.Minute)}, + } + report, err := AggregateCandidatesWithJointActivity(t.Context(), p, sessions, candidates, nil) + require.NoError(t, err) + require.Len(t, report.JointActivity, 1) + assert.Equal(t, "canonical-project", report.JointActivity[0].ProjectKey) + assert.Equal(t, 2.0, report.JointActivity[0].AgentMinutes) + assert.Equal(t, 1, report.JointActivity[0].MaxAgents) +} diff --git a/internal/activity/streaming.go b/internal/activity/streaming.go index 776f919ee4..dbff523da7 100644 --- a/internal/activity/streaming.go +++ b/internal/activity/streaming.go @@ -237,7 +237,7 @@ func BuildCandidateArtifactsFromSource( usage []UsageRow, ) (CandidateArtifacts, error) { return buildCandidateArtifactsFromSource( - ctx, p, sessions, source, usage, false, + ctx, p, sessions, source, usage, false, nil, ) } @@ -253,7 +253,7 @@ func BuildCandidateArtifactsFromSourceWithSurvivorUsage( usage []UsageRow, ) (CandidateArtifacts, error) { return buildCandidateArtifactsFromSource( - ctx, p, sessions, source, usage, true, + ctx, p, sessions, source, usage, true, nil, ) } @@ -264,6 +264,7 @@ func buildCandidateArtifactsFromSource( source CandidateSource, usage []UsageRow, usageIsSurvivorSet bool, + joint *jointActivityAccumulator, ) (CandidateArtifacts, error) { windows := rangeWindows(p) membershipWindows := secondPrecisionWindows(windows) @@ -316,6 +317,9 @@ func buildCandidateArtifactsFromSource( &report, windows, membershipWindows, aggregates, membership, words, kindBy, iv, ) + if joint != nil { + joint.add(iv) + } return nil }) if err != nil { diff --git a/internal/db/reporting_export.go b/internal/db/reporting_export.go index b4534808ba..7ca12dd3bd 100644 --- a/internal/db/reporting_export.go +++ b/internal/db/reporting_export.go @@ -20,6 +20,7 @@ type ReportingExportOptions struct { Date time.Time Now time.Time SchemaVersion int + ProjectKeys []string // afterSnapshot is a deterministic test seam for proving that every source // read uses the transaction established before this callback. @@ -40,6 +41,9 @@ func (db *DB) ExportReportingDay( "unsupported reporting schema version %d", schemaVersion, ) } + if err := export.ValidateReportingProjectScope(schemaVersion, opts.ProjectKeys); err != nil { + return export.ReportingDay{}, err + } date, _, hourCount, complete, err := resolveReportingExportRange(opts) if err != nil { return export.ReportingDay{}, err @@ -66,7 +70,7 @@ func (db *DB) ExportReportingDay( } hours, err := db.reportingHoursFromSnapshot( - ctx, tx, date, hourCount, schemaVersion, + ctx, tx, date, hourCount, schemaVersion, opts.ProjectKeys, ) if err != nil { return export.ReportingDay{}, err @@ -88,6 +92,7 @@ func (db *DB) ExportReportingDay( func (db *DB) reportingHoursFromSnapshot( ctx context.Context, tx *sql.Tx, date time.Time, hourCount, schemaVersion int, + projectKeys []string, ) ([]export.ReportingHour, error) { hours := make([]export.ReportingHour, hourCount) if hourCount == 0 { @@ -160,14 +165,19 @@ func (db *DB) reportingHoursFromSnapshot( if err != nil { return nil, err } - activityIDs := reportingSessionIDSet(ids) - activityUsage := reportingActivityUsage(usage, activityIDs) - projectLabels := activityReportProjectLabels(allSessions) projects, err := db.reportingProjectIdentityMapFrom(ctx, tx, projectLabels) if err != nil { return nil, err } + if schemaVersion == export.ReportingJointSchemaVersion { + sessions, ids, events, usage = scopeJointReporting(sessions, events, usage, sessionByID, projects, projectKeys) + for i := range sessions { + sessions[i].ProjectKey = export.ProjectKeyForEntry(projects[sessions[i].Project]) + } + } + activityIDs := reportingSessionIDSet(ids) + activityUsage := reportingActivityUsage(usage, activityIDs) createdAt, err := reportingSessionCreatedAtFrom(ctx, tx, ids) if err != nil { return nil, err @@ -186,7 +196,11 @@ func (db *DB) reportingHoursFromSnapshot( hourEnd := hourStart.Add(time.Hour) gapCap := time.Duration(query.GapCapSeconds) * time.Second candidates := activity.PairActivityEvents(events, hourStart, hourEnd, gapCap) - report, aggregateErr := activity.AggregateCandidates(ctx, activity.Params{ + aggregate := activity.AggregateCandidates + if schemaVersion == export.ReportingJointSchemaVersion { + aggregate = activity.AggregateCandidatesWithJointActivity + } + report, aggregateErr := aggregate(ctx, activity.Params{ RangeStart: hourStart, RangeEnd: hourEnd, Loc: time.UTC, @@ -227,6 +241,12 @@ func (db *DB) reportingHoursFromSnapshot( if !hour.HasData { hour = quietReportingHour(hourStart, schemaVersion) } + if schemaVersion == export.ReportingJointSchemaVersion { + hour.Joint, err = jointReportingHour(hourStart, report.JointActivity, usage, sessionByID, projects, projectKeys) + if err != nil { + return nil, err + } + } hours[i] = hour } return hours, nil @@ -612,6 +632,7 @@ func allocateReportingUsageCosts( for i, index := range indices { out[index].Cost = costs[i] out[index].CostSource = export.CostSourceReported + out[index].CostAllocated = true out[index].Priced = true out[index].Contributes = true } diff --git a/internal/db/reporting_export_test.go b/internal/db/reporting_export_test.go index b60f98ed94..9afe0cdc55 100644 --- a/internal/db/reporting_export_test.go +++ b/internal/db/reporting_export_test.go @@ -280,6 +280,17 @@ func TestReportingExportAllocatesAuthoritativeSessionCostBeforeHourPartition( } assert.Equal(t, existing.Totals.TotalCost, hourlyCost) assert.Equal(t, int64(existing.Totals.InputTokens), hourlyInputTokens) + + joint, err := d.ExportReportingDay(t.Context(), ReportingExportOptions{ + Date: time.Date(2026, 7, 28, 0, 0, 0, 0, time.UTC), + Now: time.Date(2026, 7, 29, 12, 0, 0, 0, time.UTC), SchemaVersion: 4, + }) + require.NoError(t, err) + require.Len(t, joint.Hours[10].Joint.Cells, 1) + require.Len(t, joint.Hours[11].Joint.Cells, 1) + assert.Equal(t, int64(10_000), joint.Hours[10].Joint.Cells[0].Pricing.AllocatedCost.Microdollars) + assert.Equal(t, int64(20_000), joint.Hours[11].Joint.Cells[0].Pricing.AllocatedCost.Microdollars) + assert.Zero(t, joint.Hours[10].Joint.Cells[0].Pricing.ComputedCost.Microdollars) } func TestReportingExportAllocatesAuthoritativeCostByDailyBreakdownKey( diff --git a/internal/db/reporting_joint.go b/internal/db/reporting_joint.go new file mode 100644 index 0000000000..bf5c5bed06 --- /dev/null +++ b/internal/db/reporting_joint.go @@ -0,0 +1,147 @@ +package db + +import ( + "fmt" + "slices" + "time" + + "go.kenn.io/agentsview/internal/activity" + "go.kenn.io/agentsview/internal/export" + "go.kenn.io/agentsview/internal/money" +) + +// Apply scope after the complete day's survivor selection. Filtering duplicate +// candidates first would let a second project claim the same charged usage. +func scopeJointReporting( + sessions []activity.SessionMeta, events []activity.ActivityEvent, + usage []activity.UsageRow, sessionByID map[string]activity.SessionMeta, + projects map[string]export.ProjectMapEntry, keys []string, +) ([]activity.SessionMeta, []string, []activity.ActivityEvent, []activity.UsageRow) { + selectedKeys := make(map[string]bool, len(keys)) + for _, key := range keys { + selectedKeys[key] = true + } + selected := func(project string) bool { + return len(keys) == 0 || selectedKeys[export.ProjectKeyForEntry(projects[project])] + } + keptSessions := make([]activity.SessionMeta, 0, len(sessions)) + ids := make([]string, 0, len(sessions)) + for _, session := range sessions { + if selected(session.Project) { + keptSessions = append(keptSessions, session) + ids = append(ids, session.SessionID) + } + } + keptEvents := make([]activity.ActivityEvent, 0, len(events)) + for _, event := range events { + if selected(sessionByID[event.SessionID].Project) { + keptEvents = append(keptEvents, event) + } + } + keptUsage := make([]activity.UsageRow, 0, len(usage)) + for _, row := range usage { + if selected(sessionByID[row.SessionID].Project) { + keptUsage = append(keptUsage, row) + } + } + return keptSessions, ids, keptEvents, keptUsage +} + +type reportingCellKey struct { + bucket string + projectKey, agent, model, automation string +} + +type reportingCellState struct { + cell export.ReportingCell + usage reportingUsageAccum +} + +func jointReportingHour( + start time.Time, activityCells []activity.JointActivityCell, usage []activity.UsageRow, + sessions map[string]activity.SessionMeta, projects map[string]export.ProjectMapEntry, + projectKeys []string, +) (*export.ReportingJoint, error) { + states := make(map[reportingCellKey]*reportingCellState) + cellFor := func(bucket time.Time, project, agent, model, automation string) *reportingCellState { + if model == "" { + model = "unknown" + } + if agent == "" { + agent = "unknown" + } + key := reportingCellKey{bucket.UTC().Format(time.RFC3339), + export.ProjectKeyForEntry(projects[project]), agent, model, automation} + state := states[key] + label := export.SafeProjectDisplayLabel(project) + if state == nil { + state = &reportingCellState{cell: export.ReportingCell{ + BucketStart: key.bucket, Project: label, ProjectKey: key.projectKey, + Agent: agent, Model: model, Automation: automation, + }} + states[key] = state + } else if label < state.cell.Project { + state.cell.Project = label + } + return state + } + for _, cell := range activityCells { + automation := "interactive" + if cell.IsAutomated { + automation = "automated" + } + state := cellFor(cell.BucketStart, cell.Project, cell.Agent, cell.Model, automation) + state.cell.AgentMinutes += cell.AgentMinutes + state.cell.MaxAgents = cell.MaxAgents + } + end := start.Add(time.Hour) + for _, row := range usage { + at, err := parseTimestamp(row.Timestamp) + if err != nil || at.Before(start) || !at.Before(end) { + continue + } + session, known := sessions[row.SessionID] + agent, automation := row.Agent, "unknown" + if known { + automation = "interactive" + if session.IsAutomated { + automation = "automated" + } + if agent == "" { + agent = session.Agent + } + } + state := cellFor(at.UTC().Truncate(5*time.Minute), session.Project, agent, row.Model, automation) + if err := state.usage.add(row); err != nil { + return nil, fmt.Errorf("sum joint cell usage: %w", err) + } + if !row.Priced { + state.cell.Pricing.UnpricedRows++ + } + // Token pricing may be unknown while a service fee is still known. + // Keep that fee in the cost partition as well as the usage total. + cost := &state.cell.Pricing.ComputedCost + switch { + case row.CostAllocated: + cost = &state.cell.Pricing.AllocatedCost + case row.CostSource == export.CostSourceReported: + cost = &state.cell.Pricing.ReportedCost + } + *cost, err = money.Add(*cost, row.Cost) + if err != nil { + return nil, fmt.Errorf("sum joint cell pricing: %w", err) + } + } + keys := slices.Clone(projectKeys) + slices.Sort(keys) + joint := &export.ReportingJoint{ProjectKeys: slices.Compact(keys), Cells: make([]export.ReportingCell, 0, len(states))} + for _, state := range states { + state.cell.Usage = export.ReportingUsageTotals{ + InputTokens: state.usage.inputTokens, OutputTokens: state.usage.outputTokens, + CacheCreationTokens: state.usage.cacheCreationTokens, CacheReadTokens: state.usage.cacheReadTokens, + Cost: state.usage.cost, + } + joint.Cells = append(joint.Cells, state.cell) + } + return joint, nil +} diff --git a/internal/db/reporting_joint_bench_test.go b/internal/db/reporting_joint_bench_test.go new file mode 100644 index 0000000000..51360dc610 --- /dev/null +++ b/internal/db/reporting_joint_bench_test.go @@ -0,0 +1,65 @@ +package db + +import ( + "encoding/json/jsontext" + "fmt" + "testing" + "time" + + "github.com/stretchr/testify/require" + "go.kenn.io/agentsview/internal/export" + "go.kenn.io/agentsview/internal/money" +) + +// BenchmarkReportingJointDay measures the real SQLite snapshot, aggregation, +// canonical digest and serialization, not just encoding pre-built cells. The +// archive is fixed across iterations. It does not include process startup. +func BenchmarkReportingJointDay(b *testing.B) { + for _, projects := range []int{4, 100} { + for _, version := range []int{3, 4} { + b.Run(fmt.Sprintf("projects-%d/v%d", projects, version), func(b *testing.B) { + d := testDB(b) + const sessions = 200 + start := time.Date(2026, 7, 28, 12, 0, 0, 0, time.UTC) + for model := range 8 { + require.NoError(b, d.UpsertModelPricing([]ModelPricing{{ + ModelPattern: fmt.Sprintf("model-%d", model), OutputPerMTok: money.MustParseDollars("10"), + }})) + } + for i := range sessions { + id := fmt.Sprintf("synthetic-%d", i) + at := start.Add(time.Duration(i%55) * time.Minute) + require.NoError(b, d.UpsertSession(Session{ID: id, + Project: fmt.Sprintf("project-%d", i%projects), Agent: fmt.Sprintf("agent-%d", i%3), + Machine: "synthetic", MessageCount: 3, IsAutomated: i%2 == 0, + StartedAt: Ptr(at.Format(time.RFC3339)), EndedAt: Ptr(at.Add(4 * time.Minute).Format(time.RFC3339)), + })) + require.NoError(b, d.InsertMessages([]Message{ + {SessionID: id, Ordinal: 0, Role: "user", Timestamp: at.Format(time.RFC3339)}, + {SessionID: id, Ordinal: 1, Role: "assistant", Timestamp: at.Add(2 * time.Minute).Format(time.RFC3339), + Model: fmt.Sprintf("model-%d", i%8), TokenUsage: jsontext.Value(`{"output_tokens":100}`)}, + {SessionID: id, Ordinal: 2, Role: "assistant", Timestamp: at.Add(4 * time.Minute).Format(time.RFC3339), + Model: fmt.Sprintf("model-%d", (i+1)%8), TokenUsage: jsontext.Value(`{"output_tokens":200}`)}, + })) + } + opts := ReportingExportOptions{Date: start.Truncate(24 * time.Hour), Now: start.Add(24 * time.Hour), SchemaVersion: version} + var day export.ReportingDay + var payload []byte + var err error + b.ReportAllocs() + for b.Loop() { + day, err = d.ExportReportingDay(b.Context(), opts) + require.NoError(b, err) + payload, err = export.MarshalCanonical(day) + require.NoError(b, err) + } + require.Equal(b, int64(60_000), day.Hours[12].Usage.Totals.OutputTokens) + b.ReportMetric(float64(len(payload)), "payload-bytes/op") + if version == 4 { + require.NotEmpty(b, day.Hours[12].Joint.Cells) + b.ReportMetric(float64(len(day.Hours[12].Joint.Cells)), "cells/op") + } + }) + } + } +} diff --git a/internal/db/reporting_joint_test.go b/internal/db/reporting_joint_test.go new file mode 100644 index 0000000000..565ad9cb25 --- /dev/null +++ b/internal/db/reporting_joint_test.go @@ -0,0 +1,207 @@ +package db + +import ( + "encoding/json/jsontext" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "go.kenn.io/agentsview/internal/export" + "go.kenn.io/agentsview/internal/money" +) + +func TestReportingJointCellsPreserveTimeAndDimensions(t *testing.T) { + d := testDB(t) + seedJointReporting(t, d) + opts := ReportingExportOptions{Date: time.Date(2026, 7, 28, 0, 0, 0, 0, time.UTC), + Now: time.Date(2026, 7, 29, 0, 0, 0, 0, time.UTC), SchemaVersion: 4} + day, err := d.ExportReportingDay(t.Context(), opts) + require.NoError(t, err) + hour := day.Hours[12] + require.NotNil(t, hour.Joint) + require.Len(t, hour.Joint.Cells, 4) + assert.Equal(t, 2, hour.Activity.Buckets[0].MaxAgents) + assert.Equal(t, int64(600), hour.Usage.Totals.OutputTokens) + assert.Equal(t, int64(6_000_000), hour.Usage.Totals.Cost.Microdollars) + + var projectAKey string + for _, cell := range hour.Joint.Cells { + require.NotEmpty(t, cell.ProjectKey) + if cell.Project == "project-a" { + projectAKey = cell.ProjectKey + } + switch { + case cell.Project == "project-b": + assert.Equal(t, "agent-b", cell.Agent) + assert.Equal(t, "automated", cell.Automation) + assert.Equal(t, 3.0, cell.AgentMinutes) + assert.Equal(t, int64(300), cell.Usage.OutputTokens) + case cell.Model == "model-a": + assert.Equal(t, "2026-07-28T12:00:00Z", cell.BucketStart) + assert.Equal(t, 2.0, cell.AgentMinutes) + assert.Equal(t, int64(100), cell.Usage.OutputTokens) + case cell.BucketStart == "2026-07-28T12:00:00Z": + assert.Equal(t, 3.0, cell.AgentMinutes) + assert.Zero(t, cell.Usage.OutputTokens) + default: + assert.Equal(t, "2026-07-28T12:05:00Z", cell.BucketStart) + assert.Zero(t, cell.AgentMinutes) + assert.Equal(t, int64(200), cell.Usage.OutputTokens) + } + assert.Equal(t, cell.Usage.Cost, cell.Pricing.ComputedCost) + } + + opts.ProjectKeys = []string{projectAKey, projectAKey} + scoped, err := d.ExportReportingDay(t.Context(), opts) + require.NoError(t, err) + assert.Equal(t, []string{projectAKey}, scoped.Hours[12].Joint.ProjectKeys) + assert.Equal(t, 1, scoped.Hours[12].Activity.Buckets[0].MaxAgents) + assert.Equal(t, int64(300), scoped.Hours[12].Usage.Totals.OutputTokens) + assert.NotEqual(t, hour.Digest, scoped.Hours[12].Digest) + assert.NotEqual(t, day.Hours[0].Digest, scoped.Hours[0].Digest, "quiet hours still bind their scope") + for _, cell := range scoped.Hours[12].Joint.Cells { + assert.Equal(t, "project-a", cell.Project) + } + + oracle, err := d.GetActivityReport(t.Context(), AnalyticsFilter{Timezone: "UTC"}, dayQuery(t, "2026-07-28", "UTC")) + require.NoError(t, err) + assert.Equal(t, 8.0, oracle.Totals.AgentMinutes) + assert.Equal(t, oracle.Totals.AgentMinutes, hour.Activity.Totals.AgentMinutes) + assert.Equal(t, oracle.Peak.Agents, hour.Activity.Peak.Agents) +} + +func TestReportingJointCorrectionsReplaceCellsWithinSnapshot(t *testing.T) { + d := testDB(t) + seedJointReporting(t, d) + opts := ReportingExportOptions{Date: time.Date(2026, 7, 28, 0, 0, 0, 0, time.UTC), + Now: time.Date(2026, 7, 29, 0, 0, 0, 0, time.UTC), SchemaVersion: 4} + before, err := d.ExportReportingDay(t.Context(), opts) + require.NoError(t, err) + opts.afterSnapshot = func() { + _, writeErr := d.getWriter().Exec(`UPDATE messages SET model = 'model-c' WHERE session_id = 'joint-a' AND model = 'model-a'`) + require.NoError(t, writeErr) + } + during, err := d.ExportReportingDay(t.Context(), opts) + require.NoError(t, err) + assert.Equal(t, before.Digest, during.Digest) + opts.afterSnapshot = nil + after, err := d.ExportReportingDay(t.Context(), opts) + require.NoError(t, err) + assert.NotEqual(t, before.Hours[12].Digest, after.Hours[12].Digest) + for _, cell := range after.Hours[12].Joint.Cells { + assert.NotEqual(t, "model-a", cell.Model) + if cell.Model == "model-c" { + assert.Equal(t, int64(1), cell.Pricing.UnpricedRows) + } + } + _, err = d.getWriter().Exec(`DELETE FROM sessions WHERE id IN ('joint-a', 'joint-b')`) + require.NoError(t, err) + empty, err := d.ExportReportingDay(t.Context(), opts) + require.NoError(t, err) + assert.False(t, empty.Hours[12].HasData) + assert.Empty(t, empty.Hours[12].Joint.Cells) + assert.Zero(t, empty.Hours[12].Activity.Buckets[0].MaxAgents) + assert.NotEqual(t, after.Hours[12].Digest, empty.Hours[12].Digest) +} + +func seedJointReporting(t *testing.T, d *DB) { + t.Helper() + require.NoError(t, d.UpsertModelPricing([]ModelPricing{ + {ModelPattern: "model-a", OutputPerMTok: money.MustParseDollars("10000")}, + {ModelPattern: "model-b", OutputPerMTok: money.MustParseDollars("10000")}, + })) + insertSession(t, d, "joint-a", "project-a", func(s *Session) { + s.Agent, s.StartedAt, s.EndedAt = "agent-a", Ptr("2026-07-28T12:00:00Z"), Ptr("2026-07-28T12:05:00Z") + }) + insertSession(t, d, "joint-b", "project-b", func(s *Session) { + s.Agent, s.StartedAt, s.EndedAt = "agent-b", Ptr("2026-07-28T12:01:00Z"), Ptr("2026-07-28T12:04:00Z") + s.IsAutomated = true + }) + insertMessages(t, d, + Message{SessionID: "joint-a", Ordinal: 0, Role: "user", Timestamp: "2026-07-28T12:00:00Z", Content: "synthetic private prose"}, + Message{SessionID: "joint-a", Ordinal: 1, Role: "assistant", Timestamp: "2026-07-28T12:02:00Z", Model: "model-a", TokenUsage: jsontext.Value(`{"output_tokens":100}`)}, + Message{SessionID: "joint-a", Ordinal: 2, Role: "assistant", Timestamp: "2026-07-28T12:05:00Z", Model: "model-b", TokenUsage: jsontext.Value(`{"output_tokens":200}`)}, + Message{SessionID: "joint-b", Ordinal: 0, Role: "user", Timestamp: "2026-07-28T12:01:00Z"}, + Message{SessionID: "joint-b", Ordinal: 1, Role: "assistant", Timestamp: "2026-07-28T12:04:00Z", Model: "model-b", TokenUsage: jsontext.Value(`{"output_tokens":300}`)}, + ) +} + +func TestReportingJointProjectScopeRequiresNewVersion(t *testing.T) { + d := testDB(t) + _, err := d.ExportReportingDay(t.Context(), ReportingExportOptions{ + Date: time.Date(2026, 7, 28, 0, 0, 0, 0, time.UTC), + Now: time.Date(2026, 7, 29, 0, 0, 0, 0, time.UTC), + SchemaVersion: export.ReportingSchemaVersion, ProjectKeys: []string{"project-key"}, + }) + assert.ErrorContains(t, err, "project scope requires reporting schema 4") +} + +func TestReportingJointScopeDoesNotRechargeDuplicateUsage(t *testing.T) { + d := testDB(t) + insertSession(t, d, "shared", "project-a", func(s *Session) { + s.StartedAt, s.EndedAt = Ptr("2026-07-28T09:00:00Z"), Ptr("2026-07-28T09:01:00Z") + }) + reported := money.MustParseDollars("0.002") + require.NoError(t, d.ReplaceSessionUsageEvents("shared", []UsageEvent{{ + Source: "fixture", Model: "model-a", InputTokens: 41, Cost: &reported, + CostStatus: "exact", CostSource: "reported", OccurredAt: "2026-07-28T09:05:00Z", DedupKey: "same", + }})) + require.NoError(t, d.InsertCursorUsageEvents([]CursorUsageEvent{{ + OccurredAt: "2026-07-28T09:05:00Z", Model: "standalone", Kind: "usage", + InputTokens: 17, Charged: money.MustParseDollars("0.007"), DedupKey: "shared:fixture:same", + }})) + // Give this project a real activity cell so its exported key is discoverable. + insertMessages(t, d, + Message{SessionID: "shared", Ordinal: 0, Role: "user", Timestamp: "2026-07-28T09:00:00Z"}, + Message{SessionID: "shared", Ordinal: 1, Role: "assistant", Timestamp: "2026-07-28T09:01:00Z"}, + ) + opts := ReportingExportOptions{Date: time.Date(2026, 7, 28, 0, 0, 0, 0, time.UTC), + Now: time.Date(2026, 7, 29, 0, 0, 0, 0, time.UTC), SchemaVersion: 4} + all, err := d.ExportReportingDay(t.Context(), opts) + require.NoError(t, err) + assert.Equal(t, int64(17), all.Hours[9].Usage.Totals.InputTokens) + var key string + for _, cell := range all.Hours[9].Joint.Cells { + if cell.Project == "project-a" { + key = cell.ProjectKey + } + } + require.NotEmpty(t, key) + opts.ProjectKeys = []string{key} + scoped, err := d.ExportReportingDay(t.Context(), opts) + require.NoError(t, err) + assert.Equal(t, 1.0, scoped.Hours[9].Activity.Totals.AgentMinutes) + assert.Zero(t, scoped.Hours[9].Usage.Totals.InputTokens) + assert.Zero(t, scoped.Hours[9].Usage.Totals.Cost.Microdollars) + + opts.ProjectKeys = []string{"absent-project-key"} + empty, err := d.ExportReportingDay(t.Context(), opts) + require.NoError(t, err) + assert.False(t, empty.HasData) + assert.Empty(t, empty.Hours[9].Joint.Cells) + assert.Equal(t, opts.ProjectKeys, empty.Hours[9].Joint.ProjectKeys) +} + +func TestReportingJointUnpricedTokensRetainKnownSearchFees(t *testing.T) { + d := testDB(t) + insertSession(t, d, "partial-price", "project-a", func(s *Session) { + s.Agent = "claude" + s.StartedAt, s.EndedAt = Ptr("2026-07-28T12:00:00Z"), Ptr("2026-07-28T12:01:00Z") + }) + insertMessages(t, d, Message{SessionID: "partial-price", Ordinal: 0, Role: "assistant", + Timestamp: "2026-07-28T12:01:00Z", Model: "synthetic-unpriced-model", + TokenUsage: jsontext.Value(`{"input_tokens":100,"server_tool_use":{"web_search_requests":2}}`), + }) + day, err := d.ExportReportingDay(t.Context(), ReportingExportOptions{ + Date: time.Date(2026, 7, 28, 0, 0, 0, 0, time.UTC), + Now: time.Date(2026, 7, 29, 0, 0, 0, 0, time.UTC), SchemaVersion: 4, + }) + require.NoError(t, err) + require.Len(t, day.Hours[12].Joint.Cells, 1) + cell := day.Hours[12].Joint.Cells[0] + assert.Equal(t, int64(20_000), cell.Usage.Cost.Microdollars) + assert.Equal(t, int64(20_000), cell.Pricing.ComputedCost.Microdollars) + assert.Equal(t, int64(1), cell.Pricing.UnpricedRows) +} diff --git a/internal/export/reporting.go b/internal/export/reporting.go index 5702066c1d..3e626dd2b8 100644 --- a/internal/export/reporting.go +++ b/internal/export/reporting.go @@ -15,12 +15,14 @@ const ( // ReportingSchemaVersion is the current wire version for hour, day, and // digest exports consumed by downstream integrations. ReportingSchemaVersion = 3 + // ReportingJointSchemaVersion adds opt-in session-free joint bucket cells. + ReportingJointSchemaVersion = 4 ) // IsSupportedReportingSchemaVersion reports whether reporting exports can // still produce the requested wire semantics. func IsSupportedReportingSchemaVersion(version int) bool { - return version == ReportingSchemaVersion + return version == ReportingSchemaVersion || version == ReportingJointSchemaVersion } // ReportingHour is one immutable UTC-hour export. Digest identifies the @@ -32,6 +34,7 @@ type ReportingHour struct { HasData bool `json:"has_data"` Activity ReportingActivity `json:"activity"` Usage ReportingUsage `json:"usage"` + Joint *ReportingJoint `json:"joint,omitempty"` } // ReportingDay is one UTC date exported from a single read snapshot. A @@ -211,6 +214,10 @@ func FinalizeReportingHour(hour ReportingHour) (ReportingHour, []byte, error) { return ReportingHour{}, nil, err } hour = normalizeReportingHour(hour) + hour.Joint, err = normalizeReportingJoint(hour) + if err != nil { + return ReportingHour{}, nil, err + } if err := validateReportingBuckets(hourStart, hour.Activity.Buckets); err != nil { return ReportingHour{}, nil, err } @@ -221,6 +228,7 @@ func FinalizeReportingHour(hour ReportingHour) (ReportingHour, []byte, error) { HasData: hour.HasData, Activity: hour.Activity, Usage: hour.Usage, + Joint: hour.Joint, }) if err != nil { return ReportingHour{}, nil, fmt.Errorf("digest reporting hour: %w", err) @@ -304,6 +312,7 @@ type reportingHourDigestInput struct { HasData bool `json:"has_data"` Activity ReportingActivity `json:"activity"` Usage ReportingUsage `json:"usage"` + Joint *ReportingJoint `json:"joint,omitempty"` } func parseReportingHour(value string) (time.Time, error) { diff --git a/internal/export/reporting_joint.go b/internal/export/reporting_joint.go new file mode 100644 index 0000000000..086abe2eec --- /dev/null +++ b/internal/export/reporting_joint.go @@ -0,0 +1,107 @@ +package export + +import ( + "cmp" + "fmt" + "slices" + "strings" + "time" + + "go.kenn.io/agentsview/internal/money" +) + +// ReportingJoint is a complete sparse cell set for the hour's project scope. +// An empty ProjectKeys set selects the whole archive. An empty Cells set +// retracts every previously published cell for this hour and scope. +type ReportingJoint struct { + ProjectKeys []string `json:"project_keys"` + Cells []ReportingCell `json:"cells"` +} + +// ReportingCell preserves dimension relationships; it contains no session data. +// Usage may exist without activity. Unknown automation uses "unknown", not an +// inferred interactive classification. Empty ProjectKey means unattributed. +type ReportingCell struct { + BucketStart string `json:"bucket_start"` + Project string `json:"project"` + ProjectKey string `json:"project_key"` + Agent string `json:"agent"` + Model string `json:"model"` + Automation string `json:"automation"` + AgentMinutes float64 `json:"agent_minutes"` + MaxAgents int `json:"max_agents"` + Usage ReportingUsageTotals `json:"usage"` + Pricing ReportingCellPricing `json:"pricing"` +} + +// ReportingCellPricing partitions known cost by its provenance. UnpricedRows +// records observations whose unknown cost must not be presented as free usage. +type ReportingCellPricing struct { + ComputedCost money.Money `json:"computed_cost"` + ReportedCost money.Money `json:"reported_cost"` + AllocatedCost money.Money `json:"allocated_cost"` + UnpricedRows int64 `json:"unpriced_rows"` +} + +// ValidateReportingProjectScope checks the scope before a caller opens SQLite. +func ValidateReportingProjectScope(version int, keys []string) error { + if len(keys) > 0 && version != ReportingJointSchemaVersion { + return fmt.Errorf("project scope requires reporting schema 4") + } + for _, key := range keys { + if strings.TrimSpace(key) == "" { + return fmt.Errorf("project scope contains an empty key") + } + } + return nil +} + +func normalizeReportingJoint(hour ReportingHour) (*ReportingJoint, error) { + if hour.SchemaVersion != ReportingJointSchemaVersion { + if hour.Joint != nil { + return nil, fmt.Errorf("joint cells require reporting schema 4") + } + return nil, nil + } + if hour.Joint == nil { + return nil, fmt.Errorf("reporting schema 4 requires joint cells") + } + joint := *hour.Joint + joint.ProjectKeys = cloneOrEmpty(joint.ProjectKeys) + slices.Sort(joint.ProjectKeys) + joint.ProjectKeys = slices.Compact(joint.ProjectKeys) + if err := ValidateReportingProjectScope(hour.SchemaVersion, joint.ProjectKeys); err != nil { + return nil, err + } + joint.Cells = cloneOrEmpty(joint.Cells) + slices.SortFunc(joint.Cells, compareReportingCells) + start, err := parseReportingHour(hour.Period) + if err != nil { + return nil, err + } + for i, cell := range joint.Cells { + at, err := time.Parse(time.RFC3339, cell.BucketStart) + if err != nil || at.Before(start) || !at.Before(start.Add(time.Hour)) || + at.Sub(start)%(5*time.Minute) != 0 || cell.BucketStart != at.UTC().Format(time.RFC3339) { + return nil, fmt.Errorf("joint cell has invalid bucket %q", cell.BucketStart) + } + if i > 0 && compareReportingCells(joint.Cells[i-1], cell) == 0 { + return nil, fmt.Errorf("duplicate joint cell") + } + if len(joint.ProjectKeys) > 0 && !slices.Contains(joint.ProjectKeys, cell.ProjectKey) { + return nil, fmt.Errorf("joint cell is outside project scope") + } + } + return &joint, nil +} + +func compareReportingCells(a, b ReportingCell) int { + for _, order := range []int{cmp.Compare(a.BucketStart, b.BucketStart), + cmp.Compare(a.ProjectKey, b.ProjectKey), cmp.Compare(a.Agent, b.Agent), + cmp.Compare(a.Model, b.Model), cmp.Compare(a.Automation, b.Automation)} { + if order != 0 { + return order + } + } + return 0 +} diff --git a/internal/export/reporting_joint_test.go b/internal/export/reporting_joint_test.go new file mode 100644 index 0000000000..0596b58473 --- /dev/null +++ b/internal/export/reporting_joint_test.go @@ -0,0 +1,34 @@ +package export + +import ( + "slices" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestJointReportingCanonicalIdentity(t *testing.T) { + hour := reportingHourFixture("2026-07-29-13") + hour.SchemaVersion = 4 + hour.Joint = &ReportingJoint{ProjectKeys: []string{"b", "a", "a"}, Cells: []ReportingCell{ + {BucketStart: "2026-07-29T13:00:00Z", ProjectKey: "b", Model: "model-b", Agent: "agent-a", Automation: "interactive", AgentMinutes: 1, MaxAgents: 1}, + {BucketStart: "2026-07-29T13:05:00Z", ProjectKey: "a", Model: "model-a", Agent: "agent-b", Automation: "automated", AgentMinutes: 2, MaxAgents: 1}, + }} + first, canonical, err := FinalizeReportingHour(hour) + require.NoError(t, err) + assert.Equal(t, []string{"a", "b"}, first.Joint.ProjectKeys) + slices.Reverse(hour.Joint.Cells) + slices.Reverse(hour.Joint.ProjectKeys) + _, reordered, err := FinalizeReportingHour(hour) + require.NoError(t, err) + assert.Equal(t, canonical, reordered) + + hour.Joint.Cells[0].Pricing.UnpricedRows = 1 + corrected, _, err := FinalizeReportingHour(hour) + require.NoError(t, err) + assert.NotEqual(t, first.Digest, corrected.Digest, "cell metadata is part of replacement identity") + hour.Joint.Cells = append(hour.Joint.Cells, hour.Joint.Cells[0]) + _, _, err = FinalizeReportingHour(hour) + assert.ErrorContains(t, err, "duplicate joint cell") +} From 7dac973a7c4b24ce7820a378c9653bfcfc61ad0d Mon Sep 17 00:00:00 2001 From: Wes McKinney Date: Thu, 10 Sep 2026 19:45:45 -0500 Subject: [PATCH 2/5] feat(export): make joint bucket precision selectable Reporting integrations need to choose their time precision without changing the hourly publication cadence. Accept whole-minute durations that divide an hour, declare the resolution in v3 documents and bind it to their content identity. Preserve the v1/v2 bytes and the independent inactivity-gap rule. Keep standalone usage unattributed even when a real session has an empty project label. A missing session must not borrow that project's identity or contribute to its scoped totals. Compare one- and five-minute exports on the same synthetic archives so finer precision has a measured aggregation and payload cost. --- cmd/agentsview/export_reporting.go | 36 ++++++- cmd/agentsview/export_reporting_joint_test.go | 97 +++++++++++++++++++ docs/internal/performance-gates.md | 15 +-- docs/reporting-export.md | 45 +++++++-- internal/db/reporting_export.go | 42 +++++--- internal/db/reporting_joint.go | 18 ++-- internal/db/reporting_joint_bench_test.go | 11 ++- internal/db/reporting_joint_test.go | 40 ++++++++ internal/export/reporting.go | 26 ++++- internal/export/reporting_joint.go | 48 ++++++++- internal/export/reporting_joint_test.go | 39 ++++++++ 11 files changed, 365 insertions(+), 52 deletions(-) diff --git a/cmd/agentsview/export_reporting.go b/cmd/agentsview/export_reporting.go index 237b47ea65..c7fee5479a 100644 --- a/cmd/agentsview/export_reporting.go +++ b/cmd/agentsview/export_reporting.go @@ -28,6 +28,7 @@ func defaultExportReportingDeps() exportReportingDeps { func newExportHourCommand(deps exportReportingDeps) *cobra.Command { var schemaVersion *int var projectKeys *[]string + var bucket *string command := &cobra.Command{ Use: "hour YYYY-MM-DD-HH", Short: "Export one closed UTC reporting hour", @@ -40,6 +41,9 @@ func newExportHourCommand(deps exportReportingDeps) *cobra.Command { if err := export.ValidateReportingProjectScope(*schemaVersion, *projectKeys); err != nil { return err } + if _, err := export.ParseReportingBucket(*schemaVersion, *bucket); err != nil { + return err + } now := deps.now() hourStart, err := export.ParseReportingHourKey(args[0], now) if err != nil { @@ -57,6 +61,7 @@ func newExportHourCommand(deps exportReportingDeps) *cobra.Command { Now: now, SchemaVersion: *schemaVersion, ProjectKeys: *projectKeys, + Bucket: *bucket, }, ) if err != nil { @@ -74,12 +79,14 @@ func newExportHourCommand(deps exportReportingDeps) *cobra.Command { } schemaVersion = bindReportingSchemaVersion(command) projectKeys = bindReportingProjectKeys(command) + bucket = bindReportingBucket(command) return command } func newExportDayCommand(deps exportReportingDeps) *cobra.Command { var schemaVersion *int var projectKeys *[]string + var bucket *string command := &cobra.Command{ Use: "day YYYY-MM-DD", Short: "Export all closed UTC reporting hours for a date", @@ -92,6 +99,9 @@ func newExportDayCommand(deps exportReportingDeps) *cobra.Command { if err := export.ValidateReportingProjectScope(*schemaVersion, *projectKeys); err != nil { return err } + if _, err := export.ParseReportingBucket(*schemaVersion, *bucket); err != nil { + return err + } date, err := export.ParseReportingDate(args[0]) if err != nil { return err @@ -105,7 +115,7 @@ func newExportDayCommand(deps exportReportingDeps) *cobra.Command { cmd.Context(), db.ReportingExportOptions{ Date: date, Now: deps.now(), SchemaVersion: *schemaVersion, - ProjectKeys: *projectKeys, + ProjectKeys: *projectKeys, Bucket: *bucket, }, ) if err != nil { @@ -116,6 +126,7 @@ func newExportDayCommand(deps exportReportingDeps) *cobra.Command { } schemaVersion = bindReportingSchemaVersion(command) projectKeys = bindReportingProjectKeys(command) + bucket = bindReportingBucket(command) return command } @@ -124,6 +135,7 @@ func newExportDigestCommand(deps exportReportingDeps) *cobra.Command { var toValue string var schemaVersion *int var projectKeys *[]string + var bucket *string command := &cobra.Command{ Use: "digest --from YYYY-MM-DD --to YYYY-MM-DD", Short: "Export reporting digests for a UTC date range", @@ -136,6 +148,10 @@ func newExportDigestCommand(deps exportReportingDeps) *cobra.Command { if err := export.ValidateReportingProjectScope(*schemaVersion, *projectKeys); err != nil { return err } + duration, err := export.ParseReportingBucket(*schemaVersion, *bucket) + if err != nil { + return err + } if fromValue == "" || toValue == "" { return fmt.Errorf("--from and --to are required") } @@ -171,7 +187,7 @@ func newExportDigestCommand(deps exportReportingDeps) *cobra.Command { cmd.Context(), db.ReportingExportOptions{ Date: date, Now: now, SchemaVersion: *schemaVersion, - ProjectKeys: *projectKeys, + ProjectKeys: *projectKeys, Bucket: *bucket, }, ) if err != nil { @@ -189,16 +205,21 @@ func newExportDigestCommand(deps exportReportingDeps) *cobra.Command { HourDigests: hourDigests, }) } - return writeCanonicalReportingDocument(cmd, export.ReportingDigest{ + digest := export.ReportingDigest{ SchemaVersion: *schemaVersion, From: fromValue, To: toValue, Days: days, - }) + } + if *schemaVersion == export.ReportingJointSchemaVersion { + digest.BucketSeconds = int(duration / time.Second) + } + return writeCanonicalReportingDocument(cmd, digest) }, } schemaVersion = bindReportingSchemaVersion(command) projectKeys = bindReportingProjectKeys(command) + bucket = bindReportingBucket(command) command.Flags().StringVar( &fromValue, "from", "", "First UTC date (YYYY-MM-DD)", ) @@ -226,6 +247,13 @@ func bindReportingProjectKeys(command *cobra.Command) *[]string { return keys } +func bindReportingBucket(command *cobra.Command) *string { + bucket := new(string) + command.Flags().StringVar(bucket, "bucket", "", + "Bucket duration: whole-minute divisor of one hour (schema 4; default 5m)") + return bucket +} + func validateReportingSchemaVersion(version int) error { if !export.IsSupportedReportingSchemaVersion(version) { return fmt.Errorf("unsupported reporting schema version %d", version) diff --git a/cmd/agentsview/export_reporting_joint_test.go b/cmd/agentsview/export_reporting_joint_test.go index 9d7de459f0..99e0356fec 100644 --- a/cmd/agentsview/export_reporting_joint_test.go +++ b/cmd/agentsview/export_reporting_joint_test.go @@ -91,3 +91,100 @@ func TestExportJointRejectsScopeOnOldVersionBeforeOpening(t *testing.T) { }) } } + +func TestExportJointBucketResolutionAcrossCommands(t *testing.T) { + seedExportReportingGoldenArchive(t) + now := time.Date(2026, 7, 29, 0, 0, 0, 0, time.UTC) + quietDigests := make(map[string]string) + for _, tc := range []struct { + bucket string + seconds, count int + usageBucket string + firstBucketMinutes float64 + }{ + {"1m", 60, 60, "2026-07-28T11:10:00Z", 1}, + {"2m", 120, 30, "2026-07-28T11:10:00Z", 2}, + {"5m", 300, 12, "2026-07-28T11:10:00Z", 3}, + {"15m", 900, 4, "2026-07-28T11:00:00Z", 3}, + {"1h", 3600, 1, "2026-07-28T11:00:00Z", 3}, + } { + t.Run(tc.bucket, func(t *testing.T) { + flags := []string{"--schema-version", "4", "--bucket", tc.bucket} + out, _, err := executeExportSessionsCommand(newExportReportingTestRoot(now), + append([]string{"export", "day", "2026-07-28"}, flags...)...) + require.NoError(t, err) + var day export.ReportingDay + require.NoError(t, json.Unmarshal([]byte(out), &day)) + require.Len(t, day.Hours, 24) + for _, hour := range day.Hours { + assert.Len(t, hour.Activity.Buckets, tc.count, "quiet hours use the same resolution") + } + assert.Equal(t, 3.0, day.Hours[11].Activity.Totals.AgentMinutes, "bucket size must not change the inactivity gap cap") + assert.Equal(t, tc.firstBucketMinutes, day.Hours[11].Activity.Buckets[0].AgentMinutes) + assert.Equal(t, int64(256), day.Hours[11].Usage.Totals.OutputTokens) + assert.Equal(t, int64(71_000), day.Hours[11].Usage.Totals.Cost.Microdollars) + found := false + for _, cell := range day.Hours[11].Joint.Cells { + if cell.Model == reportingGoldenLatestModel && cell.Usage.OutputTokens > 0 { + found = true + assert.Equal(t, tc.usageBucket, cell.BucketStart) + assert.Equal(t, int64(200), cell.Usage.OutputTokens) + } + } + require.True(t, found) + quietDigests[tc.bucket] = day.Hours[0].Digest + hourOut, _, err := executeExportSessionsCommand(newExportReportingTestRoot(now), + append([]string{"export", "hour", "2026-07-28-11"}, flags...)...) + require.NoError(t, err) + var hour export.ReportingHour + require.NoError(t, json.Unmarshal([]byte(hourOut), &hour)) + assert.Equal(t, day.Hours[11], hour) + digestOut, _, err := executeExportSessionsCommand(newExportReportingTestRoot(now), + append([]string{"export", "digest", "--from", "2026-07-28", "--to", "2026-07-28"}, flags...)...) + require.NoError(t, err) + var digest export.ReportingDigest + require.NoError(t, json.Unmarshal([]byte(digestOut), &digest)) + require.Len(t, digest.Days, 1) + assert.Equal(t, day.Digest, digest.Days[0].DayDigest) + assert.Equal(t, hour.Digest, digest.Days[0].HourDigests[11]) + for _, document := range []string{out, hourOut, digestOut} { + var wire map[string]any + require.NoError(t, json.Unmarshal([]byte(document), &wire)) + assert.Equal(t, float64(tc.seconds), wire["bucket_seconds"]) + } + }) + } + assert.NotEqual(t, quietDigests["1m"], quietDigests["5m"], "even empty replacements bind resolution") +} + +func TestExportJointRejectsInvalidBucketBeforeOpening(t *testing.T) { + for _, args := range [][]string{ + {"export", "hour", "2026-07-28-12"}, + {"export", "day", "2026-07-28"}, + {"export", "digest", "--from", "2026-07-28", "--to", "2026-07-28"}, + } { + for _, tc := range []struct{ version, bucket, message string }{ + {"4", "0m", "whole-minute divisor"}, + {"4", "-1m", "whole-minute divisor"}, + {"4", "30s", "whole-minute divisor"}, + {"4", "7m", "whole-minute divisor"}, + {"4", "2h", "whole-minute divisor"}, + {"4", "invalid", "invalid reporting bucket"}, + {"3", "1m", "bucket selection requires reporting schema 4"}, + } { + t.Run(args[1]+"/v"+tc.version+"/"+tc.bucket, func(t *testing.T) { + opened := false + deps := exportReportingDeps{now: time.Now, + openDatabase: func(*cobra.Command) (*db.DB, func(), error) { + opened = true + return nil, nil, errors.New("archive must not be opened") + }} + out, _, err := executeExportSessionsCommand(newExportReportingTestRootWithDeps(deps), + append(append([]string(nil), args...), "--schema-version", tc.version, "--bucket", tc.bucket)...) + assert.ErrorContains(t, err, tc.message) + assert.False(t, opened) + assert.Empty(t, out) + }) + } + } +} diff --git a/docs/internal/performance-gates.md b/docs/internal/performance-gates.md index 31bd1343ad..e36931bc8a 100644 --- a/docs/internal/performance-gates.md +++ b/docs/internal/performance-gates.md @@ -283,12 +283,15 @@ shape (call + late output, full inline signal/secret maintenance). ## Adding a benchmark to local comparisons -`BenchmarkReportingJointDay` compares reporting v3 and v4 on the same synthetic -SQLite archive: 200 sessions, eight models, three agents, model switches and -overlap, with four or 100 projects. It measures snapshot reads, aggregation, -canonical digests and serialization, and reports allocations, bytes and cell -count. Setup is outside the timer. It does not measure CLI startup, full-history -digest screening or a production-sized archive. Run it directly with +`BenchmarkReportingJointDay` compares reporting v3 and v4 at one- and +five-minute resolutions on the same synthetic SQLite archive: 200 sessions, +eight models, three agents, model switches and overlap, with four or 100 +projects. It measures snapshot reads, aggregation, canonical digests and +serialization, and reports allocations, bytes and cell count. Compare resolution +independently of project cardinality; sparse cell counts and payload sizes do +not grow in direct proportion to bucket counts. Setup is outside the timer. It +does not measure CLI startup, full-history digest screening or a +production-sized archive. Run it directly with `go test -tags fts5 ./internal/db -run '^$' -bench '^BenchmarkReportingJointDay$' -benchmem`. The local comparison includes every benchmark in the selected packages. A diff --git a/docs/reporting-export.md b/docs/reporting-export.md index fb05dd9ae3..936d54ce98 100644 --- a/docs/reporting-export.md +++ b/docs/reporting-export.md @@ -63,7 +63,8 @@ The v3 hour shape is: } ``` -Activity always contains exactly twelve consecutive five-minute buckets. +Version 3 contains exactly twelve consecutive five-minute activity buckets. +Version 4 declares the bucket duration explicitly and permits other resolutions. Activity separates interactive sessions, subagents, and automated sessions. Subagents take precedence over automation, so an automated child belongs only to the subagent category. Totals and all model, agent, and project breakdowns carry @@ -122,7 +123,7 @@ ordered closed-hour documents, and a `digest` only when all 24 hours are present. `agentsview export hour H` is constructed by the same day reader and emits byte-for-byte the canonical hour element contained by `export day D`. -## Joint bucket cells (version 3) +## Joint bucket cells (version 4) Version 4 keeps version 3's accounting rules and adds a `joint` object to every hour. Independent project, model and agent breakdowns cannot answer a combined @@ -131,6 +132,7 @@ without exporting session identifiers, titles, messages or tool content. ```sh agentsview export day --schema-version 4 2026-07-28 +agentsview export day --schema-version 4 --bucket 1m 2026-07-28 agentsview export hour --schema-version 4 --project-key 2026-07-28-13 agentsview export digest --schema-version 4 --project-key \ --from 2026-07-01 --to 2026-07-28 @@ -142,6 +144,18 @@ whole archive, including unattributed usage. An explicit key selects only that project; an unknown key produces an empty replacement, not an error. An empty key is invalid. Version 3 rejects project selection. +`--bucket` selects the duration for both activity buckets and joint cells. It +accepts any positive whole-minute duration that divides one hour, such as `1m`, +`2m`, `5m`, `15m`, or `1h`. Five minutes is the default, not a fixed contract +limit. Version 3 rejects explicit bucket selection and retains its original +bytes. Invalid durations fail before the archive is opened. + +Every v4 hour, day and digest document carries `bucket_seconds`, including empty +documents. Every hour in a day uses that duration. Bucket starts align to the +UTC hour, and activity always includes `3600 / bucket_seconds` buckets; joint +cells remain sparse. The five-minute inactivity gap cap is independent of bucket +size. One-minute exports still use the same activity and accounting rules. + The selected scope applies to the **whole hour**, including existing totals, breakdowns and per-device bucket maxima. The exporter chooses canonical usage survivors and allocates authoritative costs before applying scope. A duplicate @@ -154,7 +168,7 @@ whole archive). `joint.cells` is a sparse array with these fields: | Field | Meaning | | ----------------------------- | ----------------------------------------------------------------------------------------------- | -| `bucket_start` | UTC start of a half-open five-minute bucket | +| `bucket_start` | UTC start of a half-open bucket lasting `bucket_seconds` | | `project`, `project_key` | Safe display label and canonical archive-scoped key; an empty key is unattributed | | `agent`, `model` | Producer agent and model; `unknown` when absent | | `automation` | `interactive`, `automated`, or `unknown` for observations without session classification | @@ -174,11 +188,20 @@ over an activity interval. Activity uses the same gap cap, model attribution, clipping and overlap removal as the Activity report. Agent-minutes are not measured human working time. A -report edge inside a cell has five-minute precision; consumers must not prorate -that cell and claim an exact instant-level result. +report edge inside a cell has the declared bucket precision; consumers must not +prorate that cell and claim an exact instant-level result. ### Concurrency and corrections +Resolution is part of the hour's content identity, not a new publication period. +Changing it replaces the complete cell set and activity buckets for the same +hour and scope, including quiet hours. Consumers must remove the old resolution +instead of adding both sets together, and screen digests with the same +`--bucket` as their exports. Finer buckets do not require more frequent uploads. +Consumers can roll up to aligned multiples of the stored duration, but cannot +recover finer detail from coarse buckets or invent missing precision. Combining +different resolutions requires a common aligned coarser grain. + A model switch can create two cells for one session in the same bucket. Adding their maxima can overstate even a single device's peak. For selected cells in a bucket, use this upper bound across devices: @@ -215,7 +238,8 @@ that period. It deliberately does not mean that an agent was observed idle for - `has_data: false`; - `idle_minutes: 0`; -- twelve zero-valued five-minute buckets; +- zero-valued activity buckets covering the complete hour (twelve at the default + five-minute resolution); - empty activity and usage breakdown arrays; and - zero first-seen counters. @@ -320,11 +344,12 @@ independent concurrency peaks for each category. It retains the complete Claude snapshot selection and web-search charging introduced in version 2. Versions 1 and 2 are no longer emitted. -Version 4 adds scoped joint cells while version 3 remains the default. +Version 4 adds scoped joint cells and selectable bucket duration while version 3 +remains the default. -Integrations should request and require their intended `schema_version`, reject unknown -fields, and verify the canonical content digest before accepting an hour. The -new fields change hour and day digests, including quiet hours; refresh +Integrations should request and require their intended `schema_version`, reject +unknown fields, and verify the canonical content digest before accepting an +hour. The new fields change hour and day digests, including quiet hours; refresh previously saved digests when updating. Adding, renaming, or removing a field, changing a type or accounting rule, or changing canonicalization requires a new schema version. diff --git a/internal/db/reporting_export.go b/internal/db/reporting_export.go index 7ca12dd3bd..a7f6a9caa3 100644 --- a/internal/db/reporting_export.go +++ b/internal/db/reporting_export.go @@ -21,6 +21,7 @@ type ReportingExportOptions struct { Now time.Time SchemaVersion int ProjectKeys []string + Bucket string // afterSnapshot is a deterministic test seam for proving that every source // read uses the transaction established before this callback. @@ -44,6 +45,10 @@ func (db *DB) ExportReportingDay( if err := export.ValidateReportingProjectScope(schemaVersion, opts.ProjectKeys); err != nil { return export.ReportingDay{}, err } + bucket, err := export.ParseReportingBucket(schemaVersion, opts.Bucket) + if err != nil { + return export.ReportingDay{}, err + } date, _, hourCount, complete, err := resolveReportingExportRange(opts) if err != nil { return export.ReportingDay{}, err @@ -70,17 +75,21 @@ func (db *DB) ExportReportingDay( } hours, err := db.reportingHoursFromSnapshot( - ctx, tx, date, hourCount, schemaVersion, opts.ProjectKeys, + ctx, tx, date, hourCount, schemaVersion, opts.ProjectKeys, bucket, ) if err != nil { return export.ReportingDay{}, err } - day, _, err := export.FinalizeReportingDay(export.ReportingDay{ + day := export.ReportingDay{ SchemaVersion: schemaVersion, Date: date.Format("2006-01-02"), Complete: complete, Hours: hours, - }) + } + if schemaVersion == export.ReportingJointSchemaVersion { + day.BucketSeconds = int(bucket / time.Second) + } + day, _, err = export.FinalizeReportingDay(day) if err != nil { return export.ReportingDay{}, fmt.Errorf("finalize reporting date: %w", err) } @@ -92,7 +101,7 @@ func (db *DB) ExportReportingDay( func (db *DB) reportingHoursFromSnapshot( ctx context.Context, tx *sql.Tx, date time.Time, hourCount, schemaVersion int, - projectKeys []string, + projectKeys []string, bucket time.Duration, ) ([]export.ReportingHour, error) { hours := make([]export.ReportingHour, hourCount) if hourCount == 0 { @@ -100,15 +109,17 @@ func (db *DB) reportingHoursFromSnapshot( } end := date.Add(time.Duration(hourCount) * time.Hour) query, err := activity.ResolveQuery(activity.QueryInput{ - Preset: "custom", - From: date.Format(time.RFC3339), - To: end.Format(time.RFC3339), - Timezone: "UTC", - BucketOverride: "5m", + Preset: "custom", + From: date.Format(time.RFC3339), + To: end.Format(time.RFC3339), + Timezone: "UTC", }, end) if err != nil { return nil, fmt.Errorf("resolve reporting snapshot range: %w", err) } + // Preserve the shared range and inactivity-gap policy. Export bucket sizes + // have their own validated complete-hour contract, separate from UI presets. + query.Bucket = activity.BucketSpec{Unit: activity.BucketMinute, NominalSeconds: int(bucket / time.Second)} filter := AnalyticsFilter{ Timezone: "UTC", IncludeSubagents: true, @@ -206,7 +217,7 @@ func (db *DB) reportingHoursFromSnapshot( Loc: time.UTC, EffectiveEnd: hourEnd, GapCapSeconds: query.GapCapSeconds, - Bucket: activity.BucketSpec{Unit: activity.BucketMinute, NominalSeconds: 300}, + Bucket: query.Bucket, }, append([]activity.SessionMeta(nil), sessions...), candidates, activityUsage) if aggregateErr != nil { return nil, fmt.Errorf( @@ -239,10 +250,11 @@ func (db *DB) reportingHoursFromSnapshot( hour.Activity.Totals.ActiveMinutes > 0 || firstSeen[i].hasAny() if !hour.HasData { - hour = quietReportingHour(hourStart, schemaVersion) + hour = quietReportingHour(hourStart, schemaVersion, bucket) } if schemaVersion == export.ReportingJointSchemaVersion { - hour.Joint, err = jointReportingHour(hourStart, report.JointActivity, usage, sessionByID, projects, projectKeys) + hour.BucketSeconds = int(bucket / time.Second) + hour.Joint, err = jointReportingHour(hourStart, bucket, report.JointActivity, usage, sessionByID, projects, projectKeys) if err != nil { return nil, err } @@ -1261,10 +1273,10 @@ func resolveReportingExportRange( return } -func quietReportingHour(start time.Time, schemaVersion int) export.ReportingHour { - buckets := make([]export.ReportingActivityBucket, 12) +func quietReportingHour(start time.Time, schemaVersion int, duration time.Duration) export.ReportingHour { + buckets := make([]export.ReportingActivityBucket, int(time.Hour/duration)) for i := range buckets { - buckets[i].Start = start.Add(time.Duration(i) * 5 * time.Minute). + buckets[i].Start = start.Add(time.Duration(i) * duration). Format(time.RFC3339) } return export.ReportingHour{ diff --git a/internal/db/reporting_joint.go b/internal/db/reporting_joint.go index bf5c5bed06..0500efb0e2 100644 --- a/internal/db/reporting_joint.go +++ b/internal/db/reporting_joint.go @@ -34,13 +34,15 @@ func scopeJointReporting( } keptEvents := make([]activity.ActivityEvent, 0, len(events)) for _, event := range events { - if selected(sessionByID[event.SessionID].Project) { + session, known := sessionByID[event.SessionID] + if len(keys) == 0 || known && selected(session.Project) { keptEvents = append(keptEvents, event) } } keptUsage := make([]activity.UsageRow, 0, len(usage)) for _, row := range usage { - if selected(sessionByID[row.SessionID].Project) { + session, known := sessionByID[row.SessionID] + if len(keys) == 0 || known && selected(session.Project) { keptUsage = append(keptUsage, row) } } @@ -58,12 +60,12 @@ type reportingCellState struct { } func jointReportingHour( - start time.Time, activityCells []activity.JointActivityCell, usage []activity.UsageRow, + start time.Time, bucket time.Duration, activityCells []activity.JointActivityCell, usage []activity.UsageRow, sessions map[string]activity.SessionMeta, projects map[string]export.ProjectMapEntry, projectKeys []string, ) (*export.ReportingJoint, error) { states := make(map[reportingCellKey]*reportingCellState) - cellFor := func(bucket time.Time, project, agent, model, automation string) *reportingCellState { + cellFor := func(bucket time.Time, project, projectKey, agent, model, automation string) *reportingCellState { if model == "" { model = "unknown" } @@ -71,7 +73,7 @@ func jointReportingHour( agent = "unknown" } key := reportingCellKey{bucket.UTC().Format(time.RFC3339), - export.ProjectKeyForEntry(projects[project]), agent, model, automation} + projectKey, agent, model, automation} state := states[key] label := export.SafeProjectDisplayLabel(project) if state == nil { @@ -90,7 +92,7 @@ func jointReportingHour( if cell.IsAutomated { automation = "automated" } - state := cellFor(cell.BucketStart, cell.Project, cell.Agent, cell.Model, automation) + state := cellFor(cell.BucketStart, cell.Project, cell.ProjectKey, cell.Agent, cell.Model, automation) state.cell.AgentMinutes += cell.AgentMinutes state.cell.MaxAgents = cell.MaxAgents } @@ -102,7 +104,9 @@ func jointReportingHour( } session, known := sessions[row.SessionID] agent, automation := row.Agent, "unknown" + projectKey := "" if known { + projectKey = export.ProjectKeyForEntry(projects[session.Project]) automation = "interactive" if session.IsAutomated { automation = "automated" @@ -111,7 +115,7 @@ func jointReportingHour( agent = session.Agent } } - state := cellFor(at.UTC().Truncate(5*time.Minute), session.Project, agent, row.Model, automation) + state := cellFor(at.UTC().Truncate(bucket), session.Project, projectKey, agent, row.Model, automation) if err := state.usage.add(row); err != nil { return nil, fmt.Errorf("sum joint cell usage: %w", err) } diff --git a/internal/db/reporting_joint_bench_test.go b/internal/db/reporting_joint_bench_test.go index 51360dc610..40df7b92e0 100644 --- a/internal/db/reporting_joint_bench_test.go +++ b/internal/db/reporting_joint_bench_test.go @@ -16,8 +16,11 @@ import ( // archive is fixed across iterations. It does not include process startup. func BenchmarkReportingJointDay(b *testing.B) { for _, projects := range []int{4, 100} { - for _, version := range []int{3, 4} { - b.Run(fmt.Sprintf("projects-%d/v%d", projects, version), func(b *testing.B) { + for _, variant := range []struct { + version int + bucket string + }{{3, ""}, {4, "5m"}, {4, "1m"}} { + b.Run(fmt.Sprintf("projects-%d/v%d/%s", projects, variant.version, variant.bucket), func(b *testing.B) { d := testDB(b) const sessions = 200 start := time.Date(2026, 7, 28, 12, 0, 0, 0, time.UTC) @@ -42,7 +45,7 @@ func BenchmarkReportingJointDay(b *testing.B) { Model: fmt.Sprintf("model-%d", (i+1)%8), TokenUsage: jsontext.Value(`{"output_tokens":200}`)}, })) } - opts := ReportingExportOptions{Date: start.Truncate(24 * time.Hour), Now: start.Add(24 * time.Hour), SchemaVersion: version} + opts := ReportingExportOptions{Date: start.Truncate(24 * time.Hour), Now: start.Add(24 * time.Hour), SchemaVersion: variant.version, Bucket: variant.bucket} var day export.ReportingDay var payload []byte var err error @@ -55,7 +58,7 @@ func BenchmarkReportingJointDay(b *testing.B) { } require.Equal(b, int64(60_000), day.Hours[12].Usage.Totals.OutputTokens) b.ReportMetric(float64(len(payload)), "payload-bytes/op") - if version == 4 { + if variant.version == 4 { require.NotEmpty(b, day.Hours[12].Joint.Cells) b.ReportMetric(float64(len(day.Hours[12].Joint.Cells)), "cells/op") } diff --git a/internal/db/reporting_joint_test.go b/internal/db/reporting_joint_test.go index 565ad9cb25..ce7147a1f1 100644 --- a/internal/db/reporting_joint_test.go +++ b/internal/db/reporting_joint_test.go @@ -205,3 +205,43 @@ func TestReportingJointUnpricedTokensRetainKnownSearchFees(t *testing.T) { assert.Equal(t, int64(20_000), cell.Pricing.ComputedCost.Microdollars) assert.Equal(t, int64(1), cell.Pricing.UnpricedRows) } + +func TestReportingJointStandaloneUsageDoesNotInheritEmptyLabelProject(t *testing.T) { + d := testDB(t) + insertSession(t, d, "empty-label", "", func(s *Session) { + s.StartedAt, s.EndedAt = Ptr("2026-07-28T09:00:00Z"), Ptr("2026-07-28T09:01:00Z") + }) + insertMessages(t, d, + Message{SessionID: "empty-label", Ordinal: 0, Role: "user", Timestamp: "2026-07-28T09:00:00Z"}, + Message{SessionID: "empty-label", Ordinal: 1, Role: "assistant", Timestamp: "2026-07-28T09:01:00Z"}, + ) + require.NoError(t, d.InsertCursorUsageEvents([]CursorUsageEvent{{ + OccurredAt: "2026-07-28T09:01:00Z", Model: "standalone", Kind: "usage", + InputTokens: 17, Charged: money.MustParseDollars("0.007"), DedupKey: "standalone", + }})) + opts := ReportingExportOptions{Date: time.Date(2026, 7, 28, 0, 0, 0, 0, time.UTC), + Now: time.Date(2026, 7, 29, 0, 0, 0, 0, time.UTC), SchemaVersion: 4} + all, err := d.ExportReportingDay(t.Context(), opts) + require.NoError(t, err) + require.Len(t, all.Hours[9].Joint.Cells, 2) + var sessionProjectKey string + for _, cell := range all.Hours[9].Joint.Cells { + if cell.Model == "standalone" { + assert.Empty(t, cell.ProjectKey, "standalone cost has no session project") + assert.Equal(t, int64(17), cell.Usage.InputTokens) + assert.Equal(t, int64(7_000), cell.Usage.Cost.Microdollars) + } else { + sessionProjectKey = cell.ProjectKey + assert.Equal(t, 1.0, cell.AgentMinutes) + } + } + require.NotEmpty(t, sessionProjectKey, "an empty label still has a real session project identity") + opts.ProjectKeys = []string{sessionProjectKey} + scoped, err := d.ExportReportingDay(t.Context(), opts) + require.NoError(t, err) + assert.Equal(t, 1.0, scoped.Hours[9].Activity.Totals.AgentMinutes) + assert.Zero(t, scoped.Hours[9].Usage.Totals.InputTokens) + assert.Zero(t, scoped.Hours[9].Usage.Totals.Cost.Microdollars) + require.Len(t, scoped.Hours[9].Joint.Cells, 1) + assert.Equal(t, sessionProjectKey, scoped.Hours[9].Joint.Cells[0].ProjectKey) +} diff --git a/internal/export/reporting.go b/internal/export/reporting.go index 3e626dd2b8..b421866dc2 100644 --- a/internal/export/reporting.go +++ b/internal/export/reporting.go @@ -29,6 +29,7 @@ func IsSupportedReportingSchemaVersion(version int) bool { // canonical document body with the derived Digest field omitted. type ReportingHour struct { SchemaVersion int `json:"schema_version"` + BucketSeconds int `json:"bucket_seconds,omitzero"` Period string `json:"period"` Digest string `json:"digest"` HasData bool `json:"has_data"` @@ -42,6 +43,7 @@ type ReportingHour struct { // observed activity or usage. type ReportingDay struct { SchemaVersion int `json:"schema_version"` + BucketSeconds int `json:"bucket_seconds,omitzero"` Date string `json:"date"` Complete bool `json:"complete"` HasData bool `json:"has_data"` @@ -53,6 +55,7 @@ type ReportingDay struct { // changed completed days without transferring hour document bodies. type ReportingDigest struct { SchemaVersion int `json:"schema_version"` + BucketSeconds int `json:"bucket_seconds,omitzero"` From string `json:"from"` To string `json:"to"` Days []ReportingDigestDay `json:"days"` @@ -218,12 +221,17 @@ func FinalizeReportingHour(hour ReportingHour) (ReportingHour, []byte, error) { if err != nil { return ReportingHour{}, nil, err } - if err := validateReportingBuckets(hourStart, hour.Activity.Buckets); err != nil { + bucket, err := reportingBucketDuration(hour.SchemaVersion, hour.BucketSeconds) + if err != nil { + return ReportingHour{}, nil, err + } + if err := validateReportingBuckets(hourStart, bucket, hour.Activity.Buckets); err != nil { return ReportingHour{}, nil, err } digest, err := DigestCanonical(reportingHourDigestInput{ SchemaVersion: hour.SchemaVersion, + BucketSeconds: hour.BucketSeconds, Period: hour.Period, HasData: hour.HasData, Activity: hour.Activity, @@ -252,6 +260,9 @@ func FinalizeReportingDay(day ReportingDay) (ReportingDay, []byte, error) { if _, err := ParseReportingDate(day.Date); err != nil { return ReportingDay{}, nil, err } + if _, err := reportingBucketDuration(day.SchemaVersion, day.BucketSeconds); err != nil { + return ReportingDay{}, nil, err + } hours := cloneOrEmpty(day.Hours) sort.SliceStable(hours, func(i, j int) bool { @@ -271,6 +282,9 @@ func FinalizeReportingDay(day ReportingDay) (ReportingDay, []byte, error) { digests := make([]string, len(hours)) hasData := false for i := range hours { + if hours[i].SchemaVersion != day.SchemaVersion || hours[i].BucketSeconds != day.BucketSeconds { + return ReportingDay{}, nil, fmt.Errorf("reporting date hours must share its schema and bucket resolution") + } wantPeriod := fmt.Sprintf("%s-%02d", day.Date, i) if hours[i].Period != wantPeriod { return ReportingDay{}, nil, fmt.Errorf( @@ -308,6 +322,7 @@ func FinalizeReportingDay(day ReportingDay) (ReportingDay, []byte, error) { type reportingHourDigestInput struct { SchemaVersion int `json:"schema_version"` + BucketSeconds int `json:"bucket_seconds,omitzero"` Period string `json:"period"` HasData bool `json:"has_data"` Activity ReportingActivity `json:"activity"` @@ -381,13 +396,14 @@ func reportingProjectLess(aKey, aProject, bKey, bProject string) bool { } func validateReportingBuckets( - hourStart time.Time, buckets []ReportingActivityBucket, + hourStart time.Time, duration time.Duration, buckets []ReportingActivityBucket, ) error { - if len(buckets) != 12 { - return fmt.Errorf("reporting hour requires 12 activity buckets, got %d", len(buckets)) + count := int(time.Hour / duration) + if len(buckets) != count { + return fmt.Errorf("reporting hour requires %d activity buckets, got %d", count, len(buckets)) } for i, bucket := range buckets { - want := hourStart.Add(time.Duration(i) * 5 * time.Minute). + want := hourStart.Add(time.Duration(i) * duration). Format(time.RFC3339) if bucket.Start != want { return fmt.Errorf( diff --git a/internal/export/reporting_joint.go b/internal/export/reporting_joint.go index 086abe2eec..9fd195733d 100644 --- a/internal/export/reporting_joint.go +++ b/internal/export/reporting_joint.go @@ -10,6 +10,48 @@ import ( "go.kenn.io/agentsview/internal/money" ) +// DefaultReportingBucket is the export default, not the wire precision limit. +const DefaultReportingBucket = 5 * time.Minute + +// ParseReportingBucket resolves a CLI or database export option. Complete hours +// need an integral number of buckets; minute precision bounds output to 60 per +// hour without a list of permitted resolutions. +func ParseReportingBucket(version int, value string) (time.Duration, error) { + if value == "" { + return DefaultReportingBucket, nil + } + if version != ReportingJointSchemaVersion { + return 0, fmt.Errorf("bucket selection requires reporting schema 4") + } + bucket, err := time.ParseDuration(value) + if err != nil { + return 0, fmt.Errorf("invalid reporting bucket: %w", err) + } + return bucket, validateReportingBucketDuration(bucket) +} + +func validateReportingBucketDuration(bucket time.Duration) error { + if bucket < time.Minute || bucket > time.Hour || bucket%time.Minute != 0 || time.Hour%bucket != 0 { + return fmt.Errorf("reporting bucket must be a positive whole-minute divisor of one hour") + } + return nil +} + +func reportingBucketDuration(version, seconds int) (time.Duration, error) { + if version != ReportingJointSchemaVersion { + if seconds != 0 { + return 0, fmt.Errorf("bucket_seconds requires reporting schema 4") + } + return DefaultReportingBucket, nil + } + // Check before converting to Duration so malformed wire values cannot wrap. + if seconds < 60 || seconds > 3600 { + return 0, fmt.Errorf("reporting bucket_seconds must be between 60 and 3600") + } + bucket := time.Duration(seconds) * time.Second + return bucket, validateReportingBucketDuration(bucket) +} + // ReportingJoint is a complete sparse cell set for the hour's project scope. // An empty ProjectKeys set selects the whole archive. An empty Cells set // retracts every previously published cell for this hour and scope. @@ -79,10 +121,14 @@ func normalizeReportingJoint(hour ReportingHour) (*ReportingJoint, error) { if err != nil { return nil, err } + bucket, err := reportingBucketDuration(hour.SchemaVersion, hour.BucketSeconds) + if err != nil { + return nil, err + } for i, cell := range joint.Cells { at, err := time.Parse(time.RFC3339, cell.BucketStart) if err != nil || at.Before(start) || !at.Before(start.Add(time.Hour)) || - at.Sub(start)%(5*time.Minute) != 0 || cell.BucketStart != at.UTC().Format(time.RFC3339) { + at.Sub(start)%bucket != 0 || cell.BucketStart != at.UTC().Format(time.RFC3339) { return nil, fmt.Errorf("joint cell has invalid bucket %q", cell.BucketStart) } if i > 0 && compareReportingCells(joint.Cells[i-1], cell) == 0 { diff --git a/internal/export/reporting_joint_test.go b/internal/export/reporting_joint_test.go index 0596b58473..f15aecd8f1 100644 --- a/internal/export/reporting_joint_test.go +++ b/internal/export/reporting_joint_test.go @@ -3,6 +3,7 @@ package export import ( "slices" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -11,6 +12,7 @@ import ( func TestJointReportingCanonicalIdentity(t *testing.T) { hour := reportingHourFixture("2026-07-29-13") hour.SchemaVersion = 4 + hour.BucketSeconds = 300 hour.Joint = &ReportingJoint{ProjectKeys: []string{"b", "a", "a"}, Cells: []ReportingCell{ {BucketStart: "2026-07-29T13:00:00Z", ProjectKey: "b", Model: "model-b", Agent: "agent-a", Automation: "interactive", AgentMinutes: 1, MaxAgents: 1}, {BucketStart: "2026-07-29T13:05:00Z", ProjectKey: "a", Model: "model-a", Agent: "agent-b", Automation: "automated", AgentMinutes: 2, MaxAgents: 1}, @@ -32,3 +34,40 @@ func TestJointReportingCanonicalIdentity(t *testing.T) { _, _, err = FinalizeReportingHour(hour) assert.ErrorContains(t, err, "duplicate joint cell") } + +func TestJointReportingRejectsInconsistentResolution(t *testing.T) { + hour := reportingHourFixture("2026-07-29-13") + hour.SchemaVersion, hour.BucketSeconds = 4, 300 + hour.Joint = &ReportingJoint{Cells: []ReportingCell{{BucketStart: "2026-07-29T13:00:00Z"}}} + for _, tc := range []struct { + name string + seconds int + message string + }{ + {"missing", 0, "bucket_seconds"}, + {"non-divisor", 420, "whole-minute divisor"}, + {"wrong bucket count", 60, "requires 60 activity buckets"}, + } { + t.Run(tc.name, func(t *testing.T) { + invalid := hour + invalid.BucketSeconds = tc.seconds + _, _, err := FinalizeReportingHour(invalid) + assert.ErrorContains(t, err, tc.message) + }) + } + hour.Joint.Cells[0].BucketStart = "2026-07-29T13:01:00Z" + _, _, err := FinalizeReportingHour(hour) + assert.ErrorContains(t, err, "invalid bucket") + // The same timestamp is valid at one-minute precision. + hour.BucketSeconds = 60 + hour.Activity.Buckets = make([]ReportingActivityBucket, 60) + start := time.Date(2026, 7, 29, 13, 0, 0, 0, time.UTC) + for i := range hour.Activity.Buckets { + hour.Activity.Buckets[i].Start = start.Add(time.Duration(i) * time.Minute).Format(time.RFC3339) + } + _, _, err = FinalizeReportingHour(hour) + require.NoError(t, err) + hour.Period = "2026-07-29-00" + _, _, err = FinalizeReportingDay(ReportingDay{SchemaVersion: 4, BucketSeconds: 300, Date: "2026-07-29", Hours: []ReportingHour{hour}}) + assert.ErrorContains(t, err, "share its schema and bucket resolution") +} From b400376a50da9abc025ba86a2ba5ac8efc5e0be4 Mon Sep 17 00:00:00 2001 From: Marius van Niekerk Date: Sat, 12 Sep 2026 14:55:04 -0400 Subject: [PATCH 3/5] fix(export): preserve subagent classification in joint cells Joint exports grouped sessions by their automation flag alone, merging subagent minutes, peaks and usage into interactive or automated cells. Use the activity classifier for both interval grouping and usage labels so delegation takes precedence and matching subagents share their own cell. --- docs/reporting-export.md | 8 +++- internal/activity/activity.go | 13 ++++++ internal/activity/joint.go | 17 +++----- internal/activity/joint_test.go | 10 ++--- internal/db/reporting_joint.go | 11 +---- internal/db/reporting_joint_test.go | 62 +++++++++++++++++++++++++++++ 6 files changed, 95 insertions(+), 26 deletions(-) diff --git a/docs/reporting-export.md b/docs/reporting-export.md index 936d54ce98..aacbaff57d 100644 --- a/docs/reporting-export.md +++ b/docs/reporting-export.md @@ -171,11 +171,17 @@ whole archive). `joint.cells` is a sparse array with these fields: | `bucket_start` | UTC start of a half-open bucket lasting `bucket_seconds` | | `project`, `project_key` | Safe display label and canonical archive-scoped key; an empty key is unattributed | | `agent`, `model` | Producer agent and model; `unknown` when absent | -| `automation` | `interactive`, `automated`, or `unknown` for observations without session classification | +| `automation` | `interactive`, `subagent`, `automated`, or `unknown` without session classification | | `agent_minutes`, `max_agents` | Sum of inferred activity durations and simultaneous peak within this cell | | `usage` | Input, output, cache-creation and cache-read tokens, plus cost in integer microdollars | | `pricing` | `computed_cost`, `reported_cost`, `allocated_cost` in integer microdollars, and `unpriced_rows` | +The `automation` field uses the same disjoint categories as activity totals. +Subagents are labeled `subagent` even when their automation flag is set. +Activity and usage share this classification, so subagents remain separate from +other sessions with the same project, agent, model, and bucket. Standalone usage +keeps `unknown`. + Known cost is partitioned across the three pricing fields; their sum is the cell's usage cost. `allocated_cost` identifies an authoritative total apportioned by the existing accounting rules, not separately measured diff --git a/internal/activity/activity.go b/internal/activity/activity.go index 5495dd912c..e1f836128d 100644 --- a/internal/activity/activity.go +++ b/internal/activity/activity.go @@ -503,6 +503,19 @@ func (s SessionMeta) kind() sessionKind { return interactiveSession } +// ActivityCategory names the session's disjoint activity category for exports. +// Delegation takes precedence over the independent automation flag. +func (s SessionMeta) ActivityCategory() string { + switch s.kind() { + case subagentSession: + return "subagent" + case automatedSession: + return "automated" + default: + return "interactive" + } +} + // Sessions absent from the map are treated as interactive. func sessionKinds(sessions []SessionMeta) map[string]sessionKind { m := make(map[string]sessionKind, len(sessions)) diff --git a/internal/activity/joint.go b/internal/activity/joint.go index 977e941674..6771f31a10 100644 --- a/internal/activity/joint.go +++ b/internal/activity/joint.go @@ -16,7 +16,7 @@ type JointActivityCell struct { ProjectKey string Agent string Model string - IsAutomated bool + Category string AgentMinutes float64 MaxAgents int } @@ -59,7 +59,7 @@ func AggregateCandidatesWithJointActivity( type jointActivityKey struct { bucket int project, agent, model string - automated bool + category string } type jointActivityState struct { @@ -82,18 +82,19 @@ func (a *jointActivityAccumulator) add(iv interval) { if project == "" { project = session.Project } + category := session.ActivityCategory() for i := max(0, windowIndex(a.windows, iv.start)); i < len(a.windows) && a.windows[i].Start.Before(iv.end); i++ { window := a.windows[i] start, end := maxTime(iv.start, window.Start), minTime(iv.end, window.End) if !end.After(start) { continue } - key := jointActivityKey{i, project, session.Agent, iv.model, session.IsAutomated} + key := jointActivityKey{i, project, session.Agent, iv.model, category} state := a.cells[key] if state == nil { state = &jointActivityState{ cell: JointActivityCell{BucketStart: window.Start, Project: session.Project, ProjectKey: session.ProjectKey, - Agent: session.Agent, Model: iv.model, IsAutomated: session.IsAutomated}, + Agent: session.Agent, Model: iv.model, Category: category}, deltas: make(map[time.Time]int), } a.cells[key] = state @@ -130,13 +131,7 @@ func (a *jointActivityAccumulator) finish(ctx context.Context) ([]JointActivityC return order } } - if a.IsAutomated == b.IsAutomated { - return 0 - } - if a.IsAutomated { - return 1 - } - return -1 + return cmp.Compare(a.Category, b.Category) }) return cells, nil } diff --git a/internal/activity/joint_test.go b/internal/activity/joint_test.go index 202a571212..df36adec34 100644 --- a/internal/activity/joint_test.go +++ b/internal/activity/joint_test.go @@ -26,8 +26,8 @@ func TestJointActivityModelSwitchDoesNotDoubleCountSession(t *testing.T) { assert.Equal(t, 1, report.Buckets[0].MaxAgents) assert.Equal(t, 5.0, report.Totals.AgentMinutes) assert.Equal(t, []JointActivityCell{ - {BucketStart: start, Project: "project-a", Agent: "agent-a", Model: "model-a", AgentMinutes: 2, MaxAgents: 1}, - {BucketStart: start, Project: "project-a", Agent: "agent-a", Model: "model-b", AgentMinutes: 3, MaxAgents: 1}, + {BucketStart: start, Project: "project-a", Agent: "agent-a", Model: "model-a", Category: "interactive", AgentMinutes: 2, MaxAgents: 1}, + {BucketStart: start, Project: "project-a", Agent: "agent-a", Model: "model-b", Category: "interactive", AgentMinutes: 3, MaxAgents: 1}, }, report.JointActivity) } @@ -49,9 +49,9 @@ func TestJointActivityKeepsDimensionsAndClipsAtBuckets(t *testing.T) { report, err := AggregateCandidatesWithJointActivity(t.Context(), p, sessions, candidates, nil) require.NoError(t, err) assert.Equal(t, []JointActivityCell{ - {BucketStart: start, Project: "project-a", Agent: "agent-a", Model: "model-a", AgentMinutes: 1, MaxAgents: 1}, - {BucketStart: start.Add(5 * time.Minute), Project: "project-a", Agent: "agent-a", Model: "model-a", AgentMinutes: 3, MaxAgents: 2}, - {BucketStart: start.Add(5 * time.Minute), Project: "project-b", Agent: "agent-b", Model: "unknown", IsAutomated: true, AgentMinutes: 1, MaxAgents: 1}, + {BucketStart: start, Project: "project-a", Agent: "agent-a", Model: "model-a", Category: "interactive", AgentMinutes: 1, MaxAgents: 1}, + {BucketStart: start.Add(5 * time.Minute), Project: "project-a", Agent: "agent-a", Model: "model-a", Category: "interactive", AgentMinutes: 3, MaxAgents: 2}, + {BucketStart: start.Add(5 * time.Minute), Project: "project-b", Agent: "agent-b", Model: "unknown", Category: "automated", AgentMinutes: 1, MaxAgents: 1}, }, report.JointActivity) assert.Equal(t, 3, report.Buckets[1].MaxAgents) diff --git a/internal/db/reporting_joint.go b/internal/db/reporting_joint.go index 0500efb0e2..edcf941b47 100644 --- a/internal/db/reporting_joint.go +++ b/internal/db/reporting_joint.go @@ -88,11 +88,7 @@ func jointReportingHour( return state } for _, cell := range activityCells { - automation := "interactive" - if cell.IsAutomated { - automation = "automated" - } - state := cellFor(cell.BucketStart, cell.Project, cell.ProjectKey, cell.Agent, cell.Model, automation) + state := cellFor(cell.BucketStart, cell.Project, cell.ProjectKey, cell.Agent, cell.Model, cell.Category) state.cell.AgentMinutes += cell.AgentMinutes state.cell.MaxAgents = cell.MaxAgents } @@ -107,10 +103,7 @@ func jointReportingHour( projectKey := "" if known { projectKey = export.ProjectKeyForEntry(projects[session.Project]) - automation = "interactive" - if session.IsAutomated { - automation = "automated" - } + automation = session.ActivityCategory() if agent == "" { agent = session.Agent } diff --git a/internal/db/reporting_joint_test.go b/internal/db/reporting_joint_test.go index ce7147a1f1..f34fb141b0 100644 --- a/internal/db/reporting_joint_test.go +++ b/internal/db/reporting_joint_test.go @@ -72,6 +72,68 @@ func TestReportingJointCellsPreserveTimeAndDimensions(t *testing.T) { assert.Equal(t, oracle.Peak.Agents, hour.Activity.Peak.Agents) } +func TestReportingJointSubagentsTakePrecedenceOverAutomation(t *testing.T) { + d := testDB(t) + require.NoError(t, d.UpsertModelPricing([]ModelPricing{{ + ModelPattern: "model-a", OutputPerMTok: money.MustParseDollars("1"), + }})) + for _, session := range []struct { + id string + subagent, automated bool + }{ + {"root", false, false}, + {"child", true, false}, + {"automated-child", true, true}, + {"automated", false, true}, + } { + insertSession(t, d, session.id, "project-a", func(s *Session) { + s.Agent = "agent-a" + s.StartedAt, s.EndedAt = new("2026-07-28T12:00:00Z"), new("2026-07-28T12:01:00Z") + s.IsAutomated = session.automated + if session.subagent { + s.ParentSessionID = new("root") + s.RelationshipType = "subagent" + } + }) + insertMessages(t, d, + Message{SessionID: session.id, Ordinal: 0, Role: "user", Timestamp: "2026-07-28T12:00:00Z"}, + Message{SessionID: session.id, Ordinal: 1, Role: "assistant", Timestamp: "2026-07-28T12:01:00Z", + Model: "model-a", TokenUsage: jsontext.Value(`{"output_tokens":10}`)}, + ) + } + day, err := d.ExportReportingDay(t.Context(), ReportingExportOptions{ + Date: time.Date(2026, 7, 28, 0, 0, 0, 0, time.UTC), + Now: time.Date(2026, 7, 29, 0, 0, 0, 0, time.UTC), SchemaVersion: 4, + }) + require.NoError(t, err) + hour := day.Hours[12] + require.NotNil(t, hour.Joint) + require.Len(t, hour.Joint.Cells, 3) + for i, want := range []struct { + category string + minutes float64 + peak int + tokens int64 + }{ + {"automated", 1, 1, 10}, + {"interactive", 1, 1, 10}, + {"subagent", 2, 2, 20}, + } { + cell := hour.Joint.Cells[i] + assert.Equal(t, want.category, cell.Automation) + assert.Equal(t, want.minutes, cell.AgentMinutes) + assert.Equal(t, want.peak, cell.MaxAgents) + assert.Equal(t, want.tokens, cell.Usage.OutputTokens) + assert.Equal(t, money.Money{Microdollars: want.tokens}, cell.Usage.Cost) + assert.Equal(t, cell.Usage.Cost, cell.Pricing.ComputedCost) + } + assert.Equal(t, 1.0, hour.Activity.Totals.InteractiveAgentMinutes) + assert.Equal(t, 2.0, hour.Activity.Totals.SubagentAgentMinutes) + assert.Equal(t, 1.0, hour.Activity.Totals.AutomatedAgentMinutes) + assert.Equal(t, 2, hour.Activity.Buckets[0].MaxSubagentAgents) + assert.Equal(t, money.Money{Microdollars: 20}, hour.Activity.Totals.SubagentCost) +} + func TestReportingJointCorrectionsReplaceCellsWithinSnapshot(t *testing.T) { d := testDB(t) seedJointReporting(t, d) From 78cb83fdecce220d4c5da510c1cb6bd6e1fc8c13 Mon Sep 17 00:00:00 2001 From: Marius van Niekerk Date: Sat, 12 Sep 2026 15:13:58 -0400 Subject: [PATCH 4/5] fix(export): classify usage-only subagent sessions Usage recorded after a session's activity range comes through a separate session query. That query omitted delegation metadata, so joint exports labeled subagents as interactive or automated despite the shared classifier. Load the relationship flag with usage sessions so their labels retain subagent precedence even when no activity falls in the reporting period. --- internal/db/reporting_export.go | 4 ++- internal/db/reporting_joint_test.go | 41 +++++++++++++++++++++++++++++ 2 files changed, 44 insertions(+), 1 deletion(-) diff --git a/internal/db/reporting_export.go b/internal/db/reporting_export.go index a7f6a9caa3..ca19dc4d9e 100644 --- a/internal/db/reporting_export.go +++ b/internal/db/reporting_export.go @@ -291,7 +291,8 @@ func (db *DB) reportingUsageSessionsFrom( s.machine, COALESCE(s.started_at, ''), COALESCE(s.ended_at, ''), - COALESCE(s.is_automated, 0) + COALESCE(s.is_automated, 0), + s.relationship_type = 'subagent' FROM sessions s JOIN usage_session_ids u ON u.session_id = s.id ORDER BY s.id`, @@ -317,6 +318,7 @@ func (db *DB) reportingUsageSessionsFrom( &session.StartedAt, &session.EndedAt, &session.IsAutomated, + &session.IsSubagent, ); err != nil { return nil, nil, fmt.Errorf( "scanning reporting usage session: %w", err, diff --git a/internal/db/reporting_joint_test.go b/internal/db/reporting_joint_test.go index f34fb141b0..39352aa90c 100644 --- a/internal/db/reporting_joint_test.go +++ b/internal/db/reporting_joint_test.go @@ -134,6 +134,47 @@ func TestReportingJointSubagentsTakePrecedenceOverAutomation(t *testing.T) { assert.Equal(t, money.Money{Microdollars: 20}, hour.Activity.Totals.SubagentCost) } +func TestReportingJointUsageOnlySubagentRetainsClassification(t *testing.T) { + for _, tc := range []struct { + name string + automated bool + }{ + {"subagent", false}, + {"automated subagent", true}, + } { + t.Run(tc.name, func(t *testing.T) { + d := testDB(t) + insertSession(t, d, "child", "project-a", func(s *Session) { + s.Agent = "agent-a" + s.StartedAt, s.EndedAt = new("2026-07-27T08:00:00Z"), new("2026-07-27T08:01:00Z") + s.RelationshipType = "subagent" + s.IsAutomated = tc.automated + }) + cost := money.MustParseDollars("0.003") + require.NoError(t, d.ReplaceSessionUsageEvents("child", []UsageEvent{{ + Source: "fixture", Model: "model-a", OutputTokens: 17, + Cost: &cost, CostStatus: "exact", CostSource: "reported", + OccurredAt: "2026-07-28T09:10:00Z", DedupKey: "child-usage", + }})) + day, err := d.ExportReportingDay(t.Context(), ReportingExportOptions{ + Date: time.Date(2026, 7, 28, 0, 0, 0, 0, time.UTC), + Now: time.Date(2026, 7, 29, 0, 0, 0, 0, time.UTC), SchemaVersion: 4, + }) + require.NoError(t, err) + hour := day.Hours[9] + require.NotNil(t, hour.Joint) + require.Len(t, hour.Joint.Cells, 1) + cell := hour.Joint.Cells[0] + assert.Equal(t, "subagent", cell.Automation) + assert.Zero(t, cell.AgentMinutes) + assert.Zero(t, cell.MaxAgents) + assert.Equal(t, int64(17), cell.Usage.OutputTokens) + assert.Equal(t, money.Money{Microdollars: 3_000}, cell.Usage.Cost) + assert.Equal(t, money.Money{Microdollars: 3_000}, cell.Pricing.ReportedCost) + }) + } +} + func TestReportingJointCorrectionsReplaceCellsWithinSnapshot(t *testing.T) { d := testDB(t) seedJointReporting(t, d) From 15e0767d8f10edd98523f7cdf8f0ad23bed06759 Mon Sep 17 00:00:00 2001 From: Marius van Niekerk Date: Sat, 12 Sep 2026 15:14:04 -0400 Subject: [PATCH 5/5] test(server): use virtual time for timeout responses Windows CI observed a successful response in the timeout case. The test used competing wall-clock timers, leaving deadline assertions dependent on host scheduling. Advance the deadline and handler delay with Go's virtual clock while retaining the response status, headers and body checks. Join the handler before leaving the clock bubble after a timeout response. --- internal/server/timeout_custom_test.go | 50 +++++++++++++++----------- 1 file changed, 30 insertions(+), 20 deletions(-) diff --git a/internal/server/timeout_custom_test.go b/internal/server/timeout_custom_test.go index 220afeaf1d..c551fb3e45 100644 --- a/internal/server/timeout_custom_test.go +++ b/internal/server/timeout_custom_test.go @@ -6,6 +6,7 @@ import ( "net/http" "net/http/httptest" "testing" + "testing/synctest" "time" "github.com/stretchr/testify/assert" @@ -63,32 +64,41 @@ func TestWithTimeout(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { t.Parallel() - s := newTestServerMinimal(t, tt.timeout) - wrapped := s.withTimeout(tt.operation, tt.handler) + // Virtual time keeps the deadline and handler delay independent of host scheduling. + synctest.Test(t, func(t *testing.T) { + s := newTestServerMinimal(t, tt.timeout) + handlerDone := make(chan struct{}) + wrapped := s.withTimeout(tt.operation, func(w http.ResponseWriter, r *http.Request) { + defer close(handlerDone) + tt.handler(w, r) + }) + // A timeout returns before the handler finishes; join it while time can still advance. + defer func() { <-handlerDone }() - req := httptest.NewRequest(http.MethodGet, "/", nil) - w := httptest.NewRecorder() - wrapped.ServeHTTP(w, req) + req := httptest.NewRequest(http.MethodGet, "/", nil) + w := httptest.NewRecorder() + wrapped.ServeHTTP(w, req) - resp := w.Result() - defer resp.Body.Close() + resp := w.Result() + defer resp.Body.Close() - if tt.assertResponse != nil { - tt.assertResponse(t, resp) - return - } + if tt.assertResponse != nil { + tt.assertResponse(t, resp) + return + } - assertRecorderStatus(t, w, tt.wantStatus) + assertRecorderStatus(t, w, tt.wantStatus) - if tt.wantHeaderKey != "" { - assert.Equal(t, tt.wantHeaderVal, - resp.Header.Get(tt.wantHeaderKey), - "header %s", tt.wantHeaderKey) - } + if tt.wantHeaderKey != "" { + assert.Equal(t, tt.wantHeaderVal, + resp.Header.Get(tt.wantHeaderKey), + "header %s", tt.wantHeaderKey) + } - body, err := io.ReadAll(resp.Body) - require.NoError(t, err) - assert.Equal(t, tt.wantBody, string(body)) + body, err := io.ReadAll(resp.Body) + require.NoError(t, err) + assert.Equal(t, tt.wantBody, string(body)) + }) }) } }