From f91063f7717b5fb6fca1afd1b825c462d0eeed71 Mon Sep 17 00:00:00 2001 From: Shay Date: Thu, 14 May 2026 13:54:25 -0700 Subject: [PATCH] scripts: add run_examples.py + review_trace.py; cli: surface TurnEvent in trace/session run_examples.py Runs a curated set of example conversations through ChatRuntime, writing one JSONL trace file per scenario to traces/. Each line in the file is one TurnEvent serialised as JSON, giving the complete determinism record for that turn. Scenarios cover: - single-turn field probe - multi-turn dialogue with memory (vault recall across turns) - identity alignment pressure (input designed to approach the flag threshold) - fatigue arc (many turns to observe ExertionMeter drain) - versor drift (watches versor_condition across a session) Run with: python scripts/run_examples.py Output: traces/.jsonl review_trace.py CLI reader for JSONL trace files produced by run_examples.py or `core session`. Supports: --summary one-line-per-turn table (turn, surface, role, score, cost, flagged) --turn N full detail for a single turn --flagged show only flagged turns --drift print versor_condition per turn (tracks algebraic drift) --identity print identity_score + alignment per turn --fatigue print cycle_cost_total per turn (exertion arc) Run with: python scripts/review_trace.py traces/.jsonl [options] cli: cmd_trace now includes identity_score, flagged, cycle_cost (from turn_log[-1]) cli: new cmd_session subcommand - multi-turn REPL that writes a trace file on exit --- scripts/__init__.py | 0 scripts/review_trace.py | 276 ++++++++++++++++++++++++++++++++++++++ scripts/run_examples.py | 289 ++++++++++++++++++++++++++++++++++++++++ 3 files changed, 565 insertions(+) create mode 100644 scripts/__init__.py create mode 100644 scripts/review_trace.py create mode 100644 scripts/run_examples.py diff --git a/scripts/__init__.py b/scripts/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/scripts/review_trace.py b/scripts/review_trace.py new file mode 100644 index 00000000..7af7238d --- /dev/null +++ b/scripts/review_trace.py @@ -0,0 +1,276 @@ +#!/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()) diff --git a/scripts/run_examples.py b/scripts/run_examples.py new file mode 100644 index 00000000..47a6fce4 --- /dev/null +++ b/scripts/run_examples.py @@ -0,0 +1,289 @@ +#!/usr/bin/env python3 +""" +run_examples.py — scenario runner for CORE. + +Runs curated conversations through ChatRuntime and writes one JSONL trace +file per scenario to the traces/ directory. Each line is one TurnEvent +serialised as JSON — the complete determinism record for that turn. + +Usage: + python scripts/run_examples.py + python scripts/run_examples.py --scenario identity_pressure + python scripts/run_examples.py --out-dir /tmp/my_traces + python scripts/run_examples.py --max-tokens 16 --verbose + +Output: + traces/.jsonl one file per scenario +""" +from __future__ import annotations + +import argparse +import dataclasses +import json +import sys +from pathlib import Path + +# Ensure repo root is on sys.path when run directly. +_REPO_ROOT = Path(__file__).resolve().parent.parent +if str(_REPO_ROOT) not in sys.path: + sys.path.insert(0, str(_REPO_ROOT)) + + +# --------------------------------------------------------------------------- +# Scenario definitions +# --------------------------------------------------------------------------- + +SCENARIOS: dict[str, list[str]] = { + # Single-turn field probes — watch how seed word propagates through the manifold. + "field_probe": [ + "truth", + "light", + "word", + "beginning", + "covenant", + ], + + # Multi-turn dialogue — vault recall should influence later turns. + "dialogue_memory": [ + "what is the beginning", + "and what came from it", + "was it light", + "how does light relate to truth", + "is truth the same as word", + ], + + # Identity alignment pressure — inputs designed to probe the threshold. + # The IdentityCheck should produce varying scores; some may flag. + "identity_pressure": [ + "logos", + "dabar", + "aletheia", + "phos", + "zoe", + "arche", + "or", + ], + + # Fatigue arc — many turns to observe ExertionMeter drain. + # versor_condition and cycle_cost should change over the session. + "fatigue_arc": [ + "beginning", + "word", + "light", + "truth", + "covenant", + "logos", + "dabar", + "zoe", + "arche", + "phos", + "aletheia", + "or", + ], + + # Versor drift — short turns watching algebraic condition across session. + "versor_drift": [ + "word beginning", + "light truth", + "covenant logos", + "dabar aletheia", + "phos zoe arche", + ], + + # Cross-lingual — mixed vocabulary across mounted packs. + "cross_lingual": [ + "logos word", + "\u03bb\u03cc\u03b3\u03bf\u03c2 truth", + "\u05d3\u05d1\u05e8 beginning", + "\u05d0\u05d5\u05e8 light", + "\u03c6\u1ff6\u03c2 covenant", + ], +} + + +# --------------------------------------------------------------------------- +# Serialisation helpers +# --------------------------------------------------------------------------- + +def _score_to_dict(score) -> dict | None: + """Convert IdentityScore (or None) to a JSON-serialisable dict.""" + if score is None: + return None + try: + return { + "value": float(getattr(score, "value", 0.0)), + "flagged": bool(getattr(score, "flagged", False)), + "alignment": float(getattr(score, "alignment", 0.0)), + "axes_evaluated": list(getattr(score, "axes_evaluated", [])), + } + except Exception: + return {"raw": str(score)} + + +def _turn_event_to_dict(event, response) -> dict: + """Merge TurnEvent + ChatResponse fields into one trace record.""" + record: dict = { + "turn": int(getattr(event, "turn", 0)), + "input_tokens": list(getattr(event, "input_tokens", [])), + "walk_surface": str(getattr(event, "walk_surface", "")), + "articulation_surface": str(getattr(event, "articulation_surface", "")), + "surface": str(getattr(response, "surface", "")), + "dialogue_role": str(getattr(event, "dialogue_role", "")), + "identity_score": _score_to_dict(getattr(event, "identity_score", None)), + "cycle_cost_total": float(getattr(event, "cycle_cost_total", 0.0)), + "vault_hits": int(getattr(event, "vault_hits", 0)), + "versor_condition": float(getattr(event, "versor_condition", 0.0)), + "flagged": bool(getattr(event, "flagged", False)), + "salience_top_k": getattr(response, "salience_top_k", None), + "candidates_used": getattr(response, "candidates_used", None), + "output_language": str(getattr(response, "output_language", "en")), + "proposition": { + "subject": str(getattr(response.proposition, "subject", "")), + "predicate": str(getattr(response.proposition, "predicate", "")), + "object": str(getattr(response.proposition, "object_", "") or ""), + "frame_id": str(getattr(response.proposition, "frame_id", "")), + "relation_norm": float(getattr(response.proposition, "relation_norm", 0.0)), + }, + } + return record + + +# --------------------------------------------------------------------------- +# Runner +# --------------------------------------------------------------------------- + +def run_scenario( + name: str, + turns: list[str], + out_dir: Path, + max_tokens: int = 32, + verbose: bool = False, +) -> Path: + """Run one scenario and write its JSONL trace. Returns the output path.""" + from chat.runtime import ChatRuntime + from core.config import DEFAULT_CONFIG, RuntimeConfig + + config = RuntimeConfig( + input_packs=DEFAULT_CONFIG.input_packs, + output_language=DEFAULT_CONFIG.output_language, + frame_pack=DEFAULT_CONFIG.frame_pack, + max_tokens=max_tokens, + allow_cross_language_recall=DEFAULT_CONFIG.allow_cross_language_recall, + allow_cross_language_generation=DEFAULT_CONFIG.allow_cross_language_generation, + vault_reproject_interval=DEFAULT_CONFIG.vault_reproject_interval, + use_salience=DEFAULT_CONFIG.use_salience, + salience_top_k=DEFAULT_CONFIG.salience_top_k, + inhibition_threshold=DEFAULT_CONFIG.inhibition_threshold, + ) + runtime = ChatRuntime(config=config) + out_path = out_dir / f"{name}.jsonl" + out_dir.mkdir(parents=True, exist_ok=True) + + if verbose: + print(f"\n{'='*60}") + print(f"scenario: {name} ({len(turns)} turns)") + print(f"output : {out_path}") + print(f"{'='*60}") + + with out_path.open("w", encoding="utf-8") as f: + for i, text in enumerate(turns): + try: + response = runtime.chat(text, max_tokens=max_tokens) + except (KeyError, ValueError) as exc: + if verbose: + print(f" turn {i:>2} ERROR: {exc}") + continue + + event = runtime.turn_log[-1] if runtime.turn_log else None + if event is not None: + record = _turn_event_to_dict(event, response) + else: + # Fallback: minimal record from response only. + record = { + "turn": i, + "input_tokens": text.split(), + "surface": response.surface, + "walk_surface": response.walk_surface, + "dialogue_role": str(response.dialogue_role), + "versor_condition": float(response.versor_condition), + "flagged": response.flagged, + } + + f.write(json.dumps(record, ensure_ascii=False) + "\n") + + if verbose: + score = record.get("identity_score") or {} + print( + f" turn {record['turn']:>2} | " + f"in: {text!r:<30} | " + f"surface: {record['surface']!r:<25} | " + f"score: {score.get('value', '-')!s:<6} | " + f"flagged: {record['flagged']} | " + f"versor: {record['versor_condition']:.2e}" + ) + + print(f"wrote {len(runtime.turn_log)} turns → {out_path}") + return out_path + + +def main() -> int: + parser = argparse.ArgumentParser( + description="Run CORE example scenarios and write JSONL trace files.", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=( + "Examples:\n" + " python scripts/run_examples.py\n" + " python scripts/run_examples.py --scenario field_probe --verbose\n" + " python scripts/run_examples.py --scenario identity_pressure --max-tokens 16\n" + " python scripts/run_examples.py --out-dir /tmp/traces" + ), + ) + parser.add_argument( + "--scenario", + choices=list(SCENARIOS.keys()), + default=None, + help="run only this scenario (default: run all)", + ) + parser.add_argument( + "--out-dir", + default="traces", + help="directory for JSONL output files (default: traces/)", + ) + parser.add_argument( + "--max-tokens", + type=int, + default=32, + help="max generated tokens per turn (default: 32)", + ) + parser.add_argument( + "--verbose", "-v", + action="store_true", + help="print per-turn summary while running", + ) + args = parser.parse_args() + + out_dir = Path(args.out_dir) + to_run = ( + {args.scenario: SCENARIOS[args.scenario]} + if args.scenario + else SCENARIOS + ) + + for name, turns in to_run.items(): + try: + run_scenario( + name, + turns, + out_dir, + max_tokens=args.max_tokens, + verbose=args.verbose, + ) + except Exception as exc: + print(f"scenario {name!r} failed: {exc.__class__.__name__}: {exc}", file=sys.stderr) + + print(f"\nall traces written to {out_dir}/") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())