Files
multica/server/internal/handler/dashboard_test.go
Bohan Jiang 4d0475ce89 feat(usage): error/failure charts on the Usage page (MUL-5352) (#5991)
* feat(usage): add error/failure visibility to the Usage dashboard

The Usage page could only answer "how much did we spend"; nothing on it
showed how often agents fail, what kind of failure it was, or which agent
is responsible. Operators had to open failed tasks one at a time to spot a
pattern.

`agent_task_queue.failure_reason` already carries the refined 21-value
taxonomy from server/pkg/taskfailure, so this is a read path over data that
already exists.

Backend — two rollups, both scoped by workspace/project/window like the
existing dashboard endpoints:

  GET /api/dashboard/failures/daily     per-(date, failure_reason)
  GET /api/dashboard/failures/by-agent  per-(agent, failure_reason)

They return every terminal task, not just failures: the `failure_reason: ""`
row carries the succeeded count. That is what makes the error rate's
denominator share filters with its numerator. The run-time rollups can't
serve as that denominator — they require `started_at IS NOT NULL`, so a task
that expired in the queue (the signature of a runtime outage) contributes
nothing to their failed_count. A failed row with an empty reason column
lands in an `unclassified` bucket rather than being mistaken for a success.

Frontend:
- "Errors" joins the trend toggle, daily and weekly, stacked by failure
  class with the bucket's error rate in the tooltip.
- An Errors card breaks the window down by class and by agent, with the raw
  failure_reason strings behind a disclosure (unlocalised — an operator
  pastes them into a log search). Each agent row links to its Work tab,
  which lists the actual failed runs.
- The 21 backend reasons fold into 7 display classes in
  @multica/core/dashboard. Unknown reasons — including ones from a backend
  newer than the client — land in "other" instead of being dropped, so the
  class totals always reconcile with the failure count.

The Tasks KPI tile is deliberately left alone: its value counts started
tasks only, so quoting the failure rollup's larger count there would put two
denominators in one tile. The Errors card states its rate with the
denominator spelled out instead.

Migration 225 adds a partial index on agent_task_queue(completed_at) for
terminal statuses. The table had no completed_at index at all, so the two
pre-existing run-time rollups were already scanning it; these two new
queries would have doubled that.

Closes #4429 (MUL-5352)

Co-authored-by: multica-agent <github@multica.ai>

* fix(usage): correct the Errors drill-down, window and agent exposure

Review findings on PR #5991.

1. The drill-down pointed at the wrong page. `?view=work` renders
   ActorIssuesPanel — the issues assigned to the agent — while its runs live
   in the Overview pane's ActivityTab. Link to Overview.

   That page also could not show why a run failed: `failureReasonLabel` was a
   `Record<TaskFailureReason, string>` indexed with a cast to the old 6-value
   coarse enum, so every refined reason the backend has written since
   MUL-1949 resolved to `undefined`. It is now a function over the full
   21-value taxonomy plus the legacy coarse values, falling back to the raw
   wire string for anything newer than the client. Fixes the issue execution
   log too, which had the same cast.

2. The Errors card covered one more calendar day than the chart above it.
   `parseSinceParamInTZ` returns N+1 days of headroom on purpose and the
   dashboard trims the surplus client-side — but only a series carrying a
   date can be trimmed that way. Totals / classes / reasons now derive from
   the date-bucketed rollup after that trim, and the per-agent rollup (which
   has no date to trim on) closes its window server-side via a new
   `parseExactSinceParamInTZ`. At days=1 the card previously reported
   yesterday's failures beside a chart showing none.

3. The top-offenders list leaked agents the viewer cannot see. The failure
   rollups are workspace-scoped and deliberately skip per-agent visibility,
   but the agent list they are joined against does not — members only see a
   private agent when they own it or are owner/admin. `name ?? row.agentId`
   therefore rendered a bare UUID along with that agent's failure count,
   rate and dominant error class. Unresolvable agents now fold into one
   anonymous row, and the renderer never falls back to an id. Stricter than
   `bucketUnknownAgentRows` while the agent list loads: a transient flash of
   UUIDs is the leak, not a cosmetic glitch.

Also from the review: the Errors tooltip echoed the raw Recharts dataKey
("rate_limit") instead of the translated label the legend already carries.

Not changed — the schema's `failure_reason` default stays `""`. Defaulting a
missing field to a failure bucket guards against a deflated rate, but the
realistic drift is `omitempty` on the Go struct tag, which would strip the
field from exactly the SUCCESS rows and read as a 100% error rate. Added
TestDashboardFailureWireContractKeepsEmptyReason to pin that the server
always emits the field, which is the assumption the default rests on.

Co-authored-by: multica-agent <github@multica.ai>

* fix(usage): renumber migration and fix the anonymous bucket's failure class

Review findings on PR #5991, round 2.

1. Migration prefix 225 collided with `225_chat_message_channel_media_pending`,
   which landed on main while this branch was open — backend CI failed on
   TestMigrationNumericPrefixesStayUniqueAfterLegacySet. Merged main and
   renumbered to 231; main now carries 225 through 230, so 226 is taken too.

2. The anonymous "Other agents" bucket could announce the wrong failure class.
   It merged rows that had ALREADY collapsed to one dominant class per agent,
   then credited each agent's entire failure count to that class. An agent
   failing auth 6 / timeout 5 contributed 11 to auth and 0 to timeout, so a
   bucket whose real composition was timeout 15 / auth 6 rendered as Auth.

   Fixed by anonymizing the raw per-(agent, reason) rows instead: the sentinel
   becomes just another agent_id and `aggregateAgentFailures` computes its
   classes from real counts. That also deletes the parallel bucketing pass —
   one identity rewrite replaces it. `knownAgentIds` moves up to where both
   consumers can see it.

Also from the review:
- The wire-contract test decoded both payloads into one map. json.Unmarshal
  merges into a non-nil map rather than resetting it, so a residual
  failure_reason from the first case could have masked an omitempty
  regression in the second — exactly what the test is meant to catch. Now
  table-driven with a fresh map per case.
- A test comment still described the drill-down as pointing at the Work tab.

Co-authored-by: multica-agent <github@multica.ai>

---------

Co-authored-by: Bohan-J <bohan@devv.ai>
Co-authored-by: multica-agent <github@multica.ai>
2026-07-27 18:40:48 +08:00

1654 lines
64 KiB
Go
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
package handler
import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"testing"
"time"
)
// TestDashboardEndpoints covers the workspace-dashboard rollups:
// - daily token usage with and without project filter
// - per-agent token usage with and without project filter
// - per-agent run time
//
// Asserts that (1) tasks belonging to a project show up under the workspace
// view, (2) the project filter excludes tasks tied to issues without a
// matching project_id, and (3) run-time aggregation accumulates the
// completed_at started_at delta correctly.
func TestDashboardEndpoints(t *testing.T) {
if testHandler == nil {
t.Skip("database not available")
}
ctx := context.Background()
var runtimeID, agentID string
if err := testPool.QueryRow(ctx, `
SELECT id FROM agent_runtime WHERE workspace_id = $1 LIMIT 1
`, testWorkspaceID).Scan(&runtimeID); err != nil {
t.Fatalf("fetch runtime: %v", err)
}
if err := testPool.QueryRow(ctx, `
SELECT id FROM agent WHERE workspace_id = $1 LIMIT 1
`, testWorkspaceID).Scan(&agentID); err != nil {
t.Fatalf("fetch agent: %v", err)
}
// Two issues: one bound to a project, one not.
var projectID string
if err := testPool.QueryRow(ctx, `
INSERT INTO project (workspace_id, title)
VALUES ($1, 'dashboard test project')
RETURNING id
`, testWorkspaceID).Scan(&projectID); err != nil {
t.Fatalf("create project: %v", err)
}
t.Cleanup(func() { testPool.Exec(ctx, `DELETE FROM project WHERE id = $1`, projectID) })
// issue.number is `UNIQUE (workspace_id, number)` (migration 020) and
// defaults to 0. Two inserts into the same workspace would collide on the
// default; allocate `MAX(number) + 1` per row to stay sequential and
// avoid stepping on rows other tests have left behind in the shared
// fixture workspace.
mkIssue := func(withProject bool) string {
var id string
var pid any
if withProject {
pid = projectID
}
if err := testPool.QueryRow(ctx, `
INSERT INTO issue (workspace_id, title, creator_id, creator_type, project_id, number)
VALUES (
$1, 'dashboard test', $2, 'member', $3,
(SELECT COALESCE(MAX(number), 0) + 1 FROM issue WHERE workspace_id = $1)
)
RETURNING id
`, testWorkspaceID, testUserID, pid).Scan(&id); err != nil {
t.Fatalf("insert issue: %v", err)
}
t.Cleanup(func() { testPool.Exec(ctx, `DELETE FROM issue WHERE id = $1`, id) })
return id
}
projectIssueID := mkIssue(true)
otherIssueID := mkIssue(false)
now := time.Now().UTC()
started := now.Add(-30 * time.Minute)
completed := started.Add(10 * time.Minute) // 600s run
mkTaskWithUsage := func(issueID string, status string, tokens int64) {
var taskID string
if err := testPool.QueryRow(ctx, `
INSERT INTO agent_task_queue (agent_id, issue_id, runtime_id, status, started_at, completed_at, created_at)
VALUES ($1, $2, $3, $4, $5, $6, now())
RETURNING id
`, agentID, issueID, runtimeID, status, started, completed).Scan(&taskID); err != nil {
t.Fatalf("insert task: %v", err)
}
if _, err := testPool.Exec(ctx, `
INSERT INTO task_usage (task_id, provider, model, input_tokens, output_tokens, created_at)
VALUES ($1, 'claude', 'claude-3-5-sonnet', $2, 0, now())
`, taskID, tokens); err != nil {
t.Fatalf("insert task_usage: %v", err)
}
t.Cleanup(func() { testPool.Exec(ctx, `DELETE FROM agent_task_queue WHERE id = $1`, taskID) })
}
mkTaskWithUsage(projectIssueID, "completed", 1000)
mkTaskWithUsage(otherIssueID, "completed", 500)
// All dashboard endpoints now read from task_usage_hourly (post-RFC
// Phase 3). Drive the underlying window function directly so the
// freshly inserted fixture rows are aggregated before assertions —
// in production the cron tick handles this with a 5-min lag.
if _, err := testPool.Exec(ctx, `
SELECT rollup_task_usage_hourly_window('1970-01-01'::timestamptz, now() + interval '1 hour')
`); err != nil {
t.Fatalf("rollup window: %v", err)
}
type dailyRow struct {
Date string `json:"date"`
Model string `json:"model"`
InputTokens int64 `json:"input_tokens"`
}
type byAgentRow struct {
AgentID string `json:"agent_id"`
Model string `json:"model"`
InputTokens int64 `json:"input_tokens"`
}
type runtimeRow struct {
AgentID string `json:"agent_id"`
TotalSeconds int64 `json:"total_seconds"`
TaskCount int32 `json:"task_count"`
}
// daily — workspace-wide
{
w := httptest.NewRecorder()
testHandler.GetDashboardUsageDaily(w, newRequest("GET", "/api/dashboard/usage/daily?days=1", nil))
if w.Code != http.StatusOK {
t.Fatalf("daily ws: expected 200, got %d: %s", w.Code, w.Body.String())
}
var rows []dailyRow
_ = json.NewDecoder(w.Body).Decode(&rows)
var total int64
for _, r := range rows {
if r.Model == "claude-3-5-sonnet" {
total += r.InputTokens
}
}
if total < 1500 {
t.Errorf("daily ws: expected >=1500 tokens (1000+500), got %d", total)
}
}
// daily — project-scoped
{
w := httptest.NewRecorder()
testHandler.GetDashboardUsageDaily(w, newRequest("GET", "/api/dashboard/usage/daily?days=1&project_id="+projectID, nil))
if w.Code != http.StatusOK {
t.Fatalf("daily project: expected 200, got %d: %s", w.Code, w.Body.String())
}
var rows []dailyRow
_ = json.NewDecoder(w.Body).Decode(&rows)
var total int64
for _, r := range rows {
if r.Model == "claude-3-5-sonnet" {
total += r.InputTokens
}
}
// Project filter must exclude the 500-token "other" issue. Token total
// for this project must be >= 1000 (our task) and < 1500 (would only
// reach 1500 if filter leaked).
if total < 1000 {
t.Errorf("daily project: expected >=1000 tokens, got %d", total)
}
if total >= 1500 {
t.Errorf("daily project: filter leaked — expected <1500 tokens, got %d", total)
}
}
// by-agent — project-scoped
{
w := httptest.NewRecorder()
testHandler.GetDashboardUsageByAgent(w, newRequest("GET", "/api/dashboard/usage/by-agent?days=1&project_id="+projectID, nil))
if w.Code != http.StatusOK {
t.Fatalf("by-agent project: expected 200, got %d: %s", w.Code, w.Body.String())
}
var rows []byAgentRow
_ = json.NewDecoder(w.Body).Decode(&rows)
found := false
for _, r := range rows {
if r.AgentID == agentID && r.InputTokens >= 1000 {
found = true
}
}
if !found {
t.Errorf("by-agent project: expected agent %s with >=1000 tokens; got %v", agentID, rows)
}
}
// agent-runtime — project-scoped
{
w := httptest.NewRecorder()
testHandler.GetDashboardAgentRunTime(w, newRequest("GET", "/api/dashboard/agent-runtime?days=1&project_id="+projectID, nil))
if w.Code != http.StatusOK {
t.Fatalf("agent-runtime: expected 200, got %d: %s", w.Code, w.Body.String())
}
var rows []runtimeRow
_ = json.NewDecoder(w.Body).Decode(&rows)
var seconds int64
var tasks int32
for _, r := range rows {
if r.AgentID == agentID {
seconds += r.TotalSeconds
tasks += r.TaskCount
}
}
if tasks < 1 {
t.Errorf("agent-runtime: expected >=1 task for agent, got %d", tasks)
}
if seconds < 600 {
t.Errorf("agent-runtime: expected >=600s (one 10-minute run), got %d", seconds)
}
}
// agent-runtime — invalid project_id rejected
{
w := httptest.NewRecorder()
testHandler.GetDashboardAgentRunTime(w, newRequest("GET", "/api/dashboard/agent-runtime?project_id=not-a-uuid", nil))
if w.Code != http.StatusBadRequest {
t.Errorf("agent-runtime: expected 400 for invalid uuid, got %d", w.Code)
}
}
// Workspace-wide by-agent through the same rollup, just to confirm
// the no-project-filter shape matches up.
{
w := httptest.NewRecorder()
testHandler.GetDashboardUsageByAgent(w, newRequest("GET", "/api/dashboard/usage/by-agent?days=1", nil))
if w.Code != http.StatusOK {
t.Fatalf("by-agent ws: expected 200, got %d: %s", w.Code, w.Body.String())
}
var aRows []byAgentRow
_ = json.NewDecoder(w.Body).Decode(&aRows)
var aTotal int64
for _, r := range aRows {
if r.AgentID == agentID && r.Model == "claude-3-5-sonnet" {
aTotal += r.InputTokens
}
}
if aTotal < 1500 {
t.Errorf("by-agent ws: expected >=1500 tokens across workspace, got %d", aTotal)
}
}
}
// TestDashboardUsageDailyBucketsByViewerTimezone proves the `?tz=` query
// param drives the calendar-day boundary: the same UTC instant lands under
// a different `date` for a UTC viewer vs an America/Los_Angeles viewer.
// This is the core promise of the timezone-architecture RFC.
func TestDashboardUsageDailyBucketsByViewerTimezone(t *testing.T) {
if testHandler == nil {
t.Skip("database not available")
}
ctx := context.Background()
var runtimeID, agentID string
if err := testPool.QueryRow(ctx, `SELECT id FROM agent_runtime WHERE workspace_id = $1 LIMIT 1`, testWorkspaceID).Scan(&runtimeID); err != nil {
t.Fatalf("fetch runtime: %v", err)
}
if err := testPool.QueryRow(ctx, `SELECT id FROM agent WHERE workspace_id = $1 LIMIT 1`, testWorkspaceID).Scan(&agentID); err != nil {
t.Fatalf("fetch agent: %v", err)
}
t.Cleanup(func() {
testPool.Exec(ctx, `DELETE FROM task_usage_hourly WHERE runtime_id = $1 AND provider = 'tz-bucket-test'`, runtimeID)
})
// One bucket at 04:00 UTC two days ago. 04:00 UTC is still the
// previous evening in America/Los_Angeles (UTC-7/-8), so the UTC
// viewer and the LA viewer must see this row under different dates.
var bucketHour time.Time
if err := testPool.QueryRow(ctx, `
INSERT INTO task_usage_hourly (
bucket_hour, workspace_id, runtime_id, agent_id, project_id,
provider, model,
input_tokens, output_tokens, cache_read_tokens, cache_write_tokens, event_count
)
VALUES (
((CURRENT_DATE - 2)::timestamp + interval '4 hours') AT TIME ZONE 'UTC',
$1, $2, $3, NULL, 'tz-bucket-test', 'tz-bucket-model',
999, 0, 0, 0, 1
)
ON CONFLICT ON CONSTRAINT uq_task_usage_hourly_key DO UPDATE
SET input_tokens = EXCLUDED.input_tokens
RETURNING bucket_hour
`, testWorkspaceID, runtimeID, agentID).Scan(&bucketHour); err != nil {
t.Fatalf("seed hourly row: %v", err)
}
utcDate := bucketHour.UTC().Format("2006-01-02")
laLoc, err := time.LoadLocation("America/Los_Angeles")
if err != nil {
t.Fatalf("load LA location: %v", err)
}
laDate := bucketHour.In(laLoc).Format("2006-01-02")
if utcDate == laDate {
t.Fatalf("test setup: UTC and LA dates must differ, both %s", utcDate)
}
readDate := func(tz string) string {
w := httptest.NewRecorder()
testHandler.GetDashboardUsageDaily(w, newRequest("GET", "/api/dashboard/usage/daily?days=10&tz="+tz, nil))
if w.Code != http.StatusOK {
t.Fatalf("tz=%s: expected 200, got %d: %s", tz, w.Code, w.Body.String())
}
var rows []struct {
Date string `json:"date"`
Model string `json:"model"`
}
_ = json.NewDecoder(w.Body).Decode(&rows)
for _, r := range rows {
if r.Model == "tz-bucket-model" {
return r.Date
}
}
t.Fatalf("tz=%s: tz-bucket-model row not found in %v", tz, rows)
return ""
}
if got := readDate("UTC"); got != utcDate {
t.Errorf("UTC viewer: expected date %s, got %s", utcDate, got)
}
if got := readDate("America/Los_Angeles"); got != laDate {
t.Errorf("LA viewer: expected date %s, got %s", laDate, got)
}
}
// TestDashboardRunTimeDailyBucketsByViewerTimezone proves the `?tz=` query
// param drives the calendar-day boundary of the Time / Tasks dashboard tab:
// GetDashboardRunTimeDaily applies `@tz` to `completed_at AT TIME ZONE @tz`
// on agent_task_queue. A task completed at 04:00 UTC is still the previous
// evening in America/Los_Angeles (UTC-7/-8), so the LA viewer must see the
// row under the prior calendar date relative to a UTC viewer.
func TestDashboardRunTimeDailyBucketsByViewerTimezone(t *testing.T) {
if testHandler == nil {
t.Skip("database not available")
}
ctx := context.Background()
var runtimeID, agentID string
if err := testPool.QueryRow(ctx, `SELECT id FROM agent_runtime WHERE workspace_id = $1 LIMIT 1`, testWorkspaceID).Scan(&runtimeID); err != nil {
t.Fatalf("fetch runtime: %v", err)
}
if err := testPool.QueryRow(ctx, `SELECT id FROM agent WHERE workspace_id = $1 LIMIT 1`, testWorkspaceID).Scan(&agentID); err != nil {
t.Fatalf("fetch agent: %v", err)
}
// Issue tagged so we can clean up just this test's rows.
var issueID string
if err := testPool.QueryRow(ctx, `
INSERT INTO issue (workspace_id, title, creator_id, creator_type, number)
VALUES ($1, 'runtime-daily tz test', $2, 'member',
(SELECT COALESCE(MAX(number), 0) + 1 FROM issue WHERE workspace_id = $1))
RETURNING id
`, testWorkspaceID, testUserID).Scan(&issueID); err != nil {
t.Fatalf("create issue: %v", err)
}
t.Cleanup(func() { testPool.Exec(ctx, `DELETE FROM issue WHERE id = $1`, issueID) })
// completed_at at 04:00 UTC two days ago — still the prior evening in LA.
// started_at 10 minutes earlier so the run has a non-zero duration.
var completedAt time.Time
var taskID string
if err := testPool.QueryRow(ctx, `
INSERT INTO agent_task_queue (agent_id, issue_id, runtime_id, status, started_at, completed_at, created_at)
VALUES (
$1, $2, $3, 'completed',
((CURRENT_DATE - 2)::timestamp + interval '3 hours 50 minutes') AT TIME ZONE 'UTC',
((CURRENT_DATE - 2)::timestamp + interval '4 hours') AT TIME ZONE 'UTC',
now()
)
RETURNING id, completed_at
`, agentID, issueID, runtimeID).Scan(&taskID, &completedAt); err != nil {
t.Fatalf("insert completed task: %v", err)
}
t.Cleanup(func() { testPool.Exec(ctx, `DELETE FROM agent_task_queue WHERE id = $1`, taskID) })
utcDate := completedAt.UTC().Format("2006-01-02")
laLoc, err := time.LoadLocation("America/Los_Angeles")
if err != nil {
t.Fatalf("load LA location: %v", err)
}
laDate := completedAt.In(laLoc).Format("2006-01-02")
if utcDate == laDate {
t.Fatalf("test setup: UTC and LA dates must differ, both %s", utcDate)
}
readRow := func(tz string) (string, int64, int32) {
w := httptest.NewRecorder()
testHandler.GetDashboardRunTimeDaily(w, newRequest("GET", "/api/dashboard/runtime/daily?days=10&tz="+tz, nil))
if w.Code != http.StatusOK {
t.Fatalf("tz=%s: expected 200, got %d: %s", tz, w.Code, w.Body.String())
}
var rows []struct {
Date string `json:"date"`
TotalSeconds int64 `json:"total_seconds"`
TaskCount int32 `json:"task_count"`
}
_ = json.NewDecoder(w.Body).Decode(&rows)
want := utcDate
if tz == "America/Los_Angeles" {
want = laDate
}
for _, r := range rows {
if r.Date == want {
return r.Date, r.TotalSeconds, r.TaskCount
}
}
t.Fatalf("tz=%s: no row on expected date %s in %v", tz, want, rows)
return "", 0, 0
}
if date, secs, count := readRow("UTC"); date != utcDate || count < 1 || secs < 600 {
t.Errorf("UTC viewer: got date=%s seconds=%d count=%d, want date=%s seconds>=600 count>=1",
date, secs, count, utcDate)
}
if date, secs, count := readRow("America/Los_Angeles"); date != laDate || count < 1 || secs < 600 {
t.Errorf("LA viewer: got date=%s seconds=%d count=%d, want date=%s seconds>=600 count>=1",
date, secs, count, laDate)
}
}
// TestRollupTaskUsageHourlyIdempotentAndWatermark covers two pipeline
// invariants the deleted runtime_rollup_test.go used to guard for the
// legacy daily rollup: (1) re-running the window function over the same
// range produces identical totals, and (2) the cron entry point advances
// the rollup-state watermark and clears last_error.
func TestRollupTaskUsageHourlyIdempotentAndWatermark(t *testing.T) {
if testHandler == nil {
t.Skip("database not available")
}
// Serialise against any other package's test that touches the shared
// rollup singleton / advisory lock 4246 (MUL-3980).
lockRollupSingleton(t)
ctx := context.Background()
var runtimeID, agentID string
if err := testPool.QueryRow(ctx, `SELECT id FROM agent_runtime WHERE workspace_id = $1 LIMIT 1`, testWorkspaceID).Scan(&runtimeID); err != nil {
t.Fatalf("fetch runtime: %v", err)
}
if err := testPool.QueryRow(ctx, `SELECT id FROM agent WHERE workspace_id = $1 LIMIT 1`, testWorkspaceID).Scan(&agentID); err != nil {
t.Fatalf("fetch agent: %v", err)
}
var issueID, taskID string
if err := testPool.QueryRow(ctx, `
INSERT INTO issue (workspace_id, title, creator_id, creator_type, number)
VALUES ($1, 'rollup idempotency', $2, 'member',
(SELECT COALESCE(MAX(number), 0) + 1 FROM issue WHERE workspace_id = $1))
RETURNING id
`, testWorkspaceID, testUserID).Scan(&issueID); err != nil {
t.Fatalf("create issue: %v", err)
}
t.Cleanup(func() { testPool.Exec(ctx, `DELETE FROM issue WHERE id = $1`, issueID) })
if err := testPool.QueryRow(ctx, `
INSERT INTO agent_task_queue (agent_id, issue_id, runtime_id, status, created_at)
VALUES ($1, $2, $3, 'completed', now() - interval '20 minutes') RETURNING id
`, agentID, issueID, runtimeID).Scan(&taskID); err != nil {
t.Fatalf("insert task: %v", err)
}
t.Cleanup(func() { testPool.Exec(ctx, `DELETE FROM agent_task_queue WHERE id = $1`, taskID) })
if _, err := testPool.Exec(ctx, `
INSERT INTO task_usage (task_id, provider, model, input_tokens, output_tokens, created_at)
VALUES ($1, 'claude', 'rollup-idem-model', 3333, 0, now() - interval '20 minutes')
`, taskID); err != nil {
t.Fatalf("insert task_usage: %v", err)
}
readTotal := func() int64 {
var total int64
if err := testPool.QueryRow(ctx, `
SELECT COALESCE(SUM(input_tokens), 0) FROM task_usage_hourly
WHERE runtime_id = $1 AND model = 'rollup-idem-model'
`, runtimeID).Scan(&total); err != nil {
t.Fatalf("read total: %v", err)
}
return total
}
// Idempotency: two passes over the same range must not double-count.
for i := 0; i < 2; i++ {
if _, err := testPool.Exec(ctx, `
SELECT rollup_task_usage_hourly_window('1970-01-01'::timestamptz, now() + interval '1 hour')
`); err != nil {
t.Fatalf("rollup pass %d: %v", i, err)
}
}
if got := readTotal(); got != 3333 {
t.Errorf("idempotency: expected exactly 3333 tokens after two passes, got %d", got)
}
// Watermark advance: park the watermark an hour back, run the cron
// entry, confirm it moved forward to ~now()-5min with no error.
if _, err := testPool.Exec(ctx, `
UPDATE task_usage_hourly_rollup_state
SET watermark_at = now() - interval '1 hour', last_error = 'stale'
WHERE id = 1
`); err != nil {
t.Fatalf("park watermark: %v", err)
}
if _, err := testPool.Exec(ctx, `SELECT rollup_task_usage_hourly()`); err != nil {
t.Fatalf("rollup_task_usage_hourly: %v", err)
}
var watermarkAge time.Duration
var lastError *string
var ageSeconds float64
if err := testPool.QueryRow(ctx, `
SELECT EXTRACT(EPOCH FROM (now() - watermark_at)), last_error
FROM task_usage_hourly_rollup_state WHERE id = 1
`).Scan(&ageSeconds, &lastError); err != nil {
t.Fatalf("read rollup state: %v", err)
}
watermarkAge = time.Duration(ageSeconds) * time.Second
// Watermark should sit at now()-5min (the cron upper bound), well
// short of the 1-hour-back value we parked it at.
if watermarkAge > 10*time.Minute {
t.Errorf("watermark did not advance: still %s behind now()", watermarkAge)
}
if lastError != nil {
t.Errorf("expected last_error cleared, got %q", *lastError)
}
}
// TestRollupTaskUsageHourlyReassignBetweenRuntimes ports the invalidation
// coverage the deleted runtime_rollup_test.go held for the legacy daily
// rollup. Reassigning a task between runtimes (the runtime-merge path) must
// move its usage: the `trg_atq_dirty_hourly` trigger enqueues both the old
// and new runtime buckets, and the next window run drains the queue,
// empties the old bucket, and fills the new one. Without this the rollup
// keeps attributing usage to the merged-away runtime.
func TestRollupTaskUsageHourlyReassignBetweenRuntimes(t *testing.T) {
if testHandler == nil {
t.Skip("database not available")
}
ctx := context.Background()
oldRuntimeID := handlerTestRuntimeID(t)
var newRuntimeID string
if err := testPool.QueryRow(ctx, `
INSERT INTO agent_runtime (
workspace_id, daemon_id, name, runtime_mode, provider, status, device_info, metadata, last_seen_at
)
VALUES ($1, NULL, 'reassign-target-hourly', 'cloud', 'reassign-target-hourly', 'online', '{}'::jsonb, '{}'::jsonb, now())
RETURNING id
`, testWorkspaceID).Scan(&newRuntimeID); err != nil {
t.Fatalf("create dest runtime: %v", err)
}
t.Cleanup(func() { testPool.Exec(ctx, `DELETE FROM agent_runtime WHERE id = $1`, newRuntimeID) })
var agentID string
if err := testPool.QueryRow(ctx, `SELECT id FROM agent WHERE workspace_id = $1 LIMIT 1`, testWorkspaceID).Scan(&agentID); err != nil {
t.Fatalf("fetch agent: %v", err)
}
var issueID string
if err := testPool.QueryRow(ctx, `
INSERT INTO issue (workspace_id, title, creator_id, creator_type, number)
VALUES ($1, 'reassign hourly test', $2, 'member',
(SELECT COALESCE(MAX(number), 0) + 1 FROM issue WHERE workspace_id = $1))
RETURNING id
`, testWorkspaceID, testUserID).Scan(&issueID); err != nil {
t.Fatalf("create issue: %v", err)
}
t.Cleanup(func() { testPool.Exec(ctx, `DELETE FROM issue WHERE id = $1`, issueID) })
usageAt := time.Date(2021, 3, 14, 1, 0, 0, 0, time.UTC)
var taskID string
if err := testPool.QueryRow(ctx, `
INSERT INTO agent_task_queue (agent_id, issue_id, runtime_id, status, created_at)
VALUES ($1, $2, $3, 'completed', $4) RETURNING id
`, agentID, issueID, oldRuntimeID, usageAt).Scan(&taskID); err != nil {
t.Fatalf("insert task: %v", err)
}
t.Cleanup(func() { testPool.Exec(ctx, `DELETE FROM agent_task_queue WHERE id = $1`, taskID) })
if _, err := testPool.Exec(ctx, `
INSERT INTO task_usage (task_id, provider, model, input_tokens, output_tokens, created_at, updated_at)
VALUES ($1, 'claude', 'm-reassign-hourly', 700, 70, $2, $2)
`, taskID, usageAt); err != nil {
t.Fatalf("insert task_usage: %v", err)
}
t.Cleanup(func() {
testPool.Exec(ctx, `DELETE FROM task_usage_hourly WHERE model = 'm-reassign-hourly'`)
testPool.Exec(ctx, `DELETE FROM task_usage_hourly_dirty WHERE model = 'm-reassign-hourly'`)
})
runWindow := func(label string) {
if _, err := testPool.Exec(ctx, `
SELECT rollup_task_usage_hourly_window('-infinity'::timestamptz, 'infinity'::timestamptz)
`); err != nil {
t.Fatalf("%s: %v", label, err)
}
}
runtimeTotal := func(rt string) int64 {
var total int64
testPool.QueryRow(ctx, `
SELECT COALESCE(SUM(input_tokens), 0) FROM task_usage_hourly
WHERE runtime_id = $1 AND model = 'm-reassign-hourly'
`, rt).Scan(&total)
return total
}
runWindow("initial rollup")
if old, new := runtimeTotal(oldRuntimeID), runtimeTotal(newRuntimeID); old != 700 || new != 0 {
t.Fatalf("initial: expected old=700 new=0, got old=%d new=%d", old, new)
}
// Reassignment fires trg_atq_dirty_hourly, which enqueues the OLD and
// NEW runtime buckets (same bucket_hour, two runtime_ids).
if _, err := testPool.Exec(ctx, `UPDATE agent_task_queue SET runtime_id = $1 WHERE id = $2`, newRuntimeID, taskID); err != nil {
t.Fatalf("reassign task: %v", err)
}
var dirtyCount int
testPool.QueryRow(ctx, `SELECT COUNT(*) FROM task_usage_hourly_dirty WHERE model = 'm-reassign-hourly'`).Scan(&dirtyCount)
if dirtyCount != 2 {
t.Fatalf("expected 2 dirty entries (old+new runtime), got %d", dirtyCount)
}
runWindow("rollup after reassign")
if old, new := runtimeTotal(oldRuntimeID), runtimeTotal(newRuntimeID); old != 0 || new != 700 {
t.Fatalf("after reassign: expected old=0 new=700, got old=%d new=%d", old, new)
}
// The window function must drain every queue row whose enqueued_at
// predates p_to — a regression on that DELETE pins recomputes forever.
testPool.QueryRow(ctx, `SELECT COUNT(*) FROM task_usage_hourly_dirty WHERE model = 'm-reassign-hourly'`).Scan(&dirtyCount)
if dirtyCount != 0 {
t.Errorf("expected dirty queue drained, got %d entries", dirtyCount)
}
}
// TestRollupTaskUsageHourlyWorkspaceMismatch constructs an atq row whose
// agent.workspace_id differs from issue.workspace_id and verifies the
// hourly rollup resolves workspace_id consistently from `agent` across the
// trigger, dirty_from_updates, and the recompute join. If any path leaked
// back to issue.workspace_id the dirty key would miss the recompute join
// and the bucket would be dropped or mis-attributed across tenants. The
// schema does not enforce the two workspace_ids match, so this canary
// keeps the alignment honest.
func TestRollupTaskUsageHourlyWorkspaceMismatch(t *testing.T) {
if testHandler == nil {
t.Skip("database not available")
}
ctx := context.Background()
var foreignWorkspaceID string
if err := testPool.QueryRow(ctx, `
INSERT INTO workspace (name, slug)
VALUES ('ws-mismatch-hourly', 'ws-mismatch-hourly-' || gen_random_uuid()::text)
RETURNING id
`).Scan(&foreignWorkspaceID); err != nil {
t.Fatalf("create foreign workspace: %v", err)
}
t.Cleanup(func() { testPool.Exec(ctx, `DELETE FROM workspace WHERE id = $1`, foreignWorkspaceID) })
var foreignRuntimeID string
if err := testPool.QueryRow(ctx, `
INSERT INTO agent_runtime (
workspace_id, daemon_id, name, runtime_mode, provider, status, device_info, metadata, last_seen_at
)
VALUES ($1, NULL, 'mismatch-rt-hourly', 'cloud', 'mismatch-rt-hourly', 'online', '{}'::jsonb, '{}'::jsonb, now())
RETURNING id
`, foreignWorkspaceID).Scan(&foreignRuntimeID); err != nil {
t.Fatalf("create foreign runtime: %v", err)
}
var foreignAgentID string
if err := testPool.QueryRow(ctx, `
INSERT INTO agent (
workspace_id, name, description, runtime_mode, runtime_config,
runtime_id, visibility, max_concurrent_tasks, owner_id,
instructions, custom_env, custom_args, mcp_config
)
VALUES ($1, 'mismatch-agent-hourly', '', 'cloud', '{}'::jsonb, $2, 'private', 1, $3, '', '{}'::jsonb, '[]'::jsonb, '[]'::jsonb)
RETURNING id
`, foreignWorkspaceID, foreignRuntimeID, testUserID).Scan(&foreignAgentID); err != nil {
t.Fatalf("create foreign agent: %v", err)
}
// Issue lives in the primary test workspace; the agent lives in the
// foreign one — so agent.workspace_id != issue.workspace_id.
var issueID string
if err := testPool.QueryRow(ctx, `
INSERT INTO issue (workspace_id, title, creator_id, creator_type, number)
VALUES ($1, 'mismatch hourly test', $2, 'member',
(SELECT COALESCE(MAX(number), 0) + 1 FROM issue WHERE workspace_id = $1))
RETURNING id
`, testWorkspaceID, testUserID).Scan(&issueID); err != nil {
t.Fatalf("create issue: %v", err)
}
t.Cleanup(func() { testPool.Exec(ctx, `DELETE FROM issue WHERE id = $1`, issueID) })
usageAt := time.Date(2021, 9, 9, 1, 0, 0, 0, time.UTC)
var taskID string
if err := testPool.QueryRow(ctx, `
INSERT INTO agent_task_queue (agent_id, issue_id, runtime_id, status, created_at)
VALUES ($1, $2, $3, 'completed', $4) RETURNING id
`, foreignAgentID, issueID, foreignRuntimeID, usageAt).Scan(&taskID); err != nil {
t.Fatalf("insert atq: %v", err)
}
if _, err := testPool.Exec(ctx, `
INSERT INTO task_usage (task_id, provider, model, input_tokens, output_tokens, created_at, updated_at)
VALUES ($1, 'claude', 'm-mismatch-hourly', 333, 33, $2, $2)
`, taskID, usageAt); err != nil {
t.Fatalf("insert task_usage: %v", err)
}
t.Cleanup(func() {
testPool.Exec(ctx, `DELETE FROM task_usage_hourly WHERE model = 'm-mismatch-hourly'`)
testPool.Exec(ctx, `DELETE FROM task_usage_hourly_dirty WHERE model = 'm-mismatch-hourly'`)
})
if _, err := testPool.Exec(ctx, `
SELECT rollup_task_usage_hourly_window('-infinity'::timestamptz, 'infinity'::timestamptz)
`); err != nil {
t.Fatalf("rollup: %v", err)
}
wsTotal := func(ws string) int64 {
var total int64
testPool.QueryRow(ctx, `
SELECT COALESCE(SUM(input_tokens), 0) FROM task_usage_hourly
WHERE workspace_id = $1 AND model = 'm-mismatch-hourly'
`, ws).Scan(&total)
return total
}
if got := wsTotal(foreignWorkspaceID); got != 333 {
t.Fatalf("expected foreign workspace bucket = 333 (resolved from agent), got %d", got)
}
if got := wsTotal(testWorkspaceID); got != 0 {
t.Errorf("expected primary workspace bucket = 0 (issue.workspace_id must not leak), got %d", got)
}
}
// TestDashboardRollupReattributesOnProjectChange verifies the trigger that
// fires on `UPDATE issue SET project_id` enqueues both old + new project
// buckets so the next rollup tick re-attributes the affected tokens.
// Uses the rollup window function directly to drain the dirty queue,
// then asserts the rollup table reflects the new project_id.
func TestDashboardRollupReattributesOnProjectChange(t *testing.T) {
if testHandler == nil {
t.Skip("database not available")
}
ctx := context.Background()
var runtimeID, agentID string
if err := testPool.QueryRow(ctx, `SELECT id FROM agent_runtime WHERE workspace_id = $1 LIMIT 1`, testWorkspaceID).Scan(&runtimeID); err != nil {
t.Fatalf("fetch runtime: %v", err)
}
if err := testPool.QueryRow(ctx, `SELECT id FROM agent WHERE workspace_id = $1 LIMIT 1`, testWorkspaceID).Scan(&agentID); err != nil {
t.Fatalf("fetch agent: %v", err)
}
mkProject := func(name string) string {
var id string
if err := testPool.QueryRow(ctx, `
INSERT INTO project (workspace_id, title) VALUES ($1, $2) RETURNING id
`, testWorkspaceID, name).Scan(&id); err != nil {
t.Fatalf("create project: %v", err)
}
t.Cleanup(func() { testPool.Exec(ctx, `DELETE FROM project WHERE id = $1`, id) })
return id
}
projectA := mkProject("dashboard reattr A")
projectB := mkProject("dashboard reattr B")
var issueID string
if err := testPool.QueryRow(ctx, `
INSERT INTO issue (workspace_id, title, creator_id, creator_type, project_id, number)
VALUES ($1, 'reattr issue', $2, 'member', $3,
(SELECT COALESCE(MAX(number), 0) + 1 FROM issue WHERE workspace_id = $1))
RETURNING id
`, testWorkspaceID, testUserID, projectA).Scan(&issueID); err != nil {
t.Fatalf("create issue: %v", err)
}
t.Cleanup(func() { testPool.Exec(ctx, `DELETE FROM issue WHERE id = $1`, issueID) })
var taskID string
if err := testPool.QueryRow(ctx, `
INSERT INTO agent_task_queue (agent_id, issue_id, runtime_id, status, created_at)
VALUES ($1, $2, $3, 'completed', now()) RETURNING id
`, agentID, issueID, runtimeID).Scan(&taskID); err != nil {
t.Fatalf("insert task: %v", err)
}
t.Cleanup(func() { testPool.Exec(ctx, `DELETE FROM agent_task_queue WHERE id = $1`, taskID) })
if _, err := testPool.Exec(ctx, `
INSERT INTO task_usage (task_id, provider, model, input_tokens, output_tokens, created_at)
VALUES ($1, 'claude', 'claude-3-5-sonnet', 7777, 0, now())
`, taskID); err != nil {
t.Fatalf("insert task_usage: %v", err)
}
// First rollup pass: tokens attributed to project A.
if _, err := testPool.Exec(ctx, `
SELECT rollup_task_usage_hourly_window('1970-01-01'::timestamptz, now() + interval '1 hour')
`); err != nil {
t.Fatalf("rollup A: %v", err)
}
var aTokens int64
if err := testPool.QueryRow(ctx, `
SELECT COALESCE(SUM(input_tokens), 0) FROM task_usage_hourly
WHERE workspace_id = $1 AND project_id = $2 AND agent_id = $3
`, testWorkspaceID, projectA, agentID).Scan(&aTokens); err != nil {
t.Fatalf("read A rollup: %v", err)
}
if aTokens < 7777 {
t.Fatalf("project A: expected >=7777 tokens after first rollup, got %d", aTokens)
}
// Move the issue to project B. Trigger enqueues both A and B buckets.
if _, err := testPool.Exec(ctx, `UPDATE issue SET project_id = $1 WHERE id = $2`, projectB, issueID); err != nil {
t.Fatalf("reassign project: %v", err)
}
// Second rollup pass: A bucket drops to zero (deleted_empty), B
// bucket gets the tokens.
if _, err := testPool.Exec(ctx, `
SELECT rollup_task_usage_hourly_window('1970-01-01'::timestamptz, now() + interval '1 hour')
`); err != nil {
t.Fatalf("rollup B: %v", err)
}
var bTokens, aTokensAfter int64
if err := testPool.QueryRow(ctx, `
SELECT COALESCE(SUM(input_tokens), 0) FROM task_usage_hourly
WHERE workspace_id = $1 AND project_id = $2 AND agent_id = $3
`, testWorkspaceID, projectB, agentID).Scan(&bTokens); err != nil {
t.Fatalf("read B rollup: %v", err)
}
if err := testPool.QueryRow(ctx, `
SELECT COALESCE(SUM(input_tokens), 0) FROM task_usage_hourly
WHERE workspace_id = $1 AND project_id = $2 AND agent_id = $3
`, testWorkspaceID, projectA, agentID).Scan(&aTokensAfter); err != nil {
t.Fatalf("read A rollup after move: %v", err)
}
if bTokens < 7777 {
t.Errorf("project B: expected >=7777 tokens after reassign + rollup, got %d", bTokens)
}
if aTokensAfter != 0 {
t.Errorf("project A: expected 0 tokens after reassign + rollup, got %d", aTokensAfter)
}
}
// TestDashboardRollupClearsOnIssueDelete verifies that deleting an issue
// (which cascades to its tasks and task_usage rows) also clears the
// dashboard rollup row attributed to that issue's project. The
// `issue BEFORE DELETE` trigger has to fire ahead of the cascade so the
// dirty queue captures the original project_id while the issue row is
// still readable.
func TestDashboardRollupClearsOnIssueDelete(t *testing.T) {
if testHandler == nil {
t.Skip("database not available")
}
ctx := context.Background()
var runtimeID, agentID string
if err := testPool.QueryRow(ctx, `SELECT id FROM agent_runtime WHERE workspace_id = $1 LIMIT 1`, testWorkspaceID).Scan(&runtimeID); err != nil {
t.Fatalf("fetch runtime: %v", err)
}
if err := testPool.QueryRow(ctx, `SELECT id FROM agent WHERE workspace_id = $1 LIMIT 1`, testWorkspaceID).Scan(&agentID); err != nil {
t.Fatalf("fetch agent: %v", err)
}
var projectID string
if err := testPool.QueryRow(ctx, `
INSERT INTO project (workspace_id, title) VALUES ($1, 'dashboard cascade test') RETURNING id
`, testWorkspaceID).Scan(&projectID); err != nil {
t.Fatalf("create project: %v", err)
}
t.Cleanup(func() { testPool.Exec(ctx, `DELETE FROM project WHERE id = $1`, projectID) })
var issueID string
if err := testPool.QueryRow(ctx, `
INSERT INTO issue (workspace_id, title, creator_id, creator_type, project_id, number)
VALUES ($1, 'cascade issue', $2, 'member', $3,
(SELECT COALESCE(MAX(number), 0) + 1 FROM issue WHERE workspace_id = $1))
RETURNING id
`, testWorkspaceID, testUserID, projectID).Scan(&issueID); err != nil {
t.Fatalf("create issue: %v", err)
}
// No t.Cleanup deleting the issue — that's what the test exercises.
var taskID string
if err := testPool.QueryRow(ctx, `
INSERT INTO agent_task_queue (agent_id, issue_id, runtime_id, status, created_at)
VALUES ($1, $2, $3, 'completed', now()) RETURNING id
`, agentID, issueID, runtimeID).Scan(&taskID); err != nil {
t.Fatalf("insert task: %v", err)
}
// Don't bother cleaning up taskID either; cascade will take it.
if _, err := testPool.Exec(ctx, `
INSERT INTO task_usage (task_id, provider, model, input_tokens, output_tokens, created_at)
VALUES ($1, 'claude', 'claude-3-5-sonnet', 4242, 0, now())
`, taskID); err != nil {
t.Fatalf("insert task_usage: %v", err)
}
// First rollup: project bucket exists with 4242 tokens.
if _, err := testPool.Exec(ctx, `
SELECT rollup_task_usage_hourly_window('1970-01-01'::timestamptz, now() + interval '1 hour')
`); err != nil {
t.Fatalf("rollup before delete: %v", err)
}
var before int64
if err := testPool.QueryRow(ctx, `
SELECT COALESCE(SUM(input_tokens), 0) FROM task_usage_hourly
WHERE workspace_id = $1 AND project_id = $2
`, testWorkspaceID, projectID).Scan(&before); err != nil {
t.Fatalf("read before: %v", err)
}
if before < 4242 {
t.Fatalf("project bucket: expected >=4242 tokens before delete, got %d", before)
}
// Delete the issue. Cascade removes atq + task_usage. The issue
// BEFORE DELETE trigger should have enqueued the project bucket
// before the cascade started.
if _, err := testPool.Exec(ctx, `DELETE FROM issue WHERE id = $1`, issueID); err != nil {
t.Fatalf("delete issue: %v", err)
}
if _, err := testPool.Exec(ctx, `
SELECT rollup_task_usage_hourly_window('1970-01-01'::timestamptz, now() + interval '1 hour')
`); err != nil {
t.Fatalf("rollup after delete: %v", err)
}
var after int64
if err := testPool.QueryRow(ctx, `
SELECT COALESCE(SUM(input_tokens), 0) FROM task_usage_hourly
WHERE workspace_id = $1 AND project_id = $2
`, testWorkspaceID, projectID).Scan(&after); err != nil {
t.Fatalf("read after: %v", err)
}
if after != 0 {
t.Errorf("project bucket: expected 0 tokens after issue delete, got %d", after)
}
}
// TestDashboardRollupReattributesOnLinkTaskToIssue verifies that
// `LinkTaskToIssue` (which UPDATEs `agent_task_queue.issue_id` from NULL
// to a real issue id) re-attributes existing rollup rows from the
// no-project bucket to the linked issue's project bucket. Mirrors the
// quick-create flow in `service.task.LinkTaskToIssue`.
func TestDashboardRollupReattributesOnLinkTaskToIssue(t *testing.T) {
if testHandler == nil {
t.Skip("database not available")
}
ctx := context.Background()
var runtimeID, agentID string
if err := testPool.QueryRow(ctx, `SELECT id FROM agent_runtime WHERE workspace_id = $1 LIMIT 1`, testWorkspaceID).Scan(&runtimeID); err != nil {
t.Fatalf("fetch runtime: %v", err)
}
if err := testPool.QueryRow(ctx, `SELECT id FROM agent WHERE workspace_id = $1 LIMIT 1`, testWorkspaceID).Scan(&agentID); err != nil {
t.Fatalf("fetch agent: %v", err)
}
// Quick-create task: issue_id is NULL at creation time.
var taskID string
if err := testPool.QueryRow(ctx, `
INSERT INTO agent_task_queue (agent_id, issue_id, runtime_id, status, context, created_at)
VALUES ($1, NULL, $2, 'completed', '{}'::jsonb, now()) RETURNING id
`, agentID, runtimeID).Scan(&taskID); err != nil {
t.Fatalf("insert quick-create task: %v", err)
}
t.Cleanup(func() { testPool.Exec(ctx, `DELETE FROM agent_task_queue WHERE id = $1`, taskID) })
if _, err := testPool.Exec(ctx, `
INSERT INTO task_usage (task_id, provider, model, input_tokens, output_tokens, created_at)
VALUES ($1, 'claude', 'claude-3-5-sonnet', 1234, 0, now())
`, taskID); err != nil {
t.Fatalf("insert task_usage: %v", err)
}
// First rollup: tokens attributed to the no-project bucket (NULL).
if _, err := testPool.Exec(ctx, `
SELECT rollup_task_usage_hourly_window('1970-01-01'::timestamptz, now() + interval '1 hour')
`); err != nil {
t.Fatalf("rollup pre-link: %v", err)
}
var nullBefore int64
if err := testPool.QueryRow(ctx, `
SELECT COALESCE(SUM(input_tokens), 0) FROM task_usage_hourly
WHERE workspace_id = $1 AND project_id IS NULL AND agent_id = $2
`, testWorkspaceID, agentID).Scan(&nullBefore); err != nil {
t.Fatalf("read NULL bucket pre-link: %v", err)
}
if nullBefore < 1234 {
t.Fatalf("NULL bucket: expected >=1234 tokens pre-link, got %d", nullBefore)
}
// Create a project + issue, then run the same UPDATE LinkTaskToIssue
// uses. The atq trigger should enqueue OLD (NULL project) AND NEW
// (the project's id) so the next rollup tick zeroes the NULL bucket
// and populates the project bucket.
var projectID string
if err := testPool.QueryRow(ctx, `
INSERT INTO project (workspace_id, title) VALUES ($1, 'dashboard link test') RETURNING id
`, testWorkspaceID).Scan(&projectID); err != nil {
t.Fatalf("create project: %v", err)
}
t.Cleanup(func() { testPool.Exec(ctx, `DELETE FROM project WHERE id = $1`, projectID) })
var issueID string
if err := testPool.QueryRow(ctx, `
INSERT INTO issue (workspace_id, title, creator_id, creator_type, project_id, number)
VALUES ($1, 'link test issue', $2, 'member', $3,
(SELECT COALESCE(MAX(number), 0) + 1 FROM issue WHERE workspace_id = $1))
RETURNING id
`, testWorkspaceID, testUserID, projectID).Scan(&issueID); err != nil {
t.Fatalf("create issue: %v", err)
}
t.Cleanup(func() { testPool.Exec(ctx, `DELETE FROM issue WHERE id = $1`, issueID) })
// Mirror LinkTaskToIssue's UPDATE shape.
if _, err := testPool.Exec(ctx, `
UPDATE agent_task_queue SET issue_id = $1 WHERE id = $2 AND issue_id IS NULL
`, issueID, taskID); err != nil {
t.Fatalf("link task to issue: %v", err)
}
if _, err := testPool.Exec(ctx, `
SELECT rollup_task_usage_hourly_window('1970-01-01'::timestamptz, now() + interval '1 hour')
`); err != nil {
t.Fatalf("rollup post-link: %v", err)
}
var projectAfter, nullAfter int64
if err := testPool.QueryRow(ctx, `
SELECT COALESCE(SUM(input_tokens), 0) FROM task_usage_hourly
WHERE workspace_id = $1 AND project_id = $2 AND agent_id = $3
`, testWorkspaceID, projectID, agentID).Scan(&projectAfter); err != nil {
t.Fatalf("read project bucket post-link: %v", err)
}
if err := testPool.QueryRow(ctx, `
SELECT COALESCE(SUM(input_tokens), 0) FROM task_usage_hourly
WHERE workspace_id = $1 AND project_id IS NULL AND agent_id = $2
`, testWorkspaceID, agentID).Scan(&nullAfter); err != nil {
t.Fatalf("read NULL bucket post-link: %v", err)
}
if projectAfter < 1234 {
t.Errorf("project bucket: expected >=1234 tokens after link, got %d", projectAfter)
}
if nullAfter != 0 {
t.Errorf("NULL bucket: expected 0 tokens after link, got %d", nullAfter)
}
}
// TestPruneTaskUsageHourlyDirty covers the dirty-queue TTL. Both the RFC
// (§7.1) and the rollup-pipeline migration call this THE most-easily-missed correctness
// requirement of the hourly pipeline: without the prune, a row that escapes
// the per-tick drain (crash mid-tick, worker paused during an incident)
// pins its bucket's recompute forever and the queue grows unbounded.
func TestPruneTaskUsageHourlyDirty(t *testing.T) {
if testHandler == nil {
t.Skip("database not available")
}
// This test calls rollup_task_usage_hourly(), which advances the shared
// watermark as a side effect; serialise so it does not perturb another
// package's rollup test (MUL-3980).
lockRollupSingleton(t)
ctx := context.Background()
// task_usage_hourly_dirty carries no FKs (it is a queue), so synthetic
// UUIDs are fine. `provider` tags the rows for isolated cleanup.
const tag = "ttl-prune-test"
t.Cleanup(func() {
testPool.Exec(ctx, `DELETE FROM task_usage_hourly_dirty WHERE provider = $1`, tag)
})
seed := func(model, age string) {
if _, err := testPool.Exec(ctx, `
INSERT INTO task_usage_hourly_dirty (
bucket_hour, workspace_id, runtime_id, agent_id, project_id,
provider, model, enqueued_at
)
VALUES (
date_trunc('hour', now()), gen_random_uuid(), gen_random_uuid(),
gen_random_uuid(), NULL, $1, $2, now() - $3::interval
)
`, tag, model, age); err != nil {
t.Fatalf("seed dirty row %s: %v", model, err)
}
}
countModel := func(model string) int {
var n int
testPool.QueryRow(ctx,
`SELECT COUNT(*) FROM task_usage_hourly_dirty WHERE provider = $1 AND model = $2`,
tag, model,
).Scan(&n)
return n
}
seed("stale-row", "8 days")
seed("fresh-row", "1 day")
// Default 7-day retention: the 8-day row goes, the 1-day row stays.
var pruned int64
if err := testPool.QueryRow(ctx, `SELECT prune_task_usage_hourly_dirty()`).Scan(&pruned); err != nil {
t.Fatalf("prune (default retention): %v", err)
}
if pruned < 1 {
t.Errorf("expected prune to report at least the one stale row deleted, got %d", pruned)
}
if got := countModel("stale-row"); got != 0 {
t.Errorf("default prune: expected stale row deleted, still %d present", got)
}
if got := countModel("fresh-row"); got != 1 {
t.Errorf("default prune: expected fresh row kept, got %d", got)
}
// An explicit retention shorter than the surviving row's age drops it.
if _, err := testPool.Exec(ctx, `SELECT prune_task_usage_hourly_dirty(interval '12 hours')`); err != nil {
t.Fatalf("prune (12h retention): %v", err)
}
if got := countModel("fresh-row"); got != 0 {
t.Errorf("12h-retention prune: expected fresh row deleted, still %d present", got)
}
// The cron entry folds the prune in so operators do
// not need a second scheduled job. A single tick must drop a stale row.
seed("cron-fold-row", "9 days")
if _, err := testPool.Exec(ctx, `SELECT rollup_task_usage_hourly()`); err != nil {
t.Fatalf("rollup_task_usage_hourly: %v", err)
}
if got := countModel("cron-fold-row"); got != 0 {
t.Errorf("cron entry did not fold in the prune: stale row still present (%d)", got)
}
}
// TestRollupTaskUsageHourlyCapsWindowAtOneDay covers the catch-up cap
// in rollup_task_usage_hourly(): when the watermark has
// fallen far behind (worker paused for an incident or a migration freeze),
// a single tick must advance it by at most one day, so a multi-week backlog
// drains in bounded steps instead of one giant statement holding advisory
// lock 4246. The existing watermark test only parks the watermark one hour
// back, so the cap itself is never exercised there.
func TestRollupTaskUsageHourlyCapsWindowAtOneDay(t *testing.T) {
if testHandler == nil {
t.Skip("database not available")
}
// Serialise against any other package's rollup test (MUL-3980). Acquire
// the guard BEFORE registering the watermark-restore cleanup below so
// that cleanup (LIFO) runs while the guard is still held, and the guard
// is released last.
lockRollupSingleton(t)
ctx := context.Background()
// Other tests drive rollup_task_usage_hourly_window directly and never
// read the watermark; the idempotency test parks it itself. Restore to
// now() so nothing downstream observes a stale value.
t.Cleanup(func() {
testPool.Exec(ctx,
`UPDATE task_usage_hourly_rollup_state SET watermark_at = now(), last_error = NULL WHERE id = 1`)
})
park := func(behind string) {
if _, err := testPool.Exec(ctx, `
UPDATE task_usage_hourly_rollup_state
SET watermark_at = now() - $1::interval, last_error = NULL
WHERE id = 1
`, behind); err != nil {
t.Fatalf("park watermark: %v", err)
}
}
ageDays := func() float64 {
var sec float64
if err := testPool.QueryRow(ctx, `
SELECT EXTRACT(EPOCH FROM (now() - watermark_at))
FROM task_usage_hourly_rollup_state WHERE id = 1
`).Scan(&sec); err != nil {
t.Fatalf("read watermark: %v", err)
}
return sec / 86400
}
tick := func(label string) {
if _, err := testPool.Exec(ctx, `SELECT rollup_task_usage_hourly()`); err != nil {
t.Fatalf("%s: %v", label, err)
}
}
// Park 3 days back. One tick advances by exactly one day (v_from + 1d,
// well short of now()-5min), leaving the watermark ~2 days behind.
park("3 days")
tick("tick 1")
if age := ageDays(); age < 1.9 || age > 2.1 {
t.Fatalf("after one tick: expected watermark ~2 days behind, got %.3f days", age)
}
// A second tick advances another bounded day → ~1 day behind.
tick("tick 2")
if age := ageDays(); age < 0.9 || age > 1.1 {
t.Fatalf("after two ticks: expected watermark ~1 day behind, got %.3f days", age)
}
// Once within a day of now, the tick snaps the watermark to now()-5min
// (LEAST picks the now bound) rather than taking a further fixed day.
tick("tick 3")
if age := ageDays(); age > 0.02 {
t.Fatalf("after catch-up: expected watermark within minutes of now, got %.3f days", age)
}
}
// TestDashboardUsageDailyCrossMidnightFullPipeline runs the WHOLE timezone
// pipeline end to end: insert a raw `task_usage` row near UTC midnight →
// run `rollup_task_usage_hourly_window` to bucket it → call
// GetDashboardUsageDaily with a non-UTC viewer tz. It asserts the tokens
// land on the viewer's correct calendar day and NOT on the UTC day.
//
// This is the #2822 bug class the RFC exists to prevent. The existing
// TestDashboardUsageDailyBucketsByViewerTimezone seeds a pre-built
// task_usage_hourly row and only exercises the SQL read path; here the
// row travels from raw task_usage through the rollup, so a regression in
// task_usage_hour_bucket or the recompute join is also caught.
func TestDashboardUsageDailyCrossMidnightFullPipeline(t *testing.T) {
if testHandler == nil {
t.Skip("database not available")
}
ctx := context.Background()
var runtimeID, agentID string
if err := testPool.QueryRow(ctx, `SELECT id FROM agent_runtime WHERE workspace_id = $1 LIMIT 1`, testWorkspaceID).Scan(&runtimeID); err != nil {
t.Fatalf("fetch runtime: %v", err)
}
if err := testPool.QueryRow(ctx, `SELECT id FROM agent WHERE workspace_id = $1 LIMIT 1`, testWorkspaceID).Scan(&agentID); err != nil {
t.Fatalf("fetch agent: %v", err)
}
var issueID string
if err := testPool.QueryRow(ctx, `
INSERT INTO issue (workspace_id, title, creator_id, creator_type, number)
VALUES ($1, 'cross-midnight pipeline test', $2, 'member',
(SELECT COALESCE(MAX(number), 0) + 1 FROM issue WHERE workspace_id = $1))
RETURNING id
`, testWorkspaceID, testUserID).Scan(&issueID); err != nil {
t.Fatalf("create issue: %v", err)
}
t.Cleanup(func() { testPool.Exec(ctx, `DELETE FROM issue WHERE id = $1`, issueID) })
var taskID string
if err := testPool.QueryRow(ctx, `
INSERT INTO agent_task_queue (agent_id, issue_id, runtime_id, status, created_at)
VALUES ($1, $2, $3, 'completed', now()) RETURNING id
`, agentID, issueID, runtimeID).Scan(&taskID); err != nil {
t.Fatalf("insert task: %v", err)
}
t.Cleanup(func() { testPool.Exec(ctx, `DELETE FROM agent_task_queue WHERE id = $1`, taskID) })
// Raw task_usage at 00:30 UTC two days ago — genuinely near UTC
// midnight. 00:30 UTC is still the PRIOR evening (~16:30/17:30) in
// America/Los_Angeles (UTC-7/-8), so the UTC viewer and the LA viewer
// must see this row under different calendar days. Using CURRENT_DATE
// keeps the row inside the days=10 window without a fixed-date drift.
var usageAt time.Time
if err := testPool.QueryRow(ctx, `
INSERT INTO task_usage (task_id, provider, model, input_tokens, output_tokens, created_at)
VALUES (
$1, 'claude', 'cross-midnight-model', 8888, 0,
((CURRENT_DATE - 2)::timestamp + interval '30 minutes') AT TIME ZONE 'UTC'
)
RETURNING created_at
`, taskID).Scan(&usageAt); err != nil {
t.Fatalf("insert task_usage: %v", err)
}
t.Cleanup(func() {
testPool.Exec(ctx, `DELETE FROM task_usage_hourly WHERE model = 'cross-midnight-model'`)
})
// Run the rollup so the raw row is aggregated into task_usage_hourly.
if _, err := testPool.Exec(ctx, `
SELECT rollup_task_usage_hourly_window('1970-01-01'::timestamptz, now() + interval '1 hour')
`); err != nil {
t.Fatalf("rollup window: %v", err)
}
utcDate := usageAt.UTC().Format("2006-01-02")
laLoc, err := time.LoadLocation("America/Los_Angeles")
if err != nil {
t.Fatalf("load LA location: %v", err)
}
laDate := usageAt.In(laLoc).Format("2006-01-02")
if utcDate == laDate {
t.Fatalf("test setup: UTC and LA dates must differ, both %s", utcDate)
}
readDate := func(tz string) string {
w := httptest.NewRecorder()
testHandler.GetDashboardUsageDaily(w, newRequest("GET", "/api/dashboard/usage/daily?days=10&tz="+tz, nil))
if w.Code != http.StatusOK {
t.Fatalf("tz=%s: expected 200, got %d: %s", tz, w.Code, w.Body.String())
}
var rows []struct {
Date string `json:"date"`
Model string `json:"model"`
InputTokens int64 `json:"input_tokens"`
}
_ = json.NewDecoder(w.Body).Decode(&rows)
for _, r := range rows {
if r.Model == "cross-midnight-model" {
if r.InputTokens != 8888 {
t.Errorf("tz=%s: expected 8888 tokens, got %d", tz, r.InputTokens)
}
return r.Date
}
}
t.Fatalf("tz=%s: cross-midnight-model row not found in %v", tz, rows)
return ""
}
if got := readDate("UTC"); got != utcDate {
t.Errorf("UTC viewer: expected date %s, got %s", utcDate, got)
}
if got := readDate("America/Los_Angeles"); got != laDate {
t.Errorf("LA viewer: expected date %s, got %s; row must NOT land on the UTC day %s",
laDate, got, utcDate)
}
}
// TestRollupTaskUsageHourlyConvergesOnTaskUsageDelete covers the
// `trg_tu_dirty_hourly` trigger — a BEFORE DELETE trigger on task_usage.
// Migration 102 notes it has no production callers today and exists purely
// as defensive convergence guard, so a single minimal test is enough:
// seed a task_usage row, roll it up, DELETE the task_usage row directly,
// roll up again, and assert the hourly bucket is recomputed down to zero.
// Without the trigger the deleted row's bucket would never be re-enqueued.
func TestRollupTaskUsageHourlyConvergesOnTaskUsageDelete(t *testing.T) {
if testHandler == nil {
t.Skip("database not available")
}
ctx := context.Background()
var runtimeID, agentID string
if err := testPool.QueryRow(ctx, `SELECT id FROM agent_runtime WHERE workspace_id = $1 LIMIT 1`, testWorkspaceID).Scan(&runtimeID); err != nil {
t.Fatalf("fetch runtime: %v", err)
}
if err := testPool.QueryRow(ctx, `SELECT id FROM agent WHERE workspace_id = $1 LIMIT 1`, testWorkspaceID).Scan(&agentID); err != nil {
t.Fatalf("fetch agent: %v", err)
}
var issueID string
if err := testPool.QueryRow(ctx, `
INSERT INTO issue (workspace_id, title, creator_id, creator_type, number)
VALUES ($1, 'tu-delete trigger test', $2, 'member',
(SELECT COALESCE(MAX(number), 0) + 1 FROM issue WHERE workspace_id = $1))
RETURNING id
`, testWorkspaceID, testUserID).Scan(&issueID); err != nil {
t.Fatalf("create issue: %v", err)
}
t.Cleanup(func() { testPool.Exec(ctx, `DELETE FROM issue WHERE id = $1`, issueID) })
var taskID string
if err := testPool.QueryRow(ctx, `
INSERT INTO agent_task_queue (agent_id, issue_id, runtime_id, status, created_at)
VALUES ($1, $2, $3, 'completed', now() - interval '30 minutes') RETURNING id
`, agentID, issueID, runtimeID).Scan(&taskID); err != nil {
t.Fatalf("insert task: %v", err)
}
t.Cleanup(func() { testPool.Exec(ctx, `DELETE FROM agent_task_queue WHERE id = $1`, taskID) })
var usageID string
if err := testPool.QueryRow(ctx, `
INSERT INTO task_usage (task_id, provider, model, input_tokens, output_tokens, created_at)
VALUES ($1, 'claude', 'tu-delete-model', 5050, 0, now() - interval '30 minutes')
RETURNING id
`, taskID).Scan(&usageID); err != nil {
t.Fatalf("insert task_usage: %v", err)
}
t.Cleanup(func() {
testPool.Exec(ctx, `DELETE FROM task_usage_hourly WHERE model = 'tu-delete-model'`)
testPool.Exec(ctx, `DELETE FROM task_usage_hourly_dirty WHERE model = 'tu-delete-model'`)
})
bucketTotal := func() int64 {
var total int64
testPool.QueryRow(ctx, `
SELECT COALESCE(SUM(input_tokens), 0) FROM task_usage_hourly
WHERE runtime_id = $1 AND model = 'tu-delete-model'
`, runtimeID).Scan(&total)
return total
}
runWindow := func(label string) {
if _, err := testPool.Exec(ctx, `
SELECT rollup_task_usage_hourly_window('1970-01-01'::timestamptz, now() + interval '1 hour')
`); err != nil {
t.Fatalf("%s: %v", label, err)
}
}
runWindow("initial rollup")
if got := bucketTotal(); got != 5050 {
t.Fatalf("initial: expected bucket = 5050, got %d", got)
}
// Delete the task_usage row directly — fires trg_tu_dirty_hourly,
// which enqueues the bucket on task_usage_hourly_dirty.
if _, err := testPool.Exec(ctx, `DELETE FROM task_usage WHERE id = $1`, usageID); err != nil {
t.Fatalf("delete task_usage: %v", err)
}
var dirtyCount int
testPool.QueryRow(ctx, `SELECT COUNT(*) FROM task_usage_hourly_dirty WHERE model = 'tu-delete-model'`).Scan(&dirtyCount)
if dirtyCount != 1 {
t.Fatalf("expected 1 dirty entry from task_usage DELETE trigger, got %d", dirtyCount)
}
runWindow("rollup after delete")
if got := bucketTotal(); got != 0 {
t.Errorf("after delete: expected bucket recomputed to 0, got %d", got)
}
}
// TestDashboardFailuresCountNeverStartedTasks pins the reason the failure
// rollups exist as their own queries rather than reusing the run-time ones:
// ListDashboardRunTimeDaily / ListDashboardAgentRunTime require
// `started_at IS NOT NULL`, so a task that expired in the queue — the exact
// signature of a runtime outage — contributes nothing to their failed_count.
// The failure endpoints must count it, and must report the succeeded tasks in
// the same payload so the client's error rate has a matching denominator.
func TestDashboardFailuresCountNeverStartedTasks(t *testing.T) {
if testHandler == nil {
t.Skip("database not available")
}
ctx := context.Background()
var runtimeID, agentID string
if err := testPool.QueryRow(ctx, `SELECT id FROM agent_runtime WHERE workspace_id = $1 LIMIT 1`, testWorkspaceID).Scan(&runtimeID); err != nil {
t.Fatalf("fetch runtime: %v", err)
}
if err := testPool.QueryRow(ctx, `SELECT id FROM agent WHERE workspace_id = $1 LIMIT 1`, testWorkspaceID).Scan(&agentID); err != nil {
t.Fatalf("fetch agent: %v", err)
}
var issueID string
if err := testPool.QueryRow(ctx, `
INSERT INTO issue (workspace_id, title, creator_id, creator_type, number)
VALUES ($1, 'failures rollup test', $2, 'member',
(SELECT COALESCE(MAX(number), 0) + 1 FROM issue WHERE workspace_id = $1))
RETURNING id
`, testWorkspaceID, testUserID).Scan(&issueID); err != nil {
t.Fatalf("create issue: %v", err)
}
t.Cleanup(func() { testPool.Exec(ctx, `DELETE FROM issue WHERE id = $1`, issueID) })
now := time.Now().UTC()
started := now.Add(-30 * time.Minute)
completed := started.Add(10 * time.Minute)
// startedAt is nullable so the queue-expiry case can be modelled exactly:
// completed_at set, started_at absent.
mkTask := func(status string, failureReason any, startedAt any) {
var taskID string
if err := testPool.QueryRow(ctx, `
INSERT INTO agent_task_queue (agent_id, issue_id, runtime_id, status, started_at, completed_at, failure_reason, created_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, now())
RETURNING id
`, agentID, issueID, runtimeID, status, startedAt, completed, failureReason).Scan(&taskID); err != nil {
t.Fatalf("insert task: %v", err)
}
t.Cleanup(func() { testPool.Exec(ctx, `DELETE FROM agent_task_queue WHERE id = $1`, taskID) })
}
mkTask("completed", nil, started)
mkTask("failed", "agent_error.provider_auth_or_access", started)
mkTask("failed", "queued_expired", nil) // never started
mkTask("failed", nil, started) // unclassified: empty reason column
type failureRow struct {
Date string `json:"date"`
AgentID string `json:"agent_id"`
FailureReason string `json:"failure_reason"`
TaskCount int32 `json:"task_count"`
}
// The fixture workspace is shared, so other tests' rows may be in the
// window too. Assert on the buckets this test wrote rather than on the
// whole payload.
collect := func(rows []failureRow) map[string]int32 {
byReason := map[string]int32{}
for _, r := range rows {
byReason[r.FailureReason] += r.TaskCount
}
return byReason
}
for _, tc := range []struct {
name string
call func(w *httptest.ResponseRecorder)
}{
{"daily", func(w *httptest.ResponseRecorder) {
testHandler.GetDashboardFailuresDaily(w, newRequest("GET", "/api/dashboard/failures/daily?days=1", nil))
}},
{"by-agent", func(w *httptest.ResponseRecorder) {
testHandler.GetDashboardFailuresByAgent(w, newRequest("GET", "/api/dashboard/failures/by-agent?days=1", nil))
}},
} {
t.Run(tc.name, func(t *testing.T) {
w := httptest.NewRecorder()
tc.call(w)
if w.Code != http.StatusOK {
t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String())
}
var rows []failureRow
if err := json.NewDecoder(w.Body).Decode(&rows); err != nil {
t.Fatalf("decode: %v", err)
}
byReason := collect(rows)
if byReason["agent_error.provider_auth_or_access"] < 1 {
t.Errorf("expected the classified failure to be counted, got %v", byReason)
}
// The point of the whole endpoint: a failure with no started_at
// still lands in a bucket.
if byReason["queued_expired"] < 1 {
t.Errorf("expected the never-started failure to be counted, got %v", byReason)
}
// A failed row with an empty failure_reason must not be mistaken
// for a success — that would deflate the error rate.
if byReason["unclassified"] < 1 {
t.Errorf("expected the reason-less failure to be counted as unclassified, got %v", byReason)
}
// Succeeded tasks ride along under the empty-string key so the
// client can compute a rate from one payload.
if byReason[""] < 1 {
t.Errorf("expected succeeded tasks in the denominator bucket, got %v", byReason)
}
})
}
}
// TestDashboardFailuresByAgentUsesExactWindow pins the cutoff difference
// between the two failure endpoints.
//
// parseSinceParamInTZ deliberately returns N+1 calendar days of headroom, and
// the workspace dashboard trims the surplus client-side with `-(days-1)`. The
// by-agent rollup carries no date column, so it cannot be trimmed that way —
// it must close its own window server-side. Before that fix, `days=1` served
// the Errors card yesterday's failures while the chart beside it, correctly
// trimmed, showed none.
func TestDashboardFailuresByAgentUsesExactWindow(t *testing.T) {
if testHandler == nil {
t.Skip("database not available")
}
ctx := context.Background()
var runtimeID, agentID string
if err := testPool.QueryRow(ctx, `SELECT id FROM agent_runtime WHERE workspace_id = $1 LIMIT 1`, testWorkspaceID).Scan(&runtimeID); err != nil {
t.Fatalf("fetch runtime: %v", err)
}
if err := testPool.QueryRow(ctx, `SELECT id FROM agent WHERE workspace_id = $1 LIMIT 1`, testWorkspaceID).Scan(&agentID); err != nil {
t.Fatalf("fetch agent: %v", err)
}
var issueID string
if err := testPool.QueryRow(ctx, `
INSERT INTO issue (workspace_id, title, creator_id, creator_type, number)
VALUES ($1, 'failures window test', $2, 'member',
(SELECT COALESCE(MAX(number), 0) + 1 FROM issue WHERE workspace_id = $1))
RETURNING id
`, testWorkspaceID, testUserID).Scan(&issueID); err != nil {
t.Fatalf("create issue: %v", err)
}
t.Cleanup(func() { testPool.Exec(ctx, `DELETE FROM issue WHERE id = $1`, issueID) })
// One failure at noon YESTERDAY (UTC). days=1 means "today", so neither
// endpoint may count it. Noon avoids the midnight edge in either
// direction.
var taskID string
if err := testPool.QueryRow(ctx, `
INSERT INTO agent_task_queue (agent_id, issue_id, runtime_id, status, started_at, completed_at, failure_reason, created_at)
VALUES (
$1, $2, $3, 'failed',
((CURRENT_DATE - 1)::timestamp + interval '11 hours') AT TIME ZONE 'UTC',
((CURRENT_DATE - 1)::timestamp + interval '12 hours') AT TIME ZONE 'UTC',
'timeout', now()
)
RETURNING id
`, agentID, issueID, runtimeID).Scan(&taskID); err != nil {
t.Fatalf("insert task: %v", err)
}
t.Cleanup(func() { testPool.Exec(ctx, `DELETE FROM agent_task_queue WHERE id = $1`, taskID) })
type failureRow struct {
FailureReason string `json:"failure_reason"`
TaskCount int32 `json:"task_count"`
}
countTimeouts := func(body []byte) int32 {
var rows []failureRow
if err := json.Unmarshal(body, &rows); err != nil {
t.Fatalf("decode: %v", err)
}
var n int32
for _, r := range rows {
if r.FailureReason == "timeout" {
n += r.TaskCount
}
}
return n
}
w := httptest.NewRecorder()
testHandler.GetDashboardFailuresByAgent(w, newRequest("GET", "/api/dashboard/failures/by-agent?days=1&tz=UTC", nil))
if w.Code != http.StatusOK {
t.Fatalf("by-agent: expected 200, got %d: %s", w.Code, w.Body.String())
}
if got := countTimeouts(w.Body.Bytes()); got != 0 {
t.Errorf("days=1 must not reach yesterday's failure, but by-agent counted %d", got)
}
// days=2 covers today + yesterday, so the same row must now appear —
// proving the window was closed, not that the fixture is unreachable.
w = httptest.NewRecorder()
testHandler.GetDashboardFailuresByAgent(w, newRequest("GET", "/api/dashboard/failures/by-agent?days=2&tz=UTC", nil))
if w.Code != http.StatusOK {
t.Fatalf("by-agent days=2: expected 200, got %d: %s", w.Code, w.Body.String())
}
if got := countTimeouts(w.Body.Bytes()); got < 1 {
t.Errorf("days=2 must include yesterday's failure, got %d", got)
}
}
// TestDashboardFailureWireContractKeepsEmptyReason pins the success bucket's
// wire form. The client's zod schema defaults a missing `failure_reason` to
// "" — the succeeded bucket — which is only safe while the server always
// emits the field. Adding `omitempty` to the struct tag would strip it from
// exactly the success rows and silently turn every window into a 100% error
// rate, so that regression is caught here rather than in a dashboard.
func TestDashboardFailureWireContractKeepsEmptyReason(t *testing.T) {
// Each case decodes into its OWN map. json.Unmarshal merges into a
// non-nil map rather than resetting it, so sharing one across cases would
// leave the first payload's failure_reason in place and let a later
// omitempty regression pass unnoticed — the exact failure this test
// exists to catch.
for _, tc := range []struct {
name string
row any
}{
{"daily", DashboardFailureDailyResponse{Date: "2026-05-19", TaskCount: 3}},
{"by-agent", DashboardFailureByAgentResponse{AgentID: "a", TaskCount: 3}},
} {
t.Run(tc.name, func(t *testing.T) {
body, err := json.Marshal(tc.row)
if err != nil {
t.Fatalf("marshal: %v", err)
}
decoded := map[string]any{}
if err := json.Unmarshal(body, &decoded); err != nil {
t.Fatalf("unmarshal: %v", err)
}
if _, present := decoded["failure_reason"]; !present {
t.Errorf("succeeded rows must serialize an explicit empty failure_reason, got %s", body)
}
})
}
}