Observability¶

def get_logger(name: str, **initial_bindings: Any) -> structlog.stdlib.BoundLogger:
    """Get a structured logger bound to the given name.

    Thin wrapper over :func:`structlog.get_logger` that ensures
    consistent logger creation across the codebase.

    Usage::

        from synthorg.observability import get_logger

        logger = get_logger(__name__)
        logger.info("something happened", key="value")

    Args:
        name: Logger name, typically ``__name__``.
        **initial_bindings: Key-value pairs bound to every log entry.

    Returns:
        A bound structlog logger.
    """
    return structlog.get_logger(name, **initial_bindings)  # type: ignore[no-any-return]

def generate_correlation_id() -> str:
    """Generate a new correlation ID.

    Returns:
        A UUID4 string suitable for use as a correlation identifier.
    """
    return str(uuid.uuid4())

def bind_correlation_id(
    *,
    request_id: str | None = None,
    task_id: str | None = None,
    agent_id: str | None = None,
) -> None:
    """Bind correlation IDs to the current context.

    Only non-``None`` values are bound.  Existing bindings for
    unspecified keys are left unchanged.

    Args:
        request_id: Request correlation identifier.
        task_id: Task correlation identifier.
        agent_id: Agent correlation identifier.
    """
    bindings = _build_bindings(request_id, task_id, agent_id)
    if bindings:
        structlog.contextvars.bind_contextvars(**bindings)

def unbind_correlation_id(
    *,
    request_id: bool = False,
    task_id: bool = False,
    agent_id: bool = False,
) -> None:
    """Remove specific correlation IDs from the current context.

    Args:
        request_id: Whether to unbind the ``request_id`` key.
        task_id: Whether to unbind the ``task_id`` key.
        agent_id: Whether to unbind the ``agent_id`` key.
    """
    keys: list[str] = []
    if request_id:
        keys.append("request_id")
    if task_id:
        keys.append("task_id")
    if agent_id:
        keys.append("agent_id")
    if keys:
        structlog.contextvars.unbind_contextvars(*keys)

def clear_correlation_ids() -> None:
    """Remove all correlation IDs from the current context.

    Unbinds ``request_id``, ``task_id``, and ``agent_id``.  Other
    context variables are preserved.
    """
    structlog.contextvars.unbind_contextvars(
        "request_id",
        "task_id",
        "agent_id",
    )

@contextmanager
def correlation_scope(
    *,
    request_id: str | None = None,
    task_id: str | None = None,
    agent_id: str | None = None,
) -> Iterator[None]:
    """Scoped correlation binding that restores prior values on exit.

    Uses structlog's ``bound_contextvars`` to save and restore any
    pre-existing correlation IDs, making this safe for nested
    execution contexts (e.g. hierarchical agent delegation).

    Args:
        request_id: Request correlation identifier to bind.
        task_id: Task correlation identifier to bind.
        agent_id: Agent correlation identifier to bind.
    """
    bindings = _build_bindings(request_id, task_id, agent_id)
    if bindings:
        with structlog.contextvars.bound_contextvars(**bindings):
            yield
    else:
        yield

def with_correlation(
    *,
    request_id: str | None = None,
    task_id: str | None = None,
    agent_id: str | None = None,
) -> Callable[[Callable[_P, _T]], Callable[_P, _T]]:
    """Decorator that binds correlation IDs for a function's duration.

    Correlation IDs are bound before the function executes and unbound
    after it returns or raises.  Only non-``None`` IDs are managed.

    Note:
        This decorator is for **synchronous** functions only.  Applying
        it to an ``async def`` function raises :exc:`TypeError`.  For
        async functions, use :func:`with_correlation_async` instead.

    Args:
        request_id: Request correlation identifier to bind.
        task_id: Task correlation identifier to bind.
        agent_id: Agent correlation identifier to bind.

    Returns:
        A decorator that manages correlation ID lifecycle.

    Raises:
        TypeError: If the decorated function is a coroutine function.
    """

    def decorator(func: Callable[_P, _T]) -> Callable[_P, _T]:
        if inspect.iscoroutinefunction(func):
            msg = (
                "with_correlation() does not support async functions. "
                "Use with_correlation_async() instead."
            )
            logger.warning(
                CORRELATION_SYNC_DECORATOR_MISUSE,
                function=func.__qualname__,
            )
            raise TypeError(msg)

        @functools.wraps(func)
        def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> _T:
            bindings = _build_bindings(request_id, task_id, agent_id)
            with structlog.contextvars.bound_contextvars(**bindings):
                return func(*args, **kwargs)

        return wrapper

    return decorator

def with_correlation_async(
    *,
    request_id: str | None = None,
    task_id: str | None = None,
    agent_id: str | None = None,
) -> Callable[
    [Callable[_P, Coroutine[object, object, _T]]],
    Callable[_P, Coroutine[object, object, _T]],
]:
    """Decorator that binds correlation IDs for an async function's duration.

    Correlation IDs are bound before the coroutine executes and unbound
    after it returns or raises.  Only non-``None`` IDs are managed.

    Note:
        This decorator is for **async** functions only.  Applying it to
        a synchronous function raises :exc:`TypeError`.  For sync
        functions use :func:`with_correlation`.

    Args:
        request_id: Request correlation identifier to bind.
        task_id: Task correlation identifier to bind.
        agent_id: Agent correlation identifier to bind.

    Returns:
        A decorator that manages correlation ID lifecycle for async
        functions.

    Raises:
        TypeError: If the decorated function is not a coroutine function.
    """

    def decorator(
        func: Callable[_P, Coroutine[object, object, _T]],
    ) -> Callable[_P, Coroutine[object, object, _T]]:
        if not inspect.iscoroutinefunction(func):
            msg = (
                "with_correlation_async() requires an async function. "
                "Use with_correlation() for synchronous functions."
            )
            logger.warning(
                CORRELATION_ASYNC_DECORATOR_MISUSE,
                function=func.__qualname__,
            )
            raise TypeError(msg)

        @functools.wraps(func)
        async def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> _T:
            bindings = _build_bindings(request_id, task_id, agent_id)
            with structlog.contextvars.bound_contextvars(**bindings):
                return await func(*args, **kwargs)

        return wrapper

    return decorator

def configure_logging(
    config: LogConfig | None = None,
    *,
    routing_overrides: Mapping[str, tuple[str, ...]] | None = None,
) -> None:
    """Configure the structured logging system.

    Sets up structlog processor chains, stdlib handlers, and per-logger
    levels.  This function is **idempotent** -- calling it multiple times
    replaces the previous configuration without duplicating handlers.

    Respects the ``SYNTHORG_LOG_LEVEL`` env var to override the console
    sink level (useful for Docker deployments).

    Args:
        config: Logging configuration.  When ``None``, uses sensible
            defaults with all standard sinks.
        routing_overrides: Optional extra logger-name routing entries
            (e.g. from custom sinks) merged with the default
            ``SINK_ROUTING`` table.

    Raises:
        RuntimeError: If a critical sink fails to initialise.
    """
    if config is None:
        config = LogConfig(sinks=DEFAULT_SINKS)

    config = _apply_console_level_override(config)

    # 1. Reset structlog to a clean state
    structlog.reset_defaults()

    # 2. Clear existing stdlib root handlers
    root_logger = logging.getLogger()
    _clear_root_handlers(root_logger)

    # 3. Set root logger level from config
    root_logger.setLevel(config.root_level.value)

    # 4. Build shared processor chain (foreign pre-chain)
    shared = _build_shared_processors(
        enable_correlation=config.enable_correlation,
    )

    # 5. Configure structlog main chain
    _configure_structlog(enable_correlation=config.enable_correlation)

    # 6. Build and attach handlers for each sink
    _attach_handlers(
        config,
        root_logger,
        shared,
        routing_overrides=routing_overrides,
    )

    # 7. Tame third-party loggers (clear duplicate handlers, set defaults)
    _tame_third_party_loggers()

    # 8. Apply per-logger levels (after taming so user overrides take precedence)
    _apply_logger_levels(config)

def sanitize_sensitive_fields(
    logger: Any,  # noqa: ARG001
    method_name: str,  # noqa: ARG001
    event_dict: MutableMapping[str, Any],
) -> Mapping[str, Any]:
    """Redact values of keys matching sensitive patterns.

    Returns a new dict rather than mutating the original event dict,
    following the project's immutability convention.  Redaction is
    applied recursively to nested dicts, lists, and tuples.

    Args:
        logger: The wrapped logger object (unused, required by structlog).
        method_name: The name of the log method called (unused).
        event_dict: The event dictionary to process.

    Returns:
        A new event dict with sensitive values replaced by
        ``**REDACTED**`` at all nesting depths.
    """
    return {
        key: (
            _REDACTED
            if isinstance(key, str) and _SENSITIVE_PATTERN.search(key)
            else _redact_value(value)
        )
        for key, value in event_dict.items()
    }

def scrub_event_fields(
    logger: Any,  # noqa: ARG001
    method_name: str,  # noqa: ARG001
    event_dict: MutableMapping[str, Any],
) -> Mapping[str, Any]:
    """Deep-scrub credential patterns out of every string value.

    Belt-and-braces defence against the ``error=str(exc)`` leak
    vector: even when a caller embeds a stringified exception (or
    response body) that carries ``client_secret=...``,
    ``"access_token":"..."``, ``Authorization: Bearer ...``, or raw
    Fernet ciphertext, this processor rewrites the string so those
    substrings are masked before the renderer sees them.

    Runs *after* ``sanitize_sensitive_fields`` so keys that the
    field-name scrubber already replaced with ``**REDACTED**`` stay
    redacted.

    **Robustness contract**: this processor runs on every log record.
    If ``_scrub_value`` raises (e.g. a corrupted object whose ``repr``
    blows up, or a pathological recursive structure), we return the
    *original* event dict unchanged rather than letting the exception
    propagate and abort the caller's log call. Losing scrubbing on one
    event is preferable to silencing the entire logging pipeline at the
    moment of crisis.

    Args:
        logger: The wrapped logger object (unused, required by structlog).
        method_name: The name of the log method called (unused).
        event_dict: The event dictionary to process.

    Returns:
        A new event dict with every string value scrubbed via
        :func:`synthorg.observability.redaction.scrub_secret_tokens`,
        or the original dict if the scrub itself fails.
    """
    try:
        return {key: _scrub_value(value) for key, value in event_dict.items()}
    except MemoryError, RecursionError:
        # Interpreter-fatal errors must propagate per project convention
        # -- swallowing them here would hide exactly the class of failures
        # the rest of the codebase relies on for surfacing catastrophic
        # state.
        raise
    except Exception as exc:
        # Fail open: pass the event through unscrubbed rather than drop
        # the log line entirely.  Still safer than crashing the log
        # pipeline -- ``sanitize_sensitive_fields`` (which ran just
        # before us) has already redacted known-sensitive *field names*.
        # We write to ``sys.stderr`` directly (never via ``logger``) so
        # operators notice the scrub regression without triggering a
        # recursive log-through-logger failure. ``processors.py`` is
        # not on the ``print()`` allowlist, so we use the raw stream
        # write instead of ``print(file=sys.stderr)``.
        sys.stderr.write(
            f"WARNING: scrub_event_fields failed; event passed unscrubbed: "
            f"{type(exc).__name__}\n",
        )
        sys.stderr.flush()
        return event_dict

def build_handler(
    sink: SinkConfig,
    log_dir: Path,
    foreign_pre_chain: list[Any],
    *,
    routing: Mapping[str, tuple[str, ...]] | None = None,
) -> logging.Handler:
    """Build a stdlib logging handler from a sink configuration.

    For ``CONSOLE`` sinks a :class:`logging.StreamHandler` writing to
    ``stderr`` is created.  For ``FILE`` sinks see
    :func:`_build_file_handler`.  For ``SYSLOG`` and ``HTTP`` sinks,
    dedicated handler builders are used.

    Note: SYSLOG and HTTP sinks are built and returned by dedicated
    handler modules; they do not participate in logger-name routing.

    Args:
        sink: The sink configuration describing the handler to build.
        log_dir: Base directory for log files.
        foreign_pre_chain: Processor chain for stdlib-originated logs.
        routing: Optional routing table to use instead of the
            module-level ``SINK_ROUTING``.  When ``None``, the
            default routing is used.

    Returns:
        A configured :class:`logging.Handler` with formatter attached.
    """
    effective_routing = routing if routing is not None else SINK_ROUTING

    handler: logging.Handler
    match sink.sink_type:
        case SinkType.CONSOLE:
            handler = logging.StreamHandler(sys.stderr)
        case SinkType.FILE:
            handler = _build_file_handler(sink, log_dir)
        case SinkType.SYSLOG:
            from synthorg.observability.syslog_handler import (  # noqa: PLC0415
                build_syslog_handler,
            )

            return build_syslog_handler(sink, foreign_pre_chain)
        case SinkType.HTTP:
            from synthorg.observability.http_handler import (  # noqa: PLC0415
                build_http_handler,
            )

            return build_http_handler(sink, foreign_pre_chain)
        case SinkType.PROMETHEUS:
            # Prometheus is pull-based (scrape endpoint), not a log handler.
            # Return a no-op handler -- the /metrics controller serves metrics.
            handler = logging.NullHandler()
            handler.setLevel(sink.level.value)
            return handler
        case SinkType.OTLP:
            from synthorg.observability.otlp_handler import (  # noqa: PLC0415
                build_otlp_handler,
            )

            return build_otlp_handler(sink, foreign_pre_chain)
        case _:  # pragma: no cover
            msg = f"Unsupported sink type: {sink.sink_type}"  # type: ignore[unreachable]
            raise ValueError(msg)

    _attach_formatter_and_routing(
        handler,
        sink,
        foreign_pre_chain,
        effective_routing,
    )
    return handler

def build_log_config_from_settings(
    *,
    root_level: LogLevel,
    enable_correlation: bool,
    sink_overrides_json: str,
    custom_sinks_json: str,
    log_dir: str = "logs",
) -> SinkBuildResult:
    """Merge DEFAULT_SINKS with runtime overrides and custom sinks.

    Args:
        root_level: Root logger level.
        enable_correlation: Whether to enable correlation ID tracking.
        sink_overrides_json: JSON object of per-sink overrides.
        custom_sinks_json: JSON array of custom sink definitions.
        log_dir: Directory for log files.

    Returns:
        A :class:`SinkBuildResult` containing the validated
        :class:`LogConfig` and any routing overrides for custom sinks.

    Raises:
        ValueError: On invalid JSON, validation failures, or
            attempts to disable the console sink.
    """
    overrides = _parse_sink_overrides(sink_overrides_json)
    custom_entries = _parse_custom_sinks(custom_sinks_json)

    merged = _merge_default_sinks(overrides)
    routing = _process_custom_entries(custom_entries, merged)

    config = LogConfig(
        root_level=root_level,
        enable_correlation=enable_correlation,
        sinks=tuple(merged),
        log_dir=log_dir,
    )
    return SinkBuildResult(config=config, routing_overrides=routing)