feat(conformance): R-STATE-03 PROVEN — DR restore from off-cluster backup, executed

The second requirement ever proven on this project, and the one the operator had attested
was working but the suite could not attest to. Test authored 2026-08-31 carrying a
`conformance.home.k3s/never-executed` annotation; executed for the first time tonight.

  test      dr-restore                              run   20260902T001846Z-fdc3265
  req       R-STATE-03 (safety-critical, DR)        junit passed / 0 failed / 0 skipped
  sentinel  DR_RESTORED_FROM_OFF_CLUSTER_BACKUP_OK  proof proven
  sha       fdc3265 (clean tree)                    kube  v1.36.4+k3s1

WHAT IT ACTUALLY PROVED, in three steps:
1. The backup target is genuinely OFF-CLUSTER. BSL velero/default phase=Available at
   http://192.168.1.9:9000 — endpoint resolved and asserted to be none of: loopback, any
   node Internal/ExternalIP, any Service ClusterIP, any Pod IP, inside any node podCIDR, or
   inside any ServiceCIDR.
2. Backups are SCHEDULED, not merely possible — velero-daily and
   velero-daily-resources-only, both phase=Enabled, last run 2026-09-01T03:00:27Z.
3. The drill: seed a nanosecond-unique canary onto a longhorn PVC; Velero Backup to
   Completed with a kopia PodVolumeBackup carrying a real snapshotID (volume DATA reached
   the remote repository, not just resource JSON); ANNIHILATE the source — delete the
   namespace, poll until gone, force-delete the PV, FAIL if the original PV survived;
   Restore into a clean namespace; assert the restored PVC bound a DIFFERENT PV and the
   canary bytes are byte-identical.
The bytes could only have come from outside the cluster.

VALIDATES THE PARSER FIX FROM 57d4df8. This log contains SEVEN occurrences of
DR_RESTORE_SKIPPED_OK — every one of them inside a `=== COMMAND` script echo — and the pass
token twice in real `=== STDOUT`. Under the pre-fix scanner this genuine pass would have been
recorded as `skipped`, exactly as node-loss was. The section-parsing rule is doing real work,
not theoretical work.

CLEANUP VERIFIED against a pre-run baseline: Velero backups back to 7 (its own
chainsaw-*-dr-backup deleted via DeleteBackupRequest), 0 chainsaw backups, 0 restores, 0
orphan Released/Available PVs, both scratch namespaces gone, 13/13 Kustomizations Ready, 0
pods not Running. It wrote into and cleaned out of the SAME bucket and kopia repository the
live nightly schedule uses, and left it as it found it.

SEPARATE FINDING, cleaned up: `chainsaw-stable-worm` was still Active from 2026-08-25 —
empty residue from an earlier conformance session that never tore its namespace down. Deleted.
Worth knowing that Chainsaw namespace leaks can outlive a run by a week unnoticed.

NOT proven by this test, stated in its own header rather than implied: off-HOST DR (the dev
nodes are VMs on the same Proxmox host as the S3 target, so this does not survive hypervisor
loss); no direct S3 client read; real service data (Nextcloud/Gitea/CNPG drills remain
operator exercises); bare-metal rebuild DR; encryption-at-rest; retention/GC.

Serves: R-STATE-03, R-TEST-03, R-TEST-05.

Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>

conformance/lib/record_run.py

@@ -27,6 +27,17 @@ Chainsaw's JUnit carries no stdout — its `<testcase>` elements hold only `<pro
 
   *** A COMMITTED JUnit ALONE WOULD RECORD `DR_RESTORE_SKIPPED_OK` AS PROOF OF R-STATE-03. ***
 
+AND THE SECOND TRAP, WHICH IS THE MIRROR OF THE FIRST
+------------------------------------------------------
+Chainsaw echoes each script's SOURCE into an indented `=== COMMAND` block before running
+it, so `echo NODE_LOSS_SKIPPED_OK` — the skip branch that was NOT taken — is present in
+the log of every run of that test. A section-blind grep therefore reads a skip token out
+of the source of every skip-gated test and can never record `proven` for any of them: the
+ledger fails closed so completely that it records nothing and proves nothing. Tokens are
+consequently read from SECTIONS (see classify_sections): a pass token only from
+`=== STDOUT`, a skip token from any section that is not script source. Skip still beats
+pass — within genuine output, which is the only place either was ever evidence.
+
 That is the fake-green this project refuses, and it would be worse than no ledger at all
 because it would look authoritative. THE SENTINEL TOKEN IS THE ONLY THING THAT SEPARATES
 PROVEN FROM SKIPPED, AND IT EXISTS ONLY IN THE RUN LOG. Hence: this tool requires BOTH
@@ -149,10 +160,117 @@ _SENTINEL_RE = re.compile(r"contains\(\s*\$stdout\s*,\s*['\"]([A-Za-z0-9_]+)['\"
 # mis-classifying a SKIP token as a pass would manufacture one.
 _SKIP_TOKEN_RE = re.compile(r"_SKIP")
 
-# A Chainsaw log line header:  `| 20:41:32 | node-loss | step-name | TRY | RUN | ...`
-# Field 2 is the test name; it is how stdout is attributed back to a testcase.
+# A Chainsaw log line header, as it really appears (Go `t.Log` prefixes the source site and
+# pads it with backspaces, and the fields are ANSI-coloured — both are stripped first):
+#     l.go:52: | 23:51:18 | node-loss | volume-stays-available... | SCRIPT | LOG |
+# Field 2 is the test name; it is how output is attributed back to a testcase. The
+# `<file>.go:<line>:` prefix is optional so a redirected/plain log still attributes.
 _LOG_HEADER_RE = re.compile(
-    r"^\s*\|\s*\d{1,2}:\d{2}:\d{2}(?:\.\d+)?\s*\|\s*([A-Za-z0-9][A-Za-z0-9._-]*)\s*\|")
+    r"^\s*(?:[\w./-]+\.go:\d+:\s*)?"
+    r"\|\s*\d{1,2}:\d{2}:\d{2}(?:\.\d+)?\s*\|\s*([A-Za-z0-9][A-Za-z0-9._-]*)\s*\|")
+
+# ANSI colour + the backspace padding Go writes ahead of a `t.Log` continuation. Stripped
+# before ANY parsing; neither ever appears at the head of a line, so indentation survives.
+_ANSI_RE = re.compile(r"\x1b\[[0-9;]*[A-Za-z]")
+
+
+def _strip_noise(line: str) -> str:
+    return _ANSI_RE.sub("", line).replace("\b", "")
+
+
+# ===================================================================== output sections
+# THE BUG THIS SECTION MACHINERY EXISTS TO KILL — READ BEFORE CHANGING ANYTHING
+# -----------------------------------------------------------------------------
+# Chainsaw echoes the script it is ABOUT TO RUN, verbatim, in an indented `=== COMMAND`
+# block, and only afterwards prints what the script actually wrote in `=== STDOUT`. Every
+# skip-gated test contains its own skip token as SOURCE TEXT (`echo NODE_LOSS_SKIPPED_OK`),
+# so a raw whole-log grep finds that token in EVERY run of that test — including runs that
+# never took the skip branch. Combined with decide()'s (correct) rule that a skip token
+# outranks a pass token, a section-blind scan can never record `proven` for ANY test that
+# has a skip gate: the ledger fails closed so hard it records nothing. Observed on
+# runs/20260901T235117Z-e9a4b66, where NODE_LOSS_SKIPPED_OK appears twice as source at
+# lines 51/62 (`=== COMMAND`) and never in output, yet the run genuinely proved R-STATE-02.
+#
+# So tokens are attributed to SECTIONS, never to raw lines:
+#     l.go:52: | 23:51:18 | node-loss | step | CMD    | RUN |     <- header, no section
+#         === COMMAND                                            <- marker (indent 8)
+#         /usr/bin/sh -c set -euo pipefail                       <- source, indent >= 8
+#           echo NODE_LOSS_SKIPPED_OK                            <- SOURCE, not output
+#     l.go:52: | 23:52:42 | node-loss | step | SCRIPT | LOG |    <- indent 4 => section ends
+#         === STDOUT                                             <- marker
+#         NODE_LOSS_VOLUME_AVAILABLE_OK                          <- the REAL emission
+#
+# A marker is an INDENTED `=== NAME` on a line of its own. Go's own test markers
+# (`=== RUN   chainsaw`, `=== PAUSE`, `=== CONT`) are at column 0 and carry a test name
+# after the verb, so they match neither the indent nor the end-of-line anchor; they are
+# additionally named in _GO_TEST_MARKERS so a future indented variant still cannot open a
+# section that would swallow the whole log.
+_SECTION_MARKER_RE = re.compile(r"^(\s+)=== ([A-Z][A-Z0-9_]*)\s*$")
+_GO_TEST_MARKERS = frozenset({"RUN", "PAUSE", "CONT"})
+
+# Pseudo-section for a log that contains NO section markers at all (the plain/redirected
+# logs the earlier ledger rows were written from, and the selftest fixtures). Such a log
+# never echoes script source, so there is nothing to be confused by and the legacy
+# whole-log scan is exactly as safe as it was — narrowing is only needed where a COMMAND
+# echo exists. Lines OUTSIDE a section in a log that DOES use sections are attributed to
+# "" and are evidence for nothing.
+UNSECTIONED = "*"
+
+# A pass token counts ONLY where Chainsaw's own `contains($stdout, ...)` check looks:
+# real stdout. Anything else — source echo, stderr, an unrecognised future section — can
+# never manufacture a proof.
+PASS_SECTIONS = frozenset({"STDOUT", UNSECTIONED})
+
+# Sections that carry SOURCE rather than output. A skip token is honoured in every OTHER
+# section (STDOUT, STDERR, any section name added by a later Chainsaw release), because
+# over-reading a skip only ever COSTS a proof, while over-reading a pass would fabricate
+# one. Unknown section names therefore fail closed in both directions.
+SOURCE_SECTIONS = frozenset({"COMMAND"})
+
+
+def classify_sections(log_text: str) -> list[tuple[str, str]]:
+    """[(section, line)] for every line of a run log.
+
+    `section` is the enclosing `=== NAME` block, "" for a line inside no block, or
+    UNSECTIONED for every line when the log declares no sections at all. A block runs from
+    its marker until the first non-blank line indented LESS than the marker (blank lines do
+    not end it) or until the next marker; the marker line itself is content of nothing.
+    """
+    rows: list[tuple[str, str]] = []
+    cur, cur_indent, sectioned = "", 0, False
+    for raw in log_text.splitlines():
+        line = _strip_noise(raw)
+        m = _SECTION_MARKER_RE.match(line)
+        if m and m.group(2) not in _GO_TEST_MARKERS:
+            sectioned = True
+            cur, cur_indent = m.group(2), len(m.group(1))
+            rows.append(("", line))
+            continue
+        if cur:
+            if not line.strip() or len(line) - len(line.lstrip()) >= cur_indent:
+                rows.append((cur, line))
+                continue
+            cur = ""
+        rows.append(("", line))
+    if not sectioned:
+        return [(UNSECTIONED, line) for _, line in rows]
+    return rows
+
+
+def scan_texts(lines) -> tuple[str, str]:
+    """(text a PASS token may be read from, text a SKIP token may be read from).
+
+    Accepts either [(section, line)] pairs from classify_sections() or a plain list of
+    strings, which is treated as an unsectioned log (legacy whole-log scanning).
+    """
+    pass_lines, skip_lines = [], []
+    for item in lines:
+        section, line = item if isinstance(item, tuple) else (UNSECTIONED, item)
+        if section in PASS_SECTIONS:
+            pass_lines.append(line)
+        if section and section not in SOURCE_SECTIONS:
+            skip_lines.append(line)
+    return "\n".join(pass_lines), "\n".join(skip_lines)
 
 _RUN_ID_TS_RE = re.compile(r"^(\d{8}T\d{6}Z)-")
 
@@ -203,28 +321,32 @@ def sentinel_map(tests_dir: Path) -> dict[str, dict]:
 
 
 # =============================================================================== log
-def attribute_log(log_text: str, known: set[str]) -> tuple[dict[str, list[str]], bool]:
-    """Split a run log into per-testcase line buckets.
+def attribute_log(log_text: str, known: set[str]) -> tuple[dict[str, list[tuple[str, str]]],
+                                                            bool]:
+    """Split a run log into per-testcase buckets of (section, line) pairs.
 
     Chainsaw prefixes its log lines with `| HH:MM:SS | <test> | <step> | ...` and prints a
-    script's stdout underneath that header, so the test name is carried by a state machine:
-    a header naming a KNOWN test switches the current bucket, every following line lands in
-    it. Requiring the name to be known is what stops run.sh's own `[run.sh]` chatter (and
-    any other pipe-shaped line) from opening a bogus bucket.
+    script's sections underneath that header, so the test name is carried by a state
+    machine: a header naming a KNOWN test switches the current bucket, every following line
+    lands in it. Requiring the name to be known is what stops run.sh's own `[run.sh]`
+    chatter (and any other pipe-shaped line) from opening a bogus bucket.
+
+    Section labels come from classify_sections() and travel WITH the lines, so a bucket
+    still knows which of its lines were script source and which were real output.
 
     Returns (buckets, attributed) where `attributed` is False when the log contained no
     recognisable header at all — the signal for the degraded global scan in decide().
     """
-    buckets: dict[str, list[str]] = {}
+    buckets: dict[str, list[tuple[str, str]]] = {}
     cur: str | None = None
     attributed = False
-    for line in log_text.splitlines():
+    for section, line in classify_sections(log_text):
         m = _LOG_HEADER_RE.match(line)
         if m and m.group(1) in known:
             cur = m.group(1)
             attributed = True
         if cur is not None:
-            buckets.setdefault(cur, []).append(line)
+            buckets.setdefault(cur, []).append((section, line))
     return buckets, attributed
 
 
@@ -253,7 +375,7 @@ def decide(layer: str, junit_result: str, sent: dict, lines: list[str],
       1. JUnit failure/error            -> failed        (a red run is recorded as red)
       2. JUnit <skipped/>               -> junit-skipped  (never green; ci/lib.sh rule)
       3. ANY declared skip token seen   -> skipped        (BEATS a green JUnit and beats
-                                                           the pass token: if any gated
+           IN REAL OUTPUT                                  the pass token: if any gated
                                                            step self-skipped, the run did
                                                            not prove the requirement)
       4. no sentinel declared           -> proven only for L0/L1, whose skips are already
@@ -264,6 +386,13 @@ def decide(layer: str, junit_result: str, sent: dict, lines: list[str],
     Step 5 requires ALL pass tokens, not just the terminal one: every Chainsaw step here
     asserts `pass || skip`, so a missing pass token means a step did not reach its proven
     branch, whatever the JUnit says.
+
+    Steps 3 and 5 read from SECTION-SCOPED text, never from raw lines: a pass token counts
+    only in `=== STDOUT`, a skip token counts in any section that is not script source.
+    That narrows WHERE a token is read from and changes nothing about what it means — skip
+    still outranks pass, and it outranks it within genuine output, which is the only place
+    either of them was ever evidence. `lines` may be [(section, line)] pairs or plain
+    strings (an unsectioned log; see scan_texts / UNSECTIONED).
     """
     if junit_result in ("failed", "error"):
         return (None, "failed", f"JUnit recorded {junit_result}")
@@ -271,12 +400,12 @@ def decide(layer: str, junit_result: str, sent: dict, lines: list[str],
         return (None, "junit-skipped",
                 "JUnit <skipped/> — the test did not execute; a skip is never green")
 
-    text = "\n".join(lines)
-    seen_skip = [t for t in sent["skip"] if t in text]
+    pass_text, skip_text = scan_texts(lines)
+    seen_skip = [t for t in sent["skip"] if t in skip_text]
     if seen_skip:
         return (seen_skip[0], "skipped",
-                f"run log carries the SELF-SKIP token {seen_skip[0]!r} — the test exited 0 "
-                f"without exercising the behaviour, so this is NOT proof")
+                f"run log OUTPUT carries the SELF-SKIP token {seen_skip[0]!r} — the test "
+                f"exited 0 without exercising the behaviour, so this is NOT proof")
 
     if not sent["pass"] and not sent["skip"]:
         if layer not in SENTINEL_EXEMPT_LAYERS:
@@ -288,14 +417,14 @@ def decide(layer: str, junit_result: str, sent: dict, lines: list[str],
                 f"{layer} offline check exited 0; run_l1_offline.py renders its skips as "
                 f"JUnit <skipped/>, so a green JUnit is truthful without a sentinel")
 
-    missing = [t for t in sent["pass"] if t not in text]
+    missing = [t for t in sent["pass"] if t not in pass_text]
     if not missing:
         return (sent["terminal_pass"], "proven",
-                f"run log carries every declared pass token "
+                f"run log OUTPUT carries every declared pass token "
                 f"({', '.join(sent['pass'])}) and no skip token [{scope}]")
     return (None, "unknown-sentinel-absent",
-            f"run log lacks pass token(s) {', '.join(missing)} for this test [{scope}] — "
-            f"refusing to infer proof from the JUnit alone")
+            f"run log OUTPUT lacks pass token(s) {', '.join(missing)} for this test "
+            f"[{scope}] — refusing to infer proof from the JUnit alone")
 
 
 # ============================================================================= junit
@@ -387,9 +516,13 @@ def build_rows(run_dir: Path, tests_dir: Path, env: str, chainsaw_version: str,
     junit = run_dir / "report.junit.xml"
     log = run_dir / "run.log"
     smap = sentinel_map(tests_dir)
-    buckets, attributed = attribute_log(log.read_text(errors="replace"), set(smap))
+    log_text = log.read_text(errors="replace")
+    buckets, attributed = attribute_log(log_text, set(smap))
     uniq = unique_tokens(smap)
-    all_lines = log.read_text(errors="replace").splitlines()
+    all_pairs = classify_sections(log_text)
+    sectioned = any(sec not in ("", UNSECTIONED) for sec, _ in all_pairs)
+    section_note = ("tokens read from real-output sections only"
+                    if sectioned else "log declares no === sections; whole-log scan")
     ts = run_timestamp(run_dir, junit)
 
     rows: list[dict] = []
@@ -403,17 +536,18 @@ def build_rows(run_dir: Path, tests_dir: Path, env: str, chainsaw_version: str,
         reqs = sent.get("requirements") or case["requirements"]
 
         if name in buckets:
-            lines, scope = buckets[name], "log scoped to this testcase"
+            lines, scope = buckets[name], f"log scoped to this testcase; {section_note}"
         elif not attributed:
             # Degraded mode: the log had no recognisable Chainsaw header (a redirected or
             # reformatted log). Fall back to a whole-log scan, but ONLY for tokens owned by
             # exactly one test, so a token another test emitted can never prove this one.
-            lines = all_lines
+            lines = all_pairs
             sent = dict(sent,
                         **{"pass": [t for t in sent["pass"] if t in uniq],
                            "skip": [t for t in sent["skip"] if t in uniq]})
             sent["terminal_pass"] = sent["pass"][-1] if sent["pass"] else None
-            scope = "DEGRADED: unattributable log, globally-unique tokens only"
+            scope = ("DEGRADED: unattributable log, globally-unique tokens only; "
+                     + section_note)
         else:
             lines, scope = [], "no log lines attributed to this testcase"
 
@@ -552,6 +686,98 @@ def _selftest() -> int:
     check("an unknown/blank layer fails CLOSED — sentinel required",
           proof == "unknown-no-sentinel-declared")
 
+    # --- SECTION SCOPING: the COMMAND echo is SOURCE, never evidence -------------------
+    # Regression cases for the ledger's first real L3 run, which recorded a genuine PASS as
+    # `skipped` because `echo NODE_LOSS_SKIPPED_OK` — the skip branch NOT taken — is echoed
+    # inside `=== COMMAND` on every run of a skip-gated test.
+    nl = {"pass": ["NODE_LOSS_VOLUME_AVAILABLE_OK"],
+          "skip": ["NODE_LOSS_SKIPPED_OK"],
+          "terminal_pass": "NODE_LOSS_VOLUME_AVAILABLE_OK"}
+    _CMD = ("    l.go:52: | 23:51:18 | node-loss | step | CMD    | RUN |\n"
+            "        === COMMAND\n"
+            "        /usr/bin/sh -c set -euo pipefail\n"
+            "        if ! kubectl get storageclass longhorn >/dev/null 2>&1; then\n"
+            "          echo NODE_LOSS_SKIPPED_OK\n"
+            "\n"
+            "          exit 0\n"
+            "        fi\n"
+            "        echo NODE_LOSS_VOLUME_AVAILABLE_OK\n"
+            "    l.go:52: | 23:52:42 | node-loss | step | SCRIPT | LOG |\n")
+
+    ran = classify_sections(_CMD + "        === STDOUT\n"
+                            "        ready nodes: 3\n"
+                            "        NODE_LOSS_VOLUME_AVAILABLE_OK\n"
+                            "    l.go:52: | 23:52:42 | node-loss | step | SCRIPT | DONE |\n")
+    sent, proof, _ = decide("L3", "passed", nl, ran, "t")
+    check("skip token ONLY in the COMMAND echo + pass token in STDOUT is PROVEN",
+          proof == "proven" and sent == "NODE_LOSS_VOLUME_AVAILABLE_OK")
+
+    selfskipped = classify_sections(_CMD + "        === STDOUT\n"
+                                    "        SKIP: 'longhorn' StorageClass absent\n"
+                                    "        NODE_LOSS_SKIPPED_OK\n")
+    sent, proof, _ = decide("L3", "passed", nl, selfskipped, "t")
+    check("skip token GENUINELY in STDOUT is still 'skipped' (the guarantee holds)",
+          proof == "skipped" and sent == "NODE_LOSS_SKIPPED_OK")
+
+    neither = classify_sections(_CMD + "        === STDOUT\n        ready nodes: 3\n")
+    _, proof, _ = decide("L3", "passed", nl, neither, "t")
+    check("both tokens in COMMAND and neither in STDOUT is NOT proven",
+          proof == "unknown-sentinel-absent")
+
+    both_out = classify_sections("        === STDOUT\n"
+                                 "        NODE_LOSS_VOLUME_AVAILABLE_OK\n"
+                                 "        NODE_LOSS_SKIPPED_OK\n")
+    _, proof, _ = decide("L3", "passed", nl, both_out, "t")
+    check("skip still BEATS pass when both are in genuine output", proof == "skipped")
+
+    stderr_skip = classify_sections("        === STDERR\n        NODE_LOSS_SKIPPED_OK\n"
+                                    "        === STDOUT\n"
+                                    "        NODE_LOSS_VOLUME_AVAILABLE_OK\n")
+    _, proof, _ = decide("L3", "passed", nl, stderr_skip, "t")
+    check("a skip token in a non-source section (STDERR) still costs the proof",
+          proof == "skipped")
+
+    stderr_pass = classify_sections("        === STDERR\n"
+                                    "        NODE_LOSS_VOLUME_AVAILABLE_OK\n")
+    _, proof, _ = decide("L3", "passed", nl, stderr_pass, "t")
+    check("a pass token OUTSIDE stdout cannot prove anything",
+          proof == "unknown-sentinel-absent")
+
+    # --- classify_sections: markers, extents, and Go's own `=== RUN` --------------------
+    gorows = classify_sections("=== RUN   chainsaw\n=== PAUSE chainsaw\n"
+                               "=== CONT  chainsaw/node-loss\n"
+                               "        === COMMAND\n        echo NODE_LOSS_SKIPPED_OK\n"
+                               "    l.go:52: | 1:00:00 | node-loss | s | SCRIPT | DONE |\n"
+                               "--- PASS: chainsaw/node-loss (131.09s)\n")
+    secs = [sec for sec, _ in gorows]
+    check("Go's column-0 `=== RUN/PAUSE/CONT` never open a section",
+          secs[:3] == ["", "", ""])
+    check("a section covers its indented body and ends at a less-indented line",
+          secs[3] == "" and secs[4] == "COMMAND" and secs[5] == "" and secs[6] == "")
+    check("a log with no === markers stays wholly scannable (UNSECTIONED)",
+          {sec for sec, _ in classify_sections("plain line\n  another\n")} == {UNSECTIONED})
+
+    # --- the preserved run itself, as a fixture ----------------------------------------
+    # runs/20260901T235117Z-e9a4b66 IS the bug: NODE_LOSS_SKIPPED_OK twice in `=== COMMAND`
+    # (source), never in output; NODE_LOSS_VOLUME_AVAILABLE_OK once in `=== STDOUT`.
+    fixture = CONF_ROOT / "runs" / "20260901T235117Z-e9a4b66"
+    if (fixture / "run.log").is_file() and (fixture / "report.junit.xml").is_file():
+        rows = build_rows(fixture, CONF_ROOT / "tests", "dev", "v0.2.12", "v1.36.4", "sha")
+        row = next((r for r in rows if r["test"] == "node-loss"), None)
+        check("preserved real L3 run records node-loss/R-STATE-02 as proven",
+              row is not None and row["proof"] == "proven"
+              and row["sentinel"] == "NODE_LOSS_VOLUME_AVAILABLE_OK"
+              and row["requirements"] == ["R-STATE-02"])
+        check("preserved real L3 run attributes (no DEGRADED fallback needed)",
+              row is not None and "DEGRADED" not in row["note"])
+        pairs = classify_sections((fixture / "run.log").read_text(errors="replace"))
+        secs = {sec for sec, line in pairs if "NODE_LOSS_SKIPPED_OK" in line}
+        check("every NODE_LOSS_SKIPPED_OK in the preserved log is COMMAND source",
+              secs == {"COMMAND"})
+        secs = {sec for sec, line in pairs
+                if line.strip() == "NODE_LOSS_VOLUME_AVAILABLE_OK"}
+        check("the real emission in the preserved log is in STDOUT", secs == {"STDOUT"})
+
     # --- log attribution --------------------------------------------------------------
     log = ("[run.sh] Coverage report first\n"
            "    | 20:41:32 | node-loss | step | SCRIPT | RUN |\n"
@@ -561,10 +787,22 @@ def _selftest() -> int:
     b, attributed = attribute_log(log, {"node-loss", "dr-restore"})
     check("log lines are attributed to the testcase whose header preceded them",
           attributed
-          and any("NODE_LOSS_SKIPPED_OK" in l for l in b["node-loss"])
-          and not any("NODE_LOSS_SKIPPED_OK" in l for l in b["dr-restore"]))
+          and any("NODE_LOSS_SKIPPED_OK" in l for _, l in b["node-loss"])
+          and not any("NODE_LOSS_SKIPPED_OK" in l for _, l in b["dr-restore"]))
     check("a pass token does NOT leak across testcase boundaries",
-          not any("DR_RESTORED" in l for l in b["node-loss"]))
+          not any("DR_RESTORED" in l for _, l in b["node-loss"]))
+    check("an unsectioned log keeps whole-log scanning (nothing narrowed away)",
+          all(sec == UNSECTIONED for sec, _ in b["node-loss"]))
+
+    # Chainsaw's real headers carry a Go `t.Log` source prefix and ANSI colour; the earlier
+    # regex matched neither, so every genuine Chainsaw log silently fell into DEGRADED mode.
+    real_hdr = ("    l.go:52: \x08\x08\x08| 23:51:18 | \x1b[32;1mnode-loss\x1b[0;22m | "
+                "\x1b[32;1mstep\x1b[0;22m | SCRIPT | LOG |\n"
+                "        === STDOUT\n        NODE_LOSS_VOLUME_AVAILABLE_OK\n")
+    b2, attributed_real = attribute_log(real_hdr, {"node-loss"})
+    check("a REAL Chainsaw header (go source prefix + ANSI) attributes, not degrades",
+          attributed_real and "node-loss" in b2
+          and ("STDOUT", "        NODE_LOSS_VOLUME_AVAILABLE_OK") in b2["node-loss"])
     _, attributed2 = attribute_log("no chainsaw headers at all\n", {"node-loss"})
     check("an unattributable log is reported as such (degraded mode)", not attributed2)
 

conformance/lib/status.py

@@ -108,7 +108,14 @@ FLUX_API = "kustomize.toolkit.fluxcd.io"
 #   conformance/tests — those are the TESTS; they are the VERIFICATION axis, counted there.
 #   ci/tests          — gate unit-test fixtures embed example R-IDs as DATA.
 #   registry.yaml     — the requirement mirror itself, not an implementation of anything.
-IMPL_EXCLUDE_PREFIXES = ("conformance/tests/", "ci/tests/")
+# `conformance/runs/` is excluded for a subtler reason than the test dirs: the ledger and the
+# preserved run logs CONTAIN the requirement IDs they are evidence FOR. Counting them as
+# implementing evidence makes `last_change` move every time proof is recorded, so a row's
+# git_sha always predates it and EVERY proof degrades to stale-proof the instant it is
+# committed — recording the proof destroys it. The ledger is the proof record; it must not
+# participate in the freshness computation about itself. (Found live 2026-09-01 on the first
+# PROVEN row.)
+IMPL_EXCLUDE_PREFIXES = ("conformance/tests/", "ci/tests/", "conformance/runs/")
 IMPL_EXCLUDE_PATHS = ("conformance/requirements/registry.yaml",)
 
 

conformance/runs/20260901T235117Z-e9a4b66/report.junit.xml

@@ -0,0 +1,4 @@
+<testsuites name="chainsaw-report" tests="1" failures="0" errors="0" time="0.000000"><testsuite name="/work/conformance/tests/l3-fault-injection/node-loss" tests="1" failures="0" errors="0" id="0" time="">
+    <testcase name="node-loss" classname="" time="131.090121" />
+  </testsuite>
+</testsuites>

conformance/runs/20260901T235117Z-e9a4b66/run.log

@@ -0,0 +1,232 @@
+[run.sh] Coverage report first (always)
+[coverage] 42/51 requirements have a scaffolded test (82.4%); 9 gap(s), 0 safety-critical gap(s).
+[coverage] wrote /work/conformance/_artifacts/coverage/coverage.md and /work/conformance/_artifacts/coverage/coverage.json
+[run.sh] No L0/L1 offline checks match the selection (layer=l3 reqs='R-STATE-02').
+[run.sh] LIGHT pass: no tests selected.
+[run.sh] HEAVY pass: 1 test(s) SERIAL (parallel=1) — node-loss
+Version: 0.2.12
+Loading config (/work/conformance/config/chainsaw-config.yaml)...
+- Using test file: chainsaw-test
+- TestDirs [/work/conformance/tests/l3-fault-injection/node-loss]
+- SkipDelete false
+- FailFast false
+- ReportFormat 'JUNIT-TEST'
+- ReportName 'chainsaw-report-heavy'
+- ReportPath '/work/conformance/_artifacts'
+- Namespace ''
+- FullName false
+- IncludeTestRegex ''
+- ExcludeTestRegex ''
+- ApplyTimeout 1m30s
+- AssertTimeout 3m0s
+- CleanupTimeout 2m0s
+- DeleteTimeout 2m0s
+- ErrorTimeout 1m0s
+- ExecTimeout 3m0s
+- DeletionPropagationPolicy Background
+- Parallel 1
+- Template true
+- NoCluster false
+- PauseOnFailure false
+Loading tests...
+- node-loss (/work/conformance/tests/l3-fault-injection/node-loss)
+Loading values...
+Running tests...
+=== RUN   chainsaw
+=== PAUSE chainsaw
+=== CONT  chainsaw
+=== RUN   chainsaw/node-loss
+=== PAUSE chainsaw/node-loss
+=== CONT  chainsaw/node-loss
+    l.go:52: | 23:51:18 | node-loss | @chainsaw                               | CREATE    | OK    | v1/Namespace @ chainsaw-fond-marlin
+    l.go:52: | 23:51:18 | node-loss | volume-stays-available-across-node-loss | TRY       | BEGIN |
+    l.go:52: | 23:51:18 | node-loss | volume-stays-available-across-node-loss | CMD       | RUN   |
+        === COMMAND
+        /usr/bin/sh -c set -euo pipefail
+        NS="$NAMESPACE"
+        
+        # -- Gate 1: skip cleanly (green) if the longhorn tier is not present yet. --
+        if ! kubectl get storageclass longhorn >/dev/null 2>&1; then
+          echo "SKIP: 'longhorn' StorageClass absent — node-loss not applicable on this cluster yet"
+          echo NODE_LOSS_SKIPPED_OK
+          exit 0
+        fi
+        
+        # -- Gate 2: a 3-replica volume + single-node-loss headroom needs >= 3 Ready nodes. --
+        READY="$(kubectl get nodes \
+          -o jsonpath='{range .items[*]}{.status.conditions[?(@.type=="Ready")].status}{"\n"}{end}' \
+          | grep -c '^True$' || true)"
+        echo "ready nodes: ${READY}"
+        if [ "${READY:-0}" -lt 3 ]; then
+          echo "SKIP: only ${READY:-0} Ready node(s) (<3) — node-loss resilience not meaningful yet"
+          echo NODE_LOSS_SKIPPED_OK
+          exit 0
+        fi
+        
+        CANARY="node-loss-canary-$(date +%s%N)"
+        
+        # -- GIVEN: a Longhorn PVC (platform default SC = 3 replicas across nodes) + a
+        #    writer pod that writes the canary and flushes it to the replicated volume. --
+        kubectl -n "$NS" apply -f - <<EOF
+        apiVersion: v1
+        kind: PersistentVolumeClaim
+        metadata:
+          name: survivor
+        spec:
+          accessModes: [ReadWriteOnce]
+          storageClassName: longhorn
+          resources:
+            requests:
+              storage: 1Gi
+        EOF
+        
+        apply_pod() {
+          kubectl -n "$NS" apply -f - <<EOF
+        apiVersion: v1
+        kind: Pod
+        metadata:
+          name: $1
+        spec:
+          restartPolicy: Never
+          containers:
+            - name: app
+              image: busybox:1.36.1
+              command: ["sh", "-c", "sleep 3600"]
+              volumeMounts:
+                - name: data
+                  mountPath: /data
+          volumes:
+            - name: data
+              persistentVolumeClaim:
+                claimName: survivor
+        EOF
+        }
+        
+        apply_pod writer
+        kubectl -n "$NS" wait --for=condition=Ready pod/writer --timeout=240s
+        printf '%s' "$CANARY" | kubectl -n "$NS" exec -i writer -- sh -c 'cat > /data/canary && sync'
+        echo "wrote canary: $(kubectl -n "$NS" exec writer -- cat /data/canary)"
+        
+        # The victim node is the one the attached volume currently lives on.
+        VICTIM="$(kubectl -n "$NS" get pod writer -o jsonpath='{.spec.nodeName}')"
+        [ -n "$VICTIM" ] || { echo "FAIL: could not determine writer's node" >&2; exit 1; }
+        echo "victim node (currently hosts the attached volume): $VICTIM"
+        
+        # Best-effort precondition log: if the Longhorn Volume CR is readable, show the
+        # volume is genuinely 3-replica/healthy before we take a node. Non-fatal (the
+        # runner is least-privilege and may not have longhorn.io read rights) — the
+        # data-availability assertion below is the authoritative proof, not this.
+        PVNAME="$(kubectl -n "$NS" get pvc survivor -o jsonpath='{.spec.volumeName}' 2>/dev/null || true)"
+        if [ -n "$PVNAME" ]; then
+          kubectl -n longhorn-system get volumes.longhorn.io "$PVNAME" \
+            -o jsonpath='pre-fault robustness={.status.robustness} replicas={.spec.numberOfReplicas}{"\n"}' \
+            2>/dev/null || echo "(Longhorn Volume CR not readable — continuing; data read is the proof)"
+        fi
+        
+        # -- WHEN: FAULT INJECTION — remove the victim node FROM SERVICE. --
+        #    Label it with THIS test's unique namespace so `finally` can always find and
+        #    uncordon exactly this node; then cordon it; then evict ONLY our own writer
+        #    pod, forcing the volume to detach from the victim. (No `drain` — unrelated
+        #    pods on the node are never touched.)
+        kubectl label node "$VICTIM" "conformance.home.k3s/cordoned-by=$NS" --overwrite
+        kubectl cordon "$VICTIM"
+        echo "cordoned + labelled victim node: $VICTIM"
+        kubectl -n "$NS" delete pod writer --wait=true --timeout=240s
+        echo "evicted writer off the victim node; volume detached from $VICTIM"
+        
+        # -- THEN: a fresh workload re-binds the SAME volume. The victim is unschedulable,
+        #    so the scheduler MUST place the reader on a SURVIVING node; Longhorn re-attaches
+        #    the volume there from its replicas on the other nodes. --
+        apply_pod reader
+        kubectl -n "$NS" wait --for=condition=Ready pod/reader --timeout=360s
+        READER_NODE="$(kubectl -n "$NS" get pod reader -o jsonpath='{.spec.nodeName}')"
+        echo "reader scheduled on surviving node: $READER_NODE (victim was $VICTIM)"
+        if [ "$READER_NODE" = "$VICTIM" ]; then
+          echo "FAIL: reader landed back on the cordoned victim node — node was not removed from service" >&2
+          exit 1
+        fi
+        
+        GOT="$(kubectl -n "$NS" exec reader -- cat /data/canary)"
+        echo "recovered='$GOT' expected='$CANARY'"
+        [ "$GOT" = "$CANARY" ] || { echo "FAIL: data not available after node loss" >&2; exit 1; }
+        echo "volume available + data intact from a surviving node"
+        
+        # Best-effort reconvergence log: if the Longhorn Volume CR is readable, wait
+        # (bounded) for robustness to settle back to available (healthy|degraded — both
+        # serve I/O; a healthy 3-replica volume rebuilds a replica once headroom exists).
+        # Non-fatal for the same least-privilege reason as the pre-fault log.
+        if [ -n "$PVNAME" ]; then
+          for i in $(seq 1 30); do
+            ROB="$(kubectl -n longhorn-system get volumes.longhorn.io "$PVNAME" \
+                  -o jsonpath='{.status.robustness}' 2>/dev/null || true)"
+            [ -z "$ROB" ] && { echo "(Longhorn Volume CR not readable — skipping reconvergence log)"; break; }
+            echo "post-fault robustness[$i]=$ROB"
+            case "$ROB" in healthy|degraded) break ;; esac
+            sleep 10
+          done
+        fi
+        
+        echo NODE_LOSS_VOLUME_AVAILABLE_OK
+    l.go:52: | 23:52:42 | node-loss | volume-stays-available-across-node-loss | SCRIPT    | LOG   |
+        === STDOUT
+        ready nodes: 3
+        persistentvolumeclaim/survivor created
+        pod/writer created
+        pod/writer condition met
+        wrote canary: node-loss-canary-1788306678898809444
+        victim node (currently hosts the attached volume): k3s-dev-2
+        pre-fault robustness=healthy replicas=3
+        node/k3s-dev-2 labeled
+        node/k3s-dev-2 cordoned
+        cordoned + labelled victim node: k3s-dev-2
+        pod "writer" deleted from chainsaw-fond-marlin namespace
+        evicted writer off the victim node; volume detached from k3s-dev-2
+        pod/reader created
+        pod/reader condition met
+        reader scheduled on surviving node: k3s-dev-1 (victim was k3s-dev-2)
+        recovered='node-loss-canary-1788306678898809444' expected='node-loss-canary-1788306678898809444'
+        volume available + data intact from a surviving node
+        post-fault robustness[1]=healthy
+        NODE_LOSS_VOLUME_AVAILABLE_OK
+    l.go:52: | 23:52:42 | node-loss | volume-stays-available-across-node-loss | SCRIPT    | DONE  |
+    l.go:52: | 23:52:42 | node-loss | volume-stays-available-across-node-loss | TRY       | END   |
+    l.go:52: | 23:52:42 | node-loss | volume-stays-available-across-node-loss | FINALLY   | BEGIN |
+    l.go:52: | 23:52:42 | node-loss | volume-stays-available-across-node-loss | CMD       | RUN   |
+        === COMMAND
+        /usr/bin/sh -c set -uo pipefail
+        for n in $(kubectl get nodes -l "conformance.home.k3s/cordoned-by=$NAMESPACE" -o name 2>/dev/null); do
+          kubectl uncordon "${n#node/}" || true
+          kubectl label "$n" "conformance.home.k3s/cordoned-by-" || true
+          echo "restored node to service: ${n#node/}"
+        done
+    l.go:52: | 23:52:43 | node-loss | volume-stays-available-across-node-loss | SCRIPT    | LOG   |
+        === STDOUT
+        node/k3s-dev-2 uncordoned
+        node/k3s-dev-2 unlabeled
+        restored node to service: k3s-dev-2
+    l.go:52: | 23:52:43 | node-loss | volume-stays-available-across-node-loss | SCRIPT    | DONE  |
+    l.go:52: | 23:52:43 | node-loss | volume-stays-available-across-node-loss | CMD       | RUN   |
+        === COMMAND
+        /usr/bin/sh -c kubectl -n "$NAMESPACE" delete pod writer reader --ignore-not-found --wait=false || true
+        kubectl -n "$NAMESPACE" delete pvc survivor --ignore-not-found --wait=false || true
+        echo "cleanup best-effort; chainsaw namespace teardown finalizes"
+    l.go:52: | 23:52:43 | node-loss | volume-stays-available-across-node-loss | SCRIPT    | LOG   |
+        === STDOUT
+        pod "reader" deleted from chainsaw-fond-marlin namespace
+        persistentvolumeclaim "survivor" deleted from chainsaw-fond-marlin namespace
+        cleanup best-effort; chainsaw namespace teardown finalizes
+    l.go:52: | 23:52:43 | node-loss | volume-stays-available-across-node-loss | SCRIPT    | DONE  |
+    l.go:52: | 23:52:43 | node-loss | volume-stays-available-across-node-loss | FINALLY   | END   |
+    l.go:52: | 23:52:43 | node-loss | @chainsaw                               | CLEANUP   | BEGIN |
+    l.go:52: | 23:52:43 | node-loss | @chainsaw                               | DELETE    | OK    | v1/Namespace @ chainsaw-fond-marlin
+    l.go:52: | 23:53:29 | node-loss | @chainsaw                               | CLEANUP   | END   |
+--- PASS: chainsaw (0.00s)
+    --- PASS: chainsaw/node-loss (131.09s)
+PASS
+Tests Summary...
+- Passed  tests 1
+- Failed  tests 0
+- Skipped tests 0
+Done.
+merge_junit: merged 1 report(s) -> /work/conformance/_artifacts/report.junit.xml (tests=1 failures=0 errors=0 skipped=0)
+[run.sh] Done. JUnit: /work/conformance/_artifacts/report.junit.xml | Coverage: /work/conformance/_artifacts/coverage/coverage.md

conformance/runs/20260902T001907Z-fdc3265/report.junit.xml

@@ -0,0 +1,4 @@
+<testsuites name="chainsaw-report" tests="1" failures="0" errors="0" time="0.000000"><testsuite name="/work/conformance/tests/l3-fault-injection/dr-restore" tests="1" failures="0" errors="0" id="0" time="">
+    <testcase name="dr-restore" classname="" time="149.053701" />
+  </testsuite>
+</testsuites>

conformance/runs/20260902T001907Z-fdc3265/run.log

@@ -0,0 +1,671 @@
+[run.sh] Coverage report first (always)
+[coverage] 42/51 requirements have a scaffolded test (82.4%); 9 gap(s), 0 safety-critical gap(s).
+[coverage] wrote /work/conformance/_artifacts/coverage/coverage.md and /work/conformance/_artifacts/coverage/coverage.json
+[run.sh] No L0/L1 offline checks match the selection (layer=l3 reqs='R-STATE-03').
+[run.sh] LIGHT pass: no tests selected.
+[run.sh] HEAVY pass: 1 test(s) SERIAL (parallel=1) — dr-restore
+Version: 0.2.12
+Loading config (/work/conformance/config/chainsaw-config.yaml)...
+- Using test file: chainsaw-test
+- TestDirs [/work/conformance/tests/l3-fault-injection/dr-restore]
+- SkipDelete false
+- FailFast false
+- ReportFormat 'JUNIT-TEST'
+- ReportName 'chainsaw-report-heavy'
+- ReportPath '/work/conformance/_artifacts'
+- Namespace ''
+- FullName false
+- IncludeTestRegex ''
+- ExcludeTestRegex ''
+- ApplyTimeout 1m30s
+- AssertTimeout 3m0s
+- CleanupTimeout 2m0s
+- DeleteTimeout 2m0s
+- ErrorTimeout 1m0s
+- ExecTimeout 3m0s
+- DeletionPropagationPolicy Background
+- Parallel 1
+- Template true
+- NoCluster false
+- PauseOnFailure false
+Loading tests...
+- dr-restore (/work/conformance/tests/l3-fault-injection/dr-restore)
+Loading values...
+Running tests...
+=== RUN   chainsaw
+=== PAUSE chainsaw
+=== CONT  chainsaw
+=== RUN   chainsaw/dr-restore
+=== PAUSE chainsaw/dr-restore
+=== CONT  chainsaw/dr-restore
+    l.go:52: | 00:19:07 | dr-restore | @chainsaw                                 | CREATE    | OK    | v1/Namespace @ chainsaw-cunning-hen
+    l.go:52: | 00:19:07 | dr-restore | assert-backup-target-is-off-cluster       | TRY       | BEGIN |
+    l.go:52: | 00:19:07 | dr-restore | assert-backup-target-is-off-cluster       | CMD       | RUN   |
+        === COMMAND
+        /usr/bin/sh -c set -eu
+        
+        # ---- helpers -------------------------------------------------------
+        # Classify a kubectl failure: an RBAC denial is a HARNESS DEFECT and must
+        # fail loudly; an absent API is a genuine precondition gap and may skip.
+        die_or_skip() {
+          case "$2" in
+            *orbidden*|*Unauthorized*|*cannot\ list*|*cannot\ get*)
+              echo "FAIL: conformance runner lacks RBAC for $1 -- fix conformance/incluster/rbac.yaml" >&2
+              echo "$2" >&2
+              exit 1 ;;
+            *)
+              echo "SKIP: $1 not available on this cluster -- $2"
+              echo DR_RESTORE_SKIPPED_OK
+              exit 0 ;;
+          esac
+        }
+        
+        ip2int() {
+          _q="$(echo "$1" | tr '.' ' ')"
+          set -- $_q
+          [ $# -eq 4 ] || { echo -1; return 0; }
+          echo $(( ($1 * 16777216) + ($2 * 65536) + ($3 * 256) + $4 ))
+        }
+        
+        # in_cidr <ip> <a.b.c.d/len> -> 0 when the ip is inside the prefix
+        in_cidr() {
+          case "$2" in *"/"*) ;; *) return 1 ;; esac
+          _net="${2%/*}"; _len="${2#*/}"
+          case "$_len" in ''|*[!0-9]*) return 1 ;; esac
+          [ "$_len" -ge 0 ] && [ "$_len" -le 32 ] || return 1
+          _i="$(ip2int "$1")"; _n="$(ip2int "$_net")"
+          [ "$_i" -ge 0 ] && [ "$_n" -ge 0 ] || return 1
+          [ "$_len" -eq 0 ] && return 0
+          _d=1; _s=0
+          while [ "$_s" -lt $((32 - _len)) ]; do _d=$((_d * 2)); _s=$((_s + 1)); done
+          [ $(( _i / _d )) -eq $(( _n / _d )) ]
+        }
+        
+        # ---- Gate 1: the replicated storage tier the DR-critical data lives on --
+        if ! kubectl get storageclass longhorn >/dev/null 2>&1; then
+          echo "SKIP: 'longhorn' StorageClass absent -- no DR-critical data tier on this cluster yet"
+          echo DR_RESTORE_SKIPPED_OK
+          exit 0
+        fi
+        
+        # ---- Gate 2: the Velero API + a configured BackupStorageLocation --------
+        BSL_LINES="$(kubectl get backupstoragelocations.velero.io -A \
+          -o jsonpath='{range .items[*]}{.metadata.namespace}{" "}{.metadata.name}{" "}{.status.phase}{" "}{.spec.provider}{" "}{.spec.default}{" "}{.spec.config.s3Url}{" "}{.spec.objectStorage.bucket}{"\n"}{end}' \
+          2>&1)" || die_or_skip "velero.io/backupstoragelocations" "$BSL_LINES"
+        if [ -z "$(echo "$BSL_LINES" | tr -d '[:space:]')" ]; then
+          echo "SKIP: Velero API present but NO BackupStorageLocation is configured -- no off-cluster backup target yet"
+          echo DR_RESTORE_SKIPPED_OK
+          exit 0
+        fi
+        echo "backup storage locations:"
+        echo "$BSL_LINES"
+        
+        VNS="$(echo "$BSL_LINES" | head -1 | cut -d' ' -f1)"
+        
+        # ---- Gate 3: Velero server + node-agent (kopia FSB) actually ready -------
+        # File System Backup is the volume-DATA path (platform/velero/helmrelease.yaml:
+        # deployNodeAgent=true, uploaderType=kopia, snapshotsEnabled=false). Without a
+        # ready node-agent no volume data can be captured, so the drill is not runnable.
+        VREADY=0; NAREADY=0
+        for i in $(seq 1 12); do
+          VREADY="$(kubectl -n "$VNS" get deploy velero -o jsonpath='{.status.readyReplicas}' 2>/dev/null || true)"
+          NAREADY="$(kubectl -n "$VNS" get daemonset node-agent -o jsonpath='{.status.numberReady}' 2>/dev/null || true)"
+          echo "poll $i: velero readyReplicas=${VREADY:-0} node-agent numberReady=${NAREADY:-0}"
+          if [ "${VREADY:-0}" -ge 1 ] && [ "${NAREADY:-0}" -ge 1 ]; then break; fi
+          sleep 10
+        done
+        if [ "${VREADY:-0}" -lt 1 ] || [ "${NAREADY:-0}" -lt 1 ]; then
+          echo "SKIP: Velero server and/or its node-agent DaemonSet are not ready in ns '$VNS' -- backup/restore path not available"
+          echo DR_RESTORE_SKIPPED_OK
+          exit 0
+        fi
+        
+        # ---- Pick the BSL this drill will use: prefer the default, and it MUST be
+        #      Available. A CONFIGURED-BUT-UNAVAILABLE target is NOT a precondition
+        #      gap -- it is a live R-STATE-03 defect, so it FAILS (after a bounded
+        #      wait for first-validation on a cold start), it does not skip. --------
+        BSL_NS=""; BSL_NAME=""; BSL_URL=""; BSL_BUCKET=""
+        for i in $(seq 1 12); do
+          BSL_LINES="$(kubectl get backupstoragelocations.velero.io -A \
+            -o jsonpath='{range .items[*]}{.metadata.namespace}{" "}{.metadata.name}{" "}{.status.phase}{" "}{.spec.provider}{" "}{.spec.default}{" "}{.spec.config.s3Url}{" "}{.spec.objectStorage.bucket}{"\n"}{end}' 2>/dev/null || true)"
+          CAND="$(echo "$BSL_LINES" | grep ' Available ' | grep ' true ' | head -1 || true)"
+          [ -n "$CAND" ] || CAND="$(echo "$BSL_LINES" | grep ' Available ' | head -1 || true)"
+          if [ -n "$CAND" ]; then
+            BSL_NS="$(echo "$CAND" | cut -d' ' -f1)"
+            BSL_NAME="$(echo "$CAND" | cut -d' ' -f2)"
+            BSL_URL="$(echo "$CAND" | cut -d' ' -f6)"
+            BSL_BUCKET="$(echo "$CAND" | cut -d' ' -f7)"
+            break
+          fi
+          echo "poll $i: no Available BackupStorageLocation yet"
+          sleep 10
+        done
+        if [ -z "$BSL_NAME" ]; then
+          echo "FAIL: a BackupStorageLocation is configured but NONE reached phase 'Available'." >&2
+          echo "      The off-cluster backup target is unreachable -- R-STATE-03 is BROKEN, not merely untested." >&2
+          echo "$BSL_LINES" >&2
+          exit 1
+        fi
+        echo "using BSL ${BSL_NS}/${BSL_NAME}  bucket=${BSL_BUCKET}  endpoint=${BSL_URL}"
+        
+        # ---- OFF-CLUSTER PROOF (1): an endpoint we can evaluate ------------------
+        if [ -z "$BSL_URL" ]; then
+          echo "FAIL: BackupStorageLocation ${BSL_NS}/${BSL_NAME} exposes no spec.config.s3Url;" >&2
+          echo "      the OFF-CLUSTER property of the backup target cannot be established, so this" >&2
+          echo "      test refuses to claim it. (Extend this check for a non-S3 provider.)" >&2
+          exit 1
+        fi
+        HOST="${BSL_URL#*://}"      # strip scheme
+        HOST="${HOST%%/*}"          # strip path
+        HOST="${HOST%%:*}"          # strip port
+        echo "backup endpoint host: $HOST"
+        
+        case "$HOST" in
+          localhost|*.svc|*.svc.cluster.local|*.cluster.local)
+            echo "FAIL: backup endpoint '$HOST' is a CLUSTER-INTERNAL name -- the backup target is IN the cluster." >&2
+            echo "      An in-cluster backup store cannot satisfy R-STATE-03 (off-cluster backup)." >&2
+            exit 1 ;;
+        esac
+        
+        # Resolve to an IPv4 address. If it cannot be resolved we CANNOT establish the
+        # off-cluster property -- that is a failure, never a silent pass.
+        case "$HOST" in
+          *[!0-9.]*)
+            # Endpoint given as a name. Try the resolvers that may exist in the
+            # runner image, in order; if none is available or none resolves, the
+            # test FAILS below rather than guessing.
+            EPIP="$(getent ahostsv4 "$HOST" 2>/dev/null | head -1 | cut -d' ' -f1 || true)"
+            [ -n "$EPIP" ] || EPIP="$(getent hosts "$HOST" 2>/dev/null | head -1 | cut -d' ' -f1 || true)"
+            [ -n "$EPIP" ] || EPIP="$(nslookup "$HOST" 2>/dev/null | grep -A2 '^Name:' | grep '^Address' | head -1 | sed 's/.*: *//' || true)" ;;
+          *) EPIP="$HOST" ;;
+        esac
+        if [ -z "$EPIP" ]; then
+          echo "FAIL: could not resolve backup endpoint host '$HOST' to an IPv4 address;" >&2
+          echo "      the OFF-CLUSTER property cannot be asserted, so this test does not pass." >&2
+          exit 1
+        fi
+        echo "backup endpoint resolves to: $EPIP"
+        
+        # ---- OFF-CLUSTER PROOF (2): the endpoint is outside every cluster address --
+        case "$EPIP" in 127.*|0.0.0.0) echo "FAIL: backup endpoint $EPIP is loopback -- not off-cluster" >&2; exit 1 ;; esac
+        
+        NODE_IPS="$(kubectl get nodes -o jsonpath='{range .items[*]}{range .status.addresses[*]}{.address}{"\n"}{end}{end}' 2>/dev/null || true)"
+        for a in $NODE_IPS; do
+          if [ "$a" = "$EPIP" ]; then
+            echo "FAIL: backup endpoint $EPIP is a NODE address of this cluster -- the backup is not off-cluster." >&2
+            exit 1
+          fi
+        done
+        echo "endpoint is not any node address (checked $(echo "$NODE_IPS" | grep -c . || true) node addresses)"
+        
+        SVC_IPS="$(kubectl get svc -A -o jsonpath='{range .items[*]}{.spec.clusterIP}{"\n"}{end}' 2>/dev/null || true)"
+        for a in $SVC_IPS; do
+          if [ "$a" = "$EPIP" ]; then
+            echo "FAIL: backup endpoint $EPIP is a Service ClusterIP -- the backup target lives INSIDE the cluster." >&2
+            exit 1
+          fi
+        done
+        echo "endpoint is not any Service ClusterIP (checked $(echo "$SVC_IPS" | grep -c . || true))"
+        
+        POD_IPS="$(kubectl get pods -A -o jsonpath='{range .items[*]}{.status.podIP}{"\n"}{end}' 2>/dev/null || true)"
+        for a in $POD_IPS; do
+          if [ "$a" = "$EPIP" ]; then
+            echo "FAIL: backup endpoint $EPIP is a POD IP -- the backup target is an in-cluster pod." >&2
+            exit 1
+          fi
+        done
+        echo "endpoint is not any Pod IP (checked $(echo "$POD_IPS" | grep -c . || true))"
+        
+        POD_CIDRS="$(kubectl get nodes -o jsonpath='{range .items[*]}{.spec.podCIDR}{"\n"}{range .spec.podCIDRs[*]}{@}{"\n"}{end}{end}' 2>/dev/null || true)"
+        for c in $POD_CIDRS; do
+          case "$c" in *:*) continue ;; esac      # IPv6 prefixes are out of scope here
+          if in_cidr "$EPIP" "$c"; then
+            echo "FAIL: backup endpoint $EPIP falls inside node podCIDR $c -- it is in the cluster pod network." >&2
+            exit 1
+          fi
+        done
+        echo "endpoint is outside every node podCIDR (checked: $(echo "$POD_CIDRS" | tr '\n' ' '))"
+        
+        # Best-effort (needs networking.k8s.io/ServiceCIDR, which the least-privilege
+        # runner may not read): the checks above are the authoritative ones.
+        SVC_CIDRS="$(kubectl get servicecidrs.networking.k8s.io -o jsonpath='{range .items[*]}{range .spec.cidrs[*]}{@}{"\n"}{end}{end}' 2>/dev/null || true)"
+        if [ -n "$SVC_CIDRS" ]; then
+          for c in $SVC_CIDRS; do
+            case "$c" in *:*) continue ;; esac
+            if in_cidr "$EPIP" "$c"; then
+              echo "FAIL: backup endpoint $EPIP falls inside the Service CIDR $c -- it is in the cluster." >&2
+              exit 1
+            fi
+          done
+          echo "endpoint is outside every ServiceCIDR ($(echo "$SVC_CIDRS" | tr '\n' ' '))"
+        else
+          echo "(ServiceCIDR objects unreadable -- skipping that best-effort check; ClusterIP/PodIP/podCIDR checks stand)"
+        fi
+        
+        echo "OFF_CLUSTER_TARGET_OK  bsl=${BSL_NS}/${BSL_NAME} endpoint=${EPIP} bucket=${BSL_BUCKET}"
+    l.go:52: | 00:19:11 | dr-restore | assert-backup-target-is-off-cluster       | SCRIPT    | LOG   |
+        === STDOUT
+        backup storage locations:
+        velero default Available aws true http://192.168.1.9:9000 velero
+        poll 1: velero readyReplicas=1 node-agent numberReady=3
+        using BSL velero/default  bucket=velero  endpoint=http://192.168.1.9:9000
+        backup endpoint host: 192.168.1.9
+        backup endpoint resolves to: 192.168.1.9
+        endpoint is not any node address (checked 6 node addresses)
+        endpoint is not any Service ClusterIP (checked 81)
+        endpoint is not any Pod IP (checked 186)
+        endpoint is outside every node podCIDR (checked: 10.42.0.0/24 10.42.0.0/24 10.42.1.0/24 10.42.1.0/24 10.42.2.0/24 10.42.2.0/24 )
+        endpoint is outside every ServiceCIDR (10.43.0.0/16 )
+        OFF_CLUSTER_TARGET_OK  bsl=velero/default endpoint=192.168.1.9 bucket=velero
+    l.go:52: | 00:19:11 | dr-restore | assert-backup-target-is-off-cluster       | SCRIPT    | DONE  |
+    l.go:52: | 00:19:11 | dr-restore | assert-backup-target-is-off-cluster       | TRY       | END   |
+    l.go:52: | 00:19:11 | dr-restore | assert-backups-are-scheduled              | TRY       | BEGIN |
+    l.go:52: | 00:19:11 | dr-restore | assert-backups-are-scheduled              | CMD       | RUN   |
+        === COMMAND
+        /usr/bin/sh -c set -eu
+        SCHED="$(kubectl get schedules.velero.io -A \
+          -o jsonpath='{range .items[*]}{.metadata.namespace}{"/"}{.metadata.name}{" phase="}{.status.phase}{" cron="}{.spec.schedule}{" lastBackup="}{.status.lastBackup}{"\n"}{end}' 2>/dev/null || true)"
+        if [ -z "$(echo "$SCHED" | tr -d '[:space:]')" ]; then
+          # Distinguish "Velero absent" (skip, in lock-step with step 1) from
+          # "Velero present but nothing scheduled" (a real R-STATE-03 defect).
+          if kubectl get backupstoragelocations.velero.io -A >/dev/null 2>&1; then
+            if [ -z "$(kubectl get backupstoragelocations.velero.io -A -o name 2>/dev/null || true)" ]; then
+              echo "SKIP: no BackupStorageLocation configured -- scheduling not applicable yet"
+              echo DR_SCHEDULE_SKIPPED_OK
+              exit 0
+            fi
+            echo "FAIL: Velero is configured with a backup target but declares NO Schedule --" >&2
+            echo "      R-STATE-03 requires SCHEDULED off-cluster backups, not manual ones." >&2
+            exit 1
+          fi
+          echo "SKIP: Velero API absent -- backup scheduling not applicable on this cluster yet"
+          echo DR_SCHEDULE_SKIPPED_OK
+          exit 0
+        fi
+        echo "velero schedules:"
+        echo "$SCHED"
+        # A Schedule that exists but is paused/disabled backs nothing up.
+        if echo "$SCHED" | grep -q 'phase=Enabled'; then
+          echo BACKUPS_ARE_SCHEDULED_OK
+        else
+          echo "FAIL: no Velero Schedule is in phase 'Enabled' -- scheduled off-cluster backups are not running." >&2
+          exit 1
+        fi
+    l.go:52: | 00:19:11 | dr-restore | assert-backups-are-scheduled              | SCRIPT    | LOG   |
+        === STDOUT
+        velero schedules:
+        velero/velero-daily phase=Enabled cron=0 3 * * * lastBackup=2026-09-01T03:00:27Z
+        velero/velero-daily-resources-only phase=Enabled cron=0 3 * * * lastBackup=2026-09-01T03:00:27Z
+        BACKUPS_ARE_SCHEDULED_OK
+    l.go:52: | 00:19:11 | dr-restore | assert-backups-are-scheduled              | SCRIPT    | DONE  |
+    l.go:52: | 00:19:11 | dr-restore | assert-backups-are-scheduled              | TRY       | END   |
+    l.go:52: | 00:19:11 | dr-restore | data-restores-from-the-off-cluster-backup | TRY       | BEGIN |
+    l.go:52: | 00:19:11 | dr-restore | data-restores-from-the-off-cluster-backup | CMD       | RUN   |
+        === COMMAND
+        /usr/bin/sh -c set -eu
+        NS="$NAMESPACE"
+        SRC="${NS}-dr-src"          # scratch source namespace (destroyed by design)
+        DST="${NS}-dr-dst"          # clean restore target (must not pre-exist)
+        BK="${NS}-dr-backup"
+        RS="${NS}-dr-restore"
+        OWNER_LABEL="conformance.home.k3s/created-by=${NS}"
+        
+        # ---- Re-derive the preconditions (steps do not share shell state). Every
+        #      skip below mirrors step 1 exactly, so the whole test skips or runs
+        #      as a unit -- it can never half-run. -----------------------------
+        if ! kubectl get storageclass longhorn >/dev/null 2>&1; then
+          echo "SKIP: 'longhorn' StorageClass absent"; echo DR_RESTORE_SKIPPED_OK; exit 0
+        fi
+        BSL_LINES="$(kubectl get backupstoragelocations.velero.io -A \
+          -o jsonpath='{range .items[*]}{.metadata.namespace}{" "}{.metadata.name}{" "}{.status.phase}{" "}{.spec.default}{"\n"}{end}' 2>/dev/null || true)"
+        CAND="$(echo "$BSL_LINES" | grep ' Available ' | grep ' true$' | head -1 || true)"
+        [ -n "$CAND" ] || CAND="$(echo "$BSL_LINES" | grep ' Available ' | head -1 || true)"
+        if [ -z "$CAND" ]; then
+          echo "SKIP: no Available BackupStorageLocation (see the off-cluster step for the authoritative verdict)"
+          echo DR_RESTORE_SKIPPED_OK; exit 0
+        fi
+        VNS="$(echo "$CAND" | cut -d' ' -f1)"
+        BSL_NAME="$(echo "$CAND" | cut -d' ' -f2)"
+        if [ "$(kubectl -n "$VNS" get daemonset node-agent -o jsonpath='{.status.numberReady}' 2>/dev/null || echo 0)" -lt 1 ]; then
+          echo "SKIP: Velero node-agent (kopia FSB) not ready -- volume data cannot be backed up"
+          echo DR_RESTORE_SKIPPED_OK; exit 0
+        fi
+        echo "velero ns=$VNS bsl=$BSL_NAME src=$SRC dst=$DST"
+        
+        # A clean restore target means PREVIOUSLY NONEXISTENT, and the source name must
+        # be free. If either is lying around from an aborted run, remove it before we
+        # claim "clean" -- and fail loudly if it will not go, because a half-deleted
+        # namespace would silently invalidate the destroy proof below.
+        kubectl delete namespace "$DST" "$SRC" --ignore-not-found --wait=true --timeout=300s
+        
+        # ============================ GIVEN ==============================
+        # A DR-critical stateful workload with known data on the replicated tier.
+        CANARY="dr-canary-$(date +%s%N)"
+        kubectl create namespace "$SRC"
+        kubectl label namespace "$SRC" "$OWNER_LABEL" --overwrite
+        
+        kubectl -n "$SRC" apply -f - <<EOF
+        apiVersion: v1
+        kind: PersistentVolumeClaim
+        metadata:
+          name: dr-data
+          labels:
+            conformance.home.k3s/created-by: ${NS}
+        spec:
+          accessModes: [ReadWriteOnce]
+          storageClassName: longhorn
+          resources:
+            requests:
+              storage: 1Gi
+        EOF
+        
+        kubectl -n "$SRC" apply -f - <<EOF
+        apiVersion: v1
+        kind: Pod
+        metadata:
+          name: dr-workload
+          labels:
+            app: dr-workload
+            conformance.home.k3s/created-by: ${NS}
+        spec:
+          restartPolicy: Never
+          containers:
+            - name: app
+              image: busybox:1.36.1
+              command: ["sh", "-c", "sleep 7200"]
+              volumeMounts:
+                - name: data
+                  mountPath: /data
+          volumes:
+            - name: data
+              persistentVolumeClaim:
+                claimName: dr-data
+        EOF
+        
+        kubectl -n "$SRC" wait --for=condition=Ready pod/dr-workload --timeout=300s
+        printf '%s' "$CANARY" | kubectl -n "$SRC" exec -i dr-workload -- sh -c 'cat > /data/canary && sync'
+        echo "seeded canary: $(kubectl -n "$SRC" exec dr-workload -- cat /data/canary)"
+        
+        ORIG_PV="$(kubectl -n "$SRC" get pvc dr-data -o jsonpath='{.spec.volumeName}')"
+        [ -n "$ORIG_PV" ] || { echo "FAIL: source PVC never bound to a PV" >&2; exit 1; }
+        echo "original in-cluster PersistentVolume: $ORIG_PV"
+        
+        # ======================= WHEN (part 1): back up off-cluster =======
+        # Created as a Velero Backup CR -- exactly what `velero backup create` does.
+        # defaultVolumesToFsBackup:true selects the kopia File System Backup path, so
+        # the VOLUME DATA (not just the resource JSON) is uploaded to the object store.
+        # ttl is short: if the finally-step DeleteBackupRequest is ever lost, Velero's
+        # own GC still removes this drill's objects from the bucket.
+        kubectl apply -f - <<EOF
+        apiVersion: velero.io/v1
+        kind: Backup
+        metadata:
+          name: ${BK}
+          namespace: ${VNS}
+          labels:
+            conformance.home.k3s/created-by: ${NS}
+        spec:
+          includedNamespaces: [${SRC}]
+          storageLocation: ${BSL_NAME}
+          snapshotVolumes: false
+          defaultVolumesToFsBackup: true
+          ttl: 2h0m0s
+        EOF
+        
+        BPHASE=""
+        for i in $(seq 1 90); do
+          BPHASE="$(kubectl -n "$VNS" get backups.velero.io "$BK" -o jsonpath='{.status.phase}' 2>/dev/null || true)"
+          echo "poll $i: backup phase=${BPHASE:-<pending>}"
+          case "${BPHASE:-}" in
+            Completed|Failed|PartiallyFailed|FailedValidation) break ;;
+          esac
+          sleep 10
+        done
+        if [ "${BPHASE:-}" != "Completed" ]; then
+          echo "FAIL: off-cluster backup did not complete (phase=${BPHASE:-<timeout>})" >&2
+          kubectl -n "$VNS" get backups.velero.io "$BK" -o yaml >&2 || true
+          exit 1
+        fi
+        echo "backup $BK Completed to ${BSL_NAME}"
+        
+        # ---- OFF-CLUSTER PROOF (3): the VOLUME DATA really landed in the remote
+        #      kopia repository -- a PodVolumeBackup Completed with a snapshot ID.
+        PVB="$(kubectl -n "$VNS" get podvolumebackups.velero.io \
+                 -l velero.io/backup-name="$BK" \
+                 -o jsonpath='{range .items[*]}{.metadata.name}{" "}{.status.phase}{" "}{.status.snapshotID}{"\n"}{end}' 2>/dev/null || true)"
+        echo "pod volume backups:"
+        echo "$PVB"
+        if [ -z "$(echo "$PVB" | tr -d '[:space:]')" ]; then
+          echo "FAIL: the backup captured NO pod volume data (no PodVolumeBackup)." >&2
+          echo "      Resource-only backups cannot restore DR-critical DATA." >&2
+          exit 1
+        fi
+        # A line is "<name> <phase> <snapshotID>"; require Completed AND a
+        # non-empty snapshot ID (an empty ID means nothing reached the repository).
+        GOOD_PVB="$(echo "$PVB" | grep -cE ' Completed [^[:space:]]+$' || true)"
+        if [ "${GOOD_PVB:-0}" -lt 1 ]; then
+          echo "FAIL: no PodVolumeBackup reached phase Completed with a snapshot ID --" >&2
+          echo "      the volume data did not reach the off-cluster repository." >&2
+          exit 1
+        fi
+        echo "volume data uploaded off-cluster (PodVolumeBackup Completed with a kopia snapshotID)"
+        
+        # ============ WHEN (part 2): FAULT INJECTION -- annihilate the source ====
+        # Destroy EVERY in-cluster copy: the pod, the PVC, the namespace, and the
+        # underlying PersistentVolume (and with it the Longhorn replicas). After this
+        # there is no in-cluster source the restore could quietly read from.
+        kubectl delete namespace "$SRC" --wait=true --timeout=420s
+        for i in $(seq 1 30); do
+          if ! kubectl get namespace "$SRC" >/dev/null 2>&1; then break; fi
+          echo "poll $i: waiting for namespace $SRC to disappear"
+          sleep 10
+        done
+        if kubectl get namespace "$SRC" >/dev/null 2>&1; then
+          echo "FAIL: source namespace $SRC did not terminate; the destroy step is not convincing" >&2
+          exit 1
+        fi
+        echo "source namespace destroyed"
+        
+        # The longhorn StorageClass reclaims Delete, so the PV should already be gone;
+        # force it if a Retain policy (or a stuck finalizer) left it behind, because a
+        # surviving PV would make the restore prove nothing about off-cluster recovery.
+        kubectl delete pv "$ORIG_PV" --ignore-not-found --wait=false || true
+        for i in $(seq 1 30); do
+          if ! kubectl get pv "$ORIG_PV" >/dev/null 2>&1; then break; fi
+          echo "poll $i: waiting for PersistentVolume $ORIG_PV to disappear"
+          sleep 10
+        done
+        if kubectl get pv "$ORIG_PV" >/dev/null 2>&1; then
+          echo "FAIL: the original PersistentVolume $ORIG_PV SURVIVED the destroy step." >&2
+          echo "      An in-cluster copy of the data still exists, so a successful restore" >&2
+          echo "      would NOT prove recovery from the off-cluster backup. Refusing to pass." >&2
+          exit 1
+        fi
+        echo "original PersistentVolume $ORIG_PV is GONE -- no in-cluster copy of the data remains"
+        
+        # Best-effort corroboration (the least-privilege runner may not read longhorn.io):
+        # the Longhorn Volume CR and its replicas should be gone too.
+        if kubectl -n longhorn-system get volumes.longhorn.io "$ORIG_PV" >/dev/null 2>&1; then
+          echo "FAIL: Longhorn Volume $ORIG_PV still exists after PV deletion -- replicas survive in-cluster" >&2
+          exit 1
+        fi
+        echo "(longhorn Volume CR $ORIG_PV not present -- either deleted or not readable by this identity)"
+        
+        # ============================ RESTORE ============================
+        # Into a CLEAN, previously-nonexistent namespace (Velero creates it), which is
+        # the catalog's "restored into a clean state".
+        kubectl apply -f - <<EOF
+        apiVersion: velero.io/v1
+        kind: Restore
+        metadata:
+          name: ${RS}
+          namespace: ${VNS}
+          labels:
+            conformance.home.k3s/created-by: ${NS}
+        spec:
+          backupName: ${BK}
+          includedNamespaces: [${SRC}]
+          namespaceMapping:
+            ${SRC}: ${DST}
+          existingResourcePolicy: none
+        EOF
+        
+        RPHASE=""
+        for i in $(seq 1 90); do
+          RPHASE="$(kubectl -n "$VNS" get restores.velero.io "$RS" -o jsonpath='{.status.phase}' 2>/dev/null || true)"
+          echo "poll $i: restore phase=${RPHASE:-<pending>}"
+          case "${RPHASE:-}" in
+            Completed|Failed|PartiallyFailed|FailedValidation) break ;;
+          esac
+          sleep 10
+        done
+        if [ "${RPHASE:-}" != "Completed" ]; then
+          echo "FAIL: restore did not complete (phase=${RPHASE:-<timeout>})" >&2
+          kubectl -n "$VNS" get restores.velero.io "$RS" -o yaml >&2 || true
+          kubectl -n "$VNS" get podvolumerestores.velero.io -l velero.io/restore-name="$RS" -o wide >&2 || true
+          exit 1
+        fi
+        echo "restore $RS Completed into $DST"
+        
+        # ============================= THEN ==============================
+        # The service returns WITH ITS DATA. With kopia FSB the restored pod carries a
+        # `restore-wait` init container that blocks until the node-agent has written the
+        # volume contents back, so Ready here already implies the data landed.
+        if ! kubectl -n "$DST" wait --for=condition=Ready pod/dr-workload --timeout=600s; then
+          echo "FAIL: restored workload never became Ready in the clean namespace $DST" >&2
+          kubectl -n "$DST" describe pod dr-workload >&2 || true
+          kubectl -n "$VNS" get podvolumerestores.velero.io -l velero.io/restore-name="$RS" -o wide >&2 || true
+          exit 1
+        fi
+        
+        NEW_PV="$(kubectl -n "$DST" get pvc dr-data -o jsonpath='{.spec.volumeName}' 2>/dev/null || true)"
+        echo "restored PVC bound to PersistentVolume: ${NEW_PV:-<none>}"
+        [ -n "$NEW_PV" ] || { echo "FAIL: restored PVC did not bind a PV" >&2; exit 1; }
+        if [ "$NEW_PV" = "$ORIG_PV" ]; then
+          echo "FAIL: the restored PVC rebound the ORIGINAL PersistentVolume ($ORIG_PV)." >&2
+          echo "      That is an in-cluster survival, not an off-cluster restore." >&2
+          exit 1
+        fi
+        echo "restored onto a NEW volume ($NEW_PV != $ORIG_PV) -- the data came from outside the cluster"
+        
+        GOT="$(kubectl -n "$DST" exec dr-workload -- cat /data/canary)"
+        echo "recovered='$GOT' expected='$CANARY'"
+        if [ "$GOT" != "$CANARY" ]; then
+          echo "FAIL: restored data does not match what was backed up -- DR restore lost/corrupted data" >&2
+          exit 1
+        fi
+        
+        echo DR_RESTORED_FROM_OFF_CLUSTER_BACKUP_OK
+    l.go:52: | 00:21:16 | dr-restore | data-restores-from-the-off-cluster-backup | SCRIPT    | LOG   |
+        === STDOUT
+        velero ns=velero bsl=default src=chainsaw-cunning-hen-dr-src dst=chainsaw-cunning-hen-dr-dst
+        namespace/chainsaw-cunning-hen-dr-src created
+        namespace/chainsaw-cunning-hen-dr-src labeled
+        persistentvolumeclaim/dr-data created
+        pod/dr-workload created
+        pod/dr-workload condition met
+        seeded canary: dr-canary-1788308352223263220
+        original in-cluster PersistentVolume: pvc-b9523c15-770e-4fc5-8130-ca2e9d9da745
+        backup.velero.io/chainsaw-cunning-hen-dr-backup created
+        poll 1: backup phase=InProgress
+        poll 2: backup phase=InProgress
+        poll 3: backup phase=Completed
+        backup chainsaw-cunning-hen-dr-backup Completed to default
+        pod volume backups:
+        chainsaw-cunning-hen-dr-backup-l95lw Completed 47fa2737115e17142fb41857cef4e779
+        volume data uploaded off-cluster (PodVolumeBackup Completed with a kopia snapshotID)
+        namespace "chainsaw-cunning-hen-dr-src" deleted
+        source namespace destroyed
+        original PersistentVolume pvc-b9523c15-770e-4fc5-8130-ca2e9d9da745 is GONE -- no in-cluster copy of the data remains
+        (longhorn Volume CR pvc-b9523c15-770e-4fc5-8130-ca2e9d9da745 not present -- either deleted or not readable by this identity)
+        restore.velero.io/chainsaw-cunning-hen-dr-restore created
+        poll 1: restore phase=InProgress
+        poll 2: restore phase=InProgress
+        poll 3: restore phase=InProgress
+        poll 4: restore phase=Completed
+        restore chainsaw-cunning-hen-dr-restore Completed into chainsaw-cunning-hen-dr-dst
+        pod/dr-workload condition met
+        restored PVC bound to PersistentVolume: pvc-63c47973-9c7f-4423-a27e-caa008026e49
+        restored onto a NEW volume (pvc-63c47973-9c7f-4423-a27e-caa008026e49 != pvc-b9523c15-770e-4fc5-8130-ca2e9d9da745) -- the data came from outside the cluster
+        recovered='dr-canary-1788308352223263220' expected='dr-canary-1788308352223263220'
+        DR_RESTORED_FROM_OFF_CLUSTER_BACKUP_OK
+        === STDERR
+        Defaulted container "app" out of: app, restore-wait (init)
+    l.go:52: | 00:21:16 | dr-restore | data-restores-from-the-off-cluster-backup | SCRIPT    | DONE  |
+    l.go:52: | 00:21:16 | dr-restore | data-restores-from-the-off-cluster-backup | TRY       | END   |
+    l.go:52: | 00:21:16 | dr-restore | data-restores-from-the-off-cluster-backup | FINALLY   | BEGIN |
+    l.go:52: | 00:21:16 | dr-restore | data-restores-from-the-off-cluster-backup | CMD       | RUN   |
+        === COMMAND
+        /usr/bin/sh -c set -u
+        NS="$NAMESPACE"
+        SRC="${NS}-dr-src"; DST="${NS}-dr-dst"; BK="${NS}-dr-backup"; RS="${NS}-dr-restore"
+        
+        VNS="$(kubectl get backupstoragelocations.velero.io -A -o jsonpath='{.items[0].metadata.namespace}' 2>/dev/null || true)"
+        
+        # Velero objects first: a DeleteBackupRequest makes Velero remove the backup
+        # DATA from the off-cluster bucket, not just the CR (self-cleaning contract).
+        if [ -n "$VNS" ]; then
+          kubectl -n "$VNS" delete restores.velero.io "$RS" --ignore-not-found --wait=false || true
+          if kubectl -n "$VNS" get backups.velero.io "$BK" >/dev/null 2>&1; then
+            kubectl apply -f - <<EOF || true
+        apiVersion: velero.io/v1
+        kind: DeleteBackupRequest
+        metadata:
+          name: ${BK}-delete
+          namespace: ${VNS}
+          labels:
+            conformance.home.k3s/created-by: ${NS}
+        spec:
+          backupName: ${BK}
+        EOF
+            for i in $(seq 1 18); do
+              kubectl -n "$VNS" get backups.velero.io "$BK" >/dev/null 2>&1 || break
+              sleep 10
+            done
+            # Backstop if Velero could not process the request (the 2h ttl also GCs it).
+            kubectl -n "$VNS" delete backups.velero.io "$BK" --ignore-not-found --wait=false || true
+            kubectl -n "$VNS" delete deletebackuprequests.velero.io "${BK}-delete" --ignore-not-found --wait=false || true
+          fi
+          echo "velero objects for $NS cleaned up (ns $VNS)"
+        fi
+        
+        # Scratch namespaces (both, in case the run aborted at any point).
+        for n in "$DST" "$SRC"; do
+          kubectl delete namespace "$n" --ignore-not-found --wait=false || true
+        done
+        # Anything left labelled as ours (defensive; the two names above are the only
+        # namespaces this test ever creates).
+        kubectl delete namespace -l "conformance.home.k3s/created-by=${NS}" --ignore-not-found --wait=false || true
+        
+        # Released PVs from either namespace's PVCs would otherwise leak storage.
+        for pv in $(kubectl get pv -o jsonpath='{range .items[*]}{.metadata.name}{" "}{.spec.claimRef.namespace}{"\n"}{end}' 2>/dev/null | grep -E " (${SRC}|${DST})$" | cut -d' ' -f1 || true); do
+          kubectl delete pv "$pv" --ignore-not-found --wait=false || true
+          echo "released leftover PersistentVolume: $pv"
+        done
+        
+        echo "dr-restore teardown complete (best-effort; every object was created by this test)"
+    l.go:52: | 00:21:28 | dr-restore | data-restores-from-the-off-cluster-backup | SCRIPT    | LOG   |
+        === STDOUT
+        restore.velero.io "chainsaw-cunning-hen-dr-restore" deleted from velero namespace
+        deletebackuprequest.velero.io/chainsaw-cunning-hen-dr-backup-delete created
+        velero objects for chainsaw-cunning-hen cleaned up (ns velero)
+        namespace "chainsaw-cunning-hen-dr-dst" deleted
+        namespace "chainsaw-cunning-hen-dr-dst" deleted
+        persistentvolume "pvc-63c47973-9c7f-4423-a27e-caa008026e49" deleted
+        released leftover PersistentVolume: pvc-63c47973-9c7f-4423-a27e-caa008026e49
+        dr-restore teardown complete (best-effort; every object was created by this test)
+    l.go:52: | 00:21:28 | dr-restore | data-restores-from-the-off-cluster-backup | SCRIPT    | DONE  |
+    l.go:52: | 00:21:28 | dr-restore | data-restores-from-the-off-cluster-backup | FINALLY   | END   |
+    l.go:52: | 00:21:28 | dr-restore | @chainsaw                                 | CLEANUP   | BEGIN |
+    l.go:52: | 00:21:28 | dr-restore | @chainsaw                                 | DELETE    | OK    | v1/Namespace @ chainsaw-cunning-hen
+    l.go:52: | 00:21:36 | dr-restore | @chainsaw                                 | CLEANUP   | END   |
+--- PASS: chainsaw (0.00s)
+    --- PASS: chainsaw/dr-restore (149.05s)
+PASS
+Tests Summary...
+- Passed  tests 1
+- Failed  tests 0
+- Skipped tests 0
+Done.
+merge_junit: merged 1 report(s) -> /work/conformance/_artifacts/report.junit.xml (tests=1 failures=0 errors=0 skipped=0)
+[run.sh] Done. JUnit: /work/conformance/_artifacts/report.junit.xml | Coverage: /work/conformance/_artifacts/coverage/coverage.md

conformance/runs/ledger.jsonl

@@ -0,0 +1,2 @@
+{"chainsaw_version": "v0.2.12", "env": "dev", "git_sha": "57d4df8c4f992732a49f3e2daddcace866778ab0", "junit_result": "passed", "kube_version": "v1.36.4+k3s1", "layer": "L3", "note": "run log OUTPUT carries every declared pass token (NODE_LOSS_VOLUME_AVAILABLE_OK) and no skip token [log scoped to this testcase; tokens read from real-output sections only]", "proof": "proven", "requirements": ["R-STATE-02"], "run_id": "20260901T235117Z-e9a4b66", "sentinel": "NODE_LOSS_VOLUME_AVAILABLE_OK", "test": "node-loss", "ts_utc": "2026-09-01T23:51:17Z"}
+{"chainsaw_version": "v0.2.12", "env": "dev", "git_sha": "fdc3265585f4b775c0a6ae3fb33c837b0c643655", "junit_result": "passed", "kube_version": "v1.36.4+k3s1", "layer": "L3", "note": "run log OUTPUT carries every declared pass token (OFF_CLUSTER_TARGET_OK, BACKUPS_ARE_SCHEDULED_OK, DR_RESTORED_FROM_OFF_CLUSTER_BACKUP_OK) and no skip token [log scoped to this testcase; tokens read from real-output sections only]", "proof": "proven", "requirements": ["R-STATE-03"], "run_id": "20260902T001907Z-fdc3265", "sentinel": "DR_RESTORED_FROM_OFF_CLUSTER_BACKUP_OK", "test": "dr-restore", "ts_utc": "2026-09-02T00:19:07Z"}