log_format

log_format

Agent log formatters for structured container log output.

Provides pluggable formatters that transform raw container logs (e.g. Claude stream-json NDJSON) into human-readable, color-coded terminal output.

The AgentLogFormatter protocol defines the interface: call feed_line() for each log line, and finish() at the end for any summary output.

AgentLogFormatter

Bases: Protocol

Interface for agent log formatters.

feed_line(line)

Process one line of log output.

Source code in src/terok/lib/domain/log_format.py

def feed_line(self, line: str) -> None:
    """Process one line of log output."""
    ...

finish()

Called after all lines have been fed. Print any summary.

Source code in src/terok/lib/domain/log_format.py

def finish(self) -> None:
    """Called after all lines have been fed. Print any summary."""
    ...

PlainTextFormatter

Pass-through formatter that prints lines unchanged.

feed_line(line)

Print line as-is.

Source code in src/terok/lib/domain/log_format.py

def feed_line(self, line: str) -> None:
    """Print *line* as-is."""
    print(line, flush=True)

finish()

No-op; plain text has no summary.

Source code in src/terok/lib/domain/log_format.py

def finish(self) -> None:
    """No-op; plain text has no summary."""
    pass

ClaudeStreamJsonFormatter(*, streaming=True, color=None)

Formats Claude stream-json NDJSON into colored terminal output.

When streaming is True, processes content_block_start/delta/stop events for a typewriter effect on assistant text. When False, only processes coalesced messages (assistant, result, system).

Parameters:

Name	Type	Description	Default
streaming	bool	Enable partial streaming (typewriter text deltas).	True
color	bool | None	Enable ANSI colors (auto-detected from terminal if None).	None

Initialise formatter with streaming and color preferences.

Source code in src/terok/lib/domain/log_format.py

def __init__(self, *, streaming: bool = True, color: bool | None = None) -> None:
    """Initialise formatter with streaming and color preferences."""
    self._streaming = streaming
    self._color = color if color is not None else supports_color()
    self._state = _StreamState.IDLE
    self._tool_input_buf: list[str] = []
    self._current_tool_name: str = ""
    # Accumulated result for finish() summary
    self._result: dict | None = None

feed_line(line)

Parse a single NDJSON log line and print formatted output.

Source code in src/terok/lib/domain/log_format.py

def feed_line(self, line: str) -> None:
    """Parse a single NDJSON log line and print formatted output."""
    if not line.strip():
        return
    stripped = line.strip()
    try:
        data = json.loads(stripped)
    except (json.JSONDecodeError, ValueError):
        # Not JSON — print as plain text, preserving leading whitespace
        print(line.rstrip("\r\n"), flush=True)
        return

    msg_type = data.get("type", "")

    if msg_type == "system":
        self._handle_system(data)
    elif msg_type == "assistant":
        self._handle_assistant(data)
    elif msg_type == "user":
        self._handle_user(data)
    elif msg_type == "result":
        self._handle_result(data)
    elif self._streaming and msg_type == "content_block_start":
        self._handle_block_start(data)
    elif self._streaming and msg_type == "content_block_delta":
        self._handle_block_delta(data)
    elif self._streaming and msg_type == "content_block_stop":
        self._handle_block_stop(data)

finish()

Flush pending output and print the result summary if available.

Source code in src/terok/lib/domain/log_format.py

def finish(self) -> None:
    """Flush pending output and print the result summary if available."""
    # Flush any in-progress streaming block
    if self._state == _StreamState.TEXT_BLOCK:
        print(flush=True)
    elif self._state == _StreamState.TOOL_USE_BLOCK:
        accumulated = "".join(self._tool_input_buf)
        if accumulated:
            print(self._yellow(f"  {accumulated}"), flush=True)
    self._state = _StreamState.IDLE

    if self._result:
        self._print_result_summary()

auto_detect_formatter(mode, *, streaming=True, color=None, agent=None)

Return the appropriate formatter for a task's mode and agent.

Parameters:

Name	Type	Description	Default
mode	str | None	Task mode ("run" for headless/unattended, "cli", "web").	required
streaming	bool	Enable partial streaming for supported formatters.	True
color	bool | None	Force color on/off. None auto-detects from terminal.	None
agent	str | None	Headless agent name. When mode is "run" and agent is "claude" (or None), returns the Claude stream-json formatter. Other agents get plain text.	None

Source code in src/terok/lib/domain/log_format.py

def auto_detect_formatter(
    mode: str | None,
    *,
    streaming: bool = True,
    color: bool | None = None,
    agent: str | None = None,
) -> AgentLogFormatter:
    """Return the appropriate formatter for a task's mode and agent.

    Args:
        mode: Task mode (``"run"`` for headless/unattended, ``"cli"``, ``"web"``).
        streaming: Enable partial streaming for supported formatters.
        color: Force color on/off. ``None`` auto-detects from terminal.
        agent: Headless agent name.  When mode is ``"run"`` and
            agent is ``"claude"`` (or ``None``), returns the Claude
            stream-json formatter.  Other agents get plain text.
    """
    if mode == "run":
        effective_agent = agent or "claude"
        if effective_agent == "claude":
            return ClaudeStreamJsonFormatter(streaming=streaming, color=color)
        return PlainTextFormatter()
    return PlainTextFormatter()