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 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 _:  # 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)