trace_view.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/core/trace_view.py
 
"""
Read, format, and summarise the persisted M2 event stream — M2.6.
 
Purpose
-------
M2.1–M2.5 produce a per-session NDJSON event log (``events.NNNN.jsonl`` under
the writer's base dir). This module is the first *reader* of that log: it
locates a session's event files, parses each line back into a frozen
``TurnEvent``, renders a compact human-readable trace (for ``/trace``), and
assembles a minimal machine-readable ``packet-draft/v0`` bundle (for the draft
``/packet``). The full merge-readiness packet is M7; this is observability only.
 
Responsibilities
----------------
- :func:`read_session_events` — resolve the session's event dir, read every
  rotation slot in order, parse lines into ``TurnEvent``s, tolerate missing
  dirs and torn/malformed lines.
- :func:`format_trace` — compact per-event text rendering with an optional tail
  limit and a trailing per-kind count summary.
- :func:`build_packet_draft` — a pure ``dict`` snapshot (no disk writes) of the
  session identity, per-kind counts, turn count, and last final answer.
 
Diagnostics
-----------
Domain: OC_CHAT.TRACE
Levels:
    L2 — read/packet lifecycle.
    L3 — file/line counts and skipped malformed lines.
    L4 — module-load coverage marker.
 
Contracts
---------
- Reading never raises on missing dirs or torn lines; it returns ``[]`` or
  skips the bad line and counts it.
- ``build_packet_draft`` is pure and writes nothing.
- The packet schema is ``"packet-draft/v0"`` — explicitly a draft distinct from
  the M7 packet.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.core.event_bootstrap (active_event_dir — locate the session's log)
- oc_chat.core.events (TurnEvent — parse persisted lines)
- oc_chat.core.session (lazy; ChatSession, for type hints only)
"""
from __future__ import annotations
 
import re
from datetime import datetime, timezone
from pathlib import Path
from typing import TYPE_CHECKING, Any, Optional
 
from oc_diagnostics import _dbg
 
from oc_chat.core.event_bootstrap import active_event_dir
from oc_chat.core.events import TurnEvent
 
if TYPE_CHECKING:
    from oc_chat.core.session import ChatSession
 
_dbg("OC_CHAT.TRACE", "trace_view", "module", 4)
 
PACKET_SCHEMA = "packet-draft/v0"
_SLOT_RE = re.compile(r"^events\.(\d+)\.jsonl$")
_DEFAULT_TRACE_LIMIT = 200
 
 
def _slot_seq(path: Path) -> int:
    """Return the rotation sequence encoded in an ``events.NNNN.jsonl`` filename."""
    match = _SLOT_RE.match(path.name)
    return int(match.group(1)) if match else 0
 
 
class EventHistoryCorruption(RuntimeError):
    """Two persisted lines share an ``event_id`` but differ in bytes.
 
    Surfaced rather than silently resolved: picking one of two conflicting
    records for the same event id would fabricate an audit trail. Fail closed
    (plan 152 ADR-8 / R10-05).
    """
 
 
def _read_one_root(base: Path, session_id: str) -> tuple[list[TurnEvent], int, int]:
    """Parse one candidate root, keeping ONLY events whose payload sid matches.
 
    Returns ``(events, torn, foreign)`` — torn/malformed lines and lines
    belonging to a DIFFERENT session are each counted, never echoed (an event
    payload carries prompts, assistant text and tool arguments).
    """
    slots = sorted(
        (p for p in base.iterdir() if p.is_file() and _SLOT_RE.match(p.name)),
        key=_slot_seq,
    )
    events: list[TurnEvent] = []
    torn = 0
    foreign = 0
    for slot in slots:
        try:
            text = slot.read_text(encoding="utf-8")
        except Exception:
            torn += 1
            continue
        for line in text.splitlines():
            if not line.strip():
                continue
            try:
                event = TurnEvent.model_validate_json(line)
            except Exception:
                torn += 1                       # a crash can leave the last line partial
                continue
            if str(getattr(event, "session_id", "") or "") != session_id:
                foreign += 1                    # a directory is a SOURCE, never an identity
                continue
            events.append(event)
    return events, torn, foreign
 
 
def read_session_events(session: "ChatSession") -> list[TurnEvent]:
    """Return every persisted ``TurnEvent`` for *session*, in write order.
 
    Reads the IDENTITY-FILTERED MERGE of every candidate root that
    :func:`~oc_chat.core.event_bootstrap.event_read_dirs` reports — the
    ``session_id`` dir plus any legacy name-slug / persisted-alias / old direct
    override roots — instead of a single directory (plan 152 ADR-8). Deriving
    the write dir from the immutable ``session_id`` would otherwise hide a
    pre-migration session's history behind the new directory.
 
    Guarantees:
 
    - **Identity filter.** Only events whose payload ``session_id`` equals this
      session's are included. Legacy slug directories are measurably NOT
      identity partitions — real ones hold several sessions' streams — so
      merging by directory name would attach another session's prompts and tool
      arguments to this trace. Mismatches are skipped and COUNTED, never echoed.
    - **Dedupe by ``event_id``.** The same event reachable through two roots
      appears once. Two lines sharing an ``event_id`` but differing in bytes are
      CORRUPTION and raise :class:`EventHistoryCorruption` — no silent pick.
    - **Total order.** All legacy-root events precede all ``session_id``-dir
      events (a migration-epoch fact: post-cutover writes go only to the id
      dir), and within each root, slot order then line order — the writer's own
      order. Equal timestamps never decide anything.
    - **Read-only.** Nothing here creates, mutates or dirties session state.
 
    A torn/malformed line is skipped and counted; missing dirs yield no events.
    """
    func = "read_session_events"
    from oc_chat.core.event_bootstrap import event_read_dirs
 
    session_id = str(getattr(session, "session_id", "") or "")
    roots = event_read_dirs(session)
    events: list[TurnEvent] = []
    by_id: dict[str, str] = {}
    torn_total = 0
    foreign_total = 0
    contributing_legacy = 0
 
    for index, base in enumerate(roots):
        if not base.exists():
            continue
        found, torn, foreign = _read_one_root(base, session_id)
        torn_total += torn
        foreign_total += foreign
        if found and index < len(roots) - 1:
            contributing_legacy += 1
        for event in found:
            event_id = str(getattr(event, "event_id", "") or "")
            payload = event.model_dump_json()
            if event_id:
                previous = by_id.get(event_id)
                if previous is not None:
                    if previous != payload:
                        raise EventHistoryCorruption(
                            f"event_id {event_id} appears twice with different content "
                            f"across event roots for session {session_id}"
                        )
                    continue                    # exact duplicate across roots — keep one
                by_id[event_id] = payload
            events.append(event)
 
    if contributing_legacy > 1:
        # Legacy-only ambiguity: cross-root interleave cannot be recovered, and a
        # timestamp guess would invent an order. Surfaced, never resolved.
        _dbg("OC_CHAT.TRACE", func,
             f"{contributing_legacy} legacy roots contributed events; "
             f"cross-root interleave order is indeterminate (legacy-before-id order applied)", 2)
    _dbg("OC_CHAT.TRACE", func,
         f"read {len(events)} events from {len(roots)} root(s); "
         f"torn={torn_total} foreign_skipped={foreign_total}", 3)
    return events
 
 
def _short(value: Optional[str], width: int = 8) -> str:
    """Return the leading *width* chars of *value* (``"-"`` when empty)."""
    return (value or "-")[:width]
 
 
def _event_summary(event: TurnEvent) -> str:
    """Return a one-line key-field summary for *event* by kind."""
    p: Any = event.payload
    if event.kind == "user_input_received":
        return f"text={_short(getattr(p, 'text', ''), 60)!r}"
    if event.kind == "assistant_message":
        final = getattr(p, "is_final", False)
        return f"final={final} text={_short(getattr(p, 'text', ''), 60)!r}"
    if event.kind == "tool_call_proposed":
        return f"tool={getattr(p, 'tool_name', '?')}"
    if event.kind == "tool_call_started":
        return f"tool={getattr(p, 'tool_name', '?')} id={_short(getattr(p, 'tool_call_id', ''))}"
    if event.kind == "tool_call_completed":
        return f"id={_short(getattr(p, 'tool_call_id', ''))} outcome={getattr(p, 'outcome', '?')}"
    if event.kind in ("subagent_started", "subagent_completed"):
        sid = _short(getattr(p, "subagent_session_id", ""))
        outcome = getattr(p, "outcome", None)
        return f"subagent={sid}" + (f" outcome={outcome}" if outcome else "")
    if event.kind == "error":
        return f"error={_short(getattr(p, 'error_summary', ''), 60)!r}"
    return ""
 
 
def format_trace(events: list[TurnEvent], *, limit: Optional[int] = _DEFAULT_TRACE_LIMIT) -> str:
    """Render *events* as compact lines + a per-kind count summary.
 
    When *limit* is set, only the last *limit* events are rendered (the summary
    still counts the full stream).
    """
    if not events:
        return "(no events recorded for this session yet)"
    shown = events[-limit:] if limit else events
    lines = ["── Event trace ──"]
    if len(shown) < len(events):
        lines.append(f"(showing last {len(shown)} of {len(events)} events)")
    for ev in shown:
        lines.append(f"  {ev.ts}  {ev.kind:<22} turn={_short(ev.turn_id)}  {_event_summary(ev)}")
    counts: dict[str, int] = {}
    for ev in events:
        counts[ev.kind] = counts.get(ev.kind, 0) + 1
    lines.append("── Summary ──")
    for kind in sorted(counts):
        lines.append(f"  {kind:<22} {counts[kind]}")
    return "\n".join(lines)
 
 
def build_packet_draft(session: "ChatSession", events: list[TurnEvent]) -> dict:
    """Assemble a minimal ``packet-draft/v0`` dict for *session* (pure; no I/O).
 
    The full merge-readiness packet (diff, tests, rollback feasibility,
    inverse-patches, signing) is M7; this draft proves the stream is consumable.
    """
    counts: dict[str, int] = {}
    turn_ids: set[str] = set()
    last_final: Optional[str] = None
    for ev in events:
        counts[ev.kind] = counts.get(ev.kind, 0) + 1
        if ev.turn_id:
            turn_ids.add(ev.turn_id)
        if ev.kind == "assistant_message" and getattr(ev.payload, "is_final", False):
            last_final = getattr(ev.payload, "text", None)
    return {
        "schema": PACKET_SCHEMA,
        "session_id": getattr(session, "session_id", ""),
        "session_name": session.name,
        "model": session.model,
        "active_plan_id": getattr(session, "active_plan_id", None),
        "event_counts": counts,
        "turn_count": len(turn_ids),
        "last_final_answer": last_final,
        "generated_at": datetime.now(timezone.utc).isoformat(timespec="milliseconds"),
    }