core-labs/core
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

scripts: add run_examples.py + review_trace.py; cli: surface TurnEvent in trace/session

Browse source 
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/<scenario>.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/<scenario>.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
This commit is contained in:
Shay 2026-05-14 13:54:25 -07:00
parent 97bfd3b1d8
commit f91063f771
3 changed files with 565 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

0
scripts/__init__.py Normal file
View file

276
scripts/review_trace.py Normal file
View file

@ -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())

289
scripts/run_examples.py Normal file
View file

@ -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/<scenario>.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())