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

feat(cli): ADR-0024 chain test-suite aliases + core demo subcommand

Browse source 
Two layers of CLI surface so reviewers / investors / industry
observers can run the ADR-0024 chain evidence end-to-end without
typing test file paths or hunting for runner scripts.

Layer 1 — test-suite aliases:
  core test --suite refusal     (Phase 2 typed refusals)
  core test --suite margin      (Phase 3 / ADR-0026 ranked-with-margin)
  core test --suite rotor       (Phase 4 / ADR-0025 rotor admissibility)
  core test --suite inner-loop  (ADR-0024 inner-loop, all 4 sub-tests)
  core test --suite phase5      (stratified mechanism-isolation)
  core test --suite phase6      (3-condition comparative demo)
  core test --suite adr-0024    (full chain, 98 tests, ~2 min)

Layer 2 — `core demo` subcommand:
  core demo phase5              stratified pass/refuse table + per-family
                                breakdown (5 families, both modes)
  core demo phase6              three head-to-head verdicts vs baseline
                                (replay determinism / traced rejection /
                                coherent refusal)
  core demo all                 both phases + combined summary
  core demo list-results        index every JSON report in the central
                                results directory with headline metrics

All demo runs:
  - Write fresh JSON to evals/forward_semantic_control/results/
  - Refresh the results/index.json manifest so reviewers see every
    available report in one place
  - Accept --json for machine-readable output

Central results directory: evals/forward_semantic_control/results/
  phase2_inner_loop_report.json
  phase3_v2_report.json
  phase4_characterization_*.json
  phase5_report.json
  phase5_benign_inner_loop_report.json
  phase6_demo_report.json
  index.json (auto-generated manifest)

Files:
  core/cli.py                   — +9 suite aliases, +cmd_demo (3 targets +
                                  list-results), +index manifest writer
  tests/test_cli_demo.py        — 14 contract tests pinning Layer 1 + 2
  evals/forward_semantic_control/results/index.json — auto-generated

Tests: 1099 passed, 2 skipped (+14 from Phase 6 baseline).
This commit is contained in:
Shay 2026-05-17 16:21:37 -07:00
parent a0765066b4
commit 36aad75202
3 changed files with 452 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

236
core/cli.py
View file

@ -94,6 +94,41 @@ _TEST_SUITES: dict[str, tuple[str, ...]] = {
"proof": (
"tests/test_proof_properties.py",
),
# ADR-0024 chain suites (Phases 2-6). Each phase has its own
# contract tests so investors / reviewers can run them
# independently; ``adr-0024`` runs the full chain end-to-end.
"refusal": (
"tests/test_refusal_contract.py",
),
"margin": (
"tests/test_margin_admissibility.py",
),
"rotor": (
"tests/test_rotor_admissibility.py",
),
"inner-loop": (
"tests/test_inner_loop_admissibility.py",
"tests/test_inner_loop_phase2.py",
"tests/test_inner_loop_phase3.py",
"tests/test_inner_loop_phase4.py",
),
"phase5": (
"tests/test_phase5_corpus.py",
),
"phase6": (
"tests/test_phase6_demo.py",
),
"adr-0024": (
"tests/test_refusal_contract.py",
"tests/test_margin_admissibility.py",
"tests/test_rotor_admissibility.py",
"tests/test_inner_loop_admissibility.py",
"tests/test_inner_loop_phase2.py",
"tests/test_inner_loop_phase3.py",
"tests/test_inner_loop_phase4.py",
"tests/test_phase5_corpus.py",
"tests/test_phase6_demo.py",
),
"full": ("tests/",),
}
@ -632,6 +667,185 @@ def cmd_pulse(args: argparse.Namespace) -> int:
return 0
_DEMO_RESULTS_DIR = Path("evals/forward_semantic_control/results")
_DEMO_CORPUS_DIR = Path("evals/forward_semantic_control/public")
def _format_phase5_table(metrics: dict[str, Any], per_family: dict[str, Any]) -> str:
lines = [
"",
"Phase 5 — Stratified Mechanism-Isolation (ADR-0024 / ADR-0026)",
"=" * 68,
f" cases: {metrics.get('case_count', 0)}",
f" margin (δ): {metrics.get('margin', 0)}",
f" pass_rate (threshold): {metrics.get('pass_rate_threshold', 0):.2%}",
f" pass_rate (margin): {metrics.get('pass_rate_margin', 0):.2%}",
f" mechanism_isolated (thr): {metrics.get('mechanism_isolated_threshold', False)}",
f" mechanism_isolated (mgn): {metrics.get('mechanism_isolated_margin', False)}",
"",
f" {'family':38s} {'cases':>6s} {'pass(thr)':>11s} {'pass(mgn)':>11s} {'refuse(mgn)':>13s}",
" " + "-" * 84,
]
for fam, b in per_family.items():
lines.append(
f" {fam:38s} {b.get('case_count', 0):>6d} "
f"{b.get('pass_rate_threshold', 0):>10.2%} "
f"{b.get('pass_rate_margin', 0):>10.2%} "
f"{b.get('refusal_rate_margin', 0):>12.2%}"
)
return "\n".join(lines) + "\n"
def _format_phase6_table(metrics: dict[str, Any]) -> str:
def pf(b: bool) -> str:
return "PASS" if b else "FAIL"
lines = [
"",
"Phase 6 — Comparative Demo: CORE vs In-System Baseline (ADR-0023 ablation)",
"=" * 76,
f" total cases: {metrics.get('case_count', 0)}",
f" replay reruns: {metrics.get('replay_reruns', 0)}",
"",
" C1 Replay determinism",
f" baseline stable: {metrics.get('c1_replay_stable_baseline', 0)} / {metrics.get('c1_eligible', 0)}",
f" CORE stable: {metrics.get('c1_replay_stable_core', 0)} / {metrics.get('c1_eligible', 0)}",
f" verdict: {pf(metrics.get('c1_pass', False))}",
"",
" C2 Traced rejection",
f" baseline emits forbidden: {metrics.get('c2_baseline_emits_forbidden', 0)} / {metrics.get('c2_case_count', 0)}",
f" baseline admits forbidden: {metrics.get('c2_baseline_admits_forbidden', 0)} / {metrics.get('c2_case_count', 0)}",
f" CORE corrects-or-refuses: {metrics.get('c2_core_corrects_or_refuses', 0)} / {metrics.get('c2_case_count', 0)}",
f" CORE rejection in trace: {metrics.get('c2_core_rejection_traced', 0)} / {metrics.get('c2_case_count', 0)}",
f" verdict: {pf(metrics.get('c2_pass', False))}",
"",
" C3 Coherent refusal",
f" baseline typed refusals: {metrics.get('c3_baseline_refused_typed', 0)} / {metrics.get('c3_case_count', 0)}",
f" baseline emits inadmiss.: {metrics.get('c3_baseline_emitted_inadmissible', 0)} / {metrics.get('c3_case_count', 0)}",
f" CORE typed refusals: {metrics.get('c3_core_refused_typed', 0)} / {metrics.get('c3_case_count', 0)}",
f" verdict: {pf(metrics.get('c3_pass', False))}",
"",
f" ALL THREE CONDITIONS: {pf(metrics.get('all_three_conditions_pass', False))}",
]
return "\n".join(lines) + "\n"
def _write_results_index() -> Path:
"""Write/refresh the results index manifest.
Lists every ``*_report.json`` in the results directory with its
headline metric (or a short summary). Reviewers can read this to
discover all available evidence in one place.
"""
results_dir = _DEMO_RESULTS_DIR
results_dir.mkdir(parents=True, exist_ok=True)
entries: list[dict[str, Any]] = []
for p in sorted(results_dir.glob("*.json")):
if p.name == "index.json":
continue
try:
data = json.loads(p.read_text())
except (OSError, json.JSONDecodeError):
continue
metrics = data.get("metrics", {}) if isinstance(data, dict) else {}
entries.append({
"file": p.name,
"size_bytes": p.stat().st_size,
"headline": {
k: v for k, v in metrics.items()
if k in (
"case_count", "pass_rate", "pass_rate_threshold",
"pass_rate_margin", "mechanism_isolated",
"mechanism_isolated_threshold", "mechanism_isolated_margin",
"all_three_conditions_pass", "c1_pass", "c2_pass", "c3_pass",
"best_threshold", "best_separation_quality",
)
},
})
index_path = results_dir / "index.json"
index_path.write_text(json.dumps({
"results_dir": str(results_dir),
"reports": entries,
}, indent=2))
return index_path
def _run_demo_phase5(emit_json: bool) -> dict[str, Any]:
from evals.forward_semantic_control.phase5_runner import run_lane
cases_path = _DEMO_CORPUS_DIR / "v2_phase5" / "cases.jsonl"
cases = [json.loads(l) for l in cases_path.read_text().splitlines() if l.strip()]
report = run_lane(cases)
out = _DEMO_RESULTS_DIR / "phase5_report.json"
out.parent.mkdir(parents=True, exist_ok=True)
out.write_text(json.dumps({
"metrics": report.metrics,
"per_family": report.per_family,
"case_details": report.case_details,
}, indent=2))
if emit_json:
print(json.dumps({"metrics": report.metrics, "per_family": report.per_family}, indent=2))
else:
print(_format_phase5_table(report.metrics, report.per_family))
print(f" full report: {out}")
return report.metrics
def _run_demo_phase6(emit_json: bool) -> dict[str, Any]:
from evals.forward_semantic_control.phase6_demo import run_lane
cases_path = _DEMO_CORPUS_DIR / "v2_phase6_demo" / "cases.jsonl"
cases = [json.loads(l) for l in cases_path.read_text().splitlines() if l.strip()]
report = run_lane(cases)
out = _DEMO_RESULTS_DIR / "phase6_demo_report.json"
out.parent.mkdir(parents=True, exist_ok=True)
out.write_text(json.dumps({
"metrics": report.metrics,
"case_details": report.case_details,
}, indent=2))
if emit_json:
print(json.dumps({"metrics": report.metrics}, indent=2))
else:
print(_format_phase6_table(report.metrics))
print(f" full report: {out}")
return report.metrics
def cmd_demo(args: argparse.Namespace) -> int:
"""Run the ADR-0024 chain comparative demos for investors / reviewers."""
target = args.target
if target == "list-results":
index_path = _write_results_index()
data = json.loads(index_path.read_text())
if args.json:
print(json.dumps(data, indent=2))
else:
print(f"\nresults directory: {data['results_dir']}\n")
for entry in data["reports"]:
print(f" {entry['file']:55s} {entry['size_bytes']:>9d} bytes")
for k, v in entry["headline"].items():
print(f" {k}: {v}")
return 0
if target == "phase5":
_run_demo_phase5(args.json)
elif target == "phase6":
_run_demo_phase6(args.json)
elif target == "all":
p5 = _run_demo_phase5(args.json)
p6 = _run_demo_phase6(args.json)
if not args.json:
print("\n" + "=" * 76)
print("Combined demo summary")
print("=" * 76)
print(f" Phase 5 pass_rate (margin): {p5.get('pass_rate_margin', 0):.2%}")
print(f" Phase 5 mechanism_isolated: {p5.get('mechanism_isolated_margin', False)}")
print(f" Phase 6 all three conditions: {p6.get('all_three_conditions_pass', False)}")
print("")
else:
_die(f"unknown demo target: {target}")
_write_results_index()
return 0
def cmd_bench(args: argparse.Namespace) -> int:
"""Run benchmark harness."""
# "cost" suite has its own runtime contract — wall/CPU-seconds and
@ -807,6 +1021,28 @@ def build_parser() -> argparse.ArgumentParser:
bench.add_argument("--report", metavar="PATH", help="write JSON report to file")
bench.set_defaults(func=cmd_bench)
demo = subparsers.add_parser(
"demo",
help="run ADR-0024 chain comparative demos (phase5 / phase6 / all)",
description=(
"Run the comparative demo evidence for the ADR-0024 chain. "
"Designed for showcasing CORE's deterministic-cognition mechanisms "
"to reviewers / investors / industry observers."
),
)
demo.add_argument(
"target",
choices=["phase5", "phase6", "all", "list-results"],
help=(
"phase5: stratified 5-family mechanism-isolation. "
"phase6: 3-condition head-to-head vs in-system baseline. "
"all: run both and print a combined summary. "
"list-results: index every JSON report in the results directory."
),
)
demo.add_argument("--json", action="store_true", help="emit machine-readable JSON")
demo.set_defaults(func=cmd_demo)
eval_cmd = subparsers.add_parser("eval", help="run eval lanes")
eval_cmd.add_argument("lane", nargs="?", help="eval lane name (e.g. cognition)")
eval_cmd.add_argument("--list", dest="list_lanes", action="store_true", help="list available eval lanes")

82
evals/forward_semantic_control/results/index.json Normal file
View file

@ -0,0 +1,82 @@
{
"results_dir": "evals/forward_semantic_control/results",
"reports": [
{
"file": "phase2_inner_loop_report.json",
"size_bytes": 16773,
"headline": {
"case_count": 9
}
},
{
"file": "phase3_v2_report.json",
"size_bytes": 4302,
"headline": {
"case_count": 5,
"pass_rate": 1.0,
"mechanism_isolated": true
}
},
{
"file": "phase4_characterization_combined.json",
"size_bytes": 3753,
"headline": {
"best_threshold": 1.0,
"best_separation_quality": 0.6,
"case_count": 14
}
},
{
"file": "phase4_characterization_v1_plus_dev.json",
"size_bytes": 2587,
"headline": {
"best_threshold": -1.0,
"best_separation_quality": 0.0,
"case_count": 9
}
},
{
"file": "phase4_characterization_v2.json",
"size_bytes": 3194,
"headline": {
"best_threshold": 1.0,
"best_separation_quality": 0.6,
"case_count": 5
}
},
{
"file": "phase4_summary.json",
"size_bytes": 1387,
"headline": {}
},
{
"file": "phase5_benign_inner_loop_report.json",
"size_bytes": 18673,
"headline": {
"case_count": 10
}
},
{
"file": "phase5_report.json",
"size_bytes": 36903,
"headline": {
"case_count": 20,
"pass_rate_threshold": 1.0,
"pass_rate_margin": 1.0,
"mechanism_isolated_threshold": true,
"mechanism_isolated_margin": true
}
},
{
"file": "phase6_demo_report.json",
"size_bytes": 15874,
"headline": {
"case_count": 8,
"c1_pass": true,
"c2_pass": true,
"c3_pass": true,
"all_three_conditions_pass": true
}
}
]
}

134
tests/test_cli_demo.py Normal file
View file

@ -0,0 +1,134 @@
"""CLI tests for the ADR-0024 chain demo subcommand."""
from __future__ import annotations
import json
from pathlib import Path
import pytest
from core import cli
class TestADR0024SuiteAliases:
"""Layer 1: pin the new suite aliases so they don't drift."""
@pytest.mark.parametrize(
"suite,must_contain",
[
("refusal", "tests/test_refusal_contract.py"),
("margin", "tests/test_margin_admissibility.py"),
("rotor", "tests/test_rotor_admissibility.py"),
("inner-loop", "tests/test_inner_loop_admissibility.py"),
("phase5", "tests/test_phase5_corpus.py"),
("phase6", "tests/test_phase6_demo.py"),
],
)
def test_suite_alias_resolves(
self, monkeypatch, suite: str, must_contain: str
) -> None:
calls: list[tuple[str, ...]] = []
def fake_run(*args: str, check: bool = False, cwd=None) -> int:
calls.append(args)
return 0
monkeypatch.setattr(cli, "_run", fake_run)
rc = cli.main(["test", "--suite", suite, "-q"])
assert rc == 0
assert calls, f"no _run invocation for suite {suite!r}"
assert must_contain in calls[0]
def test_adr_0024_alias_runs_full_chain(self, monkeypatch) -> None:
calls: list[tuple[str, ...]] = []
monkeypatch.setattr(
cli, "_run", lambda *a, **kw: (calls.append(a) or 0)
)
rc = cli.main(["test", "--suite", "adr-0024", "-q"])
assert rc == 0
command = calls[0]
# All six Phase 2-6 contract files must be present.
for path in (
"tests/test_refusal_contract.py",
"tests/test_margin_admissibility.py",
"tests/test_rotor_admissibility.py",
"tests/test_phase5_corpus.py",
"tests/test_phase6_demo.py",
):
assert path in command, f"missing {path} in adr-0024 expansion"
def test_list_suites_includes_new_aliases(self, capsys) -> None:
rc = cli.main(["test", "--list-suites"])
captured = capsys.readouterr()
assert rc == 0
out = captured.out.splitlines()
for alias in (
"adr-0024",
"refusal",
"margin",
"rotor",
"inner-loop",
"phase5",
"phase6",
):
assert alias in out, f"alias {alias!r} missing from --list-suites"
class TestDemoSubcommand:
"""Layer 2: pin the `core demo` subcommand surface."""
def test_demo_help_lists_targets(self, capsys) -> None:
with pytest.raises(SystemExit) as exc:
cli.main(["demo", "--help"])
assert exc.value.code == 0
captured = capsys.readouterr()
for target in ("phase5", "phase6", "all", "list-results"):
assert target in captured.out
def test_demo_phase6_runs_and_writes_report(self, capsys) -> None:
rc = cli.main(["demo", "phase6"])
assert rc == 0
captured = capsys.readouterr()
# Headline text on stdout.
assert "ALL THREE CONDITIONS" in captured.out
assert "PASS" in captured.out
# Report file present and well-formed.
report = Path("evals/forward_semantic_control/results/phase6_demo_report.json")
assert report.exists()
data = json.loads(report.read_text())
assert data["metrics"]["all_three_conditions_pass"] is True
def test_demo_phase6_json_emits_machine_readable(self, capsys) -> None:
rc = cli.main(["demo", "phase6", "--json"])
assert rc == 0
captured = capsys.readouterr()
# First non-blank chunk of stdout must be a parseable JSON
# object containing the headline keys.
payload = json.loads(captured.out.split("\n\n")[0])
assert "metrics" in payload
assert "all_three_conditions_pass" in payload["metrics"]
def test_demo_list_results_indexes_reports(self, capsys) -> None:
rc = cli.main(["demo", "list-results"])
assert rc == 0
captured = capsys.readouterr()
assert "results directory" in captured.out
assert "phase6_demo_report.json" in captured.out
def test_demo_list_results_json_well_formed(self, capsys) -> None:
rc = cli.main(["demo", "list-results", "--json"])
assert rc == 0
captured = capsys.readouterr()
data = json.loads(captured.out)
assert "results_dir" in data
assert isinstance(data["reports"], list)
names = [e["file"] for e in data["reports"]]
assert "phase6_demo_report.json" in names
def test_demo_index_file_refreshed_after_run(self) -> None:
cli.main(["demo", "phase6"])
index_path = Path("evals/forward_semantic_control/results/index.json")
assert index_path.exists()
data = json.loads(index_path.read_text())
names = [e["file"] for e in data["reports"]]
assert "phase6_demo_report.json" in names