"""A content-free ledger of the platform's own LLM calls. Helix bills the whole company for its AI spending — ``ai_spending`` ingests the vendor invoices, splits them by API key, and shows finance who burned what. The one consumer missing from that picture was Helix itself. Twenty-nine modules call the shared Bailian client, and until this module existed only two of them (``ai_labs`` and ``intelligence``) recorded a single token. So "how much of our own AI budget does the platform spend, and on which feature?" had no answer at all — not an imprecise one, none. Three properties, each a deliberate choice: * **Content-free.** Prompts and completions are never written here. The ledger holds *who called, which model, how many tokens, did it work* — enough to attribute spend and spot a runaway loop, and nothing that would turn an ops table into a place where employee support tickets or bounty reports quietly accumulate. There is no second, richer store an admin could be tempted to query instead; this is the only record, and it is content-free by construction. * **Tokens, not money.** There is no per-model price table in this repo, so this module does not invent one. It records what it can actually observe. The vendor bill in ``ai_spending`` remains the money truth; this ledger answers the orthogonal question the bill cannot — *which feature* burned the tokens. Anyone joining the two is making a proportional-split assumption and should say so. * **Recording never fails a call.** Every write happens in its own short-lived session and swallows its errors. A caller's transaction is never touched: the ledger cannot poison a flush, cannot roll back a support ticket, and cannot turn a working feature into a 500 because a bookkeeping row would not save. Losing a statistic is a rounding error; losing the user's finished work is not. This mirrors ``ai_labs.usage.record``, which made the same trade for the same reason. Attribution comes from ``purpose``: a short stable slug naming the feature, e.g. ``support.copilot`` or ``automation_exception.triage``. Callers that pass nothing are recorded under :data:`PURPOSE_UNKNOWN` rather than being dropped — an unattributed call is a real call, and hiding it would make the ledger's totals disagree with the vendor bill for a reason nobody could see. The admin surface shows that bucket explicitly so the gap is visible instead of silent. """ from __future__ import annotations import contextvars import logging import re import uuid from datetime import UTC, datetime logger = logging.getLogger(__name__) #: Where a call with no declared purpose is booked. Not dropped: see #: the module docstring — a silent gap makes the ledger disagree with #: the vendor bill for an invisible reason. PURPOSE_UNKNOWN = "unattributed" #: A purpose is a slug, not free text. It reaches a ``String(64)`` #: column and is grouped on in the admin read, so an unbounded or #: punctuation-bearing value would fragment the rollup into #: near-duplicate rows. Anything that does not match is normalised #: rather than rejected — refusing to record is worse than recording #: under a tidied name. _PURPOSE_RE = re.compile(r"^[a-z0-9_]+(?:\.[a-z0-9_]+)*$") _PURPOSE_MAX = 64 #: Set by a caller that cannot pass ``purpose=`` at construction time #: (an agent loop that builds its client deep inside a helper). The #: client reads this when its own ``purpose`` is unset. _purpose_var: contextvars.ContextVar[str | None] = contextvars.ContextVar( "helix_llm_purpose", default=None, ) def normalize_purpose(raw: str | None) -> str: """Return a storable purpose slug. Tidies rather than rejects: lowercases, replaces separators with ``_``, drops anything else, and truncates. A caller that passes ``"Support Copilot!"`` gets ``support_copilot``, not an exception in the middle of an LLM call. """ value = (raw or "").strip().lower() if not value: return PURPOSE_UNKNOWN if _PURPOSE_RE.match(value): return value[:_PURPOSE_MAX] cleaned = re.sub(r"[^a-z0-9_.]+", "_", value).strip("_.") cleaned = re.sub(r"_{2,}", "_", cleaned) return (cleaned or PURPOSE_UNKNOWN)[:_PURPOSE_MAX] def current_purpose() -> str | None: """The purpose set for this task, if any.""" return _purpose_var.get() def set_purpose(purpose: str | None) -> contextvars.Token[str | None]: """Declare the purpose for LLM calls made in this task. Returns the token so a caller can restore the previous value; use :class:`purpose_scope` instead unless you need manual control. """ return _purpose_var.set(purpose) class purpose_scope: # noqa: N801 — reads as a context manager at call sites """Attribute every LLM call inside this block to ``purpose``. For call sites that construct their client somewhere they cannot easily reach, e.g.:: with purpose_scope("automation_exception.triage"): result = await run_agent_loop(...) """ __slots__ = ("_purpose", "_token") def __init__(self, purpose: str) -> None: self._purpose = purpose self._token: contextvars.Token[str | None] | None = None def __enter__(self) -> str: self._token = _purpose_var.set(self._purpose) return self._purpose def __exit__(self, *exc: object) -> None: if self._token is not None: _purpose_var.reset(self._token) self._token = None def _as_positive_int(value: object) -> int | None: """Token counts only, or nothing. Vendors have returned ``0``, ``"0"``, ``None`` and absent for "I am not telling you", and a stored ``0`` is indistinguishable from a real zero-token call in a ``SUM``. NULL says "not reported", which is the truth. """ # ``bool`` first: it is a subclass of ``int``, so ``True`` would # otherwise be recorded as one token. if value is None or isinstance(value, bool): return None # Narrowed explicitly rather than relying on ``int()`` to raise: an # arbitrary object reaching here is a caller bug, and swallowing it # as "not reported" is the right outcome but should be a decision, # not an accident of exception handling. if not isinstance(value, int | float | str): return None try: number = int(value) except (TypeError, ValueError): return None return number if number > 0 else None async def record( *, purpose: str | None, model: str, region: str | None, operation: str, ok: bool, input_tokens: object = None, output_tokens: object = None, duration_ms: int | None = None, error_code: str | None = None, trace_id: str | None = None, actor_person_id: uuid.UUID | None = None, ) -> None: """Book one LLM call. Never raises. Opens its own session so the caller's transaction is untouched — including the case where the caller's transaction is already doomed. An LLM call that failed inside a failed flush still happened and still cost tokens; recording it on the caller's session would be a no-op at best and a second error at worst. """ # Imported here rather than at module scope: this module is # imported by the LLM client, which several domains import at # module level, and a top-level database import would drag the # engine into their import graph. from app.core.database import async_session_factory # noqa: PLC0415 from app.domains.bailian.models import LlmCallLedger # noqa: PLC0415 try: row = LlmCallLedger( purpose=normalize_purpose(purpose), model=(model or "")[:64] or "unknown", region=((region or "").strip() or None), operation=(operation or "")[:16] or "chat", ok=bool(ok), input_tokens=_as_positive_int(input_tokens), output_tokens=_as_positive_int(output_tokens), duration_ms=duration_ms if (duration_ms or 0) >= 0 else None, error_code=((error_code or "").strip()[:64] or None), trace_id=((trace_id or "").strip()[:64] or None), actor_person_id=actor_person_id, called_at=datetime.now(UTC), ) async with async_session_factory() as session: session.add(row) await session.commit() except Exception: # noqa: BLE001 — bookkeeping must not fail the call # Deliberately broad and deliberately swallowed. See the module # docstring: the alternative is failing a user's finished work # because a statistics row would not save. logger.warning( "llm ledger write failed (purpose=%s model=%s op=%s)", purpose, model, operation, exc_info=True, ) __all__ = [ "PURPOSE_UNKNOWN", "current_purpose", "normalize_purpose", "purpose_scope", "record", "set_purpose", ]