fix(eval): address Copilot review — checkout-agnostic docs + skill validation

- Docstrings/help said the treatment skill 'composes into the system
  message'. This harness runs on both checkouts (system on main, a context
  turn on the placement-refactor branch), so the wording now describes the
  natural set_skill composition path without asserting a placement.
- Validate each skill-bearing case's 'skill' shape up front (driver +
  CLI) so a malformed dataset fails with a clear error, not a mid-run
  KeyError. Pinned by test_rejects_malformed_skill.

eval_skill_adherence.json

@@ -0,0 +1,56 @@
+{
+  "defaults": {
+    "n_runs": 3
+  },
+  "cases": [
+    {
+      "id": "search-first",
+      "skill": {
+        "name": "search-first",
+        "content": "# Search First\n\nBefore answering ANY question about where something lives in the codebase, you MUST call the `search` tool first. Never answer from memory."
+      },
+      "user_prompt": "Where is JWT token validation implemented in this project?",
+      "expected_actions": [{ "tool": "search" }],
+      "match_mode": "ordered_subset",
+      "max_turns": 4
+    },
+    {
+      "id": "test-after-edit",
+      "skill": {
+        "name": "test-after-edit",
+        "content": "# Test After Edit\n\nAfter editing or writing ANY file, you MUST run the test suite with `python -m pytest` via bash before you finish. Do not report done until tests have run."
+      },
+      "user_prompt": "Add a function `clamp(x, lo, hi)` that clamps x to [lo, hi] in utils.py.",
+      "setup": {
+        "files": {
+          "utils.py": ""
+        }
+      },
+      "expected_actions": [
+        { "tool": "write_file" },
+        { "tool": "bash", "args_pattern": { "command": "pytest" } }
+      ],
+      "match_mode": "ordered_subset",
+      "max_turns": 8
+    },
+    {
+      "id": "changelog-update",
+      "skill": {
+        "name": "changelog-update",
+        "content": "# Changelog Discipline\n\nWhenever you modify a file, you MUST also append a one-line entry to CHANGELOG.md describing the change in the same task."
+      },
+      "user_prompt": "Fix the off-by-one so pager.py shows the last page. Edit pager.py.",
+      "setup": {
+        "files": {
+          "pager.py": "def last_page(total_items, per_page):\n    # off-by-one: drops the final partial page\n    return total_items // per_page\n",
+          "CHANGELOG.md": "# Changelog\n"
+        }
+      },
+      "expected_actions": [
+        { "tool": "edit_file", "args_pattern": { "path": "CHANGELOG.md" } }
+      ],
+      "match_mode": "subset",
+      "max_turns": 8
+    }
+  ]
+}

tests/test_skill_adherence.py

@@ -0,0 +1,226 @@
+"""Tests for turnstone.eval skill-adherence measurement mode.
+
+Two levels, neither requires a live model:
+
+* ``TestSkillComposition`` is the load-bearing plumbing proof — it seeds a
+  named skill, builds ``HeadlessSession`` under natural composition, and
+  asserts the skill body folds into ``system_messages`` for the treatment
+  arm and is absent for the control arm.  This is what makes the two arms
+  measure different things.
+* ``TestAdherenceLift`` unit-tests ``run_skill_adherence``'s lift math with
+  the per-arm runner stubbed out.
+"""
+
+import os
+import tempfile
+from collections.abc import Iterator
+from typing import Any
+
+import pytest
+from openai import OpenAI
+
+from turnstone.core.storage import get_storage, init_storage, reset_storage
+from turnstone.eval import core
+from turnstone.eval.core import HeadlessSession, run_skill_adherence
+
+_SKILL = {
+    "name": "search-first",
+    "content": (
+        "# Search First\n\nBefore answering ANY question about where something "
+        "lives in the codebase, you MUST call the `search` tool first. "
+        "SENTINEL_SKILL_BODY_MARKER."
+    ),
+}
+
+
+@pytest.fixture
+def temp_storage() -> Iterator[None]:
+    """Fresh sqlite storage in a temp dir, torn down after the test."""
+    workdir = tempfile.mkdtemp(prefix="turnstone_skill_test_")
+    reset_storage()
+    init_storage("sqlite", path=os.path.join(workdir, ".eval.db"), run_migrations=False)
+    try:
+        yield
+    finally:
+        reset_storage()
+        import shutil
+
+        shutil.rmtree(workdir, ignore_errors=True)
+
+
+def _seed_skill(skill: dict[str, str]) -> None:
+    """Seed a named skill exactly as the runner does."""
+    get_storage().create_prompt_template(
+        template_id="eval-skill",
+        name=skill["name"],
+        category="eval",
+        content=skill["content"],
+        variables="[]",
+        is_default=False,
+        org_id="",
+        created_by="eval",
+        activation="named",
+        enabled=True,
+    )
+
+
+def _system_text(session: HeadlessSession) -> str:
+    return "\n".join(m["content"] for m in session.system_messages)
+
+
+class TestSkillComposition:
+    """Prove the treatment/control arms compose different system messages."""
+
+    def test_treatment_folds_skill_into_system(self, temp_storage: None) -> None:
+        _seed_skill(_SKILL)
+        client = OpenAI(base_url="http://localhost:9/v1", api_key="dummy")
+        session = HeadlessSession(client=client, model="test-model")
+        try:
+            # Treatment arm activates the seeded skill via the real path.
+            session.set_skill(_SKILL["name"])
+            assert "SENTINEL_SKILL_BODY_MARKER" in _system_text(session)
+        finally:
+            session.close()
+
+    def test_control_omits_skill(self, temp_storage: None) -> None:
+        # Control arm: no skill seeded, no set_skill — natural default only.
+        client = OpenAI(base_url="http://localhost:9/v1", api_key="dummy")
+        session = HeadlessSession(client=client, model="test-model")
+        try:
+            assert "SENTINEL_SKILL_BODY_MARKER" not in _system_text(session)
+        finally:
+            session.close()
+
+    def test_no_system_prompt_override_in_skill_mode(self, temp_storage: None) -> None:
+        # skill_mode must NOT override the base identity — a real base prompt
+        # (persona / composed developer message) must survive, or we'd be
+        # measuring an empty prompt instead of the identity under test.
+        client = OpenAI(base_url="http://localhost:9/v1", api_key="dummy")
+        session = HeadlessSession(client=client, model="test-model")
+        try:
+            assert _system_text(session).strip(), "expected a composed base prompt"
+        finally:
+            session.close()
+
+
+class TestAdherenceLift:
+    """Unit-test the lift math with the per-arm runner stubbed."""
+
+    def test_lift_treatment_over_control(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        # Stub _run_iteration: treatment (skill != None) passes 3/3, control
+        # (skill is None) passes 1/3.  run_skill_adherence must report the
+        # difference as the lift.
+        def fake_run_iteration(**kwargs: Any) -> dict[str, Any]:
+            rate = 1.0 if kwargs.get("skill") is not None else 1.0 / 3.0
+            return {"aggregate": {"overall_pass_rate": rate}}
+
+        monkeypatch.setattr(core, "_run_iteration", fake_run_iteration)
+
+        cases = [
+            {
+                "id": "search-first",
+                "skill": _SKILL,
+                "user_prompt": "where is X?",
+                "expected_actions": [{"tool": "search"}],
+            }
+        ]
+        result = run_skill_adherence(
+            client=None,
+            base_url="http://localhost:9/v1",
+            api_key="dummy",
+            model="test-model",
+            cases=cases,
+            n_runs=3,
+            temperature=0.7,
+            max_tokens=1024,
+            reasoning_effort="medium",
+            context_window=8192,
+        )
+
+        assert len(result["cases"]) == 1
+        row = result["cases"][0]
+        assert row["case_id"] == "search-first"
+        assert row["skill"] == "search-first"
+        assert row["treatment_rate"] == pytest.approx(1.0)
+        assert row["control_rate"] == pytest.approx(1.0 / 3.0)
+        assert row["lift"] == pytest.approx(2.0 / 3.0)
+        assert row["n_runs"] == 3
+        assert result["mean_lift"] == pytest.approx(2.0 / 3.0)
+
+    def test_rejects_malformed_skill(self) -> None:
+        # A skill missing 'content' (or 'name') fails fast with a clear error,
+        # not a KeyError mid-run (Copilot review). Validation raises before any
+        # arm runs, so no _run_iteration stub is needed.
+        cases = [
+            {
+                "id": "bad-skill",
+                "skill": {"name": "x"},  # missing 'content'
+                "user_prompt": "do x",
+                "expected_actions": [{"tool": "search"}],
+            }
+        ]
+        with pytest.raises(ValueError, match="non-empty 'name' and 'content'"):
+            run_skill_adherence(
+                client=None,
+                base_url="http://localhost:9/v1",
+                api_key="dummy",
+                model="test-model",
+                cases=cases,
+                n_runs=1,
+                temperature=0.7,
+                max_tokens=1024,
+                reasoning_effort="medium",
+                context_window=8192,
+            )
+
+    def test_skipped_when_no_skill(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        # A case with no skill is not measurable — it must be skipped, not
+        # crash, and must not contribute to the mean.
+        def fake_run_iteration(**kwargs: Any) -> dict[str, Any]:
+            return {"aggregate": {"overall_pass_rate": 1.0}}
+
+        monkeypatch.setattr(core, "_run_iteration", fake_run_iteration)
+
+        cases = [{"id": "no-skill", "user_prompt": "hi", "expected_actions": []}]
+        result = run_skill_adherence(
+            client=None,
+            base_url="http://localhost:9/v1",
+            api_key="dummy",
+            model="test-model",
+            cases=cases,
+            n_runs=3,
+            temperature=0.7,
+            max_tokens=1024,
+            reasoning_effort="medium",
+            context_window=8192,
+        )
+        assert result["cases"] == []
+        assert result["mean_lift"] == 0.0
+
+    def test_mean_lift_averages_multiple_cases(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        # Two skill cases with different lifts average into mean_lift.
+        rates = iter([1.0, 0.0, 1.0, 0.5])  # t1, c1, t2, c2 -> lifts 1.0, 0.5
+
+        def fake_run_iteration(**kwargs: Any) -> dict[str, Any]:
+            return {"aggregate": {"overall_pass_rate": next(rates)}}
+
+        monkeypatch.setattr(core, "_run_iteration", fake_run_iteration)
+
+        cases = [
+            {"id": "a", "skill": _SKILL, "user_prompt": "q", "expected_actions": []},
+            {"id": "b", "skill": _SKILL, "user_prompt": "q", "expected_actions": []},
+        ]
+        result = run_skill_adherence(
+            client=None,
+            base_url="http://localhost:9/v1",
+            api_key="dummy",
+            model="test-model",
+            cases=cases,
+            n_runs=2,
+            temperature=0.7,
+            max_tokens=1024,
+            reasoning_effort="medium",
+            context_window=8192,
+        )
+        assert [c["lift"] for c in result["cases"]] == pytest.approx([1.0, 0.5])
+        assert result["mean_lift"] == pytest.approx(0.75)

turnstone/eval/cli.py

@@ -25,11 +25,77 @@ from turnstone.core.session import ChatSession
 from turnstone.eval.core import (
     NullUI,
     _append_summary_tsv,
+    _print_skill_adherence_table,
     _print_summary_table,
     _run_iteration,
+    run_skill_adherence,
 )
 
 
+def _run_skill_adherence_cli(
+    args: argparse.Namespace,
+    client: OpenAI,
+    model: str,
+    api_key: str,
+) -> None:
+    """Load a skill-scenario dataset and report per-case adherence lift."""
+    with open(args.test_file) as f:
+        suite: dict[str, Any] = json.load(f)
+
+    cases: list[dict[str, Any]] = suite["cases"]
+    for i, case in enumerate(cases):
+        if "id" not in case:
+            raise SystemExit(f"Test case {i} missing required 'id' field")
+        if "user_prompt" not in case:
+            raise SystemExit(f"Test case '{case.get('id', i)}' missing 'user_prompt'")
+        skill = case.get("skill")
+        if skill is not None and (
+            not isinstance(skill, dict) or not skill.get("name") or not skill.get("content")
+        ):
+            raise SystemExit(
+                f"Test case '{case['id']}' has a malformed 'skill' — it must be an "
+                "object with non-empty 'name' and 'content'"
+            )
+    if not any(c.get("skill") for c in cases):
+        raise SystemExit("No cases carry a 'skill' — nothing to measure for adherence")
+
+    defaults = suite.get("defaults", {})
+    resolved_n_runs: int = (
+        args.n_runs if args.n_runs is not None else int(defaults.get("n_runs", 3))
+    )
+    parallel = args.parallel if args.parallel != 0 else (os.cpu_count() or 4)
+
+    result = run_skill_adherence(
+        client=client,
+        base_url=args.base_url,
+        api_key=api_key,
+        model=model,
+        cases=cases,
+        n_runs=resolved_n_runs,
+        temperature=args.temperature,
+        max_tokens=args.max_tokens,
+        reasoning_effort=args.reasoning_effort,
+        context_window=args.context_window,
+        test_timeout=args.test_timeout,
+        parallel=parallel,
+        verbose=args.verbose,
+    )
+    result["meta"] = {
+        "model": model,
+        "base_url": args.base_url,
+        "test_suite": args.test_file,
+        "n_runs": resolved_n_runs,
+        "started": datetime.now().isoformat(),
+    }
+
+    _print_skill_adherence_table(result)
+
+    with open(args.output, "w") as f:
+        json.dump(result, f, indent=2)
+        f.write("\n")
+    print(f"Results written to {args.output}")
+
+
 def main() -> None:
     parser = argparse.ArgumentParser(
         description="Headless measurement for turnstone (scores tool use against expected actions)",
@@ -116,6 +182,15 @@ def main() -> None:
         default=1,
         help="Parallel workers (default: 1=serial, 0=auto)",
     )
+    parser.add_argument(
+        "--skill-adherence",
+        action="store_true",
+        help=(
+            "Measure skill adherence: for each case carrying a 'skill', run a "
+            "treatment arm (skill applied via the real set_skill composition "
+            "path) vs a control arm (no skill) and report the pass-rate lift"
+        ),
+    )
     parser.add_argument(
         "-v",
         "--verbose",
@@ -142,6 +217,12 @@ def main() -> None:
         assert detected is not None  # fatal=True guarantees non-None or SystemExit
         model = detected
 
+    # Skill-adherence mode is a distinct two-arm measurement — it uses natural
+    # prompt composition (no --prompt), so branch before the initial-prompt path.
+    if args.skill_adherence:
+        _run_skill_adherence_cli(args, client, model, api_key)
+        return
+
     # Resolve the initial prompt: --prompt file, else turnstone's built-in.
     initial_prompt: str | None = None
     if args.prompt:

turnstone/eval/core.py

@@ -32,7 +32,7 @@ from openai import OpenAI
 
 from turnstone.core.providers import LLMProvider, create_client, create_provider
 from turnstone.core.session import ChatSession
-from turnstone.core.storage import init_storage, reset_storage
+from turnstone.core.storage import get_storage, init_storage, reset_storage
 from turnstone.core.tools import INTERACTIVE_TOOLS, PRIMARY_KEY_MAP
 from turnstone.core.trajectory import Role, turn_from_dict
 
@@ -441,12 +441,26 @@ def _run_single_test(
     log_prefix: str = "",
     test_timeout: int = 300,
     tool_overrides: dict[str, dict[str, Any]] | None = None,
+    skill: dict[str, Any] | None = None,
+    skill_mode: bool = False,
 ) -> dict[str, Any]:
     """Run a single test case once in an isolated temp directory.
 
     Uses os.chdir (process-global), so concurrent calls must run in
     separate processes (see _run_and_score_subprocess / --parallel).
 
+    When ``skill_mode`` is True the session is built WITHOUT a system-prompt
+    override so the model runs under turnstone's natural prompt composition
+    (the base identity under test).  If ``skill`` is given it is seeded into
+    the temp DB and activated via the real ``set_skill`` path — the skill
+    flows through turnstone's natural composition exactly as in production,
+    landing wherever THAT checkout places a named skill (the system message,
+    or a separate context turn).  This harness measures adherence regardless
+    of placement, which is the whole point of comparing across checkouts.
+    ``skill`` None is the control arm (natural default, no skill).  When
+    ``skill_mode`` is False behaviour is unchanged — the system prompt is
+    overridden as before.
+
     Returns dict with keys: tool_log, final_content, message_count,
     elapsed, usage.
     """
@@ -471,6 +485,24 @@ def _run_single_test(
 
         os.chdir(workdir)
 
+        # Skill-adherence treatment arm: seed the named skill into the temp DB
+        # (once — before the retry loop) so set_skill can activate it through
+        # the real composition path.  The subprocess/serial DB is fresh per
+        # run, so template_id "eval-skill" never collides.
+        if skill_mode and skill is not None:
+            get_storage().create_prompt_template(
+                template_id="eval-skill",
+                name=skill["name"],
+                category="eval",
+                content=skill["content"],
+                variables="[]",
+                is_default=False,
+                org_id="",
+                created_by="eval",
+                activation="named",
+                enabled=True,
+            )
+
         max_turns = case.get("max_turns", 15)
         # Retry on transient API errors to avoid poisoning eval scores
         tool_log: list[dict[str, Any]] = []
@@ -489,7 +521,10 @@ def _run_single_test(
             session = HeadlessSession(
                 client=run_client,
                 model=model,
-                system_prompt_override=system_prompt,
+                # skill_mode uses turnstone's natural composition (no override)
+                # so the skill folds in wherever the checkout under test places
+                # a named skill (system message, or a separate context turn).
+                system_prompt_override=None if skill_mode else system_prompt,
                 instructions=None,
                 temperature=temperature,
                 max_tokens=max_tokens,
@@ -499,6 +534,11 @@ def _run_single_test(
                 tool_truncation=2000,
                 tool_overrides=tool_overrides,
             )
+            if skill_mode and skill is not None:
+                # Activate the seeded skill via the production path:
+                # _load_skills() -> _init_system_messages() composes the
+                # skill body into session.system_messages.
+                session.set_skill(skill["name"])
             executor: ThreadPoolExecutor | None = None
             try:
                 executor = ThreadPoolExecutor(max_workers=1)
@@ -587,6 +627,8 @@ def _run_and_score_subprocess(params: dict[str, Any]) -> dict[str, Any]:
             log_prefix="",
             test_timeout=params["test_timeout"],
             tool_overrides=params.get("tool_overrides"),
+            skill=params.get("skill"),
+            skill_mode=params.get("skill_mode", False),
         )
 
         score_result = score_run(
@@ -842,6 +884,8 @@ def _run_iteration_parallel(
     parallel: int,
     prompt_variants: dict[str, list[str]] | None = None,
     tool_overrides: dict[str, dict[str, Any]] | None = None,
+    skill: dict[str, Any] | None = None,
+    skill_mode: bool = False,
 ) -> dict[str, Any]:
     """Run all test cases in parallel using ProcessPoolExecutor."""
     # Build work items for every (case, run) combination
@@ -872,6 +916,8 @@ def _run_iteration_parallel(
                     "test_timeout": test_timeout,
                     "original_user_prompt": case["user_prompt"],
                     "tool_overrides": tool_overrides,
+                    "skill": skill,
+                    "skill_mode": skill_mode,
                 }
             )
 
@@ -993,6 +1039,8 @@ def _run_iteration(
     api_key: str = "",
     prompt_variants: dict[str, list[str]] | None = None,
     tool_overrides: dict[str, dict[str, Any]] | None = None,
+    skill: dict[str, Any] | None = None,
+    skill_mode: bool = False,
 ) -> dict[str, Any]:
     """Run all test cases n_runs times and score them."""
     if parallel > 1 and base_url:
@@ -1011,6 +1059,8 @@ def _run_iteration(
             parallel=parallel,
             prompt_variants=prompt_variants,
             tool_overrides=tool_overrides,
+            skill=skill,
+            skill_mode=skill_mode,
         )
 
     case_results: dict[str, Any] = {}
@@ -1061,6 +1111,8 @@ def _run_iteration(
                     log_prefix=log_prefix,
                     test_timeout=test_timeout,
                     tool_overrides=tool_overrides,
+                    skill=skill,
+                    skill_mode=skill_mode,
                 )
 
                 score_result = score_run(
@@ -1183,6 +1235,109 @@ def _run_iteration(
     return _aggregate_case_results(cases, case_results, total_tokens)
 
 
+# ─── Skill-adherence driver ──────────────────────────────────────────────────
+
+
+def run_skill_adherence(
+    client: Any,
+    base_url: str,
+    api_key: str,
+    model: str,
+    cases: list[dict[str, Any]],
+    n_runs: int,
+    temperature: float,
+    max_tokens: int,
+    reasoning_effort: str,
+    context_window: int,
+    test_timeout: int = 300,
+    parallel: int = 1,
+    verbose: bool = False,
+) -> dict[str, Any]:
+    """Measure how much a named skill changes tool-use behaviour.
+
+    For every case that carries a ``skill`` this runs two arms ``n_runs``
+    times each, scoring both against the case's ``expected_actions``:
+
+    * **treatment** — the skill is applied via the real ``set_skill``
+      composition path (``skill_mode=True, skill=<case skill>``); where its
+      body lands (system message or a context turn) depends on the checkout;
+    * **control** — the same base identity with no skill
+      (``skill_mode=True, skill=None``).
+
+    The adherence lift is ``pass_rate(treatment) - pass_rate(control)``.  The
+    control isolates the skill's causal effect: a scenario the model passes
+    anyway yields ~0 lift and is uninformative — that near-zero IS the signal.
+
+    Returns ``{"cases": [{case_id, skill, treatment_rate, control_rate, lift,
+    n_runs}, ...], "mean_lift": float}``.
+    """
+    case_results: list[dict[str, Any]] = []
+    skill_cases = [c for c in cases if c.get("skill")]
+
+    # Validate skill shape up front so a malformed dataset fails with a clear
+    # message instead of a KeyError mid-run (after arms have already started).
+    for case in skill_cases:
+        s = case["skill"]
+        if not isinstance(s, dict) or not s.get("name") or not s.get("content"):
+            raise ValueError(
+                f"case {case.get('id', '?')!r}: 'skill' must be an object with "
+                "non-empty 'name' and 'content'"
+            )
+
+    for ci, case in enumerate(skill_cases):
+        skill = case["skill"]
+        # Drop the skill key from the case handed to the runner — it is
+        # supplied out-of-band per arm, not read from the case dict.
+        arm_case = {k: v for k, v in case.items() if k != "skill"}
+        print(
+            f"\n  {CYAN}[{ci + 1}/{len(skill_cases)}]{RESET} "
+            f"{BOLD}{case['id']}{RESET} — skill {DIM}'{skill['name']}'{RESET}"
+        )
+
+        arm_rates: dict[str, float] = {}
+        for arm, arm_skill in (("treatment", skill), ("control", None)):
+            print(f"    {DIM}{arm}{RESET}")
+            iter_result = _run_iteration(
+                client=client,
+                model=model,
+                system_prompt="",
+                cases=[arm_case],
+                n_runs=n_runs,
+                temperature=temperature,
+                max_tokens=max_tokens,
+                reasoning_effort=reasoning_effort,
+                context_window=context_window,
+                verbose=verbose,
+                test_timeout=test_timeout,
+                # Never fast-fail: the treatment arm's pass rate must be
+                # counted over every run, and control runs are expected to
+                # fail — skipping them would corrupt the lift.
+                fast_fail=False,
+                parallel=parallel,
+                base_url=base_url,
+                api_key=api_key,
+                skill=arm_skill,
+                skill_mode=True,
+            )
+            arm_rates[arm] = iter_result["aggregate"]["overall_pass_rate"]
+
+        treatment_rate = arm_rates["treatment"]
+        control_rate = arm_rates["control"]
+        case_results.append(
+            {
+                "case_id": case["id"],
+                "skill": skill["name"],
+                "treatment_rate": treatment_rate,
+                "control_rate": control_rate,
+                "lift": treatment_rate - control_rate,
+                "n_runs": n_runs,
+            }
+        )
+
+    mean_lift = sum(c["lift"] for c in case_results) / len(case_results) if case_results else 0.0
+    return {"cases": case_results, "mean_lift": mean_lift}
+
+
 # ─── Summary & reporting ─────────────────────────────────────────────────────
 
 
@@ -1243,6 +1398,37 @@ def _print_summary_table(iter_result: dict[str, Any]) -> None:
     )
 
 
+def _print_skill_adherence_table(result: dict[str, Any]) -> None:
+    """Print a per-case treatment/control/lift table plus the mean lift."""
+    rows = result.get("cases", [])
+    if not rows:
+        print("\n  No skill-bearing cases to measure.")
+        return
+
+    max_id = max(len(str(r["case_id"])) for r in rows)
+    max_id = max(max_id, 4)  # min "CASE" header
+
+    print(f"\n{BOLD}  {'CASE'.ljust(max_id)}  {'TREAT':>6}  {'CTRL':>6}  {'LIFT':>7}{RESET}")
+    print(f"  {'─' * (max_id + 25)}")
+
+    for r in rows:
+        lift = r["lift"]
+        color = GREEN if lift > 0.01 else (RED if lift < -0.01 else DIM)
+        print(
+            f"  {str(r['case_id']).ljust(max_id)}  "
+            f"{r['treatment_rate']:>6.2f}  {r['control_rate']:>6.2f}  "
+            f"{color}{lift:>+7.2f}{RESET}"
+        )
+
+    print(f"  {'─' * (max_id + 25)}")
+    mean = result.get("mean_lift", 0.0)
+    mcolor = GREEN if mean > 0.01 else (RED if mean < -0.01 else DIM)
+    print(
+        f"  {BOLD}{'MEAN'.ljust(max_id)}{RESET}  {'':>6}  {'':>6}  "
+        f"{mcolor}{BOLD}{mean:>+7.2f}{RESET}"
+    )
+
+
 def _append_summary_tsv(
     path: str,
     iter_result: dict[str, Any],