From 4669d3c0896a89f1a650018bf6390679c657660e Mon Sep 17 00:00:00 2001 From: codeingforcoffee <105403354+codeingforcoffee@users.noreply.github.com> Date: Sat, 4 Jul 2026 23:14:46 +0800 Subject: [PATCH] feat(gateway): cache-aware cost accounting (#3920) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(gateway): cache-aware cost accounting + /api/console observability endpoints - Capture prompt-cache hits (usage_metadata.input_token_details.cache_read) in RunJournal and SubagentTokenCollector as a sparse cache_read_tokens key in token_usage_by_model (JSON field — no schema migration; legacy bucket shapes unchanged) - New read-only /api/console router: GET /stats (headline counters), GET /runs (cross-thread paginated history joined with thread titles), GET /usage (zero-filled daily token series + per-model breakdown); user-scoped, 503 on the memory database backend - Optional models[*].pricing (currency, input_per_million, output_per_million, input_cache_hit_per_million) powers real spend estimation; cache-hit input tokens are billed at the hit price (omitted hit price falls back to the miss price as a conservative upper bound); unpriced models yield cost: null - create_chat_model strips the presentation-only pricing block so it never reaches the provider client (unknown kwargs are forwarded into the completion payload and break live calls) - Tests: console router SQLite round-trips, journal/collector cache capture incl. a DeepSeek raw-usage pin test, factory strip regression Co-Authored-By: Claude Fable 5 * refactor: address review feedback on cost sum and sparse cache_read_tokens - console.py: replace the walrus-in-generator total-cost sum with an explicit loop (review noted the multi-line form reads ambiguously) - token_collector.py: omit cache_read_tokens from usage records when the provider reported no cache hits, matching the journal's sparse per-model bucket shape; absent is treated as 0 downstream - add a regression test pinning the sparse record shape Co-Authored-By: Claude Fable 5 --------- Co-authored-by: coffeeFish Co-authored-by: Claude Fable 5 --- backend/AGENTS.md | 1 + backend/app/gateway/app.py | 4 + backend/app/gateway/routers/console.py | 492 ++++++++++++++++++ .../harness/deerflow/models/factory.py | 4 + .../harness/deerflow/runtime/journal.py | 25 +- .../deerflow/subagents/token_collector.py | 31 +- backend/tests/test_console_router.py | 289 ++++++++++ backend/tests/test_model_factory.py | 17 + backend/tests/test_token_usage_by_model.py | 136 +++++ config.example.yaml | 11 + 10 files changed, 998 insertions(+), 12 deletions(-) create mode 100644 backend/app/gateway/routers/console.py create mode 100644 backend/tests/test_console_router.py diff --git a/backend/AGENTS.md b/backend/AGENTS.md index dd954f743..1d06e3013 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -298,6 +298,7 @@ CORS is same-origin by default when requests enter through nginx on port 2026. S |--------|-----------| | **Models** (`/api/models`) | `GET /` - list models; `GET /{name}` - model details | | **Features** (`/api/features`) | `GET /` - report config-gated feature availability (currently `agents_api.enabled`) for frontend UI gating | +| **Console** (`/api/console`) | Read-only cross-thread observability for the current user (the data layer for an operations dashboard or external monitoring): `GET /stats` - headline counters (runs/threads/agents/tokens/cost); `GET /runs` - paginated run history joined with thread titles (per-run cost); `GET /usage` - zero-filled daily token series + per-model breakdown with spend. Queries `runs`/`threads_meta` directly as a reporting layer (no new `RunStore` methods); requires a SQL database backend — returns 503 on `database.backend: memory`. Real-cost estimation reads optional `models[*].pricing` (`currency`, `input_per_million`, `output_per_million`, `input_cache_hit_per_million`; `ModelConfig` is `extra="allow"`, so no schema change) and prices each run from its `token_usage_by_model` input/output split. Pricing is **cache-aware**: `RunJournal` accumulates prompt-cache hits from `usage_metadata.input_token_details.cache_read` into a sparse `cache_read_tokens` bucket key (also threaded through `SubagentTokenCollector` → `record_external_llm_usage_records`), and cache-hit input tokens are billed at `input_cache_hit_per_million` (omitted → billed at the miss price, a conservative upper bound). Legacy rows fall back to run-level totals at `model_name`; unpriced models yield `cost: null` and cost fields are null when no pricing is configured | | **MCP** (`/api/mcp`) | `GET /config` - get config; `PUT /config` - update config (saves to extensions_config.json) | | **Skills** (`/api/skills`) | `GET /` - list skills; `GET /{name}` - details; `PUT /{name}` - update enabled; `POST /install` - install from .skill archive (accepts standard optional frontmatter like `version`, `author`, `compatibility`) | | **Memory** (`/api/memory`) | `GET /` - memory data; `POST /reload` - force reload; `GET /config` - config; `GET /status` - config + data | diff --git a/backend/app/gateway/app.py b/backend/app/gateway/app.py index 8c9830de8..fd1c71b72 100644 --- a/backend/app/gateway/app.py +++ b/backend/app/gateway/app.py @@ -18,6 +18,7 @@ from app.gateway.routers import ( auth, channel_connections, channels, + console, features, feedback, github_webhooks, @@ -414,6 +415,9 @@ This gateway provides runtime endpoints for agent runs plus custom endpoints for # Features API is mounted at /api/features app.include_router(features.router) + # Console API (cross-thread observability) is mounted at /api/console + app.include_router(console.router) + # MCP API is mounted at /api/mcp app.include_router(mcp.router) diff --git a/backend/app/gateway/routers/console.py b/backend/app/gateway/routers/console.py new file mode 100644 index 000000000..fbdef19f9 --- /dev/null +++ b/backend/app/gateway/routers/console.py @@ -0,0 +1,492 @@ +"""Read-only operations-console endpoints. + +Aggregates observability data across all of the current user's threads: run +history, token spend over time, and asset counts — the data layer for an +operations dashboard or any external monitoring consumer. + +This is a reporting layer, not a runtime path: it issues short-lived read-only +queries against the harness-owned ``runs`` / ``threads_meta`` tables instead of +widening the runtime ``RunStore`` surface. Requires a SQL database backend +(``database.backend: sqlite | postgres``); returns 503 on the memory backend, +which persists no run history to report on. +""" + +import asyncio +import logging +from datetime import UTC, datetime, time, timedelta +from typing import NamedTuple + +from fastapi import APIRouter, HTTPException, Query, Request +from pydantic import BaseModel, Field +from sqlalchemy import func, select + +from app.gateway.authz import require_permission +from app.gateway.deps import get_current_user +from deerflow.config import get_app_config +from deerflow.config.agents_config import list_custom_agents +from deerflow.persistence.engine import get_session_factory +from deerflow.persistence.run.model import RunRow +from deerflow.persistence.thread_meta.model import ThreadMetaRow + +logger = logging.getLogger(__name__) +router = APIRouter(prefix="/api/console", tags=["console"]) + +_ACTIVE_STATUSES = ("pending", "running") +_FAILED_STATUSES = ("error", "timeout") + +# Cap the error excerpt in list responses; the full text stays on the run row. +_ERROR_EXCERPT_CHARS = 300 + + +# --------------------------------------------------------------------------- +# Response models +# --------------------------------------------------------------------------- + + +class ConsoleStatsResponse(BaseModel): + """Headline counters for the console dashboard.""" + + total_runs: int = Field(..., description="All recorded runs for the current user") + active_runs: int = Field(..., description="Runs currently pending or running") + failed_runs: int = Field(..., description="Runs that ended in error or timeout") + total_threads: int = Field(..., description="Conversation threads owned by the current user") + total_agents: int = Field(..., description="Custom agents owned by the current user") + total_tokens: int = Field(..., description="Tokens consumed across all recorded runs") + total_cost: float | None = Field(default=None, description="Estimated spend across priced runs; null when no models[*].pricing is configured") + currency: str | None = Field(default=None, description="Display currency taken from the first configured pricing entry") + + +class ConsoleRunItem(BaseModel): + """One run in the cross-thread run listing.""" + + run_id: str + thread_id: str + thread_title: str | None = Field(default=None, description="Display name from threads_meta, if tracked") + assistant_id: str | None = None + status: str + model_name: str | None = None + created_at: datetime | None = None + updated_at: datetime | None = None + duration_seconds: float | None = Field(default=None, description="Wall-clock duration; live elapsed time for active runs") + total_tokens: int = 0 + message_count: int = 0 + cost: float | None = Field(default=None, description="Estimated spend for this run; null when its models are unpriced") + error: str | None = Field(default=None, description="Error excerpt for failed runs") + + +class ConsoleRunsResponse(BaseModel): + """Paginated cross-thread run listing, newest first.""" + + runs: list[ConsoleRunItem] + has_more: bool + + +class ConsoleUsageDay(BaseModel): + """Token usage aggregated over one local-time day.""" + + date: str = Field(..., description="Local date (YYYY-MM-DD) per the requested tz offset") + total_tokens: int = 0 + input_tokens: int = 0 + output_tokens: int = 0 + runs: int = 0 + cost: float = Field(default=0.0, description="Estimated spend for the day across priced runs") + + +class ConsoleUsageModelBreakdown(BaseModel): + """Token usage attributed to one model.""" + + tokens: int = 0 + runs: int = Field(default=0, description="Runs that used this model (non-exclusive)") + cost: float | None = Field(default=None, description="Estimated spend for this model; null when unpriced") + input_tokens: int = Field(default=0, description="Input tokens attributed to this model") + cache_read_tokens: int = Field(default=0, description="Prompt-cache-hit input tokens attributed to this model") + + +class ConsoleUsageResponse(BaseModel): + """Daily token-usage series plus per-model breakdown for the window.""" + + days: list[ConsoleUsageDay] + by_model: dict[str, ConsoleUsageModelBreakdown] + total_tokens: int + total_runs: int + total_cost: float | None = Field(default=None, description="Estimated spend for the window; null when no pricing is configured") + currency: str | None = Field(default=None, description="Display currency taken from the first configured pricing entry") + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _session_factory_or_503(): + sf = get_session_factory() + if sf is None: + raise HTTPException( + status_code=503, + detail="Console requires a SQL database backend; set database.backend to sqlite or postgres in config.yaml.", + ) + return sf + + +def _as_utc(dt: datetime | None) -> datetime | None: + """Normalize DB timestamps: SQLite round-trips them naive, Postgres aware.""" + if dt is None: + return None + return dt.replace(tzinfo=UTC) if dt.tzinfo is None else dt + + +# --------------------------------------------------------------------------- +# Pricing — real spend estimation +# --------------------------------------------------------------------------- + + +class _ModelPricing(NamedTuple): + input_per_million: float + output_per_million: float + currency: str + # Price for prompt-cache-hit input tokens. None → hits are billed at the + # full input price (conservative upper bound for providers that don't + # discount, or when the operator hasn't configured the hit price). + input_cache_hit_per_million: float | None = None + + +def _build_pricing_map() -> dict[str, _ModelPricing]: + """Collect per-model prices from ``models[*].pricing`` in config.yaml. + + ``ModelConfig`` allows extra fields, so operators can annotate each model + with e.g. ``pricing: {currency: CNY, input_per_million: 8, + output_per_million: 32, input_cache_hit_per_million: 0.8}`` without any + schema change. Entries are keyed by both the config ``name`` and the + provider ``model`` id (plus lowercase variants), because + ``token_usage_by_model`` buckets carry the provider-reported model name. + """ + try: + models = get_app_config().models + except Exception: # pragma: no cover - defensive: cost display must not break the console + logger.warning("console: failed to load model pricing from config", exc_info=True) + return {} + + pricing: dict[str, _ModelPricing] = {} + for model_cfg in models or []: + raw = getattr(model_cfg, "pricing", None) + if not isinstance(raw, dict): + continue + try: + input_price = float(raw.get("input_per_million") or 0) + output_price = float(raw.get("output_per_million") or 0) + raw_hit_price = raw.get("input_cache_hit_per_million") + cache_hit_price = float(raw_hit_price) if raw_hit_price is not None else None + except (TypeError, ValueError): + logger.warning("console: ignoring malformed pricing on model %s", model_cfg.name) + continue + if input_price <= 0 and output_price <= 0: + continue + currency = str(raw.get("currency") or "USD").upper() + entry = _ModelPricing(input_price, output_price, currency, cache_hit_price) + for key in (model_cfg.name, getattr(model_cfg, "model", None)): + if key: + pricing.setdefault(key, entry) + pricing.setdefault(key.lower(), entry) + return pricing + + +def _pricing_currency(pricing: dict[str, _ModelPricing]) -> str | None: + """Display currency: the first configured entry's (one currency per deployment).""" + return next(iter(pricing.values())).currency if pricing else None + + +def _lookup_pricing(pricing: dict[str, _ModelPricing], model: str | None) -> _ModelPricing | None: + if not model: + return None + return pricing.get(model) or pricing.get(model.lower()) + + +def _token_cost(input_tokens: int, output_tokens: int, price: _ModelPricing, cache_read_tokens: int = 0) -> float: + """Cache-aware spend: cache-hit input tokens are billed at the hit price. + + ``cache_read_tokens`` is clamped into ``[0, input_tokens]``; the remainder + is billed at the full (cache-miss) input price. Without a configured hit + price all input is billed at the miss price. + """ + cache_read = min(max(int(cache_read_tokens or 0), 0), max(int(input_tokens or 0), 0)) + uncached = max(int(input_tokens or 0), 0) - cache_read + hit_price = price.input_cache_hit_per_million if price.input_cache_hit_per_million is not None else price.input_per_million + return (uncached / 1_000_000) * price.input_per_million + (cache_read / 1_000_000) * hit_price + (output_tokens / 1_000_000) * price.output_per_million + + +def _run_cost( + pricing: dict[str, _ModelPricing], + *, + model_name: str | None, + total_input_tokens: int | None, + total_output_tokens: int | None, + token_usage_by_model: dict | None, +) -> float | None: + """Estimate one run's spend, or None when none of its models are priced. + + Prefers the per-model breakdown (accurate for multi-model runs, e.g. + subagents on a different model); falls back to run-level totals priced at + ``model_name`` for legacy rows. Buckets without an input/output split are + skipped rather than guessed. + """ + cost = 0.0 + priced = False + if isinstance(token_usage_by_model, dict): + for model, usage in token_usage_by_model.items(): + if not isinstance(usage, dict): + continue + price = _lookup_pricing(pricing, model) + if price is None: + continue + input_tokens = int(usage.get("input_tokens") or 0) + output_tokens = int(usage.get("output_tokens") or 0) + if input_tokens == 0 and output_tokens == 0: + continue + cost += _token_cost(input_tokens, output_tokens, price, int(usage.get("cache_read_tokens") or 0)) + priced = True + if priced: + return cost + price = _lookup_pricing(pricing, model_name) + if price is None: + return None + input_tokens = int(total_input_tokens or 0) + output_tokens = int(total_output_tokens or 0) + if input_tokens == 0 and output_tokens == 0: + return None + return _token_cost(input_tokens, output_tokens, price) + + +# --------------------------------------------------------------------------- +# Endpoints +# --------------------------------------------------------------------------- + + +@router.get( + "/stats", + response_model=ConsoleStatsResponse, + summary="Console Stats", + description="Headline counters (runs, threads, agents, tokens) scoped to the current user.", +) +@require_permission("runs", "read") +async def console_stats(request: Request) -> ConsoleStatsResponse: + """Return the dashboard's headline counters.""" + sf = _session_factory_or_503() + user_id = await get_current_user(request) + run_where = (RunRow.user_id == user_id,) if user_id else () + thread_where = (ThreadMetaRow.user_id == user_id,) if user_id else () + + pricing = _build_pricing_map() + + async with sf() as session: + total_runs = await session.scalar(select(func.count()).select_from(RunRow).where(*run_where)) or 0 + active_runs = await session.scalar(select(func.count()).select_from(RunRow).where(RunRow.status.in_(_ACTIVE_STATUSES), *run_where)) or 0 + failed_runs = await session.scalar(select(func.count()).select_from(RunRow).where(RunRow.status.in_(_FAILED_STATUSES), *run_where)) or 0 + total_tokens = await session.scalar(select(func.coalesce(func.sum(RunRow.total_tokens), 0)).where(*run_where)) or 0 + total_threads = await session.scalar(select(func.count()).select_from(ThreadMetaRow).where(*thread_where)) or 0 + + total_cost: float | None = None + if pricing: + cost_rows = ( + await session.execute( + select( + RunRow.model_name, + RunRow.total_input_tokens, + RunRow.total_output_tokens, + RunRow.token_usage_by_model, + ).where(*run_where) + ) + ).all() + cost_sum = 0.0 + for model_name, input_tokens, output_tokens, usage_map in cost_rows: + cost = _run_cost( + pricing, + model_name=model_name, + total_input_tokens=input_tokens, + total_output_tokens=output_tokens, + token_usage_by_model=usage_map, + ) + if cost is not None: + cost_sum += cost + total_cost = round(cost_sum, 6) + + try: + # Filesystem scan; resolves the effective user internally (AuthMiddleware + # sets the context for real requests, "default" in no-auth mode). + agents = await asyncio.to_thread(list_custom_agents) + total_agents = len(agents) + except Exception: # pragma: no cover - defensive: stats must not 500 on a bad agents dir + logger.warning("console_stats: failed to list custom agents", exc_info=True) + total_agents = 0 + + return ConsoleStatsResponse( + total_runs=total_runs, + active_runs=active_runs, + failed_runs=failed_runs, + total_threads=total_threads, + total_agents=total_agents, + total_tokens=total_tokens, + total_cost=total_cost, + currency=_pricing_currency(pricing), + ) + + +@router.get( + "/runs", + response_model=ConsoleRunsResponse, + summary="List Runs Across Threads", + description="Cross-thread run history for the current user, newest first, joined with thread titles.", +) +@require_permission("runs", "read") +async def console_runs( + request: Request, + limit: int = Query(default=20, ge=1, le=100), + offset: int = Query(default=0, ge=0), + status: str | None = Query(default=None, description="Filter by run status (e.g. running, success, error)"), +) -> ConsoleRunsResponse: + """Return a page of the user's runs across all threads.""" + sf = _session_factory_or_503() + user_id = await get_current_user(request) + + stmt = select(RunRow, ThreadMetaRow.display_name).join(ThreadMetaRow, ThreadMetaRow.thread_id == RunRow.thread_id, isouter=True).order_by(RunRow.created_at.desc(), RunRow.run_id.desc()).limit(limit + 1).offset(offset) + if user_id: + stmt = stmt.where(RunRow.user_id == user_id) + if status: + stmt = stmt.where(RunRow.status == status) + + async with sf() as session: + rows = (await session.execute(stmt)).all() + + pricing = _build_pricing_map() + has_more = len(rows) > limit + now = datetime.now(UTC) + items: list[ConsoleRunItem] = [] + for row, display_name in rows[:limit]: + created = _as_utc(row.created_at) + updated = _as_utc(row.updated_at) + if row.status in _ACTIVE_STATUSES: + duration = (now - created).total_seconds() if created else None + else: + duration = (updated - created).total_seconds() if created and updated else None + cost = _run_cost( + pricing, + model_name=row.model_name, + total_input_tokens=row.total_input_tokens, + total_output_tokens=row.total_output_tokens, + token_usage_by_model=row.token_usage_by_model, + ) + items.append( + ConsoleRunItem( + run_id=row.run_id, + thread_id=row.thread_id, + thread_title=display_name, + assistant_id=row.assistant_id, + status=row.status, + model_name=row.model_name, + created_at=created, + updated_at=updated, + duration_seconds=max(duration, 0.0) if duration is not None else None, + total_tokens=row.total_tokens or 0, + message_count=row.message_count or 0, + cost=round(cost, 6) if cost is not None else None, + error=row.error[:_ERROR_EXCERPT_CHARS] if row.error else None, + ) + ) + return ConsoleRunsResponse(runs=items, has_more=has_more) + + +@router.get( + "/usage", + response_model=ConsoleUsageResponse, + summary="Token Usage Over Time", + description="Daily token-usage series (zero-filled) plus per-model breakdown over the requested window.", +) +@require_permission("runs", "read") +async def console_usage( + request: Request, + days: int = Query(default=14, ge=1, le=90), + tz_offset_minutes: int = Query(default=0, ge=-840, le=840, description="Local-time offset from UTC for day bucketing"), +) -> ConsoleUsageResponse: + """Aggregate token usage by local day and by model.""" + sf = _session_factory_or_503() + user_id = await get_current_user(request) + + tz_delta = timedelta(minutes=tz_offset_minutes) + today_local = (datetime.now(UTC) + tz_delta).date() + start_local = today_local - timedelta(days=days - 1) + window_start_utc = datetime.combine(start_local, time.min, tzinfo=UTC) - tz_delta + + stmt = select(RunRow).where(RunRow.created_at >= window_start_utc) + if user_id: + stmt = stmt.where(RunRow.user_id == user_id) + + async with sf() as session: + rows = (await session.execute(stmt)).scalars().all() + + day_buckets: dict[str, ConsoleUsageDay] = {} + for i in range(days): + d = (start_local + timedelta(days=i)).isoformat() + day_buckets[d] = ConsoleUsageDay(date=d) + + pricing = _build_pricing_map() + by_model: dict[str, ConsoleUsageModelBreakdown] = {} + total_tokens = 0 + total_runs = 0 + total_cost = 0.0 if pricing else None + for row in rows: + created = _as_utc(row.created_at) + if created is None: + continue + local_date = ((created + tz_delta).date()).isoformat() + bucket = day_buckets.get(local_date) + if bucket is None: + # Row sits just outside the local window (UTC-window over-fetch); skip. + continue + run_tokens = row.total_tokens or 0 + bucket.total_tokens += run_tokens + bucket.input_tokens += row.total_input_tokens or 0 + bucket.output_tokens += row.total_output_tokens or 0 + bucket.runs += 1 + total_tokens += run_tokens + total_runs += 1 + + run_cost = _run_cost( + pricing, + model_name=row.model_name, + total_input_tokens=row.total_input_tokens, + total_output_tokens=row.total_output_tokens, + token_usage_by_model=row.token_usage_by_model, + ) + if run_cost is not None and total_cost is not None: + bucket.cost = round(bucket.cost + run_cost, 6) + total_cost = round(total_cost + run_cost, 6) + + usage_map = row.token_usage_by_model or {} + if isinstance(usage_map, dict) and usage_map: + for model, usage in usage_map.items(): + entry = by_model.setdefault(model, ConsoleUsageModelBreakdown()) + entry.runs += 1 + if not isinstance(usage, dict): + continue + entry.tokens += int(usage.get("total_tokens", 0) or 0) + entry.input_tokens += int(usage.get("input_tokens") or 0) + entry.cache_read_tokens += int(usage.get("cache_read_tokens") or 0) + price = _lookup_pricing(pricing, model) + if price is not None: + model_cost = _token_cost(int(usage.get("input_tokens") or 0), int(usage.get("output_tokens") or 0), price, int(usage.get("cache_read_tokens") or 0)) + entry.cost = round((entry.cost or 0.0) + model_cost, 6) + elif row.model_name and run_tokens > 0: + # Legacy rows predating token_usage_by_model: fall back to the run's model. + entry = by_model.setdefault(row.model_name, ConsoleUsageModelBreakdown()) + entry.tokens += run_tokens + entry.runs += 1 + if run_cost is not None: + entry.cost = round((entry.cost or 0.0) + run_cost, 6) + + return ConsoleUsageResponse( + days=list(day_buckets.values()), + by_model=by_model, + total_tokens=total_tokens, + total_runs=total_runs, + total_cost=total_cost, + currency=_pricing_currency(pricing), + ) diff --git a/backend/packages/harness/deerflow/models/factory.py b/backend/packages/harness/deerflow/models/factory.py index 57eb15001..dce3ca1ba 100644 --- a/backend/packages/harness/deerflow/models/factory.py +++ b/backend/packages/harness/deerflow/models/factory.py @@ -121,6 +121,10 @@ def create_chat_model(name: str | None = None, thinking_enabled: bool = False, * "when_thinking_disabled", "thinking", "supports_vision", + # Presentation-only metadata (consumed by the console's cost + # display) — must never reach the provider client, which would + # forward unknown kwargs into the completion request payload. + "pricing", }, ) # Compute effective when_thinking_enabled by merging in the `thinking` shortcut field. diff --git a/backend/packages/harness/deerflow/runtime/journal.py b/backend/packages/harness/deerflow/runtime/journal.py index 9875595d3..7d7b7d4d6 100644 --- a/backend/packages/harness/deerflow/runtime/journal.py +++ b/backend/packages/harness/deerflow/runtime/journal.py @@ -317,7 +317,7 @@ class RunJournal(BaseCallbackHandler): per_call_model: str | None = None if isinstance(response_metadata, Mapping): per_call_model = response_metadata.get("model_name") or response_metadata.get("model") - self._record_model_usage(per_call_model, input_tk, output_tk, total_tk) + self._record_model_usage(per_call_model, input_tk, output_tk, total_tk, self._extract_cache_read(usage_dict)) self._schedule_progress_flush() @@ -433,12 +433,18 @@ class RunJournal(BaseCallbackHandler): input_tokens: int, output_tokens: int, total_tokens: int, + cache_read_tokens: int = 0, ) -> None: """Add a single LLM call's token usage to the per-model accumulator. Missing / empty ``model_name`` collapses into a shared ``"unknown"`` bucket so the breakdown stays usable when a provider doesn't surface ``response_metadata.model_name``. + + ``cache_read_tokens`` (prompt-cache hits, from + ``usage_metadata.input_token_details.cache_read``) is stored as a + sparse bucket key — only written when non-zero — so buckets from + providers without cache reporting keep their historical shape. """ if total_tokens <= 0: return @@ -449,6 +455,19 @@ class RunJournal(BaseCallbackHandler): bucket["input_tokens"] += int(input_tokens or 0) bucket["output_tokens"] += int(output_tokens or 0) bucket["total_tokens"] += int(total_tokens) + if cache_read_tokens > 0: + bucket["cache_read_tokens"] = bucket.get("cache_read_tokens", 0) + int(cache_read_tokens) + + @staticmethod + def _extract_cache_read(usage_dict: dict) -> int: + """Prompt-cache-hit input tokens from LangChain's normalized usage.""" + details = usage_dict.get("input_token_details") or {} + if not isinstance(details, Mapping): + return 0 + try: + return max(int(details.get("cache_read") or 0), 0) + except (TypeError, ValueError): + return 0 # -- Public methods (called by worker) -- @@ -466,6 +485,7 @@ class RunJournal(BaseCallbackHandler): input_tokens: Input token count output_tokens: Output token count total_tokens: Total token count (computed from input+output if 0/missing) + cache_read_tokens: Optional prompt-cache-hit input tokens """ if not self._track_tokens: return @@ -500,7 +520,8 @@ class RunJournal(BaseCallbackHandler): else: self._lead_agent_tokens += total_tk - self._record_model_usage(record.get("model_name"), input_tk, output_tk, total_tk) + cache_read_tk = record.get("cache_read_tokens", 0) or 0 + self._record_model_usage(record.get("model_name"), input_tk, output_tk, total_tk, int(cache_read_tk)) self._schedule_progress_flush() diff --git a/backend/packages/harness/deerflow/subagents/token_collector.py b/backend/packages/harness/deerflow/subagents/token_collector.py index 54ddcb946..642236c41 100644 --- a/backend/packages/harness/deerflow/subagents/token_collector.py +++ b/backend/packages/harness/deerflow/subagents/token_collector.py @@ -47,6 +47,14 @@ class SubagentTokenCollector(BaseCallbackHandler): total_tk = input_tk + output_tk if total_tk <= 0: continue + # Prompt-cache hits (needed for cache-aware cost accounting) + details = usage_dict.get("input_token_details") or {} + cache_read_tk = 0 + if isinstance(details, Mapping): + try: + cache_read_tk = max(int(details.get("cache_read") or 0), 0) + except (TypeError, ValueError): + cache_read_tk = 0 # Capture the model that actually produced this response so the # parent journal can bucket tokens by real model rather than the # lead agent's resolved model @@ -55,16 +63,19 @@ class SubagentTokenCollector(BaseCallbackHandler): if isinstance(response_metadata, Mapping): model_name = response_metadata.get("model_name") or response_metadata.get("model") self._counted_run_ids.add(rid) - self._records.append( - { - "source_run_id": rid, - "caller": self.caller, - "model_name": model_name, - "input_tokens": input_tk, - "output_tokens": output_tk, - "total_tokens": total_tk, - } - ) + record: dict[str, int | str | None] = { + "source_run_id": rid, + "caller": self.caller, + "model_name": model_name, + "input_tokens": input_tk, + "output_tokens": output_tk, + "total_tokens": total_tk, + } + # Sparse, matching the journal's per-model buckets: the key is + # only present when the provider actually reported cache hits. + if cache_read_tk > 0: + record["cache_read_tokens"] = cache_read_tk + self._records.append(record) return def snapshot_records(self) -> list[dict[str, int | str | None]]: diff --git a/backend/tests/test_console_router.py b/backend/tests/test_console_router.py new file mode 100644 index 000000000..c85be34d1 --- /dev/null +++ b/backend/tests/test_console_router.py @@ -0,0 +1,289 @@ +"""Tests for the console router (cross-thread observability endpoints). + +Covers: +1. /api/console/stats — headline counters +2. /api/console/runs — cross-thread listing, thread-title join, pagination, status filter +3. /api/console/usage — daily zero-filled buckets + per-model breakdown (incl. legacy fallback) +4. user scoping — rows filtered when the request resolves to a user +5. 503 when no SQL session factory is available (memory backend) + +Uses a real temp-file SQLite database (NullPool, so seeding in one event loop +and serving TestClient requests in another never share a connection). +""" + +import asyncio +from datetime import UTC, datetime, timedelta +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest +from _router_auth_helpers import make_authed_test_app +from fastapi.testclient import TestClient +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine +from sqlalchemy.pool import NullPool + +from app.gateway.routers import console +from deerflow.persistence.base import Base +from deerflow.persistence.run.model import RunRow +from deerflow.persistence.thread_meta.model import ThreadMetaRow + +NOW = datetime.now(UTC) + + +def _seed_rows() -> tuple[list[ThreadMetaRow], list[RunRow]]: + threads = [ + ThreadMetaRow(thread_id="t1", user_id="user-a", display_name="调研鹿角再生"), + ThreadMetaRow(thread_id="t2", user_id="user-a", display_name="Card assistant chat"), + ] + runs = [ + RunRow( + run_id="r1", + thread_id="t1", + user_id="user-a", + status="success", + model_name="minimax-m2", + total_tokens=1200, + total_input_tokens=800, + total_output_tokens=400, + token_usage_by_model={"minimax-m2": {"input_tokens": 800, "output_tokens": 400, "total_tokens": 1200, "cache_read_tokens": 500}}, + message_count=4, + created_at=NOW - timedelta(hours=1), + updated_at=NOW - timedelta(hours=1) + timedelta(seconds=30), + ), + RunRow( + run_id="r2", + thread_id="t1", + user_id="user-a", + status="running", + model_name="minimax-m2", + total_tokens=300, + total_input_tokens=200, + total_output_tokens=100, + token_usage_by_model={"minimax-m2": {"input_tokens": 200, "output_tokens": 100, "total_tokens": 300}}, + message_count=1, + created_at=NOW - timedelta(seconds=60), + updated_at=NOW - timedelta(seconds=1), + ), + RunRow( + run_id="r3", + thread_id="t2", + user_id="user-a", + status="error", + model_name="gpt-x", + error="Boom: provider exploded", + total_tokens=50, + total_input_tokens=50, + total_output_tokens=0, + token_usage_by_model={}, # legacy row → model_name fallback path + message_count=2, + created_at=NOW - timedelta(days=1), + updated_at=NOW - timedelta(days=1) + timedelta(seconds=5), + ), + RunRow( + run_id="r4", + thread_id="t2", + user_id="user-a", + status="success", + model_name="minimax-m2", + total_tokens=999, + token_usage_by_model={"minimax-m2": {"input_tokens": 600, "output_tokens": 399, "total_tokens": 999}}, + created_at=NOW - timedelta(days=40), # outside the default usage window + updated_at=NOW - timedelta(days=40) + timedelta(seconds=10), + ), + RunRow( + run_id="r5", + thread_id="t3", # no threads_meta row → exercises the outer join + user_id="user-b", + status="success", + model_name="qwen", + total_tokens=70, + total_input_tokens=40, + total_output_tokens=30, + token_usage_by_model={"qwen": {"input_tokens": 40, "output_tokens": 30, "total_tokens": 70}}, + created_at=NOW - timedelta(hours=2), + updated_at=NOW - timedelta(hours=2) + timedelta(seconds=8), + ), + ] + return threads, runs + + +@pytest.fixture() +def session_factory(tmp_path): + engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'console.db'}", poolclass=NullPool) + sf = async_sessionmaker(engine, expire_on_commit=False) + + async def _setup() -> None: + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + threads, runs = _seed_rows() + async with sf() as session: + session.add_all(threads) + session.add_all(runs) + await session.commit() + + asyncio.run(_setup()) + yield sf + asyncio.run(engine.dispose()) + + +@pytest.fixture() +def client(session_factory, monkeypatch): + monkeypatch.setattr(console, "get_session_factory", lambda: session_factory) + monkeypatch.setattr(console, "get_current_user", AsyncMock(return_value=None)) + monkeypatch.setattr(console, "list_custom_agents", lambda: [object(), object()]) + # No pricing configured by default; TestPricing patches its own config. + monkeypatch.setattr(console, "get_app_config", lambda: SimpleNamespace(models=[])) + app = make_authed_test_app() + app.include_router(console.router) + return TestClient(app) + + +class TestConsoleStats: + def test_headline_counters(self, client): + resp = client.get("/api/console/stats") + assert resp.status_code == 200 + data = resp.json() + assert data["total_runs"] == 5 + assert data["active_runs"] == 1 # r2 running + assert data["failed_runs"] == 1 # r3 error + assert data["total_threads"] == 2 + assert data["total_agents"] == 2 + assert data["total_tokens"] == 1200 + 300 + 50 + 999 + 70 + + +class TestConsoleRuns: + def test_listing_orders_paginates_and_joins_titles(self, client): + resp = client.get("/api/console/runs", params={"limit": 3}) + assert resp.status_code == 200 + data = resp.json() + assert data["has_more"] is True + ids = [r["run_id"] for r in data["runs"]] + assert ids == ["r2", "r1", "r5"] # newest first + by_id = {r["run_id"]: r for r in data["runs"]} + assert by_id["r2"]["thread_title"] == "调研鹿角再生" + assert by_id["r5"]["thread_title"] is None # t3 has no threads_meta row + # Terminal run: duration from created→updated; active run: live elapsed > 0. + assert by_id["r1"]["duration_seconds"] == pytest.approx(30.0, abs=1.0) + assert by_id["r2"]["duration_seconds"] > 0 + + def test_offset_pagination(self, client): + resp = client.get("/api/console/runs", params={"limit": 3, "offset": 3}) + data = resp.json() + assert [r["run_id"] for r in data["runs"]] == ["r3", "r4"] + assert data["has_more"] is False + + def test_status_filter(self, client): + resp = client.get("/api/console/runs", params={"status": "error"}) + data = resp.json() + assert [r["run_id"] for r in data["runs"]] == ["r3"] + assert data["runs"][0]["error"].startswith("Boom") + + +class TestConsoleUsage: + def test_daily_buckets_and_model_breakdown(self, client): + resp = client.get("/api/console/usage", params={"days": 14}) + assert resp.status_code == 200 + data = resp.json() + assert len(data["days"]) == 14 + # r4 (40 days old) excluded from the window + assert data["total_runs"] == 4 + assert data["total_tokens"] == 1200 + 300 + 50 + 70 + # Zero-filled series sums to the window total + assert sum(d["total_tokens"] for d in data["days"]) == data["total_tokens"] + yesterday = (NOW - timedelta(days=1)).date().isoformat() + day_map = {d["date"]: d for d in data["days"]} + assert day_map[yesterday]["total_tokens"] == 50 + assert data["by_model"]["minimax-m2"]["tokens"] == 1500 + assert data["by_model"]["qwen"]["tokens"] == 70 + # Legacy row (empty token_usage_by_model) falls back to model_name + assert data["by_model"]["gpt-x"]["tokens"] == 50 + + def test_window_excludes_old_rows_but_stats_include_them(self, client): + usage = client.get("/api/console/usage", params={"days": 7}).json() + assert all(r != 999 for d in usage["days"] for r in [d["total_tokens"]]) + stats = client.get("/api/console/stats").json() + assert stats["total_tokens"] >= 999 + + +def _priced_config(*, cache_hit_price: float | None = 0.8): + pricing = {"currency": "CNY", "input_per_million": 8, "output_per_million": 32} + if cache_hit_price is not None: + pricing["input_cache_hit_per_million"] = cache_hit_price + return SimpleNamespace( + models=[ + SimpleNamespace(name="minimax-m2", model="MiniMax-M2", pricing=pricing), + SimpleNamespace(name="gpt-x", model="gpt-x-1", pricing=None), # explicitly unpriced + ] + ) + + +# Expected per-run costs at 8 (miss) / 0.8 (hit) / 32 (output) per million: +# r1: 500 of 800 input tokens were cache hits → 300*8 + 500*0.8 + 400*32 (µ¥) +_R1_COST_CACHED = 300 * 8e-6 + 500 * 0.8e-6 + 400 * 32e-6 # 0.0156 +_R1_COST_UNCACHED = 800 * 8e-6 + 400 * 32e-6 # 0.0192 (no hit price configured) +_R2_COST = 200 * 8e-6 + 100 * 32e-6 # 0.0048 (no cache hits recorded) +_R4_COST = 600 * 8e-6 + 399 * 32e-6 # 0.017568 + + +class TestPricing: + def test_costs_use_cache_hit_price(self, client, monkeypatch): + monkeypatch.setattr(console, "get_app_config", lambda: _priced_config()) + stats = client.get("/api/console/stats").json() + assert stats["currency"] == "CNY" + # r3 (gpt-x) and r5 (qwen) are unpriced and excluded. + assert stats["total_cost"] == pytest.approx(_R1_COST_CACHED + _R2_COST + _R4_COST) + + usage = client.get("/api/console/usage").json() + assert usage["currency"] == "CNY" + assert usage["total_cost"] == pytest.approx(_R1_COST_CACHED + _R2_COST) # r4 outside window + assert usage["by_model"]["minimax-m2"]["cost"] == pytest.approx(_R1_COST_CACHED + _R2_COST) + assert usage["by_model"]["minimax-m2"]["input_tokens"] == 1000 + assert usage["by_model"]["minimax-m2"]["cache_read_tokens"] == 500 + assert usage["by_model"]["gpt-x"]["cost"] is None + assert sum(d["cost"] for d in usage["days"]) == pytest.approx(_R1_COST_CACHED + _R2_COST) + + runs = client.get("/api/console/runs", params={"limit": 50}).json() + by_id = {r["run_id"]: r for r in runs["runs"]} + assert by_id["r1"]["cost"] == pytest.approx(_R1_COST_CACHED) + assert by_id["r3"]["cost"] is None # unpriced model + + def test_cache_hits_billed_at_miss_price_without_hit_price(self, client, monkeypatch): + """No input_cache_hit_per_million configured → conservative upper bound.""" + monkeypatch.setattr(console, "get_app_config", lambda: _priced_config(cache_hit_price=None)) + runs = client.get("/api/console/runs", params={"limit": 50}).json() + by_id = {r["run_id"]: r for r in runs["runs"]} + assert by_id["r1"]["cost"] == pytest.approx(_R1_COST_UNCACHED) + + def test_costs_null_without_pricing(self, client): + stats = client.get("/api/console/stats").json() + assert stats["total_cost"] is None + assert stats["currency"] is None + usage = client.get("/api/console/usage").json() + assert usage["total_cost"] is None + runs = client.get("/api/console/runs").json() + assert all(r["cost"] is None for r in runs["runs"]) + + +class TestUserScoping: + def test_rows_filtered_by_resolved_user(self, client, monkeypatch): + monkeypatch.setattr(console, "get_current_user", AsyncMock(return_value="user-a")) + stats = client.get("/api/console/stats").json() + assert stats["total_runs"] == 4 # r5 (user-b) excluded + assert stats["total_tokens"] == 1200 + 300 + 50 + 999 + runs = client.get("/api/console/runs", params={"limit": 50}).json() + assert all(r["run_id"] != "r5" for r in runs["runs"]) + usage = client.get("/api/console/usage").json() + assert "qwen" not in usage["by_model"] + + +class TestNoSqlBackend: + def test_503_when_memory_backend(self, session_factory, monkeypatch): + monkeypatch.setattr(console, "get_session_factory", lambda: None) + monkeypatch.setattr(console, "get_current_user", AsyncMock(return_value=None)) + app = make_authed_test_app() + app.include_router(console.router) + c = TestClient(app) + for path in ("/api/console/stats", "/api/console/runs", "/api/console/usage"): + resp = c.get(path) + assert resp.status_code == 503 + assert "SQL database backend" in resp.json()["detail"] diff --git a/backend/tests/test_model_factory.py b/backend/tests/test_model_factory.py index 87118d7da..706749341 100644 --- a/backend/tests/test_model_factory.py +++ b/backend/tests/test_model_factory.py @@ -103,6 +103,23 @@ def test_raises_when_model_not_found(monkeypatch): factory_module.create_chat_model(name="ghost-model") +def test_pricing_metadata_never_reaches_the_provider_client(monkeypatch): + """`models[*].pricing` is console-only metadata (issue: ChatOpenAI forwards + unknown kwargs into the completion request payload, so an un-stripped + `pricing` block breaks every live LLM call with + ``Completions.create() got an unexpected keyword argument 'pricing'``).""" + model = _make_model("priced") + # ModelConfig is extra="allow" — pricing rides along as an extra field. + model.pricing = {"currency": "CNY", "input_per_million": 8, "output_per_million": 32, "input_cache_hit_per_million": 0.8} + cfg = _make_app_config([model]) + _patch_factory(monkeypatch, cfg) + + FakeChatModel.captured_kwargs = {} + factory_module.create_chat_model(name="priced") + + assert "pricing" not in FakeChatModel.captured_kwargs + + def test_appends_all_tracing_callbacks(monkeypatch): cfg = _make_app_config([_make_model("alpha")]) _patch_factory(monkeypatch, cfg) diff --git a/backend/tests/test_token_usage_by_model.py b/backend/tests/test_token_usage_by_model.py index 7fa6aad62..867b021e9 100644 --- a/backend/tests/test_token_usage_by_model.py +++ b/backend/tests/test_token_usage_by_model.py @@ -429,3 +429,139 @@ async def test_include_active_picks_up_running_progress_snapshot(tmp_path): assert active["total_tokens"] == 70 finally: await _close_sql_engine() + + +# --------------------------------------------------------------------------- +# Prompt-cache-hit accounting (powers cache-aware cost estimation in +# /api/console): cache_read_tokens is a *sparse* bucket key — present only +# when a provider reported cache hits — so pre-existing bucket shapes and +# exact-equality assertions above stay valid. +# --------------------------------------------------------------------------- + + +class TestJournalCacheRead: + def test_cache_read_accumulates_as_sparse_key(self) -> None: + j = _journal() + j.on_llm_end( + _make_llm_response( + usage={"input_tokens": 100, "output_tokens": 10, "total_tokens": 110, "input_token_details": {"cache_read": 80}}, + model_name="m", + ), + run_id=uuid4(), + parent_run_id=None, + tags=["lead_agent"], + ) + # Second call without cache hits still accumulates into the same bucket. + j.on_llm_end( + _make_llm_response(usage={"input_tokens": 50, "output_tokens": 5, "total_tokens": 55}, model_name="m"), + run_id=uuid4(), + parent_run_id=None, + tags=["lead_agent"], + ) + data = j.get_completion_data() + assert data["token_usage_by_model"]["m"] == { + "input_tokens": 150, + "output_tokens": 15, + "total_tokens": 165, + "cache_read_tokens": 80, + } + + def test_bucket_without_cache_hits_keeps_legacy_shape(self) -> None: + j = _journal() + j.on_llm_end( + _make_llm_response(usage={"input_tokens": 10, "output_tokens": 5, "total_tokens": 15}, model_name="m"), + run_id=uuid4(), + parent_run_id=None, + tags=["lead_agent"], + ) + assert j.get_completion_data()["token_usage_by_model"]["m"] == { + "input_tokens": 10, + "output_tokens": 5, + "total_tokens": 15, + } + + def test_external_records_carry_cache_read(self) -> None: + j = _journal() + j.record_external_llm_usage_records( + [ + { + "source_run_id": "sub-1", + "caller": "subagent:general-purpose", + "model_name": "sub-m", + "input_tokens": 40, + "output_tokens": 10, + "total_tokens": 50, + "cache_read_tokens": 25, + }, + ], + ) + assert j.get_completion_data()["token_usage_by_model"]["sub-m"]["cache_read_tokens"] == 25 + + def test_deepseek_raw_usage_normalizes_to_cache_read(self) -> None: + """Pin the DeepSeek chat-completions shape end-to-end: the raw + ``prompt_tokens_details.cached_tokens`` field is what langchain-openai's + ``_create_usage_metadata`` normalizes into + ``input_token_details.cache_read`` (DeepSeek's top-level + ``prompt_cache_hit/miss_tokens`` are redundant aliases LangChain does + not read), and the journal captures it. The derived cache-miss count + (input − cache_read) must equal DeepSeek's own + ``prompt_cache_miss_tokens``, which is what cache-aware pricing bills + at the full input price.""" + from langchain_openai.chat_models.base import _create_usage_metadata + + raw = { + "prompt_tokens": 106, + "completion_tokens": 112, + "total_tokens": 218, + "prompt_tokens_details": {"cached_tokens": 64}, + "prompt_cache_hit_tokens": 64, + "prompt_cache_miss_tokens": 42, + } + usage = _create_usage_metadata(raw) + j = _journal() + j.on_llm_end( + _make_llm_response(usage=dict(usage), model_name="deepseek-chat"), + run_id=uuid4(), + parent_run_id=None, + tags=["lead_agent"], + ) + bucket = j.get_completion_data()["token_usage_by_model"]["deepseek-chat"] + assert bucket == { + "input_tokens": 106, + "output_tokens": 112, + "total_tokens": 218, + "cache_read_tokens": 64, + } + assert bucket["input_tokens"] - bucket["cache_read_tokens"] == raw["prompt_cache_miss_tokens"] + + def test_collector_extracts_cache_read_from_usage_metadata(self) -> None: + from deerflow.subagents.token_collector import SubagentTokenCollector + + collector = SubagentTokenCollector("subagent:general-purpose") + collector.on_llm_end( + _make_llm_response( + usage={"input_tokens": 30, "output_tokens": 6, "total_tokens": 36, "input_token_details": {"cache_read": 20}}, + model_name="sub-m", + ), + run_id=uuid4(), + ) + records = collector.snapshot_records() + assert len(records) == 1 + assert records[0]["cache_read_tokens"] == 20 + + def test_collector_omits_cache_read_key_when_no_cache_hits(self) -> None: + from deerflow.subagents.token_collector import SubagentTokenCollector + + collector = SubagentTokenCollector("subagent:general-purpose") + collector.on_llm_end( + _make_llm_response( + usage={"input_tokens": 30, "output_tokens": 6, "total_tokens": 36}, + model_name="sub-m", + ), + run_id=uuid4(), + ) + records = collector.snapshot_records() + assert len(records) == 1 + # Sparse record shape: no explicit 0 when the provider reported no + # cache hits (record_external_llm_usage_records treats absent as 0). + assert "cache_read_tokens" not in records[0] diff --git a/config.example.yaml b/config.example.yaml index 03db0f47b..288c1f10c 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -70,6 +70,17 @@ max_recursion_limit: 1000 # Models Configuration # ============================================================================ # Configure available LLM models for the agent to use +# +# Optional per-model pricing (powers the real-cost display on the workspace +# console). Add a `pricing` block to any model entry; use ONE currency across +# all models. Prices are per one million tokens. +# +# pricing: +# currency: CNY # ISO code shown in the console (CNY, USD, ...) +# input_per_million: 8.0 # price per 1M input tokens (cache miss) +# output_per_million: 32.0 # price per 1M output tokens +# input_cache_hit_per_million: 0.8 # price per 1M cache-hit input tokens +# # (optional; omit → hits billed at miss price) models: # Example: Volcengine (Doubao) model