fix: let a full provision run compose with a published extract

docs/hosted-deployment.md

@@ -208,6 +208,10 @@ Evidence cell after the operation succeeds. The running cell continues to read
 the old mounted extract until that restart. A malformed, metadata-mismatched,
 or non-newer publication fails closed.
 
+A later full provision run composes with a published extract. It looks past the
+superseded extract and the superseded runtime it left behind, so a rollback keeps
+both, and still verifies everything it stages itself.
+
 ## Deployment order
 
 Deploy the reset alongside the existing deployment in this order:

scripts/provision-hosted-runtime.py

@@ -100,10 +100,11 @@
 # ceiling is lifted rather than removed so a future-dated extract is still
 # refused.
 REUSE_MAX_EXTRACT_AGE_SECONDS = 100 * 365 * 24 * 60 * 60
-ROLLBACK_RUNTIME = re.compile(
-    r"^runtime\.rollback-(?:cra-birth|nia-population|sro-poverty)-"
-    r"[0-9]{8}T[0-9]{6}(?:[0-9]{6})?Z\.yaml$"
-)
+# `publish-extract` names the extract it stages and the runtime it supersedes
+# after the same publication, so both patterns are built from one fragment.
+PUBLICATION = r"(?:cra-birth|nia-population|sro-poverty)-[0-9]{8}T[0-9]{6}(?:[0-9]{6})?Z"
+ROLLBACK_RUNTIME = re.compile(rf"^runtime\.rollback-{PUBLICATION}\.yaml$")
+PUBLISHED_EXTRACT = re.compile(rf"^{PUBLICATION}\.sqlite$")
 
 
 class ProvisionError(RuntimeError):
@@ -560,6 +561,26 @@ def _preserve_extract_rollback(relative: str, value: tuple[str, int]) -> bool:
     )
 
 
+def _preserve_superseded_extracts(
+    active: str,
+) -> Callable[[str, tuple[str, int]], bool]:
+    """Look past the extracts `publish-extract` superseded, never past this run's
+    own. A rollback needs the older file to stay readable, so provisioning has to
+    accept it beside the one it stages; the staged extract is this run's output
+    and stays verified against what it wrote."""
+
+    def preserve(relative: str, value: tuple[str, int]) -> bool:
+        return (
+            "/" not in relative
+            and relative != active
+            and PUBLISHED_EXTRACT.fullmatch(relative) is not None
+            and value[0] != "directory"
+            and value[1] == 0o444
+        )
+
+    return preserve
+
+
 def _copy_tree(source: Path, destination: Path) -> None:
     shutil.copytree(source, destination)
 
@@ -913,7 +934,9 @@ def _stage_evidence(
     observed_at: str,
     mint_origin: str,
     relay_origin: str | None,
-) -> None:
+) -> str | None:
+    """Stage one evidence cell and report the extract it bound, if it holds one."""
+
     source = assets / "evidence" / "cells" / cell
     _copy_tree(source, runtime)
     public = _public_jwk(_read_secret(secrets, "signing-public.jwk"), allow_rsa=False)
@@ -955,6 +978,7 @@ def _stage_evidence(
     for directory in [secret_output, *secret_output.rglob("*")]:
         if directory.is_dir():
             directory.chmod(0o700)
+    return extract_name
 
 
 def _stage_mint(
@@ -1103,7 +1127,7 @@ def _provision_target(args: argparse.Namespace) -> None:
                 if cell in DIRECT
                 else now
             )
-            _stage_evidence(
+            extract_name = _stage_evidence(
                 assets,
                 cell,
                 args.secrets.absolute(),
@@ -1117,15 +1141,32 @@ def _provision_target(args: argparse.Namespace) -> None:
                 relay_origin,
             )
             runtime_preserve = _preserve_extract_rollback if cell in DIRECT else None
+            # `publish-extract` leaves the superseded extract beside the one it
+            # bound, exactly as it leaves the superseded runtime, so a later full
+            # provision has to look past both to recognise its own tree.
+            extract_preserve = (
+                _preserve_superseded_extracts(extract_name)
+                if extract_name is not None
+                else None
+            )
             _check_secret_install_tree(secret_output, args.secret_output.resolve())
             if cell in DIRECT:
-                _check_install_tree(extracts, args.extract_output.resolve())
+                _check_install_tree(
+                    extracts,
+                    args.extract_output.resolve(),
+                    preserve=extract_preserve,
+                )
             _check_install_tree(
                 runtime, args.runtime_output.resolve(), preserve=runtime_preserve
             )
             _install_secret_tree(secret_output, args.secret_output.resolve())
             if cell in DIRECT:
-                _install_tree(extracts, args.extract_output.resolve(), root_mode=0o555)
+                _install_tree(
+                    extracts,
+                    args.extract_output.resolve(),
+                    root_mode=0o555,
+                    preserve=extract_preserve,
+                )
         else:
             raise ProvisionError("invalid target")
         _install_tree(

scripts/test_provision_hosted_runtime.py

@@ -1049,6 +1049,137 @@ def stage(_assets, _cell, destination, _published_at, _observed_at):
                 previous_name,
             )
 
+    def _published_extract_provision(
+        self, root: Path, superseded: str, active: str
+    ) -> object:
+        """Lay out the state `publish-extract` leaves behind and return the
+        parsed provision arguments that a later full run would use."""
+        runtime_output = root / "runtime-output"
+        extract_output = root / "extract-output"
+        runtime_output.mkdir()
+        extract_output.mkdir()
+        provisioner._write(extract_output / superseded, b"superseded", 0o444)
+        provisioner._write(extract_output / active, b"active", 0o444)
+        provisioner._write(runtime_output / "runtime.yaml", b"runtime", 0o444)
+        provisioner._write(
+            runtime_output
+            / f"runtime.rollback-{superseded.removesuffix('.sqlite')}.yaml",
+            b"rollback",
+            0o444,
+        )
+        return provisioner.parser().parse_args(
+            [
+                "provision",
+                "--target",
+                "sro-evidence",
+                "--assets",
+                str(root / "assets"),
+                "--secrets",
+                str(root / "secrets"),
+                "--runtime-output",
+                str(runtime_output),
+                "--secret-output",
+                str(root / "secret-output"),
+                "--extract-output",
+                str(extract_output),
+                "--bind-host",
+                provisioner.EXPECTED_BIND_HOST,
+                "--mint-origin",
+                provisioner.MINT_ORIGIN,
+            ]
+        )
+
+    @contextlib.contextmanager
+    def _staged_provision(self, active: str):
+        def stage(
+            _assets,
+            _cell,
+            _secrets,
+            runtime,
+            _secret_output,
+            extracts,
+            _bind_host,
+            _published_at,
+            _observed_at,
+            _mint_origin,
+            _relay_origin,
+        ):
+            provisioner._write(runtime / "runtime.yaml", b"runtime", 0o444)
+            provisioner._write(extracts / active, b"active", 0o444)
+            return active
+
+        with (
+            mock.patch.object(provisioner, "verify_assets"),
+            mock.patch.object(provisioner, "_validate_secret_inventory"),
+            mock.patch.object(provisioner, "_confine_secret_inventory"),
+            mock.patch.object(provisioner, "_consume_secret_inventory"),
+            mock.patch.object(
+                provisioner, "_publication_time", return_value="2026-08-12T10:00:00Z"
+            ),
+            mock.patch.object(provisioner, "_stage_evidence", side_effect=stage),
+            mock.patch.object(provisioner, "_check_secret_install_tree"),
+            mock.patch.object(provisioner, "_install_secret_tree"),
+        ):
+            yield
+
+    def test_provision_composes_with_a_published_extract(self) -> None:
+        """`publish-extract` appends the extract it binds beside the one it
+        supersedes and leaves a rollback runtime next to the runtime it rewrote.
+        A later full provision has to look past both to recognise its own tree,
+        and must leave the superseded extract where a rollback can still find
+        it."""
+        with tempfile.TemporaryDirectory() as temporary:
+            root = Path(temporary)
+            superseded = "sro-poverty-20260812T090000Z.sqlite"
+            active = "sro-poverty-20260812T100000Z.sqlite"
+            arguments = self._published_extract_provision(root, superseded, active)
+            with self._staged_provision(active):
+                provisioner.provision(arguments)
+
+            extract_output = root / "extract-output"
+            self.assertEqual((extract_output / superseded).read_bytes(), b"superseded")
+            self.assertEqual((extract_output / active).read_bytes(), b"active")
+            self.assertEqual(
+                stat.S_IMODE((extract_output / superseded).stat().st_mode), 0o444
+            )
+            self.assertEqual(stat.S_IMODE(extract_output.stat().st_mode), 0o555)
+
+    def test_provision_still_refuses_an_unpublished_extract_file(self) -> None:
+        """The allowance is only for extracts this provisioner published, so an
+        extract output carrying anything else is still a refusal rather than a
+        directory the check has stopped reading."""
+        with tempfile.TemporaryDirectory() as temporary:
+            root = Path(temporary)
+            superseded = "sro-poverty-20260812T090000Z.sqlite"
+            active = "sro-poverty-20260812T100000Z.sqlite"
+            arguments = self._published_extract_provision(root, superseded, active)
+            provisioner._write(
+                root / "extract-output" / "sro-poverty-restored.sqlite",
+                b"restored",
+                0o444,
+            )
+            with self._staged_provision(active), self.assertRaisesRegex(
+                provisioner.ProvisionError, "existing output mismatch"
+            ):
+                provisioner.provision(arguments)
+
+    def test_only_a_superseded_published_extract_is_preserved(self) -> None:
+        active = "sro-poverty-20260812T100000Z.sqlite"
+        superseded = "sro-poverty-20260812T090000Z.sqlite"
+        preserve = provisioner._preserve_superseded_extracts(active)
+        self.assertTrue(preserve(superseded, ("digest", 0o444)))
+        # The extract this run staged is its own output and stays verified.
+        self.assertFalse(preserve(active, ("digest", 0o444)))
+        # Only the three direct authorities publish extracts, only at the top of
+        # the output, only as immutable files.
+        self.assertFalse(
+            preserve("sipf-pension-20260812T090000Z.sqlite", ("digest", 0o444))
+        )
+        self.assertFalse(preserve(f"nested/{superseded}", ("digest", 0o444)))
+        self.assertFalse(preserve(superseded, ("directory", 0o444)))
+        self.assertFalse(preserve(superseded, ("digest", 0o644)))
+        self.assertFalse(preserve(f"{superseded}.bak", ("digest", 0o444)))
+
     def test_extract_append_never_overwrites_a_mismatched_filename(self) -> None:
         with tempfile.TemporaryDirectory() as temporary:
             root = Path(temporary)