symfonic.core.observability

observability

symfonic.core.observability — Instrumentation hooks.

NODE_LABELS module-attribute

NODE_LABELS: dict[str, str] = {NodeName.REACT.value: 'React Loop', NodeName.METACOGNITION_CRITIC.value: 'Metacognition Critic', NodeName.METACOGNITION_SKILL_GATE.value: 'Skill-Gate Critic', NodeName.CONTEXT_WINDOW_SUMMARISER.value: 'Context Window Summariser', NodeName.INSIGHT_EXTRACTOR.value: 'Insight Extractor', NodeName.ENTITY_EXTRACTOR_LLM.value: 'Entity Extractor (LLM)', NodeName.CONSOLIDATION_EXTRACTOR.value: 'Consolidation Extractor', NodeName.PROCEDURAL_ROUTER.value: 'Procedural Router', NodeName.PROCEDURAL_EXTRACTOR_LLM.value: 'Procedural Extractor'}

UI-friendly labels for the canonical node_name set.

Adopters wiring an admin dashboard can surface these directly instead of synthesising labels at the UI layer::

from symfonic.core.observability import human_label

row_label = human_label(event.node_name)  # round-trips unknown names

NODE_NAMES module-attribute

NODE_NAMES: frozenset[str] = frozenset(n.value for n in NodeName)

Set of canonical node_name strings emitted by the framework.

Adopters can pre-populate dashboards, validate pricing-table coverage, or allocate per-node budget ceilings using this set::

from symfonic.core.observability import NODE_NAMES

for node_name in NODE_NAMES:
    budget_ceilings[node_name] = float(env.get(f"BUDGET_{node_name.upper()}", 0.50))

AuditLogEntry

Bases: BaseModel

Immutable audit event emitted for every destructive operation.

Fields intentionally kept small so both in-memory and Postgres implementations can store them without a schema migration dance. metadata is the escape hatch for per-action context (counts, previous value, confirmation token, ...).

action class-attribute instance-attribute

action: str = Field(..., min_length=1, max_length=64)

Free-form action code. Convention: snake_case verb + optional resource suffix, e.g. delete_memory, bulk_delete, export_data, erase_all.

resource_type class-attribute instance-attribute

resource_type: str = Field(..., min_length=1, max_length=64)

E.g. memory_node, edge, procedure, tenant.

to_dict

to_dict() -> dict[str, Any]

Return a JSON-safe dict representation.

Source code in src/symfonic/core/observability/audit.py

def to_dict(self) -> dict[str, Any]:
    """Return a JSON-safe dict representation."""
    return {
        "tenant_id": self.tenant_id,
        "user_id": self.user_id,
        "action": self.action,
        "resource_type": self.resource_type,
        "resource_id": self.resource_id,
        "timestamp": self.timestamp.isoformat(),
        "metadata": self.metadata,
        "ip_address": self.ip_address,
        "user_agent": self.user_agent,
    }

AuditLogger

Bases: Protocol

Destination for audit events.

Implementations MAY persist synchronously or asynchronously. The contract is "record this entry, or raise" — retry / batching is an implementation detail. Callers in the router layer shield failures (fire-and-forget) so a sick audit store never 500s a user request.

BudgetCheckResult dataclass

BudgetCheckResult(allowed: bool, reason: str | None = None, daily_used_usd: float = 0.0, daily_limit_usd: float | None = None, monthly_used_usd: float = 0.0, monthly_limit_usd: float | None = None)

Result of a pre-call budget check.

BudgetStore

Bases: Protocol

Optional persistence layer for daily aggregates.

Implementations should be idempotent on conflict — the tracker calls upsert_daily after every LLM call, so the underlying storage must handle UPSERT semantics (ON CONFLICT ... DO UPDATE).

The optional read_today / read_month methods hydrate the tracker's in-memory buckets on demand so that multiple app replicas converge on the same view of a tenant's cumulative usage. A store that only implements upsert_daily still works (writes only) — multi-replica correctness requires all three.

read_month async

read_month(tenant_id: str, year_month: str) -> BudgetUsage | None

Return the aggregated month-to-date usage for tenant_id.

year_month is the YYYY-MM key produced by :func:month_key. Optional — see :meth:read_today.

Source code in src/symfonic/core/observability/budget_models.py

async def read_month(  # pragma: no cover
    self, tenant_id: str, year_month: str,
) -> BudgetUsage | None:
    """Return the aggregated month-to-date usage for ``tenant_id``.

    ``year_month`` is the ``YYYY-MM`` key produced by
    :func:`month_key`.  Optional — see :meth:`read_today`.
    """
    ...

read_today async

read_today(tenant_id: str) -> BudgetUsage | None

Return today's usage row for tenant_id (None if absent).

Optional — implementations without this method get write-through semantics only (single-process correctness).

Source code in src/symfonic/core/observability/budget_models.py

async def read_today(  # pragma: no cover
    self, tenant_id: str,
) -> BudgetUsage | None:
    """Return today's usage row for ``tenant_id`` (None if absent).

    Optional — implementations without this method get write-through
    semantics only (single-process correctness).
    """
    ...

BudgetUsage dataclass

BudgetUsage(tenant_id: str, window: str, window_key: str, input_tokens: int = 0, output_tokens: int = 0, cached_tokens: int = 0, cache_write_tokens: int = 0, cost_usd: float = 0.0, request_count: int = 0, updated_at: datetime = (lambda: datetime.now(UTC))())

Cumulative usage for a single (tenant, window) pair.

ConversationMetricsCollector

ConversationMetricsCollector(metrics_sink: object = None)

Bases: MetricsQuiescenceMixin, MetricsAttemptMixin, MetricsExecutionMixin, MetricsQueryMixin

Aggregates token usage and cost per conversation (session_id).

Satisfies the CallbackHandler protocol via structural typing.

Shared hosts partition records by tenant while preserving public session IDs.

Source code in src/symfonic/core/observability/metrics.py

def __init__(self, metrics_sink: object = None) -> None:
    self._callback_barrier = CallbackBarrier()
    self._conversations: dict[tuple[str, str], ConversationRecord] = {}
    self._metrics_sink = metrics_sink
    self._attempt_bindings: dict[str, str] = {}
    # run_id → most-recent TokenCompositionEvent
    self._current_composition: dict[str, TokenCompositionEvent] = {}
    # run_id → conversation_id
    self._current_conversation: dict[str, str] = {}
    # run_id → tenant_id (required for TokenBudgetTracker fan-out)
    self._current_tenant: dict[str, str] = {}
    # run_id -> (root_run_id, immediate parent_run_id)
    self._current_lineage: dict[str, tuple[str, str | None]] = {}
    self._execution_events: deque[dict[str, object]] = deque(
        maxlen=MAX_CALL_RECORDS_PER_CONVERSATION * MAX_CONVERSATIONS
    )
    # Only rows not yet handed to the sink, indexed by their run.  The
    # retained deque serves reads; scanning its global history on every
    # live event made request cost grow with the worker's lifetime.
    self._pending_execution_events: dict[str, list[dict[str, object]]] = defaultdict(list)
    self._execution_sequence: dict[str, int] = defaultdict(int)

on_agent_end async

on_agent_end(event: AgentEndEvent) -> None

Increment turn count and purge run-scoped bookkeeping.

Source code in src/symfonic/core/observability/metrics.py

@guarded_callback
async def on_agent_end(self, event: AgentEndEvent) -> None:
    """Increment turn count and purge run-scoped bookkeeping."""
    self._finish_run(self._event_key(event))

on_llm_end async

on_llm_end(event: LLMEndEvent) -> None

Accumulate token/cost metrics into the matching conversation record.

Also fans the same usage out to the process-wide :class:TokenBudgetTracker (if one has been registered via set_budget_tracker) so per-tenant ceilings can fire on the NEXT call. Failures in the tracker must never break the LLM flow.

Source code in src/symfonic/core/observability/metrics.py

@guarded_callback
async def on_llm_end(self, event: LLMEndEvent) -> None:
    """Accumulate token/cost metrics into the matching conversation record.

    Also fans the same usage out to the process-wide
    :class:`TokenBudgetTracker` (if one has been registered via
    ``set_budget_tracker``) so per-tenant ceilings can fire on the
    NEXT call.  Failures in the tracker must never break the LLM flow.
    """
    key = self._event_key(event)
    usage = TokenUsage.from_dict(event.usage, event.model)
    conv_id = self._current_conversation.get(key, event.run_id)
    self._current_conversation.setdefault(key, conv_id)
    tenant_id = self._current_tenant.get(key, "")

    conv = self._conversation_for_run(key)
    self._evict_conversations_if_over_cap()
    conv.llm_calls += 1
    conv.input_tokens += usage.input_tokens
    conv.output_tokens += usage.output_tokens
    conv.cached_tokens += usage.cached_tokens
    conv.cost_usd = round(conv.cost_usd + usage.cost_usd, 6)
    conv.last_seen = datetime.now(UTC)

    # Fan usage into the budget tracker (optional / best-effort).
    # Budget tracking must never break the LLM flow, but failures
    # MUST be observable -- silent suppression previously hid real
    # tracker regressions for weeks.
    from symfonic.core.observability.budget import get_budget_tracker
    tracker = get_budget_tracker()
    if tracker is not None and conv.tenant_id:
        try:
            await tracker.record(
                conv.tenant_id,
                event.model,
                {
                    "input_tokens": usage.input_tokens,
                    "output_tokens": usage.output_tokens,
                    "cached_tokens": usage.cached_tokens,
                    "cache_write_tokens": getattr(
                        usage, "cache_creation_tokens", 0,
                    ),
                },
            )
        except Exception as exc:
            logger.warning(
                "budget tracker fan-out failed: %s", exc, exc_info=True,
            )

    comp = self._current_composition.get(key)
    root_run_id, parent_run_id = self._current_lineage.get(
        key, (event.run_id, None)
    )
    conv.call_records.append(
        LLMCallRecord(
            call_id=event.run_id,
            conversation_id=conv_id,
            model=event.model,
            # Real call-site discriminator emitted since v7.4.3.  Fall
            # back to "react" only when the emitter left it empty (pre-
            # v7.4.3 payloads) so the historical default is preserved.
            node=event.node_name or "react",
            input_tokens=usage.input_tokens,
            cached_tokens=usage.cached_tokens,
            output_tokens=usage.output_tokens,
            cost_usd=usage.cost_usd,
            # iteration 0 is the turn's opening call; later React hops
            # are tool follow-ups.  The event carries no explicit
            # call_type, so derive it from the iteration index.
            call_type="initial" if event.iteration_index == 0 else "tool_followup",
            timestamp=datetime.now(UTC),
            composition={
                "system_prompt": comp.system_prompt_tokens if comp else 0,
                "tool_definitions": comp.tool_definitions_tokens if comp else 0,
                "conversation_history": comp.conversation_history_tokens if comp else 0,
                "memory_context": comp.memory_context_tokens if comp else 0,
            },
            cache_creation_tokens=usage.cache_creation_tokens,
            reasoning_tokens=usage.reasoning_tokens,
            duration_ms=event.duration_ms,
            iteration_index=event.iteration_index,
            turn_index=event.turn_index,
            root_run_id=root_run_id,
            parent_run_id=parent_run_id,
        )
    )
    self.record_model_round(event, conv_id, tenant_id)
    self._persist_execution_events(key)

    # v8.8.0 — fan the same per-call detail into the durable metrics
    # store, if one is registered, so the admin dashboards survive a
    # worker restart.  Best-effort + synchronous O(1) enqueue (the
    # BufferedMetricsSink batches the actual DB I/O off the hot path);
    # a sink failure must NEVER break the LLM flow.
    sink = self._metrics_sink
    if sink is None:
        from symfonic.core.observability.metrics_store import get_metrics_store

        sink = get_metrics_store()
    if sink is not None:
        try:
            sink.record(_record_to_snapshot(conv.call_records[-1], conv.tenant_id))
        except Exception as exc:  # noqa: BLE001 - telemetry never propagates
            logger.warning("metrics sink fan-out failed: %s", exc, exc_info=True)

on_token_composition async

on_token_composition(event: TokenCompositionEvent) -> None

Cache the composition breakdown keyed by run_id.

Source code in src/symfonic/core/observability/metrics.py

@guarded_callback
async def on_token_composition(self, event: TokenCompositionEvent) -> None:
    """Cache the composition breakdown keyed by run_id."""
    self._current_composition[self._event_key(event)] = event

set_conversation

set_conversation(run_id: str, conversation_id: str) -> None

Associate a run_id with a conversation (session) id.

Call this before the agent run so metrics land under the right conversation bucket (e.g. use the session_id as conversation_id).

Source code in src/symfonic/core/observability/metrics.py

@guarded_binding
def set_conversation(self, run_id: str, conversation_id: str) -> None:
    """Associate a run_id with a conversation (session) id.

    Call this before the agent run so metrics land under the right
    conversation bucket (e.g. use the session_id as conversation_id).
    """
    self._callback_barrier.require_open(run_id, self._current_tenant.get(run_id, ""))
    self._current_conversation[run_id] = conversation_id
    for row in self._pending_execution_events.get(run_id, ()):
        if not row["conversation_id"]:
            row["conversation_id"] = conversation_id
    self._persist_execution_events(run_id)

set_lineage

set_lineage(run_id: str, *, root_run_id: str, parent_run_id: str | None = None) -> None

Associate one execution with its delegation root and parent.

Source code in src/symfonic/core/observability/metrics.py

@guarded_binding
def set_lineage(
    self,
    run_id: str,
    *,
    root_run_id: str,
    parent_run_id: str | None = None,
) -> None:
    """Associate one execution with its delegation root and parent."""
    self._callback_barrier.require_open(run_id, self._current_tenant.get(run_id, ""))
    self._current_lineage[run_id] = (root_run_id or run_id, parent_run_id)

set_tenant

set_tenant(run_id: str, tenant_id: str) -> None

Associate a run_id with a tenant_id.

Call this before the agent run so on_llm_end can fan usage into the registered TokenBudgetTracker. Without this the tracker never fires because the ConversationRecord's tenant_id stays empty.

Source code in src/symfonic/core/observability/metrics.py

@guarded_binding
def set_tenant(self, run_id: str, tenant_id: str) -> None:
    """Associate a run_id with a tenant_id.

    Call this before the agent run so ``on_llm_end`` can fan usage
    into the registered ``TokenBudgetTracker``.  Without this the
    tracker never fires because the ConversationRecord's tenant_id
    stays empty.
    """
    if tenant_id:
        self._callback_barrier.require_open(run_id, tenant_id)
        if self._current_tenant.get(run_id, tenant_id) != tenant_id:
            raise ValueError("metrics tenant binding is immutable")
        self._current_tenant[run_id] = tenant_id
        for row in self._pending_execution_events.get(run_id, ()):
            if not row["tenant_id"]:
                row["tenant_id"] = tenant_id
        self._persist_execution_events(run_id)

InMemoryAuditLogger

InMemoryAuditLogger(max_entries: int = 10000)

Default in-process logger — keeps entries in an in-memory list.

Primarily for unit tests and local development. Production deployments should register a durable implementation (e.g. the scaffolded Postgres logger). The list is bounded by max_entries to prevent unbounded growth when no one is draining it.

Source code in src/symfonic/core/observability/audit.py

def __init__(self, max_entries: int = 10_000) -> None:
    self._entries: list[AuditLogEntry] = []
    self._max = max_entries
    self._lock: asyncio.Lock = asyncio.Lock()

drain async

drain() -> list[AuditLogEntry]

Return all buffered entries, clearing the buffer.

Tests use this to assert what was emitted during a request.

Source code in src/symfonic/core/observability/audit.py

async def drain(self) -> list[AuditLogEntry]:
    """Return all buffered entries, clearing the buffer.

    Tests use this to assert what was emitted during a request.
    """
    async with self._lock:
        snapshot = list(self._entries)
        self._entries.clear()
        return snapshot

entries

entries() -> list[AuditLogEntry]

Return a non-draining snapshot (sync, for inspection).

Not guaranteed to be consistent under concurrent record() calls — prefer drain() in assertions.

Source code in src/symfonic/core/observability/audit.py

def entries(self) -> list[AuditLogEntry]:
    """Return a non-draining snapshot (sync, for inspection).

    Not guaranteed to be consistent under concurrent ``record()``
    calls — prefer ``drain()`` in assertions.
    """
    return list(self._entries)

record async

record(entry: AuditLogEntry) -> None

Append entry to the in-memory buffer (drops oldest when full).

Source code in src/symfonic/core/observability/audit.py

async def record(self, entry: AuditLogEntry) -> None:
    """Append ``entry`` to the in-memory buffer (drops oldest when full)."""
    async with self._lock:
        self._entries.append(entry)
        if len(self._entries) > self._max:
            overflow = len(self._entries) - self._max
            del self._entries[:overflow]

LLMLifecycleHook

Bases: Protocol

Async instrumentation for LLM call start/end lifecycle.

NoOpObservabilityHook

Zero-overhead default hook. All methods are no-ops.

Always registered in BaseAgentDeps so nodes can safely call deps.require(ObservabilityHook) without risk of MissingCapabilityError.

LSP: signatures match ObservabilityHook protocol exactly.

NodeLifecycleHook

Bases: Protocol

Async instrumentation for graph-node start/end/error lifecycle.

NodeName

Bases: StrEnum

Canonical LLM emission-site discriminators (v7.15.0).

ObservabilityHook

Bases: NodeLifecycleHook, LLMLifecycleHook, ToolLifecycleHook, Protocol

Composed protocol -- union of all lifecycle hooks.

Backward-compatible: any class implementing all 7 methods satisfies this protocol. New consumers should prefer the granular protocols when they only need a subset of methods.

TokenBudgetTracker

TokenBudgetTracker(*, default_daily_budget_usd: float | None = None, default_monthly_budget_usd: float | None = None, store: BudgetStore | None = None)

Tracks cumulative token usage + cost per tenant.

Two in-memory buckets per tenant (day, month) roll over lazily on date/month boundaries. Limits configured globally via default_*_budget_usd or per-tenant via meth:set_tenant_budget; None means unlimited. Admin callers bypass the breaker.

Source code in src/symfonic/core/observability/budget.py

def __init__(
    self,
    *,
    default_daily_budget_usd: float | None = None,
    default_monthly_budget_usd: float | None = None,
    store: BudgetStore | None = None,
) -> None:
    self._default_daily = default_daily_budget_usd
    self._default_monthly = default_monthly_budget_usd
    self._store = store

    # {tenant_id: BudgetUsage}  — always keyed on the CURRENT window.
    # Rollover is lazy: if the window key no longer matches, a fresh
    # BudgetUsage is created on first access.
    self._daily: dict[str, BudgetUsage] = {}
    self._monthly: dict[str, BudgetUsage] = {}

    # Per-tenant overrides.  None value = explicit "unlimited".
    self._tenant_daily: dict[str, float | None] = {}
    self._tenant_monthly: dict[str, float | None] = {}

    # Lazy per-loop lock — binding at __init__ breaks mixed-loop tests.
    self._lock: asyncio.Lock | None = None
    self._lock_loop: Any = None

check_budget async

check_budget(tenant_id: str, *, is_admin: bool = False) -> BudgetCheckResult

Return whether tenant_id may incur another LLM call.

When a BudgetStore with read_today / read_month is registered, the in-memory buckets are first hydrated from the store so every replica shares one view of the running total. Store errors are logged at WARNING and fall through to in-memory state — the ledger is never allowed to break auth. Admin callers bypass the breaker.

Source code in src/symfonic/core/observability/budget.py

async def check_budget(
    self,
    tenant_id: str,
    *,
    is_admin: bool = False,
) -> BudgetCheckResult:
    """Return whether ``tenant_id`` may incur another LLM call.

    When a ``BudgetStore`` with ``read_today`` / ``read_month`` is
    registered, the in-memory buckets are first hydrated from the
    store so every replica shares one view of the running total.
    Store errors are logged at WARNING and fall through to in-memory
    state — the ledger is never allowed to break auth.  Admin
    callers bypass the breaker.
    """
    from symfonic.core.observability.budget_hydrate import (
        hydrate_from_store,
    )
    await hydrate_from_store(self, tenant_id)

    day = await self.usage_today(tenant_id)
    month = await self.usage_this_month(tenant_id)
    daily_limit, monthly_limit = self.get_tenant_limits(tenant_id)

    if is_admin:
        return BudgetCheckResult(
            allowed=True,
            daily_used_usd=day.cost_usd,
            daily_limit_usd=daily_limit,
            monthly_used_usd=month.cost_usd,
            monthly_limit_usd=monthly_limit,
        )

    if daily_limit is not None and day.cost_usd >= daily_limit:
        return BudgetCheckResult(
            allowed=False,
            reason=(
                f"daily limit ${daily_limit:.2f} reached "
                f"(used: ${day.cost_usd:.4f})"
            ),
            daily_used_usd=day.cost_usd,
            daily_limit_usd=daily_limit,
            monthly_used_usd=month.cost_usd,
            monthly_limit_usd=monthly_limit,
        )
    if monthly_limit is not None and month.cost_usd >= monthly_limit:
        return BudgetCheckResult(
            allowed=False,
            reason=(
                f"monthly limit ${monthly_limit:.2f} reached "
                f"(used: ${month.cost_usd:.4f})"
            ),
            daily_used_usd=day.cost_usd,
            daily_limit_usd=daily_limit,
            monthly_used_usd=month.cost_usd,
            monthly_limit_usd=monthly_limit,
        )

    return BudgetCheckResult(
        allowed=True,
        daily_used_usd=day.cost_usd,
        daily_limit_usd=daily_limit,
        monthly_used_usd=month.cost_usd,
        monthly_limit_usd=monthly_limit,
    )

clear_tenant_budget

clear_tenant_budget(tenant_id: str) -> None

Remove per-tenant overrides so the defaults apply again.

Source code in src/symfonic/core/observability/budget.py

def clear_tenant_budget(self, tenant_id: str) -> None:
    """Remove per-tenant overrides so the defaults apply again."""
    self._tenant_daily.pop(tenant_id, None)
    self._tenant_monthly.pop(tenant_id, None)

get_tenant_limits

get_tenant_limits(tenant_id: str) -> tuple[float | None, float | None]

Return (daily_limit, monthly_limit) applied to tenant_id.

Source code in src/symfonic/core/observability/budget.py

def get_tenant_limits(
    self, tenant_id: str,
) -> tuple[float | None, float | None]:
    """Return ``(daily_limit, monthly_limit)`` applied to ``tenant_id``."""
    daily = self._tenant_daily.get(tenant_id, self._default_daily) \
        if tenant_id in self._tenant_daily else self._default_daily
    monthly = self._tenant_monthly.get(tenant_id, self._default_monthly) \
        if tenant_id in self._tenant_monthly else self._default_monthly
    return daily, monthly

record async

record(tenant_id: str, model: str, usage: dict[str, int]) -> float

Record token usage for a single LLM call; returns its cost_usd.

Source code in src/symfonic/core/observability/budget.py

async def record(
    self,
    tenant_id: str,
    model: str,
    usage: dict[str, int],
) -> float:
    """Record token usage for a single LLM call; returns its cost_usd."""
    input_tokens = int(usage.get("input_tokens", 0) or 0)
    output_tokens = int(usage.get("output_tokens", 0) or 0)
    cached = int(usage.get("cached_tokens", 0) or 0)
    cache_write = int(
        usage.get("cache_write_tokens")
        or usage.get("cache_creation_tokens")
        or 0,
    )
    cost = calculate_cost(
        model, input_tokens, output_tokens, cached, cache_write,
    )

    async with self._get_lock():
        day = self._bucket(tenant_id, "day")
        month = self._bucket(tenant_id, "month")
        for b in (day, month):
            b.input_tokens += input_tokens
            b.output_tokens += output_tokens
            b.cached_tokens += cached
            b.cache_write_tokens += cache_write
            b.cost_usd = round(b.cost_usd + cost, 6)
            b.request_count += 1
            b.updated_at = datetime.now(UTC)

        daily_snapshot = copy_usage(day)

    # Persistence happens OUTSIDE the lock — the store may do I/O.
    # Log but don't propagate; telemetry failures never break LLM flow.
    if self._store is not None:
        try:
            await self._store.upsert_daily(daily_snapshot)
        except Exception:  # pragma: no cover - integration tested
            logger.warning(
                "BudgetStore.upsert_daily failed for tenant=%s "
                "(in-memory counters still accurate)",
                tenant_id,
                exc_info=True,
            )
    return cost

reset async

reset(tenant_id: str) -> None

Reset in-memory day + month counters for a tenant.

When a persistent store is wired, the shared Postgres row is NOT cleared — use the admin billing endpoint to zero the ledger.

Source code in src/symfonic/core/observability/budget.py

async def reset(self, tenant_id: str) -> None:
    """Reset in-memory day + month counters for a tenant.

    When a persistent store is wired, the shared Postgres row is NOT
    cleared — use the admin billing endpoint to zero the ledger.
    """
    async with self._get_lock():
        self._daily.pop(tenant_id, None)
        self._monthly.pop(tenant_id, None)

set_tenant_budget

set_tenant_budget(tenant_id: str, *, daily_usd: float | None = None, monthly_usd: float | None = None) -> None

Override budget ceilings for a tenant; None = unlimited.

Source code in src/symfonic/core/observability/budget.py

def set_tenant_budget(
    self,
    tenant_id: str,
    *,
    daily_usd: float | None = None,
    monthly_usd: float | None = None,
) -> None:
    """Override budget ceilings for a tenant; ``None`` = unlimited."""
    self._tenant_daily[tenant_id] = daily_usd
    self._tenant_monthly[tenant_id] = monthly_usd

ToolLifecycleHook

Bases: Protocol

Async instrumentation for tool invocation start/end lifecycle.

emit_audit_event

emit_audit_event(entry: AuditLogEntry) -> None

Schedule an audit write without blocking the caller.

Never raises — failures are logged. If no logger is registered the call is a complete no-op (zero allocations beyond the entry itself).

The FastAPI routers use this directly in handlers so a slow / sick audit backend can never 500 a user request. The task is scheduled on the currently-running loop; when called outside an async context (defensive guard) we log a warning and drop the entry — no audit logger is expected to live that far up the stack.

Source code in src/symfonic/core/observability/audit.py

def emit_audit_event(entry: AuditLogEntry) -> None:
    """Schedule an audit write without blocking the caller.

    Never raises — failures are logged.  If no logger is registered the
    call is a complete no-op (zero allocations beyond the entry itself).

    The FastAPI routers use this directly in handlers so a slow / sick
    audit backend can never 500 a user request.  The task is scheduled
    on the currently-running loop; when called outside an async context
    (defensive guard) we log a warning and drop the entry — no audit
    logger is expected to live that far up the stack.
    """
    logger_ = _audit_logger
    if logger_ is None:
        return

    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        logger.warning(
            "emit_audit_event called outside a running loop; dropping entry "
            "(action=%s resource=%s)",
            entry.action, entry.resource_type,
        )
        return

    coro: Awaitable[None] = logger_.record(entry)
    task = loop.create_task(coro)
    task.add_done_callback(_audit_task_done)

exchanges_for_rows

exchanges_for_rows(rows: list[dict[str, Any]]) -> list[tuple[str | None, str | None]]

Align (prompt, response) pairs onto one run's snapshot rows.

rows must be the timestamp-ordered snapshots of a single run group (as build_conversation_detail groups them). Position is the join, input_tokens the verification: a mismatch attaches nothing rather than a neighboring call's text (fail-closed on ring skew, e.g. the flag toggled mid-process). All-None when the flag is off or nothing was captured.

Source code in src/symfonic/core/observability/debug_exchanges.py

def exchanges_for_rows(
    rows: list[dict[str, Any]],
) -> list[tuple[str | None, str | None]]:
    """Align (prompt, response) pairs onto one run's snapshot rows.

    ``rows`` must be the timestamp-ordered snapshots of a single run
    group (as ``build_conversation_detail`` groups them). Position is
    the join, ``input_tokens`` the verification: a mismatch attaches
    nothing rather than a neighboring call's text (fail-closed on ring
    skew, e.g. the flag toggled mid-process). All-``None`` when the
    flag is off or nothing was captured.
    """
    empty: list[tuple[str | None, str | None]] = [(None, None)] * len(rows)
    if not rows or not debug_exchanges_enabled():
        return empty
    run_ids = {str(r.get("run_id") or r.get("call_id") or "") for r in rows}
    if len(run_ids) != 1:
        return empty
    (run_id,) = run_ids
    if not run_id:
        return empty
    by_ordinal = {e["ordinal"]: e for e in get_exchanges(run_id)}
    aligned: list[tuple[str | None, str | None]] = []
    for ordinal, row in enumerate(rows):
        entry = by_ordinal.get(ordinal)
        if entry is None:
            aligned.append((None, None))
            continue
        if int(row.get("input_tokens", 0) or 0) != entry["input_tokens"]:
            aligned.append((None, None))
            continue
        aligned.append((entry["prompt"], entry["response"]))
    return aligned

get_audit_logger

get_audit_logger() -> AuditLogger | None

Return the currently registered audit logger, if any.

Source code in src/symfonic/core/observability/audit.py

def get_audit_logger() -> AuditLogger | None:
    """Return the currently registered audit logger, if any."""
    return _audit_logger

get_budget_tracker

get_budget_tracker() -> TokenBudgetTracker | None

Return the currently registered tracker, if any.

Source code in src/symfonic/core/observability/budget.py

def get_budget_tracker() -> TokenBudgetTracker | None:
    """Return the currently registered tracker, if any."""
    return _budget_tracker

human_label

human_label(node_name: str) -> str

Return the UI-friendly label for node_name.

Unknown names (adopter-extended sites) round-trip unchanged so dashboards do not lose attribution -- this is the explicit non-validation contract documented at module level.

Parameters:

Name	Type	Description	Default
node_name	str	The discriminator string from LLMEndEvent.node_name.	required

Returns:

Type	Description
str	The matching label from :data:NODE_LABELS if present, otherwise
str	node_name itself.

Source code in src/symfonic/core/observability/node_names.py

def human_label(node_name: str) -> str:
    """Return the UI-friendly label for ``node_name``.

    Unknown names (adopter-extended sites) round-trip unchanged so
    dashboards do not lose attribution -- this is the explicit
    non-validation contract documented at module level.

    Args:
        node_name: The discriminator string from ``LLMEndEvent.node_name``.

    Returns:
        The matching label from :data:`NODE_LABELS` if present, otherwise
        ``node_name`` itself.
    """
    return NODE_LABELS.get(node_name, node_name)

observed_node

observed_node(node_name: str, *, swallow_errors: bool = False, always_log: bool = False) -> Callable[..., Callable[..., Any]]

Decorator factory for observable graph nodes.

Parameters:

Name	Type	Description	Default
node_name	str	Identifier recorded in node_execution_log and passed to ObservabilityHook callbacks.	required
swallow_errors	bool	When True, catch non-programming exceptions (not MissingCapabilityError, ConfigurationError) and return {} instead of propagating. Used by compaction.	False
always_log	bool	When True, append to node_execution_log even when the wrapped function returns {}. Used by nodes like context_injection that always execute meaningfully.	False

Usage::

@observed_node("react")
async def react(state: dict[str, Any]) -> dict[str, Any]:
    ...  # pure business logic

Source code in src/symfonic/core/observability/decorators.py

def observed_node(
    node_name: str,
    *,
    swallow_errors: bool = False,
    always_log: bool = False,
) -> Callable[..., Callable[..., Any]]:
    """Decorator factory for observable graph nodes.

    Args:
        node_name: Identifier recorded in node_execution_log and
            passed to ObservabilityHook callbacks.
        swallow_errors: When True, catch non-programming exceptions
            (not MissingCapabilityError, ConfigurationError) and return
            ``{}`` instead of propagating. Used by compaction.
        always_log: When True, append to node_execution_log even when
            the wrapped function returns ``{}``. Used by nodes like
            context_injection that always execute meaningfully.

    Usage::

        @observed_node("react")
        async def react(state: dict[str, Any]) -> dict[str, Any]:
            ...  # pure business logic
    """

    def decorator(fn: Callable[..., Any]) -> Callable[..., Any]:
        @functools.wraps(fn)
        async def wrapper(state: dict[str, Any]) -> dict[str, Any]:
            from ..callbacks.manager import CallbackManager
            from ..contracts.callbacks import NodeEndEvent, NodeErrorEvent, NodeStartEvent
            from ..deps import BaseAgentDeps, MissingCapabilityError
            from ..graph import ConfigurationError
            from .protocol import NodeLifecycleHook, ObservabilityHook

            deps: BaseAgentDeps = state["deps"]
            hook: NodeLifecycleHook = deps.require(ObservabilityHook)
            run_id: str = state.get("run_id", "") or ""

            # Prefer merged manager from state (propagated by runtime for
            # per-invocation callbacks), fall back to deps-level manager.
            callback_mgr: CallbackManager | None = (
                state.get("_callback_manager") or deps.get(CallbackManager)
            )

            try:
                await hook.on_node_start(node_name, state)
            except Exception:
                logger.exception("ObservabilityHook.on_node_start failed for %s", node_name)

            if callback_mgr and not callback_mgr.is_noop:
                await callback_mgr.on_node_start(
                    NodeStartEvent(node_name=node_name, run_id=run_id)
                )

            start = time.monotonic()

            try:
                result = await fn(state)

                # Append to node_execution_log when the function produced
                # a non-empty result, or when always_log is set (nodes like
                # context_injection that always execute meaningfully).
                if result or always_log:
                    result["node_execution_log"] = (
                        state.get("node_execution_log", []) + [node_name]
                    )

                duration_ms = (time.monotonic() - start) * 1000
                try:
                    await hook.on_node_end(node_name, result, duration_ms)
                except Exception:
                    logger.exception("ObservabilityHook.on_node_end failed for %s", node_name)

                if callback_mgr and not callback_mgr.is_noop:
                    await callback_mgr.on_node_end(
                        NodeEndEvent(
                            node_name=node_name,
                            run_id=run_id,
                            duration_ms=duration_ms,
                        )
                    )
                return result

            except Exception as e:
                duration_ms = (time.monotonic() - start) * 1000

                if swallow_errors and not isinstance(
                    e, (MissingCapabilityError, ConfigurationError)
                ):
                    # BUG-3: Emit error event, not success, when swallowing
                    try:
                        await hook.on_error(node_name, e, run_id=run_id)
                    except Exception:
                        logger.exception(
                            "ObservabilityHook.on_error failed for %s", node_name
                        )

                    if callback_mgr and not callback_mgr.is_noop:
                        await callback_mgr.on_node_error(
                            NodeErrorEvent(
                                node_name=node_name,
                                run_id=run_id,
                                error=str(e),
                                error_type=type(e).__name__,
                            )
                        )
                    return {}

                try:
                    await hook.on_error(node_name, e, run_id=run_id)
                except Exception:
                    logger.exception(
                        "ObservabilityHook.on_error failed for %s", node_name
                    )

                if callback_mgr and not callback_mgr.is_noop:
                    await callback_mgr.on_node_error(
                        NodeErrorEvent(
                            node_name=node_name,
                            run_id=run_id,
                            error=str(e),
                            error_type=type(e).__name__,
                        )
                    )
                raise

        return wrapper

    return decorator

set_audit_logger

set_audit_logger(logger_: AuditLogger | None) -> None

Register (or clear) the process-wide audit logger.

Call once at app startup. Passing None disables audit emission — useful for tests that registered a logger and want to tear down.

Source code in src/symfonic/core/observability/audit.py

def set_audit_logger(logger_: AuditLogger | None) -> None:
    """Register (or clear) the process-wide audit logger.

    Call once at app startup.  Passing ``None`` disables audit emission
    — useful for tests that registered a logger and want to tear down.
    """
    global _audit_logger
    _audit_logger = logger_

set_budget_tracker

set_budget_tracker(tracker: TokenBudgetTracker | None) -> None

Register (or clear) the process-wide budget tracker.

Consulted by check_budget_before_llm and SymfonicAgent.run(); when unset both skip the check.

Source code in src/symfonic/core/observability/budget.py

def set_budget_tracker(tracker: TokenBudgetTracker | None) -> None:
    """Register (or clear) the process-wide budget tracker.

    Consulted by ``check_budget_before_llm`` and ``SymfonicAgent.run()``;
    when unset both skip the check.
    """
    global _budget_tracker
    _budget_tracker = tracker