#!/usr/bin/env python3
"""Isolated six-node SDP assurance using real Helix services and independent disks.

Run against an authoritative candidate or submitted export via --source. No host
mounts or published ports are used. Failures retain labelled nodes and evidence;
--cleanup removes only this prefix's verified labelled resources.
"""
import argparse
import hashlib
import io
import json
import marshal
import os
from pathlib import Path
import re
import secrets
import shlex
import subprocess
import tempfile
import time

LABEL = "io.perforce.sdp.disposable-topology"
NODES = {"commit": ("master", "commit"), "edge": ("edge", "commit"),
         "standby": ("standby", "commit"), "edge-standby": ("edge-standby", "edge"),
         "broker": ("broker", "commit"), "proxy": ("proxy", "commit")}
HERE = Path(__file__).resolve().parent
REQUIRED_SCENARIOS = {"six_node_install", "exact_fixture", "replication_and_local_shelves",
                      "broker_proxy_routes", "checkpoint_database_switch", "retained_generation_replay",
                      "missing_archive_repair", "restart_failure_preserves_data",
                      "interrupted_recovery_preserves_data", "standby_promotion"}


def decode(value):
    if isinstance(value, bytes):
        return value.decode("utf-8", "surrogateescape")
    if isinstance(value, dict):
        return {decode(k): decode(v) for k, v in value.items()}
    if isinstance(value, list):
        return [decode(v) for v in value]
    return value


class Topology:
    def __init__(self, args):
        self.args = args
        if not args.confirm_disposable or not re.fullmatch(r"sdp-assurance-topology-[a-z0-9-]+", args.prefix):
            raise ValueError("Explicit confirmation and a unique sdp-assurance-topology-* prefix are required")
        self.source = Path(args.source).resolve()
        if not (self.source / ".export-manifest.json").is_file():
            raise ValueError("An authoritative export manifest is required")
        self.results = Path(args.results).resolve()
        self.results.mkdir(parents=True, exist_ok=True)
        self.password = "SdpFixture" + secrets.token_hex(12) + "!"
        self.records = []
        self.shelves = {}
        self.expected = {}
        self.docker = [args.docker]
        self.network = args.prefix + "-net"
        self.node_image = args.prefix + ":node"
        self.started = time.time()
        self.command_index = 0
        self.node_evidence = {}
        self.image_evidence = {}
        self.python_bins = {}
        self.driver_evidence = {}
        for name in ("topology_integration.py", "topology_bootstrap.sh"):
            payload = (HERE / name).read_bytes()
            self.driver_evidence[name] = hashlib.sha256(payload).hexdigest()
            if not args.cleanup:
                (self.results / ("driver-" + name)).write_bytes(payload)
        state = self.results / ".fixture-state.json"
        if args.resume:
            saved = json.loads(state.read_text())
            if saved["prefix"] != args.prefix or saved["source"] != str(self.source):
                raise ValueError("Resume state does not match the requested deployment")
            self.password = saved["password"]
            self.shelves = saved["shelves"]
            self.expected = saved["expected"]
            self.node_evidence = json.loads((self.results / "topology-results.json").read_text()).get("nodes", {})
            self.image_evidence = json.loads((self.results / "topology-results.json").read_text()).get("image_evidence", {})
            self.records = [r for r in json.loads((self.results / "topology-results.json").read_text())["scenarios"] if r["status"] == "passed"]
            for node in NODES:
                obj = json.loads(self.cmd(self.docker + ["inspect", self.name(node)]).stdout)[0]
                if obj["Config"].get("Labels", {}).get(LABEL) != args.prefix:
                    raise ValueError("Resume refuses an unowned container")

    def cmd(self, argv, *, data=None, check=True, timeout=300, record=True):
        started = time.monotonic()
        result = subprocess.run([str(x) for x in argv], input=data, capture_output=True, timeout=timeout)
        self.command_index += 1
        if record:
            entry = {"command": [str(x).replace(self.password, "<fixture credential>") for x in argv],
                     "returncode": result.returncode, "seconds": round(time.monotonic() - started, 3)}
            with (self.results / "commands.jsonl").open("a") as out:
                out.write(json.dumps(entry) + "\n")
            # Raw P4 marshal output may contain credentials. Save only diagnostics
            # on failure, redacted, and explicit inventories elsewhere.
        if check and result.returncode:
            detail = (result.stdout + result.stderr).decode(errors="replace").replace(self.password, "<fixture credential>")
            raise RuntimeError("Command failed (%d): %s\n%s" % (result.returncode, shlex.join(entry["command"] if record else argv), detail[-8000:]))
        return result

    def name(self, node):
        if node not in NODES:
            raise ValueError("Unknown fixture node")
        return self.args.prefix + "-" + node

    def execute(self, node, argv, **kwargs):
        user = kwargs.pop("user", "perforce")
        if argv and argv[0] == "python3":
            if node not in self.python_bins:
                result = self.cmd(self.docker + ["exec", self.name(node), "printenv", "SDP_DISTRO_PYTHON"])
                interpreter = result.stdout.decode().strip()
                if not interpreter.startswith("/") or "\n" in interpreter:
                    raise RuntimeError("Image must declare an absolute SDP_DISTRO_PYTHON path")
                self.python_bins[node] = interpreter
            argv = [self.python_bins[node], *argv[1:]]
        return self.cmd(self.docker + ["exec", "-i", "--user", user, self.name(node)] + list(argv), **kwargs)

    def shell(self, node, source, **kwargs):
        return self.execute(node, ["bash", "-c", source], **kwargs)

    def write(self, node, path, data, *, user="perforce", mode=None):
        script = "import pathlib,sys; p=pathlib.Path(sys.argv[1]); p.parent.mkdir(parents=True,exist_ok=True); p.write_bytes(sys.stdin.buffer.read())"
        if mode is not None:
            script += "; p.chmod(int(sys.argv[2],8))"
        argv = ["python3", "-c", script, str(path)] + ([oct(mode)[2:]] if mode is not None else [])
        self.execute(node, argv, user=user, data=data.encode() if isinstance(data, str) else data)

    def p4(self, node, *args, data=None, raw=False, check=True, port="127.0.0.1:1666", user="perforce", timeout=90):
        argv = ["env", "P4TICKETS=/p4/1/.p4tickets", "/p4/1/bin/p4_1", "-p", port, "-u", user]
        if not raw:
            argv += ["-G"]
        result = self.execute(node, argv + list(args), data=data, check=False, timeout=timeout)
        if raw:
            if check and result.returncode:
                raise RuntimeError("P4 command failed: %s %s" % (args, result.stderr.decode(errors="replace")))
            return result
        records = []
        stream = io.BytesIO(result.stdout)
        try:
            while stream.tell() < len(result.stdout):
                records.append(decode(marshal.load(stream)))
        except (ValueError, EOFError) as error:
            raise RuntimeError("Invalid P4 marshaled response: " + repr(result.stdout[-1000:])) from error
        errors = [r for r in records if r.get("code") == "error" and int(r.get("severity", 0)) >= 3]
        if check and (result.returncode or errors):
            raise RuntimeError("P4 %s on %s failed: %s %s" % (args, node, errors or records, result.stderr.decode(errors="replace")))
        return records

    def form(self, node, command, fields, *flags, client=None):
        return self.p4(node, *(["-c", client] if client else []), command, *flags, "-i", data=marshal.dumps({k.encode(): str(v).encode() for k, v in fields.items()}, 0))

    def script(self, node, name, *args, check=True, timeout=900):
        result = self.execute(node, ["bash", "/p4/common/bin/" + name, *map(str, args or ("1",))], check=False, timeout=timeout)
        if check and result.returncode:
            logs = self.shell(node, "tail -100 /p4/1/logs/checkpoint.log /p4/1/logs/convert_standby_to_master.log 2>/dev/null", check=False).stdout
            raise RuntimeError(name + " failed: " + (result.stdout + result.stderr + logs).decode(errors="replace")[-10000:])
        if check and self.execute(node, ["test", "-e", "/p4/1/logs/ckp_running.txt"], check=False).returncode == 0:
            raise RuntimeError(name + " returned success but retained its recovery lock")
        return result

    def service(self, node, action, daemon="p4d", check=True):
        return self.execute(node, ["systemctl", action, daemon + "_1.service"], user="root", check=check, timeout=180)

    def scenario(self, name, scripts, operation):
        print("START " + name, flush=True)
        start = time.monotonic()
        row = {"name": name, "scripts": scripts, "kind": "live", "required": True}
        try:
            row["evidence"] = operation() or {}
            row["status"] = "passed"
        except (OSError, ValueError, RuntimeError, subprocess.SubprocessError) as error:
            row.update(status="failed", error=str(error))
        row["seconds"] = round(time.monotonic() - start, 3)
        self.records.append(row)
        self.save()
        print(json.dumps(row), flush=True)
        return row["status"] == "passed"

    def save(self):
        manifest = self.source / ".export-manifest.json"
        output = {"source": str(self.source), "source_manifest_sha256": hashlib.sha256(manifest.read_bytes()).hexdigest(),
                  "image": self.args.image, "node_image": self.node_image, "prefix": self.args.prefix,
                  "driver_sha256": self.driver_evidence,
                  "journal": "/p4/1/journals/journal", "shared_storage": False, "verification_probe_journaling": "off; read-only inventory queries on scratch restored copies",
                  "image_evidence": self.image_evidence, "nodes": self.node_evidence, "scenarios": self.records,
                  "all_required_passed": {r["name"] for r in self.records} == REQUIRED_SCENARIOS and all(r["status"] == "passed" for r in self.records)}
        (self.results / "topology-results.json").write_text(json.dumps(output, indent=2) + "\n")
        state = self.results / ".fixture-state.json"
        state.write_text(json.dumps({"prefix": self.args.prefix, "source": str(self.source), "password": self.password, "shelves": self.shelves, "expected": self.expected}))
        state.chmod(0o600)

    def provision(self):
        binaries = Path(self.args.binaries).resolve()
        for name in ("p4", "p4d", "p4broker", "p4p", "SHA256SUMS"):
            if not (binaries / name).is_file():
                raise ValueError("Missing vendor-checked binary input " + name)
        hashes = {}
        for line in (binaries / "SHA256SUMS").read_text().splitlines():
            fields = line.split()
            if len(fields) == 2:
                hashes[fields[1].lstrip("*")] = fields[0].lower()
        for name in ("p4", "p4d", "p4broker", "p4p"):
            if hashlib.sha256((binaries / name).read_bytes()).hexdigest() != hashes.get(name):
                raise ValueError("Vendor manifest checksum mismatch: " + name)
        # systemd-binfmt in a privileged container can unregister the VM's
        # shared QEMU handlers. Mask only that unrelated host-global service.
        if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9./:@_-]*", self.args.image):
            raise ValueError("Invalid image reference")
        self.image_evidence["base_id"] = json.loads(self.cmd(self.docker + ["image", "inspect", self.args.image]).stdout)[0]["Id"]
        dockerfile = "FROM " + self.args.image + "\nRUN ln -sf /dev/null /etc/systemd/system/systemd-binfmt.service\n"
        self.cmd(self.docker + ["build", "--network=none", "--label", LABEL + "=" + self.args.prefix, "-t", self.node_image, "-"], data=dockerfile.encode(), timeout=120)
        self.image_evidence["node_id"] = json.loads(self.cmd(self.docker + ["image", "inspect", self.node_image]).stdout)[0]["Id"]
        self.cmd(self.docker + ["network", "create", "--internal", "--label", LABEL + "=" + self.args.prefix, self.network])
        for node, (role, upstream) in NODES.items():
            self.cmd(self.docker + ["run", "-d", "--name", self.name(node), "--hostname", node,
                                   "--network", self.network, "--network-alias", node,
                                   "--label", LABEL + "=" + self.args.prefix, "--privileged", "--cgroupns=private",
                                   "--tmpfs", "/run", "--tmpfs", "/run/lock", "--tmpfs", "/tmp",
                                   self.node_image, "/sbin/init"])
            self.execute(node, ["mkdir", "-p", "/inputs", "/results"], user="root")
            self.execute(node, ["touch", "/run/sdp-topology-test"], user="root")
            self.cmd(self.docker + ["cp", str(self.source), self.name(node) + ":/inputs/sdp"])
            self.cmd(self.docker + ["cp", str(binaries), self.name(node) + ":/inputs/binaries"])
            self.cmd(self.docker + ["cp", str(HERE / "topology_bootstrap.sh"), self.name(node) + ":/inputs/topology_bootstrap.sh"])
            self.execute(node, ["bash", "/inputs/topology_bootstrap.sh", node, role, upstream, self.password], user="root", timeout=600)
            self.execute(node, ["test", "!", "-e", "/p4/1/logs/journal"])
        info = self.p4("commit", "info")
        if not any(r.get("serverID", r.get("ServerID")) == "commit" for r in info):
            raise RuntimeError("Commit identity mismatch")
        return {"nodes": list(NODES), "installer": "mkdirs.sh with separate mkdirs.cfg per node", "vendor_sha256_verified": True}

    def client(self, node, name):
        self.execute(node, ["mkdir", "-p", "/p4/1/tmp/" + name])
        self.form(node, "client", {"Client": name, "Owner": "perforce", "Root": "/p4/1/tmp/" + name,
                                  "Options": "noallwrite noclobber nocompress unlocked nomodtime normdir",
                                  "LineEnd": "unix", "View0": "//assurance/... //" + name + "/..."})

    def submit(self, node, filename, payload, *, client="fixture-commit", add=False, filetype="binary+F"):
        path = "/p4/1/tmp/" + client + "/" + filename
        if not add:
            self.p4(node, "-c", client, "edit", path)
        self.write(node, path, payload)
        if add:
            self.p4(node, "-c", client, "add", "-t", filetype, path)
        self.p4(node, "-c", client, "submit", "-d", "assurance " + filename, path)

    def create_shelf(self, node, client, filename):
        self.p4(node, "-c", client, "edit", "//assurance/" + filename)
        payload = ("shelved bytes on " + node + "\n").encode()
        self.write(node, "/p4/1/tmp/" + client + "/" + filename, payload)
        records = self.form(node, "change", {"Change": "new", "Client": client, "User": "perforce", "Status": "new", "Description": "assurance shelf " + node + "\n"}, client=client)
        found = re.search(r"Change (\d+) created", str(records))
        if not found:
            raise RuntimeError("Could not identify new fixture change")
        change = found.group(1)
        self.p4(node, "-c", client, "reopen", "-c", change, "//assurance/" + filename)
        self.p4(node, "-c", client, "shelve", "-c", change)
        self.p4(node, "-c", client, "revert", "//assurance/" + filename)
        self.shelves[node] = {"change": change, "filename": filename, "sha256": hashlib.sha256(payload).hexdigest()}

    def fixture(self):
        if "commit" in self.shelves:
            inventory = self.inventory("commit")
            if len(inventory["revisions"]) != 6 or len(inventory["contents"]) != 5:
                raise RuntimeError("Existing fixture is incomplete")
            self.expected["commit"] = inventory
            return {"revision_records": 6, "readable_revision_hashes": 5, "commit_shelf": self.shelves["commit"]}
        self.form("commit", "depot", {"Depot": "assurance", "Owner": "perforce", "Type": "local", "Map": "assurance/...", "Description": "Disposable assurance fixture\n"})
        self.client("commit", "fixture-commit")
        self.submit("commit", "alpha.txt", b"alpha revision one\n", add=True, filetype="text+C")
        self.submit("commit", "alpha.txt", b"alpha revision two\n", filetype="text+C")
        self.submit("commit", "binary.dat", bytes(range(256)) * 128, add=True)
        self.submit("commit", "space file.txt", b"whitespace filename\n", add=True, filetype="text+C")
        self.submit("commit", "deleted.txt", b"retained deleted revision\n", add=True, filetype="text+C")
        self.p4("commit", "-c", "fixture-commit", "delete", "//assurance/deleted.txt")
        self.p4("commit", "-c", "fixture-commit", "submit", "-d", "assurance delete")
        self.form("commit", "label", {"Label": "fixture-label", "Owner": "perforce", "Description": "assurance label\n", "Options": "unlocked noautoreload", "View0": "//assurance/..."})
        self.p4("commit", "tag", "-l", "fixture-label", "//assurance/...#head")
        self.form("commit", "group", {"Group": "fixture-group", "MaxResults": "unlimited", "Users0": "perforce"})
        self.create_shelf("commit", "fixture-commit", "alpha.txt")
        inventory = self.inventory("commit")
        if len(inventory["revisions"]) != 6 or len(inventory["contents"]) != 5:
            raise RuntimeError("Initial fixture is incomplete: " + repr(inventory))
        expected_hashes = {"//assurance/alpha.txt#1": hashlib.sha256(b"alpha revision one\n").hexdigest(),
                           "//assurance/alpha.txt#2": hashlib.sha256(b"alpha revision two\n").hexdigest(),
                           "//assurance/binary.dat#1": hashlib.sha256(bytes(range(256)) * 128).hexdigest(),
                           "//assurance/space file.txt#1": hashlib.sha256(b"whitespace filename\n").hexdigest(),
                           "//assurance/deleted.txt#1": hashlib.sha256(b"retained deleted revision\n").hexdigest()}
        if inventory["contents"] != expected_hashes:
            raise RuntimeError("Initial server bytes differ from independently known fixture payloads")
        self.expected["commit"] = inventory
        (self.results / "fixture-initial.json").write_text(json.dumps(inventory, indent=2) + "\n")
        return {"revision_records": 6, "readable_revision_hashes": 5, "commit_shelf": self.shelves["commit"]}

    def inventory(self, node, *, port="127.0.0.1:1666", shelves=None):
        records = self.p4(node, "files", "-a", "//assurance/...", port=port)
        revisions, contents = [], {}
        for row in records:
            if "depotFile" not in row:
                continue
            revisions.append({k: row[k] for k in ("depotFile", "rev", "action", "type")})
            if row["action"] not in ("delete", "move/delete"):
                name = row["depotFile"] + "#" + row["rev"]
                payload = self.p4(node, "print", "-q", name, raw=True, port=port).stdout
                contents[name] = hashlib.sha256(payload).hexdigest()
        specs = {}
        for command, name, keys in (("label", "fixture-label", ("Label", "Description", "Options", "View0")),
                                    ("group", "fixture-group", ("Group", "MaxResults", "Users0")),
                                    ("user", "perforce", ("User", "FullName", "Email", "Type"))):
            rows = self.p4(node, command, "-o", name, port=port)
            spec = next((r for r in rows if r.get("code") == "stat"), {})
            specs[command] = {k: spec.get(k) for k in keys}
        labels = sorted([r["depotFile"], r["rev"]] for r in self.p4(node, "files", "//assurance/...@fixture-label", port=port) if "depotFile" in r)
        shelf_data = {}
        for owner, shelf in (shelves if shelves is not None else self.shelves).items():
            if owner == "edge" and node not in ("edge", "edge-standby"):
                continue
            number = shelf["change"]
            desc = self.p4(node, "describe", "-S", "-s", number, port=port)
            payload = self.p4(node, "print", "-q", "//assurance/" + shelf["filename"] + "@=" + number, raw=True, port=port).stdout
            digest = hashlib.sha256(payload).hexdigest()
            if digest != shelf["sha256"]:
                raise RuntimeError("Shelf content hash mismatch")
            shelf_data[owner] = {"sha256": digest, "description": [{k: v for k, v in r.items() if k.startswith(("depotFile", "rev", "action", "type")) or k in ("desc", "user", "client", "status")} for r in desc if r.get("code") == "stat"]}
        return {"revisions": sorted(revisions, key=lambda r: (r["depotFile"], int(r["rev"]))),
                "contents": contents, "specs": specs, "label_members": labels, "shelves": shelf_data}

    def assert_inventory(self, node, expected, *, port="127.0.0.1:1666", shelves=None):
        actual = self.inventory(node, port=port, shelves=shelves)
        if actual != expected:
            (self.results / (node + "-mismatch.json")).write_text(json.dumps({"expected": expected, "actual": actual}, indent=2))
            raise RuntimeError("Exact fixture inventory mismatch on " + node)
        return actual

    def wait_inventory(self, node, expected, timeout=90):
        deadline = time.monotonic() + timeout
        last = ""
        while time.monotonic() < deadline:
            try:
                self.assert_inventory(node, expected)
                return
            except RuntimeError as error:
                last = str(error)
                time.sleep(1)
        raise RuntimeError("Replication did not converge on %s: %s" % (node, last))

    def login(self, node, port="127.0.0.1:1666", user="perforce"):
        self.p4(node, "login", "-a", port=port, user=user, raw=True, data=(self.password + "\n").encode())

    def copy_between(self, source, path, target, destination):
        with tempfile.TemporaryDirectory(prefix="sdp-topology-transfer-") as tmp:
            local = Path(tmp) / "payload"
            self.cmd(self.docker + ["cp", self.name(source) + ":" + path, str(local)])
            self.cmd(self.docker + ["cp", str(local), self.name(target) + ":" + destination])
            self.execute(target, ["chown", "-R", "perforce:perforce", destination], user="root")

    def latest_checkpoint(self, node):
        query = "import pathlib,re,json; p=pathlib.Path('/p4/1/checkpoints'); a=[(int(re.fullmatch(r'p4_1.ckp.([0-9]+)(?:.gz)?',x.name)[1]),str(x),x.is_dir()) for x in p.iterdir() if re.fullmatch(r'p4_1.ckp.([0-9]+)(?:.gz)?',x.name)]; print(json.dumps(sorted(a)[-1]))"
        _, path, parallel = json.loads(self.execute(node, ["python3", "-c", query]).stdout)
        return path, parallel

    def replication(self):
        for script, node, target in (("mkedge.sh", "edge", "commit"), ("mkstandby.sh", "standby", "commit"), ("mkedge-standby.sh", "edge-standby", "edge")):
            args = ["1", node, self.password, target + ":1666"]
            if script != "mkedge.sh":
                args.append(target)
            self.script("commit", script, *args)
            spec = next(r for r in self.p4("commit", "server", "-o", node) if r.get("code") == "stat")
            spec.pop("code", None)
            spec["Address"] = node + ":1666"
            if node == "edge":
                spec["ExternalAddress"] = "edge:1666"
            self.form("commit", "server", spec)
        self.script("commit", "live_checkpoint.sh")
        checkpoint, parallel = self.latest_checkpoint("commit")
        for node in ("edge", "standby"):
            self.copy_between("commit", checkpoint, node, "/p4/1/tmp/seed.gz")
            if node == "standby":
                self.copy_between("commit", checkpoint + ".md5", node, "/p4/1/tmp/seed.gz.md5")
                self.script(node, "load_checkpoint.sh", "/p4/1/tmp/seed.gz", "-i", "1", "-s", node, "-t", "ha", "-l", "-y", "-no_start")
            else:
                self.execute(node, ["/p4/1/bin/p4d_1", "-r", "/p4/1/root", "-Redge-server", "-z", *(["-jrp", "-N", "4"] if parallel else ["-jr"]), "/p4/1/tmp/seed.gz"], timeout=300)
            self.script(node, "p4login", "1", "-service")
            self.service(node, "start")
            self.login(node)
            self.wait_inventory(node, self.expected["commit"])
        self.client("edge", "fixture-edge")
        self.p4("edge", "-c", "fixture-edge", "sync", "//assurance/...")
        self.create_shelf("edge", "fixture-edge", "space file.txt")
        self.p4("edge", "-c", "fixture-edge", "edit", "//assurance/binary.dat")
        self.write("edge", "/p4/1/tmp/fixture-edge/binary.dat", b"unsubmitted edge working bytes\n")
        self.expected["edge"] = self.inventory("edge")
        # Seed edge standby from an actual edge checkpoint, preserving its local shelf tables.
        self.execute("edge", ["/p4/1/bin/p4d_1", "-r", "/p4/1/root", "-jd", "-z", "/p4/1/tmp/edge-seed.gz"])
        checkpoint = "/p4/1/tmp/edge-seed.gz"
        self.copy_between("edge", checkpoint, "edge-standby", "/p4/1/tmp/seed.gz")
        self.execute("edge-standby", ["/p4/1/bin/p4d_1", "-r", "/p4/1/root", "-z", "-jr", "/p4/1/tmp/seed.gz"], timeout=300)
        self.script("edge-standby", "p4login", "1", "-service")
        self.service("edge-standby", "start")
        self.login("edge-standby")
        self.wait_inventory("edge-standby", self.expected["edge"])
        # Distinct new submit proves pull/journalcopy is active after the seed.
        self.submit("commit", "replicated.txt", b"post-seed replication bytes\n", add=True)
        self.expected["commit"] = self.inventory("commit")
        self.wait_inventory("standby", self.expected["commit"])
        expected_edge = dict(self.expected["commit"], shelves=self.expected["edge"]["shelves"])
        self.wait_inventory("edge", expected_edge)
        self.wait_inventory("edge-standby", expected_edge)
        self.expected["edge"] = expected_edge
        for node in ("edge", "standby", "edge-standby"):
            self.p4(node, "verify", "-q", "//assurance/...", raw=True)
            # Archive revisions include their change number; derive the one
            # physical full-file revision rather than assuming a 1.1 name.
            query = "import pathlib,hashlib; files=list(pathlib.Path('/p4/1/depots/assurance/binary.dat,d').glob('*')); assert len(files)==1 and files[0].is_file(); print(hashlib.sha256(files[0].read_bytes()).hexdigest())"
            physical = self.execute(node, ["python3", "-c", query]).stdout.decode().strip()
            if physical != self.expected["commit"]["contents"]["//assurance/binary.dat#1"]:
                raise RuntimeError("Independent physical archive differs on " + node)
        return {"topology": "commit -> edge, standby; edge -> edge-standby", "independent_archive_hashes": True, "post_seed_revision": "//assurance/replicated.txt#1", "edge_shelf": self.shelves["edge"]}

    def observe_nodes(self):
        for node, (role, _) in NODES.items():
            daemon = "p4broker" if role == "broker" else "p4p" if role == "proxy" else "p4d"
            info = next(r for r in self.p4(node, "info") if r.get("code") == "stat")
            expected_services = {"commit": "commit-server", "edge": "edge-server", "standby": "forwarding-standby", "edge-standby": "forwarding-standby"}
            if node in expected_services and (info.get("serverServices") != expected_services[node] or info.get("serverID", info.get("ServerID")) != node):
                raise RuntimeError("Actual role/identity differs from the requested node " + node)
            if node == "broker" and not info.get("brokerVersion"):
                raise RuntimeError("Broker endpoint did not identify an actual broker")
            if node == "proxy" and not info.get("proxyVersion"):
                raise RuntimeError("Proxy endpoint did not identify an actual proxy")
            pid = self.execute(node, ["systemctl", "show", daemon + "_1.service", "-p", "MainPID", "--value"], user="root").stdout.decode().strip()
            if not pid.isdigit() or int(pid) <= 0:
                raise RuntimeError("No actual service process for " + node)
            command = self.execute(node, ["cat", "/proc/" + pid + "/cmdline"], user="root").stdout.replace(b"\0", b" ").decode()
            if daemon == "p4d" and "-J /p4/1/journals/journal" not in command:
                raise RuntimeError("Running journal location differs on " + node)
            container = json.loads(self.cmd(self.docker + ["inspect", self.name(node)]).stdout)[0]
            running_image = container["Image"]
            if running_image != self.image_evidence["node_id"]:
                raise RuntimeError("Running node image differs from the recorded image")
            cgroup_namespace = container["HostConfig"].get("CgroupnsMode")
            if cgroup_namespace != "private":
                raise RuntimeError("Node does not have its own cgroup namespace: " + node)
            source_hash = self.execute(node, ["sha256sum", "/inputs/sdp/.export-manifest.json"]).stdout.decode().split()[0]
            if source_hash != hashlib.sha256((self.source / ".export-manifest.json").read_bytes()).hexdigest():
                raise RuntimeError("Copied installation source differs on " + node)
            self.node_evidence[node] = {"role": role, "service": daemon, "main_pid": int(pid), "command": command,
                                        "running_image_id": running_image, "source_manifest_sha256": source_hash,
                                        "cgroup_namespace_mode": cgroup_namespace,
                                        "server_id": info.get("serverID", info.get("ServerID")), "services": info.get("serverServices"),
                                        "journal": "/p4/1/journals/journal" if daemon == "p4d" else None,
                                        "proxy_version": info.get("proxyVersion"), "broker_version": info.get("brokerVersion")}

    def routing(self):
        config = '''target = commit:1666;
listen = 1666;
directory = /p4/1/logs;
logfile = /p4/1/logs/p4broker.log;
debug-level = server=1;
admin-name = "Disposable assurance";
admin-phone = "none";
admin-email = "disposable@localhost";
'''
        self.write("broker", "/p4/common/config/p4_1.broker.commit.cfg", config)
        self.service("broker", "start", "p4broker")
        self.service("proxy", "start", "p4p")
        for node in ("broker", "proxy"):
            self.login(node)
            self.assert_inventory(node, self.expected["commit"])
            self.p4(node, "counter", "assurance.route." + node, "314159")
            rows = self.p4("commit", "counter", "assurance.route." + node)
            if not any(r.get("value") == "314159" for r in rows):
                raise RuntimeError("Route did not carry a write to the commit")
            self.service(node, "restart", "p4broker" if node == "broker" else "p4p")
            self.assert_inventory(node, self.expected["commit"])
        self.client("proxy", "fixture-proxy")
        self.p4("proxy", "-c", "fixture-proxy", "sync", "//assurance/binary.dat")
        payload = self.execute("proxy", ["cat", "/p4/1/tmp/fixture-proxy/binary.dat"]).stdout
        if hashlib.sha256(payload).hexdigest() != self.expected["commit"]["contents"]["//assurance/binary.dat#1"]:
            raise RuntimeError("Proxy sync changed the submitted bytes")
        files = self.shell("proxy", "find -L /p4/1/cache -type f | wc -l").stdout.strip()
        if int(files) == 0:
            raise RuntimeError("Proxy reads did not populate independent cache storage")
        self.observe_nodes()
        return {"routes": ["broker", "proxy"], "read_hashes_specs_shelves_equal": True, "writes_forwarded": True, "proxy_cache_files": int(files)}

    def workspace_snapshot(self, node, client=None, *, port="127.0.0.1:1666"):
        client = client or "fixture-" + node
        spec = next(r for r in self.p4(node, "client", "-o", client, port=port) if r.get("code") == "stat")
        selected = {k: v for k, v in spec.items() if k in ("Client", "Owner", "Root", "Options", "LineEnd", "ServerID") or k.startswith("View")}
        have = [{k: v for k, v in r.items() if k in ("depotFile", "clientFile", "haveRev")} for r in self.p4(node, "-c", client, "have", "//assurance/...", port=port) if r.get("code") == "stat"]
        opened = [{k: v for k, v in r.items() if k in ("depotFile", "clientFile", "rev", "action", "type", "change", "client", "user")} for r in self.p4(node, "-c", client, "opened", "//assurance/...", port=port) if r.get("code") == "stat"]
        return {"client_spec": selected, "have": sorted(have, key=lambda r: r["depotFile"]), "opened": sorted(opened, key=lambda r: r["depotFile"])}

    def recovery(self):
        downstream_before = None
        for node in ("commit", "edge"):
            if node == "edge":
                # The commit's post-recovery submit must reach the edge before
                # taking its comparison snapshot or a healthy pull looks like
                # an unexpected revision introduced by recovery.
                expected_edge = dict(self.inventory("commit"), shelves=self.inventory("edge")["shelves"])
                self.wait_inventory("edge", expected_edge)
                self.wait_inventory("edge-standby", expected_edge)
                downstream_before = self.p4("edge-standby", "journalcopy", "-l", raw=True).stdout.decode()
                # Rebuilding an edge from a commit checkpoint resets its
                # journal history in both deployed and candidate scripts.
                # Fence its dependent standby before this discontinuity, then
                # explicitly reseed it below; do not claim uninterrupted pull.
                self.service("edge-standby", "stop")
            expected = self.inventory(node)
            workspace = self.workspace_snapshot(node)
            before = self.execute(node, ["readlink", "-f", "/p4/1/root"]).stdout.strip()
            if node == "commit":
                self.script(node, "live_checkpoint.sh")
                self.script(node, "recreate_offline_db.sh")
                self.script(node, "recreate_db_checkpoint.sh")
            else:
                self.script("commit", "live_checkpoint.sh")
                checkpoint, _ = self.latest_checkpoint("commit")
                self.copy_between("commit", checkpoint, node, "/p4/1/tmp/master-checkpoint")
                self.copy_between("commit", checkpoint + ".md5", node, "/p4/1/tmp/master-checkpoint.md5")
                self.script(node, "recover_edge.sh", "1", "/p4/1/tmp/master-checkpoint")
            after = self.execute(node, ["readlink", "-f", "/p4/1/root"]).stdout.strip()
            if before == after:
                raise RuntimeError("Database roots did not switch on " + node)
            self.assert_inventory(node, expected)
            if self.workspace_snapshot(node) != workspace:
                raise RuntimeError("Recovery lost client/have/opened metadata on " + node)
            self.submit(node, "after-recovery-" + node + ".txt", (node + " new transaction after recovery\n").encode(), client="fixture-" + node, add=True)
            expected = self.inventory(node)
            self.expected[node] = expected
            if node == "edge":
                # Exercise normal maintenance after the nested recovery stage,
                # then recover again to expose any ancestor-cleanup regression.
                self.script(node, "daily_checkpoint.sh")
                self.script(node, "recreate_db_checkpoint.sh")
                self.assert_inventory(node, expected)
                self.script("commit", "live_checkpoint.sh")
                checkpoint, _ = self.latest_checkpoint("commit")
                self.copy_between("commit", checkpoint, node, "/p4/1/tmp/second-master-checkpoint")
                self.copy_between("commit", checkpoint + ".md5", node, "/p4/1/tmp/second-master-checkpoint.md5")
                self.script(node, "recover_edge.sh", "1", "/p4/1/tmp/second-master-checkpoint")
                self.assert_inventory(node, expected)
        self.expected["commit"] = self.inventory("commit")
        self.wait_inventory("standby", self.expected["commit"])
        self.expected["edge"] = self.inventory("edge")
        checkpoint = "/p4/1/tmp/edge-post-recovery.gz"
        self.execute("edge", ["/p4/1/bin/p4d_1", "-r", "/p4/1/root", "-jd", "-z", checkpoint])
        self.copy_between("edge", checkpoint, "edge-standby", checkpoint)
        self.copy_between("edge", checkpoint + ".md5", "edge-standby", checkpoint + ".md5")
        self.script("edge-standby", "load_checkpoint.sh", checkpoint, "-i", "1", "-s", "edge-standby", "-t", "ha", "-l", "-y", "-no_start")
        self.script("edge-standby", "p4login", "1", "-service")
        self.service("edge-standby", "start")
        self.login("edge-standby")
        self.wait_inventory("edge-standby", self.expected["edge"])
        if self.workspace_snapshot("edge-standby", "fixture-edge") != self.workspace_snapshot("edge"):
            raise RuntimeError("Reseeded edge standby lost edge-local client/have/opened metadata")
        self.p4("edge-standby", "verify", "-q", "//assurance/...", raw=True)
        self.submit("edge", "after-standby-reseed.txt", b"replication after explicit dependent standby reseed\n", client="fixture-edge", add=True)
        self.expected["edge"] = self.inventory("edge")
        self.wait_inventory("edge-standby", self.expected["edge"])
        self.expected["commit"] = self.inventory("commit")
        self.wait_inventory("standby", self.expected["commit"])
        return {"roles": ["commit", "edge"], "checks": "all fixture revision hashes, label membership, user/group/client specs, have/opened metadata, commit and edge shelf hashes, post-recovery submits",
                "downstream_edge_standby": {"uninterrupted_replication": False, "operator_sequence": "stop dependent standby before upstream edge rebuild; load fresh edge checkpoint; start; verify full inventory and a new replicated submit", "prior_journalcopy_state": downstream_before, "reseed_and_post_reseed_write_verified": True}}

    def restore_server(self, node, root, expected, *, shelves=None, workspace=None):
        """Serve restored DBs on a different loopback port and read exact state."""
        self.write(node, root + "/server.id", "assurance-restore\n")
        self.execute(node, ["/p4/1/bin/p4d_1", "-r", root, "-p", "127.0.0.1:1668", "-J", "off", "-L", root + "/log", "--daemonsafe", "--pid-file=" + root + "/pid"])
        try:
            self.login(node, "127.0.0.1:1668")
            self.assert_inventory(node, expected, port="127.0.0.1:1668", shelves=shelves)
            if workspace is not None and self.workspace_snapshot(node, port="127.0.0.1:1668") != workspace:
                raise RuntimeError("Restored database lost client/have/opened metadata")
        finally:
            pid = self.execute(node, ["cat", root + "/pid"]).stdout.decode().strip()
            self.p4(node, "admin", "stop", port="127.0.0.1:1668", check=False)
            wait = "import pathlib,time,sys; p=pathlib.Path('/proc')/sys.argv[1]; end=time.monotonic()+30\nwhile p.exists() and time.monotonic()<end: time.sleep(.05)\nassert not p.exists(), 'Read-only restored probe did not stop'"
            self.execute(node, ["python3", "-c", wait, pid], timeout=35)

    def retained_generation_replay(self):
        node = "commit"
        for number in range(3):
            self.submit(node, "retention-%d.txt" % number, ("retained generation %d\n" % number).encode(), add=True)
            self.script(node, "live_checkpoint.sh")
        # A final transaction exists only in a rotated journal, not any retained checkpoint.
        self.submit(node, "journal-only.txt", b"must replay journals to find this revision\n", add=True)
        self.script(node, "rotate_journal.sh")
        expected = self.inventory(node)
        workspace = self.workspace_snapshot(node)
        self.expected[node] = expected
        query = "import pathlib,re,json; p=pathlib.Path('/p4/1/checkpoints'); print(json.dumps(sorted([(int(m[1]),str(x),x.is_dir()) for x in p.iterdir() if (m:=re.fullmatch(r'p4_1.ckp.([0-9]+)(?:.gz)?',x.name))])))"
        checkpoints = json.loads(self.execute(node, ["python3", "-c", query]).stdout)
        if len(checkpoints) != 2:
            raise RuntimeError("KEEPCKPS=2 did not retain exactly two complete generations")
        active = int(next(r["value"] for r in self.p4(node, "counter", "journal") if "value" in r))
        replayed = []
        for number, path, parallel in checkpoints:
            root = "/p4/1/tmp/retained-restore-" + str(number)
            self.execute(node, ["mkdir", root])
            self.execute(node, ["/p4/1/bin/p4d_1", "-r", root, "-z", *(["-jrp", "-N", "4"] if parallel else ["-jr"]), path], timeout=300)
            journals = []
            for journal in range(number, active):
                paths = ["/p4/1/" + directory + "/p4_1.jnl." + str(journal) + suffix for directory in ("checkpoints", "journals.rotated") for suffix in ("", ".gz")]
                found = next((p for p in paths if self.execute(node, ["test", "-f", p], check=False).returncode == 0), None)
                if found is None:
                    raise RuntimeError("Retained checkpoint %s cannot reach current state: journal %s missing" % (number, journal))
                self.execute(node, ["/p4/1/bin/p4d_1", "-r", root, *(["-z"] if found.endswith(".gz") else []), "-jr", found], timeout=300)
                journals.append(found)
            self.restore_server(node, root, expected, shelves={"commit": self.shelves["commit"]}, workspace=workspace)
            replayed.append({"checkpoint": path, "journals": journals, "all_revision_and_shelf_hashes_equal": True})
        return {"keepckps": 2, "keepjnls": 1, "restored_every_retained_generation": replayed, "journal_only_revision_verified": True}

    def missing_archive_repair(self):
        # Disable automatic archive transfers, while metadata/journalcopy remains
        # real, so only rsync_missing.sh can repair this deliberately hidden file.
        self.wait_inventory("standby", self.expected["commit"])
        for number in range(3, 13):
            self.p4("commit", "configure", "unset", "standby#startup." + str(number))
        self.service("standby", "restart")
        self.login("standby")
        archive = "/p4/1/depots/assurance/binary.dat,d"
        source_hash = self.shell("commit", "find " + archive + " -type f -exec sha256sum {} +").stdout
        self.execute("standby", ["mv", archive, archive + ".fixture-quarantine"])
        missing = self.p4("standby", "-s", "verify", "-q", "//assurance/binary.dat#1", raw=True, check=False)
        report = missing.stdout + missing.stderr
        if missing.returncode == 0 or b"MISSING!" not in report:
            raise RuntimeError("Actual verify did not detect the removed destination archive")
        self.write("standby", "/p4/1/logs/verify_output.log", report)
        self.execute("standby", ["mkdir", "-p", "/home/perforce/.ssh"])
        self.execute("standby", ["chmod", "700", "/home/perforce/.ssh"])
        self.execute("standby", ["ssh-keygen", "-q", "-t", "ed25519", "-N", "", "-f", "/home/perforce/.ssh/topology"])
        public = self.execute("standby", ["cat", "/home/perforce/.ssh/topology.pub"]).stdout
        self.write("commit", "/home/perforce/.ssh/authorized_keys", public, mode=0o600)
        self.execute("commit", ["chmod", "700", "/home/perforce/.ssh"])
        self.execute("commit", ["ssh-keygen", "-A"], user="root")
        # Use an explicit fixture configuration/port: older RHEL sshd configs
        # need not include sshd_config.d, and a distro sshd may already run.
        ssh_config = "/p4/1/tmp/topology_sshd_config"
        self.write("commit", ssh_config, "Port 2222\nListenAddress 0.0.0.0\nHostKey /etc/ssh/ssh_host_ed25519_key\nPidFile /run/sdp-topology-sshd.pid\nAuthorizedKeysFile .ssh/authorized_keys\nPasswordAuthentication no\nAuthenticationMethods publickey\nUsePAM yes\nAllowUsers perforce\n", user="root")
        self.execute("commit", ["/usr/sbin/sshd", "-f", ssh_config, "-E", "/p4/1/logs/sshd-assurance.log"], user="root")
        self.write("standby", "/home/perforce/.ssh/config", "Host commit\n  Port 2222\n  IdentityFile ~/.ssh/topology\n  IdentitiesOnly yes\n  BatchMode yes\n", mode=0o600)
        self.script("standby", "rsync_missing.sh")
        destination_hash = self.shell("standby", "find " + archive + " -type f -exec sha256sum {} +").stdout
        if destination_hash != source_hash:
            raise RuntimeError("Transferred archive hashes differ from independent source")
        repaired = self.p4("standby", "-s", "verify", "-q", "//assurance/binary.dat#1", raw=True, check=False)
        if repaired.returncode or b"MISSING!" in repaired.stdout + repaired.stderr or b"BAD!" in repaired.stdout + repaired.stderr:
            raise RuntimeError("Repaired archive did not pass a fresh server verify")
        self.assert_inventory("standby", self.expected["commit"])
        # Restore one real archive worker before later fault/promotion writes.
        self.p4("commit", "configure", "set", "standby#startup.3=pull -u -i 1")
        self.service("standby", "restart")
        return {"source_node": "commit", "destination_node": "standby", "destination_absent_before_transfer": True,
                "actual_verify_failure": report.decode(errors="replace"), "source_and_destination_archive_sha256": source_hash.decode(), "fresh_verify_passed": True}

    def preserve_failed_switch(self, *, interrupt=False):
        node = "commit"
        self.script(node, "live_checkpoint.sh")
        # Retention and an earlier deliberately interrupted switch can leave
        # the offline database stale/unusable. Prepare a fresh candidate before
        # injecting a service failure, so the fault is at the intended boundary.
        self.script(node, "recreate_offline_db.sh")
        expected = self.inventory(node)
        workspace = self.workspace_snapshot(node)
        before = self.execute(node, ["readlink", "-f", "/p4/1/root"]).stdout.decode().strip()
        fixture = "/etc/systemd/system/p4d_1.service.d/assurance-failure.conf"
        if interrupt:
            self.write(node, fixture, "[Service]\nExecStartPre=/bin/sh -c 'touch /p4/1/tmp/assurance-restart-boundary; sleep 120'\n", user="root")
        else:
            self.write(node, fixture, "[Service]\nExecStartPre=/bin/false\n", user="root")
        self.execute(node, ["systemctl", "daemon-reload"], user="root")
        marker = "/p4/1/logs/ckp_running.txt"
        try:
            if interrupt:
                # Isolate the real script in a new session; kill its exact PID,
                # never a broad name match. The service blocks after the actual
                # DB swap, leaving both old saved DB and replacement available.
                launch = "setsid bash -c 'echo $$ > /p4/1/tmp/recovery-fixture.pid; exec bash /p4/common/bin/recreate_db_checkpoint.sh 1' > /p4/1/tmp/recovery-interrupt.log 2>&1 < /dev/null &"
                self.shell(node, launch)
                deadline = time.monotonic() + 180
                while time.monotonic() < deadline:
                    if self.execute(node, ["test", "-f", "/p4/1/tmp/assurance-restart-boundary"], check=False, user="root").returncode == 0:
                        break
                    time.sleep(0.25)
                else:
                    raise RuntimeError("Real recovery did not reach the restart boundary")
                pid = self.execute(node, ["cat", "/p4/1/tmp/recovery-fixture.pid"]).stdout.decode().strip()
                if not pid.isdigit():
                    raise RuntimeError("Invalid fixture process ID")
                self.execute(node, ["kill", "-KILL", "--", "-" + pid], user="root")
                self.service(node, "stop", check=False)
                failed_status = "SIGKILL of the actual recovery session after physical DB switch"
            else:
                result = self.script(node, "recreate_db_checkpoint.sh", check=False)
                if result.returncode == 0:
                    raise RuntimeError("Deliberate real service restart failure was reported as success")
                failed_status = result.returncode
            marker_exists = self.execute(node, ["test", "-e", marker], check=False).returncode == 0
            if interrupt and not marker_exists:
                raise RuntimeError("SIGKILL recovery unexpectedly released its lock without executing die")
            if not interrupt and marker_exists:
                raise RuntimeError("Handled recovery failure did not release its owned checkpoint lock")
            current = self.execute(node, ["readlink", "-f", "/p4/1/root"]).stdout.decode().strip()
            if current == before:
                raise RuntimeError("Fault was not injected after a real database swap")
            self.execute(node, ["test", "-s", before + "/save/db.counters"])
            # Prove the saved original is readable using a separate temporary
            # copy. Never modify the preserved rollback evidence itself.
            root = "/p4/1/tmp/" + ("interrupted" if interrupt else "failed") + "-saved-restore"
            self.execute(node, ["mkdir", root])
            self.shell(node, "cp " + shlex.quote(before) + "/save/db.* " + shlex.quote(root) + "/")
            self.restore_server(node, root, expected, shelves={"commit": self.shelves["commit"]}, workspace=workspace)
        finally:
            self.execute(node, ["rm", "-f", fixture], user="root")
            self.execute(node, ["systemctl", "daemon-reload"], user="root")
            self.execute(node, ["systemctl", "reset-failed", "p4d_1.service"], user="root", check=False)
        # Explicit operator-style continuation. This is not an automatic
        # rollback guarantee: original saved data and failure logs stay retained.
        self.service(node, "start")
        self.assert_inventory(node, expected)
        if self.workspace_snapshot(node) != workspace:
            raise RuntimeError("Replacement database lost client/have/opened metadata after the injected failure")
        if interrupt:
            self.execute(node, ["mv", marker, marker + ".interrupted-evidence"])
        self.submit(node, "post-fault-" + ("interrupt" if interrupt else "restart") + ".txt", b"write after operator recovery\n", add=True)
        self.expected[node] = self.inventory(node)
        return {"injected_failure": failed_status, "old_saved_database_inventory_equal": True,
                "replacement_database_inventory_equal": True, "owned_checkpoint_lock_released_by_die": not interrupt,
                "checkpoint_lock_retained_after_sigkill": interrupt,
                "post_recovery_submit": True, "automatic_rollback_claimed": False}

    def standby_promotion(self):
        self.wait_inventory("standby", self.expected["commit"])
        expected = self.inventory("standby")
        # Fence the actual old primary. Keep it stopped for the entire promotion
        # and the post-promotion submit; no DNS/host service is touched.
        self.service("commit", "stop")
        self.service("commit", "is-active", check=False)
        if self.p4("commit", "info", raw=True, check=False).returncode == 0:
            raise RuntimeError("Old commit remained reachable after fencing")
        self.script("standby", "convert_standby_to_master.sh")
        identity = self.execute("standby", ["cat", "/p4/1/root/server.id"]).stdout.decode().strip()
        if identity != "commit":
            raise RuntimeError("Promotion did not install commit identity")
        info = next(r for r in self.p4("standby", "info") if r.get("code") == "stat")
        if info.get("serverServices") != "commit-server" or info.get("serverID", info.get("ServerID")) != "commit":
            raise RuntimeError("Promoted endpoint did not run as the commit server")
        self.assert_inventory("standby", expected)
        self.submit("standby", "post-promotion.txt", b"write on promoted standby\n", add=True)
        self.p4("standby", "-s", "verify", "-q", "//assurance/...", raw=True)
        return {"old_commit_fenced": True, "promoted_identity": identity, "promoted_services": info["serverServices"], "inventory_equal_before_new_write": True,
                "post_promotion_revision": "//assurance/post-promotion.txt#1", "verify_passed": True}

    def collect(self):
        for node in NODES:
            target = self.results / node
            target.mkdir(exist_ok=True)
            self.cmd(self.docker + ["cp", self.name(node) + ":/results/.", str(target)], check=False)
            (target / "logs").mkdir(exist_ok=True)
            self.cmd(self.docker + ["cp", self.name(node) + ":/p4/1/logs/.", str(target / "logs")], check=False)

    def cleanup(self):
        for node in NODES:
            result = self.cmd(self.docker + ["inspect", self.name(node)], check=False)
            if result.returncode:
                continue
            obj = json.loads(result.stdout)[0]
            if obj["Config"].get("Labels", {}).get(LABEL) != self.args.prefix:
                raise RuntimeError("Refusing unowned container " + self.name(node))
            self.cmd(self.docker + ["rm", "-f", self.name(node)])
        result = self.cmd(self.docker + ["network", "inspect", self.network], check=False)
        if result.returncode == 0:
            obj = json.loads(result.stdout)[0]
            if obj.get("Labels", {}).get(LABEL) != self.args.prefix:
                raise RuntimeError("Refusing unowned network")
            self.cmd(self.docker + ["network", "rm", self.network])

    def run(self):
        if self.args.cleanup:
            self.cleanup()
            return 0
        operations = [
            ("six_node_install", ["mkdirs.sh", "configure_new_server.sh", "p4d_base"], self.provision),
            ("exact_fixture", [], self.fixture),
            ("replication_and_local_shelves", ["mkedge.sh", "mkstandby.sh", "mkedge-standby.sh", "p4login", "load_checkpoint.sh"], self.replication),
            ("broker_proxy_routes", ["p4broker_base", "p4p_base"], self.routing),
            ("checkpoint_database_switch", ["live_checkpoint.sh", "recreate_offline_db.sh", "recreate_db_checkpoint.sh", "recover_edge.sh", "load_checkpoint.sh", "backup_functions.sh"], self.recovery),
            ("retained_generation_replay", ["live_checkpoint.sh", "rotate_journal.sh", "backup_functions.sh"], self.retained_generation_replay),
            ("missing_archive_repair", ["rsync_missing.sh"], self.missing_archive_repair),
            ("restart_failure_preserves_data", ["recreate_db_checkpoint.sh", "backup_functions.sh", "p4d_base"], self.preserve_failed_switch),
            ("interrupted_recovery_preserves_data", ["recreate_db_checkpoint.sh", "backup_functions.sh"], lambda: self.preserve_failed_switch(interrupt=True)),
            ("standby_promotion", ["convert_standby_to_master.sh", "backup_functions.sh"], self.standby_promotion),
        ]
        try:
            for name, scripts, operation in operations:
                if any(r["name"] == name for r in self.records):
                    continue
                if not self.scenario(name, scripts, operation):
                    for pending, paths, _ in operations[len(self.records):]:
                        self.records.append({"name": pending, "scripts": paths, "required": True, "status": "blocked", "error": "Earlier required scenario failed"})
                    break
        finally:
            self.save()
            self.collect()
        return int(any(row["status"] != "passed" for row in self.records))


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--source", required=True)
    parser.add_argument("--binaries", required=True)
    parser.add_argument("--results", required=True)
    parser.add_argument("--prefix", required=True)
    parser.add_argument("--image", default="sdp-assurance-ubuntu:26.04")
    parser.add_argument("--docker", default="/opt/homebrew/bin/docker")
    parser.add_argument("--confirm-disposable", action="store_true")
    parser.add_argument("--cleanup", action="store_true")
    parser.add_argument("--resume", action="store_true", help="Resume only this labelled fixture after investigating a failure")
    args = parser.parse_args()
    try:
        return Topology(args).run()
    except (OSError, ValueError, RuntimeError, subprocess.SubprocessError) as error:
        print(json.dumps({"guard_or_runner_error": str(error)}), flush=True)
        return 2


if __name__ == "__main__":
    raise SystemExit(main())