ss-tools/backend/src/core/cot_logger.py

# #region CotLoggerModule [C:4] [TYPE Module] [SEMANTICS logging, cot, json, trace, structured]
# @BRIEF Structured JSON logger implementing the decision audit logging protocol.
#           Uses ContextVar for trace_id and span_id propagation across async contexts.
#           Provides log(), MarkerLogger, seed_trace_id(), push_span(), pop_span().
# @LAYER Core
# @RELATION CALLED_BY -> [TraceContextMiddleware]
# @RELATION CALLED_BY -> [CotLoggerModule]
# @PRE  Python 3.7+ (ContextVar available).
# @POST  JSON log records written to the 'cot' logger at appropriate levels.
# @SIDE_EFFECT  Writes structured JSON to the 'cot' Python logger.
# @DATA_CONTRACT  Log call -> Single-line JSON to logging.StreamHandler/file.

from contextvars import ContextVar
import logging
from typing import Any
import uuid

# #region cot_trace_context [C:1] [TYPE Data] [SEMANTICS contextvar, trace_id, span_id, propagation]
# @BRIEF ContextVars for trace ID and span ID propagation across async boundaries.
_trace_id: ContextVar[str] = ContextVar("_trace_id", default="")
_span_id: ContextVar[str] = ContextVar("_span_id", default="")
# #endregion cot_trace_context

# #region cot_logger_instance [C:1] [TYPE Data] [SEMANTICS logger, instance]
# @BRIEF Dedicated Python logger for all decision audit log output.
cot_logger = logging.getLogger("cot")
# #endregion cot_logger_instance

__all__ = [
    "MarkerLogger",
    "get_trace_id",
    "log",
    "pop_span",
    "push_span",
    "seed_trace_id",
    "set_trace_id",
]

# #region seed_trace_id [C:1] [TYPE Function] [SEMANTICS trace_id, uuid, contextvar, set]
# @BRIEF Generate a new UUID4 trace_id, set it in ContextVar, and return it.
def seed_trace_id() -> str:
    """Generate a new UUID4 trace ID and store it in the thread-local ContextVar.

    Returns:
        str: The newly generated trace ID.
    """
    trace_id = str(uuid.uuid4())
    _trace_id.set(trace_id)
    return trace_id


# #endregion seed_trace_id

# #region set_trace_id [C:1] [TYPE Function] [SEMANTICS trace_id, contextvar, set, public]
# @BRIEF Set an explicit trace_id into the ContextVar (e.g. from an incoming header).
def set_trace_id(trace_id: str) -> None:
    """Set a specific trace_id into the ContextVar.

    This is used by TraceContextMiddleware when an ``X-Trace-ID`` header
    is present, enabling cross-service trace chaining.

    Args:
        trace_id: The trace ID to set.
    """
    _trace_id.set(trace_id)


# #endregion set_trace_id

# #region get_trace_id [C:1] [TYPE Function] [SEMANTICS trace_id, contextvar, get, public]
# @BRIEF Get the current trace_id from ContextVar.
def get_trace_id() -> str:
    """Get the current trace_id from the thread-local ContextVar.

    Returns:
        str: The current trace ID, or empty string if none set.
    """
    return _trace_id.get()


# #endregion get_trace_id

# #region push_span [C:1] [TYPE Function] [SEMANTICS span_id, contextvar, stack]
# @BRIEF Set a new span_id in ContextVar and return the previous span_id for later restoration.
def push_span(span: str) -> str:
    """Push a new span ID onto the context and return the previous span ID.

    Args:
        span: The new span identifier (e.g. function or operation name).

    Returns:
        str: The previous span ID, suitable for passing to pop_span().
    """
    prev = _span_id.get()
    _span_id.set(span)
    return prev


# #endregion push_span

# #region pop_span [C:1] [TYPE Function] [SEMANTICS span_id, contextvar, restore]
# @BRIEF Restore a previous span_id into the ContextVar.
def pop_span(prev: str) -> None:
    """Restore a previous span ID into the ContextVar.

    Args:
        prev: The span ID to restore (previously returned by push_span).
    """
    _span_id.set(prev)


# #endregion pop_span

# #region cot_log_function [C:2] [TYPE Function] [SEMANTICS log, json, structured, marker]
# @BRIEF Core structured logging function that emits a single-line JSON record.
def log(
    src: str,
    marker: str,
    intent: str,
    payload: dict[str, Any] | None = None,
    error: str | None = None,
    level: str | None = None,
) -> None:
    """Emit a single-line structured JSON log record through the 'cot' logger.

    Args:
        src: Qualified function or component name (e.g. 'AuthRepository.get_user').
        marker: Protocol marker — one of 'REASON', 'REFLECT', 'EXPLORE'.
        intent: Short one-line description of the intent or action.
        payload: Optional structured data dict to include in the record.
        error: Required for EXPLORE markers; describes the violated assumption.
        level: Log level override. Auto-inferred from marker if omitted
               (REASON/REFLECT -> INFO, EXPLORE -> WARNING).

    Side effects:
        Writes a single-line JSON string to the 'cot' Python logger.
    """
    if level is None:
        level = "WARNING" if marker == "EXPLORE" else "INFO"

    # Build structured extra data for the LogRecord.
    # CotJsonFormatter will read these attributes to produce the final JSON output,
    # consistent with the main superset_tools_app logger.
    extra: dict[str, Any] = {
        "marker": marker,
        "intent": intent,
        "src": src,
    }

    if payload is not None:
        extra["payload"] = payload

    if error is not None:
        extra["error"] = error

    log_func = {
        "WARNING": cot_logger.warning,
        "ERROR": cot_logger.error,
        "DEBUG": cot_logger.debug,
    }.get(level, cot_logger.info)

    log_func(intent, extra=extra)


# #endregion cot_log_function

# #region MarkerLogger [C:2] [TYPE Class] [SEMANTICS logger, proxy, marker, syntactic-sugar]
# @BRIEF Thin proxy class providing .reason(), .reflect(), .explore() convenience methods.
class MarkerLogger:
    """Thin proxy over the cot_logger.log() function.

    Usage::

        log = MarkerLogger("AuthService")
        log.reason("Validating credentials", payload={"user_id": "..."})
        log.reflect("Validation complete", payload={"result": "ok"})
        log.explore("Token expired", error="JWT expired, attempting refresh")

    Each method delegates to :func:`log` with the corresponding marker.
    """

    # #region MarkerLogger.__init__ [TYPE Function]
    # @BRIEF Store the module/component name used as the 'src' field in all log calls.
    def __init__(self, module_name: str) -> None:
        """Initialise the MarkerLogger with a module or component name.

        Args:
            module_name: The value used for the 'src' field (e.g. 'AuthService').
        """
        self._module_name = module_name

    # #endregion MarkerLogger.__init__

    # #region MarkerLogger.reason [TYPE Function]
    # @BRIEF Log a REASON marker — strict deduction, core logic.
    def reason(
        self, intent: str, *, payload: dict[str, Any] | None = None
    ) -> None:
        """Log a REASON marker (INFO level)."""
        log(self._module_name, "REASON", intent, payload=payload)

    # #endregion MarkerLogger.reason

    # #region MarkerLogger.reflect [TYPE Function]
    # @BRIEF Log a REFLECT marker — self-check, structural validation.
    def reflect(
        self, intent: str, *, payload: dict[str, Any] | None = None
    ) -> None:
        """Log a REFLECT marker (INFO level)."""
        log(self._module_name, "REFLECT", intent, payload=payload)

    # #endregion MarkerLogger.reflect

    # #region MarkerLogger.explore [TYPE Function]
    # @BRIEF Log an EXPLORE marker — searching, alternatives, violated assumptions.
    def explore(
        self,
        intent: str,
        *,
        payload: dict[str, Any] | None = None,
        error: str | None = None,
    ) -> None:
        """Log an EXPLORE marker (WARNING level).

        Args:
            intent: Short description of what was being attempted.
            payload: Optional structured data.
            error: Describes the violated assumption or error context.
        """
        log(self._module_name, "EXPLORE", intent, payload=payload, error=error)

    # #endregion MarkerLogger.explore


# #endregion MarkerLogger
# #endregion CotLoggerModule