Display & logging

Live console output, structured logging, and fast trajectory stats. See the Display & logging guide.

Display

Display

Display(mode: DisplayMode | str = SHORT)

Mode-aware console display for agent events.

Uses rich.text.Text objects (not markup strings) to avoid escaping issues with model output containing [brackets].

Source code in caw/display.py

def __init__(self, mode: DisplayMode | str = DisplayMode.SHORT) -> None:
    if isinstance(mode, str):
        mode = DisplayMode(mode)
    self.mode = mode
    self.console = Console()
    self._lock = threading.RLock()
    self._last_result_text: str = ""
    self._pending_text: TextBlock | None = None

on_metadata

on_metadata(**kwargs: str) -> None

Print metadata key-value pairs (agent, model, session, etc.).

Source code in caw/display.py

def on_metadata(self, **kwargs: str) -> None:
    """Print metadata key-value pairs (agent, model, session, etc.)."""
    with self._lock:
        if self.mode == DisplayMode.OFF:
            return
        pairs = [f"{k}={v}" for k, v in kwargs.items() if v]
        if not pairs:
            return
        line = Text()
        line.append("[Metadata] ", style="dim bold")
        line.append("  ".join(pairs), style="dim")
        self.console.print(line)

on_user_message

on_user_message(message: str) -> None

Print the user's message.

Source code in caw/display.py

def on_user_message(self, message: str) -> None:
    """Print the user's message."""
    with self._lock:
        if self.mode in (DisplayMode.RESULT, DisplayMode.OFF):
            return

        line = Text()
        line.append("[User] ", style="bold green")
        if self.mode == DisplayMode.FULL:
            line.append(message, style="bold")
        else:
            line.append(_truncate(message), style="bold")
        self.console.print(line)

on_text

on_text(block: TextBlock) -> None

Buffer an assistant text block (printed on next event or at turn end).

Source code in caw/display.py

def on_text(self, block: TextBlock) -> None:
    """Buffer an assistant text block (printed on next event or at turn end)."""
    with self._lock:
        if self.mode == DisplayMode.OFF:
            return

        if self.mode == DisplayMode.RESULT:
            self._last_result_text = block.text
            return

        # Flush any previous text block (not the final one, so not bold)
        self._flush_pending_text(bold=False)
        self._pending_text = block

on_thinking

on_thinking(block: ThinkingBlock) -> None

Print a thinking block.

Source code in caw/display.py

def on_thinking(self, block: ThinkingBlock) -> None:
    """Print a thinking block."""
    with self._lock:
        if self.mode in (DisplayMode.RESULT, DisplayMode.OFF):
            return

        line = Text()
        line.append("[Thinking] ", style="dim magenta")
        if self.mode == DisplayMode.FULL:
            line.append(block.text, style="dim")
        else:
            line.append(_truncate(block.text), style="dim")
        self.console.print(line)

on_tool_call

on_tool_call(block: ToolUse) -> None

Print a tool call (args only — result not yet known).

Source code in caw/display.py

def on_tool_call(self, block: ToolUse) -> None:
    """Print a tool call (args only — result not yet known)."""
    with self._lock:
        if self.mode in (DisplayMode.RESULT, DisplayMode.OFF):
            return

        self._flush_pending_text(bold=False)

        line = Text()
        line.append("[Tool] ", style="bold yellow")
        line.append(block.name, style="bold cyan")
        line.append(" ")

        if self.mode == DisplayMode.FULL:
            args_str = json.dumps(block.arguments, indent=2)
            line.append(args_str, style="dim")
        else:
            args_str = json.dumps(block.arguments, separators=(",", ":"))
            line.append(_truncate(args_str), style="dim")
        self.console.print(line)

on_tool_result

on_tool_result(block: ToolUse) -> None

Print a tool result (output now available on the block).

Source code in caw/display.py

def on_tool_result(self, block: ToolUse) -> None:
    """Print a tool result (output now available on the block)."""
    with self._lock:
        if self.mode in (DisplayMode.RESULT, DisplayMode.OFF):
            return

        tag_style = "bold red" if block.is_error else "bold yellow"
        line = Text()
        line.append("[Result] ", style=tag_style)
        line.append(block.name, style="bold cyan")

        output = block.output
        if output:
            line.append("\n")
            text = self.mode == DisplayMode.FULL and output or _first_n_lines(output)
            # Parse ANSI escapes so colorful tool output keeps its colors;
            # uncolored portions fall back to the dim base style.
            result_text = Text.from_ansi(text, style="dim")
            line.append_text(result_text)
        self.console.print(line)

on_turn_end

on_turn_end(result: str, usage: UsageStats, duration_ms: int) -> None

Print end-of-turn stats or deferred result text.

Source code in caw/display.py

def on_turn_end(self, result: str, usage: UsageStats, duration_ms: int) -> None:
    """Print end-of-turn stats or deferred result text."""
    with self._lock:
        if self.mode == DisplayMode.OFF:
            return

        if self.mode == DisplayMode.RESULT:
            if self._last_result_text:
                self.console.print(Panel(self._last_result_text, border_style="green", expand=False))
                self._last_result_text = ""
            return

        # Flush the last text block as bold (it's the final assistant message)
        self._flush_pending_text(bold=True)

        # Stats as metadata
        tokens = f"{usage.input_tokens}in/{usage.output_tokens}out"
        meta: dict[str, str] = {
            "duration": f"{duration_ms}ms",
            "tokens": tokens,
        }
        if usage.cost_usd:
            meta["cost"] = f"${usage.cost_usd:.4f}"
        self.on_metadata(**meta)

DisplayMode

DisplayMode

Bases: str, Enum

Print modes for agent output.

Global display

get_global_display

get_global_display() -> Display | None

Return the global Display. Falls back to CAW_LOG env var on first call.

Source code in caw/display.py

def get_global_display() -> Display | None:
    """Return the global Display. Falls back to CAW_LOG env var on first call."""
    global _global_display, _global_display_resolved
    if _global_display is None and not _global_display_resolved:
        _global_display_resolved = True
        env_mode = os.environ.get(LOG_ENV_VAR, "short")
        _global_display = Display(mode=env_mode)
    return _global_display

set_global_display

set_global_display(display: Display | None) -> None

Set (or clear) the global Display instance.

Source code in caw/display.py

def set_global_display(display: Display | None) -> None:
    """Set (or clear) the global Display instance."""
    global _global_display, _global_display_resolved
    _global_display = display
    _global_display_resolved = True

AgentLogger

AgentLogger

Bases: Protocol

Minimal logger surface used by caw.

FastStats

FastStats dataclass

FastStats(path: Optional[Path] = None, agent: str = '', model: str = '', session_id: str = '', created_at: str = '', completed_at: str = '', usage_limited: bool = False, duration_ms: int = 0, cost_usd: float = 0.0, input_tokens: int = 0, output_tokens: int = 0, cache_read_tokens: int = 0, cache_write_tokens: int = 0)

Lightweight statistics for a CAW trajectory.

The class is intentionally narrow: it exposes only the fields that consumers ask for repeatedly without paying for a full trajectory parse. For everything else (turns, tool calls, content blocks) load the file via caw.agent.Session.load_trajectory instead.

All cost_usd / token values come from the trajectory's total_usage (recursive across subagents) when present, falling back to usage for older trajectories that did not record it separately.

to_dict

to_dict() -> dict[str, Any]

Return a JSON-serializable dict (path is stringified).

Source code in caw/faststats.py

def to_dict(self) -> dict[str, Any]:
    """Return a JSON-serializable dict (``path`` is stringified)."""
    d = asdict(self)
    d["path"] = str(self.path) if self.path is not None else None
    return d

from_trajectory classmethod

from_trajectory(trajectory: Trajectory, *, path: str | Path | None = None) -> FastStats

Build FastStats from an in-memory Trajectory.

Source code in caw/faststats.py

@classmethod
def from_trajectory(cls, trajectory: Trajectory, *, path: str | Path | None = None) -> FastStats:
    """Build `FastStats` from an in-memory `Trajectory`."""
    usage = trajectory.total_usage
    return cls(
        path=Path(path) if path is not None else None,
        agent=trajectory.agent,
        model=trajectory.model,
        session_id=trajectory.session_id,
        created_at=trajectory.created_at,
        completed_at=trajectory.completed_at,
        usage_limited=trajectory.usage_limited,
        duration_ms=trajectory.duration_ms,
        cost_usd=usage.cost_usd,
        input_tokens=usage.input_tokens,
        output_tokens=usage.output_tokens,
        cache_read_tokens=usage.cache_read_tokens,
        cache_write_tokens=usage.cache_write_tokens,
    )

from_path classmethod

from_path(path: str | Path) -> Optional[FastStats]

Read fast stats from path.

Returns None if the file does not exist, is empty, or is not a recognizable trajectory file. Tries the head/tail fast path first and falls back to a full JSON parse on failure.

Source code in caw/faststats.py

@classmethod
def from_path(cls, path: str | Path) -> Optional[FastStats]:
    """Read fast stats from *path*.

    Returns ``None`` if the file does not exist, is empty, or is not a
    recognizable trajectory file. Tries the head/tail fast path first
    and falls back to a full JSON parse on failure.
    """
    path = Path(path)
    try:
        size = path.stat().st_size
    except OSError:
        return None
    if size == 0:
        return None

    try:
        with open(path, "rb") as f:
            head_len = min(_HEAD_BYTES, size)
            head = f.read(head_len).decode("utf-8", errors="replace")
            if size <= _HEAD_BYTES:
                tail = head
            elif size <= _HEAD_BYTES + _TAIL_BYTES:
                tail = head + f.read().decode("utf-8", errors="replace")
            else:
                f.seek(size - _TAIL_BYTES)
                tail = f.read(_TAIL_BYTES).decode("utf-8", errors="replace")
    except OSError:
        return None

    stats = cls._fast_extract(head, tail, path)
    if stats is not None:
        return stats

    # Fallback: parse the full document. ``Trajectory.from_dict``
    # tolerates missing keys, so we explicitly require ``model`` to be
    # present and non-empty before treating the file as a trajectory.
    try:
        data = json.loads(path.read_bytes())
    except (OSError, ValueError):
        return None
    if not isinstance(data, dict) or not data.get("model"):
        return None
    try:
        traj = Trajectory.from_dict(data)
    except (ValueError, KeyError, TypeError):
        return None
    return cls.from_trajectory(traj, path=path)

iter_directory classmethod

iter_directory(directory: str | Path, *, patterns: Iterable[str] = ('**/trajectory.json', '**/*.traj.json'), skip_parts: Iterable[str] = ()) -> Iterator[FastStats]

Yield FastStats for every trajectory file under directory.

patterns is a list of globs (relative to directory) to scan; the default catches both the canonical CAW layout (sessions/<id>/trajectory.json) and the .traj.json files produced by ad-hoc writers. Files whose path contains any directory component listed in skip_parts are excluded. Unreadable or malformed files are silently dropped.

Source code in caw/faststats.py

@classmethod
def iter_directory(
    cls,
    directory: str | Path,
    *,
    patterns: Iterable[str] = ("**/trajectory.json", "**/*.traj.json"),
    skip_parts: Iterable[str] = (),
) -> Iterator[FastStats]:
    """Yield `FastStats` for every trajectory file under *directory*.

    ``patterns`` is a list of globs (relative to *directory*) to scan;
    the default catches both the canonical CAW layout
    (``sessions/<id>/trajectory.json``) and the ``.traj.json`` files
    produced by ad-hoc writers. Files whose path contains any directory
    component listed in ``skip_parts`` are excluded. Unreadable or
    malformed files are silently dropped.
    """
    directory = Path(directory)
    if not directory.is_dir():
        return
    skip_set = set(skip_parts)
    seen: set[Path] = set()
    for pattern in patterns:
        for file in directory.glob(pattern):
            if file in seen or not file.is_file():
                continue
            seen.add(file)
            if skip_set:
                try:
                    rel_parts = file.relative_to(directory).parts
                except ValueError:
                    rel_parts = file.parts
                if any(part in skip_set for part in rel_parts):
                    continue
            stats = cls.from_path(file)
            if stats is not None:
                yield stats

directory_total_cost classmethod

directory_total_cost(directory: str | Path, **kwargs: Any) -> float

Sum cost_usd across every trajectory under directory.

Extra keyword arguments are forwarded to iter_directory.

Source code in caw/faststats.py

@classmethod
def directory_total_cost(
    cls,
    directory: str | Path,
    **kwargs: Any,
) -> float:
    """Sum ``cost_usd`` across every trajectory under *directory*.

    Extra keyword arguments are forwarded to `iter_directory`.
    """
    return sum(s.cost_usd for s in cls.iter_directory(directory, **kwargs))