diff --git a/.github/workflows/db-backup-freshness.yml b/.github/workflows/db-backup-freshness.yml
index 7301357..d676a65 100644
--- a/.github/workflows/db-backup-freshness.yml
+++ b/.github/workflows/db-backup-freshness.yml
@@ -81,14 +81,9 @@ jobs:
         id: select
         run: |
           set -euo pipefail
-          test "$(aws sts get-caller-identity --query Account --output text)" = '992382684924'
-          aws s3api get-public-access-block --bucket "$BACKUP_BUCKET" \
-            --query 'PublicAccessBlockConfiguration.[BlockPublicAcls,IgnorePublicAcls,BlockPublicPolicy,RestrictPublicBuckets]' \
-            --output text | grep -q $'True\tTrue\tTrue\tTrue'
-          test "$(aws s3api get-bucket-encryption --bucket "$BACKUP_BUCKET" \
-            --query 'ServerSideEncryptionConfiguration.Rules[0].ApplyServerSideEncryptionByDefault.SSEAlgorithm' \
-            --output text)" = 'AES256'
+          scripts/check_live_database_backup_target.sh "$BACKUP_BUCKET"
           aws s3api list-objects-v2 --bucket "$BACKUP_BUCKET" --prefix daily/ \
+            --expected-bucket-owner 992382684924 \
             > /tmp/database-backup-inventory.json
           python scripts/check_database_backup_freshness.py select \
             --inventory /tmp/database-backup-inventory.json \
@@ -101,10 +96,12 @@ jobs:
           set -euo pipefail
           aws s3api get-object --bucket "$BACKUP_BUCKET" \
             --key '${{ steps.select.outputs.manifest_key }}' \
+            --expected-bucket-owner 992382684924 \
             /tmp/database-backup-manifest.json > /dev/null
           aws s3api get-object-attributes --bucket "$BACKUP_BUCKET" \
             --key '${{ steps.select.outputs.ciphertext_key }}' \
             --object-attributes Checksum,ObjectSize,StorageClass \
+            --expected-bucket-owner 992382684924 \
             > /tmp/database-backup-attributes.json
           python scripts/check_database_backup_freshness.py verify \
             --selection /tmp/database-backup-selection.json \
diff --git a/.github/workflows/db-backup.yml b/.github/workflows/db-backup.yml
index 6668f26..3643168 100644
--- a/.github/workflows/db-backup.yml
+++ b/.github/workflows/db-backup.yml
@@ -22,7 +22,12 @@ concurrency:
 
 jobs:
   dump:
-    runs-on: ubuntu-latest
+    # The complete production database is plaintext until the local age step.
+    # Route this job only to the workflow-restricted production backup group.
+    # GitHub-hosted runners and the default self-hosted group are not permitted.
+    runs-on:
+      group: production-backup
+      labels: self-hosted
     timeout-minutes: 30
     environment: production-backup
     permissions:
@@ -39,6 +44,8 @@ jobs:
       GITHUB_REF_PROTECTED: ${{ github.ref_protected }}
     steps:
       - name: Validate the protected environment configuration
+        env:
+          RUNNER_BOUNDARY: ${{ runner.environment }}
         run: |
           set -euo pipefail
           missing=()
@@ -58,6 +65,10 @@ jobs:
             echo '::error::The database backup refuses an unprotected main branch.'
             exit 1
           fi
+          if [ "${RUNNER_BOUNDARY}" != 'self-hosted' ]; then
+            echo '::error::The database backup requires its isolated self-hosted production boundary.'
+            exit 1
+          fi
           echo 'The four required production-backup settings are present.'
 
       - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
@@ -139,10 +150,7 @@ jobs:
             echo '::error::The production-backup environment needs AWS_BACKUP_BUCKET.'
             exit 1
           fi
-          test "$(aws sts get-caller-identity --query Account --output text)" = '992382684924'
-          aws s3api get-public-access-block --bucket "$BACKUP_BUCKET" \
-            --query 'PublicAccessBlockConfiguration.[BlockPublicAcls,IgnorePublicAcls,BlockPublicPolicy,RestrictPublicBuckets]' \
-            --output text | grep -q $'True\tTrue\tTrue\tTrue'
+          scripts/check_live_database_backup_target.sh "$BACKUP_BUCKET"
           python scripts/database_backup_contract.py validate-source \
             --db-url "$SUPABASE_DB_URL" \
             --project-ref "$SUPABASE_PROJECT_REF" \
@@ -221,7 +229,8 @@ jobs:
           aws s3 cp artifact-manifest.json \
             "s3://${BACKUP_BUCKET}/${prefix}/artifact-manifest.json" \
             --only-show-errors --sse AES256 \
-            --content-type application/json --checksum-algorithm SHA256
+            --content-type application/json --checksum-algorithm SHA256 \
+            --expected-bucket-owner 992382684924
 
           aws s3api get-object-attributes \
             --bucket "$BACKUP_BUCKET" --key "${prefix}/${cipher}" \
diff --git a/ops/PRODUCTION_OPERATIONS.md b/ops/PRODUCTION_OPERATIONS.md
index 3849055..b550cf0 100644
--- a/ops/PRODUCTION_OPERATIONS.md
+++ b/ops/PRODUCTION_OPERATIONS.md
@@ -104,17 +104,20 @@ this order:
 2. Create `production-backup` and `production-backup-monitor`. Give each
    environment one exact custom `main` branch policy and no other deployment
    policy.
-3. Deploy the reviewed backup CloudFormation stack in AWS account
+3. Create the workflow-restricted `production-backup` runner group and an
+   ephemeral runner inside the production data boundary. Put the runner only in
+   that group. Permit only the exact backup workflow on `refs/heads/main`.
+4. Deploy the reviewed backup CloudFormation stack in AWS account
    `992382684924`.
-4. Configure the four settings in the `production-backup` GitHub environment.
-5. Configure the two variables in the `production-backup-monitor` environment.
-6. Store a second copy of the private `age` key in a team vault or offline
+5. Configure the four settings in the `production-backup` GitHub environment.
+6. Configure the two variables in the `production-backup-monitor` environment.
+7. Store a second copy of the private `age` key in a team vault or offline
    medium.
-7. Create an isolated scratch Supabase project and complete the first recovery
+8. Create an isolated scratch Supabase project and complete the first recovery
    drill.
-8. Select the primary operator, the secondary operator, the support hours, and
+9. Select the primary operator, the secondary operator, the support hours, and
    the response targets for human halts.
-9. Configure an external monitor for the production health and backup freshness
+10. Configure an external monitor for the production health and backup freshness
    schedules.
-10. Complete the first genuine customer transaction when an authorized customer
+11. Complete the first genuine customer transaction when an authorized customer
    is available. Do not create a founder self-charge as evidence.
diff --git a/ops/backup/RESTORE_DRILL.md b/ops/backup/RESTORE_DRILL.md
index 4ee817d..45bbc8f 100644
--- a/ops/backup/RESTORE_DRILL.md
+++ b/ops/backup/RESTORE_DRILL.md
@@ -30,6 +30,7 @@ the matching protected-environment configuration failure. As of that check:
 - provider PITR is not enabled;
 - the AWS stack is not deployed;
 - `main` has no repository ruleset or branch protection;
+- no workflow-restricted production backup runner group or runner is confirmed;
 - the `production-backup` GitHub environment has none of its four required
   settings and has no deployment-branch restriction;
 - the `production-backup-monitor` GitHub environment is not configured;
@@ -77,7 +78,17 @@ deployment policy is the exact branch gate.
    select every protected branch. Do not add a tag or wildcard policy.
 4. Create the `production-backup-monitor` GitHub environment.
 5. Give it the same single custom `main` branch policy.
-6. Do not require a manual environment approval. An approval wait would prevent
+6. Create an organization runner group named `production-backup`. Permit only
+   this repository. Restrict it to
+   `OpenAdaptAI/openadapt-ops/.github/workflows/db-backup.yml@refs/heads/main`.
+   Do not put the runner in the default group and do not permit another
+   workflow to use the group.
+7. Register an ephemeral Linux runner inside the declared OpenAdapt production
+   boundary and put it only in the `production-backup` group. Use a clean
+   encrypted work volume for each job and remove the runner after the job.
+   Permit network access only to GitHub Actions, the exact production Supabase
+   database endpoints, and the private AWS backup target.
+8. Do not require a manual environment approval. An approval wait would prevent
    the scheduled jobs.
 
 Verify that `main` reports as protected. Verify that each environment reports
@@ -85,7 +96,10 @@ Verify that `main` reports as protected. Verify that each environment reports
 the exact name `main`. Both workflows repeat this check after the environment
 admits the job and before they request AWS credentials. The job has read-only
 Actions permission for this API check. Do not deploy the AWS stack until this
-gate passes.
+gate passes. Also verify that the runner group is restricted to the exact
+workflow and that the selected runner reports `self-hosted`. A self-hosted
+runner attached to this public repository without the exact workflow
+restriction is unsafe because pull-request code can target it.
 
 ## One-time AWS setup
 
@@ -111,6 +125,13 @@ SHA-256. S3 validates that checksum before it accepts the object. This launch
 path refuses an encrypted archive above 5 GiB before upload. Build and qualify a
 multipart contract before a production database can exceed that limit.
 
+The writer, monitor, and restore paths read the live bucket controls before
+they use an object. They require AWS account `992382684924`, region
+`us-east-1`, complete public-access blocking, SSE-S3, versioning,
+bucket-owner-enforced ownership, the exact 90-day and 365-day lifecycle rules,
+TLS-only transport, and the exact encryption policy. A drifted target stops the
+operation before database access or recovery-point selection.
+
 An AWS principal with CloudFormation, IAM, and S3 administration rights must
 run:
 
@@ -162,6 +183,12 @@ The workflow validates that the URL belongs to the declared Supabase project.
 It also checks AWS account `992382684924`, complete S3 public-access blocking,
 and the committed `age` recipient before it reads the database.
 
+The dump job does not use a GitHub-hosted runner. The complete database exists
+in plaintext on the isolated backup runner until local `age` encryption. The
+cleanup trap removes the plaintext before the runner is destroyed. The
+freshness monitor can use a GitHub-hosted runner because it reads only the
+redacted manifest and S3 object metadata. It cannot read the ciphertext.
+
 The public recipient is in `ops/backup/age-recipients.txt`. Store its private
 key with mode `0600` on an encrypted trusted device. Make a second copy in a
 team vault or an offline medium before the first backup. Without a second copy,
@@ -231,7 +258,9 @@ The script:
 6. decrypts into a private temporary directory;
 7. extracts only four exact regular files and rejects unsafe archive members;
 8. restores with `ON_ERROR_STOP` in one transaction;
-9. dumps the scratch schema and data again and compares their digests;
+9. dumps the scratch schema and data again, normalizes only the matched random
+   PostgreSQL restriction guard outside `COPY` data, and compares their
+   digests;
 10. writes a new database-only evidence file without overwriting old evidence;
 11. uploads that metadata-only evidence below `drills/database-only/`; and
 12. removes all temporary plaintext.
@@ -240,6 +269,13 @@ RTO starts before AWS role assumption and download. It ends after the scratch
 database redump and validation. RPO is measured from the backup recovery point
 to the same start time.
 
+Patched PostgreSQL clients create a new `\restrict` key for each plain-text
+dump. Supabase CLI comments that key, so two correct dumps have different raw
+digests. The verifier keeps the raw source digests in the evidence and uses a
+second comparison digest that replaces only one matched `restrict` and
+`unrestrict` pair outside `COPY` blocks. A guard-shaped database value remains
+data and a changed value still fails the drill.
+
 The script does not delete the scratch project. Review the evidence first.
 Then decommission the project through its authorized owner process.
 
diff --git a/ops/backup/aws-backup-target.yml b/ops/backup/aws-backup-target.yml
index f666379..7b85eb2 100644
--- a/ops/backup/aws-backup-target.yml
+++ b/ops/backup/aws-backup-target.yml
@@ -121,7 +121,14 @@ Resources:
             Version: '2012-10-17'
             Statement:
               - Effect: Allow
-                Action: s3:GetBucketPublicAccessBlock
+                Action:
+                  - s3:GetBucketPublicAccessBlock
+                  - s3:GetEncryptionConfiguration
+                  - s3:GetBucketVersioning
+                  - s3:GetBucketOwnershipControls
+                  - s3:GetBucketLocation
+                  - s3:GetLifecycleConfiguration
+                  - s3:GetBucketPolicy
                 Resource: !GetAtt BackupBucket.Arn
               - Effect: Allow
                 Action:
@@ -154,6 +161,11 @@ Resources:
                 Action:
                   - s3:GetBucketPublicAccessBlock
                   - s3:GetEncryptionConfiguration
+                  - s3:GetBucketVersioning
+                  - s3:GetBucketOwnershipControls
+                  - s3:GetBucketLocation
+                  - s3:GetLifecycleConfiguration
+                  - s3:GetBucketPolicy
                 Resource: !GetAtt BackupBucket.Arn
               - Effect: Allow
                 Action: s3:ListBucket
@@ -185,7 +197,14 @@ Resources:
             Version: '2012-10-17'
             Statement:
               - Effect: Allow
-                Action: s3:GetBucketPublicAccessBlock
+                Action:
+                  - s3:GetBucketPublicAccessBlock
+                  - s3:GetEncryptionConfiguration
+                  - s3:GetBucketVersioning
+                  - s3:GetBucketOwnershipControls
+                  - s3:GetBucketLocation
+                  - s3:GetLifecycleConfiguration
+                  - s3:GetBucketPolicy
                 Resource: !GetAtt BackupBucket.Arn
               - Effect: Allow
                 Action: s3:GetObject
diff --git a/scripts/check_database_backup_target.py b/scripts/check_database_backup_target.py
new file mode 100644
index 0000000..2e06485
--- /dev/null
+++ b/scripts/check_database_backup_target.py
@@ -0,0 +1,220 @@
+#!/usr/bin/env python3
+"""Verify the exact live S3 database-backup target contract."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+
+
+class TargetError(ValueError):
+    """The live S3 target does not match the reviewed backup contract."""
+
+
+def read_object(path: str, name: str) -> dict[str, object]:
+    value = json.loads(Path(path).read_text(encoding="utf-8"))
+    if not isinstance(value, dict):
+        raise TargetError(f"the {name} response is not an object")
+    return value
+
+
+def require_exact_public_block(value: dict[str, object]) -> None:
+    expected = {
+        "BlockPublicAcls": True,
+        "IgnorePublicAcls": True,
+        "BlockPublicPolicy": True,
+        "RestrictPublicBuckets": True,
+    }
+    if value.get("PublicAccessBlockConfiguration") != expected:
+        raise TargetError("the backup bucket public-access block is incomplete")
+
+
+def require_exact_encryption(value: dict[str, object]) -> None:
+    configuration = value.get("ServerSideEncryptionConfiguration")
+    rules = configuration.get("Rules") if isinstance(configuration, dict) else None
+    if not isinstance(rules, list) or len(rules) != 1:
+        raise TargetError("the backup bucket encryption rule is not exact")
+    rule = rules[0]
+    if not isinstance(rule, dict):
+        raise TargetError("the backup bucket encryption rule is invalid")
+    default = rule.get("ApplyServerSideEncryptionByDefault")
+    if not isinstance(default, dict) or default.get("SSEAlgorithm") != "AES256":
+        raise TargetError("the backup bucket does not default to SSE-S3")
+    if default.get("KMSMasterKeyID") is not None:
+        raise TargetError("the backup bucket has an unexpected KMS key")
+
+
+def require_exact_versioning(value: dict[str, object]) -> None:
+    if value.get("Status") != "Enabled":
+        raise TargetError("the backup bucket versioning is not enabled")
+    if value.get("MFADelete") not in {None, "Disabled"}:
+        raise TargetError("the backup bucket has an unexpected MFA-delete state")
+
+
+def require_exact_ownership(value: dict[str, object]) -> None:
+    if value.get("OwnershipControls") != {
+        "Rules": [{"ObjectOwnership": "BucketOwnerEnforced"}]
+    }:
+        raise TargetError("the backup bucket ownership control is not exact")
+
+
+def require_us_east_1(value: dict[str, object]) -> None:
+    # S3 reports a null location constraint for us-east-1.
+    if value.get("LocationConstraint") is not None:
+        raise TargetError("the backup bucket is not in us-east-1")
+
+
+def rule_prefix(rule: dict[str, object]) -> object:
+    if "Prefix" in rule:
+        return rule.get("Prefix")
+    filter_value = rule.get("Filter")
+    if isinstance(filter_value, dict):
+        return filter_value.get("Prefix")
+    return None
+
+
+def require_exact_lifecycle(value: dict[str, object]) -> None:
+    rules = value.get("Rules")
+    if not isinstance(rules, list) or len(rules) != 2:
+        raise TargetError("the backup bucket lifecycle rule set is not exact")
+    by_id = {
+        rule.get("ID"): rule
+        for rule in rules
+        if isinstance(rule, dict) and isinstance(rule.get("ID"), str)
+    }
+    expected = {
+        "DeleteExpiredBackups": ("daily/", 90),
+        "DeleteExpiredDrillEvidence": ("drills/", 365),
+    }
+    if set(by_id) != set(expected):
+        raise TargetError("the backup bucket lifecycle rule identities are not exact")
+    for rule_id, (prefix, days) in expected.items():
+        rule = by_id[rule_id]
+        if (
+            rule.get("Status") != "Enabled"
+            or rule_prefix(rule) != prefix
+            or rule.get("Expiration") != {"Days": days}
+            or rule.get("NoncurrentVersionExpiration") != {"NoncurrentDays": 7}
+        ):
+            raise TargetError(f"the backup bucket lifecycle rule is invalid: {rule_id}")
+        abort = rule.get("AbortIncompleteMultipartUpload")
+        if rule_id == "DeleteExpiredBackups":
+            if abort != {"DaysAfterInitiation": 1}:
+                raise TargetError("the daily backup multipart cleanup is invalid")
+        elif abort is not None:
+            raise TargetError("the drill evidence lifecycle has an unexpected cleanup")
+
+
+def actions(statement: dict[str, object]) -> set[str]:
+    value = statement.get("Action")
+    if isinstance(value, str):
+        return {value}
+    if isinstance(value, list) and all(isinstance(item, str) for item in value):
+        return set(value)
+    return set()
+
+
+def resources(statement: dict[str, object]) -> set[str]:
+    value = statement.get("Resource")
+    if isinstance(value, str):
+        return {value}
+    if isinstance(value, list) and all(isinstance(item, str) for item in value):
+        return set(value)
+    return set()
+
+
+def policy_document(value: dict[str, object]) -> dict[str, object]:
+    document = value.get("Policy")
+    if isinstance(document, str):
+        document = json.loads(document)
+    if not isinstance(document, dict):
+        raise TargetError("the backup bucket policy is invalid")
+    return document
+
+
+def require_exact_policy(value: dict[str, object], bucket: str) -> None:
+    statements = policy_document(value).get("Statement")
+    if not isinstance(statements, list) or len(statements) != 3:
+        raise TargetError("the backup bucket policy statement set is not exact")
+    by_sid = {
+        statement.get("Sid"): statement
+        for statement in statements
+        if isinstance(statement, dict) and isinstance(statement.get("Sid"), str)
+    }
+    expected_ids = {
+        "DenyInsecureTransport",
+        "RequireSseS3",
+        "RequireSseS3ForDrillEvidence",
+    }
+    if set(by_sid) != expected_ids:
+        raise TargetError("the backup bucket policy identities are not exact")
+
+    bucket_arn = f"arn:aws:s3:::{bucket}"
+    tls = by_sid["DenyInsecureTransport"]
+    if (
+        tls.get("Effect") != "Deny"
+        or tls.get("Principal") != "*"
+        or actions(tls) != {"s3:*"}
+        or resources(tls) != {bucket_arn, f"{bucket_arn}/*"}
+        or tls.get("Condition") != {"Bool": {"aws:SecureTransport": "false"}}
+    ):
+        raise TargetError("the backup bucket TLS-only policy is invalid")
+
+    for sid, prefix in (
+        ("RequireSseS3", "daily"),
+        ("RequireSseS3ForDrillEvidence", "drills"),
+    ):
+        statement = by_sid[sid]
+        if (
+            statement.get("Effect") != "Deny"
+            or statement.get("Principal") != "*"
+            or actions(statement) != {"s3:PutObject"}
+            or resources(statement) != {f"{bucket_arn}/{prefix}/*"}
+            or statement.get("Condition")
+            != {
+                "StringNotEquals": {
+                    "s3:x-amz-server-side-encryption": "AES256"
+                }
+            }
+        ):
+            raise TargetError(f"the backup bucket encryption policy is invalid: {sid}")
+
+
+def validate(args: argparse.Namespace) -> dict[str, object]:
+    require_exact_public_block(read_object(args.public_access_block, "public-access"))
+    require_exact_encryption(read_object(args.encryption, "encryption"))
+    require_exact_versioning(read_object(args.versioning, "versioning"))
+    require_exact_ownership(read_object(args.ownership, "ownership"))
+    require_us_east_1(read_object(args.location, "location"))
+    require_exact_lifecycle(read_object(args.lifecycle, "lifecycle"))
+    require_exact_policy(read_object(args.policy, "policy"), args.bucket)
+    return {"valid": True, "bucket": args.bucket, "region": "us-east-1"}
+
+
+def parser() -> argparse.ArgumentParser:
+    result = argparse.ArgumentParser()
+    result.add_argument("--bucket", required=True)
+    result.add_argument("--public-access-block", required=True)
+    result.add_argument("--encryption", required=True)
+    result.add_argument("--versioning", required=True)
+    result.add_argument("--ownership", required=True)
+    result.add_argument("--location", required=True)
+    result.add_argument("--lifecycle", required=True)
+    result.add_argument("--policy", required=True)
+    return result
+
+
+def main() -> int:
+    args = parser().parse_args()
+    try:
+        print(json.dumps(validate(args), sort_keys=True))
+    except (TargetError, OSError, json.JSONDecodeError) as error:
+        print(f"error: {error}", file=sys.stderr)
+        return 2
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/scripts/check_live_database_backup_target.sh b/scripts/check_live_database_backup_target.sh
new file mode 100755
index 0000000..31cc826
--- /dev/null
+++ b/scripts/check_live_database_backup_target.sh
@@ -0,0 +1,48 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+if [[ $# -ne 1 || -z "$1" ]]; then
+  echo 'error: one backup bucket name is required' >&2
+  exit 2
+fi
+
+bucket=$1
+expected_account=992382684924
+account=$(aws sts get-caller-identity --query Account --output text)
+if [[ "$account" != "$expected_account" ]]; then
+  echo "error: AWS credentials do not resolve to account ${expected_account}" >&2
+  exit 2
+fi
+
+root=$(mktemp -d "${TMPDIR:-/tmp}/openadapt-db-backup-target.XXXXXX")
+chmod 700 "$root"
+cleanup() {
+  find "$root" -type f -delete 2>/dev/null || true
+  rmdir "$root" 2>/dev/null || true
+}
+trap cleanup EXIT INT TERM
+
+aws s3api get-public-access-block --bucket "$bucket" \
+  --expected-bucket-owner "$expected_account" > "$root/public-access.json"
+aws s3api get-bucket-encryption --bucket "$bucket" \
+  --expected-bucket-owner "$expected_account" > "$root/encryption.json"
+aws s3api get-bucket-versioning --bucket "$bucket" \
+  --expected-bucket-owner "$expected_account" > "$root/versioning.json"
+aws s3api get-bucket-ownership-controls --bucket "$bucket" \
+  --expected-bucket-owner "$expected_account" > "$root/ownership.json"
+aws s3api get-bucket-location --bucket "$bucket" \
+  --expected-bucket-owner "$expected_account" > "$root/location.json"
+aws s3api get-bucket-lifecycle-configuration --bucket "$bucket" \
+  --expected-bucket-owner "$expected_account" > "$root/lifecycle.json"
+aws s3api get-bucket-policy --bucket "$bucket" \
+  --expected-bucket-owner "$expected_account" > "$root/policy.json"
+
+python scripts/check_database_backup_target.py \
+  --bucket "$bucket" \
+  --public-access-block "$root/public-access.json" \
+  --encryption "$root/encryption.json" \
+  --versioning "$root/versioning.json" \
+  --ownership "$root/ownership.json" \
+  --location "$root/location.json" \
+  --lifecycle "$root/lifecycle.json" \
+  --policy "$root/policy.json"
diff --git a/scripts/database_backup_contract.py b/scripts/database_backup_contract.py
index cfcc9c3..bbbfcfb 100755
--- a/scripts/database_backup_contract.py
+++ b/scripts/database_backup_contract.py
@@ -21,7 +21,7 @@
 
 CONTRACT_SCHEMA = "openadapt.database-backup-contract/v2"
 ARTIFACT_SCHEMA = "openadapt.database-backup-artifact/v2"
-RESTORE_EVIDENCE_SCHEMA = "openadapt.database-restore-evidence/v1"
+RESTORE_EVIDENCE_SCHEMA = "openadapt.database-restore-evidence/v2"
 S3_UPLOAD_SCHEMA = "openadapt.database-backup-s3-upload/v1"
 S3_SINGLE_PUT_MAX_BYTES = 5 * 1024 * 1024 * 1024
 PROJECT_REF = re.compile(r"^[a-z0-9]{8,64}$")
@@ -34,6 +34,11 @@
     re.MULTILINE | re.IGNORECASE,
 )
 COPY_PROGRAM = re.compile(r"\bCOPY\b[^;]*\bPROGRAM\b", re.IGNORECASE | re.DOTALL)
+RESTRICT_MARKER = re.compile(
+    r"^-- \\(?P<operation>un)?restrict (?P<key>[A-Za-z0-9]+)\r?\n?$"
+)
+COPY_FROM_STDIN = re.compile(r"^COPY\s.+\sFROM\sstdin;\r?\n?$", re.IGNORECASE)
+SHA256 = re.compile(r"^[0-9a-f]{64}$")
 
 
 class ContractError(ValueError):
@@ -125,6 +130,49 @@ def dump_inventory(root: Path) -> list[dict[str, object]]:
     return result
 
 
+def comparison_sha256(path: Path) -> str:
+    """Hash a dump while normalizing only pg_dump's random guard key.
+
+    Patched PostgreSQL clients generate a fresh ``\\restrict`` key for every
+    plain-text dump. Supabase CLI comments those two guard commands instead of
+    removing them. A source dump and a correct scratch redump therefore differ
+    in those random keys. The restore check must ignore that non-data value,
+    but it must never rewrite a matching line inside COPY data.
+    """
+
+    digest = hashlib.sha256()
+    markers: list[tuple[str, str]] = []
+    in_copy = False
+    with path.open("r", encoding="utf-8", errors="strict", newline="") as stream:
+        for line in stream:
+            marker = RESTRICT_MARKER.fullmatch(line) if not in_copy else None
+            if marker is not None:
+                operation = "unrestrict" if marker.group("operation") else "restrict"
+                markers.append((operation, marker.group("key")))
+                digest.update(f"-- \\{operation} OPENADAPT_RANDOM_KEY\n".encode())
+                continue
+
+            digest.update(line.encode())
+            if not in_copy and COPY_FROM_STDIN.fullmatch(line):
+                in_copy = True
+            elif in_copy and line.rstrip("\r\n") == r"\.":
+                in_copy = False
+
+    if in_copy:
+        raise ContractError(f"the restored dump has an unterminated COPY block: {path.name}")
+    if markers:
+        if (
+            len(markers) != 2
+            or markers[0][0] != "restrict"
+            or markers[1][0] != "unrestrict"
+            or markers[0][1] != markers[1][1]
+        ):
+            raise ContractError(
+                f"the restored dump has an invalid pg_dump restriction guard: {path.name}"
+            )
+    return digest.hexdigest()
+
+
 def parse_time(value: str, name: str) -> datetime:
     try:
         result = datetime.fromisoformat(value.replace("Z", "+00:00"))
@@ -420,18 +468,24 @@ def extract_artifact(args: argparse.Namespace) -> None:
 def verify_restored_dumps(args: argparse.Namespace) -> None:
     source = {entry["name"]: entry for entry in dump_inventory(Path(args.source_dir))}
     restored: dict[str, dict[str, object]] = {}
+    comparison: dict[str, dict[str, str]] = {}
     for name in ("schema.sql", "data.sql"):
-        path = Path(args.restored_dir) / name
-        if not path.is_file() or path.stat().st_size <= 0:
+        source_path = Path(args.source_dir) / name
+        restored_path = Path(args.restored_dir) / name
+        if not restored_path.is_file() or restored_path.stat().st_size <= 0:
             raise ContractError(f"the restored dump is missing or empty: {name}")
         restored[name] = {
             "name": name,
-            "bytes": path.stat().st_size,
-            "sha256": sha256_file(path),
+            "bytes": restored_path.stat().st_size,
+            "sha256": sha256_file(restored_path),
+        }
+        comparison[name] = {
+            "source": comparison_sha256(source_path),
+            "restored": comparison_sha256(restored_path),
         }
     # Roles are target-specific. Schema and data must reproduce exactly.
     for name in ("schema.sql", "data.sql"):
-        if source[name]["sha256"] != restored[name]["sha256"]:
+        if comparison[name]["source"] != comparison[name]["restored"]:
             raise ContractError(f"the restored {name} does not match the backup")
     print(
         json.dumps(
@@ -439,6 +493,8 @@ def verify_restored_dumps(args: argparse.Namespace) -> None:
                 "valid": True,
                 "schema_sha256": source["schema.sql"]["sha256"],
                 "data_sha256": source["data.sql"]["sha256"],
+                "schema_comparison_sha256": comparison["schema.sql"]["source"],
+                "data_comparison_sha256": comparison["data.sql"]["source"],
             },
             sort_keys=True,
         )
@@ -475,6 +531,12 @@ def record_restore(args: argparse.Namespace) -> None:
     for name in ("schema", "data"):
         if verification.get(f"{name}_sha256") != expected_digests.get(f"{name}.sql"):
             raise ContractError("the restore verification does not match the backup contract")
+        comparison_digest = verification.get(f"{name}_comparison_sha256")
+        if (
+            not isinstance(comparison_digest, str)
+            or SHA256.fullmatch(comparison_digest) is None
+        ):
+            raise ContractError("the restore comparison digest is invalid")
     started = parse_time(args.started_at, "started-at")
     completed = parse_time(args.completed_at, "completed-at")
     if completed < started:
@@ -493,6 +555,8 @@ def record_restore(args: argparse.Namespace) -> None:
         "rto_seconds": int((completed - started).total_seconds()),
         "schema_sha256": verification["schema_sha256"],
         "data_sha256": verification["data_sha256"],
+        "schema_comparison_sha256": verification["schema_comparison_sha256"],
+        "data_comparison_sha256": verification["data_comparison_sha256"],
         "database_restored": True,
         "storage_restored": False,
     }
diff --git a/scripts/run_database_restore_drill.sh b/scripts/run_database_restore_drill.sh
index b28d800..35d8e55 100755
--- a/scripts/run_database_restore_drill.sh
+++ b/scripts/run_database_restore_drill.sh
@@ -85,26 +85,22 @@ if [[ "$account" != '992382684924' ]]; then
   echo 'error: AWS_PROFILE does not resolve to OpenAdapt account 992382684924' >&2
   exit 2
 fi
-block=$(aws s3api get-public-access-block \
-  --bucket "$AWS_BACKUP_BUCKET" \
-  --query 'PublicAccessBlockConfiguration.[BlockPublicAcls,IgnorePublicAcls,BlockPublicPolicy,RestrictPublicBuckets]' \
-  --output text)
-if [[ "$block" != $'True\tTrue\tTrue\tTrue' ]]; then
-  echo 'error: backup bucket public-access blocking is incomplete' >&2
-  exit 2
-fi
+scripts/check_live_database_backup_target.sh "$AWS_BACKUP_BUCKET"
 
 cipher="db-backup-${BACKUP_STAMP}.tar.gz.age"
 plain="db-backup-${BACKUP_STAMP}.tar.gz"
 prefix="daily/${BACKUP_STAMP}"
 aws s3 cp \
-  "s3://${AWS_BACKUP_BUCKET}/${prefix}/${cipher}" "$root/$cipher" --only-show-errors
+  "s3://${AWS_BACKUP_BUCKET}/${prefix}/${cipher}" "$root/$cipher" \
+  --only-show-errors --expected-bucket-owner 992382684924
 aws s3 cp \
   "s3://${AWS_BACKUP_BUCKET}/${prefix}/artifact-manifest.json" \
-  "$root/artifact-manifest.json" --only-show-errors
+  "$root/artifact-manifest.json" --only-show-errors \
+  --expected-bucket-owner 992382684924
 
 remote_sha=$(aws s3api head-object \
   --bucket "$AWS_BACKUP_BUCKET" --key "${prefix}/${cipher}" \
+  --expected-bucket-owner 992382684924 \
   --query 'Metadata.sha256' --output text)
 local_sha=$(shasum -a 256 "$root/$cipher" | awk '{print $1}')
 if [[ "$remote_sha" != "$local_sha" ]]; then
@@ -150,6 +146,6 @@ chmod 600 "$output"
 aws s3 cp "$output" \
   "s3://${AWS_BACKUP_BUCKET}/drills/database-only/${BACKUP_STAMP}/$(basename "$output")" \
   --only-show-errors --sse AES256 --content-type application/json \
-  --checksum-algorithm SHA256
+  --checksum-algorithm SHA256 --expected-bucket-owner 992382684924
 echo "Database-only restore evidence: $output"
 echo 'Run the openadapt-cloud database-plus-Storage drill before recording the canonical retention receipt.'
diff --git a/tests/test_database_backup_contract.py b/tests/test_database_backup_contract.py
index 9378308..22ca28d 100644
--- a/tests/test_database_backup_contract.py
+++ b/tests/test_database_backup_contract.py
@@ -323,6 +323,67 @@ def test_restore_target_can_never_be_production() -> None:
         )
 
 
+def test_restore_comparison_normalizes_only_pg_dump_guard_keys(tmp_path: Path) -> None:
+    source = tmp_path / "source"
+    restored = tmp_path / "restored"
+    write_dumps(source)
+    write_dumps(restored)
+    source_data = source / "data.sql"
+    restored_data = restored / "data.sql"
+    source_data.write_text(
+        "-- \\restrict Source123\n"
+        + source_data.read_text()
+        + "-- \\unrestrict Source123\n"
+    )
+    restored_data.write_text(
+        "-- \\restrict Restored456\n"
+        + restored_data.read_text()
+        + "-- \\unrestrict Restored456\n"
+    )
+
+    assert backup.comparison_sha256(source_data) == backup.comparison_sha256(
+        restored_data
+    )
+
+
+def test_restore_comparison_preserves_guard_shaped_copy_data(tmp_path: Path) -> None:
+    source = tmp_path / "source.sql"
+    restored = tmp_path / "restored.sql"
+    source.write_text(
+        "-- \\restrict Source123\n"
+        "COPY public.runs (value) FROM stdin;\n"
+        "-- \\restrict customer_value\n"
+        "\\.\n"
+        "-- \\unrestrict Source123\n"
+    )
+    restored.write_text(
+        "-- \\restrict Restored456\n"
+        "COPY public.runs (value) FROM stdin;\n"
+        "-- \\restrict changed_value\n"
+        "\\.\n"
+        "-- \\unrestrict Restored456\n"
+    )
+
+    assert backup.comparison_sha256(source) != backup.comparison_sha256(restored)
+
+
+@pytest.mark.parametrize(
+    "value",
+    [
+        "-- \\restrict one\n",
+        "-- \\unrestrict one\n-- \\restrict one\n",
+        "-- \\restrict one\n-- \\unrestrict two\n",
+    ],
+)
+def test_restore_comparison_rejects_invalid_guard_pairs(
+    tmp_path: Path, value: str
+) -> None:
+    path = tmp_path / "data.sql"
+    path.write_text(value)
+    with pytest.raises(backup.ContractError, match="restriction guard"):
+        backup.comparison_sha256(path)
+
+
 def test_restore_evidence_is_bound_to_exact_backup_and_scratch(tmp_path: Path) -> None:
     dumps, _, contract = make_contract(tmp_path)
     restored = tmp_path / "restored"
@@ -335,6 +396,8 @@ def test_restore_evidence_is_bound_to_exact_backup_and_scratch(tmp_path: Path) -
                 "valid": True,
                 "schema_sha256": source["schema.sql"]["sha256"],
                 "data_sha256": source["data.sql"]["sha256"],
+                "schema_comparison_sha256": source["schema.sql"]["sha256"],
+                "data_comparison_sha256": source["data.sql"]["sha256"],
             }
         )
     )
@@ -369,6 +432,8 @@ def test_restore_evidence_is_bound_to_exact_backup_and_scratch(tmp_path: Path) -
     value = json.loads(evidence.read_text())
     assert value["database_restored"] is True
     assert value["storage_restored"] is False
+    assert value["schema_comparison_sha256"] == source["schema.sql"]["sha256"]
+    assert value["data_comparison_sha256"] == source["data.sql"]["sha256"]
     assert value["rto_seconds"] == 300
     assert value["rpo_seconds_at_start"] == 2220
 
@@ -378,6 +443,8 @@ def test_restore_evidence_is_bound_to_exact_backup_and_scratch(tmp_path: Path) -
                 "valid": True,
                 "schema_sha256": "0" * 64,
                 "data_sha256": source["data.sql"]["sha256"],
+                "schema_comparison_sha256": source["schema.sql"]["sha256"],
+                "data_comparison_sha256": source["data.sql"]["sha256"],
             }
         )
     )
@@ -401,6 +468,33 @@ def test_restore_evidence_is_bound_to_exact_backup_and_scratch(tmp_path: Path) -
                 "valid": True,
                 "schema_sha256": source["schema.sql"]["sha256"],
                 "data_sha256": source["data.sql"]["sha256"],
+                "schema_comparison_sha256": "not-a-digest",
+                "data_comparison_sha256": source["data.sql"]["sha256"],
+            }
+        )
+    )
+    with pytest.raises(backup.ContractError, match="comparison digest"):
+        backup.record_restore(
+            Namespace(
+                manifest=str(manifest),
+                contract=str(contract),
+                verification=str(verification),
+                source_project_ref=SOURCE_REF,
+                scratch_project_ref=SCRATCH_REF,
+                started_at="2026-08-08T08:00:00Z",
+                completed_at="2026-08-08T08:05:00Z",
+                output=str(tmp_path / "missing-comparison-evidence.json"),
+            )
+        )
+
+    verification.write_text(
+        json.dumps(
+            {
+                "valid": True,
+                "schema_sha256": source["schema.sql"]["sha256"],
+                "data_sha256": source["data.sql"]["sha256"],
+                "schema_comparison_sha256": source["schema.sql"]["sha256"],
+                "data_comparison_sha256": source["data.sql"]["sha256"],
             }
         )
     )
diff --git a/tests/test_database_backup_target.py b/tests/test_database_backup_target.py
new file mode 100644
index 0000000..f7eac9e
--- /dev/null
+++ b/tests/test_database_backup_target.py
@@ -0,0 +1,188 @@
+from __future__ import annotations
+
+import importlib.util
+import json
+from argparse import Namespace
+from pathlib import Path
+
+import pytest
+
+MODULE_PATH = Path(__file__).parents[1] / "scripts" / "check_database_backup_target.py"
+SPEC = importlib.util.spec_from_file_location("check_database_backup_target", MODULE_PATH)
+assert SPEC and SPEC.loader
+target = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(target)
+
+BUCKET = "openadapt-production-db-backups-992382684924"
+
+
+def documents(tmp_path: Path) -> Namespace:
+    bucket_arn = f"arn:aws:s3:::{BUCKET}"
+    values = {
+        "public_access_block": {
+            "PublicAccessBlockConfiguration": {
+                "BlockPublicAcls": True,
+                "IgnorePublicAcls": True,
+                "BlockPublicPolicy": True,
+                "RestrictPublicBuckets": True,
+            }
+        },
+        "encryption": {
+            "ServerSideEncryptionConfiguration": {
+                "Rules": [
+                    {
+                        "ApplyServerSideEncryptionByDefault": {
+                            "SSEAlgorithm": "AES256"
+                        }
+                    }
+                ]
+            }
+        },
+        "versioning": {"Status": "Enabled"},
+        "ownership": {
+            "OwnershipControls": {
+                "Rules": [{"ObjectOwnership": "BucketOwnerEnforced"}]
+            }
+        },
+        "location": {"LocationConstraint": None},
+        "lifecycle": {
+            "Rules": [
+                {
+                    "ID": "DeleteExpiredBackups",
+                    "Status": "Enabled",
+                    "Filter": {"Prefix": "daily/"},
+                    "Expiration": {"Days": 90},
+                    "NoncurrentVersionExpiration": {"NoncurrentDays": 7},
+                    "AbortIncompleteMultipartUpload": {"DaysAfterInitiation": 1},
+                },
+                {
+                    "ID": "DeleteExpiredDrillEvidence",
+                    "Status": "Enabled",
+                    "Filter": {"Prefix": "drills/"},
+                    "Expiration": {"Days": 365},
+                    "NoncurrentVersionExpiration": {"NoncurrentDays": 7},
+                },
+            ]
+        },
+        "policy": {
+            "Policy": json.dumps(
+                {
+                    "Version": "2012-10-17",
+                    "Statement": [
+                        {
+                            "Sid": "DenyInsecureTransport",
+                            "Effect": "Deny",
+                            "Principal": "*",
+                            "Action": "s3:*",
+                            "Resource": [bucket_arn, f"{bucket_arn}/*"],
+                            "Condition": {
+                                "Bool": {"aws:SecureTransport": "false"}
+                            },
+                        },
+                        {
+                            "Sid": "RequireSseS3",
+                            "Effect": "Deny",
+                            "Principal": "*",
+                            "Action": "s3:PutObject",
+                            "Resource": f"{bucket_arn}/daily/*",
+                            "Condition": {
+                                "StringNotEquals": {
+                                    "s3:x-amz-server-side-encryption": "AES256"
+                                }
+                            },
+                        },
+                        {
+                            "Sid": "RequireSseS3ForDrillEvidence",
+                            "Effect": "Deny",
+                            "Principal": "*",
+                            "Action": "s3:PutObject",
+                            "Resource": f"{bucket_arn}/drills/*",
+                            "Condition": {
+                                "StringNotEquals": {
+                                    "s3:x-amz-server-side-encryption": "AES256"
+                                }
+                            },
+                        },
+                    ],
+                }
+            )
+        },
+    }
+    paths: dict[str, str] = {}
+    for name, value in values.items():
+        path = tmp_path / f"{name}.json"
+        path.write_text(json.dumps(value), encoding="utf-8")
+        paths[name] = str(path)
+    return Namespace(bucket=BUCKET, **paths)
+
+
+def change(args: Namespace, name: str, update) -> None:
+    path = Path(getattr(args, name))
+    value = json.loads(path.read_text())
+    update(value)
+    path.write_text(json.dumps(value))
+
+
+def weaken_tls_policy(value: dict[str, object]) -> None:
+    policy = json.loads(value["Policy"])
+    policy["Statement"][0]["Condition"]["Bool"]["aws:SecureTransport"] = "true"
+    value["Policy"] = json.dumps(policy)
+
+
+def test_exact_live_bucket_contract_passes(tmp_path: Path) -> None:
+    assert target.validate(documents(tmp_path)) == {
+        "valid": True,
+        "bucket": BUCKET,
+        "region": "us-east-1",
+    }
+
+
+@pytest.mark.parametrize(
+    ("name", "update", "message"),
+    [
+        (
+            "public_access_block",
+            lambda value: value["PublicAccessBlockConfiguration"].update(
+                BlockPublicPolicy=False
+            ),
+            "public-access",
+        ),
+        (
+            "encryption",
+            lambda value: value["ServerSideEncryptionConfiguration"]["Rules"][0][
+                "ApplyServerSideEncryptionByDefault"
+            ].update(SSEAlgorithm="aws:kms"),
+            "SSE-S3",
+        ),
+        ("versioning", lambda value: value.update(Status="Suspended"), "versioning"),
+        (
+            "ownership",
+            lambda value: value["OwnershipControls"]["Rules"][0].update(
+                ObjectOwnership="ObjectWriter"
+            ),
+            "ownership",
+        ),
+        (
+            "location",
+            lambda value: value.update(LocationConstraint="us-west-2"),
+            "us-east-1",
+        ),
+        (
+            "lifecycle",
+            lambda value: value["Rules"][0]["Expiration"].update(Days=30),
+            "DeleteExpiredBackups",
+        ),
+        (
+            "policy",
+            weaken_tls_policy,
+            "TLS-only",
+        ),
+    ],
+)
+def test_live_bucket_drift_is_rejected(
+    tmp_path: Path, name: str, update, message: str
+) -> None:
+    args = documents(tmp_path)
+    change(args, name, update)
+    with pytest.raises(target.TargetError, match=message):
+        target.validate(args)
diff --git a/tests/test_production_ops_workflows.py b/tests/test_production_ops_workflows.py
index 963838a..4aef4c4 100644
--- a/tests/test_production_ops_workflows.py
+++ b/tests/test_production_ops_workflows.py
@@ -20,7 +20,14 @@ def test_backup_configuration_fails_before_credentials_or_tool_install() -> None
     assert preflight < environment_gate < credentials < supabase
     assert "actions: read" in workflow
     assert "GITHUB_REF_PROTECTED" in workflow
+    assert "group: production-backup" in workflow
+    assert "labels: self-hosted" in workflow
+    assert "RUNNER_BOUNDARY: ${{ runner.environment }}" in workflow
+    assert '"${RUNNER_BOUNDARY}" != \'self-hosted\'' in workflow
     assert "check_github_environment_gate.py" in workflow
+    target = workflow.index("check_live_database_backup_target.sh")
+    dump = workflow.index("supabase db dump")
+    assert environment_gate < credentials < target < dump
     for name in (
         "AWS_BACKUP_ROLE_ARN",
         "AWS_BACKUP_BUCKET",
@@ -58,6 +65,8 @@ def test_backup_monitor_cannot_download_or_change_ciphertext() -> None:
     assert "age --decrypt" not in workflow
     assert "get-object-attributes" in workflow
     assert "steps.select.outputs.ciphertext_key" in workflow
+    assert "check_live_database_backup_target.sh" in workflow
+    assert "--expected-bucket-owner 992382684924" in workflow
     assert workflow.index("Verify the exact GitHub environment gate") < workflow.index(
         "aws-actions/configure-aws-credentials"
     )