fix(backup): make Host110 runtime receipt durable

docs/proposals/AWOOOI-AG99-H110-BACKUP-RUNTIME-V9.md

@@ -0,0 +1,133 @@
+# AWOOOI Agent99 Host110 Backup Runtime V9
+
+Status: proposed; owner approval required before any production apply.
+
+This artifact supersedes V2 through V8. The following V8 identities are void
+for approval and remain audit evidence only:
+
+- candidate/revision: `0b9c0284006c8d892c4617c2bfc16869bb6637f0`
+- artifact SHA-256: `ea9537613c462a07f3155263da5ad0f444bda0ba6773b49208de31b0acdef228`
+- exact diff SHA-256: `0867d9221ecfa7bcbe6b073a6a02da401bbed5087b9f2bd7fa494f2953b2ef2e`
+- every other proposal/content hash derived from V8 candidate bytes
+
+No V8 hash, ref, review, or receipt grants production execution authority.
+
+## Scope
+
+- Agent99 runtime bundle: 19 files, guarded by one producer/consumer parity
+  test.
+- Host110 backup payload: 18 ordered files, comprising 17 backup scripts and
+  `backup-health-textfile-exporter.py`.
+- Host110 remote stage: 21 unique basenames, comprising the ordered 18-file
+  payload, fixed executor, independent verifier, and `manifest.json`.
+- Fixed target: `wooo@192.168.0.110`, dispatched by Windows99 Agent99 from an
+  exact Gitea revision and digest manifest.
+- Gitea offline backup and container lifecycle changes remain excluded;
+  `backup-gitea.sh` has no candidate diff from the production base.
+
+## V8 Defects Closed
+
+### Durable Receipt Transaction
+
+The executor no longer captures `write_receipt` through command substitution
+and cannot mask an internal failure with a later `printf`. Receipt publication
+uses an exclusive temporary file, exact full-write check, file `fsync`,
+temporary-byte SHA-256 readback, no-replace hard link, directory `fsync`, exact
+final-byte/document/SHA-256 readback, temporary unlink, and a second directory
+`fsync`. Any write, link, fsync, cleanup, or readback failure is nonzero and
+removes a final path created by that failed attempt when possible.
+
+`APPLY_COMPLETE=1` is set only after the durable verified receipt passes this
+transaction. Stage and rollback prestate are never cleared before that point.
+If verified receipt publication fails, the EXIT path performs rollback first,
+writes a distinct failure receipt, and only removes rollback prestate after
+both rollback verification and durable failure-receipt publication succeed.
+Payload files, destination directories, rollback prestate files, and the
+rollback directory are explicitly fsynced before their corresponding durable
+claim can be published.
+
+### Run-Owned Temporary Files
+
+Every apply and rollback temporary destination is tracked for the exact run.
+Rollback removes and reads back all tracked paths, including a failure between
+`install` and `mv`. Canonical content/metadata verification plus zero hidden
+residue are both required before rollback can be called verified.
+
+### Rollback Truth
+
+Receipt schema `agent99_host110_backup_runtime_deploy_receipt_v2` records
+`rollbackAttempted`, `rollbackPerformed`, `rollbackVerified`, and
+`zeroResidueVerified` independently. A partial, unknown, or residue-bearing
+rollback terminates as `rollback_unverified`; only a fully restored and
+zero-residue transaction may terminate as `failed_rolled_back`.
+
+### Producer/Consumer Parity
+
+The integration test parses both actual sources and requires the broker's 18
+payload basenames to equal the executor payload in count, exact set, and exact
+order. It separately requires executor, verifier, and `manifest.json`, yielding
+21 unique remote-stage basenames.
+
+## Fault-Injection Contract
+
+The bounded local harness executes the real executor control flow against
+isolated temporary destinations and covers:
+
+- successful apply, durable receipt, verifier, and post-receipt prestate cleanup;
+- receipt write, link, fsync, and final-readback failures;
+- apply failure after hidden-temp installation but before canonical `mv`;
+- forced temp-cleanup failure and `rollback_unverified` evidence preservation.
+
+Every receipt fault must restore exact prestate, emit no success JSON, and
+produce a durable `failed_rolled_back` receipt. A cleanup fault must preserve
+`prestate.tsv`, expose rollback attempted/performed separately, and must not
+claim rollback verification.
+
+Fault hooks require an explicit test enable flag and a transformed
+non-production destination. The canonical `/backup/scripts` executor can never
+enable them, even if a fault environment variable is present.
+
+## Exact Diff Serialization
+
+Substitute the V9 Gitea live-ref SHA for `<candidate_git_sha>` and hash the
+exact stdout bytes from this command:
+
+```sh
+git diff --binary --full-index \
+  e7e3bcf8a5a8a19c1253d73f51d9dca248af0aa3..<candidate_git_sha> |
+  shasum -a 256
+```
+
+The review message independently binds the Gitea live ref and Git SHA, this
+tracked artifact path and bytes hash, and the serialized diff hash.
+
+## Apply Effects
+
+- Windows99 Agent99 runtime promotion writes the 19 runtime files plus bounded
+  manifest and receipt state.
+- Host110 Apply writes/replaces the 18 payload paths through one exclusive,
+  digest-bound filesystem transaction.
+- Apply does not run a backup, invoke a Gitea container lifecycle, sync Google
+  Drive, prune snapshots, restart a VM or service, or change a database,
+  network, or firewall.
+
+## Validation
+
+- The exact broker payload/order and 21-file stage parity test must pass.
+- Success and all six fault-injection paths must execute and pass.
+- Focused runtime, broker, backup, parity, Ansible, shell, and Python checks
+  must pass.
+- Changed PowerShell must parse on Windows99 with zero errors when applicable.
+- Central review must independently reproduce the live ref, artifact hash,
+  exact diff hash, 19/18/21 counts, and zero `backup-gitea.sh` diff.
+
+## Rollback
+
+- Source rollback reverts the exact V9 diff to
+  `e7e3bcf8a5a8a19c1253d73f51d9dca248af0aa3` without force-pushing.
+- Agent99 rollback restores the prior runtime files and manifest and verifies
+  prior hashes independently.
+- Host110 rollback restores all prior payload content, SHA-256, uid, gid, and
+  mode under the exclusive lock and proves zero run-owned temporary residue.
+- Any canonical, metadata, receipt, or residue mismatch terminates as
+  `rollback_unverified`; rollback prestate remains available for diagnosis.

scripts/backup/host110-backup-runtime-executor.sh

@@ -44,13 +44,19 @@ EXECUTOR_DIGEST=""
 VERIFIER_DIGEST=""
 STAGE_DIR=""
 ROLLBACK_DIR=""
+RECEIPT_PATH=""
 APPLY_STARTED=0
 APPLY_COMPLETE=0
+ROLLBACK_ATTEMPTED=0
+ROLLBACK_PERFORMED=0
 ROLLBACK_VERIFIED=0
+RUN_TEMP_RESIDUE_VERIFIED=0
+FAULT_INJECTION_ENABLED=0
 declare -A FILE_EXISTED=()
 declare -A PREVIOUS_DIGEST=()
 declare -A PREVIOUS_METADATA=()
 declare -A EXPECTED_DIGEST=()
+declare -A RUN_OWNED_TEMP_PATHS=()
 
 usage() {
   printf '%s\n' \
@@ -96,6 +102,11 @@ case "$MODE" in check|apply|verify) ;; *) fail "invalid_mode" ;; esac
 [[ "$RUN_ID" =~ ^[A-Za-z0-9][A-Za-z0-9._-]{0,95}$ ]] || fail "invalid_run_id"
 expected_source_stage="/tmp/agent99-host110-backup-runtime-${RUN_ID}"
 [ "$SOURCE_STAGE" = "$expected_source_stage" ] || fail "invalid_source_stage"
+RECEIPT_PATH="$STATUS_ROOT/agent99-backup-runtime-deploy-${RUN_ID}.json"
+if [ "${HOST110_BACKUP_RUNTIME_ENABLE_TEST_FAULTS:-}" = "1" ] \
+  && [ "$DEST_ROOT" != "/backup/scripts" ]; then
+  FAULT_INJECTION_ENABLED=1
+fi
 
 for command_name in awk bash cp dirname flock grep hostname id install mv pgrep python3 readlink rm sha256sum stat timeout tr; do
   command -v "$command_name" >/dev/null 2>&1 || fail "required_command_missing_${command_name}"
@@ -174,10 +185,10 @@ working_digest() {
   || fail "verifier_identity_failed"
 
 validate_file() {
-  local path="$1"
-  case "$path" in
-    *.sh) bash -n "$path" ;;
-    *.py) python3 -c 'import ast, pathlib, sys; ast.parse(pathlib.Path(sys.argv[1]).read_text(encoding="utf-8"))' "$path" ;;
+  local validation_path="$1"
+  case "$validation_path" in
+    *.sh) bash -n "$validation_path" ;;
+    *.py) python3 -c 'import ast, pathlib, sys; ast.parse(pathlib.Path(sys.argv[1]).read_text(encoding="utf-8"))' "$validation_path" ;;
     *) return 64 ;;
   esac
 }
@@ -222,12 +233,75 @@ verify_destination() {
   done
 }
 
+fsync_paths_and_parents() {
+  python3 - "$@" <<'PY'
+import os
+import stat
+import sys
+from pathlib import Path
+
+parents = set()
+for raw_path in sys.argv[1:]:
+    path = Path(raw_path)
+    parents.add(path.parent)
+    if not os.path.lexists(path):
+        continue
+    if path.is_symlink() or not path.is_file():
+        raise SystemExit(76)
+    flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
+    descriptor = os.open(path, flags)
+    try:
+        if not stat.S_ISREG(os.fstat(descriptor).st_mode):
+            raise SystemExit(76)
+        os.fsync(descriptor)
+    finally:
+        os.close(descriptor)
+for parent in sorted(parents, key=str):
+    if parent.is_symlink() or not parent.is_dir():
+        raise SystemExit(76)
+    flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0)
+    descriptor = os.open(parent, flags)
+    try:
+        os.fsync(descriptor)
+    finally:
+        os.close(descriptor)
+PY
+}
+
+fsync_destination_state() {
+  local name dest_path
+  local -a fsync_targets=()
+  for name in "${PAYLOAD_FILES[@]}"; do
+    dest_path="$(destination_path "$name")"
+    fsync_targets+=("$dest_path")
+  done
+  fsync_paths_and_parents "${fsync_targets[@]}"
+}
+
+fsync_rollback_prestate() {
+  local name
+  local -a fsync_targets=("$ROLLBACK_DIR/prestate.tsv")
+  for name in "${PAYLOAD_FILES[@]}"; do
+    if [ "${FILE_EXISTED[$name]:-}" = "1" ]; then
+      fsync_targets+=("$ROLLBACK_DIR/$name")
+    fi
+  done
+  fsync_paths_and_parents "${fsync_targets[@]}"
+}
+
 write_receipt() {
-  local status="$1"
-  local rollback_verified="$2"
-  local receipt_path="$STATUS_ROOT/agent99-backup-runtime-deploy-${RUN_ID}.json"
-  install -d -m 700 "$STATUS_ROOT"
-  python3 - "$receipt_path" "$status" "$SOURCE_REVISION" "$SOURCE_HEAD" "$RUN_ID" "$rollback_verified" "$VERIFIER_DIGEST" <<'PY'
+  local receipt_status="$1"
+  local rollback_attempted="$2"
+  local rollback_performed="$3"
+  local rollback_verified="$4"
+  local zero_residue_verified="$5"
+  install -d -m 700 "$STATUS_ROOT" || return 1
+  [ -d "$STATUS_ROOT" ] && [ ! -L "$STATUS_ROOT" ] || return 1
+  [ "$(stat -c '%U:%G:%a' "$STATUS_ROOT")" = "wooo:wooo:700" ] || return 1
+  if ! python3 - "$RECEIPT_PATH" "$receipt_status" "$SOURCE_REVISION" "$SOURCE_HEAD" "$RUN_ID" \
+    "$rollback_attempted" "$rollback_performed" "$rollback_verified" \
+    "$zero_residue_verified" "$VERIFIER_DIGEST" "$FAULT_INJECTION_ENABLED" <<'PY'
+import hashlib
 import json
 import os
 import sys
@@ -235,64 +309,176 @@ import time
 from pathlib import Path
 
 path = Path(sys.argv[1])
+status = sys.argv[2]
+rollback_attempted = sys.argv[6] == "1"
+rollback_performed = sys.argv[7] == "1"
+rollback_verified = sys.argv[8] == "1"
+zero_residue_verified = sys.argv[9] == "1"
+if status not in {"verified", "failed_rolled_back", "rollback_unverified"}:
+    raise SystemExit(78)
+if status == "verified" and (rollback_attempted or rollback_performed or rollback_verified or not zero_residue_verified):
+    raise SystemExit(78)
+if status == "failed_rolled_back" and not (
+    rollback_attempted and rollback_performed and rollback_verified and zero_residue_verified
+):
+    raise SystemExit(78)
+if status == "rollback_unverified" and (not rollback_attempted or rollback_verified):
+    raise SystemExit(78)
 document = {
-    "schemaVersion": "agent99_host110_backup_runtime_deploy_receipt_v1",
-    "status": sys.argv[2],
+    "schemaVersion": "agent99_host110_backup_runtime_deploy_receipt_v2",
+    "status": status,
     "sourceRevision": sys.argv[3],
     "sourceHead": sys.argv[4],
     "runId": sys.argv[5],
     "fileCount": 18,
     "backupScriptFileCount": 17,
     "backupHealthExporterIncluded": True,
-    "rollbackPerformed": sys.argv[6] == "1",
-    "rollbackVerified": sys.argv[6] == "1",
-    "verifierSha256": sys.argv[7],
+    "rollbackAttempted": rollback_attempted,
+    "rollbackPerformed": rollback_performed,
+    "rollbackVerified": rollback_verified,
+    "zeroResidueVerified": zero_residue_verified,
+    "verifierSha256": sys.argv[10],
     "executorHost": "192.168.0.110",
     "productionServiceRestarted": False,
     "secretValuesRead": False,
     "writtenAt": int(time.time()),
 }
+payload = (json.dumps(document, ensure_ascii=True, sort_keys=True, separators=(",", ":")) + "\n").encode("utf-8")
+expected_digest = hashlib.sha256(payload).hexdigest()
 temporary = path.with_name(f".{path.name}.tmp-{os.getpid()}")
-temporary.write_text(json.dumps(document, ensure_ascii=True, sort_keys=True) + "\n", encoding="utf-8")
-os.chmod(temporary, 0o600)
+fault_enabled = sys.argv[11] == "1"
+fault = (
+    os.environ.get("HOST110_BACKUP_RUNTIME_TEST_RECEIPT_FAULT", "")
+    if status == "verified" and fault_enabled
+    else ""
+)
+if fault not in {"", "write", "link", "fsync", "readback"}:
+    raise SystemExit(78)
+created_final = False
+directory_fd = None
 try:
+    if os.path.lexists(path) or os.path.lexists(temporary):
+        raise FileExistsError(path)
+    if fault == "write":
+        raise OSError("fault_injected_receipt_write")
+    descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+    with os.fdopen(descriptor, "wb") as handle:
+        if handle.write(payload) != len(payload):
+            raise OSError("receipt_short_write")
+        handle.flush()
+        if fault == "fsync":
+            raise OSError("fault_injected_receipt_fsync")
+        os.fsync(handle.fileno())
+    os.chmod(temporary, 0o600)
+    temporary_readback = temporary.read_bytes()
+    if temporary_readback != payload or hashlib.sha256(temporary_readback).hexdigest() != expected_digest:
+        raise OSError("receipt_temporary_readback_failed")
+    if fault == "link":
+        raise OSError("fault_injected_receipt_link")
     os.link(temporary, path)
+    created_final = True
+    directory_flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)
+    directory_fd = os.open(path.parent, directory_flags)
+    os.fsync(directory_fd)
+    if fault == "readback":
+        raise OSError("fault_injected_receipt_readback")
+    readback = path.read_bytes()
+    if readback != payload or hashlib.sha256(readback).hexdigest() != expected_digest:
+        raise OSError("receipt_final_readback_failed")
+    if json.loads(readback.decode("utf-8")) != document:
+        raise OSError("receipt_document_readback_failed")
+    temporary.unlink()
+    os.fsync(directory_fd)
+except BaseException:
+    try:
+        if created_final:
+            path.unlink(missing_ok=True)
+        temporary.unlink(missing_ok=True)
+        if directory_fd is None and path.parent.is_dir():
+            directory_flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)
+            directory_fd = os.open(path.parent, directory_flags)
+        if directory_fd is not None:
+            os.fsync(directory_fd)
+    finally:
+        raise
 finally:
-    temporary.unlink(missing_ok=True)
+    if directory_fd is not None:
+        os.close(directory_fd)
 PY
-  printf '%s' "$receipt_path"
+  then
+    return 1
+  fi
+  [ -f "$RECEIPT_PATH" ] && [ ! -L "$RECEIPT_PATH" ] || return 1
+  return 0
 }
 
 record_prestate() {
   local name dest_path digest metadata
-  : > "$ROLLBACK_DIR/prestate.tsv"
-  chmod 600 "$ROLLBACK_DIR/prestate.tsv"
+  : > "$ROLLBACK_DIR/prestate.tsv" || return 74
+  chmod 600 "$ROLLBACK_DIR/prestate.tsv" || return 74
   for name in "${PAYLOAD_FILES[@]}"; do
     dest_path="$(destination_path "$name")"
     if [ -f "$dest_path" ] && [ ! -L "$dest_path" ]; then
       [ "$(stat -c '%U:%G' "$dest_path")" = "wooo:wooo" ] || return 71
       FILE_EXISTED[$name]=1
-      digest="$(working_digest "$dest_path")"
-      metadata="$(stat -c '%u:%g:%a' "$dest_path")"
+      digest="$(working_digest "$dest_path")" || return 72
+      metadata="$(stat -c '%u:%g:%a' "$dest_path")" || return 72
       PREVIOUS_DIGEST[$name]="$digest"
       PREVIOUS_METADATA[$name]="$metadata"
-      printf '%s\t1\t%s\t%s\n' "$name" "$digest" "$metadata" >> "$ROLLBACK_DIR/prestate.tsv"
-      cp -p -- "$dest_path" "$ROLLBACK_DIR/$name"
+      printf '%s\t1\t%s\t%s\n' "$name" "$digest" "$metadata" >> "$ROLLBACK_DIR/prestate.tsv" || return 72
+      cp -p -- "$dest_path" "$ROLLBACK_DIR/$name" || return 72
       [ "$(working_digest "$ROLLBACK_DIR/$name")" = "$digest" ] || return 72
     elif [ ! -e "$dest_path" ]; then
       FILE_EXISTED[$name]=0
       PREVIOUS_DIGEST[$name]="-"
       PREVIOUS_METADATA[$name]="-"
-      printf '%s\t0\t-\t-\n' "$name" >> "$ROLLBACK_DIR/prestate.tsv"
+      printf '%s\t0\t-\t-\n' "$name" >> "$ROLLBACK_DIR/prestate.tsv" || return 72
     else
       return 73
     fi
   done
+  fsync_rollback_prestate || return 75
+}
+
+register_run_owned_temp() {
+  RUN_OWNED_TEMP_PATHS["$1"]=1
+}
+
+verify_run_owned_temp_absence() {
+  local temporary
+  for temporary in "${!RUN_OWNED_TEMP_PATHS[@]}"; do
+    [ ! -e "$temporary" ] && [ ! -L "$temporary" ] || return 1
+  done
+}
+
+cleanup_run_owned_temps() {
+  local temporary failed=0 preserved=0
+  local fault=""
+  if [ "$FAULT_INJECTION_ENABLED" -eq 1 ]; then
+    fault="${HOST110_BACKUP_RUNTIME_TEST_ROLLBACK_FAULT:-}"
+  fi
+  for temporary in "${!RUN_OWNED_TEMP_PATHS[@]}"; do
+    if [ "$fault" = "preserve_first_temp" ] \
+      && [ "$preserved" -eq 0 ] \
+      && { [ -e "$temporary" ] || [ -L "$temporary" ]; }; then
+      preserved=1
+      failed=1
+      continue
+    fi
+    rm -f -- "$temporary" || failed=1
+  done
+  verify_run_owned_temp_absence || failed=1
+  [ "$failed" -eq 0 ]
 }
 
 rollback_transaction() {
   local name dest_path destination_parent backup_path temporary failed=0
+  ROLLBACK_ATTEMPTED=1
+  ROLLBACK_PERFORMED=0
+  ROLLBACK_VERIFIED=0
+  RUN_TEMP_RESIDUE_VERIFIED=0
   set +e
+  cleanup_run_owned_temps || failed=1
   for name in "${PAYLOAD_FILES[@]}"; do
     [ -n "${FILE_EXISTED[$name]+x}" ] || { failed=1; continue; }
     dest_path="$(destination_path "$name")"
@@ -300,13 +486,22 @@ rollback_transaction() {
       backup_path="$ROLLBACK_DIR/$name"
       destination_parent="$(dirname "$dest_path")"
       temporary="$destination_parent/.${name}.agent99-rollback-${RUN_ID}"
-      cp -p -- "$backup_path" "$temporary" \
-        && mv -f -- "$temporary" "$dest_path" \
-        || failed=1
+      register_run_owned_temp "$temporary"
+      if cp -p -- "$backup_path" "$temporary" && mv -f -- "$temporary" "$dest_path"; then
+        ROLLBACK_PERFORMED=1
+      else
+        failed=1
+      fi
     else
-      rm -f -- "$dest_path" || failed=1
+      if rm -f -- "$dest_path"; then
+        ROLLBACK_PERFORMED=1
+      else
+        failed=1
+      fi
     fi
   done
+  cleanup_run_owned_temps || failed=1
+  fsync_destination_state || failed=1
   for name in "${PAYLOAD_FILES[@]}"; do
     dest_path="$(destination_path "$name")"
     if [ "${FILE_EXISTED[$name]:-}" = "1" ]; then
@@ -320,8 +515,14 @@ rollback_transaction() {
       failed=1
     fi
   done
+  if verify_run_owned_temp_absence; then
+    RUN_TEMP_RESIDUE_VERIFIED=1
+  else
+    failed=1
+    RUN_TEMP_RESIDUE_VERIFIED=0
+  fi
   set -e
-  if [ "$failed" -eq 0 ]; then
+  if [ "$failed" -eq 0 ] && [ "$ROLLBACK_PERFORMED" -eq 1 ] && [ "$RUN_TEMP_RESIDUE_VERIFIED" -eq 1 ]; then
     ROLLBACK_VERIFIED=1
     return 0
   fi
@@ -330,20 +531,28 @@ rollback_transaction() {
 }
 
 cleanup() {
-  local status=$?
+  local exit_status=$?
   local rollback_status="rollback_unverified"
+  local failure_receipt_written=0
   trap - EXIT HUP INT TERM
   if [ "$APPLY_STARTED" -eq 1 ] && [ "$APPLY_COMPLETE" -ne 1 ]; then
     if rollback_transaction; then
       rollback_status="failed_rolled_back"
     fi
-    write_receipt "$rollback_status" "$ROLLBACK_VERIFIED" >/dev/null || true
+    if write_receipt "$rollback_status" "$ROLLBACK_ATTEMPTED" "$ROLLBACK_PERFORMED" \
+      "$ROLLBACK_VERIFIED" "$RUN_TEMP_RESIDUE_VERIFIED"; then
+      failure_receipt_written=1
+    fi
   fi
-  [ -z "$STAGE_DIR" ] || rm -rf -- "$STAGE_DIR"
-  if [ "$ROLLBACK_VERIFIED" -eq 1 ]; then
-    [ -z "$ROLLBACK_DIR" ] || rm -rf -- "$ROLLBACK_DIR"
+  if [ -n "$STAGE_DIR" ]; then
+    rm -rf -- "$STAGE_DIR" || true
   fi
-  exit "$status"
+  if [ "$ROLLBACK_VERIFIED" -eq 1 ] && [ "$failure_receipt_written" -eq 1 ]; then
+    if [ -n "$ROLLBACK_DIR" ]; then
+      rm -rf -- "$ROLLBACK_DIR" || true
+    fi
+  fi
+  exit "$exit_status"
 }
 
 validate_source || fail "source_validation_failed"
@@ -359,7 +568,7 @@ if [ "$MODE" = "verify" ]; then
   exit 0
 fi
 
-[ ! -e "$STATUS_ROOT/agent99-backup-runtime-deploy-${RUN_ID}.json" ] || fail "run_identity_receipt_exists"
+[ ! -e "$RECEIPT_PATH" ] && [ ! -L "$RECEIPT_PATH" ] || fail "run_identity_receipt_exists"
 STAGE_DIR="$STAGE_ROOT/$RUN_ID"
 ROLLBACK_DIR="$ROLLBACK_ROOT/$RUN_ID"
 [ ! -e "$STAGE_DIR" ] || fail "run_identity_stage_exists"
@@ -399,10 +608,17 @@ for name in "${PAYLOAD_FILES[@]}"; do
   dest_path="$(destination_path "$name")"
   destination_parent="$(dirname "$dest_path")"
   temporary="$destination_parent/.${name}.agent99-${RUN_ID}"
+  register_run_owned_temp "$temporary"
   install -m 755 "$STAGE_DIR/$name" "$temporary"
+  if [ "$FAULT_INJECTION_ENABLED" -eq 1 ] \
+    && [ "${HOST110_BACKUP_RUNTIME_TEST_APPLY_FAULT:-}" = "after_temp_install" ] \
+    && [ "$name" = "${PAYLOAD_FILES[0]}" ]; then
+    fail "fault_injected_after_temp_install"
+  fi
   mv -f -- "$temporary" "$dest_path"
 done
 
+fsync_destination_state || fail "post_apply_durability_failed"
 verify_destination || fail "post_apply_verification_failed"
 verifier_result="$(BACKUP_RUNTIME_DEPLOY_CONTEXT=1 python3 "$SOURCE_STAGE/verify-host110-backup-runtime.py" \
   --manifest "$SOURCE_STAGE/manifest.json" \
@@ -433,13 +649,26 @@ if not (
     raise SystemExit(1)
 PY
 
+verify_run_owned_temp_absence || fail "post_apply_temp_residue_detected"
+RUN_TEMP_RESIDUE_VERIFIED=1
+if ! write_receipt "verified" 0 0 0 "$RUN_TEMP_RESIDUE_VERIFIED"; then
+  fail "durable_receipt_failed"
+fi
+receipt_path="$RECEIPT_PATH"
 APPLY_COMPLETE=1
-receipt_path="$(write_receipt "verified" 0)"
-rm -rf -- "$STAGE_DIR" "$ROLLBACK_DIR"
+trap - EXIT HUP INT TERM
+stage_cleanup_verified=0
+rollback_prestate_cleanup_verified=0
+if rm -rf -- "$STAGE_DIR" && [ ! -e "$STAGE_DIR" ] && [ ! -L "$STAGE_DIR" ]; then
+  stage_cleanup_verified=1
+fi
+if rm -rf -- "$ROLLBACK_DIR" && [ ! -e "$ROLLBACK_DIR" ] && [ ! -L "$ROLLBACK_DIR" ]; then
+  rollback_prestate_cleanup_verified=1
+fi
 STAGE_DIR=""
 ROLLBACK_DIR=""
-trap - EXIT HUP INT TERM
-python3 - "$SOURCE_REVISION" "$SOURCE_HEAD" "$receipt_path" "$verifier_result" <<'PY'
+python3 - "$SOURCE_REVISION" "$SOURCE_HEAD" "$receipt_path" "$verifier_result" \
+  "$stage_cleanup_verified" "$rollback_prestate_cleanup_verified" <<'PY'
 import json
 import sys
 
@@ -452,7 +681,12 @@ print(json.dumps({
     "fileCount": 18,
     "backupScriptFileCount": 17,
     "backupHealthExporterIncluded": True,
+    "rollbackAttempted": False,
     "rollbackPerformed": False,
+    "rollbackVerified": False,
+    "zeroResidueVerified": True,
+    "stageCleanupVerified": sys.argv[5] == "1",
+    "rollbackPrestateCleanupVerified": sys.argv[6] == "1",
     "receipt": sys.argv[3],
     "verifier": json.loads(sys.argv[4]),
 }, ensure_ascii=True, sort_keys=True))

scripts/reboot-recovery/tests/test_agent99_host110_backup_runtime_broker.py

@@ -1,10 +1,15 @@
 from __future__ import annotations
 
 import base64
+import grp
 import hashlib
 import importlib.util
 import json
+import os
+import pwd
 import re
+import shutil
+import subprocess
 import sys
 from pathlib import Path
 
@@ -66,6 +71,216 @@ def _powershell_array(source: str, variable: str) -> tuple[str, ...]:
     return tuple(re.findall(r'"([^"]+)"', match.group(1)))
 
 
+def _executor_payload_order(source: str) -> tuple[str, ...]:
+    match = re.search(r"readonly -a RUNTIME_FILES=\((.*?)\n\)", source, re.DOTALL)
+    assert match is not None
+    runtime_files = tuple(re.findall(r"^\s+([A-Za-z0-9_.-]+)\s*$", match.group(1), re.MULTILINE))
+    exporter = re.search(r'^readonly EXPORTER_FILE="([^"]+)"$', source, re.MULTILINE)
+    assert exporter is not None
+    return (*runtime_files, exporter.group(1))
+
+
+def _write_executable(path: Path, source: str) -> None:
+    path.write_text(source, encoding="utf-8")
+    path.chmod(0o755)
+
+
+def _build_executor_harness(tmp_path: Path, run_id: str) -> dict[str, object]:
+    destination = tmp_path / "backup-scripts"
+    exporter_root = tmp_path / "exporter"
+    status_root = tmp_path / "status"
+    stage_root = tmp_path / "stage"
+    rollback_root = tmp_path / "rollback"
+    source_stage = tmp_path / f"source-{run_id}"
+    lock_path = tmp_path / "runtime.lock"
+    fake_bin = tmp_path / "fake-bin"
+    for path in (destination, exporter_root, source_stage, fake_bin):
+        path.mkdir(parents=True)
+
+    user_name = pwd.getpwuid(os.getuid()).pw_name
+    group_name = grp.getgrgid(os.getgid()).gr_name
+    shell_path = shutil.which("bash") or "/bin/bash"
+    shell_version = subprocess.run(
+        [shell_path, "-c", 'printf "%s" "${BASH_VERSINFO[0]:-0}"'],
+        check=False,
+        capture_output=True,
+        text=True,
+    ).stdout
+    use_zsh_adapter = not shell_version.isdigit() or int(shell_version) < 4
+    if use_zsh_adapter:
+        shell_path = shutil.which("zsh") or "/bin/zsh"
+    executor_source = EXECUTOR.read_text(encoding="utf-8")
+    replacements = {
+        'readonly EXPECTED_USER="wooo"': f'readonly EXPECTED_USER="{user_name}"',
+        'readonly DEST_ROOT="/backup/scripts"': f'readonly DEST_ROOT="{destination}"',
+        'readonly EXPORTER_ROOT="/home/wooo/scripts"': f'readonly EXPORTER_ROOT="{exporter_root}"',
+        'readonly STATUS_ROOT="/backup/status"': f'readonly STATUS_ROOT="{status_root}"',
+        'readonly STAGE_ROOT="/backup/.agent99-backup-runtime-stage"': f'readonly STAGE_ROOT="{stage_root}"',
+        'readonly ROLLBACK_ROOT="/backup/.agent99-backup-runtime-rollback"': f'readonly ROLLBACK_ROOT="{rollback_root}"',
+        'readonly LOCK_PATH="/tmp/agent99-host110-backup-runtime.lock"': f'readonly LOCK_PATH="{lock_path}"',
+        'expected_source_stage="/tmp/agent99-host110-backup-runtime-${RUN_ID}"': f'expected_source_stage="{source_stage}"',
+        '"wooo:wooo:755"': f'"{user_name}:{group_name}:755"',
+        '"wooo:wooo:700"': f'"{user_name}:{group_name}:700"',
+        '"wooo:wooo"': f'"{user_name}:{group_name}"',
+    }
+    for before, after in replacements.items():
+        assert before in executor_source
+        executor_source = executor_source.replace(before, after)
+    if use_zsh_adapter:
+        executor_source = executor_source.replace(
+            '"${!RUN_OWNED_TEMP_PATHS[@]}"',
+            '"${(@k)RUN_OWNED_TEMP_PATHS}"',
+        ).replace(
+            'RUN_OWNED_TEMP_PATHS["$1"]=1',
+            'RUN_OWNED_TEMP_PATHS[$1]=1',
+        ).replace('${PAYLOAD_FILES[0]}', '${PAYLOAD_FILES[1]}')
+    executor_path = source_stage / EXECUTOR.name
+    _write_executable(executor_path, executor_source)
+
+    verifier_source = VERIFIER.read_text(encoding="utf-8")
+    verifier_replacements = {
+        'EXPECTED_DESTINATION = Path("/backup/scripts")': f"EXPECTED_DESTINATION = Path({str(destination)!r})",
+        'EXPECTED_EXPORTER_DESTINATION = Path("/home/wooo/scripts/backup-health-textfile-exporter.py")': (
+            f"EXPECTED_EXPORTER_DESTINATION = Path({str(exporter_root / 'backup-health-textfile-exporter.py')!r})"
+        ),
+        'RUNTIME_LOCK = Path("/tmp/agent99-host110-backup-runtime.lock")': f"RUNTIME_LOCK = Path({str(lock_path)!r})",
+    }
+    for before, after in verifier_replacements.items():
+        assert before in verifier_source
+        verifier_source = verifier_source.replace(before, after)
+    verifier_path = source_stage / VERIFIER.name
+    _write_executable(verifier_path, verifier_source)
+
+    payload_order = _executor_payload_order(executor_source)
+    rows = []
+    before_state: dict[str, tuple[bytes, int]] = {}
+    for name in payload_order:
+        source = (
+            ROOT / "scripts" / "ops" / name
+            if name == "backup-health-textfile-exporter.py"
+            else ROOT / "scripts" / "backup" / name
+        )
+        staged = source_stage / name
+        shutil.copy2(source, staged)
+        staged.chmod(0o755)
+        rows.append({"name": name, "sha256": _sha256(staged)})
+        target = exporter_root / name if name == "backup-health-textfile-exporter.py" else destination / name
+        target.write_text(f"prestate:{name}\n", encoding="utf-8")
+        target.chmod(0o755)
+        before_state[name] = (target.read_bytes(), target.stat().st_mode & 0o777)
+
+    source_revision = "a" * 40
+    manifest = {
+        "schemaVersion": "agent99_host110_backup_runtime_source_manifest_v1",
+        "sourceRevision": source_revision,
+        "sourceHead": "b" * 40,
+        "runId": run_id,
+        "executorSha256": _sha256(executor_path),
+        "verifierSha256": _sha256(verifier_path),
+        "files": rows,
+    }
+    (source_stage / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8")
+
+    _write_executable(fake_bin / "hostname", "#!/bin/sh\nprintf '%s\\n' '192.168.0.110'\n")
+    _write_executable(fake_bin / "pgrep", "#!/bin/sh\nexit 1\n")
+    _write_executable(fake_bin / "flock", "#!/bin/sh\nexit 0\n")
+    _write_executable(fake_bin / "timeout", "#!/bin/sh\nexit 0\n")
+    _write_executable(
+        fake_bin / "readlink",
+        """#!/usr/bin/env python3
+import pathlib
+import sys
+if len(sys.argv) != 3 or sys.argv[1] != "-f":
+    raise SystemExit(2)
+print(pathlib.Path(sys.argv[2]).resolve(strict=True))
+""",
+    )
+    _write_executable(
+        fake_bin / "stat",
+        """#!/usr/bin/env python3
+import grp
+import os
+import pwd
+import sys
+if len(sys.argv) != 4 or sys.argv[1] != "-c":
+    raise SystemExit(2)
+row = os.stat(sys.argv[3])
+mode = format(row.st_mode & 0o777, "o")
+values = {
+    "%U:%G:%a": f"{pwd.getpwuid(row.st_uid).pw_name}:{grp.getgrgid(row.st_gid).gr_name}:{mode}",
+    "%U:%G": f"{pwd.getpwuid(row.st_uid).pw_name}:{grp.getgrgid(row.st_gid).gr_name}",
+    "%u:%g:%a": f"{row.st_uid}:{row.st_gid}:{mode}",
+}
+if sys.argv[2] not in values:
+    raise SystemExit(2)
+print(values[sys.argv[2]])
+""",
+    )
+    return {
+        "executor": executor_path,
+        "source_stage": source_stage,
+        "source_revision": source_revision,
+        "run_id": run_id,
+        "destination": destination,
+        "exporter_root": exporter_root,
+        "status_root": status_root,
+        "stage_root": stage_root,
+        "rollback_root": rollback_root,
+        "fake_bin": fake_bin,
+        "shell": shell_path,
+        "use_zsh_adapter": use_zsh_adapter,
+        "payload_order": payload_order,
+        "before_state": before_state,
+    }
+
+
+def _run_executor_harness(harness: dict[str, object], **extra_env: str) -> subprocess.CompletedProcess[str]:
+    environment = os.environ.copy()
+    for name in (
+        "HOST110_BACKUP_RUNTIME_TEST_RECEIPT_FAULT",
+        "HOST110_BACKUP_RUNTIME_TEST_APPLY_FAULT",
+        "HOST110_BACKUP_RUNTIME_TEST_ROLLBACK_FAULT",
+    ):
+        environment.pop(name, None)
+    environment["HOST110_BACKUP_RUNTIME_ENABLE_TEST_FAULTS"] = "1"
+    environment.update(extra_env)
+    environment["PATH"] = f"{harness['fake_bin']}:{environment['PATH']}"
+    command = [str(harness["shell"])]
+    if harness["use_zsh_adapter"]:
+        environment["ZDOTDIR"] = str(harness["fake_bin"])
+        command.append("-f")
+    command.extend(
+        [
+            str(harness["executor"]),
+            "--mode",
+            "apply",
+            "--source-revision",
+            str(harness["source_revision"]),
+            "--run-id",
+            str(harness["run_id"]),
+            "--source-stage",
+            str(harness["source_stage"]),
+        ]
+    )
+    return subprocess.run(
+        command,
+        check=False,
+        capture_output=True,
+        text=True,
+        env=environment,
+    )
+
+
+def _assert_prestate_restored(harness: dict[str, object]) -> None:
+    destination = Path(harness["destination"])
+    exporter_root = Path(harness["exporter_root"])
+    before_state = harness["before_state"]
+    assert isinstance(before_state, dict)
+    for name in harness["payload_order"]:
+        target = exporter_root / name if name == "backup-health-textfile-exporter.py" else destination / name
+        assert (target.read_bytes(), target.stat().st_mode & 0o777) == before_state[name]
+
+
 def test_windows99_broker_fetches_exact_gitea_revision_and_is_bounded() -> None:
     source = BROKER.read_text(encoding="utf-8")
 
@@ -96,6 +311,8 @@ def test_broker_scp_basenames_match_executor_source_stage_contract() -> None:
     executor = EXECUTOR.read_text(encoding="utf-8")
 
     runtime_paths = _powershell_array(broker, "RuntimeRelativePaths")
+    broker_payload_basenames = tuple(Path(path).name for path in runtime_paths)
+    executor_payload_basenames = _executor_payload_order(executor)
     executor_path = _powershell_string(broker, "ExecutorRelativePath")
     verifier_path = _powershell_string(broker, "VerifierRelativePath")
     manifest_match = re.search(
@@ -105,9 +322,15 @@ def test_broker_scp_basenames_match_executor_source_stage_contract() -> None:
     assert manifest_match is not None
     manifest_name = manifest_match.group(1)
 
-    staged_basenames = tuple(
-        Path(path).name
-        for path in (*runtime_paths, executor_path, verifier_path, manifest_name)
+    assert len(broker_payload_basenames) == 18
+    assert len(executor_payload_basenames) == 18
+    assert set(broker_payload_basenames) == set(executor_payload_basenames)
+    assert broker_payload_basenames == executor_payload_basenames
+    staged_basenames = (
+        *broker_payload_basenames,
+        Path(executor_path).name,
+        Path(verifier_path).name,
+        manifest_name,
     )
     assert len(staged_basenames) == 21
     assert len(set(staged_basenames)) == 21
@@ -156,10 +379,30 @@ def test_host110_executor_is_manifest_bound_atomic_and_fail_closed() -> None:
     assert source.index("for name in \"${PAYLOAD_FILES[@]}\"") < source.index("mv -f -- \"$temporary\"")
     assert 'rollback_unverified' in source
     assert 'failed_rolled_back' in source
+    assert 'ROLLBACK_ATTEMPTED=1' in source
+    assert 'ROLLBACK_PERFORMED=1' in source
+    assert '"rollbackAttempted": rollback_attempted' in source
+    assert '"rollbackPerformed": rollback_performed' in source
+    assert '"rollbackVerified": rollback_verified' in source
+    assert '"zeroResidueVerified": zero_residue_verified' in source
+    assert 'cleanup_run_owned_temps' in source
+    assert 'verify_run_owned_temp_absence' in source
+    assert 'HOST110_BACKUP_RUNTIME_TEST_RECEIPT_FAULT' in source
+    assert 'HOST110_BACKUP_RUNTIME_TEST_APPLY_FAULT' in source
+    assert 'HOST110_BACKUP_RUNTIME_TEST_ROLLBACK_FAULT' in source
+    assert 'HOST110_BACKUP_RUNTIME_ENABLE_TEST_FAULTS' in source
+    assert '[ "$DEST_ROOT" != "/backup/scripts" ]' in source
+    assert 'fsync_rollback_prestate || return 75' in source
+    assert 'fsync_destination_state || fail "post_apply_durability_failed"' in source
     assert 'working_digest "$dest_path"' in source
     assert 'PREVIOUS_METADATA' in source
     assert "stat -c '%u:%g:%a'" in source
     assert 'os.link(temporary, path)' in source
+    assert 'os.fsync(handle.fileno())' in source
+    assert 'os.fsync(directory_fd)' in source
+    assert 'hashlib.sha256(readback).hexdigest()' in source
+    assert source.index('write_receipt "verified"') < source.index("\nAPPLY_COMPLETE=1")
+    assert source.index("\nAPPLY_COMPLETE=1") < source.rindex('rm -rf -- "$ROLLBACK_DIR"')
     assert 'run_identity_receipt_exists' in source
     assert 'run_identity_stage_exists' in source
     assert 'run_identity_rollback_exists' in source
@@ -174,6 +417,119 @@ def test_host110_executor_is_manifest_bound_atomic_and_fail_closed() -> None:
     assert "systemctl " not in source
 
 
+def test_apply_success_publishes_durable_receipt_before_prestate_cleanup(tmp_path: Path) -> None:
+    run_id = "apply-success"
+    harness = _build_executor_harness(tmp_path, run_id)
+
+    result = _run_executor_harness(harness)
+
+    assert result.returncode == 0, result.stderr
+    output = json.loads(result.stdout)
+    assert output["ok"] is True
+    assert output["rollbackAttempted"] is False
+    assert output["rollbackPerformed"] is False
+    assert output["rollbackVerified"] is False
+    assert output["zeroResidueVerified"] is True
+    assert output["stageCleanupVerified"] is True
+    assert output["rollbackPrestateCleanupVerified"] is True
+    receipt_path = Path(harness["status_root"]) / f"agent99-backup-runtime-deploy-{run_id}.json"
+    receipt = json.loads(receipt_path.read_text(encoding="utf-8"))
+    assert receipt["status"] == "verified"
+    assert receipt["rollbackAttempted"] is False
+    assert receipt["rollbackPerformed"] is False
+    assert receipt["rollbackVerified"] is False
+    assert receipt["zeroResidueVerified"] is True
+    assert not (Path(harness["rollback_root"]) / run_id).exists()
+    assert not (Path(harness["stage_root"]) / run_id).exists()
+    for name in harness["payload_order"]:
+        target = (
+            Path(harness["exporter_root"]) / name
+            if name == "backup-health-textfile-exporter.py"
+            else Path(harness["destination"]) / name
+        )
+        assert target.read_bytes() == (Path(harness["source_stage"]) / name).read_bytes()
+
+
+@pytest.mark.parametrize("fault", ["write", "link", "fsync", "readback"])
+def test_receipt_fault_rolls_back_without_false_success_or_premature_prestate_cleanup(
+    tmp_path: Path,
+    fault: str,
+) -> None:
+    run_id = f"receipt-{fault}"
+    harness = _build_executor_harness(tmp_path, run_id)
+
+    result = _run_executor_harness(
+        harness,
+        HOST110_BACKUP_RUNTIME_TEST_RECEIPT_FAULT=fault,
+    )
+    assert result.returncode != 0
+    assert '"ok": true' not in result.stdout.lower()
+    assert "durable_receipt_failed" in result.stderr
+    _assert_prestate_restored(harness)
+    receipt_path = Path(harness["status_root"]) / f"agent99-backup-runtime-deploy-{run_id}.json"
+    receipt = json.loads(receipt_path.read_text(encoding="utf-8"))
+    assert receipt["status"] == "failed_rolled_back"
+    assert receipt["rollbackAttempted"] is True
+    assert receipt["rollbackPerformed"] is True
+    assert receipt["rollbackVerified"] is True
+    assert receipt["zeroResidueVerified"] is True
+    assert not (Path(harness["rollback_root"]) / run_id).exists()
+    assert not (Path(harness["stage_root"]) / run_id).exists()
+    for parent in (Path(harness["destination"]), Path(harness["exporter_root"])):
+        assert not list(parent.glob(f".*.agent99-*{run_id}"))
+
+
+def test_apply_temp_fault_is_removed_and_zero_residue_is_verified(tmp_path: Path) -> None:
+    run_id = "apply-temp-cleanup"
+    harness = _build_executor_harness(tmp_path, run_id)
+
+    result = _run_executor_harness(
+        harness,
+        HOST110_BACKUP_RUNTIME_TEST_APPLY_FAULT="after_temp_install",
+    )
+
+    assert result.returncode != 0
+    assert '"ok": true' not in result.stdout.lower()
+    assert "fault_injected_after_temp_install" in result.stderr
+    _assert_prestate_restored(harness)
+    receipt_path = Path(harness["status_root"]) / f"agent99-backup-runtime-deploy-{run_id}.json"
+    receipt = json.loads(receipt_path.read_text(encoding="utf-8"))
+    assert receipt["status"] == "failed_rolled_back"
+    assert receipt["rollbackAttempted"] is True
+    assert receipt["rollbackPerformed"] is True
+    assert receipt["rollbackVerified"] is True
+    assert receipt["zeroResidueVerified"] is True
+    for parent in (Path(harness["destination"]), Path(harness["exporter_root"])):
+        assert not list(parent.glob(f".*.agent99-*{run_id}"))
+
+
+def test_temp_cleanup_fault_is_rollback_unverified_and_preserves_prestate(tmp_path: Path) -> None:
+    run_id = "temp-residue-unverified"
+    harness = _build_executor_harness(tmp_path, run_id)
+
+    result = _run_executor_harness(
+        harness,
+        HOST110_BACKUP_RUNTIME_TEST_APPLY_FAULT="after_temp_install",
+        HOST110_BACKUP_RUNTIME_TEST_ROLLBACK_FAULT="preserve_first_temp",
+    )
+
+    assert result.returncode != 0
+    assert '"ok": true' not in result.stdout.lower()
+    _assert_prestate_restored(harness)
+    receipt_path = Path(harness["status_root"]) / f"agent99-backup-runtime-deploy-{run_id}.json"
+    receipt = json.loads(receipt_path.read_text(encoding="utf-8"))
+    assert receipt["status"] == "rollback_unverified"
+    assert receipt["rollbackAttempted"] is True
+    assert receipt["rollbackPerformed"] is True
+    assert receipt["rollbackVerified"] is False
+    assert receipt["zeroResidueVerified"] is False
+    assert (Path(harness["rollback_root"]) / run_id / "prestate.tsv").is_file()
+    residue = []
+    for parent in (Path(harness["destination"]), Path(harness["exporter_root"])):
+        residue.extend(parent.glob(f".*.agent99-*{run_id}"))
+    assert residue
+
+
 def test_backup_runtime_entrypoints_share_the_deployment_gate() -> None:
     common = (ROOT / "scripts/backup/common.sh").read_text(encoding="utf-8")
     restore = (ROOT / "scripts/backup/gitea-full-backup-restore-drill.sh").read_text(encoding="utf-8")