zzos/OpenViking
1
0
Fork 0
mirror of https://github.com/volcengine/OpenViking.git synced 2026-09-30 17:28:07 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(observability): persist usage/audit in UTC and bucket per request tz (#2190)

Browse Source 
* fix(observability): persist usage/audit in UTC and bucket per request tz

Dashboard, token-trend, and context-commit-heatmap previously bucketed by
server-process local timezone (`server/config.py` default `local`, which
on Railway / Docker without `TZ` is UTC). UTC+8 viewers saw "today" and
4h heatmap edges shifted relative to their wall clock.

Root cause: `projection.py` applied `astimezone(self._tz)` before writing
the rollup PKs, so date/hour columns were already locked to the
container's tz; a read-side `?timezone=` could not recover the buckets
because the raw UTC instants were no longer in storage.

Fix: persist all time-keyed columns (`date_utc`, `hour_utc`,
`created_at`) in UTC and accept `?timezone=` per request on
`/api/v1/console/{dashboard/summary,tokens,context-commits}`. The viewer
tz is resolved with `zoneinfo`, then the SQLite reads pull the spanning
UTC window and rebucket in Python to user-local days / 4h buckets. DST
is handled by `ZoneInfo` automatically.

Schema bump (v1 -> v2): `usage_token_daily` becomes `usage_token_hourly`
and `usage_retrieval_daily` becomes `usage_retrieval_hourly` (extra
`hour_utc` column in PK so cross-tz "today" slicing is precise). Other
tables keep their shape; `date` is renamed to `date_utc` to make the
semantic shift unambiguous. On boot, the store DROPs the legacy tables
once and recreates the new layout - acceptable because retention is 14
days and the product is pre-GA.

Frontend (`web-studio`) now passes
`Intl.DateTimeFormat().resolvedOptions().timeZone` on every console
fetch; `getLastDaysRange` derives the date range in the viewer's tz so
its boundaries match the backend interpretation. No visual / component
changes were needed.

Tests:
- Backend: existing 15 cases ported to the new keyword-only `tz=` API;
  three new cases exercise UTC+8 day-boundary, America/New_York day
  rebucketing across UTC midnight, and Asia/Shanghai 4h heatmap hour
  shift. All 20 observability tests pass.
- Frontend: typecheck + lint clean (no new errors).

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

* fix(observability): reset legacy usage audit sqlite schema

---------

Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
Zayn Jarvis
2026-05-22 17:03:14 +08:00
committed by GitHub
co-authored by Claude Opus 4.7
parent 96431c17b4
commit 0d63f0e3c4
16 changed files with 988 additions and 200 deletions
Download Patch File Download Diff File Expand all files Collapse all files
openviking/observability/usage_audit/README.md
+4 -3
View File
@@ -83,7 +83,7 @@ Observability Event Bus
| --- | --- | --- |
| `enabled` | `true` | 是否启用 Usage/Audit |
| `backend` | `"sqlite"` | 当前仅支持 SQLite |
| `sqlite_path` | `null` | SQLite 文件路径；为空时使用 OpenViking workspace 下的 `_system/usage_audit/usage_audit.sqlite3` |
| `sqlite_path` | `null` | SQLite 文件路径；为空时使用当前 OpenViking workspace 下的 `_system/usage_audit/usage_audit.sqlite3` |
| `queue_size` | `10000` | 后台写入队列大小 |
| `batch_size` | `500` | 单次批量写入的最大事件数 |
| `flush_interval_seconds` | `1.0` | worker 定时 flush 间隔 |
@@ -91,7 +91,7 @@ Observability Event Bus
| `usage_retention_days` | `14` | 统计聚合数据保留天数，包含 Token、检索、上下文写入热力图、Agent 活跃；`0` 表示不按天裁剪 |
| `audit_retention_days` | `7` | 请求审计日志保留天数；`0` 表示不按天裁剪 |
| `audit_retention_per_account` | `1000` | 每个 account 保留的最新请求审计条数；`0` 表示不按条数裁剪 |
| `timezone` | `"local"` | 统计日期使用的时区；可填 `"local"`、`"UTC"` 或 IANA 时区名 |
| `timezone` | `"local"` | Console 请求未传 `timezone` 时的兜底查询时区；写入始终按 UTC 保存。`"local"` 表示 server 进程所在机器/容器的本地时区 |
| `inventory_ttl_seconds` | `10.0` | 上下文当前数据量查询缓存时间 |
本地版使用 SQLite 没问题。分布式生产环境如果多实例同时提供 Console，建议后续增加共享
@@ -391,7 +391,8 @@ curl "http://127.0.0.1:1933/api/v1/console/dashboard/summary" \
- VLM 需要产生 `vlm.call`
- Embedding 需要产生 `embedding.call`
统计日期使用 `server.observability.usage_audit.timezone`。如果时区配置不同，数据可能落在另一天。
统计数据写入时按 UTC 保存；Console 查询会优先使用请求里的 `timezone` 参数做读端分桶。
如果请求没有传 `timezone`，才会使用 `server.observability.usage_audit.timezone` 作为兜底。
### 为什么请求日志里没有 Console 自己的请求？
openviking/observability/usage_audit/api_service.py
+44 -20
View File
@@ -1,6 +1,11 @@
# Copyright (c) 2026 Beijing Volcano Engine Technology Co., Ltd.
# SPDX-License-Identifier: AGPL-3.0
"""Read service for product Usage/Audit APIs."""
"""Read service for product Usage/Audit APIs.
The store always persists UTC. The viewer's timezone is resolved per request
from the `?timezone=` query parameter (IANA name, e.g. `Asia/Shanghai`) and
falls back to the server-default tz from config when absent or invalid.
"""
from __future__ import annotations
@@ -12,7 +17,7 @@ from openviking_cli.exceptions import InvalidArgumentError
from .inventory import ContextInventoryProvider
from .store import UsageAuditStore
from .time import resolve_usage_timezone
from .time import resolve_usage_timezone, resolve_user_timezone
class UsageAuditQueryService:
@@ -27,24 +32,33 @@ class UsageAuditQueryService:
) -> None:
self._store = store
self._inventory = inventory
self._tz = resolve_usage_timezone(timezone_name)
self._default_tz = resolve_usage_timezone(timezone_name)
def today(self) -> str:
"""Return today's date in the configured usage/audit timezone."""
return datetime.now(self._tz).date().isoformat()
def _resolve_tz(self, timezone_name: str | None):
return resolve_user_timezone(timezone_name, fallback=self._default_tz)
async def dashboard_summary(self, ctx: RequestContext) -> dict[str, Any]:
"""Return all Dashboard top-card data."""
today = self.today()
def today(self, timezone_name: str | None = None) -> str:
"""Return today's date in the viewer's timezone (or the default)."""
tz = self._resolve_tz(timezone_name)
return datetime.now(tz).date().isoformat()
async def dashboard_summary(
self, ctx: RequestContext, *, timezone_name: str | None = None
) -> dict[str, Any]:
"""Return all Dashboard top-card data for the viewer's tz."""
tz = self._resolve_tz(timezone_name)
today = datetime.now(tz).date().isoformat()
context_counts = await self._inventory.get_counts(ctx)
today_tokens = await self._store.get_today_tokens(account_id=ctx.account_id, date=today)
today_tokens = await self._store.get_today_tokens(
account_id=ctx.account_id, user_date=today, tz=tz
)
today_retrievals = await self._store.get_today_retrievals(
account_id=ctx.account_id,
date=today,
account_id=ctx.account_id, user_date=today, tz=tz
)
agent_overview = await self._store.get_agent_overview(
account_id=ctx.account_id,
date=today,
user_date=today,
tz=tz,
limit=5,
)
return {
@@ -61,14 +75,17 @@ class UsageAuditQueryService:
start_date: str,
end_date: str,
bucket: str,
timezone_name: str | None = None,
) -> dict[str, Any]:
"""Return token usage series for a date range."""
"""Return token usage series in the viewer's tz."""
tz = self._resolve_tz(timezone_name)
self._validate_date_range(start_date, end_date)
items = await self._store.get_token_series(
account_id=ctx.account_id,
start_date=start_date,
end_date=end_date,
start_user_date=start_date,
end_user_date=end_date,
bucket=bucket,
tz=tz,
)
return {"start_date": start_date, "end_date": end_date, "bucket": bucket, "items": items}
@@ -79,14 +96,17 @@ class UsageAuditQueryService:
start_date: str,
end_date: str,
bucket: str,
timezone_name: str | None = None,
) -> dict[str, Any]:
"""Return context write heatmap rows for a date range."""
"""Return context write heatmap rows in the viewer's tz."""
tz = self._resolve_tz(timezone_name)
self._validate_date_range(start_date, end_date)
items = await self._store.get_context_commit_heatmap(
account_id=ctx.account_id,
start_date=start_date,
end_date=end_date,
start_user_date=start_date,
end_user_date=end_date,
bucket=bucket,
tz=tz,
)
return {"start_date": start_date, "end_date": end_date, "bucket": bucket, "items": items}
@@ -100,7 +120,11 @@ class UsageAuditQueryService:
page: int,
page_size: int,
) -> dict[str, Any]:
"""Return filtered request audit rows."""
"""Return filtered request audit rows.
`created_at` is returned as a UTC ISO string; the client formats it in
the viewer's locale on render.
"""
return await self._store.query_audit_logs(
account_id=ctx.account_id,
request_id=request_id,
openviking/observability/usage_audit/projection.py
+28 -10
View File
@@ -66,10 +66,13 @@ def safe_float(value: Any, default: float = 0.0) -> float:
def project_events(
events: Sequence[ObservabilityEvent],
*,
tz,
) -> UsageAuditProjection:
"""Project generic events into product usage/audit rows."""
"""Project generic events into product usage/audit rows.
All time-keyed columns (`date`, `hour_*`) are written in UTC. The
user-facing timezone is applied at read time so the same store can serve
viewers from any region.
"""
token_rows: defaultdict[tuple, int] = defaultdict(int)
retrieval_rows: defaultdict[tuple, tuple[int, int]] = defaultdict(lambda: (0, 0))
context_rows: defaultdict[tuple, int] = defaultdict(int)
@@ -81,9 +84,10 @@ def project_events(
account_id = normalize_identity(event.account_id, unknown=True)
user_id = normalize_identity(event.user_id)
agent_id = normalize_identity(event.agent_id)
local_dt = _local_dt(event, tz)
event_date = local_dt.date().isoformat()
created_at = local_dt.isoformat()
utc_dt = _utc_dt(event)
event_date = utc_dt.date().isoformat()
event_hour = int(utc_dt.hour)
created_at = utc_dt.isoformat()
payload = event.payload
if event.event_name == "vlm.call":
@@ -93,6 +97,7 @@ def project_events(
user_id=user_id,
agent_id=agent_id,
event_date=event_date,
event_hour=event_hour,
source="vlm",
provider=payload.get("provider"),
model_name=payload.get("model_name"),
@@ -108,6 +113,7 @@ def project_events(
user_id=user_id,
agent_id=agent_id,
event_date=event_date,
event_hour=event_hour,
source="embedding",
provider=payload.get("provider"),
model_name=payload.get("model_name"),
@@ -123,6 +129,7 @@ def project_events(
user_id=user_id,
agent_id=agent_id,
event_date=event_date,
event_hour=event_hour,
source="rerank",
provider=payload.get("provider"),
model_name=payload.get("model_name"),
@@ -135,7 +142,7 @@ def project_events(
_project_http_request(
event,
event_date=event_date,
hour=int(local_dt.hour),
hour=event_hour,
created_at=created_at,
retrieval_rows=retrieval_rows,
context_rows=context_rows,
@@ -154,11 +161,11 @@ def project_events(
)
def _local_dt(event: ObservabilityEvent, tz) -> datetime:
def _utc_dt(event: ObservabilityEvent) -> datetime:
ts = event.timestamp
if ts.tzinfo is None:
ts = ts.replace(tzinfo=timezone.utc)
return ts.astimezone(tz)
return ts.astimezone(timezone.utc)
def _add_token_rows(
@@ -168,6 +175,7 @@ def _add_token_rows(
user_id: str,
agent_id: str,
event_date: str,
event_hour: int,
source: str,
provider: Any,
model_name: Any,
@@ -184,6 +192,7 @@ def _add_token_rows(
user_id,
agent_id,
event_date,
event_hour,
source,
"input",
provider_key,
@@ -196,6 +205,7 @@ def _add_token_rows(
user_id,
agent_id,
event_date,
event_hour,
source,
"output",
provider_key,
@@ -237,7 +247,15 @@ def _project_http_request(
retrieval_operation = retrieval_operation_for_http(method, route)
if retrieval_operation:
key = (audit_account, row_user, row_agent, event_date, retrieval_operation, status)
key = (
audit_account,
row_user,
row_agent,
event_date,
hour,
retrieval_operation,
status,
)
prev_count, prev_results = retrieval_rows[key]
retrieval_rows[key] = (prev_count + 1, prev_results)
openviking/observability/usage_audit/runtime.py
-1
View File
@@ -69,7 +69,6 @@ async def init_usage_audit_from_server_config(
usage_retention_days=usage_config.usage_retention_days,
audit_retention_days=usage_config.audit_retention_days,
audit_retention_per_account=usage_config.audit_retention_per_account,
timezone_name=usage_config.timezone,
)
await store.initialize()
openviking/observability/usage_audit/schema.py
+54 -16
View File
@@ -1,13 +1,31 @@
# Copyright (c) 2026 Beijing Volcano Engine Technology Co., Ltd.
# SPDX-License-Identifier: AGPL-3.0
"""SQLite schema for product usage/audit projections."""
"""SQLite schema for product usage/audit projections.
All time-keyed columns (`date_utc`, `hour_utc`, `created_at`) are persisted in
UTC. The viewer's timezone is applied at read time so a single store can
serve any region. Token and retrieval rollups are hour-grained so cross-tz
"today" queries can slice at user-local day boundaries; agent activity stays
daily, filtered by `last_seen_at` at read time.
"""
# Bump when the table layout changes incompatibly. Stored on the `_schema_meta`
# row so `SQLiteUsageAuditStore.initialize` can reset the local SQLite store
# when an older snapshot is detected.
SCHEMA_VERSION = 3
SQLITE_SCHEMA = """
CREATE TABLE IF NOT EXISTS usage_token_daily (
CREATE TABLE IF NOT EXISTS _schema_meta (
key TEXT PRIMARY KEY,
value TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS usage_token_hourly (
account_id TEXT NOT NULL,
user_id TEXT NOT NULL,
agent_id TEXT NOT NULL,
date TEXT NOT NULL,
date_utc TEXT NOT NULL,
hour_utc INTEGER NOT NULL,
source TEXT NOT NULL,
token_type TEXT NOT NULL,
provider TEXT NOT NULL,
@@ -15,52 +33,58 @@ CREATE TABLE IF NOT EXISTS usage_token_daily (
token_count INTEGER NOT NULL DEFAULT 0,
updated_at TEXT NOT NULL,
PRIMARY KEY (
account_id, user_id, agent_id, date, source, token_type, provider, model_name
account_id, user_id, agent_id, date_utc, hour_utc,
source, token_type, provider, model_name
)
);
CREATE INDEX IF NOT EXISTS idx_usage_token_account_date
ON usage_token_daily(account_id, date);
ON usage_token_hourly(account_id, date_utc, hour_utc);
CREATE TABLE IF NOT EXISTS usage_retrieval_daily (
CREATE TABLE IF NOT EXISTS usage_retrieval_hourly (
account_id TEXT NOT NULL,
user_id TEXT NOT NULL,
agent_id TEXT NOT NULL,
date TEXT NOT NULL,
date_utc TEXT NOT NULL,
hour_utc INTEGER NOT NULL,
operation TEXT NOT NULL,
status TEXT NOT NULL,
request_count INTEGER NOT NULL DEFAULT 0,
result_count INTEGER NOT NULL DEFAULT 0,
updated_at TEXT NOT NULL,
PRIMARY KEY (account_id, user_id, agent_id, date, operation, status)
PRIMARY KEY (
account_id, user_id, agent_id, date_utc, hour_utc, operation, status
)
);
CREATE INDEX IF NOT EXISTS idx_usage_retrieval_account_date
ON usage_retrieval_daily(account_id, date);
ON usage_retrieval_hourly(account_id, date_utc, hour_utc);
CREATE TABLE IF NOT EXISTS usage_context_write_bucket (
account_id TEXT NOT NULL,
user_id TEXT NOT NULL,
agent_id TEXT NOT NULL,
date TEXT NOT NULL,
hour_bucket INTEGER NOT NULL,
date_utc TEXT NOT NULL,
hour_utc INTEGER NOT NULL,
operation TEXT NOT NULL,
count INTEGER NOT NULL DEFAULT 0,
updated_at TEXT NOT NULL,
PRIMARY KEY (account_id, user_id, agent_id, date, hour_bucket, operation)
PRIMARY KEY (
account_id, user_id, agent_id, date_utc, hour_utc, operation
)
);
CREATE INDEX IF NOT EXISTS idx_usage_context_write_account_date
ON usage_context_write_bucket(account_id, date, hour_bucket);
ON usage_context_write_bucket(account_id, date_utc, hour_utc);
CREATE TABLE IF NOT EXISTS usage_agent_activity_daily (
account_id TEXT NOT NULL,
agent_id TEXT NOT NULL,
date TEXT NOT NULL,
date_utc TEXT NOT NULL,
request_count INTEGER NOT NULL DEFAULT 0,
last_seen_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
PRIMARY KEY (account_id, agent_id, date)
PRIMARY KEY (account_id, agent_id, date_utc)
);
CREATE INDEX IF NOT EXISTS idx_usage_agent_activity_account_date
ON usage_agent_activity_daily(account_id, date, last_seen_at);
ON usage_agent_activity_daily(account_id, date_utc, last_seen_at);
CREATE TABLE IF NOT EXISTS request_audit (
id INTEGER PRIMARY KEY AUTOINCREMENT,
@@ -84,3 +108,17 @@ CREATE INDEX IF NOT EXISTS idx_request_audit_account_api
CREATE INDEX IF NOT EXISTS idx_request_audit_account_status
ON request_audit(account_id, status_code);
"""
# The SQLite backend is a local pre-GA store with short retention. When the
# schema changes incompatibly, reset all local usage/audit tables instead of
# carrying partial migrations that can leave date/date_utc or daily/hourly
# shapes mixed in one database.
RESET_ON_SCHEMA_UPGRADE_TABLES = (
"usage_token_daily",
"usage_retrieval_daily",
"usage_token_hourly",
"usage_retrieval_hourly",
"usage_context_write_bucket",
"usage_agent_activity_daily",
"request_audit",
)
openviking/observability/usage_audit/sqlite_store.py
+307 -107
View File
@@ -1,20 +1,26 @@
# Copyright (c) 2026 Beijing Volcano Engine Technology Co., Ltd.
# SPDX-License-Identifier: AGPL-3.0
"""SQLite implementation of the product usage/audit store."""
"""SQLite implementation of the product usage/audit store.
All time-keyed columns are persisted in UTC. Read methods accept a
viewer-supplied `tz` (an `Iana zoneinfo`-style `tzinfo`) and rebucket the
underlying hourly rows into user-local days / hours.
"""
from __future__ import annotations
import asyncio
import sqlite3
from datetime import date, datetime, timedelta, timezone
from datetime import date, datetime, time, timedelta, timezone, tzinfo
from pathlib import Path
from typing import Any, Iterable, Sequence
from openviking.observability.events import ObservabilityEvent
from .projection import UsageAuditProjection, project_events, safe_int
from .schema import SQLITE_SCHEMA
from .time import resolve_usage_timezone
from .schema import RESET_ON_SCHEMA_UPGRADE_TABLES, SCHEMA_VERSION, SQLITE_SCHEMA
UTC = timezone.utc
def _date_range(start_date: str, end_date: str) -> Iterable[str]:
@@ -26,6 +32,13 @@ def _date_range(start_date: str, end_date: str) -> Iterable[str]:
return ((start + timedelta(days=offset)).isoformat() for offset in range(days + 1))
def _user_day_window_utc(day: date, tz: tzinfo) -> tuple[datetime, datetime]:
"""Return the UTC `[start, end)` instants spanning one user-tz day."""
start_local = datetime.combine(day, time.min, tzinfo=tz)
end_local = datetime.combine(day + timedelta(days=1), time.min, tzinfo=tz)
return start_local.astimezone(UTC), end_local.astimezone(UTC)
class SQLiteUsageAuditStore:
"""Async wrapper around a SQLite usage/audit database."""
@@ -36,13 +49,11 @@ class SQLiteUsageAuditStore:
usage_retention_days: int = 14,
audit_retention_days: int = 7,
audit_retention_per_account: int = 1000,
timezone_name: str = "local",
) -> None:
self._db_path = Path(db_path)
self._usage_retention_days = int(usage_retention_days)
self._audit_retention_days = int(audit_retention_days)
self._audit_retention_per_account = int(audit_retention_per_account)
self._tz = resolve_usage_timezone(timezone_name)
self._conn: sqlite3.Connection | None = None
self._lock = asyncio.Lock()
@@ -60,9 +71,30 @@ class SQLiteUsageAuditStore:
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA synchronous=NORMAL")
conn.execute("PRAGMA foreign_keys=ON")
# Reset incompatible pre-v3 local tables before creating the new
# UTC/hourly layout. The SQLite backend has short retention and may live
# inside one workspace, so dropping stale rollups is safer than mixing
# old daily/local columns with the new UTC columns.
self._migrate_legacy_sync(conn)
conn.executescript(SQLITE_SCHEMA)
conn.execute(
"INSERT OR REPLACE INTO _schema_meta (key, value) VALUES (?, ?)",
("version", str(SCHEMA_VERSION)),
)
self._conn = conn
@staticmethod
def _migrate_legacy_sync(conn: sqlite3.Connection) -> None:
try:
row = conn.execute("SELECT value FROM _schema_meta WHERE key = 'version'").fetchone()
except sqlite3.OperationalError:
row = None
current = int(row["value"]) if row and row["value"] else 0
if current >= SCHEMA_VERSION:
return
for table in RESET_ON_SCHEMA_UPGRADE_TABLES:
conn.execute(f"DROP TABLE IF EXISTS {table}")
async def close(self) -> None:
await asyncio.to_thread(self._close_sync)
@@ -74,13 +106,13 @@ class SQLiteUsageAuditStore:
async def record_batch(self, events: Sequence[ObservabilityEvent]) -> None:
if not events:
return
projection = project_events(events, tz=self._tz)
projection = project_events(events)
async with self._lock:
await asyncio.to_thread(self._record_projection_sync, projection)
def _record_projection_sync(self, projection: UsageAuditProjection) -> None:
assert self._conn is not None
updated_at = datetime.now(timezone.utc).isoformat()
updated_at = datetime.now(UTC).isoformat()
conn = self._conn
conn.execute("BEGIN")
try:
@@ -100,13 +132,15 @@ class SQLiteUsageAuditStore:
def _write_token_rows(conn, rows: dict[tuple, int], updated_at: str) -> None:
conn.executemany(
"""
INSERT INTO usage_token_daily (
account_id, user_id, agent_id, date, source, token_type,
provider, model_name, token_count, updated_at
INSERT INTO usage_token_hourly (
account_id, user_id, agent_id, date_utc, hour_utc,
source, token_type, provider, model_name,
token_count, updated_at
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT (
account_id, user_id, agent_id, date, source, token_type, provider, model_name
account_id, user_id, agent_id, date_utc, hour_utc,
source, token_type, provider, model_name
)
DO UPDATE SET
token_count = token_count + excluded.token_count,
@@ -119,12 +153,14 @@ class SQLiteUsageAuditStore:
def _write_retrieval_rows(conn, rows: dict[tuple, tuple[int, int]], updated_at: str) -> None:
conn.executemany(
"""
INSERT INTO usage_retrieval_daily (
account_id, user_id, agent_id, date, operation, status,
request_count, result_count, updated_at
INSERT INTO usage_retrieval_hourly (
account_id, user_id, agent_id, date_utc, hour_utc,
operation, status, request_count, result_count, updated_at
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT (
account_id, user_id, agent_id, date_utc, hour_utc, operation, status
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT (account_id, user_id, agent_id, date, operation, status)
DO UPDATE SET
request_count = request_count + excluded.request_count,
result_count = result_count + excluded.result_count,
@@ -141,10 +177,13 @@ class SQLiteUsageAuditStore:
conn.executemany(
"""
INSERT INTO usage_context_write_bucket (
account_id, user_id, agent_id, date, hour_bucket, operation, count, updated_at
account_id, user_id, agent_id, date_utc, hour_utc,
operation, count, updated_at
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT (account_id, user_id, agent_id, date, hour_bucket, operation)
ON CONFLICT (
account_id, user_id, agent_id, date_utc, hour_utc, operation
)
DO UPDATE SET
count = count + excluded.count,
updated_at = excluded.updated_at
@@ -157,10 +196,10 @@ class SQLiteUsageAuditStore:
conn.executemany(
"""
INSERT INTO usage_agent_activity_daily (
account_id, agent_id, date, request_count, last_seen_at, updated_at
account_id, agent_id, date_utc, request_count, last_seen_at, updated_at
)
VALUES (?, ?, ?, ?, ?, ?)
ON CONFLICT (account_id, agent_id, date)
ON CONFLICT (account_id, agent_id, date_utc)
DO UPDATE SET
request_count = request_count + excluded.request_count,
last_seen_at = MAX(last_seen_at, excluded.last_seen_at),
@@ -220,22 +259,26 @@ class SQLiteUsageAuditStore:
)
for account_id, cutoff_date in cutoff_by_account.items():
for table in (
"usage_token_daily",
"usage_retrieval_daily",
"usage_token_hourly",
"usage_retrieval_hourly",
"usage_context_write_bucket",
"usage_agent_activity_daily",
):
conn.execute(
f"DELETE FROM {table} WHERE account_id = ? AND date < ?",
f"DELETE FROM {table} WHERE account_id = ? AND date_utc < ?",
(account_id, cutoff_date),
)
@staticmethod
def _usage_max_dates(projection: UsageAuditProjection) -> dict[str, str]:
max_dates: dict[str, str] = {}
# token_rows: (account, user, agent, date_utc, hour_utc, source, ...)
SQLiteUsageAuditStore._merge_max_dates(max_dates, projection.token_rows, date_index=3)
# retrieval_rows: (account, user, agent, date_utc, hour_utc, op, status)
SQLiteUsageAuditStore._merge_max_dates(max_dates, projection.retrieval_rows, date_index=3)
# context_rows: (account, user, agent, date_utc, hour_utc, op)
SQLiteUsageAuditStore._merge_max_dates(max_dates, projection.context_rows, date_index=3)
# agent_rows: (account, agent, date_utc)
SQLiteUsageAuditStore._merge_max_dates(max_dates, projection.agent_rows, date_index=2)
return max_dates
@@ -278,82 +321,102 @@ class SQLiteUsageAuditStore:
max_dates[account_id] = str(row["max_date"])
return max_dates
async def get_today_tokens(self, *, account_id: str, date: str) -> dict[str, int]:
async with self._lock:
return await asyncio.to_thread(self._get_today_tokens_sync, account_id, date)
# ------------------------------------------------------------------
# Read API. All `*_user_date` arguments are interpreted in the supplied
# viewer timezone (`tz`). The underlying rows are stored in UTC and
# rebucketed in Python after a windowed SQL query.
# ------------------------------------------------------------------
def _get_today_tokens_sync(self, account_id: str, event_date: str) -> dict[str, int]:
async def get_today_tokens(
self, *, account_id: str, user_date: str, tz: tzinfo
) -> dict[str, int]:
async with self._lock:
return await asyncio.to_thread(self._get_today_tokens_sync, account_id, user_date, tz)
def _get_today_tokens_sync(self, account_id: str, user_date: str, tz: tzinfo) -> dict[str, int]:
assert self._conn is not None
cur = self._conn.execute(
"""
SELECT source, token_type, SUM(token_count) AS total
FROM usage_token_daily
WHERE account_id = ? AND date = ?
GROUP BY source, token_type
""",
(account_id, event_date),
)
day = date.fromisoformat(user_date)
utc_start, utc_end = _user_day_window_utc(day, tz)
rows = self._fetch_hourly_token_rows(account_id, utc_start, utc_end)
result = {"vlm_input": 0, "vlm_output": 0, "embedding_input": 0}
for row in cur.fetchall():
key = f"{row['source']}_{row['token_type']}"
for source, token_type, _, _, total in rows:
key = f"{source}_{token_type}"
if key in result:
result[key] = int(row["total"] or 0)
result[key] += total
result["total"] = sum(result.values())
return result
async def get_today_retrievals(self, *, account_id: str, date: str) -> dict[str, int]:
async def get_today_retrievals(
self, *, account_id: str, user_date: str, tz: tzinfo
) -> dict[str, int]:
async with self._lock:
return await asyncio.to_thread(self._get_today_retrievals_sync, account_id, date)
return await asyncio.to_thread(
self._get_today_retrievals_sync, account_id, user_date, tz
)
def _get_today_retrievals_sync(self, account_id: str, event_date: str) -> dict[str, int]:
def _get_today_retrievals_sync(
self, account_id: str, user_date: str, tz: tzinfo
) -> dict[str, int]:
assert self._conn is not None
cur = self._conn.execute(
"""
SELECT operation, SUM(request_count) AS total
FROM usage_retrieval_daily
WHERE account_id = ? AND date = ? AND status = 'success'
GROUP BY operation
""",
(account_id, event_date),
)
day = date.fromisoformat(user_date)
utc_start, utc_end = _user_day_window_utc(day, tz)
result = {"find": 0, "search": 0}
for row in cur.fetchall():
operation = str(row["operation"])
for operation, total in self._fetch_hourly_retrieval_rows(account_id, utc_start, utc_end):
if operation in result:
result[operation] = int(row["total"] or 0)
result[operation] += total
result["total"] = sum(result.values())
return result
async def get_agent_overview(
self, *, account_id: str, date: str, limit: int = 5
self,
*,
account_id: str,
user_date: str,
tz: tzinfo,
limit: int = 5,
) -> dict[str, Any]:
async with self._lock:
return await asyncio.to_thread(
self._get_agent_overview_sync, account_id, date, int(limit)
self._get_agent_overview_sync,
account_id,
user_date,
tz,
int(limit),
)
def _get_agent_overview_sync(
self, account_id: str, event_date: str, limit: int
self, account_id: str, user_date: str, tz: tzinfo, limit: int
) -> dict[str, Any]:
assert self._conn is not None
day = date.fromisoformat(user_date)
utc_start, utc_end = _user_day_window_utc(day, tz)
utc_start_iso = utc_start.isoformat()
utc_end_iso = utc_end.isoformat()
# last_seen_at is stored as a UTC ISO string by projection.py; compare
# lexicographically — ISO-8601 with fixed UTC offset is sort-safe.
total_cur = self._conn.execute(
"""
SELECT COUNT(DISTINCT agent_id) AS total
FROM usage_agent_activity_daily
WHERE account_id = ? AND date = ?
WHERE account_id = ?
AND last_seen_at >= ?
AND last_seen_at < ?
""",
(account_id, event_date),
(account_id, utc_start_iso, utc_end_iso),
)
total = int(total_cur.fetchone()["total"] or 0)
cur = self._conn.execute(
"""
SELECT agent_id, last_seen_at
SELECT agent_id, MAX(last_seen_at) AS last_seen_at
FROM usage_agent_activity_daily
WHERE account_id = ? AND date = ?
WHERE account_id = ?
AND last_seen_at >= ?
AND last_seen_at < ?
GROUP BY agent_id
ORDER BY last_seen_at DESC
LIMIT ?
""",
(account_id, event_date, limit),
(account_id, utc_start_iso, utc_end_iso, limit),
)
return {
"total": total,
@@ -364,79 +427,211 @@ class SQLiteUsageAuditStore:
}
async def get_token_series(
self, *, account_id: str, start_date: str, end_date: str, bucket: str
self,
*,
account_id: str,
start_user_date: str,
end_user_date: str,
bucket: str,
tz: tzinfo,
) -> list[dict[str, Any]]:
async with self._lock:
return await asyncio.to_thread(
self._get_token_series_sync, account_id, start_date, end_date, bucket
self._get_token_series_sync,
account_id,
start_user_date,
end_user_date,
bucket,
tz,
)
def _get_token_series_sync(
self, account_id: str, start_date: str, end_date: str, bucket: str
self,
account_id: str,
start_user_date: str,
end_user_date: str,
bucket: str,
tz: tzinfo,
) -> list[dict[str, Any]]:
assert self._conn is not None
cur = self._conn.execute(
"""
SELECT date, source, token_type, SUM(token_count) AS total
FROM usage_token_daily
WHERE account_id = ? AND date >= ? AND date <= ?
GROUP BY date, source, token_type
""",
(account_id, start_date, end_date),
)
by_date = {
start_day = date.fromisoformat(start_user_date)
end_day = date.fromisoformat(end_user_date)
utc_start, _ = _user_day_window_utc(start_day, tz)
_, utc_end = _user_day_window_utc(end_day, tz)
by_date: dict[str, dict[str, Any]] = {
d: {"date": d, "vlm_input": 0, "vlm_output": 0, "embedding_input": 0}
for d in _date_range(start_date, end_date)
for d in _date_range(start_user_date, end_user_date)
}
for row in cur.fetchall():
key = f"{row['source']}_{row['token_type']}"
if key in {"vlm_input", "vlm_output", "embedding_input"}:
by_date.setdefault(
row["date"],
{"date": row["date"], "vlm_input": 0, "vlm_output": 0, "embedding_input": 0},
)[key] = int(row["total"] or 0)
return list(by_date.values())
for source, token_type, date_utc, hour_utc, total in self._fetch_hourly_token_rows(
account_id, utc_start, utc_end
):
local_date = (
datetime(
*_parse_ymd(date_utc),
hour_utc,
tzinfo=UTC,
)
.astimezone(tz)
.date()
.isoformat()
)
slot = by_date.setdefault(
local_date,
{"date": local_date, "vlm_input": 0, "vlm_output": 0, "embedding_input": 0},
)
key = f"{source}_{token_type}"
if key in slot:
slot[key] += total
# Drop any local dates outside the requested range (a UTC hour at the
# boundary may rebucket to a neighbour day after tz conversion).
return [by_date[d] for d in _date_range(start_user_date, end_user_date) if d in by_date]
async def get_context_commit_heatmap(
self, *, account_id: str, start_date: str, end_date: str, bucket: str
self,
*,
account_id: str,
start_user_date: str,
end_user_date: str,
bucket: str,
tz: tzinfo,
) -> list[dict[str, Any]]:
async with self._lock:
return await asyncio.to_thread(
self._get_context_commit_heatmap_sync,
account_id,
start_date,
end_date,
start_user_date,
end_user_date,
bucket,
tz,
)
def _get_context_commit_heatmap_sync(
self, account_id: str, start_date: str, end_date: str, bucket: str
self,
account_id: str,
start_user_date: str,
end_user_date: str,
bucket: str,
tz: tzinfo,
) -> list[dict[str, Any]]:
assert self._conn is not None
bucket_size = 4 if bucket == "4h" else 1
cur = self._conn.execute(
"""
SELECT date, hour_bucket, operation, SUM(count) AS total
FROM usage_context_write_bucket
WHERE account_id = ? AND date >= ? AND date <= ?
GROUP BY date, hour_bucket, operation
""",
(account_id, start_date, end_date),
)
start_day = date.fromisoformat(start_user_date)
end_day = date.fromisoformat(end_user_date)
utc_start, _ = _user_day_window_utc(start_day, tz)
_, utc_end = _user_day_window_utc(end_day, tz)
rows: dict[tuple[str, int], dict[str, Any]] = {}
for event_date in _date_range(start_date, end_date):
for event_date in _date_range(start_user_date, end_user_date):
for hour in range(0, 24, bucket_size):
rows[(event_date, hour)] = self._empty_context_row(event_date, hour)
cur = self._conn.execute(
"""
SELECT date_utc, hour_utc, operation, SUM(count) AS total
FROM usage_context_write_bucket
WHERE account_id = ?
AND (
date_utc > ? OR (date_utc = ? AND hour_utc >= ?)
)
AND (
date_utc < ? OR (date_utc = ? AND hour_utc < ?)
)
GROUP BY date_utc, hour_utc, operation
""",
(
account_id,
utc_start.date().isoformat(),
utc_start.date().isoformat(),
utc_start.hour,
utc_end.date().isoformat(),
utc_end.date().isoformat(),
utc_end.hour,
),
)
for row in cur.fetchall():
hour = int(row["hour_bucket"])
normalized_hour = (hour // bucket_size) * bucket_size
key = (row["date"], normalized_hour)
item = rows.setdefault(key, self._empty_context_row(row["date"], normalized_hour))
local_dt = datetime(
*_parse_ymd(row["date_utc"]),
int(row["hour_utc"]),
tzinfo=UTC,
).astimezone(tz)
local_date = local_dt.date().isoformat()
local_hour = local_dt.hour
normalized_hour = (local_hour // bucket_size) * bucket_size
key = (local_date, normalized_hour)
item = rows.setdefault(key, self._empty_context_row(local_date, normalized_hour))
operation_key = str(row["operation"]).replace(".", "_")
value = int(row["total"] or 0)
if operation_key in item:
item[operation_key] += int(row["total"] or 0)
item["total"] += int(row["total"] or 0)
return [rows[key] for key in sorted(rows)]
item[operation_key] += value
item["total"] += value
# Only return rows inside the requested user-tz range; UTC hours at
# boundaries that fall outside the range after rebucketing are dropped.
in_range = set(_date_range(start_user_date, end_user_date))
return [rows[key] for key in sorted(rows) if key[0] in in_range]
def _fetch_hourly_token_rows(
self, account_id: str, utc_start: datetime, utc_end: datetime
) -> list[tuple[str, str, str, int, int]]:
"""Return rows [(source, token_type, date_utc, hour_utc, total)]."""
assert self._conn is not None
utc_start_d = utc_start.date().isoformat()
utc_end_d = utc_end.date().isoformat()
cur = self._conn.execute(
"""
SELECT source, token_type, date_utc, hour_utc, SUM(token_count) AS total
FROM usage_token_hourly
WHERE account_id = ?
AND (date_utc > ? OR (date_utc = ? AND hour_utc >= ?))
AND (date_utc < ? OR (date_utc = ? AND hour_utc < ?))
GROUP BY source, token_type, date_utc, hour_utc
""",
(
account_id,
utc_start_d,
utc_start_d,
utc_start.hour,
utc_end_d,
utc_end_d,
utc_end.hour,
),
)
return [
(
str(row["source"]),
str(row["token_type"]),
str(row["date_utc"]),
int(row["hour_utc"]),
int(row["total"] or 0),
)
for row in cur.fetchall()
]
def _fetch_hourly_retrieval_rows(
self, account_id: str, utc_start: datetime, utc_end: datetime
) -> list[tuple[str, int]]:
"""Return [(operation, total_request_count)] for successful retrievals."""
assert self._conn is not None
utc_start_d = utc_start.date().isoformat()
utc_end_d = utc_end.date().isoformat()
cur = self._conn.execute(
"""
SELECT operation, SUM(request_count) AS total
FROM usage_retrieval_hourly
WHERE account_id = ?
AND status = 'success'
AND (date_utc > ? OR (date_utc = ? AND hour_utc >= ?))
AND (date_utc < ? OR (date_utc = ? AND hour_utc < ?))
GROUP BY operation
""",
(
account_id,
utc_start_d,
utc_start_d,
utc_start.hour,
utc_end_d,
utc_end_d,
utc_end.hour,
),
)
return [(str(row["operation"]), int(row["total"] or 0)) for row in cur.fetchall()]
@staticmethod
def _empty_context_row(event_date: str, hour: int) -> dict[str, Any]:
@@ -556,3 +751,8 @@ class SQLiteUsageAuditStore:
if not clauses:
return "", []
return "(" + " OR ".join(clauses) + ")", params
def _parse_ymd(date_str: str) -> tuple[int, int, int]:
parts = date_str.split("-")
return int(parts[0]), int(parts[1]), int(parts[2])
openviking/observability/usage_audit/store.py
+31 -10
View File
@@ -4,13 +4,18 @@
from __future__ import annotations
from datetime import tzinfo
from typing import Any, Protocol, Sequence
from openviking.observability.events import ObservabilityEvent
class UsageAuditStore(Protocol):
"""Persistence contract for product usage/audit data."""
"""Persistence contract for product usage/audit data.
Time-keyed columns are stored in UTC; read methods accept a viewer-supplied
`tz` and rebucket on the fly.
"""
async def initialize(self) -> None:
"""Initialize the store."""
@@ -21,26 +26,42 @@ class UsageAuditStore(Protocol):
async def record_batch(self, events: Sequence[ObservabilityEvent]) -> None:
"""Persist a batch of observability events."""
async def get_today_tokens(self, *, account_id: str, date: str) -> dict[str, int]:
"""Return token totals for one account and date."""
async def get_today_tokens(
self, *, account_id: str, user_date: str, tz: tzinfo
) -> dict[str, int]:
"""Return token totals for one account and viewer-local date."""
async def get_today_retrievals(self, *, account_id: str, date: str) -> dict[str, int]:
async def get_today_retrievals(
self, *, account_id: str, user_date: str, tz: tzinfo
) -> dict[str, int]:
"""Return successful find/search counts for one account and date."""
async def get_agent_overview(
self, *, account_id: str, date: str, limit: int = 5
self, *, account_id: str, user_date: str, tz: tzinfo, limit: int = 5
) -> dict[str, Any]:
"""Return distinct agent count and recent agents for one account/date."""
"""Return distinct agent count and recent agents for the viewer's day."""
async def get_token_series(
self, *, account_id: str, start_date: str, end_date: str, bucket: str
self,
*,
account_id: str,
start_user_date: str,
end_user_date: str,
bucket: str,
tz: tzinfo,
) -> list[dict[str, Any]]:
"""Return token series rows for a date range."""
"""Return token series rows for a viewer-local date range."""
async def get_context_commit_heatmap(
self, *, account_id: str, start_date: str, end_date: str, bucket: str
self,
*,
account_id: str,
start_user_date: str,
end_user_date: str,
bucket: str,
tz: tzinfo,
) -> list[dict[str, Any]]:
"""Return context write bucket rows for a date range."""
"""Return context write bucket rows for a viewer-local date range."""
async def query_audit_logs(
self,
openviking/observability/usage_audit/time.py
+20 -1
View File
@@ -12,7 +12,11 @@ logger = logging.getLogger(__name__)
def resolve_usage_timezone(timezone_name: str) -> tzinfo:
"""Resolve configured Usage/Audit timezone with local fallback."""
"""Resolve the server-default Usage/Audit timezone with local fallback.
Used only as the fallback when a request does not specify its own
`?timezone=` parameter. Writes are always in UTC regardless of this value.
"""
if not timezone_name or timezone_name == "local":
return datetime.now().astimezone().tzinfo or timezone.utc
try:
@@ -20,3 +24,18 @@ def resolve_usage_timezone(timezone_name: str) -> tzinfo:
except ZoneInfoNotFoundError:
logger.warning("Unknown usage_audit timezone %s; falling back to local", timezone_name)
return datetime.now().astimezone().tzinfo or timezone.utc
def resolve_user_timezone(timezone_name: str | None, *, fallback: tzinfo) -> tzinfo:
"""Resolve a request-supplied IANA tz name with a server-side fallback.
Accepts e.g. `Asia/Shanghai`, `America/New_York`, or `UTC`. An unknown or
empty value falls back to the server default and emits a debug log entry.
"""
if not timezone_name:
return fallback
try:
return ZoneInfo(timezone_name)
except ZoneInfoNotFoundError:
logger.debug("Unknown request timezone %r; falling back to server default", timezone_name)
return fallback
openviking/server/routers/console.py
+19 -5
View File
@@ -47,21 +47,29 @@ def _disabled_response():
@router.get("/dashboard/summary")
async def dashboard_summary(
request: Request,
timezone: Optional[str] = Query(
None,
description="IANA viewer timezone (e.g. Asia/Shanghai). Defaults to server tz.",
),
_ctx: RequestContext = require_role(Role.ROOT, Role.ADMIN),
):
"""Return Dashboard top-card data."""
service = _runtime_service(request)
if service is None:
return _disabled_response()
return _ok_response(await service.dashboard_summary(_ctx))
return _ok_response(await service.dashboard_summary(_ctx, timezone_name=timezone))
@router.get("/tokens")
async def token_series(
request: Request,
start_date: str = Query(..., description="Start date in YYYY-MM-DD format"),
end_date: str = Query(..., description="End date in YYYY-MM-DD format"),
start_date: str = Query(..., description="Start date (viewer-local) in YYYY-MM-DD"),
end_date: str = Query(..., description="End date (viewer-local) in YYYY-MM-DD"),
bucket: str = Query("day", pattern="^(day)$"),
timezone: Optional[str] = Query(
None,
description="IANA viewer timezone (e.g. Asia/Shanghai). Defaults to server tz.",
),
_ctx: RequestContext = require_role(Role.ROOT, Role.ADMIN),
):
"""Return token usage trend for a date range."""
@@ -73,6 +81,7 @@ async def token_series(
start_date=start_date,
end_date=end_date,
bucket=bucket,
timezone_name=timezone,
)
return _ok_response(result)
@@ -80,9 +89,13 @@ async def token_series(
@router.get("/context-commits")
async def context_commits(
request: Request,
start_date: str = Query(..., description="Start date in YYYY-MM-DD format"),
end_date: str = Query(..., description="End date in YYYY-MM-DD format"),
start_date: str = Query(..., description="Start date (viewer-local) in YYYY-MM-DD"),
end_date: str = Query(..., description="End date (viewer-local) in YYYY-MM-DD"),
bucket: str = Query("hour", pattern="^(hour|4h)$"),
timezone: Optional[str] = Query(
None,
description="IANA viewer timezone (e.g. Asia/Shanghai). Defaults to server tz.",
),
_ctx: RequestContext = require_role(Role.ROOT, Role.ADMIN),
):
"""Return context write heatmap rows for a date range."""
@@ -94,6 +107,7 @@ async def context_commits(
start_date=start_date,
end_date=end_date,
bucket=bucket,
timezone_name=timezone,
)
return _ok_response(result)
tests/observability/test_console_router.py
+72 -1
View File
@@ -53,9 +53,38 @@ def _app_with_runtime(runtime=None, *, request_context=_admin_ctx) -> FastAPI:
class FakeConsoleService:
def __init__(self) -> None:
self.audit_call = None
self.token_series_call = None
self.dashboard_call = None
self.context_commits_call = None
async def token_series(self, **kwargs):
raise InvalidArgumentError("bad date range")
self.token_series_call = kwargs
if kwargs.get("end_date", "") < kwargs.get("start_date", ""):
raise InvalidArgumentError("bad date range")
return {
"start_date": kwargs["start_date"],
"end_date": kwargs["end_date"],
"bucket": kwargs["bucket"],
"items": [],
}
async def context_commits(self, **kwargs):
self.context_commits_call = kwargs
return {
"start_date": kwargs["start_date"],
"end_date": kwargs["end_date"],
"bucket": kwargs["bucket"],
"items": [],
}
async def dashboard_summary(self, ctx, **kwargs):
self.dashboard_call = {"ctx": ctx, **kwargs}
return {
"context_counts": {},
"today_tokens": {},
"today_retrievals": {},
"agent_overview": {"total": 0, "items": []},
}
async def audit_logs(self, **kwargs):
self.audit_call = kwargs
@@ -127,3 +156,45 @@ async def test_console_router_invalid_arguments_return_http_400():
assert response.status_code == 400
assert response.json()["error"]["code"] == "INVALID_ARGUMENT"
@pytest.mark.asyncio
async def test_console_router_passes_timezone_to_dashboard_and_series():
service = FakeConsoleService()
transport = httpx.ASGITransport(app=_app_with_runtime(FakeRuntime(service)))
async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
await client.get(
"/api/v1/console/dashboard/summary",
params={"timezone": "Asia/Shanghai"},
)
await client.get(
"/api/v1/console/tokens",
params={
"start_date": "2026-05-01",
"end_date": "2026-05-12",
"timezone": "America/New_York",
},
)
await client.get(
"/api/v1/console/context-commits",
params={
"start_date": "2026-05-01",
"end_date": "2026-05-12",
"bucket": "4h",
"timezone": "Europe/Berlin",
},
)
assert service.dashboard_call["timezone_name"] == "Asia/Shanghai"
assert service.token_series_call["timezone_name"] == "America/New_York"
assert service.context_commits_call["timezone_name"] == "Europe/Berlin"
@pytest.mark.asyncio
async def test_console_router_defaults_timezone_to_none_when_missing():
service = FakeConsoleService()
transport = httpx.ASGITransport(app=_app_with_runtime(FakeRuntime(service)))
async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
await client.get("/api/v1/console/dashboard/summary")
assert service.dashboard_call["timezone_name"] is None
tests/observability/test_usage_audit_runtime.py
+4 -1
View File
@@ -46,9 +46,12 @@ async def test_usage_audit_runtime_subscribes_to_shared_event_bus(tmp_path):
)
await runtime.worker.close(timeout_seconds=1.0)
from zoneinfo import ZoneInfo
retrievals = await runtime.store.get_today_retrievals(
account_id="acct-runtime",
date=runtime.api_service.today(),
user_date=runtime.api_service.today(),
tz=ZoneInfo("UTC"),
)
audit = await runtime.store.query_audit_logs(account_id="acct-runtime")
finally:
tests/observability/test_usage_audit_store.py
+334 -10
View File
@@ -3,19 +3,29 @@
from __future__ import annotations
import sqlite3
from datetime import datetime, timezone
from zoneinfo import ZoneInfo
import pytest
from openviking.observability.events import ObservabilityEvent
from openviking.observability.usage_audit.sqlite_store import SQLiteUsageAuditStore
UTC = ZoneInfo("UTC")
def _event(event_name: str, payload: dict, *, agent_id: str | None = None) -> ObservabilityEvent:
def _event(
event_name: str,
payload: dict,
*,
agent_id: str | None = None,
ts: datetime | None = None,
) -> ObservabilityEvent:
return ObservabilityEvent(
event_name=event_name,
payload=payload,
timestamp=datetime(2026, 5, 12, 1, 2, 3, tzinfo=timezone.utc),
timestamp=ts or datetime(2026, 5, 12, 1, 2, 3, tzinfo=timezone.utc),
request_id=payload.get("request_id"),
account_id="acct-1",
user_id="user-1",
@@ -23,9 +33,90 @@ def _event(event_name: str, payload: dict, *, agent_id: str | None = None) -> Ob
)
def _create_legacy_usage_audit_db(db_path) -> None:
conn = sqlite3.connect(db_path)
try:
conn.executescript(
"""
CREATE TABLE usage_token_daily (
account_id TEXT NOT NULL,
user_id TEXT NOT NULL,
agent_id TEXT NOT NULL,
date TEXT NOT NULL,
source TEXT NOT NULL,
token_type TEXT NOT NULL,
provider TEXT NOT NULL,
model_name TEXT NOT NULL,
token_count INTEGER NOT NULL DEFAULT 0,
updated_at TEXT NOT NULL,
PRIMARY KEY (
account_id, user_id, agent_id, date, source, token_type, provider, model_name
)
);
CREATE TABLE usage_retrieval_daily (
account_id TEXT NOT NULL,
user_id TEXT NOT NULL,
agent_id TEXT NOT NULL,
date TEXT NOT NULL,
operation TEXT NOT NULL,
status TEXT NOT NULL,
request_count INTEGER NOT NULL DEFAULT 0,
result_count INTEGER NOT NULL DEFAULT 0,
updated_at TEXT NOT NULL,
PRIMARY KEY (account_id, user_id, agent_id, date, operation, status)
);
CREATE TABLE usage_context_write_bucket (
account_id TEXT NOT NULL,
user_id TEXT NOT NULL,
agent_id TEXT NOT NULL,
date TEXT NOT NULL,
hour_bucket INTEGER NOT NULL,
operation TEXT NOT NULL,
count INTEGER NOT NULL DEFAULT 0,
updated_at TEXT NOT NULL,
PRIMARY KEY (account_id, user_id, agent_id, date, hour_bucket, operation)
);
CREATE TABLE usage_agent_activity_daily (
account_id TEXT NOT NULL,
agent_id TEXT NOT NULL,
date TEXT NOT NULL,
request_count INTEGER NOT NULL DEFAULT 0,
last_seen_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
PRIMARY KEY (account_id, agent_id, date)
);
CREATE TABLE request_audit (
id INTEGER PRIMARY KEY AUTOINCREMENT,
request_id TEXT,
account_id TEXT NOT NULL,
user_id TEXT,
agent_id TEXT,
method TEXT NOT NULL,
route TEXT NOT NULL,
api_type TEXT NOT NULL,
status_code INTEGER NOT NULL,
duration_ms REAL NOT NULL,
created_at TEXT NOT NULL
);
INSERT INTO request_audit (
request_id, account_id, user_id, agent_id, method, route,
api_type, status_code, duration_ms, created_at
)
VALUES (
'legacy-req', 'acct-1', 'user-1', 'agent-1', 'GET',
'/api/v1/system/status', 'system', 200, 1.0,
'2026-05-11T00:00:00+08:00'
);
"""
)
conn.commit()
finally:
conn.close()
@pytest.mark.asyncio
async def test_sqlite_usage_audit_store_aggregates_dashboard_data(tmp_path):
store = SQLiteUsageAuditStore(tmp_path / "usage.sqlite3", timezone_name="UTC")
store = SQLiteUsageAuditStore(tmp_path / "usage.sqlite3")
await store.initialize()
try:
await store.record_batch(
@@ -105,27 +196,33 @@ async def test_sqlite_usage_audit_store_aggregates_dashboard_data(tmp_path):
]
)
assert await store.get_today_tokens(account_id="acct-1", date="2026-05-12") == {
assert await store.get_today_tokens(
account_id="acct-1", user_date="2026-05-12", tz=UTC
) == {
"vlm_input": 3,
"vlm_output": 2,
"embedding_input": 5,
"total": 10,
}
assert await store.get_today_retrievals(account_id="acct-1", date="2026-05-12") == {
assert await store.get_today_retrievals(
account_id="acct-1", user_date="2026-05-12", tz=UTC
) == {
"find": 1,
"search": 0,
"total": 1,
}
commits = await store.get_context_commit_heatmap(
account_id="acct-1",
start_date="2026-05-12",
end_date="2026-05-12",
start_user_date="2026-05-12",
end_user_date="2026-05-12",
bucket="hour",
tz=UTC,
)
assert any(row["hour"] == 1 and row["session_add_message"] == 1 for row in commits)
agents = await store.get_agent_overview(
account_id="acct-1",
date="2026-05-12",
user_date="2026-05-12",
tz=UTC,
)
assert agents["total"] == 1
assert agents["items"][0]["agent_id"] == "agent-1"
@@ -137,12 +234,94 @@ async def test_sqlite_usage_audit_store_aggregates_dashboard_data(tmp_path):
await store.close()
@pytest.mark.asyncio
async def test_sqlite_usage_audit_store_resets_incompatible_legacy_schema(tmp_path):
db_path = tmp_path / "usage.sqlite3"
_create_legacy_usage_audit_db(db_path)
store = SQLiteUsageAuditStore(db_path)
await store.initialize()
try:
await store.record_batch(
[
_event(
"vlm.call",
{
"provider": "p",
"model_name": "m",
"prompt_tokens": 8,
"completion_tokens": 2,
},
),
_event(
"http.request",
{
"request_id": "req-message",
"method": "POST",
"route": "/api/v1/sessions/{session_id}/messages",
"status": "200",
"duration_seconds": 0.1,
},
agent_id="agent-1",
),
]
)
assert await store.get_today_tokens(
account_id="acct-1", user_date="2026-05-12", tz=UTC
) == {
"vlm_input": 8,
"vlm_output": 2,
"embedding_input": 0,
"total": 10,
}
commits = await store.get_context_commit_heatmap(
account_id="acct-1",
start_user_date="2026-05-12",
end_user_date="2026-05-12",
bucket="hour",
tz=UTC,
)
assert any(row["hour"] == 1 and row["session_add_message"] == 1 for row in commits)
agents = await store.get_agent_overview(
account_id="acct-1",
user_date="2026-05-12",
tz=UTC,
)
assert agents["total"] == 1
audit = await store.query_audit_logs(account_id="acct-1", page_size=10)
assert audit["total"] == 1
assert audit["items"][0]["request_id"] == "req-message"
finally:
await store.close()
conn = sqlite3.connect(db_path)
try:
tables = {
row[0] for row in conn.execute("SELECT name FROM sqlite_master WHERE type = 'table'")
}
assert "usage_token_daily" not in tables
assert "usage_retrieval_daily" not in tables
for table in ("usage_context_write_bucket", "usage_agent_activity_daily"):
columns = {row[1] for row in conn.execute(f"PRAGMA table_info({table})")}
assert "date_utc" in columns
assert "date" not in columns
context_columns = {
row[1] for row in conn.execute("PRAGMA table_info(usage_context_write_bucket)")
}
assert "hour_utc" in context_columns
assert "hour_bucket" not in context_columns
version = conn.execute("SELECT value FROM _schema_meta WHERE key = 'version'").fetchone()
assert version == ("3",)
finally:
conn.close()
@pytest.mark.asyncio
async def test_sqlite_usage_audit_store_trims_audit_per_account(tmp_path):
store = SQLiteUsageAuditStore(
tmp_path / "usage.sqlite3",
audit_retention_per_account=2,
timezone_name="UTC",
)
await store.initialize()
try:
@@ -170,7 +349,7 @@ async def test_sqlite_usage_audit_store_trims_audit_per_account(tmp_path):
@pytest.mark.asyncio
async def test_sqlite_usage_audit_store_success_filter_includes_3xx(tmp_path):
store = SQLiteUsageAuditStore(tmp_path / "usage.sqlite3", timezone_name="UTC")
store = SQLiteUsageAuditStore(tmp_path / "usage.sqlite3")
await store.initialize()
try:
await store.record_batch(
@@ -225,3 +404,148 @@ async def test_sqlite_usage_audit_store_success_filter_includes_3xx(tmp_path):
assert explicit_2xx["items"][0]["request_id"] == "req-200"
finally:
await store.close()
@pytest.mark.asyncio
async def test_today_tokens_rebuckets_for_utc_plus_8(tmp_path):
"""A UTC+8 user querying "today" must include UTC events that fall inside
their local day, even when they span two UTC dates.
Two events bracket a UTC day boundary at 16:00 UTC = 00:00 next day
in Asia/Shanghai. From the SH viewer's perspective on 2026-05-13 both
events belong to "today"; from a UTC viewer they straddle two days.
"""
store = SQLiteUsageAuditStore(tmp_path / "usage.sqlite3")
await store.initialize()
try:
await store.record_batch(
[
_event( # 2026-05-12 23:30 UTC == 2026-05-13 07:30 SH
"vlm.call",
{
"provider": "p",
"model_name": "m",
"prompt_tokens": 11,
"completion_tokens": 0,
},
ts=datetime(2026, 5, 12, 23, 30, tzinfo=timezone.utc),
),
_event( # 2026-05-13 04:30 UTC == 2026-05-13 12:30 SH
"vlm.call",
{
"provider": "p",
"model_name": "m",
"prompt_tokens": 22,
"completion_tokens": 0,
},
ts=datetime(2026, 5, 13, 4, 30, tzinfo=timezone.utc),
),
_event( # 2026-05-13 16:30 UTC == 2026-05-14 00:30 SH (next day)
"vlm.call",
{
"provider": "p",
"model_name": "m",
"prompt_tokens": 9999,
"completion_tokens": 0,
},
ts=datetime(2026, 5, 13, 16, 30, tzinfo=timezone.utc),
),
]
)
shanghai = ZoneInfo("Asia/Shanghai")
sh_today = await store.get_today_tokens(
account_id="acct-1", user_date="2026-05-13", tz=shanghai
)
# Includes the 23:30 UTC event but excludes the 16:30 UTC event that
# already rolled into the next SH day.
assert sh_today["vlm_input"] == 33
assert sh_today["total"] == 33
utc_today = await store.get_today_tokens(
account_id="acct-1", user_date="2026-05-13", tz=UTC
)
# UTC viewer's day starts at 00:00 UTC, so misses the 23:30 prior-day
# event but includes the 16:30 event.
assert utc_today["vlm_input"] == 22 + 9999
finally:
await store.close()
@pytest.mark.asyncio
async def test_token_series_rebuckets_to_user_tz_days(tmp_path):
"""An event at 03:00 UTC must land in `2026-05-11` for an America/New_York
viewer (UTC-4 with DST in May = 23:00 of the previous day local)."""
store = SQLiteUsageAuditStore(tmp_path / "usage.sqlite3")
await store.initialize()
try:
await store.record_batch(
[
_event(
"vlm.call",
{
"provider": "p",
"model_name": "m",
"prompt_tokens": 7,
"completion_tokens": 0,
},
ts=datetime(2026, 5, 12, 3, 0, tzinfo=timezone.utc),
),
]
)
ny = ZoneInfo("America/New_York")
series = await store.get_token_series(
account_id="acct-1",
start_user_date="2026-05-10",
end_user_date="2026-05-12",
bucket="day",
tz=ny,
)
by_date = {row["date"]: row for row in series}
# 03:00 UTC on 2026-05-12 falls on 2026-05-11 23:00 in New York.
assert by_date["2026-05-11"]["vlm_input"] == 7
assert by_date["2026-05-12"]["vlm_input"] == 0
finally:
await store.close()
@pytest.mark.asyncio
async def test_context_heatmap_rebuckets_hour_for_shanghai(tmp_path):
"""A context commit at 23:30 UTC must show up at hour 7 of the next day in
Asia/Shanghai (UTC+8)."""
store = SQLiteUsageAuditStore(tmp_path / "usage.sqlite3")
await store.initialize()
try:
await store.record_batch(
[
_event(
"http.request",
{
"request_id": "req-late",
"method": "POST",
"route": "/api/v1/sessions/{session_id}/messages",
"status": "200",
"duration_seconds": 0.01,
},
ts=datetime(2026, 5, 12, 23, 30, tzinfo=timezone.utc),
agent_id="agent-1",
),
]
)
shanghai = ZoneInfo("Asia/Shanghai")
rows = await store.get_context_commit_heatmap(
account_id="acct-1",
start_user_date="2026-05-13",
end_user_date="2026-05-13",
bucket="hour",
tz=shanghai,
)
match = next(
(row for row in rows if row["date"] == "2026-05-13" and row["hour"] == 7),
None,
)
assert match is not None
assert match["session_add_message"] == 1
assert match["total"] == 1
finally:
await store.close()
web-studio/src/gen/ov-client/types.gen.ts
+24 -5
View File
@@ -2292,7 +2292,14 @@ export type GetConsoleDashboardSummaryData = {
'X-OpenViking-Agent'?: string | null;
};
path?: never;
query?: never;
query?: {
/**
* Timezone
*
* IANA viewer timezone (e.g. Asia/Shanghai). Defaults to server tz.
*/
timezone?: string | null;
};
url: '/api/v1/console/dashboard/summary';
};
@@ -2341,19 +2348,25 @@ export type GetConsoleTokensData = {
/**
* Start Date
*
* Start date in YYYY-MM-DD format
* Start date (viewer-local) in YYYY-MM-DD
*/
start_date: string;
/**
* End Date
*
* End date in YYYY-MM-DD format
* End date (viewer-local) in YYYY-MM-DD
*/
end_date: string;
/**
* Bucket
*/
bucket?: string;
/**
* Timezone
*
* IANA viewer timezone (e.g. Asia/Shanghai). Defaults to server tz.
*/
timezone?: string | null;
};
url: '/api/v1/console/tokens';
};
@@ -2403,19 +2416,25 @@ export type GetConsoleContextCommitsData = {
/**
* Start Date
*
* Start date in YYYY-MM-DD format
* Start date (viewer-local) in YYYY-MM-DD
*/
start_date: string;
/**
* End Date
*
* End date in YYYY-MM-DD format
* End date (viewer-local) in YYYY-MM-DD
*/
end_date: string;
/**
* Bucket
*/
bucket?: string;
/**
* Timezone
*
* IANA viewer timezone (e.g. Asia/Shanghai). Defaults to server tz.
*/
timezone?: string | null;
};
url: '/api/v1/console/context-commits';
};
web-studio/src/routes/home/-lib/api.ts
+8 -4
View File
@@ -12,30 +12,34 @@ import type {
ConsoleDashboardSummaryResult,
ConsoleTokenSeriesResult,
} from '@ov-server/api/v1/console'
import { getLastDaysRange } from './format'
import { getLastDaysRange, getViewerTimezone } from './format'
export function fetchConsoleDashboardSummary(): Promise<ConsoleDashboardSummaryResult> {
return getOvResult<ConsoleDashboardSummaryResult>(
getConsoleDashboardSummary(),
getConsoleDashboardSummary({ query: { timezone: getViewerTimezone() } }),
)
}
export function fetchConsoleTokenSeries(): Promise<ConsoleTokenSeriesResult> {
const range = getLastDaysRange(TOKEN_SERIES_DAYS)
const tz = getViewerTimezone()
const range = getLastDaysRange(TOKEN_SERIES_DAYS, tz)
const query: ConsoleSeriesQuery = {
bucket: 'day',
end_date: range.endDate,
start_date: range.startDate,
timezone: tz,
}
return getOvResult<ConsoleTokenSeriesResult>(getConsoleTokens({ query }))
}
export function fetchConsoleContextCommits(): Promise<ConsoleContextCommitsResult> {
const range = getLastDaysRange(COMMIT_SERIES_DAYS)
const tz = getViewerTimezone()
const range = getLastDaysRange(COMMIT_SERIES_DAYS, tz)
const query: ConsoleSeriesQuery = {
bucket: '4h',
end_date: range.endDate,
start_date: range.startDate,
timezone: tz,
}
return getOvResult<ConsoleContextCommitsResult>(
getConsoleContextCommits({ query }),
web-studio/src/routes/home/-lib/format.ts
+34 -6
View File
@@ -35,16 +35,44 @@ export function parseDateKey(value: string | undefined): Date {
return new Date(year, month - 1, day)
}
export function getLastDaysRange(days: number): {
export function getViewerTimezone(): string {
// Falls back to UTC when Intl is unavailable or returns an empty string
// (e.g. very old Safari, jest jsdom edge cases).
try {
return Intl.DateTimeFormat().resolvedOptions().timeZone || 'UTC'
} catch {
return 'UTC'
}
}
function formatDateInTimezone(d: Date, timeZone: string): string {
// `en-CA` produces YYYY-MM-DD, which matches the backend `date` query
// parameter shape exactly.
return new Intl.DateTimeFormat('en-CA', {
timeZone,
year: 'numeric',
month: '2-digit',
day: '2-digit',
}).format(d)
}
export function getLastDaysRange(
days: number,
timeZone?: string,
): {
endDate: string
startDate: string
} {
const end = new Date()
const start = new Date(end)
start.setDate(end.getDate() - days + 1)
const tz = timeZone ?? getViewerTimezone()
const now = new Date()
// Walk back `days - 1` UTC midnights from now and convert each instant in
// the viewer's tz; using millisecond arithmetic keeps the math safe across
// DST transitions because we never inspect a local hour.
const dayMs = 24 * 60 * 60 * 1000
const earlier = new Date(now.getTime() - (days - 1) * dayMs)
return {
endDate: formatDateKey(end),
startDate: formatDateKey(start),
endDate: formatDateInTimezone(now, tz),
startDate: formatDateInTimezone(earlier, tz),
}
}
web-studio/types/ov-server/api/v1/console.ts
+5
View File
@@ -47,6 +47,11 @@ export type ConsoleSeriesQuery = {
bucket?: ConsoleSeriesBucket
end_date: string
start_date: string
timezone?: string | null
}
export type ConsoleDashboardSummaryQuery = {
timezone?: string | null
}
export type ConsoleSeriesResult<TItem> = OvMaybeDisabled & {