Skip to content

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 
60
61
62
63
64
65
66
67
68
69
70
71
72
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 
 94
 95
 96
 97
 98
 99
100
101
102
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 
84
85
86
87
88
89
90
91
92
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.

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 
 98
 99
100
101
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 
111
112
113
114
115
116
117
118
119
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 
121
122
123
124
125
126
127
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 
103
104
105
106
107
108
109
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 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
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 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
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 
79
80
81
82
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 
84
85
86
87
88
89
90
91
92
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 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
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 
235
236
237
238
239
240
241
242
243
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 
68
69
70
71
72
73
74
75
76
77
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 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
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)

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 
147
148
149
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 
286
287
288
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 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
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 
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
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 
137
138
139
140
141
142
143
144
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 
276
277
278
279
280
281
282
283
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