pigway-frame-flow/scripts/production-deploy.py

"""Opt-in production deployment; host paths verified against the running container."""

import json
import os
import re
import signal
import subprocess
import tempfile
import time
from datetime import UTC, datetime
from pathlib import Path
from uuid import uuid4

import yaml

ROOT = Path(__file__).resolve().parents[1]
REQUIRED = (
    "PROD_PROJECT",
    "PROD_CONTAINER",
    "PROD_IMAGE_REPO",
    "PROD_DATA_DIR",
    "PROD_CONFIG_DIR",
    "PROD_WEB_BIND",
    "PROD_WEB_PORT",
    "PROD_STATE_VOLUME",
    "PROD_MOCK",
    "PROD_WRITE_ENABLED",
)


class Deploy:
    def __init__(self):
        self.env = os.environ.copy()
        self.owner = self.env.get("FRAME_FLOW_DEPLOY_OWNER") or uuid4().hex
        if not re.fullmatch(r"[0-9a-f]{32}", self.owner):
            raise ValueError("Invalid deployment owner")
        self.lock_id = None
        self.state = None
        self.old_image = None
        self.secret_values = [
            self.env[key]
            for key in (
                "PROD_DATA_DIR",
                "PROD_CONFIG_DIR",
                "PROD_PHOTOS_DIR",
                "PROD_WEB_BIND",
                "SYNOLOGY_PHOTOS_URL",
                "SYNOLOGY_PHOTOS_CERT_SHA256",
            )
            if self.env.get(key)
        ]
        self.docker = self.env.get("DOCKER_BIN", "docker")

    def run(self, *args, data=None, check=True):
        result = subprocess.run(
            [self.docker, *args],
            input=data,
            capture_output=True,
            env=self.env | {"DOCKER_BUILDKIT": "1"},
        )
        if check and result.returncode:
            # Never print Docker inspect/config/env output or secret-bearing commands.
            raise RuntimeError(
                f"Docker {args[0]} failed (exit {result.returncode}); inspect scoped container logs privately"
            )
        return result

    def inspect(self, name):
        return json.loads(self.run("inspect", name).stdout)[0]

    def settings(self):
        missing = [key for key in REQUIRED if not self.env.get(key)]
        if missing:
            raise ValueError("Missing Variables: " + ", ".join(missing))
        for key in (
            "PROD_PROJECT",
            "PROD_CONTAINER",
            "PROD_IMAGE_REPO",
            "PROD_STATE_VOLUME",
        ):
            if not re.fullmatch(r"[a-z][a-z0-9-]{2,80}", self.env[key]) or self.env[
                key
            ].startswith("ffci-"):
                raise ValueError("Unsafe production resource name: " + key)
        for key in ("PROD_DATA_DIR", "PROD_CONFIG_DIR"):
            if not self.env[key].startswith("/") or self.env[key] == "/":
                raise ValueError("Absolute existing NAS path required: " + key)
        if self.env["PROD_DATA_DIR"] == self.env["PROD_CONFIG_DIR"]:
            raise ValueError("data/config must use distinct directories")
        if any(
            self.env[key] not in ("true", "false")
            for key in ("PROD_MOCK", "PROD_WRITE_ENABLED")
        ):
            raise ValueError("Explicit true/false required for mock and writes")
        if not 1 <= int(self.env["PROD_WEB_PORT"]) <= 65535:
            raise ValueError("Invalid production port")

    def preflight(self, require_running=True):
        self.settings()
        old = self.inspect(self.env["PROD_CONTAINER"])
        labels = old["Config"].get("Labels") or {}
        if (
            labels.get("com.docker.compose.project") != self.env["PROD_PROJECT"]
            or labels.get("com.docker.compose.service") != "pigway-frame-flow"
        ):
            raise ValueError(
                "Existing container does not belong to the configured production project/service"
            )
        if require_running and not old["State"]["Running"]:
            raise ValueError("Existing production container must be running")
        mounts = {m["Destination"]: m for m in old["Mounts"]}
        for target, key in (("/data", "PROD_DATA_DIR"), ("/config", "PROD_CONFIG_DIR")):
            m = mounts.get(target, {})
            if (
                m.get("Type") != "bind"
                or m.get("Source") != self.env[key]
                or not m.get("RW")
            ):
                raise ValueError("Production mount mismatch: " + key)
        photos = self.env.get("PROD_PHOTOS_DIR")
        if "/photos" in mounts and (
            not photos
            or mounts["/photos"]["Source"] != photos
            or mounts["/photos"]["RW"]
        ):
            raise ValueError(
                "Confirm PROD_PHOTOS_DIR for the existing read-only photo mount"
            )
        if photos and "/photos" not in mounts:
            raise ValueError(
                "Configured photos directory does not match current container"
            )
        if set(mounts) != {"/data", "/config"} | ({"/photos"} if photos else set()):
            raise ValueError("Unrecognized persistent mount; review before adopting CD")
        bindings = old["HostConfig"].get("PortBindings", {}).get("18787/tcp")
        if bindings != [
            {"HostIp": self.env["PROD_WEB_BIND"], "HostPort": self.env["PROD_WEB_PORT"]}
        ]:
            raise ValueError(
                "Port/bind mismatch; CD must not change production networking"
            )
        if (
            old["HostConfig"].get("NetworkMode")
            != self.env["PROD_PROJECT"] + "_default"
        ):
            raise ValueError("Custom production network requires explicit review")
        config = yaml.safe_load((ROOT / "docker-compose.yml").read_text())
        config["name"] = self.env["PROD_PROJECT"]
        service = config["services"]["pigway-frame-flow"]
        service.pop("build")
        service.pop("env_file")
        service["container_name"] = self.env["PROD_CONTAINER"]
        service["image"] = old["Image"]
        service["environment"] = dict(
            value.split("=", 1) for value in old["Config"]["Env"] if "=" in value
        )
        self.secret_values.extend(
            value
            for key, value in service["environment"].items()
            if value
            and any(
                word in key
                for word in ("KEY", "TOKEN", "PASSWORD", "SECRET", "SESSION")
            )
        )
        service["ports"] = [
            f"{self.env['PROD_WEB_BIND']}:{self.env['PROD_WEB_PORT']}:18787"
        ]
        service["volumes"] = [
            {
                "type": "bind",
                "source": self.env[key],
                "target": target,
                "bind": {"create_host_path": False},
            }
            for key, target in (
                ("PROD_DATA_DIR", "/data"),
                ("PROD_CONFIG_DIR", "/config"),
            )
        ]
        service["group_add"] = old["HostConfig"].get("GroupAdd") or []
        if photos:
            service["volumes"].append(
                {
                    "type": "bind",
                    "source": photos,
                    "target": "/photos",
                    "read_only": True,
                    "bind": {"create_host_path": False},
                }
            )
        # Adoption must preserve runtime constraints, not silently change them.
        if (
            old["Config"]["User"] != service["user"]
            or old["HostConfig"]["RestartPolicy"]["Name"] != service["restart"]
        ):
            raise ValueError("Production user/restart policy differs from repository")
        self.old_image = old["Image"]
        volume = self.run(
            "volume", "inspect", self.env["PROD_STATE_VOLUME"], check=False
        )
        if (
            volume.returncode == 0
            and (json.loads(volume.stdout)[0].get("Labels") or {}).get(
                "frameflow.deploy.project"
            )
            != self.env["PROD_PROJECT"]
        ):
            raise ValueError("State volume is not owned by this deployment project")
        return config

    def lock(self):
        result = self.run(
            "create",
            "--name",
            self.env["PROD_CONTAINER"] + "-deploy-lock",
            "--label",
            "frameflow.deploy.owner=" + self.owner,
            "--network",
            "none",
            "--entrypoint",
            "/bin/true",
            self.old_image,
        )
        self.lock_id = result.stdout.decode().strip()

    def unlock(self):
        if self.lock_id:
            self.run("rm", self.lock_id)
            self.lock_id = None

    def state_io(self, payload=None):
        script = (
            "import json,os,sys;from pathlib import Path;p=Path('/state/deploy.json');"
            "os.umask(0o077);os.chmod('/state',0o700);v=json.load(sys.stdin);"
            "old=json.loads(p.read_text()) if p.exists() else {};"
            "owner=old.get('owner','');"
            "assert not owner or (len(owner)==32 and all(c in '0123456789abcdef' for c in owner));"
            "archive=p.with_name('deploy.'+owner+'.json');"
            "archive.write_text(p.read_text()) if owner and owner!=v.get('owner') and not archive.exists() else None;"
            "q=p.with_suffix('.tmp');q.write_text(json.dumps(v));q.replace(p)"
            if payload is not None
            else "from pathlib import Path;print(Path('/state/deploy.json').read_text())"
        )
        return self.run(
            "run",
            "--rm",
            "-i",
            "--user",
            "0:0",
            "--network",
            "none",
            "--mount",
            f"type=volume,source={self.env['PROD_STATE_VOLUME']},target=/state",
            "--entrypoint",
            "python",
            self.old_image,
            "-c",
            script,
            data=json.dumps(payload).encode() if payload is not None else None,
        ).stdout

    def activate(self, config):
        with tempfile.NamedTemporaryFile(mode="w", suffix=".yml") as file:
            yaml.safe_dump(json.loads(json.dumps(config).replace("$", "$$")), file)
            file.flush()
            self.run(
                "compose",
                "-p",
                self.env["PROD_PROJECT"],
                "-f",
                file.name,
                "config",
                "--quiet",
            )
            self.run(
                "compose",
                "-p",
                self.env["PROD_PROJECT"],
                "-f",
                file.name,
                "up",
                "-d",
                "--no-build",
                "--pull",
                "never",
                "pigway-frame-flow",
            )

    def verify(self, config):
        container = self.env["PROD_CONTAINER"]
        code = "import urllib.request;urllib.request.urlopen('http://127.0.0.1:18787/ready',timeout=3)"
        for _ in range(60):
            if (
                self.run(
                    "exec", container, "python", "-c", code, check=False
                ).returncode
                == 0
            ):
                break
            time.sleep(2)
        else:
            raise RuntimeError("Production /ready timeout")
        self.run(
            "exec",
            "-e",
            "FRAME_FLOW_URL=http://127.0.0.1:18787",
            "-i",
            container,
            "python",
            "-",
            data=(ROOT / "scripts/smoke-test.py").read_bytes(),
        )
        expected = config["services"]["pigway-frame-flow"]["environment"].get(
            "BUILD_COMMIT"
        )
        value = json.loads(
            self.run(
                "exec",
                container,
                "python",
                "-c",
                "import json,urllib.request;print(json.dumps(json.load(urllib.request.urlopen('http://127.0.0.1:18787/api/v1/version'))))",
            ).stdout
        )
        if value["commit"] != expected:
            raise RuntimeError("Running commit mismatch")
        print(
            "Production /ready and all smoke endpoints: HTTP 200; commit",
            value["commit"],
            flush=True,
        )
        return value

    def rollback(self, previous):
        self.activate(previous)
        self.verify(previous)
        print("ROLLBACK VERIFIED", flush=True)

    def secret_config(self):
        secret = self.env.get("PRODUCTION_ENV", "")
        proposed = dict(
            line.split("=", 1)
            for line in secret.splitlines()
            if line.strip() and not line.startswith("#") and "=" in line
        )
        for key in ("FRAME_FLOW_TOKEN", "KOKONNA_API_KEY"):
            if key not in proposed:
                raise ValueError("Missing PRODUCTION_ENV key: " + key)
        if self.env["PROD_MOCK"] == "false" and not proposed["KOKONNA_API_KEY"]:
            raise ValueError("KoKonna API key required for non-mock production")
        if not proposed["FRAME_FLOW_TOKEN"]:
            raise ValueError("FRAME_FLOW_TOKEN required for smoke checks")
        self.secret_values.extend(v for v in proposed.values() if v)
        return proposed

    def build(self, config):
        commit = (
            subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=ROOT)
            .decode()
            .strip()
        )
        main = (
            subprocess.check_output(
                ["git", "ls-remote", "origin", "refs/heads/main"], cwd=ROOT
            )
            .decode()
            .split()[0]
        )
        if commit != main:
            raise ValueError("Stale main revision; refusing automatic downgrade")
        image = (
            self.env["PROD_IMAGE_REPO"] + ":release-" + commit + "-" + self.owner[:8]
        )
        command = ["build", "--pull", "-t", image]
        now = datetime.now(UTC).isoformat()
        for key, value in {
            "DOCKER_PIP_INDEX_URL": self.env.get("DOCKER_PIP_INDEX_URL")
            or "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/simple",
            "DOCKER_PIP_FALLBACK_INDEX_URL": self.env.get(
                "DOCKER_PIP_FALLBACK_INDEX_URL"
            )
            or "https://pypi.org/simple",
            "DEBIAN_MIRROR": self.env.get("DEBIAN_MIRROR") or "deb.debian.org",
            "DEBIAN_SECURITY_MIRROR": self.env.get("DEBIAN_SECURITY_MIRROR")
            or "deb.debian.org",
            "BUILD_COMMIT": commit,
            "BUILD_TIME": now,
        }.items():
            command += ["--build-arg", key + "=" + value]
        self.run(*command, str(ROOT))
        service = config["services"]["pigway-frame-flow"]
        service["image"] = image
        proposed = self.secret_config()
        service["environment"].update(proposed)
        for key in ("SYNOLOGY_PHOTOS_URL", "SYNOLOGY_PHOTOS_CERT_SHA256"):
            if self.env.get(key):
                service["environment"][key] = self.env[key]
        service["environment"].update(
            KOKONNA_MOCK=self.env["PROD_MOCK"],
            KOKONNA_WRITE_ENABLED=self.env["PROD_WRITE_ENABLED"],
            BUILD_COMMIT=commit,
            BUILD_TIME=now,
            DATA_DIR="/data",
            CONFIG_DIR="/config",
            TZ="Asia/Shanghai",
        )
        return config

    def container_optional(self):
        result = self.run("container", "ls", "-a", "--format", "{{.Names}}")
        if self.env["PROD_CONTAINER"] not in result.stdout.decode().splitlines():
            return None
        return self.inspect(self.env["PROD_CONTAINER"])

    def host_paths(self, paths):
        # Inspect the daemon host, never the job container workspace. Read-only root.
        code = "import json,sys;from pathlib import Path;print(json.dumps([{'exists':p.exists() or p.is_symlink() or any(a.is_symlink() for a in p.parents),'empty':p.is_dir() and not p.is_symlink() and not any(a.is_symlink() for a in p.parents) and not any(p.iterdir())} for p in [Path('/host'+v) for v in json.load(sys.stdin)]]))"
        return json.loads(
            self.run(
                "run",
                "--rm",
                "-i",
                "--user",
                "0:0",
                "--network",
                "none",
                "--mount",
                "type=bind,source=/,target=/host,readonly",
                "--entrypoint",
                "python",
                self.old_image,
                "-c",
                code,
                data=json.dumps(paths).encode(),
            ).stdout
        )

    def bootstrap_check(self):
        self.settings()
        if self.env["PROD_MOCK"] != "true" or self.env["PROD_WRITE_ENABLED"] != "false":
            raise ValueError("Bootstrap requires mock=true and write=false")
        if self.env.get("PROD_PHOTOS_DIR"):
            raise ValueError("First bootstrap must not attach Photos")
        paths = [self.env[k] for k in ("PROD_DATA_DIR", "PROD_CONFIG_DIR")]
        if any(".." in Path(p).parts or p.endswith("/") for p in paths) or any(
            Path(a) in Path(b).parents
            for a, b in ((paths[0], paths[1]), (paths[1], paths[0]))
        ):
            raise ValueError("Unsafe or overlapping bootstrap directories")
        old = self.container_optional()
        if old:
            labels = old["Config"].get("Labels") or {}
            state = old["State"]
            if (
                state.get("Status") != "created"
                or state.get("Running")
                or state.get("StartedAt") != "0001-01-01T00:00:00Z"
                or labels.get("com.docker.compose.project") != self.env["PROD_PROJECT"]
                or labels.get("com.docker.compose.service") != "pigway-frame-flow"
            ):
                raise ValueError(
                    "Bootstrap refused: existing container is not a never-started owned residue"
                )
            mounts = old["Mounts"]
            if (
                len(mounts) != 2
                or {m["Destination"] for m in mounts} != {"/data", "/config"}
                or any(m["Type"] != "bind" for m in mounts)
            ):
                raise ValueError("Bootstrap refused: unknown persistent mount")
            if any(v["exists"] for v in self.host_paths([m["Source"] for m in mounts])):
                raise ValueError("Bootstrap refused: old mount has existing data/path")
        if any(v["exists"] and not v["empty"] for v in self.host_paths(paths)):
            raise ValueError("Bootstrap refused: target directory nonempty or unsafe")
        volume = self.run("volume", "ls", "--format", "{{.Name}}")
        if self.env["PROD_STATE_VOLUME"] in volume.stdout.decode().splitlines():
            volume_info = json.loads(
                self.run("volume", "inspect", self.env["PROD_STATE_VOLUME"]).stdout
            )[0]
            if (volume_info.get("Labels") or {}).get(
                "frameflow.deploy.project"
            ) != self.env["PROD_PROJECT"]:
                raise ValueError("Bootstrap refused: foreign journal volume")
            record = json.loads(self.state_io())
            candidate = record.get("candidate", {})
            service = candidate.get("services", {}).get("pigway-frame-flow", {})
            sources = {
                m.get("target"): m.get("source") for m in service.get("volumes", [])
            }
            if (
                record.get("kind") != "bootstrap"
                or record.get("status") != "failed"
                or candidate.get("name") != self.env["PROD_PROJECT"]
                or service.get("container_name") != self.env["PROD_CONTAINER"]
                or sources != {"/data": paths[0], "/config": paths[1]}
                or service.get("ports")
                != [f"{self.env['PROD_WEB_BIND']}:{self.env['PROD_WEB_PORT']}:18787"]
            ):
                raise ValueError(
                    "Bootstrap refused: journal is not a failed, matching first install"
                )
            print(
                "Manual retry of failed bootstrap with empty targets; prior journal will be archived",
                flush=True,
            )
        return old

    def initialize_directories(self):
        mounts, names = [], []
        for key, target in (
            ("PROD_DATA_DIR", "/parent-data"),
            ("PROD_CONFIG_DIR", "/parent-config"),
        ):
            path = Path(self.env[key])
            if path.parent == Path("/"):
                raise ValueError("Refusing writable NAS root mount")
            mounts += ["--mount", f"type=bind,source={path.parent},target={target}"]
            names.append(str(Path(target) / path.name))
        # Synology rejects missing bind sources even with -v. Mount only existing
        # explicit parents, then mkdir/chown/chmod only the two validated leaves.
        code = (
            "import json,os,sys;from pathlib import Path;"
            "paths=[Path(v) for v in json.load(sys.stdin)];"
            "assert all(not p.is_symlink() and (not p.exists() or (p.is_dir() and not any(p.iterdir()))) for p in paths);"
            "[(p.mkdir(exist_ok=True),os.chown(p,10333,10333),os.chmod(p,0o750)) for p in paths]"
        )
        self.run(
            "run",
            "--rm",
            "-i",
            "--user",
            "0:0",
            "--network",
            "none",
            *mounts,
            "--entrypoint",
            "python",
            self.old_image,
            "-c",
            code,
            data=json.dumps(names).encode(),
        )

    def bootstrap_config(self):
        config = yaml.safe_load((ROOT / "docker-compose.yml").read_text())
        config["name"] = self.env["PROD_PROJECT"]
        service = config["services"]["pigway-frame-flow"]
        service.pop("build")
        service.pop("env_file")
        service["container_name"] = self.env["PROD_CONTAINER"]
        service["environment"] = {}
        service["ports"] = [
            f"{self.env['PROD_WEB_BIND']}:{self.env['PROD_WEB_PORT']}:18787"
        ]
        service["labels"] = {"frameflow.bootstrap.owner": self.owner}
        service["volumes"] = [
            {
                "type": "bind",
                "source": self.env[key],
                "target": target,
                "bind": {"create_host_path": False},
            }
            for key, target in (
                ("PROD_DATA_DIR", "/data"),
                ("PROD_CONFIG_DIR", "/config"),
            )
        ]
        return config

    def bootstrap_cleanup(self):
        old = self.container_optional()
        if old:
            if (old["Config"].get("Labels") or {}).get(
                "frameflow.bootstrap.owner"
            ) != self.owner:
                raise ValueError("Refusing cleanup of another container")
            self.run("rm", "-f", old["Id"])
        if self.state and self.state.get("network_created"):
            name = self.env["PROD_PROJECT"] + "_default"
            result = self.run("network", "inspect", name, check=False)
            if result.returncode == 0:
                network = json.loads(result.stdout)[0]
                if (
                    network.get("Containers")
                    or (network.get("Labels") or {}).get("com.docker.compose.project")
                    != self.env["PROD_PROJECT"]
                ):
                    raise ValueError("Refusing cleanup of occupied/unknown network")
                self.run("network", "rm", network["Id"])

    def bootstrap(self):
        self.settings()
        if self.env["PROD_MOCK"] != "true" or self.env["PROD_WRITE_ENABLED"] != "false":
            raise ValueError("Bootstrap requires mock=true and write=false")
        self.secret_config()
        # Build before host inspection so helpers use this verified repository image.
        candidate = self.build(self.bootstrap_config())
        self.old_image = candidate["services"]["pigway-frame-flow"]["image"]
        self.lock()
        safe_unlock = True
        try:
            residue = self.bootstrap_check()
            network_name = self.env["PROD_PROJECT"] + "_default"
            networks = (
                self.run("network", "ls", "--format", "{{.Name}}")
                .stdout.decode()
                .splitlines()
            )
            if network_name in networks:
                network = json.loads(
                    self.run("network", "inspect", network_name).stdout
                )[0]
                if (
                    set(network.get("Containers") or {})
                    - ({residue["Id"]} if residue else set())
                    or (network.get("Labels") or {}).get("com.docker.compose.project")
                    != self.env["PROD_PROJECT"]
                ):
                    raise ValueError(
                        "Bootstrap network identity/occupancy requires review"
                    )
            self.state = {
                "owner": self.owner,
                "kind": "bootstrap",
                "status": "prepared",
                "candidate": candidate,
                "network_created": network_name not in networks,
            }
            self.run(
                "volume",
                "create",
                "--label",
                "frameflow.deploy.project=" + self.env["PROD_PROJECT"],
                self.env["PROD_STATE_VOLUME"],
            )
            self.state_io(self.state)
            if residue:
                self.run("rm", residue["Id"])
            self.initialize_directories()
            self.activate(candidate)
            self.verify(candidate)
            self.preflight()
            self.run("restart", self.env["PROD_CONTAINER"])
            self.verify(candidate)
            self.preflight()
            info = self.inspect(self.env["PROD_CONTAINER"])
            for _ in range(60):
                if info["State"].get("Health", {}).get("Status") == "healthy":
                    break
                time.sleep(2)
                info = self.inspect(self.env["PROD_CONTAINER"])
            else:
                raise RuntimeError("Bootstrap health timeout")
            expected_image = json.loads(
                self.run(
                    "image",
                    "inspect",
                    candidate["services"]["pigway-frame-flow"]["image"],
                ).stdout
            )[0]["Id"]
            if info["Image"] != expected_image:
                raise RuntimeError("Bootstrap image mismatch")
            self.state["status"] = "success"
            self.state_io(self.state)
            print("BOOTSTRAP VERIFIED", flush=True)
        except BaseException:
            print("BOOTSTRAP FAILED", flush=True)
            self.diagnostics()
            if self.state:
                safe_unlock = False
                self.bootstrap_cleanup()
                self.state["status"] = "failed"
                self.state_io(self.state)
                safe_unlock = True
            raise
        finally:
            if safe_unlock:
                self.unlock()

    def diagnostics(self):
        try:
            self._diagnostics()
        except Exception:
            print(
                "Scoped diagnostics unavailable; rollback still attempted", flush=True
            )

    def _diagnostics(self):
        result = self.run("inspect", self.env["PROD_CONTAINER"], check=False)
        if result.returncode == 0:
            info = json.loads(result.stdout)[0]
            self.secret_values.extend(
                value
                for key, value in (
                    entry.split("=", 1)
                    for entry in info.get("Config", {}).get("Env", [])
                    if "=" in entry
                )
                if value
                and any(
                    word in key
                    for word in ("KEY", "TOKEN", "PASSWORD", "SECRET", "SESSION")
                )
            )
            state = info["State"]
            print(
                "Scoped production diagnostic:",
                json.dumps(
                    {
                        "image": info["Image"],
                        "status": state.get("Status"),
                        "exit_code": state.get("ExitCode"),
                        "health": state.get("Health", {}).get("Status"),
                    }
                ),
                flush=True,
            )
        logs = self.run("logs", "--tail", "60", self.env["PROD_CONTAINER"], check=False)
        text = (logs.stdout + logs.stderr).decode(errors="replace")
        for value in sorted(self.secret_values, key=len, reverse=True):
            text = text.replace(value, "[REDACTED]")
        text = re.sub(r"https?://[^\s\"'<>]+", "[REDACTED-URL]", text)
        print("Scoped container logs (redacted):", text, flush=True)

    def cleanup_images(self):
        """Retain container references and the verified deployment's rollback baseline."""
        try:
            protected = {self.state["previous_image"]}
            container_ids = self.run("ps", "-aq").stdout.decode().split()
            if container_ids:
                containers = json.loads(self.run("inspect", *container_ids).stdout)
                protected.update(container["Image"] for container in containers)
            repo = self.env["PROD_IMAGE_REPO"]
            image_ids = (
                self.run("image", "ls", "--no-trunc", "-q", repo)
                .stdout.decode()
                .split()
            )
            if not image_ids:
                return
            images = json.loads(
                self.run("image", "inspect", *sorted(set(image_ids))).stdout
            )
            removed = 0
            for image in images:
                tags = image.get("RepoTags") or []
                # Shared images and unrelated tags are never cleanup targets.
                if image["Id"] in protected or not tags:
                    continue
                if not all(
                    re.fullmatch(
                        re.escape(repo)
                        + r":(?:release-[0-9a-f]{40}-[0-9a-f]{8}|rollback-[0-9a-f]{32})",
                        tag,
                    )
                    for tag in tags
                ):
                    continue
                result = self.run("image", "rm", *tags, check=False)
                if result.returncode:
                    print("IMAGE RETENTION: referenced image retained", flush=True)
                else:
                    removed += 1
            print(f"IMAGE RETENTION: removed {removed} obsolete images", flush=True)
        except Exception:
            # Cleanup failure must never roll back an already verified deployment.
            print("IMAGE RETENTION WARNING: cleanup deferred", flush=True)

    def deploy(self):
        previous = self.preflight()
        self.lock()
        touched = False
        safe_unlock = True
        try:
            # Refresh baseline under the daemon-level lock.
            previous = self.preflight()
            version = self.verify(previous)
            pin = self.env["PROD_IMAGE_REPO"] + ":rollback-" + self.owner
            self.run("image", "tag", self.old_image, pin)
            candidate = self.build(json.loads(json.dumps(previous)))
            self.state = {
                "owner": self.owner,
                "status": "prepared",
                "previous": previous,
                "candidate": candidate,
                "previous_image": self.old_image,
                "previous_version": version,
                "rollback_tag": pin,
            }
            self.run(
                "volume",
                "create",
                "--label",
                "frameflow.deploy.project=" + self.env["PROD_PROJECT"],
                self.env["PROD_STATE_VOLUME"],
            )
            self.state_io(self.state)
            # Online backup is additive; rollback never restores/deletes user data.
            backup = "/data/deploy-backups/" + self.owner + ".db"
            self.run(
                "exec",
                self.env["PROD_CONTAINER"],
                "python",
                "-c",
                "import sqlite3;from pathlib import Path;p=Path('"
                + backup
                + "');p.parent.mkdir(exist_ok=True);assert Path('/data/frameflow.db').is_file();sqlite3.connect('/data/frameflow.db').backup(sqlite3.connect(p))",
            )
            touched = True
            self.activate(candidate)
            self.verify(candidate)
            self.state["status"] = "success"
            self.state_io(self.state)
            print("DEPLOY VERIFIED", flush=True)
            self.cleanup_images()
        except BaseException as exc:
            print("DEPLOY FAILED:", type(exc).__name__, flush=True)
            self.diagnostics()
            if touched:
                safe_unlock = False
                try:
                    self.rollback(previous)
                except BaseException:
                    print(
                        "ROLLBACK FAILED: deployment lock retained; always() recovery will retry",
                        flush=True,
                    )
                    self.diagnostics()
                    raise
                safe_unlock = True
                self.state["status"] = "rolled_back"
                self.state_io(self.state)
            raise
        finally:
            if safe_unlock:
                self.unlock()

    def recover(self, manual=False):
        self.settings()
        lock = self.run(
            "inspect", self.env["PROD_CONTAINER"] + "-deploy-lock", check=False
        )
        if manual:
            self.preflight(require_running=False)
            self.lock()
        elif lock.returncode == 0:
            info = json.loads(lock.stdout)[0]
            if (info["Config"].get("Labels") or {}).get(
                "frameflow.deploy.owner"
            ) != self.owner:
                return
            self.lock_id = info["Id"]
            self.old_image = info["Image"]
        else:
            return
        recovered = False
        try:
            self.run("volume", "inspect", self.env["PROD_STATE_VOLUME"])
            volume = json.loads(
                self.run("volume", "inspect", self.env["PROD_STATE_VOLUME"]).stdout
            )[0]
            if (volume.get("Labels") or {}).get("frameflow.deploy.project") != self.env[
                "PROD_PROJECT"
            ]:
                raise ValueError("Recovery state volume ownership mismatch")
            state = json.loads(self.state_io())
            if state.get("kind") == "bootstrap":
                self.state = state
                if state["owner"] != self.owner:
                    raise ValueError("Bootstrap journal owner mismatch")
                if state["status"] == "prepared":
                    self.bootstrap_cleanup()
                    state["status"] = "failed"
                    self.state_io(state)
                recovered = True
                return
            previous = state["previous"]
            service = previous["services"]["pigway-frame-flow"]
            if (
                previous["name"] != self.env["PROD_PROJECT"]
                or service["container_name"] != self.env["PROD_CONTAINER"]
            ):
                raise ValueError("Rollback journal resource mismatch")
            sources = {m["target"]: m["source"] for m in service["volumes"]}
            if (
                sources.get("/data") != self.env["PROD_DATA_DIR"]
                or sources.get("/config") != self.env["PROD_CONFIG_DIR"]
            ):
                raise ValueError("Rollback journal persistence mismatch")
            if manual:
                state["owner"] = self.owner
                state["status"] = "prepared"
                self.state_io(state)
            if manual or (
                state["owner"] == self.owner
                and state["status"] not in ("success", "rolled_back")
            ):
                self.rollback(state["previous"])
                state["status"] = "rolled_back"
                self.state_io(state)
            recovered = True
        except BaseException:
            print("RECOVERY FAILED: production lock retained", flush=True)
            self.diagnostics()
            raise
        finally:
            if recovered:
                self.unlock()


def main():
    tool = Deploy()
    action = tool.env.get("CD_ACTION") or (
        "deploy" if tool.env.get("PROD_DEPLOY_ENABLED") == "true" else "plan"
    )
    if action == "recover":
        if tool.env.get("FRAME_FLOW_DEPLOY_OWNER"):
            tool.recover()
        return
    if action != "plan" and tool.env.get("PROD_DEPLOY_ENABLED") != "true":
        raise SystemExit(
            "Production CD disabled; set PROD_DEPLOY_ENABLED=true after safe trial"
        )
    if action == "plan":
        if any(not tool.env.get(key) for key in REQUIRED):
            print("PLAN ONLY: configure required Variables:", ", ".join(REQUIRED))
            return
        if tool.container_optional() is None:
            print("PLAN ONLY: production not bootstrapped / bootstrap required")
            return
        info = tool.inspect(tool.env["PROD_CONTAINER"])
        if info["State"].get("Status") == "created":
            print(
                "PLAN ONLY: production not bootstrapped; created residue requires bootstrap safety checks"
            )
            return
        baseline = tool.preflight()
        tool.secret_config()
        tool.verify(baseline)
        print(
            "PLAN ONLY: production resource identity/mounts/port checked; no resources modified"
        )
        return
    if tool.env.get("GITHUB_ENV"):
        with Path(tool.env["GITHUB_ENV"]).open("a") as file:
            file.write("FRAME_FLOW_DEPLOY_OWNER=" + tool.owner + "\n")

    def cancel(*_):
        raise SystemExit(143)

    signal.signal(signal.SIGTERM, cancel)
    if action == "bootstrap":
        if tool.env.get("GITHUB_EVENT_NAME") != "workflow_dispatch":
            raise SystemExit("Bootstrap requires explicit workflow_dispatch")
        try:
            tool.bootstrap()
        except BaseException:
            print(
                "BOOTSTRAP FAILED: inspect redacted diagnostics; data/config retained",
                flush=True,
            )
            raise
    elif action == "deploy":
        info = tool.container_optional()
        if info is None or info["State"].get("Status") == "created":
            print("production not bootstrapped / bootstrap required")
            return
        tool.deploy()
    elif action == "rollback":
        tool.recover(manual=True)
    else:
        raise SystemExit("Unknown CD_ACTION")


if __name__ == "__main__":
    main()