fix(site-publish): retain immutable history across deploys

Authored-By: @architect <architect@fritzlab.net>

README.md

@@ -92,8 +92,10 @@ routes:
 ```
 
 The caller supplies each declared credential name as an environment variable
-on the action step. Credential values are passed to `aws` only through its
+on the action step. Names must be matched `<NAME>_S3_ACCESS_KEY` and
-environment and never appear in a logged command or process argument.
+`<NAME>_S3_SECRET_KEY` pairs; arbitrary environment variables cannot become
+publication credentials. Values pass to `aws` only through its environment and
+never appear in a logged command or process argument.
 
 ```yaml
       - uses: https://code.fritzlab.net/action/site-publish@v1
@@ -126,6 +128,24 @@ cache policy, content type, and bytes. That content address makes concurrent
 writes identical even though Garage v2.2.0 has no conditional destination
 write. An identical retry converges; a changed object, missing digest metadata,
 wrong address, or nested policy under that immutable prefix fails publication.
+Every immutable target across every artifact is validated and published before
+any route's mutable objects change.
+Mutable default and override partitions receive their final cache policy before
+the matching prefix-scoped stale deletion, so publication never exposes a
+provisional cache policy or a pointer to a missing immutable target.
+An append-only `site-publish-history.json` beside each generated site retains every bucket's access
+class and absolute immutable prefixes, including removed routes and rules. When a route move places
+a retired prefix inside the new sync scope, that subtree is excluded; a current-file collision
+fails publication. Protected access remains sticky across artifact renames and legacy mode, so
+publishing the same artifact publicly requires a new bucket. Decommissioning removes the live
+application and manifests while retaining this history because its Garage bucket is not purged.
+
+Artifact input directories must be pairwise disjoint after filesystem
+resolution. Publication stops before build or upload if one contains another or
+escapes the repository. Symlinked roots, components, and descendants are also rejected, preventing
+protected input from entering a public artifact through dereference. Split
+storage endpoints are pinned to Garage, and each website
+authority is derived from its bucket; a site cannot expose an arbitrary backend.
 
 Each split route gets a bucket-specific `<bucket>.web.sjc001.fritzlab.net`
 ExternalName Service annotated to disable pass-host-header and a separate Ingress. Route

scripts/build.py

@@ -5,7 +5,7 @@ import subprocess
 import tempfile
 from pathlib import Path
 
-from utils import EXCLUDE_FILES, env, parse_site_yaml, run
+from utils import EXCLUDE_FILES, env, parse_site_yaml, run, validate_artifact_inputs
 
 
 def build_artifact(site_dir, artifact):
@@ -64,5 +64,7 @@ def cmd_build():
         print("Site disabled — skipping build")
         return
 
+    validate_artifact_inputs(site_dir, cfg)
+
     for artifact in cfg["artifacts"]:
         build_artifact(site_dir, artifact)

scripts/deploy.py

@@ -8,8 +8,7 @@ import os
 import re
 import shutil
 import subprocess
-import tempfile
+from pathlib import Path, PurePosixPath
-from pathlib import Path
 from urllib.error import HTTPError, URLError
 from urllib.request import Request, urlopen
 
@@ -23,11 +22,13 @@ from utils import (
     parse_site_yaml,
     render_templates,
     run,
+    validate_artifact_inputs,
 )
 
 GARAGE_ADMIN_ENDPOINT = os.environ.get(
     "GARAGE_ADMIN_ENDPOINT", "http://garage.storage.svc:3903"
 )
+HISTORY_FILE = "site-publish-history.json"
 
 
 def validate_artifact_output(site_dir, artifact):
@@ -149,9 +150,8 @@ def publish_immutable_rule(artifact, route, rule, html_dir, aws_env):
         )
 
 
-def s3_sync(artifact, route, site_dir, credential_env_names=None):
+def publication_aws_env(artifact, credential_env_names=None):
-    endpoint = artifact["s3_endpoint"]
+    """Build the route-scoped AWS environment without leaking other credentials."""
-    html_dir = site_dir / artifact["build_dir"]
     access_key = env(artifact["credentials"]["access_key_env"])
     secret_key = env(artifact["credentials"]["secret_key_env"])
     aws_env = os.environ.copy()
@@ -165,10 +165,58 @@ def s3_sync(artifact, route, site_dir, credential_env_names=None):
         "AWS_SECRET_ACCESS_KEY": secret_key,
         "AWS_DEFAULT_REGION": os.environ.get("AWS_DEFAULT_REGION", "sjc001"),
     })
+    return aws_env
+
+
+def publish_route_immutables(artifact, route, site_dir, credential_env_names=None):
+    """Publish one route's immutable partitions during the global preflight."""
+    html_dir = site_dir / artifact["build_dir"]
+    aws_env = publication_aws_env(artifact, credential_env_names)
+    for rule in artifact["cache_rules"]:
+        if _is_immutable(rule):
+            publish_immutable_rule(artifact, route, rule, html_dir, aws_env)
+
+
+def retired_immutable_filters(artifact, route, html_dir, previous_contract):
+    """Protect every historical immutable prefix inside the current sync scope."""
+    if not previous_contract:
+        return []
+    current_prefix = route["path"].strip("/")
+    filters = []
+    current_immutable = set(immutable_prefixes(artifact, route))
+    for immutable_prefix in previous_contract["immutable_prefixes"]:
+        if current_prefix:
+            marker = f"{current_prefix}/"
+            if not immutable_prefix.startswith(marker):
+                continue
+            relative_path = immutable_prefix[len(marker):]
+        else:
+            relative_path = immutable_prefix
+        collision = html_dir / relative_path
+        if (immutable_prefix not in current_immutable and collision.exists()
+                and any(path.is_file() for path in collision.rglob("*"))):
+            raise RuntimeError(
+                f"current artifact collides with retired immutable partition: {relative_path}"
+            )
+        filters.extend(("--exclude", f"{relative_path}/*"))
+    return filters
+
+
+def immutable_prefixes(artifact, route):
+    route_prefix = route["path"].strip("/")
+    return sorted({
+        "/".join(part for part in (route_prefix, rule["path"]) if part)
+        for rule in artifact["cache_rules"] if _is_immutable(rule)
+    })
+
+
+def s3_sync(artifact, route, site_dir, credential_env_names=None, previous_contract=None):
+    endpoint = artifact["s3_endpoint"]
+    html_dir = site_dir / artifact["build_dir"]
+    aws_env = publication_aws_env(artifact, credential_env_names)
     bucket = artifact["bucket"]
     object_prefix = route["path"].strip("/")
-    bucket_destination = f"s3://{bucket}/"
+    destination = f"s3://{bucket}/{object_prefix + '/' if object_prefix else ''}"
-    destination = f"{bucket_destination}{object_prefix + '/' if object_prefix else ''}"
     default_cache = next(rule["cache_control"] for rule in artifact["cache_rules"] if not rule["path"])
     immutable_paths = [rule["path"] for rule in artifact["cache_rules"] if _is_immutable(rule)]
     # `excludes` are patterns (site.yaml `excludes:` list) that should never
@@ -178,48 +226,29 @@ def s3_sync(artifact, route, site_dir, credential_env_names=None):
     if artifact["excludes"]:
         print(f"Excluding patterns: {artifact['excludes']}")
     print(f"Syncing artifact {artifact['name']} → {destination} via {endpoint}")
-    # `sync --delete` handles new/changed/orphaned files. Partitioned
+    # Upload with the final cache policy before cleanup. Sync and deletion are
-    # `cp --recursive` calls then re-upload each file once to refresh metadata
+    # scoped to the same current route prefix and cache partition. A route move
-    # (cache-control, content-type) on objects sync skipped as unchanged.
+    # leaves its old bucket partition intact but unreachable after the old
+    # Ingress disappears, while stale mutable keys on the serving prefix are
+    # deleted. Immutable subtrees are structurally excluded. `cp --recursive`
+    # refreshes metadata atomically per object before `sync --delete` removes
+    # stale keys without ever exposing new bytes under a provisional policy.
     # A no-op deploy therefore transfers the artifact bytes once.
     # AWS CLI guesses Content-Type from file extension on local→S3 uploads,
     # so a fresh upload always carries the right MIME type.
-    stage = None
-    sync_source = html_dir
-    sync_excludes = [*exclude_args,
-                     *(arg for path in immutable_paths for arg in ("--exclude", f"{path}/*"))]
-    if object_prefix:
-        stage = tempfile.TemporaryDirectory()
-        sync_source = Path(stage.name)
-        staged_artifact = sync_source / object_prefix
-        staged_artifact.parent.mkdir(parents=True, exist_ok=True)
-        staged_artifact.symlink_to(html_dir.resolve(), target_is_directory=True)
-        sync_excludes = [
-            *(arg for pattern in artifact["excludes"]
-              for arg in ("--exclude", f"{object_prefix}/{pattern}")),
-            *(arg for path in immutable_paths
-              for arg in ("--exclude", f"{object_prefix}/{path}/*")),
-        ]
-    try:
-        # Sync the complete bucket authority so moving a route prefix also
-        # deletes objects under its old prefix instead of leaving them public.
-        run(["aws", "--endpoint-url", endpoint, "s3", "sync", f"{sync_source}/", bucket_destination,
-             "--delete", "--only-show-errors", "--cache-control", default_cache,
-             *sync_excludes], env=aws_env)
-    finally:
-        if stage:
-            stage.cleanup()
-    print("Re-stamping metadata on all objects...")
     specific_paths = [rule["path"] for rule in artifact["cache_rules"] if rule["path"]]
     default_filters = [arg for path in specific_paths for arg in ("--exclude", f"{path}/*")]
+    retired_filters = retired_immutable_filters(artifact, route, html_dir, previous_contract)
     run(["aws", "--endpoint-url", endpoint, "s3", "cp", f"{html_dir}/", destination,
          "--recursive", "--only-show-errors", "--cache-control", default_cache,
-         *default_filters, *exclude_args], env=aws_env)
+         *default_filters, *retired_filters, *exclude_args], env=aws_env)
+    run(["aws", "--endpoint-url", endpoint, "s3", "sync", f"{html_dir}/", destination,
+         "--delete", "--only-show-errors", "--cache-control", default_cache,
+         *default_filters, *retired_filters, *exclude_args], env=aws_env)
     for rule in artifact["cache_rules"]:
         if not rule["path"]:
             continue
         if _is_immutable(rule):
-            publish_immutable_rule(artifact, route, rule, html_dir, aws_env)
             continue
         include = f"{rule['path'].rstrip('/')}/*"
         child_filters = [arg for path in specific_paths
@@ -230,6 +259,9 @@ def s3_sync(artifact, route, site_dir, credential_env_names=None):
         run(["aws", "--endpoint-url", endpoint, "s3", "cp", f"{html_dir}/", destination,
              "--recursive", "--only-show-errors", "--cache-control", rule["cache_control"],
              "--exclude", "*", "--include", include, *child_filters, *exclude_args], env=aws_env)
+        run(["aws", "--endpoint-url", endpoint, "s3", "sync", f"{html_dir}/", destination,
+             "--delete", "--only-show-errors", "--cache-control", rule["cache_control"],
+             "--exclude", "*", "--include", include, *child_filters, *exclude_args], env=aws_env)
 
 
 def garage_admin(method, path, token, body=None):
@@ -300,6 +332,99 @@ def render_site_manifests(site_name, action_dir, app_dir, manifests_dir, cfg):
     render_templates(action_dir, template_vars, app_dir, manifests_dir)
 
 
+def previous_route_contracts(app_dir):
+    """Read the append-only bucket history kept beside generated manifests."""
+    path = app_dir / HISTORY_FILE
+    if not path.exists():
+        return {}
+    try:
+        document = json.loads(path.read_text())
+    except (OSError, json.JSONDecodeError) as exc:
+        raise RuntimeError(f"invalid site-publish route history in {path}") from exc
+    if (not isinstance(document, dict) or set(document) != {"schemaVersion", "buckets"}
+            or document["schemaVersion"] != 1 or not isinstance(document["buckets"], dict)):
+        raise RuntimeError(f"invalid site-publish route history in {path}")
+    for bucket, contract in document["buckets"].items():
+        if (not isinstance(bucket, str) or not isinstance(contract, dict)
+                or set(contract) != {"access", "artifact", "immutablePrefixes", "routePath"}
+                or contract["access"] not in {"legacy", "protected", "public"}
+                or not isinstance(contract["artifact"], str)
+                or not isinstance(contract["routePath"], str)
+                or not contract["routePath"].startswith("/")
+                or not isinstance(contract["immutablePrefixes"], list)
+                or any(not _valid_immutable_prefix(value)
+                       for value in contract["immutablePrefixes"])
+                or len(contract["immutablePrefixes"]) != len(set(contract["immutablePrefixes"]))):
+            raise RuntimeError(f"invalid site-publish route history in {path}")
+    return {
+        bucket: {
+            "access": contract["access"],
+            "artifact": contract["artifact"],
+            "immutable_prefixes": sorted(contract["immutablePrefixes"]),
+            "path": contract["routePath"],
+        }
+        for bucket, contract in document["buckets"].items()
+    }
+
+
+def next_route_contracts(cfg, previous_contracts):
+    """Carry protected access and immutable prefixes forward for every known bucket."""
+    contracts = json.loads(json.dumps(previous_contracts))
+    artifacts = {artifact["name"]: artifact for artifact in cfg["artifacts"]}
+    for route in cfg["routes"]:
+        artifact = artifacts[route["artifact"]]
+        bucket = artifact["bucket"]
+        previous = previous_contracts.get(bucket)
+        access = "protected" if (
+            route["access"] == "protected" or previous and previous["access"] == "protected"
+        ) else route["access"]
+        contracts[bucket] = {
+            "access": access,
+            "artifact": route["artifact"],
+            "immutable_prefixes": sorted(set(
+                (previous or {}).get("immutable_prefixes", []) + immutable_prefixes(artifact, route)
+            )),
+            "path": route["path"],
+        }
+    return contracts
+
+
+def write_route_contracts(app_dir, contracts):
+    app_dir.mkdir(parents=True, exist_ok=True)
+    document = {
+        "schemaVersion": 1,
+        "buckets": {
+            bucket: {
+                "access": contract["access"],
+                "artifact": contract["artifact"],
+                "immutablePrefixes": contract["immutable_prefixes"],
+                "routePath": contract["path"],
+            }
+            for bucket, contract in sorted(contracts.items())
+        },
+    }
+    (app_dir / HISTORY_FILE).write_text(f"{json.dumps(document, indent=2, sort_keys=True)}\n")
+
+
+def _valid_immutable_prefix(value):
+    return (isinstance(value, str) and value and not value.startswith("/")
+            and not value.endswith("/") and ".." not in PurePosixPath(value).parts)
+
+
+def validate_route_migrations(cfg, previous_contracts):
+    artifacts = {artifact["name"]: artifact for artifact in cfg["artifacts"]}
+    for route in cfg["routes"]:
+        artifact = artifacts[route["artifact"]]
+        previous = previous_contracts.get(artifact["bucket"])
+        if (previous and previous["access"] == "protected"
+                and route["access"] in {"public", "legacy"}
+        ):
+            raise RuntimeError(
+                f"artifact {route['artifact']} cannot become public while reusing protected "
+                f"bucket {artifact['bucket']}"
+            )
+
+
 def deploy_static(site_name, site_dir, action_dir, token, cfg):
     artifact_by_name = {artifact["name"]: artifact for artifact in cfg["artifacts"]}
     credential_env_names = {
@@ -308,15 +433,27 @@ def deploy_static(site_name, site_dir, action_dir, token, cfg):
     validate_publication_environment(cfg)
     for artifact in cfg["artifacts"]:
         validate_artifact_output(site_dir, artifact)
-    for route in cfg["routes"]:
-        s3_sync(artifact_by_name[route["artifact"]], route, site_dir, credential_env_names)
-    if cfg["compatibility"]:
-        ensure_bucket_aliases(site_name, cfg["aliases"], os.environ.get("GARAGE_ADMIN_TOKEN"))
-
     apps_dir = clone_apps(token)
     app_dir = apps_dir / "sjc001" / "websites" / site_name
     manifests_dir = app_dir / "manifests"
+    previous_contracts = previous_route_contracts(app_dir)
+    validate_route_migrations(cfg, previous_contracts)
+    # Complete immutable work across the whole publication before any route's
+    # mutable pointers can change. Partial immutable success is safe; mixing a
+    # new route with an old route after a later immutable failure is not.
+    for route in cfg["routes"]:
+        publish_route_immutables(
+            artifact_by_name[route["artifact"]], route, site_dir, credential_env_names,
+        )
+    for route in cfg["routes"]:
+        s3_sync(
+            artifact_by_name[route["artifact"]], route, site_dir, credential_env_names,
+            previous_contracts.get(artifact_by_name[route["artifact"]]["bucket"]),
+        )
+    if cfg["compatibility"]:
+        ensure_bucket_aliases(site_name, cfg["aliases"], os.environ.get("GARAGE_ADMIN_TOKEN"))
 
+    write_route_contracts(app_dir, next_route_contracts(cfg, previous_contracts))
     render_site_manifests(site_name, action_dir, app_dir, manifests_dir, cfg)
 
     commit_and_push(apps_dir, f"Deploy {site_name}", token)
@@ -329,7 +466,12 @@ def decommission(site_name, token, buckets=None):
     if not site_path.exists():
         print(f"No manifests for {site_name} — nothing to remove")
         return
+    history_path = site_path / HISTORY_FILE
+    history = history_path.read_bytes() if history_path.exists() else None
     shutil.rmtree(site_path)
+    if history is not None:
+        site_path.mkdir(parents=True)
+        (site_path / HISTORY_FILE).write_bytes(history)
     commit_and_push(apps_dir, f"Decommission {site_name}", token)
     for bucket in buckets or [site_name]:
         print(f"Bucket {bucket} and its objects are NOT purged automatically.")
@@ -350,4 +492,5 @@ def cmd_deploy():
         decommission(site_name, token, [artifact["bucket"] for artifact in cfg["artifacts"]])
         return
 
+    validate_artifact_inputs(site_dir, cfg)
     deploy_static(site_name, site_dir, action_dir, token, cfg)

scripts/utils.py

@@ -26,6 +26,7 @@ EXCLUDE_FILES = {
 VALID_TYPES = {"static", "hugo", "mkdocs"}
 NAME_RE = re.compile(r"^[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$")
 ENV_RE = re.compile(r"^[A-Z_][A-Z0-9_]*$")
+ACCESS_KEY_ENV_RE = re.compile(r"^([A-Z][A-Z0-9_]*)_S3_ACCESS_KEY$")
 BUCKET_RE = re.compile(r"^[a-z0-9](?:[a-z0-9.-]{1,61}[a-z0-9])?$")
 MIDDLEWARE_RE = re.compile(r"^[A-Za-z0-9](?:[A-Za-z0-9_-]{0,61}[A-Za-z0-9])?$")
 
@@ -238,7 +239,7 @@ def _artifact(item, index):
     if not isinstance(name, str) or not NAME_RE.fullmatch(name):
         raise ConfigError(f"{label}.name must be a DNS label")
     publish = _mapping(item.get("publish"), f"{label}.publish")
-    _known_keys(publish, {"bucket", "endpoint", "website_authority", "credentials"}, f"{label}.publish")
+    _known_keys(publish, {"bucket", "credentials"}, f"{label}.publish")
     bucket = publish.get("bucket")
     if not isinstance(bucket, str) or not BUCKET_RE.fullmatch(bucket):
         raise ConfigError(f"{label}.publish.bucket is not a valid bucket name")
@@ -250,6 +251,15 @@ def _artifact(item, index):
         if not isinstance(value, str) or not ENV_RE.fullmatch(value):
             raise ConfigError(f"{label}.publish.credentials.{key} must name an environment variable")
         normalized_credentials[key] = value
+    access_match = ACCESS_KEY_ENV_RE.fullmatch(normalized_credentials["access_key_env"])
+    expected_secret = (
+        f"{access_match.group(1)}_S3_SECRET_KEY" if access_match else None
+    )
+    if normalized_credentials["secret_key_env"] != expected_secret:
+        raise ConfigError(
+            f"{label}.publish.credentials must be a matched "
+            "<NAME>_S3_ACCESS_KEY and <NAME>_S3_SECRET_KEY pair"
+        )
     cache = _mapping(item.get("cache"), f"{label}.cache")
     _known_keys(cache, {"rules"}, f"{label}.cache")
     rules = _list(cache.get("rules"), f"{label}.cache.rules")
@@ -278,10 +288,7 @@ def _artifact(item, index):
     for path in immutable_paths:
         if any(other.startswith(f"{path}/") for other in paths):
             raise ConfigError(f"{label}.cache.rules cannot nest another policy under immutable /{path}")
-    authority = _hostname(
+    authority = f"{bucket}.{DEFAULT_WEBSITE_SUFFIX}"
-        publish.get("website_authority", f"{bucket}.{DEFAULT_WEBSITE_SUFFIX}"),
-        f"{label}.publish.website_authority",
-    )
     if not isinstance(item.get("tidy", True), bool):
         raise ConfigError(f"{label}.tidy must be a boolean")
     return {
@@ -292,7 +299,7 @@ def _artifact(item, index):
         "excludes": _strings(item.get("excludes") or [], f"{label}.excludes"),
         "build_dir": f".site-publish/{name}/html",
         "bucket": bucket,
-        "s3_endpoint": _endpoint(publish.get("endpoint", DEFAULT_S3_ENDPOINT), f"{label}.publish.endpoint"),
+        "s3_endpoint": DEFAULT_S3_ENDPOINT,
         "website_authority": authority,
         "credentials": normalized_credentials,
         "cache_rules": cache_rules,
@@ -418,6 +425,42 @@ def normalize_site_config(raw, site_name):
     return cfg
 
 
+def validate_artifact_inputs(site_dir, cfg):
+    """Reject source containment before a public or protected build starts."""
+    if cfg["compatibility"]:
+        return
+    root = Path(site_dir).resolve()
+    sources = []
+    for artifact in cfg["artifacts"]:
+        declared = root
+        for component in Path(artifact["content_dir"]).parts:
+            declared /= component
+            if declared.is_symlink():
+                raise ConfigError(
+                    f"artifact {artifact['name']} content_dir contains symlink component: "
+                    f"{declared.relative_to(root)}"
+                )
+        source = declared.resolve()
+        if source != root and root not in source.parents:
+            raise ConfigError(
+                f"artifact {artifact['name']} content_dir resolves outside the repository"
+            )
+        if source.exists():
+            symlink = next((path for path in source.rglob("*") if path.is_symlink()), None)
+            if symlink is not None:
+                raise ConfigError(
+                    f"artifact {artifact['name']} build input contains symlink: "
+                    f"{symlink.relative_to(root)}"
+                )
+        sources.append((artifact["name"], source))
+    for index, (name, source) in enumerate(sources):
+        for other_name, other_source in sources[index + 1:]:
+            if source == other_source or source in other_source.parents or other_source in source.parents:
+                raise ConfigError(
+                    f"artifact build inputs overlap after resolution: {name} and {other_name}"
+                )
+
+
 def parse_site_yaml(site_dir, site_name=None):
     path = Path(site_dir) / "site.yaml"
     if not path.exists():

tests/test_contract.py

@@ -24,7 +24,7 @@ sys.path.insert(0, str(ROOT / "scripts"))
 import deploy
 import build
 import utils
-from utils import ConfigError, normalize_site_config
+from utils import ConfigError, normalize_site_config, validate_artifact_inputs
 
 
 def fixture(name):
@@ -151,12 +151,11 @@ class ConfigContractTests(unittest.TestCase):
         )
 
     def test_publication_credentials_cannot_be_reused(self):
-        self.assert_invalid(
+        def mutate(raw):
-            lambda raw: raw["artifacts"][0]["publish"]["credentials"].__setitem__(
+            raw["artifacts"][0]["publish"]["credentials"] = copy.deepcopy(
-                "access_key_env", "DIST_S3_ACCESS_KEY"
+                raw["artifacts"][1]["publish"]["credentials"]
-            ),
-            "publication credential DIST_S3_ACCESS_KEY is reused",
             )
+        self.assert_invalid(mutate, "publication credential DIST_S3_ACCESS_KEY is reused")
 
     def test_cache_directive_contradiction_is_rejected(self):
         self.assert_invalid(
@@ -191,6 +190,56 @@ class ConfigContractTests(unittest.TestCase):
             "unknown fields: storage_bucket",
         )
 
+    def test_backend_authority_and_endpoint_are_derived(self):
+        for field, value in (
+            ("endpoint", "https://attacker.example"),
+            ("website_authority", "internal-api.default.svc.k8s.sjc001.fritzlab.net"),
+        ):
+            with self.subTest(field=field):
+                self.assert_invalid(
+                    lambda raw, field=field, value=value: raw["artifacts"][0]["publish"].__setitem__(field, value),
+                    f"unknown fields: {field}",
+                )
+
+    def test_publication_credentials_use_dedicated_matched_names(self):
+        self.assert_invalid(
+            lambda raw: raw["artifacts"][0]["publish"]["credentials"].__setitem__(
+                "access_key_env", "CI_BOT_TOKEN"
+            ),
+            "must be a matched <NAME>_S3_ACCESS_KEY",
+        )
+
+    def test_resolved_build_inputs_must_be_pairwise_disjoint(self):
+        raw = copy.deepcopy(self.raw)
+        raw["artifacts"][1]["content_dir"] = ""
+        cfg = normalize_site_config(raw, "baseline.fritzlab.net")
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            (root / "portal" / "build").mkdir(parents=True)
+            with self.assertRaisesRegex(ConfigError, "build inputs overlap after resolution"):
+                validate_artifact_inputs(root, cfg)
+
+    def test_descendant_symlink_cannot_cross_artifact_boundary(self):
+        cfg = normalize_site_config(self.raw, "baseline.fritzlab.net")
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            (root / "dist").mkdir()
+            (root / "portal" / "build").mkdir(parents=True)
+            (root / "portal" / "build" / "private.txt").write_text("private")
+            (root / "dist" / "portal-link").symlink_to(root / "portal" / "build")
+            with self.assertRaisesRegex(ConfigError, "build input contains symlink"):
+                validate_artifact_inputs(root, cfg)
+
+    def test_artifact_root_symlink_is_rejected_before_resolution(self):
+        cfg = normalize_site_config(self.raw, "baseline.fritzlab.net")
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            (root / "dist-real").mkdir()
+            (root / "dist").symlink_to(root / "dist-real")
+            (root / "portal" / "build").mkdir(parents=True)
+            with self.assertRaisesRegex(ConfigError, "content_dir contains symlink component"):
+                validate_artifact_inputs(root, cfg)
+
 
 class GenerationTests(unittest.TestCase):
     def render(self, raw):
@@ -224,6 +273,18 @@ class GenerationTests(unittest.TestCase):
         self.assertNotIn("passhostheader", files["manifests/ingress-portal.yaml"])
         self.assertIn("baseline-dist.web.sjc001.fritzlab.net", files["manifests/service-distributions.yaml"])
 
+    def test_history_keeps_yaml_ambiguous_artifact_names_as_strings(self):
+        raw = fixture("split-site.yaml")
+        raw["artifacts"][0]["name"] = "yes"
+        raw["routes"][0]["artifact"] = "yes"
+        cfg = normalize_site_config(raw, "baseline.fritzlab.net")
+        contracts = deploy.next_route_contracts(cfg, {})
+        with tempfile.TemporaryDirectory() as tmp:
+            app_dir = Path(tmp)
+            deploy.write_route_contracts(app_dir, contracts)
+            restored = deploy.previous_route_contracts(app_dir)
+        self.assertEqual("yes", restored["baseline-portal"]["artifact"])
+
     def test_generation_is_deterministic_when_input_lists_are_reversed(self):
         raw = fixture("split-site.yaml")
         first_tmp, _, first = self.render(raw)
@@ -247,12 +308,14 @@ class GenerationTests(unittest.TestCase):
             self.assertFalse(stale.exists())
 
     def test_legacy_names_and_garage_s3_target_are_preserved(self):
-        tmp, _, files = self.render(fixture("legacy-site.yaml"))
+        tmp, app_dir, files = self.render(fixture("legacy-site.yaml"))
         self.addCleanup(tmp.cleanup)
         self.assertIn("manifests/service.yaml", files)
         self.assertIn("manifests/ingress.yaml", files)
         self.assertIn("garage-s3.storage.svc.k8s.sjc001.fritzlab.net", files["manifests/service.yaml"])
         self.assertNotIn("passhostheader", files["manifests/ingress.yaml"])
+        self.assertNotIn("site-publish.fritzlab.net", files["manifests/ingress.yaml"])
+        self.assertEqual({}, deploy.previous_route_contracts(app_dir))
 
 
 class BuildTests(unittest.TestCase):
@@ -351,27 +414,35 @@ class PublishingTests(unittest.TestCase):
             (html / "channels").mkdir()
             (html / "releases" / "1.0.js").write_text("release")
             (html / "channels" / "stable.json").write_text("channel")
-            commands = []
+            commands, events = [], []
 
             def capture(command, **kwargs):
                 commands.append((command, kwargs["env"]))
+                events.append("mutable")
+
+            def publish_immutable(*_args):
+                events.append("immutable")
 
             secret = "secret-must-not-appear"
             output = io.StringIO()
             with patch.dict(os.environ, {
                 "DIST_S3_ACCESS_KEY": "dist-key", "DIST_S3_SECRET_KEY": secret
             }, clear=False), patch.object(deploy, "run", side_effect=capture), \
-                    patch.object(deploy, "publish_immutable_rule") as immutable_publish, \
+                    patch.object(deploy, "publish_immutable_rule", side_effect=publish_immutable) as immutable_publish, \
                     redirect_stdout(output):
+                deploy.publish_route_immutables(artifact, route, root)
                 deploy.s3_sync(artifact, route, root)
 
             self.assertTrue(all(secret not in " ".join(command) for command, _ in commands))
+            self.assertEqual("immutable", events[0])
             self.assertNotIn(secret, output.getvalue())
             self.assertTrue(all(call_env["AWS_ACCESS_KEY_ID"] == "dist-key" for _, call_env in commands))
             self.assertTrue(all("DIST_S3_SECRET_KEY" not in call_env for _, call_env in commands))
             rendered = [" ".join(command) for command, _ in commands]
-            self.assertIn("s3://baseline-dist/", rendered[0])
+            self.assertIn("s3://baseline-dist/dist/", rendered[0])
-            self.assertIn("dist/releases/*", rendered[0])
+            self.assertIn("releases/*", rendered[0])
+            self.assertNotIn("--delete", rendered[0])
+            self.assertIn("--delete", rendered[1])
             self.assertTrue(all("s3://baseline-dist/dist/" in command for command in rendered[1:]))
             self.assertTrue(any("channels/" in command and
                                 "public, max-age=0, must-revalidate" in command
@@ -382,6 +453,148 @@ class PublishingTests(unittest.TestCase):
                 immutable_publish.call_args.args[2]["cache_control"],
             )
 
+    def test_root_move_preserves_only_actual_retired_immutable_prefix(self):
+        cfg = normalize_site_config(fixture("split-site.yaml"), "baseline.fritzlab.net")
+        artifact = next(item for item in cfg["artifacts"] if item["name"] == "distributions")
+        route = {**next(item for item in cfg["routes"] if item["artifact"] == "distributions"),
+                 "path": "/"}
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            html = root / artifact["build_dir"]
+            (html / "releases").mkdir(parents=True)
+            (html / "channels").mkdir()
+            (html / "docs" / "releases").mkdir(parents=True)
+            (html / "channels" / "stable.json").write_text("channel")
+            (html / "docs" / "releases" / "index.html").write_text("mutable")
+            commands = []
+            with patch.dict(os.environ, {
+                "DIST_S3_ACCESS_KEY": "dist-key", "DIST_S3_SECRET_KEY": "dist-secret"
+            }, clear=False), patch.object(
+                deploy, "run", side_effect=lambda command, **_: commands.append(command)
+            ):
+                deploy.s3_sync(artifact, route, root, previous_contract={
+                    "path": "/foo", "access": "public", "artifact": "distributions",
+                    "immutable_prefixes": ["foo/releases"],
+                })
+            rendered = [" ".join(command) for command in commands]
+            self.assertTrue(all("foo/releases/*" in command for command in rendered[:2]))
+            self.assertTrue(all("*/releases/*" not in command for command in rendered))
+
+    def test_protected_bucket_cannot_become_public_across_deployments(self):
+        cfg = normalize_site_config(fixture("split-site.yaml"), "baseline.fritzlab.net")
+        previous = {
+            "baseline-dist": {
+                "path": "/dist", "access": "protected", "artifact": "old-name",
+                "immutable_prefixes": ["dist/releases"],
+            }
+        }
+        with self.assertRaisesRegex(RuntimeError, "cannot become public while reusing protected"):
+            deploy.validate_route_migrations(cfg, previous)
+        renamed = copy.deepcopy(cfg)
+        artifact = next(item for item in renamed["artifacts"] if item["name"] == "distributions")
+        artifact["name"] = "downloads"
+        route = next(item for item in renamed["routes"] if item["artifact"] == "distributions")
+        route["artifact"] = "downloads"
+        with self.assertRaisesRegex(RuntimeError, "cannot become public while reusing protected"):
+            deploy.validate_route_migrations(renamed, previous)
+        previous = {"retired-protected-bucket": previous["baseline-dist"]}
+        deploy.validate_route_migrations(cfg, previous)
+
+    def test_protected_split_bucket_cannot_become_legacy_public(self):
+        cfg = normalize_site_config(fixture("legacy-site.yaml"), "baseline.fritzlab.net")
+        previous = {
+            "baseline.fritzlab.net": {
+                "path": "/portal", "access": "protected", "artifact": "portal",
+                "immutable_prefixes": [],
+            }
+        }
+        with self.assertRaisesRegex(RuntimeError, "cannot become public while reusing protected"):
+            deploy.validate_route_migrations(cfg, previous)
+
+    def test_removed_immutable_rule_preserves_prior_keys(self):
+        cfg = normalize_site_config(fixture("split-site.yaml"), "baseline.fritzlab.net")
+        artifact = next(item for item in cfg["artifacts"] if item["name"] == "distributions")
+        artifact["cache_rules"] = [
+            rule for rule in artifact["cache_rules"] if rule["path"] != "releases"
+        ]
+        route = next(item for item in cfg["routes"] if item["artifact"] == "distributions")
+        with tempfile.TemporaryDirectory() as tmp:
+            html = Path(tmp)
+            filters = deploy.retired_immutable_filters(artifact, route, html, {
+                "path": "/dist", "access": "public", "artifact": "distributions",
+                "immutable_prefixes": ["dist/releases"],
+            })
+            self.assertEqual(["--exclude", "releases/*"], filters)
+            previous = {
+                "baseline-dist": {
+                    "path": "/dist", "access": "public", "artifact": "distributions",
+                    "immutable_prefixes": ["dist/releases"],
+                }
+            }
+            next_contracts = deploy.next_route_contracts(cfg, previous)
+            self.assertEqual(
+                ["dist/releases"], next_contracts["baseline-dist"]["immutable_prefixes"],
+            )
+            self.assertEqual(
+                ["--exclude", "releases/*"],
+                deploy.retired_immutable_filters(
+                    artifact, route, html, next_contracts["baseline-dist"],
+                ),
+            )
+            (html / "releases").mkdir()
+            (html / "releases" / "replacement.js").write_text("mutable")
+            with self.assertRaisesRegex(RuntimeError, "collides with retired immutable"):
+                deploy.retired_immutable_filters(artifact, route, html, {
+                    "path": "/dist", "access": "public", "artifact": "distributions",
+                    "immutable_prefixes": ["dist/releases"],
+                })
+
+    def test_removed_route_history_remains_append_only(self):
+        cfg = normalize_site_config(fixture("split-site.yaml"), "baseline.fritzlab.net")
+        previous = {
+            "retired-bucket": {
+                "path": "/retired", "access": "protected", "artifact": "retired",
+                "immutable_prefixes": ["retired/releases"],
+            }
+        }
+        contracts = deploy.next_route_contracts(cfg, previous)
+        self.assertEqual(previous["retired-bucket"], contracts["retired-bucket"])
+
+    def test_decommission_preserves_history_for_an_unpurged_bucket(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            apps = Path(tmp)
+            site = apps / "sjc001/websites/baseline"
+            site.mkdir(parents=True)
+            history = b'{"schemaVersion":1,"buckets":{}}\n'
+            (site / deploy.HISTORY_FILE).write_bytes(history)
+            (site / "app.yaml").write_text("live\n")
+            with patch.object(deploy, "clone_apps", return_value=apps), patch.object(
+                deploy, "commit_and_push"
+            ) as commit:
+                deploy.decommission("baseline", "token", ["baseline-dist"])
+            self.assertEqual(history, (site / deploy.HISTORY_FILE).read_bytes())
+            self.assertFalse((site / "app.yaml").exists())
+            commit.assert_called_once_with(apps, "Decommission baseline", "token")
+
+    def test_later_route_immutable_failure_stops_all_mutable_publication(self):
+        cfg = normalize_site_config(fixture("split-site.yaml"), "baseline.fritzlab.net")
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            with patch.object(deploy, "validate_publication_environment"), \
+                    patch.object(deploy, "validate_artifact_output"), \
+                    patch.object(deploy, "clone_apps", return_value=root / "apps"), patch.object(
+                        deploy, "publish_route_immutables",
+                        side_effect=[None, RuntimeError("immutable failed")],
+                    ) as immutable_publish, patch.object(deploy, "s3_sync") as mutable_sync, \
+                    self.assertRaisesRegex(
+                RuntimeError, "immutable failed"
+            ):
+                deploy.deploy_static("baseline", root, root, "token", cfg)
+
+            self.assertEqual(2, immutable_publish.call_count)
+            mutable_sync.assert_not_called()
+
+
     def test_absent_artifact_is_detected_before_publish(self):
         cfg = normalize_site_config(fixture("split-site.yaml"), "baseline.fritzlab.net")
         with tempfile.TemporaryDirectory() as tmp, redirect_stderr(io.StringIO()), \