from __future__ import annotations

import inspect
import json
import shutil
import subprocess
import sys
import tempfile
import textwrap
import unittest
from pathlib import Path

from deployments.online.runtime import consume_staging_d3_release_tag as tag_consumer
from deployments.online.runtime import prepare_multi_prod_d3_runtime_env as prod_prepare
from deployments.online.runtime.consume_staging_d3_release_tag import (
    TagError,
    consume,
    verify_staging_dispatch,
)
from deployments.online.runtime.prepare_multi_prod_d3_runtime_env import (
    PrepareError,
    prepare,
)
from deployments.online.runtime.prod_d3_ingress import (
    IngressError,
    next_slot,
    parse_prod_route,
)
from scripts import run_serviceification_d3_prod as prod


REPO_ROOT = Path(__file__).resolve().parents[1]
WORKFLOW = REPO_ROOT / ".github" / "workflows" / "deploy-multi-prod-d3-environment.yml"


class ProdD3MetadataFetchTests(unittest.TestCase):
    def setUp(self) -> None:
        temp = tempfile.TemporaryDirectory()
        self.addCleanup(temp.cleanup)
        self.root = Path(temp.name)
        self.bin_dir = self.root / "bin"
        self.bin_dir.mkdir()
        self.calls = self.root / "calls.jsonl"
        # Only curl is available to the workflow shell. Its transport fixture
        # records requests and returns metadata or curl's HTTP failure status.
        curl = self.bin_dir / "curl"
        curl.write_text(
            f"#!{sys.executable}\n" + textwrap.dedent("""\
                import json
                import os
                import sys
                from pathlib import Path

                args = sys.argv[1:]
                with Path(os.environ["FETCH_CALLS"]).open("a") as output:
                    output.write(json.dumps(args) + "\\n")
                url = next(arg for arg in args if arg.startswith("https://"))
                if os.environ.get("FAIL_ENDPOINT", "never") in url:
                    sys.exit(22)
                payload = {"id": int(url.rsplit("/", 1)[1])}
                Path(args[args.index("--output") + 1]).write_text(json.dumps(payload))
                """),
            encoding="utf-8",
        )
        curl.chmod(0o755)
        step = WORKFLOW.read_text(encoding="utf-8").split(
            "- name: Fetch Tag-selected Staging run and artifact metadata", 1
        )[1].split("\n      - name:", 1)[0]
        self.script = textwrap.dedent(step.split("        run: |\n", 1)[1])
        self.env = {
            "PATH": str(self.bin_dir),
            "GH_TOKEN": "fixture-token",
            "GITHUB_API_URL": "https://api.github.com",
            "GITHUB_REPOSITORY": "pingcap-inc/nexuscrm-pingcap-pack",
            "STAGING_RUN_ID": "33956521445",
            "ARTIFACT_ID": "9967000291",
            "RUNNER_TEMP": str(self.root),
            "FETCH_CALLS": str(self.calls),
        }

    def run_fetch(self, **env: str) -> subprocess.CompletedProcess[str]:
        return subprocess.run(
            ["/bin/bash", "--noprofile", "--norc", "-e", "-o", "pipefail", "-c", self.script],
            env={**self.env, **env}, capture_output=True, text=True, timeout=10,
        )

    def test_fetches_exact_metadata_without_github_cli(self) -> None:
        self.assertIsNone(shutil.which("gh", path=self.env["PATH"]))
        result = self.run_fetch()
        self.assertEqual(0, result.returncode, result.stderr)
        calls = [json.loads(line) for line in self.calls.read_text().splitlines()]
        self.assertEqual(2, len(calls))
        for args, endpoint, filename, identity in (
            (calls[0], "runs/33956521445", "staging-d3-workflow-run.json", 33956521445),
            (calls[1], "artifacts/9967000291", "staging-d3-evidence-artifact.json", 9967000291),
        ):
            self.assertIn(
                f"https://api.github.com/repos/pingcap-inc/nexuscrm-pingcap-pack/actions/{endpoint}",
                args,
            )
            self.assertIn("Authorization: Bearer fixture-token", args)
            self.assertIn("Accept: application/vnd.github+json", args)
            self.assertIn("X-GitHub-Api-Version: 2022-11-28", args)
            self.assertIn("--fail", args)
            self.assertEqual({"id": identity}, json.loads((self.root / filename).read_text()))
        self.assertNotIn("fixture-token", result.stdout + result.stderr)

    def test_stops_when_either_metadata_request_fails(self) -> None:
        for endpoint, expected_calls in (("/runs/", 1), ("/artifacts/", 2)):
            with self.subTest(endpoint=endpoint):
                self.calls.write_text("")
                result = self.run_fetch(FAIL_ENDPOINT=endpoint)
                self.assertEqual(22, result.returncode, result.stderr)
                self.assertEqual(expected_calls, len(self.calls.read_text().splitlines()))
                self.assertFalse((self.root / "staging-d3-evidence-artifact.json").exists())

    def test_missing_curl_fails_with_actionable_error(self) -> None:
        result = self.run_fetch(PATH=str(self.root / "missing-bin"))
        self.assertNotEqual(0, result.returncode)
        self.assertIn("missing Prod D3 host prerequisite: curl", result.stdout)
        self.assertFalse(self.calls.exists())


def nginx(port: int, *, d3: bool = False) -> str:
    marker = "# NexusCRM D3 direct ingress run=d3-prod-slot-run\n" if d3 else ""
    return (
        marker
        + "server {\n"
        + "    listen 80;\n"
        + "    server_name nexuscrm-new-prod.pingcap.com;\n"
        + f"    location /api/ {{ proxy_pass http://127.0.0.1:{port}; }}\n"
        + f"    location /health {{ proxy_pass http://127.0.0.1:{port}/health; }}\n"
        + "}\n"
    )


class ProdD3ReleaseTests(unittest.TestCase):
    def test_classic_handoff_accepts_the_routed_slots_workflow_adapter(self) -> None:
        for port, adapter, successor in (
            (3103, 3204, "slot-3104"),
            (3104, 3205, "slot-3105"),
            (3105, 3206, "slot-3106"),
            (3106, 3207, "slot-3103"),
        ):
            with self.subTest(port=port):
                classic = nginx(port).replace(
                    "    location /health",
                    "    location /hubspot/workflow-actions/options/ { "
                    f"proxy_pass http://127.0.0.1:{adapter}; }}\n"
                    "    location = /hubspot/workflow-actions/execute { "
                    f"proxy_pass http://127.0.0.1:{adapter}; }}\n"
                    "    location /health",
                )
                route = parse_prod_route(classic)
                self.assertEqual(port, route.backend_port)
                self.assertFalse(route.is_d3)
                self.assertEqual(successor, next_slot(route))

    def test_classic_handoff_rejects_other_slots_and_d3_only_proxy_ports(self) -> None:
        for port, other_adapter, integration in (
            (3103, 3205, 3403),
            (3104, 3206, 3413),
            (3105, 3207, 3423),
            (3106, 3204, 3433),
        ):
            for unexpected in (other_adapter, integration):
                with self.subTest(port=port, unexpected=unexpected):
                    classic = nginx(port).replace(
                        "    location /health",
                        "    location /other { "
                        f"proxy_pass http://127.0.0.1:{unexpected}; }}\n"
                        "    location /health",
                    )
                    with self.assertRaisesRegex(IngressError, "outside the routed slot"):
                        parse_prod_route(classic)

    def test_ingress_parser_rotates_all_four_physical_slots(self) -> None:
        expected = {
            3103: ("slot-3103", "slot-3104"),
            3104: ("slot-3104", "slot-3105"),
            3105: ("slot-3105", "slot-3106"),
            3106: ("slot-3106", "slot-3103"),
        }
        for port, (current, target) in expected.items():
            with self.subTest(port=port):
                route = parse_prod_route(nginx(port, d3=port != 3106))
                self.assertEqual(current, route.slot)
                self.assertEqual(target, next_slot(route))

    def test_ingress_parser_rejects_mixed_prod_backend_owners(self) -> None:
        mixed = nginx(3103).replace(
            "location /health",
            "location /other { proxy_pass http://127.0.0.1:3104; }\n    location /health",
        )
        with self.assertRaisesRegex(IngressError, "unique backend port"):
            parse_prod_route(mixed)

        unreviewed = nginx(3103).replace(
            "location /health",
            "location /other { proxy_pass http://127.0.0.1:3999; }\n    location /health",
        )
        with self.assertRaisesRegex(IngressError, "outside the routed slot"):
            parse_prod_route(unreviewed)

        hostname_target = nginx(3103).replace(
            "location /health",
            "location /other { proxy_pass http://localhost:3999; }\n"
            "    location /health",
        )
        with self.assertRaisesRegex(IngressError, "unreviewed proxy target"):
            parse_prod_route(hostname_target)

        reviewed_d3 = nginx(3103, d3=True).replace(
            "location /health",
            "location /adapter { proxy_pass http://127.0.0.1:3204; }\n"
            "    location /webhooks { proxy_pass http://127.0.0.1:3403; }\n"
            "    location /health",
        )
        self.assertEqual(3103, parse_prod_route(reviewed_d3).backend_port)


    def test_first_handoff_rejects_frontend_outside_routed_release(self) -> None:
        with tempfile.TemporaryDirectory() as temp:
            root = Path(temp) / "multi_prod"
            current = root / "current"
            current_env = current / "env" / "runtime.env"
            current_env.parent.mkdir(parents=True)
            unrelated_frontend = root / "unrelated" / "frontend-dist"
            unrelated_frontend.mkdir(parents=True)
            (current / "frontend-dist").symlink_to(
                unrelated_frontend, target_is_directory=True
            )
            current_env.write_text(
                f"ENV_ROOT={current}\nBACKEND_PORT=3103\n", encoding="utf-8"
            )
            with self.assertRaisesRegex(PrepareError, "valid live release"):
                prod_prepare._require_current_route_reconciled(
                    root, parse_prod_route(nginx(3103, d3=False))
                )

    def test_first_handoff_rejects_stale_d3_current_pointer(self) -> None:
        with tempfile.TemporaryDirectory() as temp:
            root = Path(temp) / "multi_prod"
            current = root / "current"
            current_env = current / "env" / "runtime.env"
            current_env.parent.mkdir(parents=True)
            (current / "frontend-dist").mkdir()
            current_env.write_text(
                f"ENV_ROOT={current}\nBACKEND_PORT=3103\n", encoding="utf-8"
            )
            stale = root / "d3" / "releases" / "stale"
            stale.mkdir(parents=True)
            (root / "d3" / "current").symlink_to(stale)
            with self.assertRaisesRegex(PrepareError, "D3 current authority"):
                prod_prepare._require_current_route_reconciled(
                    root, parse_prod_route(nginx(3103, d3=False))
                )

    def test_d3_current_marker_binds_pointer_env_and_frontend(self) -> None:
        with tempfile.TemporaryDirectory() as temp:
            root = Path(temp) / "multi_prod"
            d3_root = root / "d3"
            release = d3_root / "releases" / "staging-d3-release-current"
            other = d3_root / "releases" / "staging-d3-release-other"
            (release / "env").mkdir(parents=True)
            (release / "frontend-dist").mkdir()
            other.mkdir(parents=True)
            bom_reference = (
                "ghcr.io/pingcap-inc/nexuscrm-serviceification-bom@sha256:"
                + "a" * 64
            )
            (release / "env" / "runtime.env").write_text(
                "\n".join(
                    (
                        f"ENV_ROOT={release}",
                        "ENV_NAME=multi-prod-d3-slot-3103",
                        "BACKEND_PORT=3103",
                        "RELEASE_TAG=staging-d3-release-current",
                        f"D3_RESOLVED_BOM_REFERENCE={bom_reference}",
                    )
                )
                + "\n",
                encoding="utf-8",
            )
            (d3_root / "current").symlink_to(release)
            marker_path = d3_root / "current.json"
            marker = {
                "schema": prod_prepare.CURRENT_SCHEMA,
                "releaseTag": "staging-d3-release-current",
                "slot": "slot-3103",
                "backendPort": 3103,
                "resolvedBomReference": bom_reference,
                "releaseRoot": str(release),
            }
            marker_path.write_text(json.dumps(marker) + "\n", encoding="utf-8")
            route = parse_prod_route(nginx(3103, d3=True))
            prod_prepare._require_current_route_reconciled(root, route)

            marker["releaseRoot"] = str(other)
            marker_path.write_text(json.dumps(marker) + "\n", encoding="utf-8")
            with self.assertRaisesRegex(PrepareError, "current pointer"):
                prod_prepare._require_current_route_reconciled(root, route)

            marker["releaseRoot"] = str(release)
            marker_path.write_text(json.dumps(marker) + "\n", encoding="utf-8")
            (release / "frontend-dist").rmdir()
            outside_frontend = d3_root / "outside-frontend"
            outside_frontend.mkdir()
            (release / "frontend-dist").symlink_to(
                outside_frontend, target_is_directory=True
            )
            with self.assertRaisesRegex(PrepareError, "env or frontend"):
                prod_prepare._require_current_route_reconciled(root, route)

    def test_prod_consumes_only_the_minimal_d3_tag_contract(self) -> None:
        run_id = "d3-staging-blue-123-1"
        payload = {
            "schema": "nexuscrm.serviceification-d3-release-tag/v5",
            "tagTargetCommit": "c" * 40,
            "resolvedBomReference": (
                "ghcr.io/pingcap-inc/nexuscrm-serviceification-bom@sha256:" + "a" * 64
            ),
            "patchBundle": None,
            "staging": {
                "workflowPath": ".github/workflows/deploy-staging-blue-green-environment.yml",
                "headCommit": "b" * 40,
                "workflowRunId": "123",
                "workflowRunAttempt": 1,
                "deploymentRunId": run_id,
                "slot": "staging-blue",
                "evidenceArtifact": {
                    "id": 123456,
                    "name": f"serviceification-d3-staging-{run_id}",
                    "digest": "sha256:" + "e" * 64,
                },
            },
        }
        result = consume(
            payload,
            release_tag="staging-d3-release-20260823T010203Z-123",
            tag_target_commit="c" * 40,
        )
        self.assertEqual(payload["resolvedBomReference"], result["resolved_bom_reference"])
        self.assertEqual("1", result["staging_workflow_run_attempt"])
        self.assertEqual("", result["patch_dir"])
        self.assertEqual("", result["patch_tree_digest"])
        serialized = json.dumps(payload).lower()
        self.assertNotIn("businessacceptance", serialized)
        self.assertNotIn("candidate", serialized)
        self.assertNotIn("s3", serialized)

    def test_prod_consumes_patch_identity_bound_to_staging_evidence(self) -> None:
        patch_path = "docs/migration/patch-bundles/20260823_once"
        patch_digest = "sha256:" + "d" * 64
        payload = {
            "schema": "nexuscrm.serviceification-d3-release-tag/v5",
            "tagTargetCommit": "c" * 40,
            "resolvedBomReference": (
                "ghcr.io/pingcap-inc/nexuscrm-serviceification-bom@sha256:" + "a" * 64
            ),
            "patchBundle": {"path": patch_path, "treeDigest": patch_digest},
            "staging": {
                "workflowPath": ".github/workflows/deploy-staging-blue-green-environment.yml",
                "headCommit": "b" * 40,
                "workflowRunId": "123",
                "workflowRunAttempt": 1,
                "deploymentRunId": "d3-staging-blue-123-1",
                "slot": "staging-blue",
                "evidenceArtifact": {
                    "id": 123456,
                    "name": "serviceification-d3-staging-d3-staging-blue-123-1",
                    "digest": "sha256:" + "e" * 64,
                },
            },
        }
        result = consume(
            payload,
            release_tag="staging-d3-release-20260823T010203Z-123",
            tag_target_commit="c" * 40,
        )
        self.assertEqual(patch_path, result["patch_dir"])
        self.assertEqual(patch_digest, result["patch_tree_digest"])
        dispatch = {
            "schema": "nexuscrm.serviceification-d3-staging-dispatch/v2",
            "runId": "d3-staging-blue-123-1",
            "slot": "staging-blue",
            "resolvedBomReference": payload["resolvedBomReference"],
            "l2Commit": payload["tagTargetCommit"],
            "patchBundle": {"path": patch_path, "treeDigest": patch_digest},
        }
        workflow_run = {
            "id": 123,
            "path": ".github/workflows/deploy-staging-blue-green-environment.yml",
            "event": "workflow_dispatch",
            "head_branch": "main",
            "status": "completed",
            "conclusion": "success",
            "head_sha": "b" * 40,
            "run_attempt": 1,
        }
        artifact_metadata = {
            "id": 123456,
            "name": "serviceification-d3-staging-d3-staging-blue-123-1",
            "digest": "sha256:" + "e" * 64,
            "expired": False,
            "workflow_run": {
                "id": 123,
                "head_branch": "main",
                "head_sha": "b" * 40,
            },
        }
        verify_staging_dispatch(
            payload, dispatch, workflow_run, artifact_metadata
        )
        dispatch["patchBundle"] = None
        with self.assertRaisesRegex(TagError, "dispatch identity differs"):
            verify_staging_dispatch(payload, dispatch, workflow_run, artifact_metadata)
        dispatch["patchBundle"] = payload["patchBundle"]

        for field, replacement in (
            ("run_attempt", 2),
            ("head_sha", "f" * 40),
        ):
            with self.subTest(workflow_run_field=field):
                changed_run = dict(workflow_run)
                changed_run[field] = replacement
                with self.assertRaisesRegex(TagError, "workflow run differs"):
                    verify_staging_dispatch(
                        payload, dispatch, changed_run, artifact_metadata
                    )

        for field, replacement in (
            ("id", 654321),
            ("name", "serviceification-d3-staging-other"),
            ("digest", "sha256:" + "f" * 64),
        ):
            with self.subTest(artifact_field=field):
                changed_artifact = dict(artifact_metadata)
                changed_artifact[field] = replacement
                with self.assertRaisesRegex(TagError, "artifact differs"):
                    verify_staging_dispatch(
                        payload, dispatch, workflow_run, changed_artifact
                    )

    def test_tag_consumer_cli_requires_immutable_staging_dispatch(self) -> None:
        source = inspect.getsource(tag_consumer.main)
        self.assertIn('"--staging-dispatch", required=True', source)
        self.assertIn('"--staging-workflow-run", required=True', source)
        self.assertIn('"--staging-artifact-metadata", required=True', source)
        self.assertNotIn("if args.staging_dispatch", source)


    def test_cutover_proof_precedes_retirement_and_marker_publication(self) -> None:
        source = inspect.getsource(prod.converge_prod)
        pending = source.index("require_prod_in_progress")
        patch = source.index("preflight_and_seal_patch")
        prepare = source.index("prepare_target_for_convergence")
        converge = source.index("converge_target_with_bounded_resume")
        live_readback = source.index("require_live_target")
        public_readback = source.index("verify_public_target")
        retirement = source.index("retire_previous_runtime")
        final_readback = source.rindex("verify_public_target")
        finalize = source.index("impl.replace_state")
        publish = source.index("publish_current")
        self.assertLess(pending, patch)
        self.assertLess(patch, prepare)
        self.assertLess(pending, prepare)
        self.assertLess(prepare, converge)
        self.assertLess(converge, live_readback)
        self.assertLess(live_readback, public_readback)
        self.assertLess(public_readback, retirement)
        self.assertLess(retirement, final_readback)
        self.assertLess(final_readback, finalize)
        self.assertLess(finalize, publish)
        self.assertNotIn("activeRuntimeFence", source)
        self.assertIn("previousRuntimeRetirement", source)
        self.assertNotIn("retire_previous_runtime", inspect.getsource(prod.prepare_target_for_convergence))


    def test_nginx_switch_follows_all_d3_and_companion_readiness(self) -> None:
        start = inspect.getsource(prod.impl.start_serving)
        ingress = start.index("ensure_d3_public_ingress")
        self.assertGreater(start.rfind("ensure_detached_container", 0, ingress), 0)
        self.assertGreater(start.rfind("wait_ready=wait_ready", 0, ingress), 0)
        prod_ingress = inspect.getsource(prod.ensure_prod_ingress)
        self.assertLess(
            prod_ingress.index("converge_companion"),
            prod_ingress.index("_ORIG_ENSURE_INGRESS"),
        )


    def test_companion_comes_from_reviewed_activation_checkout(self) -> None:
        source = inspect.getsource(prod.companion_activation_authority)
        self.assertIn("REPO_ROOT.resolve()", source)
        self.assertIn('"status"', source)
        self.assertNotIn("l2Root", source)
        receipt_source = inspect.getsource(prod._companion_receipt)
        self.assertIn("companion_activation_authority", receipt_source)
        self.assertIn("reviewed-activation-checkout", receipt_source)


    def test_prod_retry_quiescence_keeps_the_live_slot_out_of_hostwide_checks(self) -> None:
        source = inspect.getsource(prod.reestablish_prod_serving_quiescence)
        self.assertIn("quiesce_target_runtime", source)
        self.assertNotIn("list-units", source)
        self.assertNotIn('Path("/proc")', source)


    def test_catalog_resume_reproves_target_quiescence_before_mutation(self) -> None:
        source = inspect.getsource(prod.apply_catalog)
        quiesce = source.index("quiesce_target_runtime")
        replay = source.index("_record_patch_replay")
        catalog = source.index("_ORIG_APPLY_CATALOG")
        self.assertLess(quiesce, replay)
        self.assertLess(replay, catalog)
        self.assertLess(quiesce, catalog)


    def test_workflow_has_one_input_and_shares_prod_mutation_lock(self) -> None:
        workflow = WORKFLOW.read_text(encoding="utf-8")
        input_block = workflow.split("    inputs:\n", 1)[1].split("\n\npermissions:", 1)[0]
        self.assertIn("release_tag:", input_block)
        self.assertNotIn("candidate", input_block.lower())
        self.assertNotIn("business", input_block.lower())
        self.assertIn("group: prod-release-tidb-boundary", workflow)
        self.assertIn("actions: read", workflow)
        self.assertIn("ref: ${{ github.sha }}", workflow)
        self.assertNotIn("ref: refs/heads/main", workflow)
        self.assertIn(
            '[[ "$(git rev-parse HEAD)" == "${GITHUB_SHA}" ]]', workflow
        )
        credential_scope = workflow.index(
            "- name: Prepare private Docker credential scope"
        )
        registry_login = workflow.index("uses: docker/login-action@")
        self.assertLess(credential_scope, registry_login)
        self.assertEqual(1, workflow.count('echo "DOCKER_CONFIG=${docker_config}"'))
        self.assertEqual(1, workflow.count('echo "D3_WORK_ROOT=${work_root}"'))
        self.assertIn("run_serviceification_d3_prod.py deploy", workflow)
        self.assertIn("--preserve-current-database", workflow)
        self.assertNotIn("Checkout exact Staging patch authority", workflow)
        self.assertNotIn("D3_PATCH_SOURCE_ROOT", workflow)
        self.assertNotIn("D3_PATCH_BUNDLE_PATH", workflow)
        self.assertIn(
            "D3_PATCH_DIR: ${{ steps.release.outputs.patch_dir }}", workflow
        )
        self.assertIn(
            "D3_PATCH_TREE_DIGEST: ${{ steps.release.outputs.patch_tree_digest }}",
            workflow,
        )
        self.assertGreaterEqual(
            workflow.count("PROD_TIDB_HOST: ${{ vars.PROD_TIDB_HOST }}"), 2
        )
        self.assertGreaterEqual(
            workflow.count(
                "PROD_RUNTIME_TIDB_USER: ${{ vars.PROD_RUNTIME_TIDB_USER }}"
            ),
            2,
        )
        guard = workflow.split(
            "- name: Guard Prod D3 host and secret prerequisites", 1
        )[1].split("- name: Prepare private Docker credential scope", 1)[0]
        self.assertIn(
            "NEXUSCRM_INTEGRATION_CONNECTED_MAIL_OAUTH_CLIENT_ID: "
            "${{ secrets.NEXUSCRM_GOOGLE_CLIENT_ID }}",
            guard,
        )
        self.assertIn(
            "NEXUSCRM_INTEGRATION_CONNECTED_MAIL_OAUTH_CLIENT_SECRET: "
            "${{ secrets.NEXUSCRM_GOOGLE_CLIENT_SECRET }}",
            guard,
        )
        self.assertIn("connected_mail_oauth_env(os.environ)", guard)
        bom_login = workflow.index("- name: Log in before immutable BOM admission")
        admission_scope = workflow.index(
            "- name: Register immutable Prod D3 admission workspace"
        )
        verify_bom = workflow.index(
            "- name: Verify and extract exact signed no-Gateway D3 BOM"
        )
        exact_l2 = workflow.index("- name: Checkout exact L2 source")
        prepare = workflow.index("- name: Prepare only the next Prod D3 target runtime")
        self.assertLess(bom_login, admission_scope)
        self.assertLess(admission_scope, verify_bom)
        self.assertLess(verify_bom, exact_l2)
        self.assertLess(exact_l2, prepare)
        self.assertNotIn(
            "prepare_multi_prod_d3_runtime_env.py", workflow[:verify_bom]
        )
        self.assertIn('[[ "${l2_commit}" == "${TAG_TARGET_COMMIT}" ]]', workflow)
        self.assertIn("signed-BOM L2 commit does not equal the D3 Tag target", workflow)
        self.assertIn("Download exact immutable Staging evidence artifact", workflow)
        self.assertIn("--staging-dispatch", workflow)
        self.assertIn("--staging-workflow-run", workflow)
        self.assertIn("--staging-artifact-metadata", workflow)
        self.assertIn("staging-d3-workflow-run.json", workflow)
        self.assertIn("staging-d3-evidence-artifact.json", workflow)
        self.assertNotIn("release_artifact_bucket", workflow.lower())
        self.assertNotIn("business_acceptance", workflow.lower())
        self.assertNotIn("D3_EXPECTED_DESIRED_BOM_DIGEST", workflow)
        self.assertIn("slot-(3103|3104|3105|3106)", workflow)
        admit = workflow.index("- name: Admit exact successful Staging evidence")
        self.assertLess(admit, prepare)
        converge = workflow.index("- name: Converge target D3 completely before cutover")
        readback = workflow.index("- name: Verify cutover readback and current marker")
        self.assertLess(converge, readback)
        self.assertIn(
            "previous runtime retirement: \\`verified after public cutover "
            "and before current publication\\`",
            workflow,
        )
        self.assertNotIn("active-runtime fence:", workflow)
        self.assertIn(".previousRuntimeRetirement", workflow)

    def test_classic_release_workflows_are_retired(self) -> None:
        for workflow_name in (
            "_emit-staging-release-tag-v3.yml",
            "_emit-staging-release-tag-v4.yml",
            "emit-staging-release-tag.yml",
            "deploy-multi-prod-environment.yml",
        ):
            with self.subTest(workflow=workflow_name):
                self.assertFalse(
                    (REPO_ROOT / ".github" / "workflows" / workflow_name).exists()
                )

    def test_classic_prod_release_runtime_is_retired(self) -> None:
        for relative_path in (
            "deploy/online/runtime/check_multi_prod_release_preflight.py",
            "deploy/online/runtime/deploy_multi_prod_env.py",
            "deploy/online/runtime/multi_prod_release_gate_catalog.json",
            "deploy/online/runtime/release_phase_ledger.py",
            "deploy/online/runtime/staging_release_gate_catalog.json",
            "deploy/online/runtime/validate_post_apply_prod_resume.py",
            "scripts/notify_feishu_prod_release.py",
        ):
            with self.subTest(path=relative_path):
                self.assertFalse((REPO_ROOT / relative_path).exists())

    def test_patch_bundle_is_tag_derived_and_not_a_prod_workflow_input(self) -> None:
        workflow = WORKFLOW.read_text(encoding="utf-8")
        runner = inspect.getsource(prod)
        prepare_source = inspect.getsource(prod_prepare.prepare)
        input_block = workflow.split("    inputs:\n", 1)[1].split(
            "\n\npermissions:", 1
        )[0]
        self.assertNotIn("patch_dir:", input_block)
        self.assertIn("D3_PATCH_DIR", workflow)
        self.assertIn("preflight_and_seal_patch", runner)
        self.assertIn("replay_patch_before_catalog", runner)
        self.assertNotIn("D3_PATCH_DIR", prepare_source)

    def test_prod_d3_ports_match_deployment_config_authority(self) -> None:
        config = json.loads(
            (REPO_ROOT / "deployments" / "deployment-config.json").read_text(
                encoding="utf-8"
            )
        )
        targets = config["environments"]["multi_prod"]["d3_topology"][
            "target_config"
        ]
        self.assertEqual(
            {
                slot: {
                    "backend_port": str(contract.sfcrm_port),
                    "workflow_adapter_port": str(contract.adapter_port),
                    "integration_port": str(contract.integration_port),
                    "runtime_grpc_port": str(contract.runtime_grpc),
                    "runtime_admin_port": str(contract.runtime_admin),
                    "worker_admin_base_port": str(contract.worker_admin_base),
                }
                for slot, contract in prod.SLOT_CONTRACT.items()
            },
            targets,
        )
        self.assertEqual(
            ["slot-3103", "slot-3104", "slot-3105", "slot-3106"],
            list(prod.SLOT_CONTRACT),
        )

    def test_all_four_prod_d3_private_port_sets_are_disjoint(self) -> None:
        allocated: dict[int, str] = {}
        for slot, contract in prod.SLOT_CONTRACT.items():
            ports = {
                contract.sfcrm_port,
                contract.adapter_port,
                contract.integration_port,
                contract.runtime_grpc,
                contract.runtime_admin,
                *(contract.worker_admin_base + index for index in range(len(prod.impl.WORKER_STEMS))),
            }
            for port in ports:
                self.assertNotIn(port, allocated, f"{slot} overlaps {allocated.get(port)}")
                allocated[port] = slot

    def test_each_physical_slot_derives_its_complete_runtime_port_contract(self) -> None:
        for slot, contract in prod.SLOT_CONTRACT.items():
            with self.subTest(slot=slot):
                ports = prod.derive_ports(
                    {
                        "ENV_NAME": f"multi-prod-d3-{slot}",
                        "BACKEND_PORT": str(contract.sfcrm_port),
                        "EXP_HUBSPOT_OPTIONS_ADAPTER_PORT": str(contract.adapter_port),
                        "SERVER_NAME": "nexuscrm-new-prod.pingcap.com",
                    },
                    slot,
                )
                self.assertEqual(contract.sfcrm_port, ports.sfcrm_http)
                self.assertEqual(contract.integration_port, ports.integration_http)
                self.assertEqual(contract.runtime_grpc, ports.runtime_grpc)
                self.assertEqual(contract.runtime_admin, ports.runtime_admin)


if __name__ == "__main__":
    unittest.main()