symfonic.core.observability.decorators

decorators

@observed_node -- DRY decorator for node observability boilerplate.

Wraps async node functions with on_node_start / on_node_end / on_error lifecycle hooks and node_execution_log bookkeeping. Also dispatches to CallbackManager for user-facing node-level callbacks.

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