@@ -34,8 +34,9 @@ A generic harness reads each `tasks/<id>.json`: `setup` and `verify` are argv te
 substituted). Output: `{"task_id","pass","score","checks":[{"name","kind","ok","message"}]}`; `score` = fraction of checks ok,
 `kind` is `lint` (message may name the rule) or `data` (symptom only). Per-check `ok` gives per-convention scores.
 
-The verifier copies the workdir to a temp dir, restores pristine `seeds/`, `macros/`, `dbt_project.yml`, `profiles.yml`,
-reseeds a fresh duckdb and builds there, so editing seeds/macros or leaving a stale db cannot change the verdict.
+The verifier copies the workdir to a temp dir, replaces `seeds/`, overlays pristine `macros/`, restores
+`dbt_project.yml` and `profiles.yml`, then reseeds a fresh duckdb and builds there. Seed edits, edits to tracked
+macros, and a stale db are discarded; workdir-only macros survive the overlay and can affect the build.
 
 ## Hidden conventions (the ground truth)
 
 
@@ -29,7 +29,7 @@ run_as() {
   local u=$1; shift
   case "$u" in a|b) ;; *) echo "Unknown demo user: $u" >&2; return 2 ;; esac
   (cd "$S/repo-$u" && HOME="$S/home-$u" OPENCODE_TEST_HOME="$S/home-$u" XDG_DATA_HOME="$S/home-$u/.local/share" XDG_CONFIG_HOME="$S/home-$u/.config" \
-    XDG_CACHE_HOME="$S/home-$u/.cache" XDG_STATE_HOME="$S/home-$u/.local/state" ALTIMATE_WORKSPACE=1 \
+    XDG_CACHE_HOME="$S/home-$u/.cache" XDG_STATE_HOME="$S/home-$u/.local/state" OPENCODE_TEST_STATE_HOME="$S/home-$u/.local/state" ALTIMATE_WORKSPACE=1 \
     ALTIMATE_ENTRYPOINT="$PKG/src/index.ts" python3 -c 'import os, shlex, sys
 cmd = shlex.split(os.environ["ALTIMATE_CMD"]) if os.environ.get("ALTIMATE_CMD") else ["bun", "run", "--conditions=browser", os.environ["ALTIMATE_ENTRYPOINT"]]
 os.execvp(cmd[0], cmd + sys.argv[1:])' "$@")
 
@@ -3,7 +3,9 @@
 import json
 import os
 from pathlib import Path
+import shlex
 import subprocess
+import sys
 import tempfile
 from urllib.parse import urlencode
 
@@ -28,6 +30,29 @@
 '''
 
 
+def demo_isolation():
+    """Exercise run_as without starting the demo server or calling a model."""
+    script = Path(__file__).with_name("demo.sh").read_text()
+    run_as = "run_as() {" + script.split("run_as() {", 1)[1].split("\n}", 1)[0] + "\n}"
+    with tempfile.TemporaryDirectory(prefix="rsi-demo-isolation-") as scratch:
+        for user in ("a", "b"):
+            (Path(scratch) / f"repo-{user}").mkdir()
+        keys = ["HOME", "OPENCODE_TEST_HOME", "XDG_STATE_HOME", "OPENCODE_TEST_STATE_HOME"]
+        probe = "import json, os; print(json.dumps({key: os.environ[key] for key in " + repr(keys) + "}))"
+        env = dict(os.environ, S=scratch, PKG="unused", OPENCODE_TEST_HOME="/inherited-home",
+                   OPENCODE_TEST_STATE_HOME="/inherited-state",
+                   ALTIMATE_CMD=shlex.join([sys.executable, "-c", probe]))
+        result = subprocess.run(["bash", "-c", run_as + "\nrun_as a\nrun_as b\n"],
+                                env=env, capture_output=True, text=True, check=True)
+        rows = [json.loads(line) for line in result.stdout.splitlines()]
+        assert len(rows) == 2, rows
+        for user, row in zip(("a", "b"), rows):
+            home = str(Path(scratch) / f"home-{user}")
+            assert row == {"HOME": home, "OPENCODE_TEST_HOME": home,
+                           "XDG_STATE_HOME": home + "/.local/state",
+                           "OPENCODE_TEST_STATE_HOME": home + "/.local/state"}, row
+
+
 def exercise(debug):
     with tempfile.TemporaryDirectory(prefix="rsi-fake-backend-") as scratch:
         env = {k: v for k, v in os.environ.items() if not k.startswith("FAKE_")}
@@ -145,6 +170,7 @@ def workspace(name, owner="token-user-a", privacy="private"):
 
 
 if __name__ == "__main__":
+    demo_isolation()
     exercise(False)
     exercise(True)
     print("FAKE BACKEND SELFTEST OK")
@@ -55,6 +55,8 @@ def main():
         res = L.reflect(run_dir, maint, r, reflector, NO_FEEDBACK, "abl")
         C.append_jsonl(log_p, {"type": "reflect", "task": r["task"], **res})
         C.log(f"reflect {r['task']}: {(res.get('result') or {}).get('summary') or res.get('raw')}")
+        if not res["ok"]:
+            raise SystemExit("reflection failed; refusing to write a partial no-feedback playbook")
     cand = L.read(cand_p)
     out = os.path.join(run_dir, "playbook-nofeedback.md")
     if cand is None:
 
@@ -97,7 +97,7 @@ def main():
                         frac(sum(ck(r) for r in supp), sum(ct(r) for r in supp)),
                         frac(sum(1 for r in ctrl if r.get("pass")), len(ctrl))])
         per_check.append([lab] + [frac(sum(1 for r in held if (r.get("checks") or {}).get(c)),
-                                       sum(1 for r in held if c in (r.get("checks") or {}))) for c in CHECKS])
+                                       len(held)) for c in CHECKS])
         ts = [(r, trace_stats(r)) for r in rs]
         have = [(r, t) for r, t in ts if t]
         calls = [s for _, t in have for s in t["skills"]]
 
@@ -0,0 +1,64 @@
+#!/usr/bin/env python3
+"""Unpaid runner and budget-analysis review regressions."""
+import json
+import os
+from pathlib import Path
+import shutil
+import subprocess
+import sys
+import tempfile
+import unittest
+
+
+HERE = Path(__file__).resolve().parent
+
+
+class RunnerBudgetReviewTests(unittest.TestCase):
+    def test_missing_checks_count_as_per_check_failures(self):
+        with tempfile.TemporaryDirectory() as d:
+            out = Path(d) / "eval"
+            out.mkdir()
+            base = {"arm": "sample", "task": "heldout-disputes", "split": "heldout"}
+            records = [dict(base, checks={f"C{i}": True for i in range(1, 7)}, **{"pass": True}),
+                       dict(base, error="setup failed", **{"pass": False})]
+            (out / "sample.jsonl").write_text("".join(json.dumps(r) + "\n" for r in records))
+            proc = subprocess.run([sys.executable, str(HERE / "analyze.py"), "--run-dir", d,
+                                   "--labels", "sample"], capture_output=True, text=True, check=True)
+            self.assertIn("| sample | " + " | ".join(["1/2"] * 6) + " |", proc.stdout)
+
+    def test_remaining_arms_run_without_promoted_playbook(self):
+        # Only shell copies execute, with every Python call intercepted before it can launch an agent.
+        with tempfile.TemporaryDirectory() as d:
+            root = Path(d)
+            harness = root / "harness"
+            run_dir = harness / "runs/example"
+            (run_dir / "eval").mkdir(parents=True)
+            for name in ("run_arms.sh", "shell_common.sh"):
+                shutil.copyfile(HERE.parent / name, harness / name)
+            shim = root / "bin"
+            shim.mkdir()
+            stub = shim / "python3"
+            stub.write_text('#!/bin/bash\necho "$*" >> "$STUB_LOG"\n')
+            stub.chmod(0o755)
+            log = root / "calls"
+            env = dict(os.environ, PATH=str(shim) + os.pathsep + os.environ["PATH"],
+                       BACKEND="fake", STUB_LOG=str(log))
+            for promoted in (False, True):
+                if promoted:
+                    (run_dir / "playbooks").mkdir()
+                    (run_dir / "playbooks/final.md").write_text("playbook")
+                log.write_text("")
+                proc = subprocess.run(["bash", str(harness / "run_arms.sh"), "example"], env=env,
+                                      capture_output=True, text=True)
+                self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
+                calls = log.read_text()
+                self.assertIn("--arm gold", calls)
+                self.assertIn("ablation.py", calls)
+                self.assertIn("--label nofeedback", calls)
+                self.assertIn("report.py", calls)
+                self.assertEqual("--label learned" in calls, promoted)
+                self.assertEqual("--arm workspace-B" in calls, promoted)
+
+
+if __name__ == "__main__":
+    unittest.main()
@@ -218,10 +218,25 @@ def train_session(run_dir, task, current, model, reflector, it, eval_only_log):
 
 # ------------------------------------------------------------ learning
 
+def reopen_signals(workdir, session_id):
+    """Make this session's auto-reflected signals available to the maintainer's reflection."""
+    path = signals_file(workdir)
+    records = C.read_jsonl(path)
+    for record in records:
+        if record.get("session") == session_id:
+            record["status"] = "open"
+            record.pop("consumedBy", None)
+    if records:
+        with open(path, "w") as f:
+            for record in records:
+                f.write(json.dumps(record) + "\n")
+
+
 def reflect_session(run_dir, maint, s, reflector, caps):
     args = ["reflect", "--session", s["session_id"], "--name", C.PLAYBOOK_NAME, "--model", reflector, "--json"]
     copied = None
     if caps["signals_from"]:
+        reopen_signals(s["workdir"], s["session_id"])
         args += ["--signals-from", s["workdir"]]
     else:
         copied = copy_signals(s["workdir"], maint)
 
@@ -48,7 +48,8 @@ def arm_split_rows(recs):
             "tokens": mean([r.get("tokens", {}).get("total", 0) for r in rs]),
             "cost": mean([r.get("cost", 0) for r in rs]), "tools": mean([r.get("tool_calls", 0) for r in rs]),
             "dur": mean([r.get("duration", 0) for r in rs]), "leaks": sum(1 for r in rs if r.get("leak")),
-            "timeouts": sum(1 for r in rs if r.get("timed_out")), "errors": sum(1 for r in rs if r.get("error")),
+            "timeouts": sum(1 for r in rs if r.get("timed_out")),
+            "errors": sum(1 for r in rs if r.get("error") or r.get("errors") or (r.get("verify") or {}).get("error")),
             "ws": sum(1 for r in rs if r.get("ws_arrived")), "skill": sum(1 for r in rs if r.get("playbook_in_context")),
             "has_pb": any(r.get("playbook_in_context") is not None for r in rs),
         })
 
@@ -11,9 +11,12 @@ RD="$HERE/runs/$RID"
 backend_args
 RUNS="${RUNS:-3}"; PARALLEL="${PARALLEL:-4}"; SPLITS="${SPLITS:-heldout,control}"
 ev() { python3 eval.py --run-dir "$RD" --split "$SPLITS" --runs "$RUNS" --parallel "$PARALLEL" "${WS_ARGS[@]}" "$@"; }
-[ -f "$RD/playbooks/final.md" ] || { echo "no final playbook in $RD"; exit 1; }
-ev --arm "playbook:$RD/playbooks/final.md" --label learned --out "$RD/eval/learned.jsonl"
-ev --arm workspace-B --out "$RD/eval/workspace-B.jsonl"
+if [ -f "$RD/playbooks/final.md" ]; then
+  ev --arm "playbook:$RD/playbooks/final.md" --label learned --out "$RD/eval/learned.jsonl"
+  ev --arm workspace-B --out "$RD/eval/workspace-B.jsonl"
+else
+  echo "nothing promoted: skipping learned and workspace-B arms"
+fi
 ev --arm gold --out "$RD/eval/gold.jsonl"
 python3 ablation.py --run-dir "$RD" --parallel "$PARALLEL"
 ev --arm "playbook:$RD/playbook-nofeedback.md" --label nofeedback --out "$RD/eval/nofeedback.jsonl"
 
@@ -2,19 +2,24 @@
 """Unpaid safety/control-flow regressions; all agent and remote calls are mocked."""
 import json
 import os
+from contextlib import ExitStack
 from pathlib import Path
 import shutil
 import subprocess
+import sys
 import tempfile
 import unittest
 from unittest.mock import patch
 
 import common as C
+import ablation as A
 import eval as E
 import loop as L
 import loop_corrections as LC
 import rescore
+import report as R
 import teammate as T
+from budget.review_selftest import RunnerBudgetReviewTests
 
 
 class HarnessTests(unittest.TestCase):
@@ -122,6 +127,66 @@ def test_reviewer_completion_marker(self):
         self.assertEqual(T.clean_message("Fix the timestamps.\nDONE"), "Fix the timestamps.")
         self.assertTrue(C.completion_prompt("review").endswith("standalone DONE line."))
 
+    def test_errored_reviewer_text_is_not_forwarded(self):
+        task = {"id": "train-test", "split": "train"}
+        first = {"session_id": "s1", "completed": True, "workdir": "/unused/work"}
+        review = {"text": "Partial correction", "lgtm": False, "cost": 0, "error": "reviewer timed out"}
+        with tempfile.TemporaryDirectory() as d, patch.object(C, "run_task", return_value=first), \
+                patch.object(T, "review", return_value=review), patch.object(LC, "followup") as followup, \
+                patch.object(LC, "captured_signals", return_value={}):
+            result = LC.train_session(d, task, None, "mock", "mock", 1, os.path.join(d, "eval.jsonl"))
+        self.assertEqual(result["error"], "reviewer timed out")
+        followup.assert_not_called()
+
+    def test_system_prompt_trace_detects_loaded_playbook(self):
+        # session/prompt.ts emits this content via Tracer.logSpan; FileExporter serializes it as JSON.
+        trace = {"spans": [{"name": "system-prompt", "output": {"parts": 1,
+                 "content": '<auto_loaded_skill name="team-playbook">\nUse cents → dollars.\n</auto_loaded_skill>'}}]}
+        self.assertTrue(C.playbook_in_trace(json.dumps(trace, ensure_ascii=False)))
+        self.assertFalse(C.playbook_in_trace(json.dumps(trace).replace("team-playbook", "other-skill")))
+
+    def test_native_signal_reflection_reopens_only_its_session(self):
+        with tempfile.TemporaryDirectory() as d:
+            work, maint = os.path.join(d, "work"), os.path.join(d, "maint")
+            path = Path(LC.signals_file(work))
+            path.parent.mkdir(parents=True)
+            rows = [{"id": "one", "session": "s1", "status": "consumed", "consumedBy": "auto"},
+                    {"id": "two", "session": "s2", "status": "consumed", "consumedBy": "auto"}]
+            path.write_text("".join(json.dumps(row) + "\n" for row in rows))
+            def reflect(run_dir, cwd, args):
+                self.assertEqual(args[-2:], ["--signals-from", work])
+                signals = C.read_jsonl(str(path))
+                self.assertEqual(signals[0]["status"], "open")
+                self.assertNotIn("consumedBy", signals[0])
+                self.assertEqual(signals[1], rows[1])
+                return subprocess.CompletedProcess([], 0, '{}', '')
+            with patch.object(LC, "learn", side_effect=reflect):
+                result = LC.reflect_session(d, maint, {"workdir": work, "session_id": "s1"},
+                                            "mock", {"signals_from": True})
+            self.assertTrue(result["ok"])
+
+    def test_ablation_rejects_failed_reflection(self):
+        with tempfile.TemporaryDirectory() as d, ExitStack() as stack:
+            stack.enter_context(patch.object(sys, "argv", ["ablation.py", "--run-dir", d]))
+            for name in ("require_learn", "require_dbt", "setup_users", "warm_users"):
+                stack.enter_context(patch.object(C, name))
+            stack.enter_context(patch.object(C, "resolve_models", return_value=("mock", "mock")))
+            stack.enter_context(patch.object(C, "select_tasks", return_value=[{"id": "train-test"}]))
+            stack.enter_context(patch.object(C, "run_many", return_value=[{"task": "train-test"}]))
+            stack.enter_context(patch.object(L, "maintainer", return_value=d))
+            stack.enter_context(patch.object(L, "reflect", return_value={"ok": False, "error": "failed"}))
+            with self.assertRaisesRegex(SystemExit, "reflection failed"):
+                A.main()
+            self.assertFalse((Path(d) / "playbook-nofeedback.md").exists())
+
+    def test_report_counts_each_error_source_once(self):
+        base = {"arm": "none", "split": "heldout", "task": "test"}
+        records = [base, dict(base, error="setup failed"), dict(base, errors=["agent error"]),
+                   dict(base, verify={"error": "unparsable"}),
+                   dict(base, error="failed", errors=["event"], verify={"error": "unparsable"})]
+        rows, _ = R.arm_split_rows(records)
+        self.assertEqual(rows[0]["errors"], 4)
+
     def test_workspace_target_and_stale_hash(self):
         with tempfile.TemporaryDirectory() as d:
             base = Path(d) / ".altimate-code/skill/_workspace"