mirror of
https://github.com/multica-ai/multica.git
synced 2026-06-17 11:48:42 +02:00
* fix(server): aggregate task_usage into daily rollup table to cut DB load ListRuntimeUsage previously did a SUM(...) GROUP BY DATE(created_at), provider, model over the raw task_usage stream once per runtime row on the runtimes list and once per detail page load, scaling O(events) per call. This is the hot read path responsible for sustained load on Postgres. Switch the read path to a materialized daily rollup table maintained by a pg_cron job: - 072_task_usage_daily_rollup: schema for task_usage_daily + task_usage_rollup_state, plus rollup_task_usage_daily_window(p_from, p_to) (window primitive used by both cron and offline backfill, idempotent via ON CONFLICT DO UPDATE adding deltas) and rollup_task_usage_daily() (cron entry point — pg_try_advisory_lock(4242) for serialization, watermark advancement, 5-minute safety lag for late-visible inserts). Also adds idx_task_usage_created_at to help the two lazy endpoints (ListRuntimeUsageByAgent / GetRuntimeUsageByHour) that still hit the raw table. - 073_task_usage_daily_pgcron: CREATE EXTENSION IF NOT EXISTS pg_cron in a DO/EXCEPTION block (mirrors the migration 032 pg_bigm pattern so envs without shared_preload_libraries=pg_cron skip gracefully) and schedules rollup_task_usage_daily() every 5 minutes when the extension is present. - queries/runtime_usage.sql ListRuntimeUsage rewritten to read from task_usage_daily; sqlc regenerated. Other usage queries unchanged. - cmd/backfill_task_usage_daily: one-shot Go command that walks task_usage in monthly slices through rollup_task_usage_daily_window, then stamps the watermark to now()-5m so the cron resumes cleanly. Run once after migrations have applied, before relying on the rollup. - runtime_test.go: TestGetRuntimeUsage_BucketsByUsageTime now invokes rollup_task_usage_daily_window after fixture inserts so the handler sees the rolled-up rows. Synthetic daily rows cleaned up after each test. - runtime_rollup_test.go: new tests covering aggregation correctness, idempotency contract of ON CONFLICT DO UPDATE, and the watermark advancing exactly to now()-5m via the cron entry point. Deployment order: apply migrations → run backfill_task_usage_daily once → pg_cron picks up subsequent windows automatically. Today bucket may be up to ~10 minutes stale (5 min cron + 5 min lag) by design. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Co-authored-by: multica-agent <github@multica.ai> * fix(server): make task_usage_daily rollup safe to overlap, replay, and correct Addresses 4 review blockers on the original PR: 1. Cron/backfill double-count race: the rollup function is now idempotent. Window calls find DIRTY KEYS via task_usage.updated_at, then RECOMPUTE each bucket from ground truth and REPLACE the daily row (no more additive ON CONFLICT). Cron and backfill can now overlap safely. 2. Silent pg_cron absence: the read path is gated behind a new USAGE_DAILY_ROLLUP_ENABLED feature flag (default off). The raw task_usage scan is preserved as the fallback. Operators flip the flag per-environment after backfill + cron are confirmed healthy (task_usage_rollup_lag_seconds() helper added for monitoring). 3. UpsertTaskUsage corrections invisible to rollup: added task_usage.updated_at column (default now(), backfilled from created_at), and bumped it on conflict. Corrections now mark the bucket dirty and the next window call recomputes it correctly. 4. CREATE INDEX blocking writes on hot table: split into separate single-statement migrations using CREATE INDEX CONCURRENTLY (074, 075), matching the 035/067 pattern. Also: cron.schedule() removed from migrations entirely. Migration 076 only enables the extension (gracefully on unsupported envs); the actual schedule is a documented operator runbook step that runs AFTER backfill. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Co-authored-by: multica-agent <github@multica.ai> * fix(server): trigger-driven invalidation + online-safe migration for task_usage_daily Round-2 review feedback on PR #2256: 1. Add explicit dirty-bucket queue (task_usage_daily_dirty) populated by triggers on agent_task_queue (UPDATE OF runtime_id, DELETE) and task_usage (DELETE). The rollup window function drains both this queue and the updated_at-based discovery, so runtime reassignment and issue-cascade deletes no longer leave the rollup divergent from the raw query. Triggers join via agent (not issue) to look up workspace_id, because when the cascade comes from issue, the issue row is already gone by the time atq's BEFORE DELETE fires; agent stays alive. 2. Make migration 072 online-safe: only ADD COLUMN updated_at TIMESTAMPTZ (nullable, no default → metadata-only ALTER, no row rewrite) and a separate ALTER for SET DEFAULT now() (also metadata-only). No bulk UPDATE on the hot task_usage table. The rollup window function's dirty_keys CTE handles legacy NULL rows via an OR branch, supported by partial index idx_task_usage_created_at_legacy. 3. Refresh stale documentation in cmd/backfill_task_usage_daily/main.go header to describe the current recompute/replace semantics, idempotent re-runnability, and the actual migration numbering (072..077). Tests: - TestRollupTaskUsageDaily_InvalidationOnReassign: verifies usage moves between runtime buckets after ReassignTasksToRuntime-style update. - TestRollupTaskUsageDaily_InvalidationOnIssueDelete: verifies daily bucket is cleared after issue delete cascades through atq → task_usage. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Co-authored-by: multica-agent <github@multica.ai> * fix(server): close dirty-queue race + move legacy partial index to its own concurrent migration Round-3 review feedback on PR #2256: 1. Blocker: dirty-queue invalidations could be silently lost under concurrency. ON CONFLICT DO NOTHING let a late trigger see the row already enqueued, no-op, and then the rollup drain (WHERE enqueued_at < p_to) would delete the original row — losing the late invalidation. Switched all three trigger enqueue paths to ON CONFLICT DO UPDATE SET enqueued_at = GREATEST(existing, EXCLUDED.enqueued_at), so any invalidation arriving during a rollup tick keeps enqueued_at > p_to (p_to = now() - 5min) and survives the post-tick drain. 2. High: idx_task_usage_created_at_legacy (partial index on hot task_usage table) was being created in the regular 077 migration without CONCURRENTLY. Moved to new migration 078 with CREATE INDEX CONCURRENTLY, matching the pattern of 074/075. 077's down migration leaves the index alone (it is owned by 078). 3. Minor: gofmt -w on runtime_rollup_test.go and backfill_task_usage_daily/main.go (tabs were lost in the original heredoc append). PR description rewritten to describe the current recompute/replace + dirty queue + feature flag design and the 072..078 migration ordering. Tests still green: TestRollupTaskUsageDaily_* (including both new invalidation regressions), TestGetRuntimeUsage_*, TestWorkspaceUsage_*. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Co-authored-by: multica-agent <github@multica.ai> * fix(server): unify workspace_id source via agent in rollup window function Round-4 review feedback (J) on PR #2256: M1 (must-fix): The dirty queue triggers resolved workspace_id via `agent.workspace_id`, but the window function's `dirty_from_updates` discovery and `recomputed` recompute join used `issue.workspace_id`. There is no schema-level FK guaranteeing `agent.workspace_id == issue.workspace_id`. Any divergence (future cross-workspace task scenarios, data repairs, migration bugs) would cause: - dirty queue rows with workspace_id from agent - recompute join filtering by workspace_id from issue - 0 matches in recompute → bucket erroneously hits the deleted_empty branch and the daily row is silently dropped - dirty_from_updates path attributing usage to the wrong workspace Replaced both CTEs to JOIN agent (not issue) so trigger / discovery / recompute share one workspace_id source. Comment in 077 explains the constraint. N1: Refreshed two stale references in cmd/backfill_task_usage_daily/main.go (header now says "072..078"; stampWatermark warning now mentions migration 073, where the rollup state table is actually introduced). Test: New TestRollupTaskUsageDaily_WorkspaceMismatch constructs an atq with agent.workspace_id != issue.workspace_id, asserts the bucket lands under agent's workspace (not issue's), and re-asserts after a runtime reassign in the foreign workspace. Acts as a canary if the schema invariant changes. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Co-authored-by: multica-agent <github@multica.ai> --------- Co-authored-by: Eve <eve@multica.ai> Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Co-authored-by: multica-agent <github@multica.ai> Co-authored-by: Devv <devv@Devvs-Mac-mini.local>
492 lines
16 KiB
Go
492 lines
16 KiB
Go
package handler
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"log/slog"
|
|
"net/http"
|
|
"strconv"
|
|
"time"
|
|
|
|
"github.com/go-chi/chi/v5"
|
|
"github.com/jackc/pgx/v5/pgtype"
|
|
"github.com/multica-ai/multica/server/pkg/agent"
|
|
db "github.com/multica-ai/multica/server/pkg/db/generated"
|
|
"github.com/multica-ai/multica/server/pkg/protocol"
|
|
)
|
|
|
|
type AgentRuntimeResponse struct {
|
|
ID string `json:"id"`
|
|
WorkspaceID string `json:"workspace_id"`
|
|
DaemonID *string `json:"daemon_id"`
|
|
Name string `json:"name"`
|
|
RuntimeMode string `json:"runtime_mode"`
|
|
Provider string `json:"provider"`
|
|
LaunchHeader string `json:"launch_header"`
|
|
Status string `json:"status"`
|
|
DeviceInfo string `json:"device_info"`
|
|
Metadata any `json:"metadata"`
|
|
OwnerID *string `json:"owner_id"`
|
|
LastSeenAt *string `json:"last_seen_at"`
|
|
CreatedAt string `json:"created_at"`
|
|
UpdatedAt string `json:"updated_at"`
|
|
}
|
|
|
|
func runtimeToResponse(rt db.AgentRuntime) AgentRuntimeResponse {
|
|
var metadata any
|
|
if rt.Metadata != nil {
|
|
json.Unmarshal(rt.Metadata, &metadata)
|
|
}
|
|
if metadata == nil {
|
|
metadata = map[string]any{}
|
|
}
|
|
|
|
return AgentRuntimeResponse{
|
|
ID: uuidToString(rt.ID),
|
|
WorkspaceID: uuidToString(rt.WorkspaceID),
|
|
DaemonID: textToPtr(rt.DaemonID),
|
|
Name: rt.Name,
|
|
RuntimeMode: rt.RuntimeMode,
|
|
Provider: rt.Provider,
|
|
LaunchHeader: agent.LaunchHeader(rt.Provider),
|
|
Status: rt.Status,
|
|
DeviceInfo: rt.DeviceInfo,
|
|
Metadata: metadata,
|
|
OwnerID: uuidToPtr(rt.OwnerID),
|
|
LastSeenAt: timestampToPtr(rt.LastSeenAt),
|
|
CreatedAt: timestampToString(rt.CreatedAt),
|
|
UpdatedAt: timestampToString(rt.UpdatedAt),
|
|
}
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Runtime Usage
|
|
// ---------------------------------------------------------------------------
|
|
|
|
type RuntimeUsageResponse struct {
|
|
RuntimeID string `json:"runtime_id"`
|
|
Date string `json:"date"`
|
|
Provider string `json:"provider"`
|
|
Model string `json:"model"`
|
|
InputTokens int64 `json:"input_tokens"`
|
|
OutputTokens int64 `json:"output_tokens"`
|
|
CacheReadTokens int64 `json:"cache_read_tokens"`
|
|
CacheWriteTokens int64 `json:"cache_write_tokens"`
|
|
}
|
|
|
|
// GetRuntimeUsage returns daily token usage for a runtime, aggregated from
|
|
// per-task usage records captured by the daemon. This is scoped to
|
|
// Daemon-executed tasks only (i.e. excludes users' local CLI usage of the
|
|
// same tool).
|
|
func (h *Handler) GetRuntimeUsage(w http.ResponseWriter, r *http.Request) {
|
|
runtimeID := chi.URLParam(r, "runtimeId")
|
|
runtimeUUID, ok := parseUUIDOrBadRequest(w, runtimeID, "runtime_id")
|
|
if !ok {
|
|
return
|
|
}
|
|
|
|
rt, err := h.Queries.GetAgentRuntime(r.Context(), runtimeUUID)
|
|
if err != nil {
|
|
writeError(w, http.StatusNotFound, "runtime not found")
|
|
return
|
|
}
|
|
|
|
if _, ok := h.requireWorkspaceMember(w, r, uuidToString(rt.WorkspaceID), "runtime not found"); !ok {
|
|
return
|
|
}
|
|
|
|
since := parseSinceParam(r, 90)
|
|
|
|
resp, err := h.listRuntimeUsage(r.Context(), rt.ID, since)
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to list usage")
|
|
return
|
|
}
|
|
|
|
writeJSON(w, http.StatusOK, resp)
|
|
}
|
|
|
|
// listRuntimeUsage dispatches between the raw task_usage scan and the
|
|
// task_usage_daily rollup based on the UseDailyRollupForRuntimeUsage
|
|
// feature flag. Both code paths return rows in the same shape, so the
|
|
// handler doesn't care which one ran.
|
|
func (h *Handler) listRuntimeUsage(ctx context.Context, runtimeID pgtype.UUID, since pgtype.Timestamptz) ([]RuntimeUsageResponse, error) {
|
|
resolvedRuntimeID := uuidToString(runtimeID)
|
|
if h.cfg.UseDailyRollupForRuntimeUsage {
|
|
rows, err := h.Queries.ListRuntimeUsageDaily(ctx, db.ListRuntimeUsageDailyParams{
|
|
RuntimeID: runtimeID,
|
|
Since: since,
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
resp := make([]RuntimeUsageResponse, len(rows))
|
|
for i, row := range rows {
|
|
resp[i] = RuntimeUsageResponse{
|
|
RuntimeID: resolvedRuntimeID,
|
|
Date: row.Date.Time.Format("2006-01-02"),
|
|
Provider: row.Provider,
|
|
Model: row.Model,
|
|
InputTokens: row.InputTokens,
|
|
OutputTokens: row.OutputTokens,
|
|
CacheReadTokens: row.CacheReadTokens,
|
|
CacheWriteTokens: row.CacheWriteTokens,
|
|
}
|
|
}
|
|
return resp, nil
|
|
}
|
|
|
|
rows, err := h.Queries.ListRuntimeUsage(ctx, db.ListRuntimeUsageParams{
|
|
RuntimeID: runtimeID,
|
|
Since: since,
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
resp := make([]RuntimeUsageResponse, len(rows))
|
|
for i, row := range rows {
|
|
resp[i] = RuntimeUsageResponse{
|
|
RuntimeID: resolvedRuntimeID,
|
|
Date: row.Date.Time.Format("2006-01-02"),
|
|
Provider: row.Provider,
|
|
Model: row.Model,
|
|
InputTokens: row.InputTokens,
|
|
OutputTokens: row.OutputTokens,
|
|
CacheReadTokens: row.CacheReadTokens,
|
|
CacheWriteTokens: row.CacheWriteTokens,
|
|
}
|
|
}
|
|
return resp, nil
|
|
}
|
|
|
|
// GetRuntimeTaskActivity returns hourly task activity distribution for a runtime.
|
|
func (h *Handler) GetRuntimeTaskActivity(w http.ResponseWriter, r *http.Request) {
|
|
runtimeID := chi.URLParam(r, "runtimeId")
|
|
runtimeUUID, ok := parseUUIDOrBadRequest(w, runtimeID, "runtime_id")
|
|
if !ok {
|
|
return
|
|
}
|
|
|
|
rt, err := h.Queries.GetAgentRuntime(r.Context(), runtimeUUID)
|
|
if err != nil {
|
|
writeError(w, http.StatusNotFound, "runtime not found")
|
|
return
|
|
}
|
|
|
|
if _, ok := h.requireWorkspaceMember(w, r, uuidToString(rt.WorkspaceID), "runtime not found"); !ok {
|
|
return
|
|
}
|
|
|
|
rows, err := h.Queries.GetRuntimeTaskHourlyActivity(r.Context(), rt.ID)
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to get task activity")
|
|
return
|
|
}
|
|
|
|
type HourlyActivity struct {
|
|
Hour int `json:"hour"`
|
|
Count int `json:"count"`
|
|
}
|
|
|
|
resp := make([]HourlyActivity, len(rows))
|
|
for i, row := range rows {
|
|
resp[i] = HourlyActivity{Hour: int(row.Hour), Count: int(row.Count)}
|
|
}
|
|
|
|
writeJSON(w, http.StatusOK, resp)
|
|
}
|
|
|
|
// RuntimeUsageByAgentResponse is one (agent, model) row of "Cost by agent".
|
|
// Model stays on the wire because cost is computed client-side from a model
|
|
// pricing table, intentionally not stored server-side so pricing changes
|
|
// don't require a back-fill. The client groups by agent_id and sums.
|
|
type RuntimeUsageByAgentResponse struct {
|
|
AgentID string `json:"agent_id"`
|
|
Model string `json:"model"`
|
|
InputTokens int64 `json:"input_tokens"`
|
|
OutputTokens int64 `json:"output_tokens"`
|
|
CacheReadTokens int64 `json:"cache_read_tokens"`
|
|
CacheWriteTokens int64 `json:"cache_write_tokens"`
|
|
TaskCount int32 `json:"task_count"`
|
|
}
|
|
|
|
// GetRuntimeUsageByAgent returns per-agent token aggregates for a runtime
|
|
// since the cutoff window. Drives the runtime-detail "Cost by agent" tab.
|
|
func (h *Handler) GetRuntimeUsageByAgent(w http.ResponseWriter, r *http.Request) {
|
|
runtimeID := chi.URLParam(r, "runtimeId")
|
|
|
|
rt, err := h.Queries.GetAgentRuntime(r.Context(), parseUUID(runtimeID))
|
|
if err != nil {
|
|
writeError(w, http.StatusNotFound, "runtime not found")
|
|
return
|
|
}
|
|
|
|
if _, ok := h.requireWorkspaceMember(w, r, uuidToString(rt.WorkspaceID), "runtime not found"); !ok {
|
|
return
|
|
}
|
|
|
|
since := parseSinceParam(r, 30)
|
|
|
|
rows, err := h.Queries.ListRuntimeUsageByAgent(r.Context(), db.ListRuntimeUsageByAgentParams{
|
|
RuntimeID: parseUUID(runtimeID),
|
|
Since: since,
|
|
})
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to list usage by agent")
|
|
return
|
|
}
|
|
|
|
resp := make([]RuntimeUsageByAgentResponse, len(rows))
|
|
for i, row := range rows {
|
|
resp[i] = RuntimeUsageByAgentResponse{
|
|
AgentID: uuidToString(row.AgentID),
|
|
Model: row.Model,
|
|
InputTokens: row.InputTokens,
|
|
OutputTokens: row.OutputTokens,
|
|
CacheReadTokens: row.CacheReadTokens,
|
|
CacheWriteTokens: row.CacheWriteTokens,
|
|
TaskCount: row.TaskCount,
|
|
}
|
|
}
|
|
|
|
writeJSON(w, http.StatusOK, resp)
|
|
}
|
|
|
|
// RuntimeUsageByHourResponse is one (hour, model) row. Hours with zero
|
|
// activity are omitted by the SQL — clients fill the gap to render a
|
|
// continuous 0..23 axis. Model is preserved for client-side cost math.
|
|
type RuntimeUsageByHourResponse struct {
|
|
Hour int `json:"hour"`
|
|
Model string `json:"model"`
|
|
InputTokens int64 `json:"input_tokens"`
|
|
OutputTokens int64 `json:"output_tokens"`
|
|
CacheReadTokens int64 `json:"cache_read_tokens"`
|
|
CacheWriteTokens int64 `json:"cache_write_tokens"`
|
|
TaskCount int32 `json:"task_count"`
|
|
}
|
|
|
|
// GetRuntimeUsageByHour returns hourly (0..23) token aggregates for a
|
|
// runtime since the cutoff window. Drives the "By hour" tab.
|
|
func (h *Handler) GetRuntimeUsageByHour(w http.ResponseWriter, r *http.Request) {
|
|
runtimeID := chi.URLParam(r, "runtimeId")
|
|
|
|
rt, err := h.Queries.GetAgentRuntime(r.Context(), parseUUID(runtimeID))
|
|
if err != nil {
|
|
writeError(w, http.StatusNotFound, "runtime not found")
|
|
return
|
|
}
|
|
|
|
if _, ok := h.requireWorkspaceMember(w, r, uuidToString(rt.WorkspaceID), "runtime not found"); !ok {
|
|
return
|
|
}
|
|
|
|
since := parseSinceParam(r, 30)
|
|
|
|
rows, err := h.Queries.GetRuntimeUsageByHour(r.Context(), db.GetRuntimeUsageByHourParams{
|
|
RuntimeID: parseUUID(runtimeID),
|
|
Since: since,
|
|
})
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to get usage by hour")
|
|
return
|
|
}
|
|
|
|
resp := make([]RuntimeUsageByHourResponse, len(rows))
|
|
for i, row := range rows {
|
|
resp[i] = RuntimeUsageByHourResponse{
|
|
Hour: int(row.Hour),
|
|
Model: row.Model,
|
|
InputTokens: row.InputTokens,
|
|
OutputTokens: row.OutputTokens,
|
|
CacheReadTokens: row.CacheReadTokens,
|
|
CacheWriteTokens: row.CacheWriteTokens,
|
|
TaskCount: row.TaskCount,
|
|
}
|
|
}
|
|
|
|
writeJSON(w, http.StatusOK, resp)
|
|
}
|
|
|
|
// GetWorkspaceUsageByDay returns daily token usage aggregated by model for the workspace.
|
|
func (h *Handler) GetWorkspaceUsageByDay(w http.ResponseWriter, r *http.Request) {
|
|
workspaceID := h.resolveWorkspaceID(r)
|
|
since := parseSinceParam(r, 30)
|
|
|
|
rows, err := h.Queries.GetWorkspaceUsageByDay(r.Context(), db.GetWorkspaceUsageByDayParams{
|
|
WorkspaceID: parseUUID(workspaceID),
|
|
Since: since,
|
|
})
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to get usage")
|
|
return
|
|
}
|
|
|
|
type DailyUsageRow struct {
|
|
Date string `json:"date"`
|
|
Model string `json:"model"`
|
|
TotalInputTokens int64 `json:"total_input_tokens"`
|
|
TotalOutputTokens int64 `json:"total_output_tokens"`
|
|
TotalCacheReadTokens int64 `json:"total_cache_read_tokens"`
|
|
TotalCacheWriteTokens int64 `json:"total_cache_write_tokens"`
|
|
TaskCount int32 `json:"task_count"`
|
|
}
|
|
|
|
resp := make([]DailyUsageRow, len(rows))
|
|
for i, row := range rows {
|
|
resp[i] = DailyUsageRow{
|
|
Date: row.Date.Time.Format("2006-01-02"),
|
|
Model: row.Model,
|
|
TotalInputTokens: row.TotalInputTokens,
|
|
TotalOutputTokens: row.TotalOutputTokens,
|
|
TotalCacheReadTokens: row.TotalCacheReadTokens,
|
|
TotalCacheWriteTokens: row.TotalCacheWriteTokens,
|
|
TaskCount: row.TaskCount,
|
|
}
|
|
}
|
|
|
|
writeJSON(w, http.StatusOK, resp)
|
|
}
|
|
|
|
// GetWorkspaceUsageSummary returns total token usage aggregated by model for the workspace.
|
|
func (h *Handler) GetWorkspaceUsageSummary(w http.ResponseWriter, r *http.Request) {
|
|
workspaceID := h.resolveWorkspaceID(r)
|
|
since := parseSinceParam(r, 30)
|
|
|
|
rows, err := h.Queries.GetWorkspaceUsageSummary(r.Context(), db.GetWorkspaceUsageSummaryParams{
|
|
WorkspaceID: parseUUID(workspaceID),
|
|
Since: since,
|
|
})
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to get usage summary")
|
|
return
|
|
}
|
|
|
|
type UsageSummaryRow struct {
|
|
Model string `json:"model"`
|
|
TotalInputTokens int64 `json:"total_input_tokens"`
|
|
TotalOutputTokens int64 `json:"total_output_tokens"`
|
|
TotalCacheReadTokens int64 `json:"total_cache_read_tokens"`
|
|
TotalCacheWriteTokens int64 `json:"total_cache_write_tokens"`
|
|
TaskCount int32 `json:"task_count"`
|
|
}
|
|
|
|
resp := make([]UsageSummaryRow, len(rows))
|
|
for i, row := range rows {
|
|
resp[i] = UsageSummaryRow{
|
|
Model: row.Model,
|
|
TotalInputTokens: row.TotalInputTokens,
|
|
TotalOutputTokens: row.TotalOutputTokens,
|
|
TotalCacheReadTokens: row.TotalCacheReadTokens,
|
|
TotalCacheWriteTokens: row.TotalCacheWriteTokens,
|
|
TaskCount: row.TaskCount,
|
|
}
|
|
}
|
|
|
|
writeJSON(w, http.StatusOK, resp)
|
|
}
|
|
|
|
// parseSinceParam parses the "days" query parameter and returns a timestamptz.
|
|
func parseSinceParam(r *http.Request, defaultDays int) pgtype.Timestamptz {
|
|
days := defaultDays
|
|
if d := r.URL.Query().Get("days"); d != "" {
|
|
if parsed, err := strconv.Atoi(d); err == nil && parsed > 0 && parsed <= 365 {
|
|
days = parsed
|
|
}
|
|
}
|
|
t := time.Now().AddDate(0, 0, -days)
|
|
return pgtype.Timestamptz{Time: t, Valid: true}
|
|
}
|
|
|
|
func (h *Handler) ListAgentRuntimes(w http.ResponseWriter, r *http.Request) {
|
|
workspaceID := h.resolveWorkspaceID(r)
|
|
|
|
var runtimes []db.AgentRuntime
|
|
var err error
|
|
|
|
if ownerFilter := r.URL.Query().Get("owner"); ownerFilter == "me" {
|
|
userID, ok := requireUserID(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
runtimes, err = h.Queries.ListAgentRuntimesByOwner(r.Context(), db.ListAgentRuntimesByOwnerParams{
|
|
WorkspaceID: parseUUID(workspaceID),
|
|
OwnerID: parseUUID(userID),
|
|
})
|
|
} else {
|
|
runtimes, err = h.Queries.ListAgentRuntimes(r.Context(), parseUUID(workspaceID))
|
|
}
|
|
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to list runtimes")
|
|
return
|
|
}
|
|
|
|
resp := make([]AgentRuntimeResponse, len(runtimes))
|
|
for i, rt := range runtimes {
|
|
resp[i] = runtimeToResponse(rt)
|
|
}
|
|
|
|
writeJSON(w, http.StatusOK, resp)
|
|
}
|
|
|
|
// DeleteAgentRuntime deletes a runtime after permission and dependency checks.
|
|
func (h *Handler) DeleteAgentRuntime(w http.ResponseWriter, r *http.Request) {
|
|
runtimeID := chi.URLParam(r, "runtimeId")
|
|
runtimeUUID, ok := parseUUIDOrBadRequest(w, runtimeID, "runtime_id")
|
|
if !ok {
|
|
return
|
|
}
|
|
|
|
rt, err := h.Queries.GetAgentRuntime(r.Context(), runtimeUUID)
|
|
if err != nil {
|
|
writeError(w, http.StatusNotFound, "runtime not found")
|
|
return
|
|
}
|
|
|
|
wsID := uuidToString(rt.WorkspaceID)
|
|
member, ok := h.requireWorkspaceMember(w, r, wsID, "runtime not found")
|
|
if !ok {
|
|
return
|
|
}
|
|
|
|
// Permission: owner/admin can delete any runtime; members can only delete their own.
|
|
userID := uuidToString(member.UserID)
|
|
isAdmin := roleAllowed(member.Role, "owner", "admin")
|
|
isOwner := rt.OwnerID.Valid && uuidToString(rt.OwnerID) == userID
|
|
if !isAdmin && !isOwner {
|
|
writeError(w, http.StatusForbidden, "you can only delete your own runtimes")
|
|
return
|
|
}
|
|
|
|
// Check if any active (non-archived) agents are bound to this runtime.
|
|
activeCount, err := h.Queries.CountActiveAgentsByRuntime(r.Context(), rt.ID)
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to check runtime dependencies")
|
|
return
|
|
}
|
|
if activeCount > 0 {
|
|
writeError(w, http.StatusConflict, "cannot delete runtime: it has active agents bound to it. Archive or reassign the agents first.")
|
|
return
|
|
}
|
|
|
|
// Remove archived agents so the FK constraint (ON DELETE RESTRICT) won't block deletion.
|
|
if err := h.Queries.DeleteArchivedAgentsByRuntime(r.Context(), rt.ID); err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to clean up archived agents")
|
|
return
|
|
}
|
|
|
|
if err := h.Queries.DeleteAgentRuntime(r.Context(), rt.ID); err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to delete runtime")
|
|
return
|
|
}
|
|
|
|
slog.Info("runtime deleted", "runtime_id", uuidToString(rt.ID), "deleted_by", userID)
|
|
|
|
// Notify frontend to refresh runtime list.
|
|
h.publish(protocol.EventDaemonRegister, wsID, "member", userID, map[string]any{
|
|
"action": "delete",
|
|
})
|
|
|
|
writeJSON(w, http.StatusOK, map[string]string{"status": "ok"})
|
|
}
|