Skip to content

forge.log

Structured JSON logging with automatic trace context propagation.

forge.log

Structured logging module — JSON and human-readable log output with context propagation.

Provides a module-aware logger factory, automatic trace ID propagation via contextvars, and configurable output formats for development and production environments.

Classes

DevFormatter

Bases: Formatter

Colorized human-readable log output for development.

Format::

[2026-06-28 12:00:00] module.name  INFO   message here
Source code in src/forge/log/formatters.py 
163
164
165
166
167
168
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
class DevFormatter(logging.Formatter):
    """
    Colorized human-readable log output for development.

    Format::

        [2026-06-28 12:00:00] module.name  INFO   message here
    """

    def format(self, record: logging.LogRecord) -> str:
        timestamp = datetime.datetime.fromtimestamp(record.created, tz=datetime.UTC).strftime(
            "%Y-%m-%d %H:%M:%S"
        )
        level = _LEVEL_LABELS.get(record.levelno, record.levelname)
        colour = _COLOUR_MAP.get(record.levelno, _GREY)

        message = record.getMessage()
        trace_id = get_trace_id()
        parts: list[str] = [
            f"{colour}[{timestamp}]{_RESET}",
            f"{record.name:<25}",
            f"{colour}{level:>5}{_RESET}",
            message,
        ]
        if trace_id:
            parts.insert(2, f"{_GREY}[{trace_id[:8]}]{_RESET}")
        elif _otel_available():
            otel_ctx = _get_otel_span_context()
            if otel_ctx:
                parts.insert(2, f"{_GREY}[{otel_ctx['trace_id'][:8]}]{_RESET}")

        if record.exc_info and record.exc_info[0] is not None:
            exc = "".join(traceback.format_exception(*record.exc_info))
            parts.append(f"\n{_RED}{exc}{_RESET}")

        return "  ".join(parts)

JSONFormatter

Bases: Formatter

Outputs structured JSON log lines for production use.

Every record produces: timestamp, level, module, message, and optionally trace_id (if one is active). Extra fields bound via LogContext are included at the top level. Circular references are replaced with "<circular>" and strings longer than 10 000 characters are truncated.

Source code in src/forge/log/formatters.py 
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
158
159
160
class JSONFormatter(logging.Formatter):
    """
    Outputs structured JSON log lines for production use.

    Every record produces: ``timestamp``, ``level``, ``module``,
    ``message``, and optionally ``trace_id`` (if one is active).
    Extra fields bound via ``LogContext`` are included at the top level.
    Circular references are replaced with ``"<circular>"`` and strings
    longer than 10 000 characters are truncated.
    """

    def format(self, record: logging.LogRecord) -> str:
        payload: dict[str, Any] = {
            "timestamp": datetime.datetime.fromtimestamp(
                record.created, tz=datetime.UTC
            ).isoformat(),
            "level": _LEVEL_LABELS.get(record.levelno, record.levelname),
            "module": record.name,
            "message": record.getMessage(),
        }

        trace_id = get_trace_id()
        if trace_id:
            payload["trace_id"] = trace_id

        if _otel_available():
            otel_ctx = _get_otel_span_context()
            if otel_ctx:
                payload.setdefault("otel_trace_id", otel_ctx["trace_id"])
                payload.setdefault("otel_span_id", otel_ctx["span_id"])

        if record.exc_info and record.exc_info[0] is not None:
            payload["exception"] = "".join(traceback.format_exception(*record.exc_info))

        extras = _extra_fields(record)
        payload.update(extras)

        safe = _truncate(_convert_for_json(payload))
        return json.dumps(safe, default=str)

LogContext

Async-safe context manager that binds extra fields to log entries.

Usage::

with LogContext(request_id="abc-123", user="alice"):
    logger.info("processing request")
    # → log entry includes request_id="abc-123", user="alice"
Source code in src/forge/log/context.py 
16
17
18
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
class LogContext:
    """
    Async-safe context manager that binds extra fields to log entries.

    Usage::

        with LogContext(request_id="abc-123", user="alice"):
            logger.info("processing request")
            # → log entry includes request_id="abc-123", user="alice"
    """

    def __init__(self, **kwargs: Any) -> None:
        self._extra = kwargs
        self._token: Token[dict[str, Any]] | None = None

    def __enter__(self) -> Self:
        current = dict(_log_context.get({}))
        current.update(self._extra)
        self._token = _log_context.set(current)
        return self

    def __exit__(self, *args: object) -> None:
        if self._token is not None:
            _log_context.reset(self._token)

    @staticmethod
    def get_current() -> dict[str, Any]:
        """Return a copy of the currently bound extra fields."""
        current = _log_context.get({})
        return dict(current)
Methods:
get_current staticmethod 
get_current() -> dict[str, Any]

Return a copy of the currently bound extra fields.

Source code in src/forge/log/context.py 
41
42
43
44
45
@staticmethod
def get_current() -> dict[str, Any]:
    """Return a copy of the currently bound extra fields."""
    current = _log_context.get({})
    return dict(current)

LogContextFilter

Bases: Filter

Logging filter that injects LogContext fields into every record.

Attach this filter to the root logger or handler to automatically propagate contextvars-based extra fields to all log entries::

root_logger.addFilter(LogContextFilter())
Source code in src/forge/log/context.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
class LogContextFilter(logging.Filter):
    """
    Logging filter that injects ``LogContext`` fields into every record.

    Attach this filter to the root logger or handler to automatically
    propagate contextvars-based extra fields to all log entries::

        root_logger.addFilter(LogContextFilter())
    """

    def filter(self, record: logging.LogRecord) -> bool:
        if _otel_available():
            ctx = get_current_span_context()
            if ctx:
                record.otel_trace_id = ctx["trace_id"]
                record.otel_span_id = ctx["span_id"]

        fields = _log_context.get({})
        for key, value in fields.items():
            if not hasattr(record, key):
                setattr(record, key, value)
        return True

LogModule

Bases: ForgeModule

Source code in src/forge/log/module.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
 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
148
149
150
151
class LogModule(ForgeModule):
    name = "log"
    dependencies: ClassVar[list[str]] = ["config"]

    def __init__(self) -> None:
        super().__init__()
        self._buffer: _BufferedHandler | None = None
        self._queue: Queue[logging.LogRecord] | None = None
        self._listener: QueueListener | None = None
        self._stream_handler: logging.StreamHandler[Any] | None = None
        self._queue_handler: QueueHandler | None = None
        self._context_filter: LogContextFilter | None = None

        self._install_buffer()

    # ── Public API ─────────────────────────────────────────────────

    def get_logger(self, name: str) -> logging.Logger:
        """Return a logger for *name* configured by this module."""
        logger = logging.getLogger(name)
        if self._context_filter is not None and logger.level == logging.NOTSET:
            logger.setLevel(logging.DEBUG)
        return logger

    def get(self, name: str) -> LoggerProxy:
        """Return a logger wrapped in LoggerProxy for *name*."""
        return LoggerProxy(self.get_logger(name))

    @contextmanager
    def context(self, **kwargs: Any) -> Generator[None, None, None]:
        """Context manager for binding key-value pairs to all log entries within the context."""
        from forge.log.context import LogContext

        with LogContext(**kwargs):
            yield

    # ── ForgeModule ────────────────────────────────────────────────

    async def setup(self, runtime: ForgeRuntime) -> None:
        init_otel()

        config_module: ConfigModule = runtime.get(ConfigModule)  # type: ignore[assignment]
        log_cfg = config_module.config.log

        level: int = getattr(logging, log_cfg.level.upper(), logging.INFO)

        if log_cfg.format == "json":
            formatter: logging.Formatter = JSONFormatter()
        else:
            formatter = DevFormatter()

        # Build non-blocking pipeline: logger → QueueHandler → Queue → QueueListener → StreamHandler
        self._stream_handler = logging.StreamHandler(sys.stderr)
        self._stream_handler.setFormatter(formatter)
        self._stream_handler.setLevel(level)

        self._queue = Queue(-1)
        self._queue_handler = QueueHandler(self._queue)
        self._queue_handler.setLevel(logging.DEBUG)

        self._listener = QueueListener(self._queue, self._stream_handler)
        self._listener.start()

        # Context filter injects LogContext fields into records
        self._context_filter = LogContextFilter()

        # Wire into root logger
        root = logging.getLogger(_ROOT)
        root.addFilter(self._context_filter)
        root.addHandler(self._queue_handler)
        root.setLevel(logging.DEBUG)

        # Remove the pre-init buffer and flush captured records
        if self._buffer is not None:
            root.removeHandler(self._buffer)
            self._buffer.flush_to(self._stream_handler)
            self._buffer = None

        # Set per-module log levels from config
        for module_name, level_name in log_cfg.levels.items():
            logger = logging.getLogger(module_name)
            logger.setLevel(getattr(logging, level_name.upper(), logging.NOTSET))

    async def teardown(self) -> None:
        if self._listener is not None:
            self._listener.stop()
            self._listener = None

        root = logging.getLogger(_ROOT)
        if self._queue_handler is not None:
            root.removeHandler(self._queue_handler)
            self._queue_handler = None

        self._queue = None
        self._stream_handler = None

    def health_check(self) -> HealthResult:
        if self._listener is None:
            return HealthResult.error("Log listener not running")
        if self._listener._thread is None or not self._listener._thread.is_alive():
            return HealthResult.error("Log listener thread died")
        return HealthResult.ok()

    # ── Internal ───────────────────────────────────────────────────

    def _install_buffer(self) -> None:
        """Install a buffered handler on the root logger before init."""
        self._buffer = _BufferedHandler()
        root = logging.getLogger(_ROOT)
        root.addHandler(self._buffer)
        root.setLevel(logging.DEBUG)
Methods:
context 
context(**kwargs: Any) -> Generator[None, None, None]

Context manager for binding key-value pairs to all log entries within the context.

Source code in src/forge/log/module.py 
69
70
71
72
73
74
75
@contextmanager
def context(self, **kwargs: Any) -> Generator[None, None, None]:
    """Context manager for binding key-value pairs to all log entries within the context."""
    from forge.log.context import LogContext

    with LogContext(**kwargs):
        yield
get 
get(name: str) -> LoggerProxy

Return a logger wrapped in LoggerProxy for name.

Source code in src/forge/log/module.py 
65
66
67
def get(self, name: str) -> LoggerProxy:
    """Return a logger wrapped in LoggerProxy for *name*."""
    return LoggerProxy(self.get_logger(name))
get_logger 
get_logger(name: str) -> logging.Logger

Return a logger for name configured by this module.

Source code in src/forge/log/module.py 
58
59
60
61
62
63
def get_logger(self, name: str) -> logging.Logger:
    """Return a logger for *name* configured by this module."""
    logger = logging.getLogger(name)
    if self._context_filter is not None and logger.level == logging.NOTSET:
        logger.setLevel(logging.DEBUG)
    return logger

LoggerProxy

Wraps a standard logging.Logger to support direct keyword arguments.

Example::

logger = log.get("module")
logger.info("user logged in", user_id=123, ip="127.0.0.1")
# → keyword arguments are merged into the structured extra fields.
Source code in src/forge/log/proxy.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
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
class LoggerProxy:
    """
    Wraps a standard logging.Logger to support direct keyword arguments.

    Example::

        logger = log.get("module")
        logger.info("user logged in", user_id=123, ip="127.0.0.1")
        # → keyword arguments are merged into the structured extra fields.
    """

    def __init__(self, logger: logging.Logger) -> None:
        self._logger = logger

    def debug(self, msg: str, *args: Any, **kwargs: Any) -> None:
        self._log(logging.DEBUG, msg, args, kwargs)

    def info(self, msg: str, *args: Any, **kwargs: Any) -> None:
        self._log(logging.INFO, msg, args, kwargs)

    def warning(self, msg: str, *args: Any, **kwargs: Any) -> None:
        self._log(logging.WARNING, msg, args, kwargs)

    def warn(self, msg: str, *args: Any, **kwargs: Any) -> None:
        self._log(logging.WARNING, msg, args, kwargs)

    def error(self, msg: str, *args: Any, **kwargs: Any) -> None:
        self._log(logging.ERROR, msg, args, kwargs)

    def critical(self, msg: str, *args: Any, **kwargs: Any) -> None:
        self._log(logging.CRITICAL, msg, args, kwargs)

    def exception(self, msg: str, *args: Any, **kwargs: Any) -> None:
        kwargs.setdefault("exc_info", True)
        self._log(logging.ERROR, msg, args, kwargs)

    def log(self, level: int, msg: str, *args: Any, **kwargs: Any) -> None:
        self._log(level, msg, args, kwargs)

    def _log(self, level: int, msg: str, args: tuple[Any, ...], kwargs: dict[str, Any]) -> None:
        # Standard logging parameters accepted by Logger._log/log:
        # exc_info, stack_info, stacklevel, extra
        std_keys = {"exc_info", "stack_info", "stacklevel"}
        extra = {}
        log_kwargs = {}

        for k, v in kwargs.items():
            if k in std_keys:
                log_kwargs[k] = v
            elif k == "extra":
                if isinstance(v, dict):
                    extra.update(v)
            else:
                extra[k] = v

        if extra:
            log_kwargs["extra"] = extra

        self._logger.log(level, msg, *args, **log_kwargs)

    def __getattr__(self, name: str) -> Any:
        return getattr(self._logger, name)

Functions:

get 

get(name: str) -> LoggerProxy

Get a module logger wrapped in LoggerProxy.

Delegates to the active runtime's LogModule if initialized, otherwise returns a default LoggerProxy.

Source code in src/forge/log/__init__.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
def get(name: str) -> LoggerProxy:
    """
    Get a module logger wrapped in LoggerProxy.

    Delegates to the active runtime's LogModule if initialized,
    otherwise returns a default LoggerProxy.
    """
    try:
        from forge.core.runtime import ForgeRuntime

        runtime = ForgeRuntime.get_active()
        log_module: LogModule = runtime.get(LogModule)  # type: ignore[assignment]
        return log_module.get(name)
    except Exception:
        # Before runtime init
        return LoggerProxy(logging.getLogger(name))