Files
state-hub/api/schemas/token_event.py
tegwick acb30978cd feat(token-tracking): repo aggregation via graph walk (task→workstream→repo)
By Repo now resolves via the full chain rather than requiring repo_id
directly on the token event:
  1. token_events.repo_id (direct)
  2. → workstreams.repo_id (via workstream_id)
  3. → task.workstream_id → workstreams.repo_id (via task_id)

Changes:
- Auto-populate repo_id on token events at creation time (both the
  token_events router and the tasks router)
- New GET /token-events/by-repo/ endpoint with RepoTokenSummary schema;
  returns tokens_in/out/total, event_count, by_model, by_note per repo
- Dashboard By Repo section uses /by-repo/ directly and shows repo_slug
  instead of a truncated UUID
- Backfilled the three existing events (userbased) with repo_id via SQL

185 tests pass.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-03-29 19:05:23 +02:00

64 lines
1.5 KiB
Python

import uuid
from datetime import datetime
from pydantic import BaseModel, ConfigDict, computed_field
class TokenEventCreate(BaseModel):
tokens_in: int
tokens_out: int
task_id: uuid.UUID | None = None
workstream_id: uuid.UUID | None = None
repo_id: uuid.UUID | None = None
session_id: str | None = None
model: str | None = None
agent: str | None = None
ref_type: str | None = None
ref_id: str | None = None
note: str | None = None
class TokenEventRead(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: uuid.UUID
tokens_in: int
tokens_out: int
task_id: uuid.UUID | None = None
workstream_id: uuid.UUID | None = None
repo_id: uuid.UUID | None = None
session_id: str | None = None
model: str | None = None
agent: str | None = None
ref_type: str | None = None
ref_id: str | None = None
note: str | None = None
created_at: datetime
@computed_field
@property
def tokens_total(self) -> int:
return self.tokens_in + self.tokens_out
class TokenSummary(BaseModel):
scope: str
scope_id: str
tokens_in: int
tokens_out: int
tokens_total: int
event_count: int
by_model: dict[str, int]
by_agent: dict[str, int]
class RepoTokenSummary(BaseModel):
repo_id: uuid.UUID
repo_slug: str
tokens_in: int
tokens_out: int
tokens_total: int
event_count: int
by_model: dict[str, int]
by_note: dict[str, int]