fix(bridge-cutover): accept the pinned post-cutover state so worker-state steps unblock

docs/bridge-cutover-sealed-runtimes.md

@@ -168,6 +168,12 @@ Both transactions revalidate the quiet point, release bindings, regular-file ide
 
 That final compare-and-swap is performed after the staged object itself has been rehashed. A live symlink, regular file, registry, or adoption target that changes after the quiet-point scan is never overwritten.
 
+The worker-state gate loads the candidate API through bundle validation, which transitively re-runs the sealed-adoption assessment; provision verification, finalize, and rollback identity attribution all take that path, while the snapshot step revalidates only the cutover quiet point.
+The sealed-adoption plan itself stays strict: it accepts live state only at the exact initial or sealed identities.
+Once the normal cutover is fully applied and the post-install irreversible boundary is marked, bundle validation instead accepts exactly one further live state - a fully sealed adoption journal, the full quiet-point contract, intact retained sealed release trees, and every adoption-managed path, including the live registry, at its exact cutover-new identity pinned from the reconstructed cutover manifest - reported as the `runtime-switched` phase.
+Any other combination, including a post-cutover registry without the marked boundary or with any drifted path, refuses exactly as before.
+This pinned post-cutover acceptance exists because the documented order crosses the runtime switch before worker-state verification, so verification, finalize, and rollback attribution must remain provable on the switched machine.
+
 ## Exact worker topology
 
 The registry contains eight explicit profiles, but only six profiles are Fleet workers.

tests/fm-account-routing.test.sh

@@ -5078,7 +5078,8 @@ test_agent_fleet_lifecycle_calls_are_bounded() {
   if out=$(FM_FAKE_AF_SELECT_SLEEP=10 FM_ACCOUNT_CONTROL_TIMEOUT=1 run_spawn "$id" "$PROJ_DIR" --account-pool claude-crew); then status=0; else status=$?; fi
   elapsed=$(( $(date +%s) - started ))
   [ "$status" -eq 0 ] || fail "timed-out lease choice was not reconciled through recovery: $out"
-  [ "$elapsed" -lt 5 ] || fail "lease choice timeout was not bounded (elapsed ${elapsed}s)"
+  # Semantic: returns well under the 10s unbounded fake sleep; 8s absorbs full-sweep scheduling load (5s flaked in-sweep while passing standalone).
+  [ "$elapsed" -lt 8 ] || fail "lease choice timeout was not bounded (elapsed ${elapsed}s)"
   assert_grep 'lease recover ' "$AF_LOG" "timed-out lease choice did not reconcile ownership"
 
   rm -f "$CASE_DIR/endpoint-live"
@@ -5087,7 +5088,8 @@ test_agent_fleet_lifecycle_calls_are_bounded() {
   if out=$(FM_FAKE_AF_RELEASE_SLEEP=10 FM_ACCOUNT_CONTROL_TIMEOUT=1 run_teardown "$id" --force 2>&1); then status=0; else status=$?; fi
   elapsed=$(( $(date +%s) - started ))
   [ "$status" -ne 0 ] || fail "ambiguous timed-out lease release unexpectedly completed teardown"
-  [ "$elapsed" -lt 5 ] || fail "lease release timeout was not bounded (elapsed ${elapsed}s)"
+  # Semantic: returns well under the 10s unbounded fake sleep; 8s absorbs full-sweep scheduling load (5s flaked in-sweep while passing standalone).
+  [ "$elapsed" -lt 8 ] || fail "lease release timeout was not bounded (elapsed ${elapsed}s)"
   assert_present "$HOME_DIR/state/$id.meta" "ambiguous lease release discarded retry metadata"
   pass "Agent Fleet lease mutations are bounded and ambiguous outcomes retain ownership state"
 }

tests/fm-bootstrap.test.sh

@@ -762,6 +762,7 @@ jq() {
 }
 SH
   out=$(PATH="$fakebin:$BASE_PATH" BASH_ENV="$bash_env" FM_HOME="$case_dir/home" FM_ROOT_OVERRIDE="$case_dir/home" \
+    FM_ACCOUNT_ROUTING_TEST_LAB=firstmate-account-routing-test-lab-v1 \
     FM_FAKE_TREEHOUSE_LEASE_HELP=1 "$ROOT/bin/fm-bootstrap.sh")
   assert_contains "$out" 'MISSING_MANUAL: agent-fleet (instructions: https://github.com/ruby-dlee/firstmate/blob/main/docs/configuration.md#agent-fleet-account-routing)' "enforce mode did not report manual Agent Fleet installation"
   assert_contains "$out" 'MISSING: jq (install: brew install jq  # or the platform' "enforce mode did not report missing jq"
@@ -774,6 +775,7 @@ SH
   add_real_jq "$fakebin"
   rm -f "$fakebin/agent-fleet"
   out=$(PATH="$fakebin:$BASE_PATH" FM_HOME="$case_dir/home" FM_ROOT_OVERRIDE="$case_dir/home" \
+    FM_ACCOUNT_ROUTING_TEST_LAB=firstmate-account-routing-test-lab-v1 \
     FM_FAKE_TREEHOUSE_LEASE_HELP=1 "$ROOT/bin/fm-bootstrap.sh")
   assert_contains "$out" 'MISSING_MANUAL: agent-fleet (instructions: https://github.com/ruby-dlee/firstmate/blob/main/docs/configuration.md#agent-fleet-account-routing)' "account-routed dispatch profile did not report manual Agent Fleet installation"
   assert_contains "$out" 'CREW_DISPATCH: active config/crew-dispatch.json' "account dependency preflight suppressed dispatch validation"

tests/test_bridge_sealed_adoption.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
+import contextlib
 import hashlib
+import io
 import json
 import os
 import sys
@@ -550,6 +552,40 @@ def quiet_then_change(value: adoption.Manifest) -> None:
                 adoption.apply(manifest)
         self.assertEqual(fixture.registry.read_bytes(), racer)
 
+    def test_post_cutover_hint_is_cli_only_and_never_in_the_refusal(self) -> None:
+        fixture = Fixture()
+        self.addCleanup(fixture.close)
+        fixture.registry.write_bytes(b'version = 1\nmode = "post-cutover"\n')
+        fixture.registry.chmod(0o600)
+
+        with self.assertRaises(adoption.AdoptionError) as caught:
+            adoption.plan(fixture.load())
+        refusal = str(caught.exception)
+        self.assertIn(adoption.UNKNOWN_LIVE_REGISTRY_REFUSAL, refusal)
+        self.assertNotIn(adoption.POST_CUTOVER_CLI_HINT, refusal)
+        self.assertNotIn("prepare_bridge_cutover.py", refusal)
+
+        stderr = io.StringIO()
+        with contextlib.redirect_stderr(stderr):
+            code = adoption.main([str(fixture.manifest_path)])
+        printed = stderr.getvalue()
+        self.assertEqual(code, 2)
+        self.assertIn(f"refused: {refusal}", printed)
+        self.assertIn(adoption.POST_CUTOVER_CLI_HINT, printed)
+
+    def test_post_cutover_hint_is_not_printed_for_unrelated_refusals(self) -> None:
+        fixture = Fixture()
+        self.addCleanup(fixture.close)
+        fixture.backend.write_text("zellij\n", encoding="utf-8")
+        fixture.backend.chmod(0o600)
+
+        stderr = io.StringIO()
+        with contextlib.redirect_stderr(stderr):
+            code = adoption.main([str(fixture.manifest_path)])
+        printed = stderr.getvalue()
+        self.assertEqual(code, 2)
+        self.assertNotIn(adoption.POST_CUTOVER_CLI_HINT, printed)
+
     def test_link_change_at_exchange_syscall_is_swapped_back_and_preserved(self) -> None:
         fixture = Fixture()
         self.addCleanup(fixture.close)

tests/test_bridge_worker_state_transaction.py

@@ -20,6 +20,7 @@
 
 import bridge_worker_state_transaction as worker_state  # noqa: E402
 from tests.test_prepare_bridge_cutover import (  # noqa: E402
+    DRIVER,
     CutoverPreparationFixture,
     prepare,
 )
@@ -151,6 +152,63 @@ def test_snapshot_verify_and_finalize_exact_six_workers(self) -> None:
         self.assertFalse(cleaned["snapshot_bytes_present"])
         self.assertFalse(self.manifest.snapshot_path.exists())
 
+    def _apply_runtime_switch(self) -> None:
+        driver = prepare._load_driver(DRIVER)
+        manifest_path = self.fixture.bundle_dir / "cutover.manifest.json"
+        driver.execute(driver.load_manifest(manifest_path), "forward")
+        driver.mark_post_install_irreversible_boundary(
+            driver.load_manifest(manifest_path)
+        )
+
+    def test_verify_and_finalize_pass_after_runtime_switch(self) -> None:
+        """Regression: the 2026-07-19 live 6b NO-GO.
+
+        The documented cutover order applies the runtime switch (step 4) and
+        marks the post-install boundary (step 5) before worker-state
+        verification (6b) and finalize (6c).  Every pre-existing test ran
+        worker-state at the pre-switch baseline, so the sealed-adoption quiet
+        point refusing the legitimately advanced live registry was first
+        observed on the real machine.  This test runs the exact live order.
+        """
+
+        self._apply_runtime_switch()
+        worker_state.begin(self.manifest)
+        self._materialize_provisioned_workers()
+        verified = worker_state.verify_provisioned(self.manifest)
+        self.assertEqual(verified["phase"], "provision_verified")
+        self.assertTrue(verified["rollback_available"])
+        self._write_identity_bundles()
+        finalized = worker_state.finalize(self.manifest)
+        self.assertEqual(finalized["phase"], "complete")
+        self.assertTrue(finalized["worker_state_ready"])
+
+    def test_rollback_attributes_identity_drift_after_runtime_switch(self) -> None:
+        """Regression: the armed rollback trap behind the 2026-07-19 NO-GO.
+
+        Once a provider identity bundle exists and drifts from the snapshot,
+        ``rollback`` must attribute the drift through ``_candidate_api`` - the
+        exact loader the sealed-adoption quiet point was refusing post-switch.
+        Rollback must keep working in that state, not die at the moment
+        identity state is what needs restoring.
+        """
+
+        self._apply_runtime_switch()
+        self._write_identity_bundles()
+        worker_state.begin(self.manifest)
+        self._materialize_provisioned_workers()
+        self.manifest.identity_bundles["codex"].write_text(
+            '{"schema":1,"provider":"codex","fresh":true}', encoding="utf-8"
+        )
+        rolled = worker_state.rollback(self.manifest)
+        self.assertEqual(rolled["phase"], "rolled_back")
+        # Attributed drift admits the rollback, which restores the snapshot.
+        self.assertEqual(
+            json.loads(
+                self.manifest.identity_bundles["codex"].read_text(encoding="utf-8")
+            ),
+            {"schema": 1, "provider": "codex"},
+        )
+
     def test_restartable_rollback_restores_exact_absence(self) -> None:
         worker_state.begin(self.manifest)
         self._materialize_provisioned_workers()

tests/test_prepare_bridge_cutover.py

@@ -2169,6 +2169,99 @@ def test_bundle_becomes_runtime_switch_ready_only_after_sealed_adoption(self) ->
         self.assertTrue(result["runtime_switch_ready"])
         self.assertFalse(result["cutover_ready"])
 
+    def _apply_runtime_switch(
+        self, mark_boundary: bool = True
+    ) -> tuple[types.ModuleType, Path]:
+        bundle = json.loads(
+            (self.fixture.bundle_dir / "bundle.json").read_text(encoding="utf-8")
+        )
+        adoption_driver = prepare._load_adoption_driver()
+        adoption_driver.apply(
+            adoption_driver.load_manifest(Path(bundle["adoption_manifest_path"]))
+        )
+        driver = prepare._load_driver(DRIVER)
+        manifest_path = Path(bundle["manifest_path"])
+        driver.execute(driver.load_manifest(manifest_path), "forward")
+        if mark_boundary:
+            driver.mark_post_install_irreversible_boundary(
+                driver.load_manifest(manifest_path)
+            )
+        return driver, manifest_path
+
+    def test_validate_accepts_post_cutover_state_only_with_boundary_marked(self) -> None:
+        self.fixture.prepare()
+        driver, manifest_path = self._apply_runtime_switch(mark_boundary=False)
+        with self.assertRaisesRegex(
+            prepare.PreparationError,
+            "sealed-adoption state is invalid: .*"
+            r"\(post-cutover probe: main cutover is not fully applied past the "
+            "marked post-install irreversible boundary\\)",
+        ):
+            prepare.validate_bundle(self.fixture.bundle_dir / "bundle.json", DRIVER)
+
+        driver.mark_post_install_irreversible_boundary(
+            driver.load_manifest(manifest_path)
+        )
+        result = prepare.validate_bundle(self.fixture.bundle_dir / "bundle.json", DRIVER)
+
+        self.assertEqual(result["cutover_phase"], "runtime-switched")
+        self.assertFalse(result["runtime_switch_ready"])
+        self.assertFalse(result["cutover_ready"])
+        # The accepted live-registry set stays pinned to exactly the three
+        # bundle-recorded identities; post-cutover the live file is the new one.
+        self.assertEqual(
+            prepare._sha256(self.fixture.live), result["new_registry_sha256"]
+        )
+
+    def test_validate_refuses_tampered_live_registry_after_runtime_switch(self) -> None:
+        self.fixture.prepare()
+        self._apply_runtime_switch()
+        payload = self.fixture.live.read_bytes()
+        self.fixture.live.write_bytes(payload + b"# drift\n")
+        self.fixture.live.chmod(0o600)
+        with self.assertRaisesRegex(prepare.PreparationError, "unknown SHA-256"):
+            prepare.validate_bundle(self.fixture.bundle_dir / "bundle.json", DRIVER)
+
+    def test_validate_refuses_candidate_registry_before_adoption(self) -> None:
+        self.fixture.prepare()
+        self.fixture.live.write_bytes(
+            (self.fixture.bundle_dir / "registry.new.toml").read_bytes()
+        )
+        self.fixture.live.chmod(0o600)
+        with self.assertRaisesRegex(prepare.PreparationError, "unknown SHA-256"):
+            prepare.validate_bundle(self.fixture.bundle_dir / "bundle.json", DRIVER)
+
+    def test_validate_refuses_partially_reverted_state_after_runtime_switch(self) -> None:
+        self.fixture.prepare()
+        self._apply_runtime_switch()
+        current = self.fixture.agent_root / "current"
+        os.unlink(current)
+        os.symlink("releases/0.1.5-old", current)
+        with self.assertRaisesRegex(
+            prepare.PreparationError,
+            "sealed-adoption state is invalid: .*"
+            r"\(post-cutover probe: observed old/new states are not a valid "
+            "transaction prefix",
+        ):
+            prepare.validate_bundle(self.fixture.bundle_dir / "bundle.json", DRIVER)
+
+    def test_post_cutover_plan_refuses_registry_pin_without_sha256(self) -> None:
+        self.fixture.prepare()
+        driver, manifest_path = self._apply_runtime_switch()
+        bundle = json.loads(
+            (self.fixture.bundle_dir / "bundle.json").read_text(encoding="utf-8")
+        )
+        adoption_driver = prepare._load_adoption_driver()
+        loaded_adoption = adoption_driver.load_manifest(
+            Path(bundle["adoption_manifest_path"])
+        )
+        pins = prepare._post_cutover_pins(driver, driver.load_manifest(manifest_path))
+        pins[str(loaded_adoption.registry_operation.path)] = {"kind": "file"}
+        with self.assertRaisesRegex(
+            adoption_driver.AdoptionError, "requires a registry file pin"
+        ):
+            adoption_driver.post_cutover_plan(loaded_adoption, pins)
+
     def test_validate_refuses_extra_trusted_project(self) -> None:
         self.fixture.prepare()
         extra = self.fixture.root / "other-project"