R-896: the volume-persistence gate refuses a host that is not a scratch host (no containers on DooPlex)

check() refuses (HARNESS REFUSED, exit 3) before any docker call unless /opt/upg exists (the bench)
or FELHOM_SCRATCH_HOST=1. TestVenue red before the guard; new end-to-end decoy.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0159rPz1ZhFKsS53msqPYxtS

scripts/check-volume-persistence.py

@@ -944,9 +944,33 @@ def collect_apps(root: Path, only=None, include_unavailable=False):
 # apart, so a summary can say whether the gate ran at all. Pinned by test_check_volume_persistence.py.
 HARNESS_REFUSED = 3
 
+# R-896 (2026-10-10): the real prober builds a canary image and runs every template on the LOCAL Docker. Run on DooPlex
+# (the production host) it acted there — and its self-test failed for every app. So it runs only where the host says it
+# is a scratch host: the bench's layout directory, or an explicit FELHOM_SCRATCH_HOST=1 from the person who knows the box.
+# Fail closed: no marker → HARNESS REFUSED before any docker call. Pinned by TestVenue (test_check_volume_persistence.py).
+SCRATCH_ENV = "FELHOM_SCRATCH_HOST"
+SCRATCH_MARKERS = (Path("/opt/upg"),)   # the bench guest's harness layout (upgrade-test.py ROOT)
+
+
+def venue_refusal() -> str:
+    """Empty when this host may run containers for the gate; otherwise the reason it may not."""
+    if os.environ.get(SCRATCH_ENV) == "1":
+        return ""
+    if any(m.is_dir() for m in SCRATCH_MARKERS):
+        return ""
+    return ("this host is not a scratch host (no %s, %s is not 1) — the gate runs real containers on the local "
+            "Docker and must not act on a production host (R-896). Run it on the bench, or set %s=1 on a scratch "
+            "host." % (" / ".join(str(m) for m in SCRATCH_MARKERS), SCRATCH_ENV, SCRATCH_ENV))
+
 
 def check(root: Path, only=None, prober=docker_prober, include_unavailable=False,
           evidence: Path | None = None, skip_self_test=False) -> int:
+    if prober is docker_prober:
+        why = venue_refusal()
+        if why:
+            print(f"ERROR: {why}", file=sys.stderr)
+            print("HARNESS REFUSED — not a scratch host; no container was started")
+            return HARNESS_REFUSED
     apps, skipped = collect_apps(root, only, include_unavailable)
     if skipped:
         print(f"skipping {len(skipped)} app(s) not offered for new installs: {', '.join(skipped)}")

scripts/test_check_volume_persistence.py

@@ -9,6 +9,7 @@ Run: python3 scripts/test_check_volume_persistence.py
 """
 import importlib.util
 import io
+import os
 import sys
 import tempfile
 import unittest
@@ -597,5 +598,64 @@ class TestEmptyBind(unittest.TestCase):
         self.assertFalse(any("R-805" in w for w in why), why)
 
 
+class TestVenue(unittest.TestCase):
+    """R-896: the REAL prober runs containers on the local Docker. On a host that is not a scratch host (DooPlex is
+    the production host) the gate refuses BEFORE any docker call — the consequence, not the helper: the prober is
+    never reached."""
+
+    def setUp(self):
+        self._env = os.environ.pop(cvp.SCRATCH_ENV, None)
+        self._markers = cvp.SCRATCH_MARKERS
+        self._real = cvp.docker_prober
+        self.calls = []
+
+    def tearDown(self):
+        if self._env is not None:
+            os.environ[cvp.SCRATCH_ENV] = self._env
+        else:
+            os.environ.pop(cvp.SCRATCH_ENV, None)
+        cvp.SCRATCH_MARKERS = self._markers
+
+    def _run(self):
+        def spy(*a, **kw):
+            self.calls.append(a)
+            raise AssertionError("the real prober was reached")
+        with tempfile.TemporaryDirectory() as td:
+            d = Path(td) / "templates" / "demo"
+            d.mkdir(parents=True)
+            (d / "docker-compose.yml").write_text("services:\n  s:\n    image: alpine:3.22\n")
+            (d / ".felhom.yml").write_text("slug: demo\n")
+            orig_self_test = cvp.self_test
+            cvp.self_test = lambda prober: spy()
+            try:
+                buf = io.StringIO()
+                with redirect_stdout(buf):
+                    rc = cvp.check(Path(td))   # the default prober = the real one
+            finally:
+                cvp.self_test = orig_self_test
+        return rc, buf.getvalue()
+
+    def test_no_marker_refuses_before_any_docker_call(self):
+        with tempfile.TemporaryDirectory() as nowhere:
+            cvp.SCRATCH_MARKERS = (Path(nowhere) / "absent",)
+            rc, out = self._run()
+        self.assertEqual(rc, cvp.HARNESS_REFUSED, out)
+        self.assertIn("not a scratch host", out)
+        self.assertEqual(self.calls, [])
+
+    def test_marker_directory_lets_it_run(self):
+        with tempfile.TemporaryDirectory() as here:
+            cvp.SCRATCH_MARKERS = (Path(here),)
+            with self.assertRaises(AssertionError):   # it went on to the self-test (the spy)
+                self._run()
+
+    def test_explicit_env_lets_it_run(self):
+        with tempfile.TemporaryDirectory() as nowhere:
+            cvp.SCRATCH_MARKERS = (Path(nowhere) / "absent",)
+            os.environ[cvp.SCRATCH_ENV] = "1"
+            with self.assertRaises(AssertionError):
+                self._run()
+
+
 if __name__ == "__main__":
     unittest.main(verbosity=2)

scripts/test_gate_decoys.py

@@ -876,7 +876,7 @@ def volume_persistence_cases():
         empty = os.path.join(ws, "nobin")
         os.makedirs(empty)
 
-        def run(name, expect_rc, must=(), path=stub, templates=True):
+        def run(name, expect_rc, must=(), path=stub, templates=True, scratch=True):
             global ran
             cat = os.path.join(ws, "cat")
             shutil.rmtree(cat, ignore_errors=True)
@@ -889,7 +889,7 @@ def volume_persistence_cases():
                     "services:\n  demo:\n    image: example.org/demo:1.0\n    volumes:\n      - demo_data:/data\n"
                     "volumes:\n  demo_data:\n")
             r = subprocess.run([sys.executable, os.path.join(cat, "scripts", "check-volume-persistence.py")],
-                               cwd=cat, env={"PATH": path}, capture_output=True, text=True, input="", timeout=300)
+                               cwd=cat, env=dict({"PATH": path}, **({"FELHOM_SCRATCH_HOST": "1"} if scratch else {})), capture_output=True, text=True, input="", timeout=300)
             out = r.stdout + r.stderr
             ran += 1
             miss = [m for m in must if m not in out]
@@ -902,6 +902,8 @@ def volume_persistence_cases():
         run("LABEL: a runtime that answers nothing", 3, ("HARNESS REFUSED", "failed its canary"))
         run("LABEL: no docker at all", 3, ("HARNESS REFUSED",), path=empty)
         run("LABEL: nothing to judge", 3, ("HARNESS REFUSED",), templates=False)
+        # R-896: a host that does not say it is a scratch host (DooPlex) — refused before the canary is even built.
+        run("LABEL: not a scratch host", 3, ("HARNESS REFUSED", "not a scratch host"), scratch=False)
     finally:
         shutil.rmtree(ws, ignore_errors=True)