mirror of
https://github.com/multica-ai/multica.git
synced 2026-06-17 03:38:32 +02:00
* feat: per-runtime timezone for token usage aggregation The runtime token-usage charts (daily and hourly tabs on the runtime-detail page) bucketed every event by the Postgres session timezone, which is UTC in production. For an operator in UTC+8 that meant a Tuesday afternoon's tasks landed in Tuesday early-morning's bar — the chart was always one off. Fix: store an IANA timezone on agent_runtime and aggregate under it. * migrations 081 / 082 add agent_runtime.timezone (TEXT NOT NULL DEFAULT 'UTC') and rebuild the rollup pipeline (window function and both trigger functions) to compute bucket_date with AT TIME ZONE rt.timezone instead of bare DATE(). * No historical backfill — task_usage_daily rows already on disk keep their UTC bucket_date; only future writes / re-touches recompute under the new tz. (Product call from MUL-1950: 'guarantee future correctness'.) * runtime_usage.sql gains a @tz parameter on ListRuntimeUsage and GetRuntimeUsageByHour and threads tz through GetRuntimeTaskHourly Activity. ListRuntimeUsageDaily reads bucket_date as-is since the rollup already wrote it in tz. * parseSinceParamInTZ replaces the raw N×24h cutoff with start-of- day-N in the runtime's tz so 'last 7 days' lines up with bucket boundaries. * Daemon registration sends the host's IANA tz (TZ env, then time.Local), and UpsertAgentRuntime preserves any user override via a CASE-on-existing-value pattern so a daemon reconnect can't silently revert the operator's setting. * New PATCH /api/runtimes/:id endpoint (UpdateAgentRuntime) lets the runtime detail page edit the tz; the editor seeds with the browser tz on first interaction. Refs: MUL-1950 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Co-authored-by: multica-agent <github@multica.ai> * fix: harden runtime timezone rollups Co-authored-by: multica-agent <github@multica.ai> * fix: address runtime timezone review nits 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: Eve <eve@multica-ai.local>
340 lines
11 KiB
Go
340 lines
11 KiB
Go
// Code generated by sqlc. DO NOT EDIT.
|
|
// versions:
|
|
// sqlc v1.30.0
|
|
// source: runtime_usage.sql
|
|
|
|
package db
|
|
|
|
import (
|
|
"context"
|
|
|
|
"github.com/jackc/pgx/v5/pgtype"
|
|
)
|
|
|
|
const getRuntimeTaskHourlyActivity = `-- name: GetRuntimeTaskHourlyActivity :many
|
|
SELECT EXTRACT(HOUR FROM started_at AT TIME ZONE $2::text)::int AS hour,
|
|
COUNT(*)::int AS count
|
|
FROM agent_task_queue
|
|
WHERE runtime_id = $1 AND started_at IS NOT NULL
|
|
GROUP BY hour
|
|
ORDER BY hour
|
|
`
|
|
|
|
type GetRuntimeTaskHourlyActivityParams struct {
|
|
RuntimeID pgtype.UUID `json:"runtime_id"`
|
|
Tz string `json:"tz"`
|
|
}
|
|
|
|
type GetRuntimeTaskHourlyActivityRow struct {
|
|
Hour int32 `json:"hour"`
|
|
Count int32 `json:"count"`
|
|
}
|
|
|
|
// Hour-of-day distribution for queue starts. Bucketed in the runtime's
|
|
// local tz so "this runtime is busy in the afternoon" actually means
|
|
// the operator's afternoon, not UTC's.
|
|
func (q *Queries) GetRuntimeTaskHourlyActivity(ctx context.Context, arg GetRuntimeTaskHourlyActivityParams) ([]GetRuntimeTaskHourlyActivityRow, error) {
|
|
rows, err := q.db.Query(ctx, getRuntimeTaskHourlyActivity, arg.RuntimeID, arg.Tz)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []GetRuntimeTaskHourlyActivityRow{}
|
|
for rows.Next() {
|
|
var i GetRuntimeTaskHourlyActivityRow
|
|
if err := rows.Scan(&i.Hour, &i.Count); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, i)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
const getRuntimeUsageByHour = `-- name: GetRuntimeUsageByHour :many
|
|
SELECT
|
|
EXTRACT(HOUR FROM tu.created_at AT TIME ZONE $2::text)::int AS hour,
|
|
tu.model,
|
|
SUM(tu.input_tokens)::bigint AS input_tokens,
|
|
SUM(tu.output_tokens)::bigint AS output_tokens,
|
|
SUM(tu.cache_read_tokens)::bigint AS cache_read_tokens,
|
|
SUM(tu.cache_write_tokens)::bigint AS cache_write_tokens,
|
|
COUNT(DISTINCT tu.task_id)::int AS task_count
|
|
FROM task_usage tu
|
|
JOIN agent_task_queue atq ON atq.id = tu.task_id
|
|
WHERE atq.runtime_id = $1
|
|
AND tu.created_at >= $3::timestamptz
|
|
GROUP BY EXTRACT(HOUR FROM tu.created_at AT TIME ZONE $2::text), tu.model
|
|
ORDER BY hour, tu.model
|
|
`
|
|
|
|
type GetRuntimeUsageByHourParams struct {
|
|
RuntimeID pgtype.UUID `json:"runtime_id"`
|
|
Tz string `json:"tz"`
|
|
Since pgtype.Timestamptz `json:"since"`
|
|
}
|
|
|
|
type GetRuntimeUsageByHourRow struct {
|
|
Hour int32 `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"`
|
|
}
|
|
|
|
// Per-(hour, model) token aggregates (hour ∈ 0..23) for a runtime since a
|
|
// cutoff. Powers the "By hour" tab — shows when in the day this runtime is
|
|
// doing real work, with model preserved for client-side cost calculation
|
|
// (same reason as ListRuntimeUsageByAgent above). Hours with zero activity
|
|
// are omitted; the client fills the 24-bucket axis.
|
|
//
|
|
// Hours are extracted in the runtime's local tz via @tz so afternoon
|
|
// work bucketed at UTC 06:00 lands in 14:00 for a UTC+8 runtime.
|
|
func (q *Queries) GetRuntimeUsageByHour(ctx context.Context, arg GetRuntimeUsageByHourParams) ([]GetRuntimeUsageByHourRow, error) {
|
|
rows, err := q.db.Query(ctx, getRuntimeUsageByHour, arg.RuntimeID, arg.Tz, arg.Since)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []GetRuntimeUsageByHourRow{}
|
|
for rows.Next() {
|
|
var i GetRuntimeUsageByHourRow
|
|
if err := rows.Scan(
|
|
&i.Hour,
|
|
&i.Model,
|
|
&i.InputTokens,
|
|
&i.OutputTokens,
|
|
&i.CacheReadTokens,
|
|
&i.CacheWriteTokens,
|
|
&i.TaskCount,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, i)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
const listRuntimeUsage = `-- name: ListRuntimeUsage :many
|
|
SELECT
|
|
DATE(tu.created_at AT TIME ZONE $2::text) AS date,
|
|
tu.provider,
|
|
tu.model,
|
|
SUM(tu.input_tokens)::bigint AS input_tokens,
|
|
SUM(tu.output_tokens)::bigint AS output_tokens,
|
|
SUM(tu.cache_read_tokens)::bigint AS cache_read_tokens,
|
|
SUM(tu.cache_write_tokens)::bigint AS cache_write_tokens
|
|
FROM task_usage tu
|
|
JOIN agent_task_queue atq ON atq.id = tu.task_id
|
|
WHERE atq.runtime_id = $1
|
|
AND tu.created_at >= $3::timestamptz
|
|
GROUP BY DATE(tu.created_at AT TIME ZONE $2::text), tu.provider, tu.model
|
|
ORDER BY DATE(tu.created_at AT TIME ZONE $2::text) DESC, tu.provider, tu.model
|
|
`
|
|
|
|
type ListRuntimeUsageParams struct {
|
|
RuntimeID pgtype.UUID `json:"runtime_id"`
|
|
Tz string `json:"tz"`
|
|
Since pgtype.Timestamptz `json:"since"`
|
|
}
|
|
|
|
type ListRuntimeUsageRow struct {
|
|
Date pgtype.Date `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"`
|
|
}
|
|
|
|
// Reads from raw `task_usage`, bucketed by the runtime's local calendar
|
|
// date via @tz (IANA name, e.g. 'Asia/Shanghai'). The Go layer resolves
|
|
// @tz from agent_runtime.timezone and computes @since as start-of-day-N
|
|
// already in that zone, so the cutoff can stay as a plain timestamptz.
|
|
// This is the always-correct fallback path; used when
|
|
// USAGE_DAILY_ROLLUP_ENABLED is false (or the rollup hasn't been
|
|
// deployed yet).
|
|
func (q *Queries) ListRuntimeUsage(ctx context.Context, arg ListRuntimeUsageParams) ([]ListRuntimeUsageRow, error) {
|
|
rows, err := q.db.Query(ctx, listRuntimeUsage, arg.RuntimeID, arg.Tz, arg.Since)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []ListRuntimeUsageRow{}
|
|
for rows.Next() {
|
|
var i ListRuntimeUsageRow
|
|
if err := rows.Scan(
|
|
&i.Date,
|
|
&i.Provider,
|
|
&i.Model,
|
|
&i.InputTokens,
|
|
&i.OutputTokens,
|
|
&i.CacheReadTokens,
|
|
&i.CacheWriteTokens,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, i)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
const listRuntimeUsageByAgent = `-- name: ListRuntimeUsageByAgent :many
|
|
SELECT
|
|
atq.agent_id,
|
|
tu.model,
|
|
SUM(tu.input_tokens)::bigint AS input_tokens,
|
|
SUM(tu.output_tokens)::bigint AS output_tokens,
|
|
SUM(tu.cache_read_tokens)::bigint AS cache_read_tokens,
|
|
SUM(tu.cache_write_tokens)::bigint AS cache_write_tokens,
|
|
COUNT(DISTINCT tu.task_id)::int AS task_count
|
|
FROM task_usage tu
|
|
JOIN agent_task_queue atq ON atq.id = tu.task_id
|
|
WHERE atq.runtime_id = $1
|
|
AND tu.created_at >= $2::timestamptz
|
|
GROUP BY atq.agent_id, tu.model
|
|
ORDER BY atq.agent_id, tu.model
|
|
`
|
|
|
|
type ListRuntimeUsageByAgentParams struct {
|
|
RuntimeID pgtype.UUID `json:"runtime_id"`
|
|
Since pgtype.Timestamptz `json:"since"`
|
|
}
|
|
|
|
type ListRuntimeUsageByAgentRow struct {
|
|
AgentID pgtype.UUID `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"`
|
|
}
|
|
|
|
// Per-(agent, model) token aggregates for a runtime since a cutoff. Powers
|
|
// the runtime-detail "Cost by agent" tab. task_usage only carries task_id,
|
|
// so we join the queue to expose agent_id. The model dimension is kept on
|
|
// purpose: cost is computed client-side from a per-model pricing table, so
|
|
// collapsing models server-side would erase the information needed to do
|
|
// that arithmetic. The client groups by agent_id and sums cost per agent.
|
|
//
|
|
// This view doesn't bucket by date, so it doesn't need @tz; only the
|
|
// @since cutoff is provided in runtime-local terms (computed in Go).
|
|
func (q *Queries) ListRuntimeUsageByAgent(ctx context.Context, arg ListRuntimeUsageByAgentParams) ([]ListRuntimeUsageByAgentRow, error) {
|
|
rows, err := q.db.Query(ctx, listRuntimeUsageByAgent, arg.RuntimeID, arg.Since)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []ListRuntimeUsageByAgentRow{}
|
|
for rows.Next() {
|
|
var i ListRuntimeUsageByAgentRow
|
|
if err := rows.Scan(
|
|
&i.AgentID,
|
|
&i.Model,
|
|
&i.InputTokens,
|
|
&i.OutputTokens,
|
|
&i.CacheReadTokens,
|
|
&i.CacheWriteTokens,
|
|
&i.TaskCount,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, i)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
const listRuntimeUsageDaily = `-- name: ListRuntimeUsageDaily :many
|
|
SELECT
|
|
bucket_date AS date,
|
|
provider,
|
|
model,
|
|
SUM(input_tokens)::bigint AS input_tokens,
|
|
SUM(output_tokens)::bigint AS output_tokens,
|
|
SUM(cache_read_tokens)::bigint AS cache_read_tokens,
|
|
SUM(cache_write_tokens)::bigint AS cache_write_tokens
|
|
FROM task_usage_daily
|
|
WHERE runtime_id = $1
|
|
AND bucket_date >= (($3::timestamptz AT TIME ZONE $2::text)::date)
|
|
GROUP BY bucket_date, provider, model
|
|
ORDER BY bucket_date DESC, provider, model
|
|
`
|
|
|
|
type ListRuntimeUsageDailyParams struct {
|
|
RuntimeID pgtype.UUID `json:"runtime_id"`
|
|
Tz string `json:"tz"`
|
|
Since pgtype.Timestamptz `json:"since"`
|
|
}
|
|
|
|
type ListRuntimeUsageDailyRow struct {
|
|
Date pgtype.Date `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"`
|
|
}
|
|
|
|
// Reads from the `task_usage_daily` rollup table maintained by
|
|
// rollup_task_usage_daily() (scheduled every 5 min via pg_cron, or any
|
|
// equivalent external scheduler that calls the function). Same shape as
|
|
// ListRuntimeUsage above. Today's bucket may lag the raw table by up to
|
|
// ~10 min (5 min cron period + 5 min rollup safety lag); intentional.
|
|
//
|
|
// Only used when USAGE_DAILY_ROLLUP_ENABLED is true AND deploy has
|
|
// verified that the rollup is fresh (see task_usage_rollup_lag_seconds
|
|
// helper from migration 076).
|
|
//
|
|
// bucket_date is already materialized in the runtime's tz (migration
|
|
// 082). The cutoff still needs @tz because DATE(timestamptz) would cast in
|
|
// the Postgres session timezone; positive-offset runtimes would otherwise
|
|
// include one extra UTC day.
|
|
//
|
|
// The PK on task_usage_daily already collapses to one row per
|
|
// (bucket_date, runtime_id, provider, model), but SUM/GROUP BY is kept
|
|
// so future schema changes (extra dimensions promoted into the table)
|
|
// don't silently change query semantics.
|
|
func (q *Queries) ListRuntimeUsageDaily(ctx context.Context, arg ListRuntimeUsageDailyParams) ([]ListRuntimeUsageDailyRow, error) {
|
|
rows, err := q.db.Query(ctx, listRuntimeUsageDaily, arg.RuntimeID, arg.Tz, arg.Since)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []ListRuntimeUsageDailyRow{}
|
|
for rows.Next() {
|
|
var i ListRuntimeUsageDailyRow
|
|
if err := rows.Scan(
|
|
&i.Date,
|
|
&i.Provider,
|
|
&i.Model,
|
|
&i.InputTokens,
|
|
&i.OutputTokens,
|
|
&i.CacheReadTokens,
|
|
&i.CacheWriteTokens,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, i)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return items, nil
|
|
}
|