refactor(cli): split cli.py god-file into cli/ package (commands/ + parser)

proteus/cli/__init__.py

@@ -0,0 +1,27 @@
+"""proteus.cli — command dispatch. Behavior identical to the former cli.py.
+
+Subcommands:
+    run             Run one session and append its trace to a JSONL file.
+    list-scenarios  List the registered scenario names.
+    replay          Print a saved trace's turns, outcome, and metrics.
+    compare         Aggregate traces by (model, difficulty) for baseline comparison.
+
+All commands are network-free except ``run`` with a real provider spec; use
+``--model fake:<name>`` for an offline smoke.
+"""
+from __future__ import annotations
+
+import proteus.game.scenarios  # noqa: F401  — side-effect: populate the scenario registry
+
+from proteus.cli.parser import build_parser
+
+
+def main(argv: list[str] | None = None) -> int:
+    """CLI entry point. Returns a process exit code."""
+    parser = build_parser()
+    args = parser.parse_args(argv)
+    return args.func(args)
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

proteus/cli/commands/__init__.py

@@ -0,0 +1 @@
+"""proteus.cli.commands — one module per subcommand handler."""

proteus/cli/commands/compare.py

@@ -0,0 +1,41 @@
+"""proteus.cli.commands.compare — the ``compare`` subcommand (aggregate traces)."""
+
+from __future__ import annotations
+
+import argparse
+import sys
+
+from proteus.game.runtime import read_traces
+
+
+def _cmd_compare(args: argparse.Namespace) -> int:
+    import json
+
+    from proteus.game.metrics.aggregate import aggregate_traces
+
+    all_traces = []
+    for path in args.trace_files:
+        try:
+            all_traces.extend(read_traces(path))
+        except FileNotFoundError:
+            print(f"trace file not found: {path}", file=sys.stderr)
+            return 2
+    if not all_traces:
+        print("no traces to compare", file=sys.stderr)
+        return 1
+
+    groups = aggregate_traces(all_traces)
+    for (model, difficulty), info in sorted(groups.items()):
+        print(f"=== model={model} difficulty={difficulty} n={info['n']} ===")
+        for key, mean in info["metrics"].items():
+            print(f"  {key}: {mean:.2f}")
+
+    if args.out:
+        serializable = {
+            f"{model}|{difficulty}": info
+            for (model, difficulty), info in groups.items()
+        }
+        with open(args.out, "w") as fh:
+            json.dump(serializable, fh, indent=2)
+        print(f"summary written to {args.out}")
+    return 0

proteus/cli/commands/list_scenarios.py

@@ -0,0 +1,13 @@
+"""proteus.cli.commands.list_scenarios — the ``list-scenarios`` subcommand."""
+
+from __future__ import annotations
+
+import argparse
+
+from proteus.game.scenarios.base import list_scenarios
+
+
+def _cmd_list_scenarios(_args: argparse.Namespace) -> int:
+    for name in list_scenarios():
+        print(name)
+    return 0

proteus/cli/commands/memory.py

@@ -0,0 +1,55 @@
+"""proteus.cli.commands.memory — the ``memory`` subcommand (generate + save checkpoint)."""
+
+from __future__ import annotations
+
+import argparse
+import sys
+
+from proteus.game.agents import VanillaAgent
+from proteus.game.engine.difficulty import Difficulty
+from proteus.game.scenarios.base import list_scenarios
+from proteus.providers import make_provider
+from proteus.game.runtime.memory import save_checkpoint
+from proteus.game.runtime.memory_gen import generate_memory
+from proteus.cli.commands.run import _resolve_persona
+
+
+def _cmd_memory(args: argparse.Namespace) -> int:
+    try:
+        provider = make_provider(args.model)
+    except ValueError as exc:
+        print(str(exc), file=sys.stderr)
+        return 2
+    if args.scenario not in list_scenarios():
+        print(
+            f"Unknown scenario {args.scenario!r}. "
+            f"Available scenarios: {', '.join(list_scenarios())}.",
+            file=sys.stderr,
+        )
+        return 2
+    persona, persona_err = _resolve_persona(args.persona)
+    if persona_err is not None:
+        print(persona_err, file=sys.stderr)
+        return 2
+    agent = VanillaAgent(provider)
+    ckpt = generate_memory(
+        args.scenario, agent,
+        difficulty=Difficulty(args.difficulty), seed=args.seed,
+        memory_turns=args.memory_turns, model_name=provider.model_name,
+        persona=persona,
+    )
+    if args.out:
+        from pathlib import Path
+
+        path = Path(args.out)
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(ckpt.model_dump_json(), encoding="utf-8")
+        written = path
+    else:
+        written = save_checkpoint(ckpt, root=args.memory_root)
+    print(
+        f"memory {ckpt.scenario} seed={ckpt.seed} {ckpt.difficulty} "
+        f"model={ckpt.model} turns={len(ckpt.memory_turns)} -> {ckpt.outcome}"
+    )
+    print(f"checkpoint written to {written}")
+    return 0

proteus/cli/commands/play.py

@@ -0,0 +1,48 @@
+"""proteus.cli.commands.play — the ``play`` subcommand (human via stdin)."""
+
+from __future__ import annotations
+
+import argparse
+import sys
+
+from proteus.game.agents import HumanAgent
+from proteus.game.engine.difficulty import Difficulty
+from proteus.game.scenarios.base import list_scenarios
+from proteus.game.runtime import SessionRunner, append_trace
+from proteus.cli.commands.run import _auto_gif
+
+
+def _cmd_play(args: argparse.Namespace) -> int:
+    if args.scenario not in list_scenarios():
+        print(
+            f"Unknown scenario {args.scenario!r}. "
+            f"Available scenarios: {', '.join(list_scenarios())}.",
+            file=sys.stderr,
+        )
+        return 2
+    agent = HumanAgent()
+    runner = SessionRunner(
+        args.scenario,
+        agent,
+        difficulty=Difficulty(args.difficulty),
+        seed=args.seed,
+        play_turns=args.play_turns,
+        use_probe=args.probe,
+    )
+    try:
+        trace = runner.run()
+    except EOFError:
+        print("stdin closed before the session finished.", file=sys.stderr)
+        return 2
+    accuracy = trace.metrics["motive_reading_accuracy"]
+    print(
+        f"{trace.scenario} seed={trace.seed} {trace.difficulty} "
+        f"model={trace.model} -> {trace.outcome} "
+        f"| motive_reading_accuracy={accuracy:.1f}% "
+        f"| reactivity_index={trace.metrics['reactivity_index']:.1f}%"
+    )
+    if args.out:
+        written = append_trace(trace, args.out)
+        print(f"trace appended to {written}")
+    _auto_gif(args.out, args.no_gif)
+    return 0

proteus/cli/commands/replay.py

@@ -0,0 +1,47 @@
+"""proteus.cli.commands.replay — the ``replay`` subcommand (print a saved trace)."""
+
+from __future__ import annotations
+
+import argparse
+import sys
+
+from proteus.game.runtime import read_traces
+
+
+def _cmd_replay(args: argparse.Namespace) -> int:
+    try:
+        traces = read_traces(args.trace_file)
+    except FileNotFoundError:
+        print(f"trace file not found: {args.trace_file}", file=sys.stderr)
+        return 2
+    if not traces:
+        print(f"no traces in {args.trace_file}", file=sys.stderr)
+        return 1
+
+    if args.visual or args.png:
+        from proteus.game.viz import reconstruct, render_terminal, write_pngs
+
+        for trace in traces:
+            steps = reconstruct(trace)
+            if args.visual:
+                render_terminal(steps, fps=args.fps)
+            if args.png:
+                paths = write_pngs(steps, args.png)
+                print(f"wrote {len(paths)} PNG frames to {args.png}")
+        return 0
+
+    for trace in traces:
+        print(
+            f"=== {trace.scenario} seed={trace.seed} {trace.difficulty} "
+            f"model={trace.model} -> {trace.outcome} ==="
+        )
+        for turn in trace.turns:
+            tag = "congruent" if turn.was_congruent else "DIVERGED"
+            diag = " [diagnostic]" if turn.is_diagnostic else ""
+            print(
+                f"  turn {turn.turn_idx}: action={turn.action} "
+                f"motive={turn.motive_action} habit={turn.habit_action} "
+                f"{tag}{diag} reward={turn.reward}"
+            )
+        print(f"  metrics: {trace.metrics}")
+    return 0

proteus/cli/commands/run.py

@@ -0,0 +1,128 @@
+"""proteus.cli.commands.run — the ``run`` subcommand and shared helpers."""
+
+from __future__ import annotations
+
+import argparse
+import sys
+
+from proteus.game.agents import VanillaAgent
+from proteus.game.engine.difficulty import Difficulty
+from proteus.game.scenarios.base import list_scenarios
+from proteus.providers import make_provider
+from proteus.game.runtime import SessionRunner, append_trace, read_traces
+from proteus.game.runtime.memory import (
+    latest_for_model,
+    load_checkpoint,
+    save_checkpoint,
+)
+from proteus.game.runtime.memory_gen import generate_memory
+from proteus.game.metrics.persona import available_personas, get_persona
+
+
+def _cmd_run(args: argparse.Namespace) -> int:
+    try:
+        provider = make_provider(args.model)
+    except ValueError as exc:
+        print(str(exc), file=sys.stderr)
+        return 2
+    if args.scenario not in list_scenarios():
+        print(
+            f"Unknown scenario {args.scenario!r}. "
+            f"Available scenarios: {', '.join(list_scenarios())}.",
+            file=sys.stderr,
+        )
+        return 2
+    agent = VanillaAgent(provider)
+    memory, memory_ref, mem_err = _resolve_memory(
+        args.memory, provider=provider, agent=agent, scenario=args.scenario,
+        difficulty=Difficulty(args.difficulty), seed=args.seed,
+        memory_turns=args.memory_turns, memory_root=args.memory_root,
+    )
+    if mem_err is not None:
+        print(mem_err, file=sys.stderr)
+        return 2
+    persona, persona_err = _resolve_persona(args.persona)
+    if persona_err is not None:
+        print(persona_err, file=sys.stderr)
+        return 2
+    runner = SessionRunner(
+        args.scenario,
+        agent,
+        difficulty=Difficulty(args.difficulty),
+        seed=args.seed,
+        play_turns=args.play_turns,
+        use_probe=not args.no_probe,
+        memory=memory,
+        memory_ref=memory_ref,
+        persona=persona,
+    )
+    trace = runner.run()
+    written = append_trace(trace, args.out)
+    accuracy = trace.metrics["motive_reading_accuracy"]
+    print(
+        f"{trace.scenario} seed={trace.seed} {trace.difficulty} "
+        f"model={trace.model} -> {trace.outcome} "
+        f"| motive_reading_accuracy={accuracy:.1f}% "
+        f"| reactivity_index={trace.metrics['reactivity_index']:.1f}%"
+    )
+    print(f"trace appended to {written}")
+    _auto_gif(args.out, args.no_gif)
+    return 0
+
+
+def _resolve_persona(persona_id: str | None):
+    """Resolve --persona ID to (PersonaWeights | None, error | None)."""
+    if persona_id is None:
+        return None, None
+    try:
+        return get_persona(persona_id), None
+    except KeyError:
+        return None, (
+            f"Unknown persona {persona_id!r}. "
+            f"Available personas: {', '.join(available_personas())}."
+        )
+
+
+def _auto_gif(out_path: str | None, no_gif: bool) -> None:
+    """Render <out_stem>.gif next to the trace file unless suppressed."""
+    if not out_path or no_gif:
+        return
+    from pathlib import Path
+
+    from proteus.game.viz import reconstruct, write_gif
+
+    gif_path = Path(out_path).with_suffix(".gif")
+    for trace in read_traces(out_path):
+        write_gif(reconstruct(trace), gif_path)
+        print(f"gif written to {gif_path}")
+        break  # one game per run/play -> one GIF
+
+
+def _resolve_memory(
+    mode: str, *, provider, agent, scenario: str, difficulty: Difficulty,
+    seed: int | None, memory_turns: int, memory_root: str,
+):
+    """Resolve --memory MODE to (checkpoint | None, memory_ref | None, error | None)."""
+    if mode == "none":
+        return None, None, None
+    if mode == "generate":
+        ckpt = generate_memory(
+            scenario, agent, difficulty=difficulty, seed=seed,
+            memory_turns=memory_turns, model_name=provider.model_name,
+        )
+        path = save_checkpoint(ckpt, root=memory_root)
+        return ckpt, str(path), None
+    if mode == "latest":
+        ckpt = latest_for_model(provider.model_name, root=memory_root)
+        if ckpt is None:
+            return None, None, (
+                f"no memory checkpoint for model {provider.model_name!r} "
+                f"under {memory_root!r}"
+            )
+        return ckpt, f"{ckpt.model}@{ckpt.created_at}", None
+    # otherwise treat MODE as a checkpoint path
+    try:
+        ckpt = load_checkpoint(mode)
+    except FileNotFoundError:
+        return None, None, f"memory checkpoint not found: {mode}"
+    return ckpt, mode, None

proteus/{cli.py → cli/parser.py}

@@ -1,297 +1,19 @@
-"""Command-line interface for the PROTEUS arena.
-
-Subcommands:
-    run             Run one session and append its trace to a JSONL file.
-    list-scenarios  List the registered scenario names.
-    replay          Print a saved trace's turns, outcome, and metrics.
-    compare         Aggregate traces by (model, difficulty) for baseline comparison.
-
-All commands are network-free except ``run`` with a real provider spec; use
-``--model fake:<name>`` for an offline smoke.
-"""
+"""proteus.cli.parser — top-level argument parser construction."""
 
 from __future__ import annotations
 
 import argparse
-import sys
 
-import proteus.game.scenarios  # noqa: F401  — side-effect: populate the scenario registry
-from proteus.game.agents import HumanAgent, VanillaAgent
 from proteus.game.engine.difficulty import Difficulty
-from proteus.game.scenarios.base import list_scenarios
-from proteus.providers import available_providers, make_provider
-from proteus.game.runtime import SessionRunner, append_trace, read_traces
-from proteus.game.runtime.memory import (
-    latest_for_model,
-    load_checkpoint,
-    save_checkpoint,
-)
-from proteus.game.runtime.memory_gen import generate_memory
-from proteus.game.metrics.persona import available_personas, get_persona
-
-
-def _cmd_run(args: argparse.Namespace) -> int:
-    try:
-        provider = make_provider(args.model)
-    except ValueError as exc:
-        print(str(exc), file=sys.stderr)
-        return 2
-    if args.scenario not in list_scenarios():
-        print(
-            f"Unknown scenario {args.scenario!r}. "
-            f"Available scenarios: {', '.join(list_scenarios())}.",
-            file=sys.stderr,
-        )
-        return 2
-    agent = VanillaAgent(provider)
-    memory, memory_ref, mem_err = _resolve_memory(
-        args.memory, provider=provider, agent=agent, scenario=args.scenario,
-        difficulty=Difficulty(args.difficulty), seed=args.seed,
-        memory_turns=args.memory_turns, memory_root=args.memory_root,
-    )
-    if mem_err is not None:
-        print(mem_err, file=sys.stderr)
-        return 2
-    persona, persona_err = _resolve_persona(args.persona)
-    if persona_err is not None:
-        print(persona_err, file=sys.stderr)
-        return 2
-    runner = SessionRunner(
-        args.scenario,
-        agent,
-        difficulty=Difficulty(args.difficulty),
-        seed=args.seed,
-        play_turns=args.play_turns,
-        use_probe=not args.no_probe,
-        memory=memory,
-        memory_ref=memory_ref,
-        persona=persona,
-    )
-    trace = runner.run()
-    written = append_trace(trace, args.out)
-    accuracy = trace.metrics["motive_reading_accuracy"]
-    print(
-        f"{trace.scenario} seed={trace.seed} {trace.difficulty} "
-        f"model={trace.model} -> {trace.outcome} "
-        f"| motive_reading_accuracy={accuracy:.1f}% "
-        f"| reactivity_index={trace.metrics['reactivity_index']:.1f}%"
-    )
-    print(f"trace appended to {written}")
-    _auto_gif(args.out, args.no_gif)
-    return 0
-
-
-def _resolve_persona(persona_id: str | None):
-    """Resolve --persona ID to (PersonaWeights | None, error | None)."""
-    if persona_id is None:
-        return None, None
-    try:
-        return get_persona(persona_id), None
-    except KeyError:
-        return None, (
-            f"Unknown persona {persona_id!r}. "
-            f"Available personas: {', '.join(available_personas())}."
-        )
-
-
-def _auto_gif(out_path: str | None, no_gif: bool) -> None:
-    """Render <out_stem>.gif next to the trace file unless suppressed."""
-    if not out_path or no_gif:
-        return
-    from pathlib import Path
-
-    from proteus.game.viz import reconstruct, write_gif
-
-    gif_path = Path(out_path).with_suffix(".gif")
-    for trace in read_traces(out_path):
-        write_gif(reconstruct(trace), gif_path)
-        print(f"gif written to {gif_path}")
-        break  # one game per run/play -> one GIF
-
-
-def _resolve_memory(
-    mode: str, *, provider, agent, scenario: str, difficulty: Difficulty,
-    seed: int | None, memory_turns: int, memory_root: str,
-):
-    """Resolve --memory MODE to (checkpoint | None, memory_ref | None, error | None)."""
-    if mode == "none":
-        return None, None, None
-    if mode == "generate":
-        ckpt = generate_memory(
-            scenario, agent, difficulty=difficulty, seed=seed,
-            memory_turns=memory_turns, model_name=provider.model_name,
-        )
-        path = save_checkpoint(ckpt, root=memory_root)
-        return ckpt, str(path), None
-    if mode == "latest":
-        ckpt = latest_for_model(provider.model_name, root=memory_root)
-        if ckpt is None:
-            return None, None, (
-                f"no memory checkpoint for model {provider.model_name!r} "
-                f"under {memory_root!r}"
-            )
-        return ckpt, f"{ckpt.model}@{ckpt.created_at}", None
-    # otherwise treat MODE as a checkpoint path
-    try:
-        ckpt = load_checkpoint(mode)
-    except FileNotFoundError:
-        return None, None, f"memory checkpoint not found: {mode}"
-    return ckpt, mode, None
-
-
-def _cmd_memory(args: argparse.Namespace) -> int:
-    try:
-        provider = make_provider(args.model)
-    except ValueError as exc:
-        print(str(exc), file=sys.stderr)
-        return 2
-    if args.scenario not in list_scenarios():
-        print(
-            f"Unknown scenario {args.scenario!r}. "
-            f"Available scenarios: {', '.join(list_scenarios())}.",
-            file=sys.stderr,
-        )
-        return 2
-    persona, persona_err = _resolve_persona(args.persona)
-    if persona_err is not None:
-        print(persona_err, file=sys.stderr)
-        return 2
-    agent = VanillaAgent(provider)
-    ckpt = generate_memory(
-        args.scenario, agent,
-        difficulty=Difficulty(args.difficulty), seed=args.seed,
-        memory_turns=args.memory_turns, model_name=provider.model_name,
-        persona=persona,
-    )
-    if args.out:
-        from pathlib import Path
-
-        path = Path(args.out)
-        path.parent.mkdir(parents=True, exist_ok=True)
-        path.write_text(ckpt.model_dump_json(), encoding="utf-8")
-        written = path
-    else:
-        written = save_checkpoint(ckpt, root=args.memory_root)
-    print(
-        f"memory {ckpt.scenario} seed={ckpt.seed} {ckpt.difficulty} "
-        f"model={ckpt.model} turns={len(ckpt.memory_turns)} -> {ckpt.outcome}"
-    )
-    print(f"checkpoint written to {written}")
-    return 0
-
-
-def _cmd_play(args: argparse.Namespace) -> int:
-    if args.scenario not in list_scenarios():
-        print(
-            f"Unknown scenario {args.scenario!r}. "
-            f"Available scenarios: {', '.join(list_scenarios())}.",
-            file=sys.stderr,
-        )
-        return 2
-    agent = HumanAgent()
-    runner = SessionRunner(
-        args.scenario,
-        agent,
-        difficulty=Difficulty(args.difficulty),
-        seed=args.seed,
-        play_turns=args.play_turns,
-        use_probe=args.probe,
-    )
-    try:
-        trace = runner.run()
-    except EOFError:
-        print("stdin closed before the session finished.", file=sys.stderr)
-        return 2
-    accuracy = trace.metrics["motive_reading_accuracy"]
-    print(
-        f"{trace.scenario} seed={trace.seed} {trace.difficulty} "
-        f"model={trace.model} -> {trace.outcome} "
-        f"| motive_reading_accuracy={accuracy:.1f}% "
-        f"| reactivity_index={trace.metrics['reactivity_index']:.1f}%"
-    )
-    if args.out:
-        written = append_trace(trace, args.out)
-        print(f"trace appended to {written}")
-    _auto_gif(args.out, args.no_gif)
-    return 0
-
-
-def _cmd_list_scenarios(_args: argparse.Namespace) -> int:
-    for name in list_scenarios():
-        print(name)
-    return 0
-
-
-def _cmd_replay(args: argparse.Namespace) -> int:
-    try:
-        traces = read_traces(args.trace_file)
-    except FileNotFoundError:
-        print(f"trace file not found: {args.trace_file}", file=sys.stderr)
-        return 2
-    if not traces:
-        print(f"no traces in {args.trace_file}", file=sys.stderr)
-        return 1
-
-    if args.visual or args.png:
-        from proteus.game.viz import reconstruct, render_terminal, write_pngs
+from proteus.providers import available_providers
+from proteus.game.metrics.persona import available_personas
 
-        for trace in traces:
-            steps = reconstruct(trace)
-            if args.visual:
-                render_terminal(steps, fps=args.fps)
-            if args.png:
-                paths = write_pngs(steps, args.png)
-                print(f"wrote {len(paths)} PNG frames to {args.png}")
-        return 0
-
-    for trace in traces:
-        print(
-            f"=== {trace.scenario} seed={trace.seed} {trace.difficulty} "
-            f"model={trace.model} -> {trace.outcome} ==="
-        )
-        for turn in trace.turns:
-            tag = "congruent" if turn.was_congruent else "DIVERGED"
-            diag = " [diagnostic]" if turn.is_diagnostic else ""
-            print(
-                f"  turn {turn.turn_idx}: action={turn.action} "
-                f"motive={turn.motive_action} habit={turn.habit_action} "
-                f"{tag}{diag} reward={turn.reward}"
-            )
-        print(f"  metrics: {trace.metrics}")
-    return 0
-
-
-def _cmd_compare(args: argparse.Namespace) -> int:
-    import json
-
-    from proteus.game.metrics.aggregate import aggregate_traces
-
-    all_traces = []
-    for path in args.trace_files:
-        try:
-            all_traces.extend(read_traces(path))
-        except FileNotFoundError:
-            print(f"trace file not found: {path}", file=sys.stderr)
-            return 2
-    if not all_traces:
-        print("no traces to compare", file=sys.stderr)
-        return 1
-
-    groups = aggregate_traces(all_traces)
-    for (model, difficulty), info in sorted(groups.items()):
-        print(f"=== model={model} difficulty={difficulty} n={info['n']} ===")
-        for key, mean in info["metrics"].items():
-            print(f"  {key}: {mean:.2f}")
-
-    if args.out:
-        serializable = {
-            f"{model}|{difficulty}": info
-            for (model, difficulty), info in groups.items()
-        }
-        with open(args.out, "w") as fh:
-            json.dump(serializable, fh, indent=2)
-        print(f"summary written to {args.out}")
-    return 0
+from proteus.cli.commands.run import _cmd_run
+from proteus.cli.commands.play import _cmd_play
+from proteus.cli.commands.memory import _cmd_memory
+from proteus.cli.commands.list_scenarios import _cmd_list_scenarios
+from proteus.cli.commands.replay import _cmd_replay
+from proteus.cli.commands.compare import _cmd_compare
 
 
 def build_parser() -> argparse.ArgumentParser:
@@ -415,10 +137,3 @@ def build_parser() -> argparse.ArgumentParser:
     compare.set_defaults(func=_cmd_compare)
 
     return parser
-
-
-def main(argv: list[str] | None = None) -> int:
-    """CLI entry point. Returns a process exit code."""
-    parser = build_parser()
-    args = parser.parse_args(argv)
-    return args.func(args)