diff --git a/docs/bridge-cutover-sealed-runtimes.md b/docs/bridge-cutover-sealed-runtimes.md index 25a00cf2ed9..f6c04d0ea83 100644 --- a/docs/bridge-cutover-sealed-runtimes.md +++ b/docs/bridge-cutover-sealed-runtimes.md @@ -174,6 +174,17 @@ Once the normal cutover is fully applied and the post-install irreversible bound 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. +## In-place identity refresh + +The prepared activation plan and worker-state manifest pin the live provider-binary identity - inode, mtime, and content SHA-256 - so a byte-exact reinstall of that provider binary gives it a fresh inode and mtime and re-stales both artifacts against the live binary, and the strict activation-plan and worker-state gates then refuse even though the machine is genuinely in the `runtime-switched` state. +A from-scratch rebuild cannot recover, because the `runtime-switched` proof is journal-bound: the post-install irreversible boundary and the sealed-adoption journal live only inside the applying bundle's own `transaction/` directory, cannot be borrowed byte-for-byte by another bundle, and cannot be re-sealed against the already-migrated live registry, so only the bundle that actually applied the switch validates as `runtime-switched`. +The `refresh` subcommand of `prepare_bridge_cutover.py` refreshes exactly those two identity artifacts of that existing bundle in place to the current provider-binary identity, while leaving the sealed cutover and sealed-adoption journals untouched; its `--help` output owns the exact invocation syntax. +It holds the worker-state advisory lock across the journal check and both atomic artifact replacements, so worker-state cannot begin against the old fingerprint during refresh. +It runs only before worker-state verification begins - before the 6a snapshot, while the worker-state journal, snapshot, and snapshot-staging paths are all absent and no worker-state transaction is bound to the manifest. +Once any of those transaction artifacts exists, refresh refuses; it never restarts, resets, concludes, cleans up, or otherwise mutates a worker-state transaction bound to the manifest. +If refresh is interrupted after replacing the bundle but before replacing the worker-state manifest, rerunning refresh repairs the stale manifest and restores an exact pair. +It also refuses unless the bundle is provably the applied `runtime-switched` one - it reuses the same journal-bound phase determination as validation - and it gates completion on a full strict `validate_bundle` that must still report the `runtime-switched` phase, so it makes the strict gates pass only because the recorded identity again equals the live binary, never by loosening any check. + ## Exact worker topology The registry contains eight explicit profiles, but only six profiles are Fleet workers. diff --git a/tests/test_bridge_worker_state_transaction.py b/tests/test_bridge_worker_state_transaction.py index 243f2ea333b..baee1f70092 100644 --- a/tests/test_bridge_worker_state_transaction.py +++ b/tests/test_bridge_worker_state_transaction.py @@ -152,6 +152,29 @@ 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 test_begin_reloads_manifest_after_lock_acquisition(self) -> None: + stale_manifest = self.manifest + bundle_path = self.fixture.bundle_dir / "bundle.json" + bundle_path.write_bytes(bundle_path.read_bytes() + b"\n") + manifest = json.loads(self.manifest_path.read_text(encoding="utf-8")) + manifest["bundle_sha256"] = hashlib.sha256( + bundle_path.read_bytes() + ).hexdigest() + self.manifest_path.write_text( + json.dumps(manifest, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + self.manifest_path.chmod(0o600) + current_manifest = worker_state.load_manifest(self.manifest_path) + self.assertNotEqual(stale_manifest.fingerprint, current_manifest.fingerprint) + worker_state.begin(stale_manifest) + journal = json.loads( + current_manifest.journal_path.read_text(encoding="utf-8") + ) + self.assertEqual( + journal["manifest_fingerprint"], current_manifest.fingerprint + ) + def _apply_runtime_switch(self) -> None: driver = prepare._load_driver(DRIVER) manifest_path = self.fixture.bundle_dir / "cutover.manifest.json" diff --git a/tests/test_prepare_bridge_cutover.py b/tests/test_prepare_bridge_cutover.py index a840322ce63..3b001329cda 100644 --- a/tests/test_prepare_bridge_cutover.py +++ b/tests/test_prepare_bridge_cutover.py @@ -2262,6 +2262,204 @@ def test_post_cutover_plan_refuses_registry_pin_without_sha256(self) -> None: ): adoption_driver.post_cutover_plan(loaded_adoption, pins) + def _read_json_file(self, path: Path) -> dict: + return json.loads(Path(path).read_text(encoding="utf-8")) + + def _write_json_file(self, path: Path, value: dict) -> None: + Path(path).write_text( + json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8" + ) + Path(path).chmod(0o600) + + def _transaction_journal_bytes(self) -> dict[str, bytes]: + transaction = self.fixture.bundle_dir / "transaction" + return { + name: (transaction / name).read_bytes() + for name in ("cutover.journal.json", "sealed-adoption.journal.json") + } + + def _restale_recorded_identity(self) -> None: + """Re-stale the recorded identity artifacts the way a byte-exact provider + reinstall does live: the sealed cutover and adoption journals stay + intact, but the recorded activation plan and worker-state manifest drift + off the current provider-binary identity. The synthetic provision model + is registry-pure (it does not pin the live binary inode/mtime the real + provision API does), so the faithful analog of that drift is to move the + recorded per-worker plan identity off what a fresh contract reconstructs. + """ + profile = prepare.WORKER_PROFILES[0] + stale = "a" * 64 + bundle_path = self.fixture.bundle_dir / "bundle.json" + bundle = self._read_json_file(bundle_path) + plans = bundle["activation_plan"]["provision"]["sealed_contract"]["plans"] + plans[profile]["plan_sha256"] = stale + self._write_json_file(bundle_path, bundle) + worker_state_path = self.fixture.bundle_dir / "worker-state.manifest.json" + worker_state = self._read_json_file(worker_state_path) + worker_state["sealed_plans"][profile]["plan_sha256"] = stale + for worker in worker_state["workers"]: + if worker["profile"] == profile: + worker["plan_sha256"] = stale + self._write_json_file(worker_state_path, worker_state) + + def test_refresh_updates_identity_and_revalidates_runtime_switched(self) -> None: + self.fixture.prepare() + self._apply_runtime_switch() + bundle_path = self.fixture.bundle_dir / "bundle.json" + worker_state_path = self.fixture.bundle_dir / "worker-state.manifest.json" + + # The applied runtime-switched bundle validates strictly before drift. + self.assertEqual( + prepare.validate_bundle(bundle_path, DRIVER)["cutover_phase"], + "runtime-switched", + ) + original_bundle_bytes = bundle_path.read_bytes() + original_worker_state_bytes = worker_state_path.read_bytes() + journal_before = self._transaction_journal_bytes() + + # A byte-exact provider reinstall re-stales the two identity artifacts. + self._restale_recorded_identity() + with self.assertRaisesRegex( + prepare.PreparationError, "activation plan is not exact" + ): + prepare.validate_bundle(bundle_path, DRIVER) + + # The in-place refresh realigns both artifacts to the live identity and + # gates completion on the journal-bound runtime-switched proof. + refreshed = prepare.refresh_bundle(bundle_path, DRIVER) + self.assertTrue(refreshed["refreshed"]) + self.assertTrue(refreshed["valid"]) + self.assertEqual(refreshed["cutover_phase"], "runtime-switched") + + # A fresh strict validation now passes with no gate loosening. + revalidated = prepare.validate_bundle(bundle_path, DRIVER) + self.assertEqual(revalidated["cutover_phase"], "runtime-switched") + self.assertFalse(revalidated["runtime_switch_ready"]) + self.assertFalse(revalidated["cutover_ready"]) + + # Both artifacts now equal the exact live reconstruction; the sealed + # cutover and adoption journals were preserved byte-for-byte. + self.assertEqual(bundle_path.read_bytes(), original_bundle_bytes) + self.assertEqual( + worker_state_path.read_bytes(), original_worker_state_bytes + ) + self.assertEqual(self._transaction_journal_bytes(), journal_before) + + def test_refresh_does_not_loosen_the_strict_identity_gate(self) -> None: + self.fixture.prepare() + self._apply_runtime_switch() + bundle_path = self.fixture.bundle_dir / "bundle.json" + self._restale_recorded_identity() + prepare.refresh_bundle(bundle_path, DRIVER) + + # Re-staling after a refresh still refuses exactly as before: the strict + # gate is unchanged; the refresh only realigned the recorded identity to + # the live binary rather than weakening the check. + self._restale_recorded_identity() + with self.assertRaisesRegex( + prepare.PreparationError, "activation plan is not exact" + ): + prepare.validate_bundle(bundle_path, DRIVER) + + def test_refresh_refuses_bundle_that_is_not_runtime_switched(self) -> None: + # Prepared but never adopted or switched: sealed-adoption-pending, not + # the applied runtime-switched bundle, so refresh must refuse and leave + # both identity artifacts untouched. + self.fixture.prepare() + bundle_path = self.fixture.bundle_dir / "bundle.json" + worker_state_path = self.fixture.bundle_dir / "worker-state.manifest.json" + before = (bundle_path.read_bytes(), worker_state_path.read_bytes()) + self.assertEqual( + prepare.validate_bundle(bundle_path, DRIVER)["cutover_phase"], + "sealed-adoption-pending", + ) + with self.assertRaisesRegex( + prepare.PreparationError, "only to an applied runtime-switched bundle" + ): + prepare.refresh_bundle(bundle_path, DRIVER) + self.assertEqual( + (bundle_path.read_bytes(), worker_state_path.read_bytes()), before + ) + + def test_refresh_refuses_runtime_switch_ready_bundle(self) -> None: + # Sealed adoption applied but the main cutover still at the sealed + # baseline: runtime-switch-ready, not yet applied. Refresh must refuse. + self.fixture.prepare() + 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"])) + ) + bundle_path = self.fixture.bundle_dir / "bundle.json" + self.assertEqual( + prepare.validate_bundle(bundle_path, DRIVER)["cutover_phase"], + "runtime-switch-ready", + ) + with self.assertRaisesRegex( + prepare.PreparationError, "only to an applied runtime-switched bundle" + ): + prepare.refresh_bundle(bundle_path, DRIVER) + + def test_refresh_refuses_when_boundary_not_marked(self) -> None: + # Cutover applied but the post-install irreversible boundary not marked: + # the journal-bound runtime-switched proof cannot be established, so the + # shared reconstruction refuses and refresh never mutates the bundle. + self.fixture.prepare() + self._apply_runtime_switch(mark_boundary=False) + bundle_path = self.fixture.bundle_dir / "bundle.json" + worker_state_path = self.fixture.bundle_dir / "worker-state.manifest.json" + before = (bundle_path.read_bytes(), worker_state_path.read_bytes()) + with self.assertRaisesRegex( + prepare.PreparationError, + "post-cutover probe: main cutover is not fully applied", + ): + prepare.refresh_bundle(bundle_path, DRIVER) + self.assertEqual( + (bundle_path.read_bytes(), worker_state_path.read_bytes()), before + ) + + def test_refresh_refuses_active_worker_state_transaction(self) -> None: + self.fixture.prepare() + self._apply_runtime_switch() + bundle_path = self.fixture.bundle_dir / "bundle.json" + worker_state_path = self.fixture.bundle_dir / "worker-state.manifest.json" + before = (bundle_path.read_bytes(), worker_state_path.read_bytes()) + worker_state_journal = ( + self.fixture.snapshot_parent + / "bridge-cutover-fixture-workers.journal.json" + ) + worker_state_journal.write_text("{}\n", encoding="utf-8") + worker_state_journal.chmod(0o600) + with self.assertRaisesRegex( + prepare.PreparationError, + "before the worker-state 6a snapshot.*strand any bound transaction", + ): + prepare.refresh_bundle(bundle_path, DRIVER) + self.assertEqual( + (bundle_path.read_bytes(), worker_state_path.read_bytes()), before + ) + + def test_refresh_repairs_partial_artifact_replacement(self) -> None: + self.fixture.prepare() + self._apply_runtime_switch() + bundle_path = self.fixture.bundle_dir / "bundle.json" + worker_state_path = self.fixture.bundle_dir / "worker-state.manifest.json" + worker_state = self._read_json_file(worker_state_path) + worker_state["bundle_sha256"] = "a" * 64 + self._write_json_file(worker_state_path, worker_state) + with self.assertRaises(prepare.PreparationError): + prepare.validate_bundle(bundle_path, DRIVER) + refreshed = prepare.refresh_bundle(bundle_path, DRIVER) + self.assertTrue(refreshed["valid"]) + self.assertEqual(refreshed["cutover_phase"], "runtime-switched") + repaired_worker_state = self._read_json_file(worker_state_path) + self.assertEqual( + repaired_worker_state["bundle_sha256"], + hashlib.sha256(bundle_path.read_bytes()).hexdigest(), + ) + def test_validate_refuses_extra_trusted_project(self) -> None: self.fixture.prepare() extra = self.fixture.root / "other-project" diff --git a/tools/bridge-cutover/bridge_worker_state_transaction.py b/tools/bridge-cutover/bridge_worker_state_transaction.py index dbf5f4c38d0..abc652b5436 100644 --- a/tools/bridge-cutover/bridge_worker_state_transaction.py +++ b/tools/bridge-cutover/bridge_worker_state_transaction.py @@ -946,6 +946,12 @@ def begin( boundaries = boundaries or BoundaryController() lock_fd = _lock(manifest) try: + current_manifest = load_manifest(manifest.path) + if current_manifest.lock_path != manifest.lock_path: + raise WorkerStateError( + "worker-state lock binding changed while acquiring the lock" + ) + manifest = current_manifest existing = _journal(manifest) if existing is not None: return plan(manifest) diff --git a/tools/bridge-cutover/prepare_bridge_cutover.py b/tools/bridge-cutover/prepare_bridge_cutover.py index f5602aee656..b905a265a31 100644 --- a/tools/bridge-cutover/prepare_bridge_cutover.py +++ b/tools/bridge-cutover/prepare_bridge_cutover.py @@ -1,16 +1,18 @@ #!/usr/bin/env python3 -"""Prepare and rehearse an exact, disabled Bridge cutover bundle. +"""Prepare, validate, refresh, and rehearse a Bridge cutover bundle. This script is deliberately installation-agnostic and standard-library only. -It never discovers a live installation and it has no defaults for user homes, -registries, release links, projects, or provider binaries. Every input is an -absolute path in a strict JSON preparation specification. +It has no defaults for user homes, registries, release links, projects, or +provider binaries. Every live path is bound by an absolute path in the strict +preparation specification or its validated bundle. ``prepare`` writes only to a new explicit output directory. ``validate`` is -read-only. ``rehearse`` copies the bound release trees and registry sources -into a disposable private directory, then exercises the existing restartable -transaction driver there. None of the commands enrolls an account, launches -a provider, changes routing, or operates a live release link. +read-only. ``refresh`` atomically replaces only an applied bundle's recorded +activation plan and worker-state manifest. ``rehearse`` copies the bound +release trees and registry sources into a disposable private directory, then +exercises the existing restartable transaction driver there. None of the +commands enrolls an account, launches a provider, changes routing, or operates +a live release link. """ from __future__ import annotations @@ -2727,6 +2729,46 @@ def _write_json(path: Path, value: Any, mode: int = 0o600) -> None: _write_bytes(path, payload, mode) +def _atomic_replace_bytes(path: Path, value: bytes, mode: int = 0o600) -> None: + """Durably replace an existing regular file with new bytes. + + ``_write_bytes`` opens with ``O_EXCL`` and therefore only ever creates a new + file. Refreshing an already-published bundle artifact must overwrite in + place, so this writes the payload to a private sibling temporary file and + atomically renames it over the target, then fsyncs the directory. The + rename is atomic, so a reader either sees the whole old file or the whole new + file, never a partial write. + """ + + directory = path.parent + descriptor, temp_name = tempfile.mkstemp(prefix=f".{path.name}.", dir=directory) + temp_path = Path(temp_name) + try: + os.fchmod(descriptor, mode) + offset = 0 + while offset < len(value): + offset += os.write(descriptor, value[offset:]) + os.fsync(descriptor) + finally: + os.close(descriptor) + try: + os.replace(temp_path, path) + except BaseException: + if os.path.lexists(temp_path): + os.unlink(temp_path) + raise + parent_fd = os.open(directory, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)) + try: + os.fsync(parent_fd) + finally: + os.close(parent_fd) + + +def _atomic_replace_json(path: Path, value: Any, mode: int = 0o600) -> None: + payload = (json.dumps(value, indent=2, sort_keys=True) + "\n").encode("utf-8") + _atomic_replace_bytes(path, payload, mode) + + def _native_dependencies(path: Path) -> list[str]: try: completed = subprocess.run( @@ -4671,6 +4713,21 @@ def _activation_plan(provision_contract: Mapping[str, Any]) -> dict[str, Any]: } +def _worker_state_transaction_paths( + spec: PreparationSpec, +) -> tuple[str, Path, Path, Path]: + transaction_id = f"{spec.transaction_id}-workers" + if len(transaction_id) > 64: + raise PreparationError("transaction_id is too long for worker-state suffix") + parent = spec.worker_state.snapshot_parent + return ( + transaction_id, + parent / f"{transaction_id}.lock", + parent / f"{transaction_id}.journal.json", + parent / f"{transaction_id}.snapshot", + ) + + def _worker_state_manifest_dict( spec: PreparationSpec, *, @@ -4696,17 +4753,17 @@ def _worker_state_manifest_dict( raise PreparationError( "worker-state snapshot parent must not overlap bundle, workers, or identity state" ) - transaction_id = f"{spec.transaction_id}-workers" - if len(transaction_id) > 64: - raise PreparationError("transaction_id is too long for worker-state suffix") + transaction_id, lock_path, journal_path, snapshot_path = ( + _worker_state_transaction_paths(spec) + ) return { "schema_version": 1, "transaction_id": transaction_id, "apply_opt_in": True, "snapshot_parent": str(parent), - "lock_path": str(parent / f"{transaction_id}.lock"), - "journal_path": str(parent / f"{transaction_id}.journal.json"), - "snapshot_path": str(parent / f"{transaction_id}.snapshot"), + "lock_path": str(lock_path), + "journal_path": str(journal_path), + "snapshot_path": str(snapshot_path), "cutover_manifest_path": str(cutover_manifest_path), "bundle_path": str(bundle_path), "bundle_sha256": bundle_sha256, @@ -5274,7 +5331,45 @@ def _spec_from_bundle(bundle_path: Path, bundle: dict[str, Any]) -> PreparationS return spec -def validate_bundle(bundle_path: Path, driver_path: Path) -> dict[str, Any]: +@dataclass(frozen=True) +class _BundleState: + """Reconstructed, journal-bound runtime state of a validated bundle. + + Carries exactly the pieces both the strict validator and the in-place + refresh need after every shared integrity check has passed but before the + two identity-bound gates (recorded activation plan, worker-state manifest) + are enforced. ``expected_provision_contract`` is recomputed against the + CURRENT provider-binary identity, so it is what the refresh writes and what + the validator compares against. + """ + + bundle_path: Path + bundle: dict[str, Any] + spec: PreparationSpec + candidate: Any + loaded: Any + loaded_adoption: Any + cutover_phase: str + manifest_path: Path + initial_path: Path + old_path: Path + new_path: Path + expected_provision_contract: dict[str, Any] + + +def _reconstruct_bundle_state(bundle_path: Path, driver_path: Path) -> _BundleState: + """Validate a bundle and reconstruct its journal-bound runtime state. + + Owns every bundle-integrity check shared by ``validate_bundle`` and the + in-place refresh: metadata exactness, the sealed runtime and quota proofs, + the three registries, the cutover and sealed-adoption manifests, the + journal-bound cutover phase, the topology summary, and the freshly recomputed + sealed provision contract (which reads the current provider-binary identity). + It stops just before the two identity-bound exactness gates so the validator + can enforce them strictly and the refresh can regenerate them. It never + mutates state. + """ + bundle_path = _normalized_absolute(str(bundle_path), "bundle path") _canonical(bundle_path, "bundle path") bundle = _read_json(bundle_path, "cutover bundle") @@ -5479,6 +5574,44 @@ def validate_bundle(bundle_path: Path, driver_path: Path) -> dict[str, Any]: if bundle["topology"] != expected_topology: raise PreparationError("bundle topology summary is not exact") expected_provision_contract = _sealed_provision_contract(candidate_api, candidate) + return _BundleState( + bundle_path=bundle_path, + bundle=bundle, + spec=spec, + candidate=candidate, + loaded=loaded, + loaded_adoption=loaded_adoption, + cutover_phase=cutover_phase, + manifest_path=manifest_path, + initial_path=initial_path, + old_path=old_path, + new_path=new_path, + expected_provision_contract=expected_provision_contract, + ) + + +def validate_bundle(bundle_path: Path, driver_path: Path) -> dict[str, Any]: + """Validate a bundle strictly, enforcing the two identity-bound gates. + + The recorded activation plan and worker-state manifest must equal the exact + reconstruction from the current provider-binary identity. These are the + journal-independent, identity-bound proofs the in-place refresh regenerates; + this function never loosens them. + """ + + state = _reconstruct_bundle_state(bundle_path, driver_path) + bundle = state.bundle + bundle_path = state.bundle_path + spec = state.spec + candidate = state.candidate + cutover_phase = state.cutover_phase + loaded = state.loaded + loaded_adoption = state.loaded_adoption + manifest_path = state.manifest_path + initial_path = state.initial_path + old_path = state.old_path + new_path = state.new_path + expected_provision_contract = state.expected_provision_contract if bundle["activation_plan"] != _activation_plan(expected_provision_contract): raise PreparationError( "bundle activation plan is not exact or generated-command-free" @@ -5548,6 +5681,97 @@ def validate_bundle(bundle_path: Path, driver_path: Path) -> dict[str, Any]: } +def refresh_bundle(bundle_path: Path, driver_path: Path) -> dict[str, Any]: + """Refresh identity artifacts per docs/bridge-cutover-sealed-runtimes.md.""" + + state = _reconstruct_bundle_state(bundle_path, driver_path) + if state.cutover_phase != "runtime-switched": + raise PreparationError( + "in-place refresh applies only to an applied runtime-switched bundle; " + f"observed cutover phase {state.cutover_phase!r}" + ) + bundle_path = state.bundle_path + spec = state.spec + # Serialize against a concurrent prepare or refresh of the same transaction. + staging = spec.output_dir.parent / ( + f".{spec.output_dir.name}.prepare-" + f"{hashlib.sha256(spec.transaction_id.encode('utf-8')).hexdigest()[:32]}" + ) + _, lock_path = _preparation_control_paths(staging) + descriptor = _open_preparation_lock(lock_path) + try: + worker_state_path = bundle_path.parent / "worker-state.manifest.json" + ( + worker_state_id, + worker_state_lock, + worker_state_journal, + worker_state_snapshot, + ) = _worker_state_transaction_paths(spec) + worker_state_staging = spec.worker_state.snapshot_parent / ( + f".{worker_state_id}.snapshot-staging" + ) + worker_state_flags = ( + os.O_RDWR | os.O_CREAT | getattr(os, "O_NOFOLLOW", 0) + ) + worker_state_descriptor = os.open( + worker_state_lock, worker_state_flags, 0o600 + ) + try: + os.fchmod(worker_state_descriptor, 0o600) + fcntl.flock(worker_state_descriptor, fcntl.LOCK_EX) + try: + if any( + os.path.lexists(path) + for path in ( + worker_state_journal, + worker_state_snapshot, + worker_state_staging, + ) + ): + raise PreparationError( + "in-place refresh must run before the worker-state 6a " + "snapshot, while no worker-state transaction is bound to " + "the manifest; worker-state transaction state currently " + "exists, so refreshing would change the manifest fingerprint " + "and strand any bound transaction and its rollback" + ) + refreshed_bundle = dict(state.bundle) + refreshed_bundle["activation_plan"] = _activation_plan( + state.expected_provision_contract + ) + bundle_payload = ( + json.dumps(refreshed_bundle, indent=2, sort_keys=True) + "\n" + ).encode("utf-8") + worker_state_manifest = _worker_state_manifest_dict( + spec, + bundle_path=bundle_path, + bundle_sha256=hashlib.sha256(bundle_payload).hexdigest(), + cutover_manifest_path=state.manifest_path, + candidate_registry_path=state.new_path, + candidate_registry_sha256=_sha256(state.new_path), + candidate=state.candidate, + provision_contract=state.expected_provision_contract, + ) + # Write the bundle first so the worker-state manifest's bundle_sha256 + # pins the refreshed bundle bytes; both replacements are atomic. + _atomic_replace_bytes(bundle_path, bundle_payload, 0o600) + _atomic_replace_json(worker_state_path, worker_state_manifest, 0o600) + finally: + fcntl.flock(worker_state_descriptor, fcntl.LOCK_UN) + finally: + os.close(worker_state_descriptor) + finally: + fcntl.flock(descriptor, fcntl.LOCK_UN) + os.close(descriptor) + result = validate_bundle(bundle_path, driver_path) + if not result.get("valid") or result.get("cutover_phase") != "runtime-switched": + raise PreparationError( + "in-place refresh did not converge on a strict runtime-switched bundle" + ) + result["refreshed"] = True + return result + + def _validate_input_state_for_existing_bundle( spec: PreparationSpec, ) -> tuple[AgentFleetAPI, AgentFleetAPI]: @@ -6033,6 +6257,14 @@ def _build_parser() -> argparse.ArgumentParser: prepare_parser.add_argument("spec", help="absolute path to the preparation JSON spec") validate_parser = commands.add_parser("validate", help="validate an existing bundle") validate_parser.add_argument("bundle", help="absolute path to bundle.json") + refresh_parser = commands.add_parser( + "refresh", + help=( + "refresh an applied runtime-switched bundle's activation plan and " + "worker-state manifest to the current provider-binary identity" + ), + ) + refresh_parser.add_argument("bundle", help="absolute path to bundle.json") rehearse_parser = commands.add_parser( "rehearse", help="run exhaustive disposable transaction recovery rehearsal" ) @@ -6053,6 +6285,8 @@ def main(argv: Sequence[str] | None = None) -> int: result = prepare(_normalized_absolute(args.spec, "spec path"), driver) elif args.command == "validate": result = validate_bundle(_normalized_absolute(args.bundle, "bundle path"), driver) + elif args.command == "refresh": + result = refresh_bundle(_normalized_absolute(args.bundle, "bundle path"), driver) else: scratch = _normalized_absolute(args.scratch_root, "scratch root") result = rehearse_bundle(