feat(console): add usage audit retention config

Author: qin-ctx

openviking/observability/usage_audit/README.md

@@ -66,6 +66,8 @@ Observability Event Bus
         "batch_size": 500,
         "flush_interval_seconds": 1.0,
         "shutdown_flush_timeout_seconds": 3.0,
+        "usage_retention_days": 14,
+        "audit_retention_days": 7,
         "audit_retention_per_account": 1000,
         "timezone": "local",
         "inventory_ttl_seconds": 10.0
@@ -86,7 +88,9 @@ Observability Event Bus
 | `batch_size` | `500` | 单次批量写入的最大事件数 |
 | `flush_interval_seconds` | `1.0` | worker 定时 flush 间隔 |
 | `shutdown_flush_timeout_seconds` | `3.0` | 服务关闭时 flush 等待时间 |
-| `audit_retention_per_account` | `1000` | 每个 account 保留的最新请求审计条数；`0` 表示不主动裁剪 |
+| `usage_retention_days` | `14` | 统计聚合数据保留天数，包含 Token、检索、上下文写入热力图、Agent 活跃；`0` 表示不按天裁剪 |
+| `audit_retention_days` | `7` | 请求审计日志保留天数；`0` 表示不按天裁剪 |
+| `audit_retention_per_account` | `1000` | 每个 account 保留的最新请求审计条数；`0` 表示不按条数裁剪 |
 | `timezone` | `"local"` | 统计日期使用的时区；可填 `"local"`、`"UTC"` 或 IANA 时区名 |
 | `inventory_ttl_seconds` | `10.0` | 上下文当前数据量查询缓存时间 |
 

openviking/observability/usage_audit/runtime.py

@@ -65,6 +65,8 @@ async def init_usage_audit_from_server_config(
 
     store = SQLiteUsageAuditStore(
         _resolve_sqlite_path(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,
     )

openviking/observability/usage_audit/sqlite_store.py

@@ -33,10 +33,14 @@ def __init__(
         self,
         db_path: Path,
         *,
+        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
@@ -85,6 +89,7 @@ def _record_projection_sync(self, projection: UsageAuditProjection) -> None:
             self._write_context_rows(conn, projection.context_rows, updated_at)
             self._write_agent_rows(conn, projection.agent_rows, updated_at)
             self._write_audit_rows(conn, projection.audit_rows)
+            self._trim_usage_rows(conn, self._usage_max_dates(projection))
             self._trim_audit_rows(conn, projection.touched_audit_accounts)
             conn.execute("COMMIT")
         except Exception:
@@ -175,6 +180,19 @@ def _write_audit_rows(conn, rows: list[tuple]) -> None:
         )
 
     def _trim_audit_rows(self, conn, accounts: set[str]) -> None:
+        if self._audit_retention_days > 0:
+            max_dates = self._audit_max_dates(conn, accounts)
+            for account_id, cutoff_date in self._cutoff_dates(
+                max_dates,
+                retention_days=self._audit_retention_days,
+            ).items():
+                conn.execute(
+                    """
+                    DELETE FROM request_audit
+                    WHERE account_id = ? AND substr(created_at, 1, 10) < ?
+                    """,
+                    (account_id, cutoff_date),
+                )
         if self._audit_retention_per_account <= 0:
             return
         for account_id in accounts:
@@ -192,6 +210,67 @@ def _trim_audit_rows(self, conn, accounts: set[str]) -> None:
                 (account_id, account_id, self._audit_retention_per_account),
             )
 
+    def _trim_usage_rows(self, conn, max_dates_by_account: dict[str, str]) -> None:
+        cutoff_by_account = self._cutoff_dates(
+            max_dates_by_account,
+            retention_days=self._usage_retention_days,
+        )
+        for account_id, cutoff_date in cutoff_by_account.items():
+            for table in (
+                "usage_token_daily",
+                "usage_retrieval_daily",
+                "usage_context_write_bucket",
+                "usage_agent_activity_daily",
+            ):
+                conn.execute(
+                    f"DELETE FROM {table} WHERE account_id = ? AND date < ?",
+                    (account_id, cutoff_date),
+                )
+
+    @staticmethod
+    def _usage_max_dates(projection: UsageAuditProjection) -> dict[str, str]:
+        max_dates: dict[str, str] = {}
+        SQLiteUsageAuditStore._merge_max_dates(max_dates, projection.token_rows, date_index=3)
+        SQLiteUsageAuditStore._merge_max_dates(max_dates, projection.retrieval_rows, date_index=3)
+        SQLiteUsageAuditStore._merge_max_dates(max_dates, projection.context_rows, date_index=3)
+        SQLiteUsageAuditStore._merge_max_dates(max_dates, projection.agent_rows, date_index=2)
+        return max_dates
+
+    @staticmethod
+    def _merge_max_dates(target: dict[str, str], rows: dict[tuple, Any], *, date_index: int) -> None:
+        for key in rows:
+            account_id = str(key[0])
+            event_date = str(key[date_index])
+            if event_date > target.get(account_id, ""):
+                target[account_id] = event_date
+
+    @staticmethod
+    def _cutoff_dates(max_dates_by_account: dict[str, str], *, retention_days: int) -> dict[str, str]:
+        if retention_days <= 0:
+            return {}
+        return {
+            account_id: (
+                date.fromisoformat(max_date) - timedelta(days=retention_days - 1)
+            ).isoformat()
+            for account_id, max_date in max_dates_by_account.items()
+        }
+
+    @staticmethod
+    def _audit_max_dates(conn, accounts: set[str]) -> dict[str, str]:
+        max_dates: dict[str, str] = {}
+        for account_id in accounts:
+            row = conn.execute(
+                """
+                SELECT MAX(substr(created_at, 1, 10)) AS max_date
+                FROM request_audit
+                WHERE account_id = ?
+                """,
+                (account_id,),
+            ).fetchone()
+            if row and row["max_date"]:
+                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)

openviking/server/config.py

@@ -95,6 +95,8 @@ class UsageAuditConfig(BaseModel):
     batch_size: int = Field(500, gt=0)
     flush_interval_seconds: float = Field(1.0, gt=0)
     shutdown_flush_timeout_seconds: float = Field(3.0, gt=0)
+    usage_retention_days: int = Field(14, ge=0)
+    audit_retention_days: int = Field(7, ge=0)
     audit_retention_per_account: int = Field(1000, ge=0)
     timezone: str = "local"
     inventory_ttl_seconds: float = Field(10.0, ge=0)