diff --git a/ci/bootstrap-recovery-contract.sh b/ci/bootstrap-recovery-contract.sh
index 39ae37b..e37e1ba 100644
--- a/ci/bootstrap-recovery-contract.sh
+++ b/ci/bootstrap-recovery-contract.sh
@@ -184,13 +184,28 @@ assert_contains "$app_init" 'trap release_cloud_compose_lifecycle_lock EXIT'
 assert_contains "$run_script" 'if cloud_compose_should_run_app_init'
 assert_contains "$run_script" 'cloud_compose_publish_marker "$current_boot_app_init_marker"'
 assert_contains "$run_script" 'cloud_compose_start_and_wait_for_oneshot cloud-compose.service "$app_wait_seconds"'
+assert_contains "$run_script" '"$fresh_filesystem_marker" "$fresh_filesystem_identity"'
 assert_contains "$run_script" 'cloud_compose_publish_marker "$durable_bootstrap_marker"'
+[[ "$(grep -Fc 'cloud_compose_consume_fresh_filesystem_marker \' "$run_script")" == "1" ]] ||
+    fail "bootstrap must have one fresh-filesystem authority consumption boundary"
+rotation_line="$(grep -nF 'bash /home/cloud-compose/rotate-keys-daily.sh' "$run_script" | cut -d: -f1)"
+fresh_marker_line="$(grep -nF 'cloud_compose_consume_fresh_filesystem_marker \' "$run_script" | cut -d: -f1)"
+fresh_marker_sync_line="$(grep -nFx 'sync' "$run_script" | cut -d: -f1)"
+key_timer_line="$(grep -nF 'systemctl enable --now cloud-compose-key-rotation.timer' "$run_script" | cut -d: -f1)"
+vault_line="$(grep -nF 'bash /home/cloud-compose/vault-agent-init.sh' "$run_script" | cut -d: -f1)"
 init_line="$(grep -nF 'run_as_cloud_compose bash /home/cloud-compose/app-init.sh' "$run_script" | cut -d: -f1)"
 boot_marker_line="$(grep -nF 'cloud_compose_publish_marker "$current_boot_app_init_marker"' "$run_script" | cut -d: -f1)"
 app_line="$(grep -nF 'cloud_compose_start_and_wait_for_oneshot cloud-compose.service' "$run_script" | cut -d: -f1)"
 durable_marker_line="$(grep -nF 'cloud_compose_publish_marker "$durable_bootstrap_marker"' "$run_script" | cut -d: -f1)"
-((init_line < boot_marker_line && boot_marker_line < app_line && app_line < durable_marker_line)) ||
-    fail "bootstrap does not publish init, await the app, then publish durable readiness"
+((rotation_line < fresh_marker_line &&
+    fresh_marker_line < fresh_marker_sync_line &&
+    fresh_marker_sync_line < key_timer_line &&
+    key_timer_line < vault_line &&
+    vault_line < init_line &&
+    init_line < boot_marker_line &&
+    boot_marker_line < app_line &&
+    app_line < durable_marker_line)) ||
+    fail "bootstrap does not consume fresh authority durably before persistent rotation, Vault, and application initialization"
 
 assert_contains "$run_bootstrap" 'if ((EUID != 0)); then'
 assert_contains "$run_bootstrap" 'exec bash /home/cloud-compose/run.sh'
@@ -283,7 +298,8 @@ rm -f -- "$durable_marker"
 # Execute the production run script from a relocated fixture so the contract can
 # inject deterministic host commands without writing to /home or /run. The
 # first application-service convergence fails after app-init succeeds. The
-# second full run must reuse the current-boot init marker and publish durable
+# fresh reconciliation authority must already be consumed before that failure;
+# the second run reuses the current-boot init marker and publishes durable
 # readiness after the service recovers.
 integration_root="$tmp/integration"
 integration_home="$integration_root/home/cloud-compose"
@@ -291,11 +307,14 @@ integration_run="$integration_root/run"
 integration_bin="$integration_root/bin"
 integration_log="$integration_root/systemctl.log"
 app_init_count="$integration_root/app-init-count"
+fresh_data_root="$integration_root/data"
+fresh_marker="$fresh_data_root/.cloud-compose/fresh-filesystem"
 first_output="$integration_root/first-attempt.log"
 retry_output="$integration_root/retry-attempt.log"
-mkdir -p "$integration_home" "$integration_run" "$integration_bin"
+mkdir -p "$integration_home" "$integration_run" "$integration_bin" "$(dirname -- "$fresh_marker")"
 : >"$integration_log"
 printf '0\n' >"$app_init_count"
+printf 'fresh\n' >"$fresh_marker"
 
 sed \
     -e "s#/home/cloud-compose#$integration_home#g" \
@@ -344,6 +363,12 @@ cat >"$integration_bin/sleep" <<'EOF'
 #!/usr/bin/env bash
 exit 0
 EOF
+cat >"$integration_bin/sync" <<'EOF'
+#!/usr/bin/env bash
+set -euo pipefail
+printf '%s sync\n' "${RUN_ATTEMPT:?}" >>"${INTEGRATION_SYSTEMCTL_LOG:?}"
+[[ "$RUN_ATTEMPT" != "sync-fail" ]]
+EOF
 cat >"$integration_bin/systemctl" <<'EOF'
 #!/usr/bin/env bash
 set -euo pipefail
@@ -379,6 +404,32 @@ case "$1" in
         ;;
 esac
 EOF
+cat >"$integration_bin/stat" <<'EOF'
+#!/usr/bin/env bash
+set -euo pipefail
+
+target="${!#}"
+format=""
+previous=""
+for argument in "$@"; do
+    if [[ "$previous" == "-c" ]]; then
+        format="$argument"
+    fi
+    previous="$argument"
+done
+marker="${INTEGRATION_FRESH_MARKER:?}"
+case "$target:$format" in
+    "$(dirname -- "$marker"):%u:%g:%a")
+        printf '%s\n' "${FRESH_MARKER_DIR_IDENTITY:-0:0:700}"
+        ;;
+    "$marker:%u:%g:%a:%h")
+        printf '%s\n' "${FRESH_MARKER_IDENTITY:-0:0:600:1}"
+        ;;
+    *)
+        exec /usr/bin/stat "$@"
+        ;;
+esac
+EOF
 chmod +x "$integration_bin"/*
 
 integration_env=(
@@ -387,8 +438,10 @@ integration_env=(
     "CLOUD_COMPOSE_SYSTEMD_POLL_SECONDS=1"
     "CLOUD_COMPOSE_PROVIDER=contract"
     "CLOUD_COMPOSE_DOCKER_PRUNE_ENABLED=false"
+    "CLOUD_COMPOSE_FRESH_FILESYSTEM_MARKER=$fresh_marker"
     "LIBOPS_INTERNAL_SERVICES_ENABLED=false"
     "LIBOPS_MANAGED_RUNTIME_ENABLED=false"
+    "INTEGRATION_FRESH_MARKER=$fresh_marker"
     "INTEGRATION_SYSTEMCTL_LOG=$integration_log"
     "PATH=$integration_bin:/usr/bin:/bin"
 )
@@ -402,6 +455,8 @@ fi
     fail "failed first attempt did not retain current-boot app-init readiness"
 [[ ! -e "$integration_home/.cloud-compose-bootstrap-complete" ]] ||
     fail "failed first attempt published durable bootstrap readiness"
+[[ ! -e "$fresh_marker" ]] ||
+    fail "failed first attempt retained fresh-filesystem authority past key convergence"
 
 env "${integration_env[@]}" RUN_ATTEMPT=recover \
     bash "$integration_home/run.sh" >"$retry_output" 2>&1
@@ -409,6 +464,8 @@ env "${integration_env[@]}" RUN_ATTEMPT=recover \
     fail "bootstrap retry repeated successful app-init"
 [[ -f "$integration_home/.cloud-compose-bootstrap-complete" ]] ||
     fail "bootstrap retry did not publish durable readiness"
+[[ ! -e "$fresh_marker" ]] ||
+    fail "successful bootstrap retry retained fresh-filesystem reconciliation authority"
 grep -Fq 'Application initialization already completed during this boot' "$retry_output" ||
     fail "bootstrap retry did not report reuse of successful app-init"
 grep -Fq 'fail reset-failed -- cloud-compose.service' "$integration_log" ||
@@ -416,4 +473,72 @@ grep -Fq 'fail reset-failed -- cloud-compose.service' "$integration_log" ||
 grep -Fq 'recover enable -- cloud-compose.service' "$integration_log" ||
     fail "bootstrap retry did not converge the application service"
 
+# Marker removal must be flushed before durable readiness can be republished.
+rm -f -- "$integration_home/.cloud-compose-bootstrap-complete"
+printf 'fresh\n' >"$fresh_marker"
+if env "${integration_env[@]}" RUN_ATTEMPT=sync-fail \
+    bash "$integration_home/run.sh" >/dev/null 2>&1; then
+    fail "bootstrap accepted a failed post-consume durability barrier"
+fi
+[[ ! -e "$fresh_marker" ]] ||
+    fail "post-consume durability coverage did not remove the fresh marker"
+[[ ! -e "$integration_home/.cloud-compose-bootstrap-complete" ]] ||
+    fail "failed post-consume durability barrier published readiness"
+env "${integration_env[@]}" RUN_ATTEMPT=recover \
+    bash "$integration_home/run.sh" >/dev/null 2>&1
+[[ -f "$integration_home/.cloud-compose-bootstrap-complete" ]] ||
+    fail "bootstrap retry did not flush an already-absent marker before readiness"
+
+# GCP never falls back to the generic non-GCP marker identity. This check runs
+# before key rotation, so missing disk identity cannot reach IAM.
+rm -f -- "$integration_home/.cloud-compose-bootstrap-complete"
+printf 'fresh\n' >"$fresh_marker"
+if env "${integration_env[@]}" CLOUD_COMPOSE_PROVIDER=gcp RUN_ATTEMPT=recover \
+    bash "$integration_home/run.sh" >/dev/null 2>&1; then
+    fail "GCP bootstrap accepted the generic fresh-filesystem identity"
+fi
+[[ -f "$fresh_marker" ]] ||
+    fail "GCP bootstrap consumed generic fresh-filesystem authority"
+[[ ! -e "$integration_home/.cloud-compose-bootstrap-complete" ]] ||
+    fail "GCP bootstrap with generic authority published durable readiness"
+
+# A marker payload for another incarnation must fail before application
+# initialization or durable readiness.
+rm -f -- "$integration_home/.cloud-compose-bootstrap-complete"
+printf 'v1:gcp-disk-id:111111111111111111\n' >"$fresh_marker"
+if env "${integration_env[@]}" RUN_ATTEMPT=recover \
+    bash "$integration_home/run.sh" >/dev/null 2>&1; then
+    fail "bootstrap accepted a mismatched fresh-filesystem marker payload"
+fi
+[[ -f "$fresh_marker" ]] ||
+    fail "mismatched fresh-filesystem marker payload was consumed"
+[[ ! -e "$integration_home/.cloud-compose-bootstrap-complete" ]] ||
+    fail "mismatched fresh-filesystem marker payload published durable readiness"
+
+# Unsafe authority must fail closed before durable readiness. The current-boot
+# app-init marker remains valid, so these attempts exercise only the early
+# marker boundary rather than repeating application initialization.
+rm -f -- "$integration_home/.cloud-compose-bootstrap-complete"
+printf 'fresh\n' >"$fresh_marker"
+if env "${integration_env[@]}" RUN_ATTEMPT=recover \
+    FRESH_MARKER_IDENTITY=1000:1000:600:1 \
+    bash "$integration_home/run.sh" >/dev/null 2>&1; then
+    fail "bootstrap accepted a fresh-filesystem marker with unsafe ownership"
+fi
+[[ -f "$fresh_marker" ]] ||
+    fail "unsafe fresh-filesystem marker was consumed"
+[[ ! -e "$integration_home/.cloud-compose-bootstrap-complete" ]] ||
+    fail "unsafe fresh-filesystem marker published durable readiness"
+
+rm -f -- "$fresh_marker"
+ln -s /dev/null "$fresh_marker"
+if env "${integration_env[@]}" RUN_ATTEMPT=recover \
+    bash "$integration_home/run.sh" >/dev/null 2>&1; then
+    fail "bootstrap accepted a symlink fresh-filesystem marker"
+fi
+[[ -L "$fresh_marker" ]] ||
+    fail "symlink fresh-filesystem marker was consumed"
+[[ ! -e "$integration_home/.cloud-compose-bootstrap-complete" ]] ||
+    fail "symlink fresh-filesystem marker published durable readiness"
+
 echo "Bootstrap recovery contract passed"
diff --git a/ci/filesystem-prep-contract.sh b/ci/filesystem-prep-contract.sh
index d616186..270e730 100644
--- a/ci/filesystem-prep-contract.sh
+++ b/ci/filesystem-prep-contract.sh
@@ -17,6 +17,9 @@ mkdir -p "$tmp/bin" "$tmp/dev" "$tmp/mount" "$tmp/systemd"
 device="$tmp/dev/data"
 : >"$device"
 calls="$tmp/calls"
+fresh_marker="$tmp/mount/.cloud-compose/fresh-filesystem"
+label_state="$tmp/label-state"
+: >"$label_state"
 
 cat >"$tmp/bin/readlink" <<'EOF'
 #!/usr/bin/env bash
@@ -41,6 +44,30 @@ EOF
 cat >"$tmp/bin/mkfs.ext4" <<'EOF'
 #!/usr/bin/env bash
 printf 'mkfs\n' >>"${CALLS:?}"
+label=""
+previous=""
+for argument in "$@"; do
+  if [[ "$previous" == "-L" ]]; then
+    label="$argument"
+  fi
+  previous="$argument"
+done
+printf '%s' "$label" >"${FAKE_LABEL_STATE:?}"
+EOF
+cat >"$tmp/bin/e2label" <<'EOF'
+#!/usr/bin/env bash
+case "$#" in
+  1)
+    cat "${FAKE_LABEL_STATE:?}"
+    ;;
+  2)
+    printf '%s' "$2" >"${FAKE_LABEL_STATE:?}"
+    printf 'label-set %s\n' "$2" >>"${CALLS:?}"
+    ;;
+  *)
+    exit 2
+    ;;
+esac
 EOF
 cat >"$tmp/bin/resize2fs" <<'EOF'
 #!/usr/bin/env bash
@@ -65,9 +92,57 @@ EOF
 cat >"$tmp/bin/mount" <<'EOF'
 #!/usr/bin/env bash
 printf 'mount\n' >>"${CALLS:?}"
-if [[ "${1:-}" == "--move" ]]; then
+status="${FAKE_MOUNT_STATUS:-0}"
+if [[ "${1:-}" == "--move" && "$status" == "0" ]]; then
   printf '%s\n' "${3:?}" >"${FAKE_MOUNT_STATE:?}"
 fi
+exit "$status"
+EOF
+cat >"$tmp/bin/stat" <<'EOF'
+#!/usr/bin/env bash
+set -euo pipefail
+
+target="${!#}"
+format=""
+previous=""
+for argument in "$@"; do
+  if [[ "$previous" == "-c" ]]; then
+    format="$argument"
+  fi
+  previous="$argument"
+done
+case "$target:$format" in
+  "${MOUNT_PATH:?}:%u:%g:%a")
+    printf '%s\n' "${FAKE_FRESH_ROOT_IDENTITY:-0:0:755}"
+    ;;
+  "$MOUNT_PATH/.cloud-compose:%u:%g:%a")
+    printf '%s\n' "${FAKE_MARKER_DIR_IDENTITY:-0:0:700}"
+    ;;
+  "$MOUNT_PATH/.cloud-compose/fresh-filesystem:%u:%g:%a:%h")
+    printf 'marker\n' >>"${CALLS:?}"
+    printf '%s\n' "${FAKE_MARKER_IDENTITY:-0:0:600:1}"
+    ;;
+  "$MOUNT_PATH/lost+found:%u:%g:%a")
+    printf '%s\n' "${FAKE_LOST_FOUND_IDENTITY:-0:0:700}"
+    ;;
+  *)
+    exec /usr/bin/stat "$@"
+    ;;
+esac
+EOF
+cat >"$tmp/bin/sync" <<'EOF'
+#!/usr/bin/env bash
+set -euo pipefail
+count=0
+if [[ -s "${FAKE_SYNC_COUNT:?}" ]]; then
+  count="$(<"$FAKE_SYNC_COUNT")"
+fi
+count=$((count + 1))
+printf '%s\n' "$count" >"$FAKE_SYNC_COUNT"
+printf 'sync\n' >>"${CALLS:?}"
+if [[ -n "${FAKE_SYNC_FAIL_AT:-}" && "$count" == "$FAKE_SYNC_FAIL_AT" ]]; then
+  exit 1
+fi
 EOF
 cat >"$tmp/bin/findmnt" <<'EOF'
 #!/usr/bin/env bash
@@ -136,18 +211,29 @@ run_prep() {
     : >"$calls"
     rm -f -- "$tmp/mount-state"
     rm -f -- "$tmp/findmnt-count"
+    printf '0\n' >"$tmp/sync-count"
     FAKE_DEVICE="$configured_device" FAKE_MOUNT_STATE="$tmp/mount-state" \
-        FAKE_FINDMNT_COUNT="$tmp/findmnt-count" CALLS="$calls" \
+        FAKE_FINDMNT_COUNT="$tmp/findmnt-count" FAKE_LABEL_STATE="$label_state" \
+        FAKE_SYNC_COUNT="$tmp/sync-count" CALLS="$calls" \
         PATH="$tmp/bin:/usr/bin:/bin" PREP_SCRIPT="$prep_script" MOUNT_PATH="$tmp/mount" \
         CLOUD_COMPOSE_SYSTEMD_DIR="$tmp/systemd" \
         DEVICE_PATH="${TEST_DEVICE_PATH:-/dev/disk/by-id/test-data}" \
+        PREP_EXTRA_ARG="${TEST_PREP_EXTRA_ARG:-}" \
+        PREP_MARKER_IDENTITY="${TEST_PREP_MARKER_IDENTITY:-}" \
         bash --noprofile --norc -c '
             source "$PREP_SCRIPT"
             if [[ -n "${TEST_PROVIDER_MOUNT:-}" ]]; then
                 digitalocean_automatic_mount() { printf "%s\n" "$TEST_PROVIDER_MOUNT"; }
             fi
             is_block_device() { return 0; }
-            main "$DEVICE_PATH" "$MOUNT_PATH"
+            prep_args=("$DEVICE_PATH" "$MOUNT_PATH")
+            if [[ -n "$PREP_EXTRA_ARG" ]]; then
+                prep_args+=("$PREP_EXTRA_ARG")
+            fi
+            if [[ -n "$PREP_MARKER_IDENTITY" ]]; then
+                prep_args+=("$PREP_MARKER_IDENTITY")
+            fi
+            main "${prep_args[@]}"
         '
 }
 
@@ -183,6 +269,144 @@ if grep -Fxq resize "$calls"; then
     fail "a newly formatted whole-disk filesystem was resized redundantly"
 fi
 
+rm -rf -- "$tmp/mount/.cloud-compose" "$tmp/mount/lost+found"
+if TEST_PREP_EXTRA_ARG=--unsupported FAKE_BLKID_STATUS=2 run_prep >/dev/null 2>&1; then
+    fail "an unsupported filesystem preparation option was accepted"
+fi
+[[ ! -e "$fresh_marker" ]] ||
+    fail "an unsupported filesystem preparation option published a fresh-filesystem marker"
+if [[ -s "$calls" ]]; then
+    fail "an unsupported filesystem preparation option mutated the filesystem"
+fi
+
+TEST_PREP_EXTRA_ARG=--publish-fresh-marker \
+    FAKE_BLKID_STATUS=0 FAKE_FILESYSTEM_TYPE=ext4 FAKE_FSCK_STATUS=0 run_prep
+[[ ! -e "$fresh_marker" ]] ||
+    fail "an existing filesystem was marked as freshly formatted"
+if grep -Fxq marker "$calls"; then
+    fail "an existing filesystem invoked fresh-filesystem marker publication"
+fi
+
+if TEST_PREP_EXTRA_ARG=--publish-fresh-marker \
+    FAKE_BLKID_STATUS=2 FAKE_MOUNT_STATUS=1 run_prep >/dev/null 2>&1; then
+    fail "a failed first mount was accepted"
+fi
+grep -Fxq mkfs "$calls" || fail "mount failure coverage did not format the unsigned device"
+grep -Fxq mount "$calls" || fail "mount failure coverage did not attempt the first mount"
+[[ ! -e "$fresh_marker" ]] ||
+    fail "a fresh-filesystem marker was published before the first mount succeeded"
+if grep -Fxq marker "$calls"; then
+    fail "a failed first mount invoked fresh-filesystem marker publication"
+fi
+[[ "$(<"$label_state")" == "cc-fresh-pending" ]] ||
+    fail "a failed first mount did not retain durable fresh-filesystem intent"
+
+printf 'not-pristine\n' >"$tmp/mount/application-data"
+if TEST_PREP_EXTRA_ARG=--publish-fresh-marker \
+    FAKE_BLKID_STATUS=0 FAKE_FILESYSTEM_TYPE=ext4 FAKE_FSCK_STATUS=0 \
+    run_prep >/dev/null 2>&1; then
+    fail "a pending ext4 label authorized a populated filesystem"
+fi
+[[ ! -e "$fresh_marker" && "$(<"$label_state")" == "cc-fresh-pending" ]] ||
+    fail "populated-filesystem rejection changed fresh-marker authority"
+rm -f -- "$tmp/mount/application-data"
+
+mkdir -p "$tmp/mount/lost+found"
+chmod 0700 "$tmp/mount/lost+found"
+printf 'recovered-data\n' >"$tmp/mount/lost+found/orphan"
+if TEST_PREP_EXTRA_ARG=--publish-fresh-marker \
+    FAKE_BLKID_STATUS=0 FAKE_FILESYSTEM_TYPE=ext4 FAKE_FSCK_STATUS=0 \
+    run_prep >/dev/null 2>&1; then
+    fail "a pending ext4 label accepted data inside lost+found"
+fi
+rm -f -- "$tmp/mount/lost+found/orphan"
+if TEST_PREP_EXTRA_ARG=--publish-fresh-marker \
+    FAKE_BLKID_STATUS=0 FAKE_FILESYSTEM_TYPE=ext4 FAKE_FSCK_STATUS=0 \
+    FAKE_LOST_FOUND_IDENTITY=1000:1000:700 run_prep >/dev/null 2>&1; then
+    fail "a pending ext4 label accepted an untrusted lost+found directory"
+fi
+
+if TEST_PREP_EXTRA_ARG=--publish-fresh-marker \
+    FAKE_BLKID_STATUS=0 FAKE_FILESYSTEM_TYPE=ext4 FAKE_FSCK_STATUS=0 \
+    FAKE_SYNC_FAIL_AT=1 run_prep >/dev/null 2>&1; then
+    fail "fresh marker publication accepted a failed durability barrier"
+fi
+[[ -f "$fresh_marker" && "$(<"$label_state")" == "cc-fresh-pending" ]] ||
+    fail "the pending label was cleared before marker publication became durable"
+if grep -Eq '^label-set ' "$calls"; then
+    fail "a failed marker durability barrier cleared the pending label"
+fi
+
+fresh_marker_inode="$(stat -c '%d:%i' "$fresh_marker")"
+printf 'v1:gcp-disk-id:111111111111111111\n' >"$fresh_marker"
+if TEST_PREP_EXTRA_ARG=--publish-fresh-marker \
+    TEST_PREP_MARKER_IDENTITY=v1:gcp-disk-id:222222222222222222 \
+    FAKE_BLKID_STATUS=0 FAKE_FILESYSTEM_TYPE=ext4 FAKE_FSCK_STATUS=0 \
+    run_prep >/dev/null 2>&1; then
+    fail "a pending label accepted a marker from another disk incarnation"
+fi
+[[ "$(<"$label_state")" == "cc-fresh-pending" ]] ||
+    fail "a mismatched disk-incarnation marker cleared the pending label"
+if grep -Eq '^label-set ' "$calls"; then
+    fail "a mismatched disk-incarnation marker cleared its pending label"
+fi
+printf 'fresh\n' >"$fresh_marker"
+if TEST_PREP_EXTRA_ARG=--publish-fresh-marker \
+    FAKE_BLKID_STATUS=0 FAKE_FILESYSTEM_TYPE=ext4 FAKE_FSCK_STATUS=0 \
+    FAKE_MARKER_IDENTITY=1000:1000:600:1 run_prep >/dev/null 2>&1; then
+    fail "a pending label accepted an untrusted existing marker"
+fi
+[[ "$(<"$label_state")" == "cc-fresh-pending" ]] ||
+    fail "an untrusted existing marker cleared the pending label"
+
+TEST_PREP_EXTRA_ARG=--publish-fresh-marker \
+    FAKE_BLKID_STATUS=0 FAKE_FILESYSTEM_TYPE=ext4 FAKE_FSCK_STATUS=0 run_prep
+[[ -f "$fresh_marker" ]] ||
+    fail "a retry did not recover fresh-filesystem marker publication"
+if grep -Fxq mkfs "$calls"; then
+    fail "fresh-filesystem marker recovery reformatted the existing filesystem"
+fi
+[[ ! -s "$label_state" ]] ||
+    fail "successful fresh-filesystem marker publication retained the pending label"
+[[ "$(stat -c '%a' "$tmp/mount/.cloud-compose")" == "700" ]] ||
+    fail "the fresh-filesystem marker directory was not mode 0700"
+[[ "$(stat -c '%a' "$fresh_marker")" == "600" ]] ||
+    fail "the fresh-filesystem marker was not mode 0600"
+[[ "$(<"$fresh_marker")" == "fresh" ]] ||
+    fail "generic fresh-filesystem marker did not preserve its exact payload"
+mount_line="$(grep -nFx mount "$calls" | cut -d: -f1)"
+marker_line="$(grep -nFx marker "$calls" | cut -d: -f1)"
+label_clear_line="$(grep -nE '^label-set $' "$calls" | cut -d: -f1)"
+mapfile -t sync_lines < <(grep -nFx sync "$calls" | cut -d: -f1)
+[[ -n "$mount_line" && -n "$marker_line" && -n "$label_clear_line" &&
+    "${#sync_lines[@]}" == "2" &&
+    "$mount_line" -lt "$marker_line" &&
+    "$marker_line" -lt "${sync_lines[0]}" &&
+    "${sync_lines[0]}" -lt "$label_clear_line" &&
+    "$label_clear_line" -lt "${sync_lines[1]}" ]] ||
+    fail "fresh authority ordering is not mount, marker, sync, label clear, sync"
+[[ "$(stat -c '%d:%i' "$fresh_marker")" == "$fresh_marker_inode" ]] ||
+    fail "fresh marker recovery replaced an already trusted marker"
+
+TEST_PREP_EXTRA_ARG=--publish-fresh-marker \
+    FAKE_BLKID_STATUS=0 FAKE_FILESYSTEM_TYPE=ext4 FAKE_FSCK_STATUS=0 run_prep
+[[ -f "$fresh_marker" && "$(stat -c '%d:%i' "$fresh_marker")" == "$fresh_marker_inode" ]] ||
+    fail "an existing fresh-filesystem marker was not preserved on a same-boot rerun"
+if grep -Fxq marker "$calls"; then
+    fail "a same-boot rerun republished the fresh-filesystem marker"
+fi
+
+rm -rf -- "$tmp/mount/.cloud-compose" "$tmp/mount/lost+found"
+printf 'cc-fresh-pending' >"$label_state"
+TEST_PREP_EXTRA_ARG=--publish-fresh-marker \
+    TEST_PREP_MARKER_IDENTITY=v1:gcp-disk-id:987654321012345678 \
+    FAKE_BLKID_STATUS=0 FAKE_FILESYSTEM_TYPE=ext4 FAKE_FSCK_STATUS=0 run_prep
+[[ "$(<"$fresh_marker")" == "v1:gcp-disk-id:987654321012345678" ]] ||
+    fail "GCP marker publication did not bind the exact data-disk identity"
+[[ ! -s "$label_state" ]] ||
+    fail "exact GCP disk identity did not complete pending marker publication"
+rm -rf -- "$tmp/mount/.cloud-compose" "$tmp/mount/lost+found"
+
 if FAKE_BLKID_STATUS=0 FAKE_FILESYSTEM_TYPE=xfs FAKE_FSCK_STATUS=0 run_prep >/dev/null 2>&1; then
     fail "a non-ext4 filesystem was accepted"
 fi
@@ -371,10 +595,33 @@ if grep -Eq 'fsck[^\n]*\|\|[^\n]*mkfs|fsck[^\n]*mkfs' "$repo_root/templates/clou
 fi
 grep -Fq 'FILESYSTEM_PREP_SCRIPT_B64' "$repo_root/templates/cloud-init.yml" || \
     fail "GCP cloud-init does not bootstrap the tested filesystem helper"
-grep -Fq 'bash "$filesystem_prep" /dev/disk/by-id/google-data /mnt/disks/data' "$repo_root/templates/cloud-init.yml" || \
-    fail "GCP cloud-init executes a temporary filesystem helper directly on potentially noexec /run"
-grep -Fq 'bash "$filesystem_prep" '\''${DATA_DEVICE}'\'' /mnt/disks/data' "$repo_root/modules/linux-vm-runtime/templates/cloud-init.yml" || \
-    fail "Linux VM cloud-init executes a temporary filesystem helper directly on potentially noexec /run"
+grep -Fq 'bash "$filesystem_prep" /dev/disk/by-id/google-data /mnt/disks/data \' \
+    "$repo_root/templates/cloud-init.yml" ||
+    fail "GCP cloud-init does not request a fresh marker for the data filesystem"
+grep -Fq -- '--publish-fresh-marker ${jsonencode(FRESH_FILESYSTEM_IDENTITY)}' \
+    "$repo_root/templates/cloud-init.yml" ||
+    fail "GCP cloud-init does not bind the marker to the rendered data-disk identity"
+grep -Fq 'bash "$filesystem_prep" '\''${DATA_DEVICE}'\'' /mnt/disks/data --publish-fresh-marker' \
+    "$repo_root/modules/linux-vm-runtime/templates/cloud-init.yml" ||
+    fail "Linux VM cloud-init does not request a fresh marker for the data filesystem"
+grep -Fq 'bash /run/cloud-compose-prepare-filesystem /dev/disk/by-id/google-docker-volumes /mnt/disks/volumes' \
+    "$repo_root/templates/cloud-init.yml" ||
+    fail "GCP cloud-init does not prepare the volumes filesystem without a fresh marker"
+grep -Fq 'bash "$filesystem_prep" '\''${VOLUMES_DEVICE}'\'' /mnt/disks/volumes' \
+    "$repo_root/modules/linux-vm-runtime/templates/cloud-init.yml" ||
+    fail "Linux VM cloud-init does not prepare the volumes filesystem without a fresh marker"
+grep -Fq '(umask 077 && mkdir -- "$marker_dir")' "$prep_script" ||
+    fail "the fresh-filesystem marker directory is not created privately"
+grep -Fq "(umask 077 && printf '%s\\n' \"\$expected_identity\" >\"\$marker\")" "$prep_script" ||
+    fail "the fresh-filesystem marker is not created privately"
+grep -Fq 'marker_root_identity="$(stat -c '\''%u:%g:%a'\'' -- "$mount_path")"' "$prep_script" ||
+    fail "marker recovery does not verify a root-owned pristine filesystem root"
+grep -Fq '! -name lost+found -print -quit' "$prep_script" ||
+    fail "marker recovery does not reject data outside lost+found"
+grep -Fq -- '-L "$fresh_filesystem_pending_label" -- "$device"' "$prep_script" ||
+    fail "new data filesystems do not persist pending marker intent in their ext4 label"
+grep -Fq 'e2label "$device" ""' "$prep_script" ||
+    fail "fresh-filesystem pending intent is not cleared after marker publication"
 if grep -Eq '^[[:space:]]*"?\$filesystem_(prep|persist)"?[[:space:]]' \
     "$repo_root/templates/cloud-init.yml" "$repo_root/modules/linux-vm-runtime/templates/cloud-init.yml"; then
     fail "cloud-init directly executes a temporary helper from potentially noexec /run"
@@ -382,6 +629,16 @@ fi
 for cloud_init_template in \
     "$repo_root/templates/cloud-init.yml" \
     "$repo_root/modules/linux-vm-runtime/templates/cloud-init.yml"; do
+    [[ "$(grep -Fc -- '--publish-fresh-marker' "$cloud_init_template")" == "1" ]] ||
+        fail "cloud-init must publish a fresh marker for the data filesystem only"
+    grep -Fq 'chown root:cloud-compose /mnt/disks/data' "$cloud_init_template" ||
+        fail "cloud-init does not preserve root ownership of the data mount root"
+    grep -Fq 'chmod 1775 /mnt/disks/data' "$cloud_init_template" ||
+        fail "cloud-init does not make the data mount root sticky and group-writable"
+    grep -Fq 'chown cloud-compose:cloud-compose /mnt/disks/volumes' "$cloud_init_template" ||
+        fail "cloud-init does not assign the volumes mount root to cloud-compose"
+    grep -Fq 'chmod 0775 /mnt/disks/volumes' "$cloud_init_template" ||
+        fail "cloud-init does not make the volumes mount root group-writable"
     grep -Fq 'for required_mount in /mnt/disks/data /mnt/disks/volumes /mnt/disks/data/docker/volumes; do' \
         "$cloud_init_template" || fail "cloud-init does not verify every required mount before initialization"
     grep -Fq 'Required cloud-compose mount is unavailable:' "$cloud_init_template" ||
diff --git a/ci/host-runtime-security.sh b/ci/host-runtime-security.sh
index c40d4ab..6a76810 100644
--- a/ci/host-runtime-security.sh
+++ b/ci/host-runtime-security.sh
@@ -176,8 +176,8 @@ assert_contains "$host_init" 'chown root:root /home/cloud-compose'
 assert_contains "$host_init" "-exec chown root:root {} +"
 assert_contains "$host_init" '/home/cloud-compose/.sitectl \'
 assert_contains "$host_init" 'install -d -m 0750 -o cloud-compose -g cloud-compose "$mutable_dir"'
-assert_contains "$host_init" 'for mutable_root in /mnt/disks/data /mnt/disks/volumes; do'
-assert_contains "$host_init" 'install -d -m 0775 -o cloud-compose -g cloud-compose "$mutable_root"'
+assert_contains "$host_init" 'install -d -m 1775 -o root -g cloud-compose /mnt/disks/data'
+assert_contains "$host_init" 'install -d -m 0775 -o cloud-compose -g cloud-compose /mnt/disks/volumes'
 assert_contains "$host_init" 'install -d -m 0775 -o cloud-compose -g cloud-compose /mnt/disks/data/libops'
 if grep -A12 -F 'source_compose_app_env()' "$repo_root/rootfs/home/cloud-compose/compose-apps.sh" | \
     grep -Eq 'source[[:space:]]+.*COMPOSE_APPS_ENV_DIR'; then
diff --git a/ci/key-rotation-contract.sh b/ci/key-rotation-contract.sh
index 8eff38f..353f7d8 100644
--- a/ci/key-rotation-contract.sh
+++ b/ci/key-rotation-contract.sh
@@ -15,10 +15,14 @@ mkdir -p "$tmp/bin" "$tmp/apps/alpha" "$tmp/apps/beta"
 order_log="$tmp/order.log"
 post_count="$tmp/post-count"
 auth_count="$tmp/auth-count"
+list_count="$tmp/list-count"
+sync_count="$tmp/sync-count"
 key_state="$tmp/key-state.json"
 : >"$order_log"
 printf '0\n' >"$post_count"
 printf '0\n' >"$auth_count"
+printf '0\n' >"$list_count"
+printf '0\n' >"$sync_count"
 
 cat >"$tmp/profile.sh" <<'EOF'
 #!/usr/bin/env bash
@@ -61,6 +65,17 @@ cat >"$tmp/bin/sleep" <<'EOF'
 #!/usr/bin/env bash
 printf 'SLEEP %s\n' "${1:-}" >>"${ORDER_LOG:?}"
 EOF
+cat >"$tmp/bin/sync" <<'EOF'
+#!/usr/bin/env bash
+set -euo pipefail
+count="$(<"${SYNC_COUNT:?}")"
+count=$((count + 1))
+printf '%s\n' "$count" >"$SYNC_COUNT"
+printf 'SYNC %s\n' "$count" >>"${ORDER_LOG:?}"
+if [[ -n "${FAKE_SYNC_FAIL_AT:-}" && "$count" == "$FAKE_SYNC_FAIL_AT" ]]; then
+  exit 1
+fi
+EOF
 cat >"$tmp/bin/openssl" <<'EOF'
 #!/usr/bin/env bash
 set -euo pipefail
@@ -77,11 +92,46 @@ cat >"$tmp/bin/systemctl" <<'EOF'
 #!/usr/bin/env bash
 set -euo pipefail
 case "${1:-}" in
+  show)
+    property=""
+    for argument in "$@"; do
+      case "$argument" in
+        --property=*) property="${argument#--property=}" ;;
+      esac
+    done
+    unit="${!#}"
+    if [[ "${FAKE_SYSTEMCTL_QUERY_FAIL:-false}" == "true" ]]; then
+      exit 1
+    fi
+    case "$property" in
+      LoadState) printf '%s\n' "${FAKE_SYSTEMCTL_LOAD_STATE:-loaded}" ;;
+      ActiveState)
+        case "$unit" in
+          cloud-compose.service) printf '%s\n' "${FAKE_APP_STATE:-inactive}" ;;
+          cloud-compose-internal-services.service) printf '%s\n' "${FAKE_INTERNAL_STATE:-inactive}" ;;
+          *) exit 1 ;;
+        esac
+        ;;
+      *) exit 2 ;;
+    esac
+    ;;
   is-active)
     unit="${!#}"
     case "$unit" in
-      cloud-compose.service) [[ "${FAKE_APP_ACTIVE:-true}" == "true" ]] ;;
-      cloud-compose-internal-services.service) [[ "${FAKE_INTERNAL_ACTIVE:-true}" == "true" ]] ;;
+      cloud-compose.service)
+        if [[ -n "${FAKE_APP_ACTIVE+x}" ]]; then
+          [[ "$FAKE_APP_ACTIVE" == "true" ]]
+        else
+          [[ "${FAKE_APP_STATE:-active}" == "active" ]]
+        fi
+        ;;
+      cloud-compose-internal-services.service)
+        if [[ -n "${FAKE_INTERNAL_ACTIVE+x}" ]]; then
+          [[ "$FAKE_INTERNAL_ACTIVE" == "true" ]]
+        else
+          [[ "${FAKE_INTERNAL_STATE:-active}" == "active" ]]
+        fi
+        ;;
       *) exit 1 ;;
     esac
     ;;
@@ -107,6 +157,31 @@ case "${1:-}" in
 esac
 EOF
 
+cat >"$tmp/bin/stat" <<'EOF'
+#!/usr/bin/env bash
+set -euo pipefail
+
+target="${!#}"
+format=""
+previous=""
+for argument in "$@"; do
+  if [[ "$previous" == "-c" ]]; then
+    format="$argument"
+  fi
+  previous="$argument"
+done
+if [[ -n "${FAKE_FRESH_MARKER:-}" ]]; then
+  marker_dir="$(dirname -- "$FAKE_FRESH_MARKER")"
+  marker_root="$(dirname -- "$marker_dir")"
+  case "$target:$format" in
+    "$marker_root:%u:%a") printf '0:1775\n'; exit 0 ;;
+    "$marker_dir:%u:%g:%a") printf '0:0:700\n'; exit 0 ;;
+    "$FAKE_FRESH_MARKER:%u:%g:%a:%h") printf '0:0:600:1\n'; exit 0 ;;
+  esac
+fi
+exec /usr/bin/stat "$@"
+EOF
+
 cat >"$tmp/bin/curl" <<'EOF'
 #!/usr/bin/env bash
 set -euo pipefail
@@ -138,7 +213,10 @@ write_response() {
 }
 
 write_status() {
-  local status="$1" body="${2:-{}}"
+  local status="$1" body="${2-}"
+  if [[ -z "$body" ]]; then
+    body='{}'
+  fi
   write_response "$body"
   if [[ -n "$write_format" ]]; then
     printf '%s' "$status"
@@ -163,8 +241,21 @@ if [[ "$url" == "https://oauth2.googleapis.com/token" ]]; then
 fi
 
 if [[ "$method" == GET && "$url" == */keys ]]; then
+  count="$(<"${LIST_COUNT:?}")"
+  count=$((count + 1))
+  printf '%s\n' "$count" >"$LIST_COUNT"
+  if [[ -n "${FAKE_INJECT_KEY_ON_LIST:-}" &&
+    "$count" == "${FAKE_INJECT_KEY_ON_LIST_NUMBER:-2}" ]]; then
+    service_account="${url#*serviceAccounts/}"
+    service_account="${service_account%/keys}"
+    injected_name="projects/test-project/serviceAccounts/${service_account}/keys/${FAKE_INJECT_KEY_ON_LIST}"
+    jq --arg name "$injected_name" \
+      '. + [{name: $name, disabled: false}] | unique_by(.name)' \
+      "$KEY_STATE" >"$KEY_STATE.tmp"
+    mv "$KEY_STATE.tmp" "$KEY_STATE"
+  fi
   jq -c '{keys: [.[] | {
-    keyType: "USER_MANAGED",
+    keyType: (.keyType // "USER_MANAGED"),
     name: .name,
     disabled: .disabled
   }]}' "${KEY_STATE:?}"
@@ -178,13 +269,22 @@ if [[ "$method" == POST && "$url" == */keys ]]; then
   service_account="${url#*serviceAccounts/}"
   service_account="${service_account%/keys}"
   key_name="projects/test-project/serviceAccounts/${service_account}/keys/${NEW_KEY_ID:?}"
-  if [[ "${FAKE_POST_MODE:-success}" != "fail-no-create" ]]; then
+  if [[ "${FAKE_POST_MODE:-success}" != "fail-no-create" &&
+    "${FAKE_POST_MODE:-success}" != "rejected-400" ]]; then
     jq --arg name "$key_name" '. + [{name: $name, disabled: false}] | unique_by(.name)' \
       "$KEY_STATE" >"$KEY_STATE.tmp"
     mv "$KEY_STATE.tmp" "$KEY_STATE"
   fi
   case "${FAKE_POST_MODE:-success}" in
     fail-no-create | ambiguous-create) exit 22 ;;
+    rejected-400)
+      write_status 400 '{"error":{"code":400,"message":"quota"}}'
+      exit 0
+      ;;
+    ambiguous-408)
+      write_status 408 '{"error":{"code":408,"message":"timeout"}}'
+      exit 0
+      ;;
   esac
   key_data="$(jq -nc --arg id "$NEW_KEY_ID" --arg email "$service_account" '{
     type: "service_account",
@@ -194,7 +294,11 @@ if [[ "$method" == POST && "$url" == */keys ]]; then
     client_email: $email,
     token_uri: "https://oauth2.googleapis.com/token"
   }' | base64 | tr -d '\n')"
-  printf '{"name":"%s","privateKeyData":"%s"}\n' "$key_name" "$key_data"
+  body="$(printf '{"name":"%s","privateKeyData":"%s"}' "$key_name" "$key_data")"
+  write_status 200 "$body"
+  if [[ "${FAKE_POST_MODE:-success}" == "late-success" ]]; then
+    exit 18
+  fi
   exit 0
 fi
 
@@ -246,6 +350,11 @@ if [[ "$method" == DELETE ]]; then
   jq --arg name "$key_name" 'map(select(.name != $name))' "$KEY_STATE" >"$KEY_STATE.tmp"
   mv "$KEY_STATE.tmp" "$KEY_STATE"
   printf 'DELETE %s\n' "$key_name" >>"${ORDER_LOG:?}"
+  if [[ -n "${FAKE_DELETE_FAIL_AFTER_MUTATION_ID:-}" &&
+    "$key_name" == */keys/"$FAKE_DELETE_FAIL_AFTER_MUTATION_ID" ]]; then
+    write_status 500
+    exit 0
+  fi
   write_status 200
   exit 0
 fi
@@ -261,6 +370,8 @@ export TEST_BIN="$tmp/bin"
 export ORDER_LOG="$order_log"
 export POST_COUNT="$post_count"
 export AUTH_COUNT="$auth_count"
+export LIST_COUNT="$list_count"
+export SYNC_COUNT="$sync_count"
 export KEY_STATE="$key_state"
 export CLOUD_COMPOSE_PROFILE_PATH="$tmp/profile.sh"
 export CLOUD_COMPOSE_COMPOSE_APPS_PATH="$tmp/compose-apps.sh"
@@ -277,6 +388,8 @@ export ROTATION_AUTH_SLEEP_SECONDS=0
 export ROTATION_RECOVERY_SETTLE_SECONDS=0
 export ROTATION_CREDENTIAL_OWNER=
 export ROTATION_CREDENTIAL_GROUP=test-group
+export ROTATION_RUNTIME_DIR="$tmp/run/key-rotation"
+export GCP_APP_SERVICE_ACCOUNT_MANAGED=false
 export NEW_KEY_ID="replacement-key"
 export EXPECTED_CONSUMER_KEY_ID="$NEW_KEY_ID"
 
@@ -284,13 +397,32 @@ key_name() {
   printf 'projects/test-project/serviceAccounts/%s/keys/%s\n' "$1" "$2"
 }
 
+reset_operation_counts() {
+  : >"$order_log"
+  printf '0\n' >"$post_count"
+  printf '0\n' >"$auth_count"
+  printf '0\n' >"$list_count"
+  printf '0\n' >"$sync_count"
+}
+
 reset_key_state() {
   local email="$1" old_id="$2"
   jq -n --arg name "$(key_name "$email" "$old_id")" \
     '[{name: $name, disabled: false}]' >"$key_state"
-  : >"$order_log"
-  printf '0\n' >"$post_count"
-  printf '0\n' >"$auth_count"
+  reset_operation_counts
+}
+
+reset_key_state_many() {
+  local email="$1" count="$2" index
+
+  jq -n '[]' >"$key_state"
+  for ((index = 1; index <= count; index++)); do
+    jq --arg name "$(key_name "$email" "orphan-$index")" \
+      '. + [{name: $name, disabled: false}]' \
+      "$key_state" >"$key_state.tmp"
+    mv "$key_state.tmp" "$key_state"
+  done
+  reset_operation_counts
 }
 
 write_credentials() {
@@ -369,6 +501,16 @@ if bash "$repo_root/rootfs/home/cloud-compose/rotate-keys.sh" status \
   app@example.invalid test-project "$invalid_state_credentials" >/dev/null 2>&1; then
   fail "rotation normalized a trailing newline in a baseline key name"
 fi
+jq --arg name "$(key_name app@example.invalid old-key)" \
+  '.baseline_key_names = [$name, $name]' \
+  "${invalid_state_credentials}.rotation-pending.json" \
+  >"${invalid_state_credentials}.rotation-pending.json.tmp"
+mv "${invalid_state_credentials}.rotation-pending.json.tmp" \
+  "${invalid_state_credentials}.rotation-pending.json"
+if bash "$repo_root/rootfs/home/cloud-compose/rotate-keys.sh" status \
+  app@example.invalid test-project "$invalid_state_credentials" >/dev/null 2>&1; then
+  fail "rotation accepted duplicate names in a durable reconciliation baseline"
+fi
 write_credentials "$APP_CREDENTIALS_FILE" old-key
 
 # Retirement deletes the currently installed remote key before removing every
@@ -431,6 +573,11 @@ grep -Fq "CHOWN -- cloud-compose $tmp/apps/alpha/secrets" "$order_log" || \
 if grep -Eq 'RESTART |DISABLE |DELETE ' "$order_log"; then
   fail "distribution failure restarted consumers or changed the previous key"
 fi
+creating_sync_line="$(grep -n '^SYNC 1$' "$order_log" | cut -d: -f1)"
+creating_post_line="$(grep -n '^POST$' "$order_log" | cut -d: -f1)"
+[[ -n "$creating_sync_line" && -n "$creating_post_line" &&
+  "$creating_sync_line" -lt "$creating_post_line" ]] ||
+  fail "ordinary key creation outran its durable creating state"
 
 # A failed reload cannot advance readiness or disable the previous key.
 if FAKE_RESTART_FAIL=true FAKE_APP_ACTIVE=true bash "$repo_root/rootfs/home/cloud-compose/rotate-keys-app.sh" >/dev/null 2>&1; then
@@ -611,6 +758,24 @@ FAKE_DELETE_FAIL=false bash "$repo_root/rootfs/home/cloud-compose/rotate-keys.sh
 [[ ! -e "${failure_retry}.rotation-pending.json" ]] || fail "delete retry did not finish rotation"
 [[ "$(<"$post_count")" == 1 ]] || fail "disable/delete retries created another replacement key"
 
+# A failed state durability barrier stops before the create request. The
+# renamed state may be visible in the live filesystem, but IAM remains
+# untouched until a later run can durably flush it.
+durability="$tmp/durability/GOOGLE_APPLICATION_CREDENTIALS"
+central_cleanup
+reset_key_state app@example.invalid old-key
+write_credentials "$durability" old-key
+export NEW_KEY_ID=durability-key
+if FAKE_SYNC_FAIL_AT=1 bash "$repo_root/rootfs/home/cloud-compose/rotate-keys.sh" prepare \
+  app@example.invalid test-project "$durability" >/dev/null 2>&1; then
+  fail "rotation accepted a failed pre-create state durability barrier"
+fi
+[[ "$(<"$post_count")" == 0 ]] ||
+  fail "failed state durability barrier issued a create request"
+if grep -Eq '^(DELETE |DISABLE |ENABLE )' "$order_log"; then
+  fail "failed state durability barrier mutated IAM"
+fi
+
 # First provisioning has no rollback key and therefore never disables/deletes
 # another service-account key.
 first="$tmp/first/GOOGLE_APPLICATION_CREDENTIALS"
@@ -625,6 +790,308 @@ if grep -Eq '^(DISABLE|DELETE) ' "$order_log"; then
   fail "first-time provisioning changed a key without a previous local credential"
 fi
 
+# A trusted marker on a newly formatted data filesystem authorizes the
+# module-owned app identity to delete its exact orphan baseline before creating
+# one replacement. Rotation itself leaves consumption to the bootstrap
+# boundary immediately after every enabled identity has converged.
+fresh_data_root="$tmp/fresh-data"
+fresh_marker="$fresh_data_root/.cloud-compose/fresh-filesystem"
+fresh_identity="v1:gcp-disk-id:987654321012345678"
+mkdir -p "$(dirname -- "$fresh_marker")"
+chmod 1775 "$fresh_data_root"
+chmod 0700 "$(dirname -- "$fresh_marker")"
+printf '%s\n' "$fresh_identity" >"$fresh_marker"
+chmod 0600 "$fresh_marker"
+export FAKE_FRESH_MARKER="$fresh_marker"
+export CLOUD_COMPOSE_FRESH_FILESYSTEM_MARKER="$fresh_marker"
+export CLOUD_COMPOSE_FRESH_FILESYSTEM_IDENTITY="$fresh_identity"
+
+# A root-owned marker copied from a snapshot of another data-disk incarnation
+# is not destructive authority. Reject it before even listing IAM keys.
+printf 'v1:gcp-disk-id:111111111111111111\n' >"$fresh_marker"
+central_cleanup
+reset_key_state_many app@example.invalid 2
+if GCP_APP_SERVICE_ACCOUNT_MANAGED=true FAKE_APP_STATE=inactive FAKE_APP_ACTIVE=false \
+  bash "$repo_root/rootfs/home/cloud-compose/rotate-keys-app.sh" >/dev/null 2>&1; then
+  fail "fresh reconciliation accepted another disk incarnation"
+fi
+if grep -Eq '^(DELETE |POST$)' "$order_log" ||
+  [[ "$(<"$list_count")" != 0 ]]; then
+  fail "mismatched disk-incarnation authority reached IAM"
+fi
+printf '%s\n' "$fresh_identity" >"$fresh_marker"
+
+central_cleanup
+reset_key_state_many app@example.invalid 1
+export NEW_KEY_ID=fresh-durability-key EXPECTED_CONSUMER_KEY_ID=fresh-durability-key
+if GCP_APP_SERVICE_ACCOUNT_MANAGED=true FAKE_APP_STATE=inactive FAKE_APP_ACTIVE=false \
+  FAKE_SYNC_FAIL_AT=1 \
+  bash "$repo_root/rootfs/home/cloud-compose/rotate-keys-app.sh" >/dev/null 2>&1; then
+  fail "fresh reconciliation accepted a failed baseline durability barrier"
+fi
+if grep -Eq '^(DELETE |POST$)' "$order_log"; then
+  fail "fresh reconciliation mutated IAM before its baseline became durable"
+fi
+
+central_cleanup
+reset_key_state_many app@example.invalid 10
+export NEW_KEY_ID=fresh-key EXPECTED_CONSUMER_KEY_ID=fresh-key
+GCP_APP_SERVICE_ACCOUNT_MANAGED=true FAKE_APP_STATE=inactive FAKE_APP_ACTIVE=false \
+  bash "$repo_root/rootfs/home/cloud-compose/rotate-keys-app.sh" >/dev/null
+[[ "$(<"$post_count")" == 1 ]] || fail "fresh managed reconciliation did not create exactly one key"
+[[ "$(grep -c '^DELETE ' "$order_log")" == 10 ]] || fail "fresh managed reconciliation did not delete its exact ten-key baseline"
+last_delete_line="$(grep -n '^DELETE ' "$order_log" | tail -n1 | cut -d: -f1)"
+post_line="$(grep -n '^POST$' "$order_log" | cut -d: -f1)"
+((last_delete_line < post_line)) || fail "fresh managed reconciliation created a key before deleting its baseline"
+first_delete_line="$(grep -n '^DELETE ' "$order_log" | head -n1 | cut -d: -f1)"
+first_sync_line="$(grep -n '^SYNC 1$' "$order_log" | cut -d: -f1)"
+second_sync_line="$(grep -n '^SYNC 2$' "$order_log" | cut -d: -f1)"
+[[ -n "$first_sync_line" && -n "$second_sync_line" &&
+  "$first_sync_line" -lt "$first_delete_line" &&
+  "$last_delete_line" -lt "$second_sync_line" &&
+  "$second_sync_line" -lt "$post_line" ]] ||
+  fail "fresh reconciliation IAM mutations outran their durable state transitions"
+jq -e --arg name "$(key_name app@example.invalid fresh-key)" \
+  'length == 1 and .[0].name == $name' "$key_state" >/dev/null ||
+  fail "fresh managed reconciliation did not converge to one replacement key"
+[[ -f "$fresh_marker" ]] || fail "key rotation consumed the shared fresh-filesystem marker inside one identity wrapper"
+
+# A delete that reaches IAM but loses its successful response retains the exact
+# baseline. The retry accepts 404 and cannot create until every baseline key is
+# absent.
+central_cleanup
+reset_key_state_many app@example.invalid 2
+export NEW_KEY_ID=fresh-retry-key EXPECTED_CONSUMER_KEY_ID=fresh-retry-key
+if GCP_APP_SERVICE_ACCOUNT_MANAGED=true FAKE_APP_STATE=inactive FAKE_APP_ACTIVE=false \
+  FAKE_DELETE_FAIL_AFTER_MUTATION_ID=orphan-1 \
+  bash "$repo_root/rootfs/home/cloud-compose/rotate-keys-app.sh" >/dev/null 2>&1; then
+  fail "fresh managed reconciliation accepted an indeterminate orphan deletion"
+fi
+jq -e '.phase == "reconciling" and (.baseline_key_names | length) == 2' \
+  "${APP_CREDENTIALS_FILE}.rotation-pending.json" >/dev/null ||
+  fail "failed orphan deletion did not retain the exact reconciliation baseline"
+[[ "$(<"$post_count")" == 0 ]] || fail "failed orphan deletion created a replacement key"
+GCP_APP_SERVICE_ACCOUNT_MANAGED=true FAKE_APP_STATE=inactive FAKE_APP_ACTIVE=false \
+  FAKE_DELETE_FAIL_AFTER_MUTATION_ID=orphan-1 \
+  bash "$repo_root/rootfs/home/cloud-compose/rotate-keys-app.sh" >/dev/null
+[[ "$(<"$post_count")" == 1 ]] || fail "orphan deletion retry did not create exactly one replacement"
+grep -Fq "DELETE404 $(key_name app@example.invalid orphan-1)" "$order_log" ||
+  fail "orphan deletion retry did not accept the already-absent key"
+
+# A key created by another actor after the durable baseline snapshot is never
+# swept into the managed set, and permanently blocks creation until audited.
+central_cleanup
+reset_key_state_many app@example.invalid 1
+export NEW_KEY_ID=blocked-fresh-key EXPECTED_CONSUMER_KEY_ID=blocked-fresh-key
+if GCP_APP_SERVICE_ACCOUNT_MANAGED=true FAKE_APP_STATE=inactive FAKE_APP_ACTIVE=false \
+  FAKE_INJECT_KEY_ON_LIST=concurrent-key \
+  bash "$repo_root/rootfs/home/cloud-compose/rotate-keys-app.sh" >/dev/null 2>&1; then
+  fail "fresh managed reconciliation accepted a concurrent key"
+fi
+[[ "$(<"$post_count")" == 0 ]] || fail "concurrent key detection issued a create"
+if GCP_APP_SERVICE_ACCOUNT_MANAGED=true FAKE_APP_STATE=inactive FAKE_APP_ACTIVE=false \
+  FAKE_INJECT_KEY_ON_LIST=concurrent-key \
+  bash "$repo_root/rootfs/home/cloud-compose/rotate-keys-app.sh" >/dev/null 2>&1; then
+  fail "fresh managed reconciliation swept a concurrent key on retry"
+fi
+jq -e --arg name "$(key_name app@example.invalid concurrent-key)" \
+  'length == 1 and .[0].name == $name' "$key_state" >/dev/null ||
+  fail "fresh managed reconciliation deleted the concurrent key"
+[[ "$(<"$post_count")" == 0 ]] || fail "concurrent key retry issued a create"
+
+# Reconciliation is unavailable to supplied identities and to any service that
+# is not exactly loaded and inactive.
+central_cleanup
+reset_key_state_many app@example.invalid 1
+export NEW_KEY_ID=unmanaged-key EXPECTED_CONSUMER_KEY_ID=unmanaged-key
+GCP_APP_SERVICE_ACCOUNT_MANAGED=false FAKE_APP_ACTIVE=false \
+  bash "$repo_root/rootfs/home/cloud-compose/rotate-keys-app.sh" >/dev/null
+if grep -Eq '^DELETE ' "$order_log"; then
+  fail "caller-supplied app identity deleted a baseline key"
+fi
+jq -e 'length == 2' "$key_state" >/dev/null ||
+  fail "caller-supplied app identity did not preserve its remote baseline"
+
+# Both independent proofs are required. A managed identity without the
+# root-owned fresh-filesystem marker follows ordinary provisioning and never
+# treats pre-existing keys as its deletion set.
+rm -f -- "$fresh_marker"
+central_cleanup
+reset_key_state_many app@example.invalid 1
+export NEW_KEY_ID=managed-without-marker EXPECTED_CONSUMER_KEY_ID=managed-without-marker
+GCP_APP_SERVICE_ACCOUNT_MANAGED=true FAKE_APP_STATE=inactive FAKE_APP_ACTIVE=false \
+  bash "$repo_root/rootfs/home/cloud-compose/rotate-keys-app.sh" >/dev/null
+if grep -Eq '^DELETE ' "$order_log"; then
+  fail "managed app identity deleted a baseline without fresh-filesystem authority"
+fi
+jq -e 'length == 2' "$key_state" >/dev/null ||
+  fail "managed app identity without a fresh marker did not preserve its remote baseline"
+
+# Ordinary rotation cannot exceed IAM's ten-key quota and must fail before
+# persisting an ambiguous create state.
+central_cleanup
+reset_key_state_many app@example.invalid 10
+if GCP_APP_SERVICE_ACCOUNT_MANAGED=true FAKE_APP_STATE=inactive FAKE_APP_ACTIVE=false \
+  bash "$repo_root/rootfs/home/cloud-compose/rotate-keys-app.sh" >/dev/null 2>&1; then
+  fail "ordinary provisioning accepted an already-full IAM key quota"
+fi
+[[ "$(<"$post_count")" == 0 ]] || fail "full ordinary key quota issued a create"
+[[ ! -e "${APP_CREDENTIALS_FILE}.rotation-pending.json" ]] ||
+  fail "full ordinary key quota persisted an ambiguous create state"
+if grep -Eq '^DELETE ' "$order_log"; then
+  fail "full ordinary key quota deleted an existing key"
+fi
+
+# Invalid identity ownership configuration fails before any IAM request.
+central_cleanup
+reset_key_state_many app@example.invalid 1
+if GCP_APP_SERVICE_ACCOUNT_MANAGED=unexpected \
+  bash "$repo_root/rootfs/home/cloud-compose/rotate-keys-app.sh" >/dev/null 2>&1; then
+  fail "app wrapper accepted invalid managed-identity configuration"
+fi
+[[ "$(<"$post_count")" == 0 && "$(<"$list_count")" == 0 ]] ||
+  fail "invalid managed-identity configuration reached IAM"
+
+printf '%s\n' "$fresh_identity" >"$fresh_marker"
+chmod 0600 "$fresh_marker"
+central_cleanup
+reset_key_state_many app@example.invalid 1
+if GCP_APP_SERVICE_ACCOUNT_MANAGED=true FAKE_APP_STATE=failed \
+  bash "$repo_root/rootfs/home/cloud-compose/rotate-keys-app.sh" >/dev/null 2>&1; then
+  fail "fresh managed reconciliation accepted a failed application service"
+fi
+[[ "$(<"$post_count")" == 0 ]] || fail "failed application service allowed key creation"
+if grep -Eq '^DELETE ' "$order_log"; then
+  fail "failed application service allowed orphan deletion"
+fi
+
+# A malformed, duplicate, or over-limit IAM inventory cannot be converted into
+# a deletion baseline.
+central_cleanup
+jq -n --arg name "$(key_name app@example.invalid duplicate-key)" \
+  '[{name: $name, disabled: false}, {name: $name, disabled: false}]' >"$key_state"
+reset_operation_counts
+if GCP_APP_SERVICE_ACCOUNT_MANAGED=true FAKE_APP_STATE=inactive FAKE_APP_ACTIVE=false \
+  bash "$repo_root/rootfs/home/cloud-compose/rotate-keys-app.sh" >/dev/null 2>&1; then
+  fail "fresh managed reconciliation accepted duplicate IAM key entries"
+fi
+if grep -Eq '^(DELETE |POST$)' "$order_log"; then
+  fail "duplicate IAM key inventory triggered a mutation"
+fi
+
+central_cleanup
+jq -n '[{
+  keyType: "USER_MANAGED",
+  name: "projects/other-project/serviceAccounts/other@example.invalid/keys/foreign-key",
+  disabled: false
+}]' >"$key_state"
+reset_operation_counts
+if GCP_APP_SERVICE_ACCOUNT_MANAGED=true FAKE_APP_STATE=inactive FAKE_APP_ACTIVE=false \
+  bash "$repo_root/rootfs/home/cloud-compose/rotate-keys-app.sh" >/dev/null 2>&1; then
+  fail "fresh managed reconciliation accepted a foreign IAM key name"
+fi
+if grep -Eq '^(DELETE |POST$)' "$order_log"; then
+  fail "foreign IAM key inventory triggered a mutation"
+fi
+
+central_cleanup
+reset_key_state_many app@example.invalid 11
+if GCP_APP_SERVICE_ACCOUNT_MANAGED=true FAKE_APP_STATE=inactive FAKE_APP_ACTIVE=false \
+  bash "$repo_root/rootfs/home/cloud-compose/rotate-keys-app.sh" >/dev/null 2>&1; then
+  fail "fresh managed reconciliation accepted more than IAM's key limit"
+fi
+if grep -Eq '^(DELETE |POST$)' "$order_log"; then
+  fail "over-limit IAM key inventory triggered a mutation"
+fi
+
+# A missing central file is recovered only when every existing distributed copy
+# is valid for the exact target and byte-identical.
+central_cleanup
+reset_key_state app@example.invalid supplied-victim-key
+write_credentials "$tmp/apps/alpha/secrets/GOOGLE_APPLICATION_CREDENTIALS" supplied-victim-key
+write_credentials "$tmp/apps/beta/secrets/GOOGLE_APPLICATION_CREDENTIALS" supplied-victim-key
+if GCP_APP_SERVICE_ACCOUNT_MANAGED=false \
+  bash "$repo_root/rootfs/home/cloud-compose/rotate-keys-app.sh" >/dev/null 2>&1; then
+  fail "caller-supplied identity promoted distributed credentials into revocation authority"
+fi
+[[ ! -e "$APP_CREDENTIALS_FILE" ]] ||
+  fail "caller-supplied identity restored an unproven central credential"
+[[ "$(<"$post_count")" == 0 && "$(<"$list_count")" == 0 ]] ||
+  fail "caller-supplied distributed credential recovery reached IAM"
+if grep -Eq '^(DELETE |POST$)' "$order_log"; then
+  fail "caller-supplied distributed credential recovery mutated IAM"
+fi
+
+central_cleanup
+reset_key_state app@example.invalid recovered-key
+write_credentials "$tmp/apps/alpha/secrets/GOOGLE_APPLICATION_CREDENTIALS" recovered-key
+write_credentials "$tmp/apps/beta/secrets/GOOGLE_APPLICATION_CREDENTIALS" recovered-key
+touch -d '10 minutes ago' \
+  "$tmp/apps/alpha/secrets/GOOGLE_APPLICATION_CREDENTIALS" \
+  "$tmp/apps/beta/secrets/GOOGLE_APPLICATION_CREDENTIALS"
+recovered_modified_at="$(stat -c %Y "$tmp/apps/alpha/secrets/GOOGLE_APPLICATION_CREDENTIALS")"
+GCP_APP_SERVICE_ACCOUNT_MANAGED=true ROTATION_MIN_AGE_SECONDS=86400 \
+  bash "$repo_root/rootfs/home/cloud-compose/rotate-keys-app.sh" >/dev/null
+cmp -s "$APP_CREDENTIALS_FILE" "$tmp/apps/alpha/secrets/GOOGLE_APPLICATION_CREDENTIALS" ||
+  fail "unanimous distributed credentials were not restored centrally"
+[[ "$(stat -c %Y "$APP_CREDENTIALS_FILE")" == "$recovered_modified_at" ]] ||
+  fail "distributed credential recovery reset the credential rotation age"
+[[ "$(<"$post_count")" == 0 ]] || fail "distributed credential recovery created a cloud key"
+if grep -Eq '^DELETE ' "$order_log"; then
+  fail "distributed credential recovery deleted a cloud key"
+fi
+
+central_cleanup
+reset_key_state app@example.invalid recovered-key
+write_credentials "$tmp/apps/alpha/secrets/GOOGLE_APPLICATION_CREDENTIALS" recovered-key
+write_credentials "$tmp/apps/beta/secrets/GOOGLE_APPLICATION_CREDENTIALS" conflicting-key
+if GCP_APP_SERVICE_ACCOUNT_MANAGED=true ROTATION_MIN_AGE_SECONDS=86400 \
+  bash "$repo_root/rootfs/home/cloud-compose/rotate-keys-app.sh" >/dev/null 2>&1; then
+  fail "conflicting distributed credentials were accepted"
+fi
+[[ ! -e "$APP_CREDENTIALS_FILE" ]] || fail "conflicting distributed credentials restored a central key"
+[[ "$(<"$post_count")" == 0 ]] || fail "conflicting distributed credentials created a cloud key"
+
+# Definite IAM rejection does not strand an ambiguous state. Request timeout
+# remains ambiguous and a complete response is salvageable after a late
+# transport failure.
+definite="$tmp/definite/GOOGLE_APPLICATION_CREDENTIALS"
+reset_key_state app@example.invalid old-key
+write_credentials "$definite" old-key
+export NEW_KEY_ID=definite-key
+if FAKE_POST_MODE=rejected-400 bash "$repo_root/rootfs/home/cloud-compose/rotate-keys.sh" prepare \
+  app@example.invalid test-project "$definite" >/dev/null 2>&1; then
+  fail "definite IAM key rejection was accepted"
+fi
+[[ ! -e "${definite}.rotation-pending.json" ]] ||
+  fail "definite IAM key rejection retained ambiguous state"
+
+timeout="$tmp/timeout/GOOGLE_APPLICATION_CREDENTIALS"
+reset_key_state app@example.invalid old-key
+write_credentials "$timeout" old-key
+export NEW_KEY_ID=timeout-key
+if FAKE_POST_MODE=ambiguous-408 bash "$repo_root/rootfs/home/cloud-compose/rotate-keys.sh" prepare \
+  app@example.invalid test-project "$timeout" >/dev/null 2>&1; then
+  fail "ambiguous HTTP 408 key creation was accepted"
+fi
+if FAKE_POST_MODE=success bash "$repo_root/rootfs/home/cloud-compose/rotate-keys.sh" prepare \
+  app@example.invalid test-project "$timeout" >/dev/null 2>&1; then
+  fail "ambiguous HTTP 408 key creation was retried"
+fi
+[[ "$(<"$post_count")" == 1 ]] || fail "ambiguous HTTP 408 issued another POST"
+jq -e '.phase == "creating"' "${timeout}.rotation-pending.json" >/dev/null ||
+  fail "ambiguous HTTP 408 did not retain creating state"
+
+late="$tmp/late/GOOGLE_APPLICATION_CREDENTIALS"
+reset_key_state app@example.invalid old-key
+write_credentials "$late" old-key
+export NEW_KEY_ID=late-key
+FAKE_POST_MODE=late-success bash "$repo_root/rootfs/home/cloud-compose/rotate-keys.sh" prepare \
+  app@example.invalid test-project "$late" >/dev/null
+jq -e '.phase == "staged" and .new_key_id == "late-key"' \
+  "${late}.rotation-pending.json" >/dev/null ||
+  fail "complete IAM response was not salvaged after a late transport failure"
+
 # The internal-service wrapper follows the same no-start rule.
 internal_email="internal-test-instance@test-project.iam.gserviceaccount.com"
 export INTERNAL_CREDENTIALS_FILE="$tmp/internal/GOOGLE_APPLICATION_CREDENTIALS"
@@ -639,4 +1106,52 @@ fi
 jq -e '.phase == "grace"' "${INTERNAL_CREDENTIALS_FILE}.rotation-pending.json" >/dev/null || \
   fail "inactive internal service did not complete authenticated propagation"
 
+# The internal identity is always module-created and uses the same trusted
+# fresh-filesystem reconciliation before first provisioning.
+rm -f -- "$INTERNAL_CREDENTIALS_FILE" "${INTERNAL_CREDENTIALS_FILE}.rotation-"*
+reset_key_state_many "$internal_email" 10
+export NEW_KEY_ID=internal-fresh EXPECTED_CONSUMER_KEY_ID=internal-fresh
+FAKE_INTERNAL_STATE=inactive FAKE_INTERNAL_ACTIVE=false \
+  bash "$repo_root/rootfs/home/cloud-compose/rotate-keys-internal.sh" >/dev/null
+[[ "$(<"$post_count")" == 1 ]] || fail "fresh internal reconciliation did not create exactly one key"
+jq -e --arg name "$(key_name "$internal_email" internal-fresh)" \
+  'length == 1 and .[0].name == $name' "$key_state" >/dev/null ||
+  fail "fresh internal reconciliation did not converge to one replacement key"
+
+# The internal identity uses the same exact inactive-service gate. Failed or
+# unqueryable unit state cannot authorize deletion.
+rm -f -- "$INTERNAL_CREDENTIALS_FILE" "${INTERNAL_CREDENTIALS_FILE}.rotation-"*
+reset_key_state_many "$internal_email" 1
+if FAKE_INTERNAL_STATE=failed \
+  bash "$repo_root/rootfs/home/cloud-compose/rotate-keys-internal.sh" >/dev/null 2>&1; then
+  fail "fresh internal reconciliation accepted a failed service"
+fi
+if grep -Eq '^(DELETE |POST$)' "$order_log"; then
+  fail "failed internal service allowed IAM mutation"
+fi
+
+rm -f -- "$INTERNAL_CREDENTIALS_FILE" "${INTERNAL_CREDENTIALS_FILE}.rotation-"*
+reset_key_state_many "$internal_email" 1
+if FAKE_SYSTEMCTL_QUERY_FAIL=true \
+  bash "$repo_root/rootfs/home/cloud-compose/rotate-keys-internal.sh" >/dev/null 2>&1; then
+  fail "fresh internal reconciliation accepted an unqueryable service state"
+fi
+if grep -Eq '^(DELETE |POST$)' "$order_log"; then
+  fail "unqueryable internal service allowed IAM mutation"
+fi
+
+if find "$ROTATION_RUNTIME_DIR" -maxdepth 1 -type f -name 'iam-create-response.*' -print -quit |
+  grep -q .; then
+  fail "rotation retained an ephemeral IAM private-key response"
+fi
+if find "$tmp" -type f -name '.iam-create-response.*' -print -quit | grep -q .; then
+  fail "rotation wrote an IAM private-key response to persistent credential storage"
+fi
+grep -Fq 'mktemp "${ROTATION_RUNTIME_DIR}/iam-create-response.XXXXXX"' \
+  "$repo_root/rootfs/home/cloud-compose/rotate-keys.sh" ||
+  fail "IAM private-key responses are not staged under the ephemeral runtime directory"
+grep -Fq 'trap cleanup_ephemeral_create_response EXIT' \
+  "$repo_root/rootfs/home/cloud-compose/rotate-keys.sh" ||
+  fail "IAM private-key response cleanup is not registered for process exit"
+
 echo "Key rotation contract passed"
diff --git a/docs/runtime-contracts.md b/docs/runtime-contracts.md
index d711f22..6dc9348 100644
--- a/docs/runtime-contracts.md
+++ b/docs/runtime-contracts.md
@@ -555,6 +555,59 @@ previous app credential with:
 sudo /home/cloud-compose/rotate-keys-app.sh rollback
 ```
 
+If the central credential for a module-created app identity is missing but
+distributed app copies remain, bootstrap first restores it only when every
+existing copy is a valid credential for the exact project and service account
+and all copies are byte-identical. The restored file retains the source
+modification time so recovery cannot reset the rotation interval. An invalid
+or conflicting copy fails before any IAM mutation. Distributed copies for a
+caller-supplied identity are not sufficient revocation authority and require
+operator-reviewed recovery.
+
+Formatting a new GCP data disk publishes a root-owned, single-use
+fresh-filesystem marker whose exact payload is
+`v1:gcp-disk-id:<google_compute_disk.data.disk_id>`. Terraform renders the same
+server-assigned disk incarnation into cloud-init and the root runtime
+environment. A marker restored from a snapshot onto a new disk therefore
+cannot authorize deletion of that new host's current IAM keys. GCP never
+accepts the generic marker used by non-GCP hosts.
+
+A supported GCP data-disk snapshot restore must run through Terraform and VM
+replacement so the replacement boot configuration receives the new disk's
+`disk_id`. Out-of-band attachment or hot-swapping a restored data disk while
+retaining an older boot configuration is unsupported and does not satisfy the
+fresh-authority invariant; converge the VM through Terraform before bootstrap.
+
+A reserved pending ext4 label carries fresh-format intent across a crash
+between formatting, mounting, and marker publication. The label can recreate a
+missing marker only while the mounted ext4 root is exactly root-owned, mode
+`0755`, and empty except for an empty, root-owned `lost+found`; a populated disk
+fails closed. An existing marker must have the exact expected disk payload,
+root ownership, private modes, and single link before the label can be cleared.
+Global durability barriers order marker publication and label clearing.
+
+While the marker exists, a module-created app identity or the
+always-module-created internal identity may reconcile leaked keys from an
+interrupted first bootstrap, but only while its exact consumer unit is loaded
+and inactive. Reconciliation durably flushes the complete validated
+user-managed-key baseline before deletion, retries deletion idempotently, and
+creates one replacement only after the empty-baseline state is also durable. A
+key that appears after the snapshot is never adopted into the deletion set and
+blocks progress for review. Host convergence keeps the data mount root owned by
+`root:cloud-compose` with mode `1775`, leaving the marker directory
+root-private while application paths remain group-creatable. Bootstrap
+consumes and flushes the marker immediately after the one-shot daily rotation
+converges every enabled identity. Consumption precedes the persistent rotation
+timer, Vault, application initialization, and application startup, closing the
+window in which a scheduled snapshot could retain destructive authority.
+
+Caller-supplied app identities never receive that destructive recovery path.
+An already-formatted disk has no fresh-filesystem attestation either: audit and
+revoke its exact orphan keys through an operator-reviewed procedure instead of
+creating the marker retroactively. Outside fresh reconciliation, reaching
+IAM's ten-user-key limit fails before a create request or pending state is
+written.
+
 To disable an existing app file credential, first leave
 `app_credentials_enabled = true`, apply the 1.0 runtime, and retire the remote
 key plus every distributed local copy:
@@ -573,13 +626,15 @@ user-managed key IDs and fails closed without deleting any of them. Audit and
 explicitly revoke those IDs before disabling Key Admin. A missing local file is
 reported as complete only when the service account has no user-managed key.
 
-An ambiguous key-creation response never triggers an automatic second create.
-`rotate-keys.sh audit ...` reports only the baseline delta/key IDs. Recovery
-requires the single audited orphan key ID as explicit confirmation to
-`rotate-keys.sh recover ... KEY_ID`; if no key was created, recovery clears the
-pending state without deleting anything. Recovery also waits
-`ROTATION_RECOVERY_SETTLE_SECONDS` (60 seconds by default) before trusting an
-empty IAM delta, avoiding a false “no key created” result during propagation.
+A definite IAM rejection that proves no key was created clears the pre-create
+state. A timeout, transport failure, or otherwise ambiguous key-creation
+response never triggers an automatic second create. `rotate-keys.sh audit ...`
+reports only the baseline delta/key IDs. Recovery requires the single audited
+orphan key ID as explicit confirmation to `rotate-keys.sh recover ... KEY_ID`;
+if no key was created, recovery clears the pending state without deleting
+anything. Recovery also waits `ROTATION_RECOVERY_SETTLE_SECONDS` (60 seconds by
+default) before trusting an empty IAM delta, avoiding a false “no key created”
+result during propagation.
 
 All Terraform entrypoints, including GCP, DigitalOcean, and Linode, support the
 same verified rootfs archive contract. When `runtime.rootfs_archive_url` is
diff --git a/modules/gcp/main.tf b/modules/gcp/main.tf
index ebf901f..9c60e06 100644
--- a/modules/gcp/main.tf
+++ b/modules/gcp/main.tf
@@ -358,6 +358,9 @@ rollout_env = var.rollout_enabled ? {
   } : {
   ROLLOUT_ENABLED = "false"
 }
+fresh_filesystem_env = {
+  CLOUD_COMPOSE_FRESH_FILESYSTEM_IDENTITY = "v1:gcp-disk-id:${google_compute_disk.data.disk_id}"
+}
 host_env = merge({
   HOME                                 = "/home/cloud-compose"
   GCP_PROJECT                          = var.project_id
@@ -387,6 +390,7 @@ host_env = merge({
   PRODUCTION                           = tostring(var.production)
   SITECTL_VERIFY_ARGS                  = join(" ", local.primary_compose_project.sitectl_verify_args)
   GCP_APP_SERVICE_ACCOUNT_EMAIL        = local.app_service_account_email
+  GCP_APP_SERVICE_ACCOUNT_MANAGED      = tostring(local.app_service_account_managed)
   GCP_APP_CREDENTIALS_ENABLED          = tostring(local.app_credentials_enabled)
   POWER_MANAGEMENT_ENABLED             = tostring(var.power_management_enabled)
   COMPOSE_PROFILES                     = local.internal_services_compose_profiles
@@ -399,7 +403,7 @@ host_env = merge({
   LIBOPS_MANAGED_RUNTIME_ENABLED       = tostring(var.libops_managed_runtime_enabled)
   LIBOPS_INTERNAL_SERVICES_ENABLED     = tostring(local.internal_services_enabled)
   LIBOPS_INTERNAL_SERVICES_AUTO_UPDATE = tostring(local.internal_services_enabled && var.libops_internal_services_auto_update)
-}, local.rollout_env)
+}, local.fresh_filesystem_env, local.rollout_env)
 env_file_content             = <<-EOT
     - path: "/home/cloud-compose/.env"
       permissions: "0640"
@@ -460,6 +464,7 @@ rollout_runcmd = var.rollout_enabled ? [
 cloud_init_yaml = templatefile("${path.module}/../../templates/cloud-init.yml", {
   FILESYSTEM_PREP_SCRIPT_B64     = filebase64("${local.rootFs}/home/cloud-compose/prepare-filesystem.sh"),
   FILESYSTEM_PERSIST_SCRIPT_B64  = filebase64("${local.rootFs}/home/cloud-compose/persist-filesystems.sh"),
+  FRESH_FILESYSTEM_IDENTITY      = "v1:gcp-disk-id:${google_compute_disk.data.disk_id}",
   WRITE_FILES_CONTENT            = local.write_files_content,
   DOCKER_COMPOSE_SCRIPTS         = local.docker_compose_scripts,
   COMPOSE_PROJECTS_FILE          = local.compose_projects_file,
@@ -480,9 +485,10 @@ vm_service_account_email = var.service_account_email != "" ? data.google_service
 vm_service_account_id    = var.service_account_email != "" ? data.google_service_account.vm[0].name : google_service_account.cloud-compose[0].id
 vm_service_account_name  = var.service_account_email != "" ? data.google_service_account.vm[0].name : google_service_account.cloud-compose[0].name
 
-app_service_account_email = var.app_service_account_email != "" ? data.google_service_account.app[0].email : google_service_account.app[0].email
-app_service_account_id    = var.app_service_account_email != "" ? data.google_service_account.app[0].name : google_service_account.app[0].id
-app_service_account_name  = var.app_service_account_email != "" ? data.google_service_account.app[0].name : google_service_account.app[0].name
+app_service_account_email   = var.app_service_account_email != "" ? data.google_service_account.app[0].email : google_service_account.app[0].email
+app_service_account_id      = var.app_service_account_email != "" ? data.google_service_account.app[0].name : google_service_account.app[0].id
+app_service_account_name    = var.app_service_account_email != "" ? data.google_service_account.app[0].name : google_service_account.app[0].name
+app_service_account_managed = var.app_service_account_email == ""
 
 app_credentials_enabled            = var.app_credentials_enabled
 internal_services_enabled          = var.libops_internal_services_enabled || var.power_management_enabled
diff --git a/modules/gcp/outputs.tf b/modules/gcp/outputs.tf
index d2ba63d..3a24703 100644
--- a/modules/gcp/outputs.tf
+++ b/modules/gcp/outputs.tf
@@ -100,9 +100,10 @@ output "appGsa" {
     email : local.app_service_account_email,
     id : local.app_service_account_id,
     name : local.app_service_account_name,
+    managed : local.app_service_account_managed,
     credentials_enabled : local.app_credentials_enabled,
   }
-  description = "The Google Service Account the app can leverage to authenticate to other Google services, including whether managed file credentials are enabled."
+  description = "The Google Service Account the app can leverage to authenticate to other Google services, including whether the module owns it and whether managed file credentials are enabled."
 }
 
 output "urls" {
diff --git a/modules/gcp/runtime_contracts.tftest.hcl b/modules/gcp/runtime_contracts.tftest.hcl
index 3b0ecc0..ea52679 100644
--- a/modules/gcp/runtime_contracts.tftest.hcl
+++ b/modules/gcp/runtime_contracts.tftest.hcl
@@ -1,5 +1,11 @@
 mock_provider "cloudinit" {}
 mock_provider "google" {
+  mock_resource "google_compute_disk" {
+    override_during = plan
+    defaults = {
+      disk_id = "987654321012345678"
+    }
+  }
   mock_resource "google_service_account" {
     override_during = plan
     defaults = {
@@ -79,6 +85,17 @@ run "disables_privileged_services_by_default" {
     error_message = "The runtime project number must come from project_id discovery, not the deprecated caller assertion."
   }
 
+  assert {
+    condition = (
+      local.host_env.CLOUD_COMPOSE_FRESH_FILESYSTEM_IDENTITY == "v1:gcp-disk-id:987654321012345678" &&
+      strcontains(
+        local.cloud_init_yaml,
+        "--publish-fresh-marker \"v1:gcp-disk-id:987654321012345678\"",
+      )
+    )
+    error_message = "GCP cloud-init and the root runtime environment must carry the same immutable data-disk identity."
+  }
+
   assert {
     condition = (
       google_project_iam_member.log.member == "serviceAccount:${local.vm_service_account_email}" &&
@@ -138,9 +155,31 @@ run "creates_app_key_management_only_when_explicitly_enabled" {
       length(google_service_account_iam_member.app-keys) == 1 &&
       google_service_account_iam_member.app-keys[0].role == "roles/iam.serviceAccountKeyAdmin" &&
       google_service_account_iam_member.app-keys[0].member == "serviceAccount:${local.vm_service_account_email}" &&
+      local.host_env.GCP_APP_SERVICE_ACCOUNT_MANAGED == "true" &&
+      output.appGsa.managed == true &&
       local.host_env.GCP_APP_CREDENTIALS_ENABLED == "true"
     )
-    error_message = "Explicit app file credentials must scope Key Admin to the app identity and enable runtime rotation."
+    error_message = "Explicit app file credentials must scope Key Admin to the module-owned app identity and enable runtime rotation."
+  }
+}
+
+run "marks_caller_supplied_app_identity_unmanaged" {
+  command = plan
+
+  variables {
+    name                      = "gcp-contract"
+    project_id                = "test-project"
+    docker_compose_repo       = "https://github.com/libops/wp.git"
+    app_service_account_email = "existing@test-project.iam.gserviceaccount.com"
+    app_credentials_enabled   = true
+  }
+
+  assert {
+    condition = (
+      local.host_env.GCP_APP_SERVICE_ACCOUNT_MANAGED == "false" &&
+      output.appGsa.managed == false
+    )
+    error_message = "A caller-supplied app identity must never authorize managed orphan-key reconciliation."
   }
 }
 
diff --git a/modules/linux-vm-runtime/templates/cloud-init.yml b/modules/linux-vm-runtime/templates/cloud-init.yml
index 6557d7b..1b8d0ce 100644
--- a/modules/linux-vm-runtime/templates/cloud-init.yml
+++ b/modules/linux-vm-runtime/templates/cloud-init.yml
@@ -69,7 +69,7 @@ runcmd:
   printf '%s' '${FILESYSTEM_PERSIST_SCRIPT_B64}' | base64 -d >"$filesystem_persist"
   chmod 0600 "$filesystem_prep" "$filesystem_persist"
 %{ endif }
-  bash "$filesystem_prep" '${DATA_DEVICE}' /mnt/disks/data
+  bash "$filesystem_prep" '${DATA_DEVICE}' /mnt/disks/data --publish-fresh-marker
   bash "$filesystem_prep" '${VOLUMES_DEVICE}' /mnt/disks/volumes
   mkdir -p /mnt/disks/data/docker/volumes
   if ! mountpoint -q /mnt/disks/data/docker/volumes; then
@@ -90,8 +90,10 @@ runcmd:
     cp -a /var/lib/cloud-compose/mounted-rootfs/mnt/disks/. /mnt/disks/
   fi
 %{ endif }
-  chown cloud-compose:cloud-compose /mnt/disks/data /mnt/disks/volumes
-  chmod 0775 /mnt/disks/data /mnt/disks/volumes
+  chown root:cloud-compose /mnt/disks/data
+  chmod 1775 /mnt/disks/data
+  chown cloud-compose:cloud-compose /mnt/disks/volumes
+  chmod 0775 /mnt/disks/volumes
   install -d -m 0775 -o cloud-compose -g cloud-compose /mnt/disks/data/libops
   rm -f /home/cloud-compose/.cloud-compose-bootstrap-complete
   bash /home/cloud-compose/start-cloud-compose-bootstrap.sh
diff --git a/rootfs/home/cloud-compose/bootstrap-helpers.sh b/rootfs/home/cloud-compose/bootstrap-helpers.sh
index 35a0256..4a4c774 100644
--- a/rootfs/home/cloud-compose/bootstrap-helpers.sh
+++ b/rootfs/home/cloud-compose/bootstrap-helpers.sh
@@ -34,6 +34,55 @@ cloud_compose_publish_marker() (
     fi
 )
 
+cloud_compose_consume_fresh_filesystem_marker() {
+    local marker="$1"
+    local expected_identity="$2"
+    local marker_dir marker_identity marker_dir_identity marker_payload marker_size
+
+    case "$expected_identity" in
+        fresh) ;;
+        v1:gcp-disk-id:*)
+            if [[ ! "$expected_identity" =~ ^v1:gcp-disk-id:[0-9]{1,32}$ ]]; then
+                echo "Unsafe fresh-filesystem identity" >&2
+                return 2
+            fi
+            ;;
+        *)
+            echo "Unsafe fresh-filesystem identity" >&2
+            return 2
+            ;;
+    esac
+
+    if [[ "$marker" != /* || "$marker" == "/" || "$marker" == *$'\n'* ||
+        "$marker" == *$'\r'* || "$marker" =~ (^|/)\.\.?(/|$) ]]; then
+        echo "Unsafe fresh-filesystem marker path: $marker" >&2
+        return 1
+    fi
+    if [[ ! -e "$marker" && ! -L "$marker" ]]; then
+        return 0
+    fi
+    marker_dir="$(dirname -- "$marker")"
+    if [[ -L "$marker_dir" || ! -d "$marker_dir" || -L "$marker" || ! -f "$marker" ]]; then
+        echo "Unsafe fresh-filesystem marker: $marker" >&2
+        return 1
+    fi
+    marker_dir_identity="$(stat -c '%u:%g:%a' -- "$marker_dir")" || return 1
+    marker_identity="$(stat -c '%u:%g:%a:%h' -- "$marker")" || return 1
+    if [[ "$marker_dir_identity" != "0:0:700" || "$marker_identity" != "0:0:600:1" ]]; then
+        echo "Unsafe fresh-filesystem marker ownership or mode: $marker" >&2
+        return 1
+    fi
+    marker_size="$(stat -c '%s' -- "$marker")" || return 1
+    marker_payload=""
+    if [[ "$marker_size" != "$((${#expected_identity} + 1))" ]] ||
+        ! IFS= read -r marker_payload <"$marker" ||
+        [[ "$marker_payload" != "$expected_identity" ]]; then
+        echo "Fresh-filesystem marker does not match this disk incarnation" >&2
+        return 1
+    fi
+    rm -f -- "$marker"
+}
+
 cloud_compose_validate_systemd_unit() {
     case "$1" in
         cloud-compose.service | cloud-compose-bootstrap.service) return 0 ;;
diff --git a/rootfs/home/cloud-compose/host-init.sh b/rootfs/home/cloud-compose/host-init.sh
index 66b577d..0c00eb4 100644
--- a/rootfs/home/cloud-compose/host-init.sh
+++ b/rootfs/home/cloud-compose/host-init.sh
@@ -81,7 +81,8 @@ fi
 # Ansible and Salt adapters can install the runtime onto pre-mounted storage.
 # Normalize only the mount roots here so every entry path can create lifecycle
 # locks without granting recursive ownership of application or control data.
-for mutable_root in /mnt/disks/data /mnt/disks/volumes; do
-  install -d -m 0775 -o cloud-compose -g cloud-compose "$mutable_root"
-done
+# The sticky, root-owned data root protects root-only bootstrap attestations
+# while still allowing the cloud-compose group to create application paths.
+install -d -m 1775 -o root -g cloud-compose /mnt/disks/data
+install -d -m 0775 -o cloud-compose -g cloud-compose /mnt/disks/volumes
 install -d -m 0775 -o cloud-compose -g cloud-compose /mnt/disks/data/libops
diff --git a/rootfs/home/cloud-compose/prepare-filesystem.sh b/rootfs/home/cloud-compose/prepare-filesystem.sh
index 7b78362..1eef331 100644
--- a/rootfs/home/cloud-compose/prepare-filesystem.sh
+++ b/rootfs/home/cloud-compose/prepare-filesystem.sh
@@ -2,12 +2,115 @@
 
 set -euo pipefail
 
+fresh_filesystem_pending_label="cc-fresh-pending"
+
 log() {
     printf '[filesystem-prep] %s\n' "$*" >&2
 }
 
+publish_fresh_filesystem_marker() {
+    local mount_path="$1"
+    local expected_identity="$2"
+    local marker_dir="$mount_path/.cloud-compose"
+    local marker="$marker_dir/fresh-filesystem"
+    local marker_dir_identity marker_identity marker_payload marker_root_identity
+    local marker_size unexpected
+
+    case "$expected_identity" in
+        fresh) ;;
+        v1:gcp-disk-id:*)
+            if [[ ! "$expected_identity" =~ ^v1:gcp-disk-id:[0-9]{1,32}$ ]]; then
+                log "Fresh-filesystem identity is unsafe"
+                return 2
+            fi
+            ;;
+        *)
+            log "Fresh-filesystem identity is unsafe"
+            return 2
+            ;;
+    esac
+
+    if [[ -e "$marker" || -L "$marker" ]]; then
+        if [[ -L "$marker_dir" || ! -d "$marker_dir" ||
+            -L "$marker" || ! -f "$marker" ]]; then
+            log "Fresh-filesystem marker path is unsafe: $marker"
+            return 1
+        fi
+        marker_dir_identity="$(stat -c '%u:%g:%a' -- "$marker_dir")" || return 1
+        marker_identity="$(stat -c '%u:%g:%a:%h' -- "$marker")" || return 1
+        if [[ "$marker_dir_identity" != "0:0:700" ||
+            "$marker_identity" != "0:0:600:1" ]]; then
+            log "Existing fresh-filesystem marker is not root-owned and private: $marker"
+            return 1
+        fi
+        marker_size="$(stat -c '%s' -- "$marker")" || return 1
+        marker_payload=""
+        if [[ "$marker_size" != "$((${#expected_identity} + 1))" ]] ||
+            ! IFS= read -r marker_payload <"$marker" ||
+            [[ "$marker_payload" != "$expected_identity" ]]; then
+            log "Existing fresh-filesystem marker does not match this disk incarnation"
+            return 1
+        fi
+        return 0
+    fi
+
+    # Recover the only safe residue possible if the host stopped after making
+    # the private marker directory but before creating its marker.
+    if [[ -e "$marker_dir" || -L "$marker_dir" ]]; then
+        if [[ -L "$marker_dir" || ! -d "$marker_dir" ]]; then
+            log "Fresh-filesystem marker directory is unsafe: $marker_dir"
+            return 1
+        fi
+        marker_dir_identity="$(stat -c '%u:%g:%a' -- "$marker_dir")" || return 1
+        unexpected="$(find "$marker_dir" -mindepth 1 -maxdepth 1 -print -quit)" || return 1
+        if [[ "$marker_dir_identity" != "0:0:700" || -n "$unexpected" ]]; then
+            log "Fresh-filesystem marker directory is not an empty root-owned staging residue: $marker_dir"
+            return 1
+        fi
+        rmdir -- "$marker_dir"
+        sync
+    fi
+
+    marker_root_identity="$(stat -c '%u:%g:%a' -- "$mount_path")" || return 1
+    if [[ "$marker_root_identity" != "0:0:755" ]]; then
+        log "Fresh-filesystem root is not pristine: $mount_path ($marker_root_identity)"
+        return 1
+    fi
+    unexpected="$(find "$mount_path" -mindepth 1 -maxdepth 1 \
+        ! -name lost+found -print -quit)" || return 1
+    if [[ -n "$unexpected" ]]; then
+        log "Fresh-filesystem root contains unexpected data: $unexpected"
+        return 1
+    fi
+    if [[ -e "$mount_path/lost+found" || -L "$mount_path/lost+found" ]]; then
+        if [[ -L "$mount_path/lost+found" || ! -d "$mount_path/lost+found" ]]; then
+            log "Fresh-filesystem lost+found path is unsafe"
+            return 1
+        fi
+        marker_dir_identity="$(stat -c '%u:%g:%a' -- "$mount_path/lost+found")" || return 1
+        unexpected="$(find "$mount_path/lost+found" -mindepth 1 -maxdepth 1 -print -quit)" || return 1
+        if [[ "$marker_dir_identity" != "0:0:700" || -n "$unexpected" ]]; then
+            log "Fresh-filesystem lost+found is not an empty root-owned directory"
+            return 1
+        fi
+    fi
+
+    if ! (umask 077 && mkdir -- "$marker_dir") ||
+        ! (umask 077 && printf '%s\n' "$expected_identity" >"$marker"); then
+        log "Fresh-filesystem marker path is unsafe: $marker"
+        return 1
+    fi
+    marker_dir_identity="$(stat -c '%u:%g:%a' -- "$marker_dir")" || return 1
+    marker_identity="$(stat -c '%u:%g:%a:%h' -- "$marker")" || return 1
+    if [[ "$marker_dir_identity" != "0:0:700" ||
+        "$marker_identity" != "0:0:600:1" ]]; then
+        log "Published fresh-filesystem marker is not root-owned and private: $marker"
+        return 1
+    fi
+}
+
 usage() {
-    echo "Usage: $0 DEVICE_PATH MOUNT_PATH" >&2
+    echo "Usage: $0 DEVICE_PATH MOUNT_PATH [--publish-fresh-marker [IDENTITY]]" >&2
 }
 
 is_block_device() {
@@ -124,13 +227,35 @@ main() {
     local device_path mount_path device filesystem_type mounted_source resolved_mounted_source
     local already_mounted blkid_status fsck_status wait_seconds sleep_seconds settle_seconds
     local device_mounts_output provider_mount="" moved_source resolved_moved_source
-    local target_contents
+    local target_contents filesystem_label
+    local publish_fresh_marker=false fresh_marker_identity="" fresh_marker_pending=false
     local -a device_mounts
 
-    if [[ $# -ne 2 ]]; then
+    if [[ $# -lt 2 || $# -gt 4 ]]; then
         usage
         return 2
     fi
+    if [[ $# -ge 3 ]]; then
+        if [[ "$3" != "--publish-fresh-marker" ]]; then
+            usage
+            return 2
+        fi
+        publish_fresh_marker=true
+        fresh_marker_identity="${4:-fresh}"
+        case "$fresh_marker_identity" in
+            fresh) ;;
+            v1:gcp-disk-id:*)
+                if [[ ! "$fresh_marker_identity" =~ ^v1:gcp-disk-id:[0-9]{1,32}$ ]]; then
+                    usage
+                    return 2
+                fi
+                ;;
+            *)
+                usage
+                return 2
+                ;;
+        esac
+    fi
 
     device_path="$1"
     mount_path="$2"
@@ -305,6 +430,15 @@ main() {
                 log "Could not grow the ext4 filesystem on $device"
                 return 1
             fi
+            if [[ "$publish_fresh_marker" == "true" ]]; then
+                filesystem_label="$(e2label "$device")" || {
+                    log "Could not inspect the ext4 label on $device"
+                    return 1
+                }
+                if [[ "$filesystem_label" == "$fresh_filesystem_pending_label" ]]; then
+                    fresh_marker_pending=true
+                fi
+            fi
             ;;
         2)
             if [[ "$already_mounted" == "true" ]]; then
@@ -313,7 +447,13 @@ main() {
             fi
             log "No filesystem signature found on $device; creating ext4"
             require_device_unmounted "$device" "filesystem creation"
-            mkfs.ext4 -m 0 -E lazy_itable_init=1,lazy_journal_init=1,nodiscard -- "$device"
+            if [[ "$publish_fresh_marker" == "true" ]]; then
+                mkfs.ext4 -m 0 -E lazy_itable_init=1,lazy_journal_init=1,nodiscard \
+                    -L "$fresh_filesystem_pending_label" -- "$device"
+                fresh_marker_pending=true
+            else
+                mkfs.ext4 -m 0 -E lazy_itable_init=1,lazy_journal_init=1,nodiscard -- "$device"
+            fi
             ;;
         *)
             log "Could not safely inspect filesystem signatures on $device (blkid status $blkid_status)"
@@ -325,6 +465,16 @@ main() {
         require_device_unmounted "$device" "cloud-compose mount"
         mount -o defaults -- "$device" "$mount_path"
     fi
+    if [[ "$fresh_marker_pending" == "true" ]]; then
+        publish_fresh_filesystem_marker "$mount_path" "$fresh_marker_identity"
+        sync
+        if ! e2label "$device" ""; then
+            log "Published the fresh-filesystem marker but could not clear the pending ext4 label on $device"
+            return 1
+        fi
+        sync
+        log "Published fresh-filesystem marker at $mount_path/.cloud-compose/fresh-filesystem"
+    fi
 }
 
 if [[ "${BASH_SOURCE[0]}" == "$0" ]]; then
diff --git a/rootfs/home/cloud-compose/rotate-keys-app.sh b/rootfs/home/cloud-compose/rotate-keys-app.sh
index 714cad1..1b15f9d 100644
--- a/rootfs/home/cloud-compose/rotate-keys-app.sh
+++ b/rootfs/home/cloud-compose/rotate-keys-app.sh
@@ -26,6 +26,9 @@ APP_CREDENTIALS_FILE="${APP_CREDENTIALS_FILE:-/mnt/disks/data/cloud-compose/app/
 ROTATION_APP_CREDENTIAL_OWNER="${ROTATION_CREDENTIAL_OWNER-100}"
 ROTATION_CENTRAL_CREDENTIAL_OWNER="${ROTATION_CENTRAL_CREDENTIAL_OWNER-root}"
 ROTATION_CREDENTIAL_GROUP="${ROTATION_CREDENTIAL_GROUP-cloud-compose}"
+CLOUD_COMPOSE_FRESH_FILESYSTEM_MARKER="${CLOUD_COMPOSE_FRESH_FILESYSTEM_MARKER:-/mnt/disks/data/.cloud-compose/fresh-filesystem}"
+CLOUD_COMPOSE_FRESH_FILESYSTEM_IDENTITY="${CLOUD_COMPOSE_FRESH_FILESYSTEM_IDENTITY:-}"
+rotation_reconcile_orphans=false
 rotation_action="${1:-rotate}"
 
 if [[ -n "$ROTATION_CREDENTIAL_GROUP" &&
@@ -55,14 +58,165 @@ case "${GCP_APP_CREDENTIALS_ENABLED:-false}" in
     ;;
 esac
 
+case "${GCP_APP_SERVICE_ACCOUNT_MANAGED:-false}" in
+  true) app_service_account_managed=true ;;
+  false) app_service_account_managed=false ;;
+  *)
+    echo "GCP_APP_SERVICE_ACCOUNT_MANAGED must be true or false" >&2
+    exit 2
+    ;;
+esac
+
 rotate_keys() {
   ROTATION_CREDENTIAL_OWNER="$ROTATION_CENTRAL_CREDENTIAL_OWNER" \
+  ROTATION_RECONCILE_ORPHANS="$rotation_reconcile_orphans" \
+  CLOUD_COMPOSE_FRESH_FILESYSTEM_MARKER="$CLOUD_COMPOSE_FRESH_FILESYSTEM_MARKER" \
+  CLOUD_COMPOSE_FRESH_FILESYSTEM_IDENTITY="$CLOUD_COMPOSE_FRESH_FILESYSTEM_IDENTITY" \
   bash "$rotate_keys_script" "$1" \
     "$GCP_APP_SERVICE_ACCOUNT_EMAIL" \
     "$GCP_PROJECT" \
     "$APP_CREDENTIALS_FILE"
 }
 
+require_inactive_app_service() {
+  local load_state active_state
+
+  load_state="$(systemctl show --property=LoadState --value -- cloud-compose.service)" || {
+    echo "Could not determine whether cloud-compose.service is loaded" >&2
+    return 1
+  }
+  active_state="$(systemctl show --property=ActiveState --value -- cloud-compose.service)" || {
+    echo "Could not determine whether cloud-compose.service is inactive" >&2
+    return 1
+  }
+  if [[ "$load_state" != "loaded" || "$active_state" != "inactive" ]]; then
+    echo "Fresh-filesystem key reconciliation requires loaded, inactive cloud-compose.service; observed ${load_state}/${active_state}" >&2
+    return 1
+  fi
+}
+
+app_credential_key_id() {
+  local file="$1" key_id
+
+  key_id="$(jq -jr '
+    (.private_key_id |
+      select(type == "string" and length > 0 and (explode | index(0) == null))),
+    "\u001f"
+  ' "$file")" || return 1
+  key_id="${key_id%$'\x1f'}"
+  [[ "$key_id" =~ ^[A-Za-z0-9_-]+$ ]] || return 1
+  printf '%s\n' "$key_id"
+}
+
+validate_app_credentials() {
+  local file="$1" key_id
+
+  [[ ! -L "$file" && -f "$file" ]] || return 1
+  key_id="$(app_credential_key_id "$file")" || return 1
+  jq -e \
+    --arg key_id "$key_id" \
+    --arg service_account "$GCP_APP_SERVICE_ACCOUNT_EMAIL" \
+    --arg project_id "$GCP_PROJECT" '
+      .type == "service_account" and
+      .private_key_id == $key_id and
+      .client_email == $service_account and
+      .project_id == $project_id and
+      .token_uri == "https://oauth2.googleapis.com/token" and
+      (.private_key | type == "string" and
+        startswith("-----BEGIN PRIVATE KEY-----") and
+        contains("-----END PRIVATE KEY-----"))
+    ' "$file" >/dev/null
+}
+
+restore_central_app_credentials() {
+  local source="$1" target_dir target_tmp
+
+  target_dir="$(dirname -- "$APP_CREDENTIALS_FILE")"
+  if [[ -L "$target_dir" || ( -e "$target_dir" && ! -d "$target_dir" ) ]]; then
+    echo "Application credential directory is unsafe: $target_dir" >&2
+    return 1
+  fi
+  install -d -m 0750 -- "$target_dir"
+  if [[ -L "$target_dir" || ! -d "$target_dir" ]]; then
+    echo "Application credential directory changed during recovery: $target_dir" >&2
+    return 1
+  fi
+
+  target_tmp="$(mktemp "${APP_CREDENTIALS_FILE}.tmp.XXXXXX")" || return 1
+  if ! install -m 0440 "$source" "$target_tmp"; then
+    rm -f -- "$target_tmp"
+    return 1
+  fi
+  if ! touch -r "$source" "$target_tmp"; then
+    rm -f -- "$target_tmp"
+    return 1
+  fi
+  if ! chown -- "$ROTATION_CENTRAL_CREDENTIAL_OWNER" "$target_tmp"; then
+    rm -f -- "$target_tmp"
+    return 1
+  fi
+  if [[ -n "$ROTATION_CREDENTIAL_GROUP" ]] &&
+    ! chgrp -- "$ROTATION_CREDENTIAL_GROUP" "$target_tmp"; then
+    rm -f -- "$target_tmp"
+    return 1
+  fi
+  if ! mv -f -- "$target_tmp" "$APP_CREDENTIALS_FILE"; then
+    rm -f -- "$target_tmp"
+    return 1
+  fi
+  validate_app_credentials "$APP_CREDENTIALS_FILE" || {
+    echo "Recovered central application credentials failed validation" >&2
+    return 1
+  }
+  echo "Recovered central application credentials from unanimous distributed copies"
+}
+
+recover_distributed_app_credentials() {
+  local first_target="" target
+  local -a pending_artifacts=(
+    "${APP_CREDENTIALS_FILE}.rotation-pending.json"
+    "${APP_CREDENTIALS_FILE}.rotation-staged.json"
+    "${APP_CREDENTIALS_FILE}.rotation-previous.json"
+    "${APP_CREDENTIALS_FILE}.rotation-replacement.json"
+  )
+
+  if [[ -f "$APP_CREDENTIALS_FILE" ]]; then
+    return 0
+  fi
+  for target in "${pending_artifacts[@]}"; do
+    if [[ -e "$target" || -L "$target" ]]; then
+      if [[ "$target" == "${APP_CREDENTIALS_FILE}.rotation-pending.json" ]]; then
+        return 0
+      fi
+      echo "Cannot recover central credentials while an orphaned rotation artifact remains: $target" >&2
+      return 1
+    fi
+  done
+
+  for target in "${app_credential_targets[@]}"; do
+    if [[ ! -e "$target" && ! -L "$target" ]]; then
+      continue
+    fi
+    if ! validate_app_credentials "$target"; then
+      echo "Distributed application credentials are invalid: $target" >&2
+      return 1
+    fi
+    if [[ -z "$first_target" ]]; then
+      first_target="$target"
+    elif ! cmp -s -- "$first_target" "$target"; then
+      echo "Distributed application credentials disagree; refusing remote key reconciliation" >&2
+      return 1
+    fi
+  done
+  if [[ -n "$first_target" ]]; then
+    if [[ "$app_service_account_managed" != "true" ]]; then
+      echo "Caller-supplied app identities require operator-reviewed central credential recovery" >&2
+      return 1
+    fi
+    restore_central_app_credentials "$first_target"
+  fi
+}
+
 install_app_credentials() {
   local target="$1"
   local target_dir target_tmp
@@ -144,6 +298,13 @@ for target in "${credential_artifacts[@]}"; do
     exit 2
   fi
 done
+for target in "${app_credential_targets[@]}"; do
+  target_dir="$(dirname -- "$target")"
+  if [[ -L "$target_dir" || ( -e "$target_dir" && ! -d "$target_dir" ) ]]; then
+    echo "Application credential directory is unsafe: $target_dir" >&2
+    exit 2
+  fi
+done
 
 if [[ "$rotation_action" == "retire" ]]; then
   rotate_keys retire
@@ -163,6 +324,14 @@ if [[ "$app_credentials_enabled" != "true" ]]; then
   exit 0
 fi
 
+recover_distributed_app_credentials
+if [[ ! -e "$APP_CREDENTIALS_FILE" && ! -L "$APP_CREDENTIALS_FILE" &&
+  "$app_service_account_managed" == "true" &&
+  ( -e "$CLOUD_COMPOSE_FRESH_FILESYSTEM_MARKER" || -L "$CLOUD_COMPOSE_FRESH_FILESYSTEM_MARKER" ) ]]; then
+  require_inactive_app_service
+  rotation_reconcile_orphans=true
+fi
+
 install -d -m 0750 "$(dirname -- "$APP_CREDENTIALS_FILE")"
 if [[ "$rotation_action" == "rollback" ]]; then
   rotate_keys rollback
diff --git a/rootfs/home/cloud-compose/rotate-keys-internal.sh b/rootfs/home/cloud-compose/rotate-keys-internal.sh
index 5546e9a..f8aca1e 100644
--- a/rootfs/home/cloud-compose/rotate-keys-internal.sh
+++ b/rootfs/home/cloud-compose/rotate-keys-internal.sh
@@ -20,6 +20,9 @@ esac
 
 INTERNAL_CREDENTIALS_FILE="${INTERNAL_CREDENTIALS_FILE:-/mnt/disks/data/libops-internal/GOOGLE_APPLICATION_CREDENTIALS}"
 ROTATION_CREDENTIAL_GROUP="${ROTATION_CREDENTIAL_GROUP-cloud-compose}"
+CLOUD_COMPOSE_FRESH_FILESYSTEM_MARKER="${CLOUD_COMPOSE_FRESH_FILESYSTEM_MARKER:-/mnt/disks/data/.cloud-compose/fresh-filesystem}"
+CLOUD_COMPOSE_FRESH_FILESYSTEM_IDENTITY="${CLOUD_COMPOSE_FRESH_FILESYSTEM_IDENTITY:-}"
+rotation_reconcile_orphans=false
 internal_service_account="internal-$GCP_INSTANCE_NAME@$GCP_PROJECT.iam.gserviceaccount.com"
 rotation_action="${1:-rotate}"
 
@@ -38,12 +41,38 @@ case "$rotation_action" in
 esac
 
 rotate_keys() {
+  ROTATION_RECONCILE_ORPHANS="$rotation_reconcile_orphans" \
+  CLOUD_COMPOSE_FRESH_FILESYSTEM_MARKER="$CLOUD_COMPOSE_FRESH_FILESYSTEM_MARKER" \
+  CLOUD_COMPOSE_FRESH_FILESYSTEM_IDENTITY="$CLOUD_COMPOSE_FRESH_FILESYSTEM_IDENTITY" \
   bash "$rotate_keys_script" "$1" \
     "$internal_service_account" \
     "$GCP_PROJECT" \
     "$INTERNAL_CREDENTIALS_FILE"
 }
 
+require_inactive_internal_service() {
+  local load_state active_state
+
+  load_state="$(systemctl show --property=LoadState --value -- cloud-compose-internal-services.service)" || {
+    echo "Could not determine whether cloud-compose-internal-services.service is loaded" >&2
+    return 1
+  }
+  active_state="$(systemctl show --property=ActiveState --value -- cloud-compose-internal-services.service)" || {
+    echo "Could not determine whether cloud-compose-internal-services.service is inactive" >&2
+    return 1
+  }
+  if [[ "$load_state" != "loaded" || "$active_state" != "inactive" ]]; then
+    echo "Fresh-filesystem key reconciliation requires loaded, inactive cloud-compose-internal-services.service; observed ${load_state}/${active_state}" >&2
+    return 1
+  fi
+}
+
+if [[ ! -e "$INTERNAL_CREDENTIALS_FILE" && ! -L "$INTERNAL_CREDENTIALS_FILE" &&
+  ( -e "$CLOUD_COMPOSE_FRESH_FILESYSTEM_MARKER" || -L "$CLOUD_COMPOSE_FRESH_FILESYSTEM_MARKER" ) ]]; then
+  require_inactive_internal_service
+  rotation_reconcile_orphans=true
+fi
+
 if [[ "$rotation_action" == "rollback" ]]; then
   rotate_keys rollback
 else
diff --git a/rootfs/home/cloud-compose/rotate-keys.sh b/rootfs/home/cloud-compose/rotate-keys.sh
index 57b4cba..9735301 100644
--- a/rootfs/home/cloud-compose/rotate-keys.sh
+++ b/rootfs/home/cloud-compose/rotate-keys.sh
@@ -93,6 +93,10 @@ ROTATION_AUTH_MAX_RETRIES="${ROTATION_AUTH_MAX_RETRIES:-6}"
 ROTATION_AUTH_SLEEP_SECONDS="${ROTATION_AUTH_SLEEP_SECONDS:-5}"
 ROTATION_RECOVERY_SETTLE_SECONDS="${ROTATION_RECOVERY_SETTLE_SECONDS:-60}"
 ROTATION_CREDENTIAL_OWNER="${ROTATION_CREDENTIAL_OWNER-100}"
+ROTATION_RECONCILE_ORPHANS="${ROTATION_RECONCILE_ORPHANS:-false}"
+CLOUD_COMPOSE_FRESH_FILESYSTEM_MARKER="${CLOUD_COMPOSE_FRESH_FILESYSTEM_MARKER:-/mnt/disks/data/.cloud-compose/fresh-filesystem}"
+CLOUD_COMPOSE_FRESH_FILESYSTEM_IDENTITY="${CLOUD_COMPOSE_FRESH_FILESYSTEM_IDENTITY:-}"
+ROTATION_RUNTIME_DIR="${ROTATION_RUNTIME_DIR:-/run/cloud-compose-key-rotation}"
 
 validate_integer_setting() {
     local name="$1" value="$2" minimum="$3" maximum="$4"
@@ -114,6 +118,13 @@ if [[ -n "$ROTATION_CREDENTIAL_OWNER" &&
     log_error "ROTATION_CREDENTIAL_OWNER must be empty, a numeric UID, or a safe local account name"
     exit 2
 fi
+case "$ROTATION_RECONCILE_ORPHANS" in
+    true | false) ;;
+    *)
+        log_error "ROTATION_RECONCILE_ORPHANS must be true or false"
+        exit 2
+        ;;
+esac
 
 if [[ -L "$credentials_dir" || ( -e "$credentials_dir" && ! -d "$credentials_dir" ) ]]; then
     log_error "Credentials directory is unsafe: $credentials_dir"
@@ -141,6 +152,19 @@ STATE_READY_AT=0
 STATE_DISABLED_AT=0
 ACCESS_TOKEN=""
 KEY_OPERATION_RESULT=""
+EPHEMERAL_CREATE_RESPONSE=""
+
+cleanup_ephemeral_create_response() {
+    if [[ -n "$EPHEMERAL_CREATE_RESPONSE" ]]; then
+        rm -f -- "$EPHEMERAL_CREATE_RESPONSE"
+        EPHEMERAL_CREATE_RESPONSE=""
+    fi
+}
+
+trap cleanup_ephemeral_create_response EXIT
+trap 'exit 129' HUP
+trap 'exit 130' INT
+trap 'exit 143' TERM
 
 now_epoch() {
     date +%s
@@ -175,6 +199,54 @@ valid_iam_key_name() {
     valid_iam_key_id "$key_id" && [[ "$key_name" == "$prefix$key_id" ]]
 }
 
+require_fresh_reconciliation_authority() {
+    local marker_dir marker_root marker_identity marker_dir_identity marker_payload
+    local marker_root_identity marker_size
+
+    if [[ "$ROTATION_RECONCILE_ORPHANS" != "true" ]]; then
+        log_error "Managed orphan-key reconciliation was not authorized"
+        return 1
+    fi
+    if [[ ! "$CLOUD_COMPOSE_FRESH_FILESYSTEM_IDENTITY" =~ ^v1:gcp-disk-id:[0-9]{1,32}$ ]]; then
+        log_error "GCP fresh-filesystem identity is missing or unsafe"
+        return 1
+    fi
+    if [[ "$CLOUD_COMPOSE_FRESH_FILESYSTEM_MARKER" != /* ||
+        "$CLOUD_COMPOSE_FRESH_FILESYSTEM_MARKER" == "/" ||
+        "$CLOUD_COMPOSE_FRESH_FILESYSTEM_MARKER" == *$'\n'* ||
+        "$CLOUD_COMPOSE_FRESH_FILESYSTEM_MARKER" == *$'\r'* ||
+        "$CLOUD_COMPOSE_FRESH_FILESYSTEM_MARKER" =~ (^|/)\.\.?(/|$) ]]; then
+        log_error "Fresh-filesystem marker path is unsafe: $CLOUD_COMPOSE_FRESH_FILESYSTEM_MARKER"
+        return 1
+    fi
+    marker_dir="$(dirname -- "$CLOUD_COMPOSE_FRESH_FILESYSTEM_MARKER")"
+    marker_root="$(dirname -- "$marker_dir")"
+    if [[ -L "$marker_root" || ! -d "$marker_root" ||
+        -L "$marker_dir" || ! -d "$marker_dir" ||
+        -L "$CLOUD_COMPOSE_FRESH_FILESYSTEM_MARKER" ||
+        ! -f "$CLOUD_COMPOSE_FRESH_FILESYSTEM_MARKER" ]]; then
+        log_error "Fresh-filesystem marker is missing or unsafe: $CLOUD_COMPOSE_FRESH_FILESYSTEM_MARKER"
+        return 1
+    fi
+    marker_root_identity="$(stat -c '%u:%a' -- "$marker_root")" || return 1
+    marker_dir_identity="$(stat -c '%u:%g:%a' -- "$marker_dir")" || return 1
+    marker_identity="$(stat -c '%u:%g:%a:%h' -- "$CLOUD_COMPOSE_FRESH_FILESYSTEM_MARKER")" || return 1
+    if [[ "$marker_root_identity" != "0:1775" ||
+        "$marker_dir_identity" != "0:0:700" ||
+        "$marker_identity" != "0:0:600:1" ]]; then
+        log_error "Fresh-filesystem marker ownership or mode is unsafe: $CLOUD_COMPOSE_FRESH_FILESYSTEM_MARKER"
+        return 1
+    fi
+    marker_size="$(stat -c '%s' -- "$CLOUD_COMPOSE_FRESH_FILESYSTEM_MARKER")" || return 1
+    marker_payload=""
+    if [[ "$marker_size" != "$((${#CLOUD_COMPOSE_FRESH_FILESYSTEM_IDENTITY} + 1))" ]] ||
+        ! IFS= read -r marker_payload <"$CLOUD_COMPOSE_FRESH_FILESYSTEM_MARKER" ||
+        [[ "$marker_payload" != "$CLOUD_COMPOSE_FRESH_FILESYSTEM_IDENTITY" ]]; then
+        log_error "Fresh-filesystem marker does not match this GCP data disk incarnation"
+        return 1
+    fi
+}
+
 write_state() {
     local state_tmp
 
@@ -210,6 +282,10 @@ write_state() {
     fi
     chmod 0600 "$state_tmp"
     mv -f -- "$state_tmp" "$PENDING_STATE"
+    # IAM mutation must never outrun the exact state that makes its retry
+    # idempotent. A global sync is available on COS and flushes both the
+    # renamed file and its parent-directory entry.
+    sync
 }
 
 load_state() {
@@ -225,7 +301,8 @@ load_state() {
         --arg credentials_file "$CREDENTIALS_FILE" '
         select(
             .version == 2 and
-            (.phase == "creating" or .phase == "staged" or .phase == "authenticated" or
+            (.phase == "reconciling" or .phase == "creating-fresh" or
+             .phase == "creating" or .phase == "staged" or .phase == "authenticated" or
              .phase == "ready" or .phase == "grace" or .phase == "rolling-back" or
              .phase == "rollback" or .phase == "revoke-new") and
             .service_account == $service_account and
@@ -235,6 +312,7 @@ load_state() {
             (.new_key_id | type == "string" and (explode | index(0) == null)) and
             (.new_key_name | type == "string" and (explode | index(0) == null)) and
             (.baseline_key_names | type == "array") and
+            (.baseline_key_names | length <= 10 and . == (sort | unique)) and
             all(.baseline_key_names[];
                 type == "string" and (explode | index(0) == null)) and
             (.created_at | type == "number" and . >= 0 and floor == .) and
@@ -288,6 +366,10 @@ load_state() {
 }
 
 cleanup_state() {
+    # A stale cleanup is safe: every preceding cloud mutation targets the
+    # exact persisted key name and its retry accepts the already-absent state.
+    # Durability barriers are required before mutations, not after removing
+    # retry metadata for an operation that has already converged.
     rm -f -- "$STAGED_CREDENTIALS" "$PREVIOUS_CREDENTIALS" \
         "$REPLACEMENT_CREDENTIALS" "$PENDING_STATE"
 }
@@ -317,10 +399,11 @@ write_access_header_file() {
 }
 
 list_user_keys() {
-    local keys_response curl_status=0
+    local keys_response normalized_keys encoded_key_name key_name curl_status=0
 
     write_access_header_file || return 1
     keys_response="$(curl -fsS --retry 5 --retry-all-errors --retry-delay 2 --retry-max-time 120 \
+        --max-filesize 1048576 \
         --connect-timeout 5 --max-time 30 \
         -H "@$ACCESS_HEADER_FILE" \
         "https://iam.googleapis.com/v1/$SA_RESOURCE/keys")" || curl_status=$?
@@ -329,17 +412,40 @@ list_user_keys() {
         log_error "Failed to list service-account keys"
         return 1
     fi
-    jq -cer --arg prefix "$SA_RESOURCE/keys/" '
+    normalized_keys="$(jq -cer --arg prefix "$SA_RESOURCE/keys/" '
         (.keys // []) as $keys |
         if ($keys | type) != "array" then error("invalid key list") else
-            [$keys[] |
-                select(.keyType == "USER_MANAGED") |
-                select((.name | type) == "string" and (.name | startswith($prefix))) |
-                {name: .name, disabled: (.disabled // false)}
-            ] |
-            if all(.[]; (.disabled | type) == "boolean") then . else error("invalid disabled state") end
+            [$keys[] | select(.keyType == "USER_MANAGED")] as $user_keys |
+            if ($user_keys | length) > 10 then error("too many user-managed keys")
+            elif any($user_keys[];
+                (.name | type) != "string" or
+                (.name | startswith($prefix) | not) or
+                ((.disabled // false) | type) != "boolean")
+            then error("invalid user-managed key")
+            elif ([$user_keys[].name] | unique | length) != ($user_keys | length)
+            then error("duplicate user-managed key")
+            else
+                [$user_keys[] | {name: .name, disabled: (.disabled // false)}] | sort_by(.name)
+            end
         end
-    ' <<<"$keys_response"
+    ' <<<"$keys_response")" || {
+        log_error "IAM returned an invalid user-managed key inventory"
+        return 1
+    }
+    while IFS= read -r encoded_key_name; do
+        key_name="$(
+            if ! printf '%s' "$encoded_key_name" | base64 --decode; then
+                exit 1
+            fi
+            printf '\037'
+        )" || return 1
+        key_name="${key_name%$'\x1f'}"
+        if ! valid_iam_key_name "$key_name"; then
+            log_error "IAM returned an invalid user-managed key name"
+            return 1
+        fi
+    done < <(jq -r '.[].name | @base64' <<<"$normalized_keys")
+    printf '%s\n' "$normalized_keys"
 }
 
 list_user_key_names() {
@@ -577,12 +683,220 @@ authenticate_credentials_with_backoff() {
     return 1
 }
 
+reset_after_definite_create_rejection() {
+    if [[ "$STATE_PHASE" == "creating-fresh" ]]; then
+        STATE_PHASE=reconciling
+        STATE_BASELINE_KEY_NAMES='[]'
+        STATE_CREATED_AT="$(now_epoch)"
+        write_state
+    else
+        cleanup_state
+    fi
+}
+
+definite_create_rejection_status() {
+    case "$1" in
+        400 | 401 | 403 | 404 | 409 | 429) return 0 ;;
+        *) return 1 ;;
+    esac
+}
+
+prepare_rotation_runtime_dir() {
+    local runtime_identity
+
+    if [[ "$ROTATION_RUNTIME_DIR" != /* || "$ROTATION_RUNTIME_DIR" == "/" ||
+        "$ROTATION_RUNTIME_DIR" == *$'\n'* || "$ROTATION_RUNTIME_DIR" == *$'\r'* ||
+        "$ROTATION_RUNTIME_DIR" =~ (^|/)\.\.?(/|$) ||
+        -L "$ROTATION_RUNTIME_DIR" ||
+        ( -e "$ROTATION_RUNTIME_DIR" && ! -d "$ROTATION_RUNTIME_DIR" ) ]]; then
+        log_error "Rotation runtime directory is unsafe: $ROTATION_RUNTIME_DIR"
+        return 1
+    fi
+    install -d -m 0700 -- "$ROTATION_RUNTIME_DIR" || return 1
+    if [[ -L "$ROTATION_RUNTIME_DIR" || ! -d "$ROTATION_RUNTIME_DIR" ]]; then
+        log_error "Rotation runtime directory changed during preparation: $ROTATION_RUNTIME_DIR"
+        return 1
+    fi
+    runtime_identity="$(stat -c '%u:%a' -- "$ROTATION_RUNTIME_DIR")" || return 1
+    if [[ "$runtime_identity" != "${EUID}:700" ]]; then
+        log_error "Rotation runtime directory has unsafe ownership or mode: $ROTATION_RUNTIME_DIR"
+        return 1
+    fi
+}
+
+create_replacement_key() {
+    local response_tmp http_status="" curl_status=0
+    local new_key_name="" new_key_id private_key_data="" staged_tmp
+
+    prepare_rotation_runtime_dir || {
+        reset_after_definite_create_rejection
+        return 1
+    }
+    response_tmp="$(mktemp "${ROTATION_RUNTIME_DIR}/iam-create-response.XXXXXX")" || {
+        reset_after_definite_create_rejection
+        return 1
+    }
+    EPHEMERAL_CREATE_RESPONSE="$response_tmp"
+    chmod 0600 "$response_tmp"
+    if ! write_access_header_file; then
+        cleanup_ephemeral_create_response
+        reset_after_definite_create_rejection
+        return 1
+    fi
+
+    log_info "Creating replacement key for $SERVICE_ACCOUNT"
+    http_status="$(curl -sS --connect-timeout 5 --max-time 30 -X POST \
+        -H "@$ACCESS_HEADER_FILE" \
+        -H "Content-Type: application/json" \
+        -o "$response_tmp" -w '%{http_code}' \
+        "https://iam.googleapis.com/v1/$SA_RESOURCE/keys")" || curl_status=$?
+    rm -f -- "$ACCESS_HEADER_FILE"
+
+    new_key_name="$(jq -er '.name | select(type == "string" and length > 0)' \
+        "$response_tmp" 2>/dev/null)" || new_key_name=""
+    if [[ -z "$new_key_name" ]]; then
+        cleanup_ephemeral_create_response
+        if definite_create_rejection_status "$http_status"; then
+            reset_after_definite_create_rejection
+            log_error "IAM rejected key creation with HTTP $http_status; no key was created"
+        else
+            log_error "Key creation had an indeterminate result; pending state prevents another key until audited recovery"
+        fi
+        return 1
+    fi
+    new_key_id="${new_key_name##*/}"
+    if [[ ! "$new_key_id" =~ ^[A-Za-z0-9_-]+$ ]] ||
+        [[ "$new_key_name" != "$SA_RESOURCE/keys/$new_key_id" ]]; then
+        cleanup_ephemeral_create_response
+        log_error "Key creation response contained an unexpected resource name; use audited recovery"
+        return 1
+    fi
+
+    STATE_NEW_KEY_ID="$new_key_id"
+    STATE_NEW_KEY_NAME="$new_key_name"
+    private_key_data="$(jq -er '.privateKeyData | select(type == "string" and length > 0)' \
+        "$response_tmp" 2>/dev/null)" || private_key_data=""
+    cleanup_ephemeral_create_response
+    if [[ -z "$private_key_data" ]]; then
+        STATE_PHASE=revoke-new
+        write_state
+        if delete_key "$new_key_name"; then
+            cleanup_state
+        fi
+        log_error "Key creation response omitted replacement credentials"
+        return 1
+    fi
+    if ((curl_status != 0)); then
+        log_info "Recovered a complete replacement credential from an interrupted IAM response"
+    fi
+
+    staged_tmp="$(mktemp "${STAGED_CREDENTIALS}.tmp.XXXXXX")" || return 1
+    if ! printf '%s' "$private_key_data" | base64 -d >"$staged_tmp" ||
+        ! validate_credentials_key "$staged_tmp" "$new_key_id"; then
+        rm -f -- "$staged_tmp"
+        STATE_PHASE=revoke-new
+        write_state
+        if delete_key "$new_key_name"; then
+            cleanup_state
+        fi
+        log_error "IAM returned invalid replacement credentials"
+        return 1
+    fi
+    chmod 0600 "$staged_tmp"
+    mv -f -- "$staged_tmp" "$STAGED_CREDENTIALS"
+
+    STATE_PHASE=staged
+    write_state
+    if ! install_staged_credentials; then
+        STATE_PHASE=revoke-new
+        write_state
+        if delete_key "$new_key_name"; then
+            cleanup_state
+        fi
+        log_error "Could not safely preserve and install replacement credentials"
+        return 1
+    fi
+    log_info "Installed replacement key $new_key_id; previous credentials remain available for rollback"
+}
+
+finish_orphan_reconciliation() {
+    local encoded_key_name key_name current_names remaining_baseline unexpected
+    local remaining_ids unexpected_ids
+
+    while IFS= read -r encoded_key_name; do
+        key_name="$(
+            if ! printf '%s' "$encoded_key_name" | base64 --decode; then
+                exit 1
+            fi
+            printf '\037'
+        )" || return 1
+        key_name="${key_name%$'\x1f'}"
+        valid_iam_key_name "$key_name" || return 1
+        log_info "Deleting fresh-filesystem orphan key ${key_name##*/}"
+        if ! delete_key "$key_name"; then
+            log_error "Failed to delete fresh-filesystem orphan key ${key_name##*/}; reconciliation will retry"
+            return 1
+        fi
+    done < <(jq -r '.[] | @base64' <<<"$STATE_BASELINE_KEY_NAMES")
+
+    current_names="$(list_user_key_names)" || return 1
+    remaining_baseline="$(jq -cn \
+        --argjson before "$STATE_BASELINE_KEY_NAMES" \
+        --argjson after "$current_names" \
+        '$after | map(select(. as $name | $before | index($name))) | sort')" || return 1
+    unexpected="$(jq -cn \
+        --argjson before "$STATE_BASELINE_KEY_NAMES" \
+        --argjson after "$current_names" \
+        '$after - $before | sort')" || return 1
+
+    if [[ "$(jq -r 'length' <<<"$unexpected")" != "0" ]]; then
+        unexpected_ids="$(jq -r '[.[] | split("/")[-1]] | join(", ")' <<<"$unexpected")" || return 1
+        log_error "Unexpected concurrent user-managed keys appeared during fresh-filesystem reconciliation: $unexpected_ids"
+        return 1
+    fi
+    if [[ "$(jq -r 'length' <<<"$remaining_baseline")" != "0" ]]; then
+        remaining_ids="$(jq -r '[.[] | split("/")[-1]] | join(", ")' <<<"$remaining_baseline")" || return 1
+        log_error "Deleted orphan keys are still visible and will be retried: $remaining_ids"
+        return 1
+    fi
+
+    STATE_PHASE=creating-fresh
+    STATE_BASELINE_KEY_NAMES='[]'
+    STATE_CREATED_AT="$(now_epoch)"
+    write_state
+    create_replacement_key
+}
+
+begin_orphan_reconciliation() {
+    require_fresh_reconciliation_authority || return 1
+    fetch_access_token || return 1
+
+    STATE_PHASE=reconciling
+    STATE_CURRENT_KEY_ID=""
+    STATE_NEW_KEY_ID=""
+    STATE_NEW_KEY_NAME=""
+    STATE_BASELINE_KEY_NAMES="$(list_user_key_names)" || return 1
+    STATE_CREATED_AT="$(now_epoch)"
+    STATE_READY_AT=0
+    STATE_DISABLED_AT=0
+    write_state
+    finish_orphan_reconciliation
+}
+
 resume_pending_prepare() {
     local recovered_key_id
 
     load_state || return 1
     case "$STATE_PHASE" in
-        creating)
+        reconciling)
+            require_fresh_reconciliation_authority || return 1
+            fetch_access_token || return 1
+            finish_orphan_reconciliation
+            ;;
+        creating | creating-fresh)
+            if [[ "$STATE_PHASE" == "creating-fresh" ]]; then
+                require_fresh_reconciliation_authority || return 1
+            fi
             if [[ -L "$STAGED_CREDENTIALS" || ! -f "$STAGED_CREDENTIALS" ]]; then
                 log_error "Key creation has an ambiguous result; run audit, then recover with an explicitly confirmed orphan key ID"
                 return 1
@@ -643,8 +957,7 @@ resume_pending_prepare() {
 
 prepare_rotation() {
     local current_key_id="" file_modified_at file_age_seconds
-    local new_key_response="" new_key_name new_key_id private_key_data staged_tmp
-    local create_succeeded=true curl_status=0
+    local baseline_key_names baseline_count baseline_ids
 
     if [[ -L "$CREDENTIALS_FILE" || ( -e "$CREDENTIALS_FILE" && ! -f "$CREDENTIALS_FILE" ) ]]; then
         log_error "Credentials path is unsafe: $CREDENTIALS_FILE"
@@ -673,83 +986,29 @@ prepare_rotation() {
             log_error "Existing credentials file does not match the rotation target: $CREDENTIALS_FILE"
             return 1
         }
+    elif [[ "$ROTATION_RECONCILE_ORPHANS" == "true" ]]; then
+        begin_orphan_reconciliation
+        return
     fi
 
     fetch_access_token || return 1
+    baseline_key_names="$(list_user_key_names)" || return 1
+    baseline_count="$(jq -r 'length' <<<"$baseline_key_names")" || return 1
+    if ((10#$baseline_count >= 10)); then
+        baseline_ids="$(jq -r '[.[] | split("/")[-1]] | join(", ")' <<<"$baseline_key_names")" || return 1
+        log_error "Service account already has the 10 user-managed keys allowed by IAM: $baseline_ids"
+        return 1
+    fi
     STATE_PHASE=creating
     STATE_CURRENT_KEY_ID="$current_key_id"
     STATE_NEW_KEY_ID=""
     STATE_NEW_KEY_NAME=""
-    STATE_BASELINE_KEY_NAMES="$(list_user_key_names)" || return 1
+    STATE_BASELINE_KEY_NAMES="$baseline_key_names"
     STATE_CREATED_AT="$(now_epoch)"
     STATE_READY_AT=0
     STATE_DISABLED_AT=0
     write_state
-
-    log_info "Creating replacement key for $SERVICE_ACCOUNT"
-    write_access_header_file || return 1
-    new_key_response="$(curl -fsS --connect-timeout 5 --max-time 30 -X POST \
-        -H "@$ACCESS_HEADER_FILE" \
-        -H "Content-Type: application/json" \
-        "https://iam.googleapis.com/v1/$SA_RESOURCE/keys")" || curl_status=$?
-    rm -f -- "$ACCESS_HEADER_FILE"
-    if ((curl_status != 0)); then
-        create_succeeded=false
-    fi
-    if [[ "$create_succeeded" != "true" ]]; then
-        log_error "Key creation had an indeterminate result; pending state prevents another key until audited recovery"
-        return 1
-    fi
-
-    new_key_name="$(jq -er '.name | select(type == "string" and length > 0)' <<<"$new_key_response")" || {
-        log_error "Key creation response omitted the key resource name; use audited recovery"
-        return 1
-    }
-    new_key_id="${new_key_name##*/}"
-    if [[ ! "$new_key_id" =~ ^[A-Za-z0-9_-]+$ ]] || [[ "$new_key_name" != "$SA_RESOURCE/keys/$new_key_id" ]]; then
-        log_error "Key creation response contained an unexpected resource name; use audited recovery"
-        return 1
-    fi
-
-    STATE_NEW_KEY_ID="$new_key_id"
-    STATE_NEW_KEY_NAME="$new_key_name"
-    private_key_data="$(jq -er '.privateKeyData | select(type == "string" and length > 0)' <<<"$new_key_response")" || {
-        STATE_PHASE=revoke-new
-        write_state
-        if delete_key "$new_key_name"; then
-            cleanup_state
-        fi
-        log_error "Key creation response omitted replacement credentials"
-        return 1
-    }
-
-    staged_tmp="$(mktemp "${STAGED_CREDENTIALS}.tmp.XXXXXX")" || return 1
-    if ! printf '%s' "$private_key_data" | base64 -d >"$staged_tmp" ||
-        ! validate_credentials_key "$staged_tmp" "$new_key_id"; then
-        rm -f -- "$staged_tmp"
-        STATE_PHASE=revoke-new
-        write_state
-        if delete_key "$new_key_name"; then
-            cleanup_state
-        fi
-        log_error "IAM returned invalid replacement credentials"
-        return 1
-    fi
-    chmod 0600 "$staged_tmp"
-    mv -f -- "$staged_tmp" "$STAGED_CREDENTIALS"
-
-    STATE_PHASE=staged
-    write_state
-    if ! install_staged_credentials; then
-        STATE_PHASE=revoke-new
-        write_state
-        if delete_key "$new_key_name"; then
-            cleanup_state
-        fi
-        log_error "Could not safely preserve and install replacement credentials"
-        return 1
-    fi
-    log_info "Installed replacement key $new_key_id; previous credentials remain available for rollback"
+    create_replacement_key
 }
 
 authenticate_rotation() {
@@ -993,7 +1252,7 @@ rotation_audit() {
         return 0
     fi
     load_state || return 1
-    if [[ "$STATE_PHASE" == "creating" ]]; then
+    if [[ "$STATE_PHASE" == "creating" || "$STATE_PHASE" == "creating-fresh" ]]; then
         candidates="$(creation_candidates)" || return 1
     fi
     if [[ "$STATE_PHASE" == "grace" ]]; then
@@ -1016,7 +1275,7 @@ rotation_audit() {
             phase: $phase,
             current_key_id: $current_key_id,
             new_key_id: $new_key_id,
-            recovery_required: ($phase == "creating"),
+            recovery_required: ($phase == "creating" or $phase == "creating-fresh"),
             candidate_key_ids: [$candidate_names[] | split("/")[-1]],
             created_at: $created_at,
             ready_at: $ready_at,
@@ -1029,7 +1288,7 @@ recover_ambiguous_creation() {
     local candidates candidate_id creation_age
 
     load_state || return 1
-    if [[ "$STATE_PHASE" != "creating" ]]; then
+    if [[ "$STATE_PHASE" != "creating" && "$STATE_PHASE" != "creating-fresh" ]]; then
         log_error "Audited recovery is valid only for an ambiguous creating state"
         return 1
     fi
@@ -1041,8 +1300,15 @@ recover_ambiguous_creation() {
     fi
     candidates="$(creation_candidates)" || return 1
     if [[ "$(jq -r 'length' <<<"$candidates")" == "0" ]]; then
-        cleanup_state
-        log_info "Audit found no key beyond the pre-create baseline; cleared ambiguous state"
+        if [[ "$STATE_PHASE" == "creating-fresh" ]]; then
+            STATE_PHASE=reconciling
+            STATE_BASELINE_KEY_NAMES='[]'
+            write_state
+            log_info "Audit found no key after fresh reconciliation; retained empty-baseline state for a safe retry"
+        else
+            cleanup_state
+            log_info "Audit found no key beyond the pre-create baseline; cleared ambiguous state"
+        fi
         return 0
     fi
     if [[ "$(jq -r 'length' <<<"$candidates")" != "1" ]]; then
@@ -1059,8 +1325,15 @@ recover_ambiguous_creation() {
     fetch_access_token || return 1
     log_info "Deleting explicitly confirmed orphan key $candidate_id from ambiguous creation"
     delete_key "$SA_RESOURCE/keys/$candidate_id" || return 1
-    cleanup_state
-    log_info "Audited recovery completed; a later run may create a new replacement"
+    if [[ "$STATE_PHASE" == "creating-fresh" ]]; then
+        STATE_PHASE=reconciling
+        STATE_BASELINE_KEY_NAMES='[]'
+        write_state
+        log_info "Audited fresh-key recovery completed; retained empty-baseline state for a safe retry"
+    else
+        cleanup_state
+        log_info "Audited recovery completed; a later run may create a new replacement"
+    fi
 }
 
 rotation_status() {
diff --git a/rootfs/home/cloud-compose/run.sh b/rootfs/home/cloud-compose/run.sh
index 251e139..470a72f 100644
--- a/rootfs/home/cloud-compose/run.sh
+++ b/rootfs/home/cloud-compose/run.sh
@@ -34,6 +34,8 @@ runtime_enabled() {
 
 durable_bootstrap_marker="/home/cloud-compose/.cloud-compose-bootstrap-complete"
 current_boot_app_init_marker="/run/cloud-compose-app-init-complete"
+fresh_filesystem_marker="${CLOUD_COMPOSE_FRESH_FILESYSTEM_MARKER:-/mnt/disks/data/.cloud-compose/fresh-filesystem}"
+fresh_filesystem_identity="${CLOUD_COMPOSE_FRESH_FILESYSTEM_IDENTITY:-fresh}"
 app_wait_seconds="${CLOUD_COMPOSE_APP_WAIT_SECONDS:-4500}"
 
 # The shared lifecycle lock must exist before the root-owned managed-runtime
@@ -63,7 +65,16 @@ bash /home/cloud-compose/converge-app-filesystems.sh
 # to run before an explicitly enabled Vault Agent has authenticated.
 run_as_cloud_compose bash /home/cloud-compose/prepare-app-sources.sh
 if [ "${CLOUD_COMPOSE_PROVIDER:-}" = "gcp" ]; then
+  if [[ ! "$fresh_filesystem_identity" =~ ^v1:gcp-disk-id:[0-9]{1,32}$ ]]; then
+    echo "GCP fresh-filesystem identity is missing or unsafe" >&2
+    exit 1
+  fi
   bash /home/cloud-compose/rotate-keys-daily.sh
+fi
+cloud_compose_consume_fresh_filesystem_marker \
+  "$fresh_filesystem_marker" "$fresh_filesystem_identity"
+sync
+if [ "${CLOUD_COMPOSE_PROVIDER:-}" = "gcp" ]; then
   if runtime_enabled "${GCP_APP_CREDENTIALS_ENABLED:-false}" ||
     runtime_enabled "${LIBOPS_INTERNAL_SERVICES_ENABLED:-false}"; then
     systemctl enable --now cloud-compose-key-rotation.timer
diff --git a/templates/cloud-init.yml b/templates/cloud-init.yml
index 79ee4d8..9f4dcb5 100644
--- a/templates/cloud-init.yml
+++ b/templates/cloud-init.yml
@@ -35,7 +35,8 @@ bootcmd:
   printf '%s' '${FILESYSTEM_PREP_SCRIPT_B64}' | base64 -d >"$filesystem_prep"
   printf '%s' '${FILESYSTEM_PERSIST_SCRIPT_B64}' | base64 -d >"$filesystem_persist"
   chmod 0600 "$filesystem_prep" "$filesystem_persist"
-  bash "$filesystem_prep" /dev/disk/by-id/google-data /mnt/disks/data
+  bash "$filesystem_prep" /dev/disk/by-id/google-data /mnt/disks/data \
+    --publish-fresh-marker ${jsonencode(FRESH_FILESYSTEM_IDENTITY)}
 
 # mount the specific Volumes disk
 - bash /run/cloud-compose-prepare-filesystem /dev/disk/by-id/google-docker-volumes /mnt/disks/volumes
@@ -88,8 +89,10 @@ ${ROOTFS_ARCHIVE_COMMAND}
 %{ for CMD in ADDITIONAL_INITCMD ~}
   ${indent(2, CMD)}
 %{ endfor ~}
-  chown cloud-compose:cloud-compose /mnt/disks/data /mnt/disks/volumes
-  chmod 0775 /mnt/disks/data /mnt/disks/volumes
+  chown root:cloud-compose /mnt/disks/data
+  chmod 1775 /mnt/disks/data
+  chown cloud-compose:cloud-compose /mnt/disks/volumes
+  chmod 0775 /mnt/disks/volumes
   install -d -m 0775 -o cloud-compose -g cloud-compose /mnt/disks/data/libops
   rm -f /home/cloud-compose/.cloud-compose-bootstrap-complete
   bash /home/cloud-compose/start-cloud-compose-bootstrap.sh