core/scripts/review_trace.py

#!/usr/bin/env python3
"""
review_trace.py — JSONL trace reader for CORE.

Reads trace files produced by run_examples.py or `core session` and
prints formatted reports for post-hoc determinism inspection.

Usage:
    python scripts/review_trace.py traces/field_probe.jsonl
    python scripts/review_trace.py traces/identity_pressure.jsonl --summary
    python scripts/review_trace.py traces/fatigue_arc.jsonl --fatigue
    python scripts/review_trace.py traces/versor_drift.jsonl --drift
    python scripts/review_trace.py traces/dialogue_memory.jsonl --turn 2
    python scripts/review_trace.py traces/identity_pressure.jsonl --flagged
    python scripts/review_trace.py traces/identity_pressure.jsonl --identity
    python scripts/review_trace.py traces/*.jsonl --summary  (glob supported)
"""
from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path
from typing import Iterator


# ---------------------------------------------------------------------------
# JSONL loading
# ---------------------------------------------------------------------------

def _load_jsonl(path: Path) -> list[dict]:
    events = []
    with path.open(encoding="utf-8") as f:
        for lineno, line in enumerate(f, 1):
            line = line.strip()
            if not line:
                continue
            try:
                events.append(json.loads(line))
            except json.JSONDecodeError as exc:
                print(f"warning: {path}:{lineno}: {exc}", file=sys.stderr)
    return events


def _load_all(paths: list[Path]) -> dict[str, list[dict]]:
    result = {}
    for p in paths:
        result[p.stem] = _load_jsonl(p)
    return result


# ---------------------------------------------------------------------------
# Report views
# ---------------------------------------------------------------------------

_COL = {
    "turn": 4, "input": 30, "surface": 28, "role": 16,
    "score": 6, "cost": 6, "flagged": 7, "versor": 9,
}


def _header() -> str:
    return (
        f"{'TURN':>{_COL['turn']}}  "
        f"{'INPUT':<{_COL['input']}}  "
        f"{'SURFACE':<{_COL['surface']}}  "
        f"{'ROLE':<{_COL['role']}}  "
        f"{'SCORE':>{_COL['score']}}  "
        f"{'COST':>{_COL['cost']}}  "
        f"{'FLAGGED':<{_COL['flagged']}}  "
        f"{'VERSOR':>{_COL['versor']}}"
    )


def _row(e: dict) -> str:
    score = (e.get("identity_score") or {})
    score_val = score.get("value", "-")
    score_str = f"{score_val:.3f}" if isinstance(score_val, float) else str(score_val)
    inp = " ".join(e.get("input_tokens", []))
    return (
        f"{e.get('turn', '?'):>{_COL['turn']}}  "
        f"{inp[:_COL['input']]:<{_COL['input']}}  "
        f"{str(e.get('surface', ''))[:_COL['surface']]:<{_COL['surface']}}  "
        f"{str(e.get('dialogue_role', ''))[:_COL['role']]:<{_COL['role']}}  "
        f"{score_str:>{_COL['score']}}  "
        f"{e.get('cycle_cost_total', 0.0):>{_COL['cost']}.1f}  "
        f"{'YES' if e.get('flagged') else 'no':<{_COL['flagged']}}  "
        f"{e.get('versor_condition', 0.0):>{_COL['versor']}.2e}"
    )


def view_summary(name: str, events: list[dict]) -> None:
    print(f"\n─── {name} ({len(events)} turns) ───")
    print(_header())
    print("─" * (sum(_COL.values()) + len(_COL) * 2 + 2))
    for e in events:
        print(_row(e))


def view_turn(name: str, events: list[dict], turn: int) -> None:
    matches = [e for e in events if e.get("turn") == turn]
    if not matches:
        print(f"{name}: no turn {turn}")
        return
    e = matches[0]
    print(f"\n─── {name} / turn {turn} ───")
    print(f"  input_tokens      : {e.get('input_tokens')}")
    print(f"  surface           : {e.get('surface')!r}")
    print(f"  walk_surface      : {e.get('walk_surface')!r}")
    print(f"  articulation_surf : {e.get('articulation_surface')!r}")
    print(f"  dialogue_role     : {e.get('dialogue_role')}")
    print(f"  versor_condition  : {e.get('versor_condition', 0.0):.4e}")
    print(f"  cycle_cost_total  : {e.get('cycle_cost_total', 0.0):.2f}")
    print(f"  vault_hits        : {e.get('vault_hits')}")
    print(f"  flagged           : {e.get('flagged')}")
    score = e.get("identity_score") or {}
    if score:
        print(f"  identity_score:")
        print(f"    value           : {score.get('value')}")
        print(f"    flagged         : {score.get('flagged')}")
        print(f"    alignment       : {score.get('alignment')}")
        print(f"    axes_evaluated  : {score.get('axes_evaluated')}")
    prop = e.get("proposition") or {}
    if prop:
        print(f"  proposition:")
        print(f"    subject         : {prop.get('subject')!r}")
        print(f"    predicate       : {prop.get('predicate')!r}")
        print(f"    object          : {prop.get('object')!r}")
        print(f"    frame_id        : {prop.get('frame_id')}")
        print(f"    relation_norm   : {prop.get('relation_norm', 0.0):.4f}")


def view_flagged(name: str, events: list[dict]) -> None:
    flagged = [e for e in events if e.get("flagged")]
    print(f"\n─── {name} ─ flagged turns ({len(flagged)}/{len(events)}) ───")
    if not flagged:
        print("  none")
        return
    print(_header())
    for e in flagged:
        print(_row(e))


def view_drift(name: str, events: list[dict]) -> None:
    print(f"\n─── {name} ─ versor_condition drift ───")
    print(f"  {'TURN':>4}  {'VERSOR_CONDITION':>18}  {'DELTA':>14}")
    prev = None
    for e in events:
        val = float(e.get("versor_condition", 0.0))
        delta = (val - prev) if prev is not None else 0.0
        sign = "+" if delta >= 0 else ""
        print(f"  {e.get('turn', '?'):>4}  {val:>18.6e}  {sign}{delta:>13.6e}")
        prev = val


def view_identity(name: str, events: list[dict]) -> None:
    print(f"\n─── {name} ─ identity scores ───")
    print(f"  {'TURN':>4}  {'VALUE':>8}  {'ALIGNMENT':>10}  {'FLAGGED':<8}  AXES")
    for e in events:
        score = e.get("identity_score") or {}
        val = score.get("value", "-")
        val_str = f"{val:.4f}" if isinstance(val, float) else str(val)
        aln = score.get("alignment", "-")
        aln_str = f"{aln:.4f}" if isinstance(aln, float) else str(aln)
        axes = ", ".join(str(a) for a in (score.get("axes_evaluated") or []))
        print(
            f"  {e.get('turn', '?'):>4}  {val_str:>8}  {aln_str:>10}  "
            f"{'YES' if score.get('flagged') else 'no':<8}  {axes}"
        )


def view_fatigue(name: str, events: list[dict]) -> None:
    print(f"\n─── {name} ─ fatigue arc (cycle_cost_total) ───")
    print(f"  {'TURN':>4}  {'COST':>8}  {'CUMULATIVE':>12}  {'VAULT_HITS':>10}")
    cumulative = 0.0
    for e in events:
        cost = float(e.get("cycle_cost_total", 0.0))
        cumulative += cost
        print(
            f"  {e.get('turn', '?'):>4}  {cost:>8.2f}  {cumulative:>12.2f}  "
            f"{e.get('vault_hits', 0):>10}"
        )


# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------

def main() -> int:
    parser = argparse.ArgumentParser(
        description="Review CORE JSONL trace files.",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=(
            "Examples:\n"
            "  python scripts/review_trace.py traces/field_probe.jsonl\n"
            "  python scripts/review_trace.py traces/identity_pressure.jsonl --flagged\n"
            "  python scripts/review_trace.py traces/fatigue_arc.jsonl --fatigue\n"
            "  python scripts/review_trace.py traces/dialogue_memory.jsonl --turn 3\n"
            "  python scripts/review_trace.py traces/versor_drift.jsonl --drift\n"
        ),
    )
    parser.add_argument(
        "files",
        nargs="+",
        help="JSONL trace file(s) to review",
    )
    parser.add_argument(
        "--summary",
        action="store_true",
        default=True,
        help="print one-line-per-turn summary table (default)",
    )
    parser.add_argument(
        "--turn",
        type=int,
        default=None,
        metavar="N",
        help="print full detail for turn N",
    )
    parser.add_argument(
        "--flagged",
        action="store_true",
        help="show only flagged turns",
    )
    parser.add_argument(
        "--drift",
        action="store_true",
        help="print versor_condition per turn",
    )
    parser.add_argument(
        "--identity",
        action="store_true",
        help="print identity_score breakdown per turn",
    )
    parser.add_argument(
        "--fatigue",
        action="store_true",
        help="print cycle_cost_total per turn (exertion arc)",
    )
    args = parser.parse_args()

    paths = [Path(f) for f in args.files]
    missing = [p for p in paths if not p.exists()]
    if missing:
        for p in missing:
            print(f"error: file not found: {p}", file=sys.stderr)
        return 1

    traces = _load_all(paths)
    any_view_requested = any([
        args.turn is not None,
        args.flagged,
        args.drift,
        args.identity,
        args.fatigue,
    ])

    for name, events in traces.items():
        if args.turn is not None:
            view_turn(name, events, args.turn)
        if args.flagged:
            view_flagged(name, events)
        if args.drift:
            view_drift(name, events)
        if args.identity:
            view_identity(name, events)
        if args.fatigue:
            view_fatigue(name, events)
        if not any_view_requested or args.summary:
            view_summary(name, events)

    return 0


if __name__ == "__main__":
    raise SystemExit(main())