Scanner SCN-5 own-database data move and rollback

Audience: Scanner implementer, database operator, live-deploy operator, rollback owner, and the owner approving the SPRINT_20260722_017 SCN-5 (M4) maintenance window.

Purpose: start from the prepared target at migration 001, create any historical partitions the copy needs, freeze both writers, copy the 84 source leaf tables into that old shape, and prove exact row, content-digest and sequence/identity parity. Only then run Scanner’s supported --migrate-only path to apply 002 and 003, prove the EPSS retirement and tenant attribution, and repoint the live estate. The source remains the rollback authority throughout retention.

This procedure is executable only in the recorded owner-attended window. Phase A prep receipts are recorded below; they do not replace the window-time assertions. Stop at the first failed assertion. Do not boot a current Scanner candidate against the migration-001 target before the data copy: its startup chain would apply 002/003 to an empty database and make the old-shape dump incompatible. Do not improvise a whole-schema restore, copy either migration ledger, copy signals.* or eventing.*, or unfreeze a writer while authority is ambiguous.

Non-negotiable boundaries

• The source scanner schema in stellaops_platform stays intact through rollback retention. This runbook drops no schema, table, database, role, volume or container. The source schema drop is SCN-6’s work and a separate destructive approval.
• signals.score_history and the whole signals schema are not Scanner’s and are never copied. The consolidated baseline does not create them (verified: signals is absent from stellaops_scanner — SELECT nspname FROM pg_namespace returns public, scanner only).
• eventing.* is never copied. scanner-web converges eventing into the family database itself and the worker mints a new stream_epoch by construction. Copying eventing.stream_state would carry the OLD epoch across the one rebuild the epoch exists for. stellaops_platform does carry an eventing schema (information_schema.schemata count = 1), so a whole-database dump would pick it up — that is why every dump in this runbook is --schema=scanner.
• scanner.schema_migrations is never copied. It must read exactly 001 before the copy and exactly 001/002/003 after --migrate-only. See hazard H2.
• A count-only comparison is insufficient. Every copied table needs count plus a canonical content digest, and sequence state is a separate parity class that can fail while every digest matches (hazard H3).
• Any parity failure happens before any repoint, key deletion or grant change. Keep the writer freeze, change nothing outward-facing, and enter rollback.
• Run from the repository root in Bash. Receipts go under the git-ignored tmp/scn5-<window-id>/. Never put a password, token, connection string or decrypted secret into a command transcript or a checked-in file. The Scanner DSN is read from the git-ignored devops/compose/.env by variable name only.

Source-derived copy contract

Scope: the scanner schema of stellaops_platform, data-only, minus the named exclusions.

The target is a strict superset of the source at the relation level (see the A2 verdict below), so a whole-schema data-only dump with explicit exclusions is both correct and drift-proof; an enumerated table list would silently omit a table a later migration adds. The exclusions are the statement of intent and are re-asserted from the dump’s own table of contents in step 7.

Copy set — 84 leaf tables, 135 rows. Measured 2026-08-28 with count(*) per leaf relation (relkind='r'), which is what the restore must reproduce:

artifact_boms is a RANGE-partitioned parent over inserted_at; its 8 rows live in the two children listed above (artifact_boms_2026_09 exists and is empty). Counting the parent and its children both is a double count — the previously circulated figure of “143 business rows” adds artifact_boms (8) to artifact_boms_2026_07 (4) and artifact_boms_2026_08 (4). The leaf-relation total is 135, and the parity harness in step 9 counts leaves only.

Explicit exclusions:

• scanner.schema_migrations — the source ledger (14 rows). The target’s ledger is minted by its own convergence and holds exactly one row. See hazard H2.
• everything outside the scanner schema — signals.*, eventing.*, policy.*, public.*. The --schema=scanner filter is the mechanism; there are no cross-schema foreign keys out of scanner (verified: pg_constraint where contype='f' and the referenced namespace <> 'scanner' returns zero rows), so the filter cannot orphan a reference.
• the 24 EPSS relations present in the migration-001 target and absent from the source — nothing is copied into them, and forward migration 002 removes them after copy parity. See hazard H4.

Phase A prep receipts (executed 2026-08-28)

A1 — the target converged from Scanner’s own startup migration

stellaops_scanner was empty (0 relations in every non-system schema) and was converged by running ScannerPersistenceExtensions.AddScannerPersistence and starting the resulting IHostedService — i.e. the GenericStartupMigrationHost path scanner-web runs at boot — against the staged STELLAOPS_POSTGRES_SCANNER_CONNECTION, as the least-privilege scanner role (hazard H8). No schema was copied from the source.

Verified by:

• resolver receipt resolved-from=STELLAOPS_POSTGRES_SCANNER_CONNECTION canonical=True, one hosted migration service registered;
• role-identity receipt taken on the same connection before anything was applied — session_user=scanner current_user=scanner rolsuper=False rolbypassrls=False row_security=on;
• host log Applying 001_scanner_consolidated_baseline.sql (Startup) → Applied 1 migration(s);
• idempotency: a second run logged Database is up to date for Scanner.Persistence and applied 0;
• scanner.schema_migrations in the target holds exactly one row, 001_scanner_consolidated_baseline.sql [startup], checksum prefix 0b5715c512b5;
• non-system schemas in the target are public and scanner only;
• every one of the 576 relations in the target scanner schema is owned by role scanner.

The harness lived under the git-ignored tmp/scratch/scn5-prep/ and is not part of the shipped source. Honest limit: it drives AddScannerPersistence + IHostedService.StartAsync directly rather than booting the scanner-web container, so it proves the migration path, not the service’s runtime wiring. The container-level positive row was already taken by SCN-4 (scratch boot on an image built from working-tree code); the worker’s own boot is still owed to this window.

A2 — shape parity verdict: the fresh baseline can hold every live object

Ten projections were captured from both databases on the same PostgreSQL 18.1 server and compared as sorted sets. MISSING = present in the live source, absent from the fresh target (the class that loses data). EXTRA = present in the target only.

Verdict: after the one remedy in hazard H1, no live object in the scanner schema is missing from a fresh consolidated baseline, and every divergence in the other direction is EPSS. RLS is identical too — a per-table diff of relrowsecurity/relforcerowsecurity across both databases is empty, giving the same 9 tables at rls=true force=false and the same 9 tenant_id = scanner.current_tenant_id() policies. Extensions match (pg_trgm@public, pgcrypto@scanner, plpgsql@pg_catalog).

The 2026-08-28 executable recheck corrected the original prep table’s one blind spot: the fuller projection includes scanner.schema_migrations’ 7 columns, 2 indexes and 6 constraints on both sides. Its rows remain deliberately outside the data copy; its shape is owner schema and must not drift. This changes the three totals above, not MISSING or EXTRA.

Read this verdict with its provenance, which is not what the sprint assumed. The live source is not “the legacy chain only”. scanner.schema_migrations in stellaops_platform shows the consolidated baseline was applied there on 2026-08-18 07:36:30 UTC by container f197f27703f0, with checksum prefix 0b5715c512b5 — byte-identical to the target’s row — on top of the 13 legacy rows from 2026-07-23/24. A live scanner host resolved a transitional key (hazard H5) and converged the consolidated authority into the shared database. So this comparison proves the fresh baseline can hold everything the live schema currently holds, which is exactly what the copy needs; it is weaker evidence than it looks for the separate claim the baseline reproduces the legacy chain’s end shape. That claim is carried by ConsolidatedScannerSchemaTests.LegacyChainAndCurrentConsolidatedChain_ConvergeTheSameSurvivingShape (src/Scanner/__Tests/StellaOps.Scanner.Persistence.Tests/ConsolidatedScannerSchemaTests.cs), which converges both chains from empty in one container and diffs them column-, index- and constraint-wise.

That test was run rather than left as an inference: GREEN. The 2026-08-28 go/no-go run used the targeted xUnit harness: the class ran 9/9, and the exact differential method ran 1/1. These are ran-vs-suite receipts, not an MTP --filter claim.

The two proofs are complements, and the window needs both — neither alone answers M4. What the differential test does not cover, read out of ConsolidatedScannerSchemaTests.cs rather than assumed:

• it runs on postgres:16-alpine(:77) while the estate is PostgreSQL 18.1, so chain equivalence is proven under PG16 semantics;
• its ShapeQueries (:420-445) compare columns, indexes and constraints only — not sequences or their ownership, partition bounds, enum types, view/matview definitions, functions, triggers or RLS flags;
• its columns projection omits is_identity/identity_generation, numeric precision/scale and udt_name, so it cannot see an identity-vs-serial divergence — precisely hazard H3’s class. Its green says nothing about oci_referrers.id;
• both chains converge from empty, so it says nothing about the live schema’s accumulated history (the 2026-08-18 baseline application, the 2026-08-19 EPSS drop);
• schema_migrations is excluded from all three projections;
• it drives MigrationRunner with a Startup category filter (:496-500), not StartupMigrationHost — no advisory lock, checksum validation or pending-manual gate;
• it proves nothing about data.

The table above is the half taken on the real server version, across ten projections including identity, sequence ownership and partition bounds. The test is the half that covers statement-order equivalence, which the table structurally cannot. Re-run the test at go/no-go: it is a source-derived proof and the source can move between now and then.

Hazards measured in prep

H1 — BLOCKING without the remedy. The fresh baseline does not create artifact_boms_2026_07, and the source holds 4 rows there. The baseline calls SELECT scanner.ensure_artifact_boms_future_partitions(1), which creates the current month and one ahead. Converged on 2026-08-28 that yields artifact_boms_2026_08 and artifact_boms_2026_09 — not July. A data-only restore then fails either with relation "scanner.artifact_boms_2026_07" does not exist (per-child COPY) or no partition of relation "artifact_boms" found for row (--load-via-partition-root). Remedy, verified: call the service’s own function SELECT scanner.create_artifact_boms_partition(2026, 7); on the scanner connection, before the restore. After that call the MISSING column above is 0 in all ten projections, and the child’s bound matches the source exactly (FOR VALUES FROM ('2026-07-01 00:00:00+00') TO ('2026-08-01 00:00:00+00')). Running it as the cluster superuser instead leaves the partition and its 7 indexes owned by stellaops — measured, then corrected by recreating it under SET ROLE scanner. The month is not a constant: recompute the set of source partitions at window time (step 3) rather than copying 2026, 7 from this page.

H2 — BLOCKING. Never restore scanner.schema_migrations. The source ledger holds 14 rows: the 13 legacy Scanner.Storage/Scanner.Triage rows and — because of H5 — a row for 001_scanner_consolidated_baseline.sql whose checksum is identical to the target’s. Restoring it would (a) collide on the ledger’s primary key (migration_name) and abort the single-transaction restore, and (b) if forced, reintroduce the two retired authorities into the new database’s ledger, undoing the consolidation this move exists to complete.

H3 — oci_referrers.id is GENERATED ALWAYS AS IDENTITY, and COPY does not advance its sequence. Measured on a throwaway table in the target:

• INSERT INTO t (id, v) VALUES (42, …) → ERROR: cannot insert a non-DEFAULT value into column "id" … defined as GENERATED ALWAYS. Any --column-inserts/hand-written-INSERT copy breaks here.
• COPY t (id, v) FROM stdin with explicit ids → succeeds, ids preserved. This is the form pg_dump --data-only emits, which is why the restore must go through pg_dump/pg_restore and never through hand-authored SQL.
• After that COPY the sequence last_value was still unset, and the next generated insert produced id = 1 — inside the range the copied rows already occupy. That is the failure this runbook’s step 6 and step 9 exist to prevent, and it is invisible to row-count and digest parity.
• pg_dump --data-only does emit SELECT pg_catalog.setval(...) for both identity-backed and serial sequences, and still emits them under --exclude-table and under a -t filter. So step 9 is a verification that the setval landed, not a hand-written fixup.

Source sequence state to reproduce (read 2026-08-28, re-read during step 6 after the step-4 writer freeze): classification_history_id_seq last_value=1, execution_phases_id_seq never called (NULL), oci_referrers_id_seq last_value=16. Ownership is identical on both sides: classification_history.id and execution_phases.id are nextval defaults (deptype='a'), oci_referrers.id is an identity column (deptype='i').

H4 — resolved by forward migration 002; do not edit applied migration 001. The 2026-08-19 26-relation EPSS drop (sprint 008, register item 12) left stellaops_platform.scanner with zero epss% relations. A fresh convergence of 001_scanner_consolidated_baseline.sql creates 24 relations (epss_changes + epss_scores partitioned parents with 7 children each, epss_config, epss_current, epss_import_runs, epss_raw, epss_signal, epss_signal_config), 2 sequences (epss_raw_raw_id_seq, epss_signal_signal_id_seq) and 6 functions (compute_epss_change_flags, compute_epss_risk_band, compute_epss_signal_dedupe_key, create_epss_partition, prune_epss_raw, prune_epss_signals), plus the org-neutral default row in epss_config. Migration 002_remove_mishomed_epss.sql is the ADR-004-compliant retirement and is embedded in Scanner persistence. It must run after the old-shape copy, through the supported --migrate-only path, and leave zero EPSS relations, sequences, and functions. There is no remaining operator choice to accept an empty EPSS surface and no permission to rewrite migration 001.

H5 — both live hosts still resolve a TRANSITIONAL key, as the cluster superuser. Read from docker inspect 2026-08-28:

• stellaops-scanner-web carries SCANNER_SCANNER__STORAGE__DSN → Database=stellaops_platform;Username=stellaops;
• stellaops-scanner-worker carries ScannerStorage__Postgres__ConnectionString → the same, plus ScannerStorage__Postgres__SchemaName=scanner (which stays — the move changes the database, not the schema);
• neither carries STELLAOPS_POSTGRES_SCANNER_CONNECTION, and neither carries VulnSurfaces__FeedTrigger__ScannerConnectionString (D-SCN1-1’s third pointer — still unset, as SCN-4 recorded).

Two consequences. First, this is what let the consolidated baseline reach stellaops_platform on 2026-08-18. Second, the repoint must delete both transitional keys, not merely add the canonical one: the resolver consults the canonical variable first, so a host that keeps the legacy key and loses the canonical one falls straight back to the shared database and reports healthy. Deleting them is also the only available fence, because the hosts connect as the cluster superuser — no REVOKE on stellaops_platform can stop them (step 14).

H6 — the containers’ historical Compose labels are evidence, not an executable cutover chain. The two com.docker.compose.project.config_files labels differ, are long, and include paths in a different clone which no longer exist. Replaying either label can therefore fail before it restores anything. The executable authority is the repository’s canonical compose-cli.sh chain. Step 12 adds only the Scanner database overlay and an ignored exact-candidate image overlay; the helper loads the durable local-estate image-ID file when present and keeps docker-compose.existing-networks.override.yml last. Step 2 preserves the outgoing image IDs under window-specific local tags and writes a matching ignored rollback overlay. The labels remain useful for drift evidence only; neither forward cut nor rollback reconstructs them.

H7 — resolved by the approved Q-25 producer/owner redesign; prove all three boundaries before the window. Concelier no longer creates, imports, exports, or writes scanner.artifact_boms and no longer references Scanner persistence. OfflineKit owns the imported-part lifecycle in its own database, extracts the Scanner artifact-BOM part, and activates it only through Scanner’s tenant-scoped owner endpoint. Scanner validates the tenant/body identity, canonical digest, and payload digest before writing its own table. Authority gives the activator client exactly the Scanner activation operation scope; the client fails closed when its secret, token, tenant, or owner API is unavailable. This removes the silent-divergence path instead of depending on a permission failure in the retired shared database. The go/no-go receipts are: zero Concelier Scanner-table or Scanner-persistence references, OfflineKit replay/ledger tests, Scanner owner-endpoint negative and idempotency tests, and a rendered activator client with the exact tenant, audiences, and one scope.

H8 — the role-identity trap: converge as the SERVICE ROLE, never as the superuser

Precedent, paid for live on 2026-08-28 by the NTF-9 window (SPRINT_20260722_015). notify-web was cut over to stellaops_notify as its own least-privilege role and crash-looped. Root cause: StartupMigrationHost.ApplyMigrationAsync runs migration DML, and every one of the 21 tables in the notify schema carries FORCE ROW LEVEL SECURITY with policies calling notify_app.require_current_tenant(). FORCE ROW LEVEL SECURITY subjects even the table owner to the policy, so the migration raised P0001: app.tenant_id session variable not set and the host died. It had never been seen because notify-web had always connected as the cluster superuser, and superusers bypass RLS unconditionally. The window was rolled back.

Converging the target as the superuser hides this entire class and hands the window a false green. The convergence must run under the same least-privilege role the service will use, and the identity must be recorded, not assumed — a connection string naming Username=scanner is not proof that the session is unprivileged.

Pre-window check (run it; do not take a number on faith): the executable source-and-target survey is in step 1, after the database and schema variables exist. Do not move that command above their definitions or reduce it to the source database only.

Scanner’s result, measured independently 2026-08-28: 0 FORCE / 9 RLS of 85. Scanner is CLEAR. Nine tables have RLS enabled but not forced, and a table owner bypasses non-forced RLS. The target agrees: the same query on stellaops_scanner returns 0 FORCE / 9 RLS, a per-table flag diff between the two databases is empty, and the baseline SQL contains zero FORCE ROW LEVEL SECURITY statements (its 24 ENABLE statements are the collapsed statement stream repeating the same 9 tables).

That “clear” rests on a detector proven to detect, not on an absent flag. In the target sandbox, acting as scanner with row_security=on:

• a table with FORCE ROW LEVEL SECURITY and a tenant-GUC policy rejected the insert with ERROR: app.tenant_id session variable not set — the identical NTF-9 failure;
• the control table with RLS enabled but not forced accepted it (INSERT 0 1), owner bypass intact.

For contrast, the same survey on sibling schemas in stellaops_platform: notify 21 FORCE / 21 RLS of 41, scheduler 13 / 13 of 21, binaries 11 FORCE / 14 RLS of 16. Those three windows will hit this wall; Scanner’s will not — but only if it converges under the service role, which is the only way the claim stays true rather than untested.

Pre-window bundle generation — detached and build-only

The tracked Scanner inventories and ignored payload are generated together. A detached build/probe worktree may generate them but, per GIT_WORKFLOW.md §7, it must never commit. Scanner Web currently mounts primary devops/plugins/scanner/base, so primary cannot receive the generated roots while either Scanner host is running. This pass only prepares and validates candidate bytes; the inventory commit and all exact-SHA image builds occur after step 4 stops both writers.

Do not execute this section until every non-candidate input in Required approvals and receipts below is satisfied. This section produces the candidate-specific receipts used by the remaining row.

Use one detached build-only worktree. CI=true is required because Directory.Build.props applies PathMap only in CI mode; without it, PDB paths can make the regenerated checksums depend on the worktree path. Scanner payload publishes also pass the latest committed src/ revision explicitly as both SourceRevisionId and RepositoryCommit. That revision changes when product source changes but remains stable across the inventory-only commit below; substituting HEAD recreates an unbounded DLL/PDB -> checksum -> manifest -> commit-SHA feedback loop.

Run this before the window while the primary checkout and its live bind mount remain untouched:

set -euo pipefail
SCN5_REPO_ROOT="$(git rev-parse --show-toplevel)"
SCN5_PRIMARY_PREP_SHA="$(git -C "${SCN5_REPO_ROOT}" rev-parse HEAD)"
SCN5_SCANNER_SOURCE_REVISION="$(git -C "${SCN5_REPO_ROOT}" log -1 --format=%H -- src)"
[[ "${SCN5_SCANNER_SOURCE_REVISION}" =~ ^[0-9a-f]{40}$ ]]
git -C "${SCN5_REPO_ROOT}" cat-file -e "${SCN5_SCANNER_SOURCE_REVISION}^{commit}"
SCN5_CANDIDATE_ENV="${SCN5_REPO_ROOT}/tmp/scn5-bundle-candidate.env"
test ! -e "${SCN5_CANDIDATE_ENV}"
SCN5_INVENTORY_WORKTREE="${SCN5_REPO_ROOT}/tmp/worktrees/scn5-inventory-${SCN5_PRIMARY_PREP_SHA:0:12}"
test ! -e "${SCN5_INVENTORY_WORKTREE}"
GIT_LFS_SKIP_SMUDGE=1 git -C "${SCN5_REPO_ROOT}" worktree add --detach \
  "${SCN5_INVENTORY_WORKTREE}" "${SCN5_PRIMARY_PREP_SHA}"
test -z "$(git -C "${SCN5_INVENTORY_WORKTREE}" branch --show-current)"
test "$(git -C "${SCN5_INVENTORY_WORKTREE}" rev-parse HEAD)" = \
  "${SCN5_PRIMARY_PREP_SHA}"
test -z "$(git -C "${SCN5_INVENTORY_WORKTREE}" status --porcelain=v1 --untracked-files=no)"

SCN5_INVENTORY_PARENT_SHA="$(git -C "${SCN5_INVENTORY_WORKTREE}" rev-parse HEAD)"
SCN5_INVENTORY_RECEIPTS="${SCN5_INVENTORY_WORKTREE}/tmp/scn5-bundle-inventory"
mkdir -p "${SCN5_INVENTORY_RECEIPTS}"
(
  cd -- "${SCN5_INVENTORY_WORKTREE}"
  CI=true IMAGE_TAG=scn5-inventory-prep \
    bash devops/docker/build-service-publish.sh scanner-worker
)

git -C "${SCN5_INVENTORY_WORKTREE}" diff --binary --no-ext-diff -- \
  devops/plugins/scanner/base \
  devops/plugins/scanner/scanner-full-analyzers \
  > "${SCN5_INVENTORY_RECEIPTS}/tracked-inventory.patch"
git -C "${SCN5_INVENTORY_WORKTREE}" diff --name-status -- \
  devops/plugins/scanner/base \
  devops/plugins/scanner/scanner-full-analyzers \
  > "${SCN5_INVENTORY_RECEIPTS}/tracked-inventory.name-status.tsv"
test -s "${SCN5_INVENTORY_RECEIPTS}/tracked-inventory.patch"
test -s "${SCN5_INVENTORY_RECEIPTS}/tracked-inventory.name-status.tsv"

# Only the two generated inventory roots may be tracked changes from this pass.
git -C "${SCN5_INVENTORY_WORKTREE}" diff --name-only \
  > "${SCN5_INVENTORY_RECEIPTS}/all-tracked-changes.txt"
git -C "${SCN5_INVENTORY_WORKTREE}" diff --name-only -- \
  devops/plugins/scanner/base \
  devops/plugins/scanner/scanner-full-analyzers \
  > "${SCN5_INVENTORY_RECEIPTS}/expected-tracked-changes.txt"
cmp "${SCN5_INVENTORY_RECEIPTS}/expected-tracked-changes.txt" \
    "${SCN5_INVENTORY_RECEIPTS}/all-tracked-changes.txt"
git -C "${SCN5_INVENTORY_WORKTREE}" diff --check -- \
  devops/plugins/scanner/base \
  devops/plugins/scanner/scanner-full-analyzers

# Read the generated tracked hunks now; repeat this review on primary after the
# stopped-writer promotion. Do not stage or commit in this detached worktree.
git -C "${SCN5_INVENTORY_WORKTREE}" diff -- \
  devops/plugins/scanner/base \
  devops/plugins/scanner/scanner-full-analyzers

{
  printf 'SCN5_PRIMARY_PREP_SHA=%q\n' "${SCN5_PRIMARY_PREP_SHA}"
  printf 'SCN5_SCANNER_SOURCE_REVISION=%q\n' "${SCN5_SCANNER_SOURCE_REVISION}"
  printf 'SCN5_INVENTORY_WORKTREE=%q\n' "${SCN5_INVENTORY_WORKTREE}"
} > "${SCN5_CANDIDATE_ENV}"

Verified by: the detached worktree remains at SCN5_PRIMARY_PREP_SHA, has no branch and no commit, its complete tracked diff is restricted to base and scanner-full-analyzers, its explicit Scanner assembly revision resolves to a real commit, and primary main is untouched. Keep it and tmp/scn5-bundle-candidate.env through the stopped-writer promotion.

Required approvals and receipts

The window is NO-GO until every row names a real reviewed artifact. Blank, TBD, verbal, or source-only evidence is a failure.

1. Establish the workspace

set -euo pipefail
# Git Bash/MSYS rewrites container-absolute paths such as /app/buildinfo.json.
# Scope path-conversion suppression to direct docker exec/cp calls only: the
# Compose helpers are child Bash processes and still need normal host-path
# conversion for this Windows checkout. Clear any inherited global workaround
# before defining the deliberately non-exported wrapper.
unset MSYS_NO_PATHCONV MSYS2_ARG_CONV_EXCL
unset -f docker 2>/dev/null || true
docker() {
  local index=0 psql_index=-1
  local -a original=("$@") forwarded=() sql_commands=()

  # This host's psql does not expand :variables inside a -c command string.
  # Route only direct docker-exec psql -c calls through stdin, where psql
  # substitution is active; preserve every other docker argument unchanged.
  if test "${original[0]:-}" = exec; then
    for index in "${!original[@]}"; do
      test "${original[$index]}" != psql || psql_index="${index}"
    done
  fi

  if test "${psql_index}" -ge 0; then
    index=0
    while test "${index}" -lt "${#original[@]}"; do
      if test "${index}" -eq 0; then
        forwarded+=(exec -i)
        index=$((index + 1))
      elif test "${original[$index]}" = -c; then
        test "$((index + 1))" -lt "${#original[@]}"
        sql_commands+=("${original[$((index + 1))]}")
        index=$((index + 2))
      else
        forwarded+=("${original[$index]}")
        index=$((index + 1))
      fi
    done
    if test "${#sql_commands[@]}" -gt 0; then
      printf '%s\n' "${sql_commands[@]}" \
        | MSYS_NO_PATHCONV=1 command docker "${forwarded[@]}"
      return
    fi
  fi

  case "${original[0]:-}" in
    exec|cp) MSYS_NO_PATHCONV=1 command docker "${original[@]}" ;;
    *) command docker "${original[@]}" ;;
  esac
}
test "$(bash -lc 'type -t docker')" = file
SCN5_REPO_ROOT="$(git rev-parse --show-toplevel)"
test -n "${SCN5_REPO_ROOT}"
test -d "${SCN5_REPO_ROOT}"
cd -- "${SCN5_REPO_ROOT}"
SCN5_CANDIDATE_ENV="${SCN5_REPO_ROOT}/tmp/scn5-bundle-candidate.env"
test -f "${SCN5_CANDIDATE_ENV}"
SCN5_LOCAL_ESTATE_PIN="${SCN5_REPO_ROOT}/devops/compose/docker-compose.local-estate-image-pins.override.yml"
test ! -e "${SCN5_LOCAL_ESTATE_PIN}"
# Values are Bash-escaped with printf %q by the pre-window materialization.
# shellcheck disable=SC1090
. "${SCN5_CANDIDATE_ENV}"
: "${SCN5_PRIMARY_PREP_SHA:?Load the exact value from the pre-window candidate.env receipt}"
: "${SCN5_INVENTORY_WORKTREE:?Load the exact value from the pre-window candidate.env receipt}"
test "$(git branch --show-current)" = main
test "$(git rev-parse HEAD)" = "${SCN5_PRIMARY_PREP_SHA}"
test "$(git -C "${SCN5_INVENTORY_WORKTREE}" rev-parse --show-toplevel)" = \
  "${SCN5_INVENTORY_WORKTREE}"
test -z "$(git -C "${SCN5_INVENTORY_WORKTREE}" branch --show-current)"
test "$(git -C "${SCN5_INVENTORY_WORKTREE}" rev-parse HEAD)" = \
  "${SCN5_PRIMARY_PREP_SHA}"
SCN5_WINDOW_ID="$(date -u +%Y%m%dT%H%M%SZ)"
SCN5_WINDOW_EPOCH_MS="$(date -u +%s%3N)"
SCN5_ROOT="tmp/scn5-${SCN5_WINDOW_ID}"
mkdir -p "${SCN5_ROOT}"/{topology,backup,parity,shape,logs}
SCN5_COMPOSE_ROOT_REL="../../${SCN5_ROOT}"

SCN5_BUNDLE_SOURCE_BASE="${SCN5_INVENTORY_WORKTREE}/devops/plugins/scanner/base"
SCN5_BUNDLE_SOURCE_FULL="${SCN5_INVENTORY_WORKTREE}/devops/plugins/scanner/scanner-full-analyzers"
SCN5_BUNDLE_TARGET_PARENT="${SCN5_REPO_ROOT}/devops/plugins/scanner"
SCN5_BUNDLE_TARGET_BASE="${SCN5_BUNDLE_TARGET_PARENT}/base"
SCN5_BUNDLE_TARGET_FULL="${SCN5_BUNDLE_TARGET_PARENT}/scanner-full-analyzers"
test "$(cd -- "${SCN5_INVENTORY_WORKTREE}" && pwd -P)" != \
  "$(cd -- "${SCN5_REPO_ROOT}" && pwd -P)"
test "$(cd -- "${SCN5_BUNDLE_TARGET_PARENT}" && pwd -P)" = \
  "$(cd -- "${SCN5_REPO_ROOT}/devops/plugins/scanner" && pwd -P)"
test "${SCN5_BUNDLE_TARGET_BASE}" = \
  "${SCN5_REPO_ROOT}/devops/plugins/scanner/base"
test "${SCN5_BUNDLE_TARGET_FULL}" = \
  "${SCN5_REPO_ROOT}/devops/plugins/scanner/scanner-full-analyzers"
test -d "${SCN5_BUNDLE_SOURCE_BASE}"
test -d "${SCN5_BUNDLE_SOURCE_FULL}"

SCN5_PG_CONTAINER=stellaops-postgres
SCN5_PG_OPERATOR=stellaops
SCN5_SCANNER_ROLE=scanner
SCN5_SOURCE_DB=stellaops_platform
SCN5_TARGET_DB=stellaops_scanner
SCN5_POLICY_DB=stellaops_policy
SCN5_POLICY_COMPATIBILITY_ID=3f7dc64d-9d0b-4cb7-9a3a-3b7cb9a4e3d1
SCN5_SCHEMA=scanner
SCN5_TENANT=default
SCN5_TENANT_STREAM="scanner.scans.tenant.$(printf '%s' "${SCN5_TENANT}" | od -An -tx1 | tr -d ' \n')"

# Derive the receipt-authentication key in memory from the already-required
# Scanner role secret. The secret travels only on stdin; neither it nor the
# derived key is persisted. Reloading that named secret reproduces the key if
# the operator shell must be reconstructed for rollback.
set +x
: "${SCANNER_POSTGRES_PASSWORD:?Inject the Scanner role password by name; do not print it}"
SCN5_TOPOLOGY_HMAC_KEY="$(
  printf '%s' "${SCANNER_POSTGRES_PASSWORD}" \
    | python -c 'import hashlib,hmac,sys; print(hmac.new(sys.stdin.buffer.read(), b"stellaops/scn5/topology-env-contract/v1", hashlib.sha256).hexdigest())'
)"
[[ "${SCN5_TOPOLOGY_HMAC_KEY}" =~ ^[0-9a-f]{64}$ ]]

scn5_render_topology() { # $1 = names-only receipt; never persist rendered secret values
  local output="$1"
  local -a pipeline_rc
  set +e
  ./devops/compose/scripts/compose-cli.sh config --format json \
    | jq '{services: (.services | with_entries(
        select(.key == "scanner-web" or .key == "scanner-worker")
        | .value |= {
            image: .image,
            environmentKeys: ((.environment // {}) | keys),
            scannerBaseSource: ([.volumes[]?
              | select(.target == "/app/plugins/scanner/base") | .source]
              | if length == 1 then (.[0] | gsub("\\\\"; "/"))
                else error("expected one Scanner base volume") end)
          }))}' \
    > "${output}"
  pipeline_rc=("${PIPESTATUS[@]}")
  set -e
  test "${pipeline_rc[0]}" -eq 0
  test "${pipeline_rc[1]}" -eq 0
}

scn5_assert_rollback_connection_contract() {
  local -a pipeline_rc
  : "${SCN5_ROLLBACK_FILES:?Set the rollback overlay before asserting it}"
  set +e
  COMPOSE_EXTRA_FILES="${SCN5_ROLLBACK_FILES}" \
    ./devops/compose/scripts/compose-cli.sh config --format json \
    | jq -e '
        .services["scanner-web"].environment as $web
        | .services["scanner-worker"].environment as $worker
        | ($web["STELLAOPS_POSTGRES_SCANNER_CONNECTION"]
            | type == "string" and length > 0)
          and $web["STELLAOPS_POSTGRES_SCANNER_CONNECTION"]
            == $web["SCANNER_SCANNER__STORAGE__DSN"]
          and $worker["STELLAOPS_POSTGRES_SCANNER_CONNECTION"]
            == $worker["ScannerStorage__Postgres__ConnectionString"]
          and $web["STELLAOPS_POSTGRES_SCANNER_CONNECTION"]
            == $worker["STELLAOPS_POSTGRES_SCANNER_CONNECTION"]
      ' >/dev/null
  pipeline_rc=("${PIPESTATUS[@]}")
  set -e
  test "${#pipeline_rc[@]}" -eq 2
  test "${pipeline_rc[0]}" -eq 0
  test "${pipeline_rc[1]}" -eq 0
}

scn5_capture_env_names() { # $1=container, $2=names-only receipt
  local container="$1" output="$2"
  SCN5_INSPECT_CONTAINER="${container}" python - <<'PY' > "${output}"
import json
import os
import subprocess

container = os.environ["SCN5_INSPECT_CONTAINER"]
document = json.loads(subprocess.check_output(["docker", "inspect", container], text=True))[0]
names = {entry.partition("=")[0] for entry in document["Config"].get("Env") or []}
for name in sorted(names):
    print(name)
PY
}

scn5_capture_image_env_names() { # $1=image, $2=names-only receipt
  local image="$1" output="$2"
  SCN5_INSPECT_IMAGE="${image}" python - <<'PY' > "${output}"
import json
import os
import subprocess

image = os.environ["SCN5_INSPECT_IMAGE"]
raw = subprocess.check_output(
    ["docker", "image", "inspect", image, "--format", "{{json .Config.Env}}"],
    text=True,
)
names = {entry.partition("=")[0] for entry in (json.loads(raw) or [])}
for name in sorted(names):
    print(name)
PY
}

scn5_capture_env_hmac() { # $1=container, $2=keyed-HMAC receipt, $3=optional excluded-name regex
  local container="$1" output="$2" exclude_regex="${3:-}"
  SCN5_INSPECT_CONTAINER="${container}" \
    SCN5_ENV_EXCLUDE_REGEX="${exclude_regex}" \
    SCN5_TOPOLOGY_HMAC_KEY="${SCN5_TOPOLOGY_HMAC_KEY}" \
    python - <<'PY' > "${output}"
import hashlib
import hmac
import json
import os
import re
import subprocess

container = os.environ["SCN5_INSPECT_CONTAINER"]
exclude_regex = os.environ["SCN5_ENV_EXCLUDE_REGEX"]
child_environment = os.environ.copy()
child_environment.pop("SCN5_INSPECT_CONTAINER", None)
child_environment.pop("SCN5_ENV_EXCLUDE_REGEX", None)
child_environment.pop("SCN5_TOPOLOGY_HMAC_KEY", None)
document = json.loads(subprocess.check_output(
    ["docker", "inspect", container], env=child_environment, text=True
))[0]
environment = {}
for entry in document["Config"].get("Env") or []:
    name, separator, value = entry.partition("=")
    if not separator:
        raise SystemExit(f"environment entry without '=' in {container}")
    if exclude_regex and re.fullmatch(exclude_regex, name):
        continue
    environment[name] = value
payload = "".join(f"{name}={environment[name]}\n" for name in sorted(environment)).encode()
key = bytes.fromhex(os.environ["SCN5_TOPOLOGY_HMAC_KEY"])
print(hmac.new(key, payload, hashlib.sha256).hexdigest())
PY
}

scn5_capture_rendered_env_hmac() { # $1=service, $2=image, $3=receipt, $4=optional excluded-name regex
  local service="$1" image="$2" output="$3" exclude_regex="${4:-}"
  SCN5_EXPECTED_SERVICE="${service}" SCN5_EXPECTED_IMAGE="${image}" \
    SCN5_ENV_EXCLUDE_REGEX="${exclude_regex}" \
    SCN5_TOPOLOGY_HMAC_KEY="${SCN5_TOPOLOGY_HMAC_KEY}" \
    python - <<'PY' > "${output}"
import hashlib
import hmac
import json
import os
import re
import subprocess

service = os.environ["SCN5_EXPECTED_SERVICE"]
image = os.environ["SCN5_EXPECTED_IMAGE"]
exclude_regex = os.environ["SCN5_ENV_EXCLUDE_REGEX"]
child_environment = os.environ.copy()
child_environment.pop("SCN5_EXPECTED_SERVICE", None)
child_environment.pop("SCN5_EXPECTED_IMAGE", None)
child_environment.pop("SCN5_ENV_EXCLUDE_REGEX", None)
child_environment.pop("SCN5_TOPOLOGY_HMAC_KEY", None)
image_raw = subprocess.check_output(
    ["docker", "image", "inspect", image, "--format", "{{json .Config.Env}}"],
    env=child_environment,
    text=True,
)
render_raw = subprocess.check_output(
    ["bash", "./devops/compose/scripts/compose-cli.sh", "config", "--format", "json"],
    env=child_environment,
    text=True,
)

environment = {}
for entry in json.loads(image_raw) or []:
    name, separator, value = entry.partition("=")
    if not separator:
        raise SystemExit(f"image environment entry without '=' in {image}")
    if exclude_regex and re.fullmatch(exclude_regex, name):
        continue
    environment[name] = value
for name, value in (json.loads(render_raw)["services"][service].get("environment") or {}).items():
    if exclude_regex and re.fullmatch(exclude_regex, name):
        continue
    if value is None:
        environment.pop(name, None)
    elif isinstance(value, bool):
        environment[name] = str(value).lower()
    else:
        environment[name] = str(value)

payload = "".join(f"{name}={environment[name]}\n" for name in sorted(environment)).encode()
key = bytes.fromhex(os.environ["SCN5_TOPOLOGY_HMAC_KEY"])
print(hmac.new(key, payload, hashlib.sha256).hexdigest())
PY
}

scn5_capture_runtime_topology() { # $1=container, $2=sanitized stable receipt
  local container="$1" output="$2"
  local -a pipeline_rc
  set +e
  docker inspect "${container}" \
    | jq '
      def sorted_object: (to_entries | sort_by(.key) | from_entries);
      .[0] | {
        config: {
          attachStdin: .Config.AttachStdin,
          attachStdout: .Config.AttachStdout,
          attachStderr: .Config.AttachStderr,
          domainname: .Config.Domainname,
          entrypoint: .Config.Entrypoint,
          cmd: .Config.Cmd,
          user: .Config.User,
          workingDir: .Config.WorkingDir,
          stopSignal: .Config.StopSignal,
          healthcheck: .Config.Healthcheck,
          exposedPorts: ((.Config.ExposedPorts // {}) | keys | sort),
          tty: .Config.Tty,
          openStdin: .Config.OpenStdin,
          stdinOnce: .Config.StdinOnce
        },
        host: {
          autoRemove: .HostConfig.AutoRemove,
          runtime: .HostConfig.Runtime,
          isolation: .HostConfig.Isolation,
          networkMode: .HostConfig.NetworkMode,
          publishAllPorts: .HostConfig.PublishAllPorts,
          portBindings: ((.HostConfig.PortBindings // {}) | to_entries | sort_by(.key)
            | map({port: .key, bindings: ((.value // []) | sort_by([.HostIp, .HostPort]))})),
          restartPolicy: .HostConfig.RestartPolicy,
          readonlyRootfs: .HostConfig.ReadonlyRootfs,
          privileged: .HostConfig.Privileged,
          capAdd: ((.HostConfig.CapAdd // []) | sort),
          capDrop: ((.HostConfig.CapDrop // []) | sort),
          securityOpt: ((.HostConfig.SecurityOpt // []) | sort),
          dns: ((.HostConfig.Dns // []) | sort),
          dnsOptions: ((.HostConfig.DnsOptions // []) | sort),
          dnsSearch: ((.HostConfig.DnsSearch // []) | sort),
          extraHosts: ((.HostConfig.ExtraHosts // []) | sort),
          groupAdd: ((.HostConfig.GroupAdd // []) | sort),
          ipcMode: .HostConfig.IpcMode,
          pidMode: .HostConfig.PidMode,
          utsMode: .HostConfig.UTSMode,
          usernsMode: .HostConfig.UsernsMode,
          cgroupnsMode: .HostConfig.CgroupnsMode,
          init: .HostConfig.Init,
          logConfig: {
            type: .HostConfig.LogConfig.Type,
            config: ((.HostConfig.LogConfig.Config // {}) | sorted_object)
          },
          memory: .HostConfig.Memory,
          memoryReservation: .HostConfig.MemoryReservation,
          memorySwap: .HostConfig.MemorySwap,
          memorySwappiness: .HostConfig.MemorySwappiness,
          nanoCpus: .HostConfig.NanoCpus,
          cpuShares: .HostConfig.CpuShares,
          cpuPeriod: .HostConfig.CpuPeriod,
          cpuQuota: .HostConfig.CpuQuota,
          cpusetCpus: .HostConfig.CpusetCpus,
          cpusetMems: .HostConfig.CpusetMems,
          pidsLimit: .HostConfig.PidsLimit,
          oomKillDisable: .HostConfig.OomKillDisable,
          oomScoreAdj: .HostConfig.OomScoreAdj,
          shmSize: .HostConfig.ShmSize,
          ulimits: ((.HostConfig.Ulimits // []) | sort_by(tostring)),
          devices: ((.HostConfig.Devices // []) | sort_by(tostring)),
          deviceCgroupRules: ((.HostConfig.DeviceCgroupRules // []) | sort),
          deviceRequests: ((.HostConfig.DeviceRequests // []) | sort_by(tostring)),
          sysctls: ((.HostConfig.Sysctls // {}) | sorted_object),
          storageOpt: ((.HostConfig.StorageOpt // {}) | sorted_object),
          tmpfs: ((.HostConfig.Tmpfs // {}) | sorted_object)
        },
        mounts: ((.Mounts // []) | map({
          type: .Type,
          name: (.Name // ""),
          source: .Source,
          destination: .Destination,
          driver: (.Driver // ""),
          mode: .Mode,
          rw: .RW,
          propagation: .Propagation
        }) | sort_by([.destination, .source])),
        networks: ((.NetworkSettings.Networks // {}) | to_entries | map({
          name: .key,
          aliases: ((.value.Aliases // []) | sort),
          ipamConfig: ((.value.IPAMConfig // {}) | sorted_object),
          driverOpts: ((.value.DriverOpts // {}) | sorted_object)
        }) | sort_by(.name))
      }' > "${output}"
  pipeline_rc=("${PIPESTATUS[@]}")
  set -e
  test "${#pipeline_rc[@]}" -eq 2
  test "${pipeline_rc[0]}" -eq 0
  test "${pipeline_rc[1]}" -eq 0
}

scn5_assert_local_estate_pin() { # $1=authority $2=policy $3=web $4=worker
  local authority_id="$1" policy_id="$2" web_id="$3" worker_id="$4" service image_id
  local -a render_rc
  test -f "${SCN5_LOCAL_ESTATE_PIN}"
  git check-ignore -q -- "${SCN5_LOCAL_ESTATE_PIN}"
  test "$(grep -Ec '^  (authority|policy-engine|scanner-web|scanner-worker):$' \
    "${SCN5_LOCAL_ESTATE_PIN}")" -eq 4
  for service in authority policy-engine scanner-web scanner-worker; do
    test "$(grep -Fxc "  ${service}:" "${SCN5_LOCAL_ESTATE_PIN}")" -eq 1
  done
  test "$(grep -Ec '^    image: sha256:[0-9a-f]{64}$' \
    "${SCN5_LOCAL_ESTATE_PIN}")" -eq 4
  test "$(grep -Fxc '    pull_policy: never' "${SCN5_LOCAL_ESTATE_PIN}")" -eq 4
  test -z "$(grep -Ev \
    '^(services:|  (authority|policy-engine|scanner-web|scanner-worker):|    image: sha256:[0-9a-f]{64}|    pull_policy: never|[[:space:]]*)$' \
    "${SCN5_LOCAL_ESTATE_PIN}")"
  set +e
  (unset COMPOSE_EXTRA_FILES
   ./devops/compose/scripts/compose-cli.sh config --format json) \
    | jq -e --arg authority "${authority_id}" --arg policy "${policy_id}" \
        --arg web "${web_id}" --arg worker "${worker_id}" '
        .services.authority.image == $authority
        and .services["policy-engine"].image == $policy
        and .services["scanner-web"].image == $web
        and .services["scanner-worker"].image == $worker' >/dev/null
  render_rc=("${PIPESTATUS[@]}")
  set -e
  test "${#render_rc[@]}" -eq 2
  test "${render_rc[0]}" -eq 0
  test "${render_rc[1]}" -eq 0
  for image_id in "${authority_id}" "${policy_id}" "${web_id}" "${worker_id}"; do
    [[ "${image_id}" =~ ^sha256:[0-9a-f]{64}$ ]]
    test "$(docker image inspect "${image_id}" --format '{{.Id}}')" = "${image_id}"
  done
}

scn5_fingerprint() {   # $1 = database to run against, $2 = output file
  local gen
  gen=$(docker exec "${SCN5_PG_CONTAINER}" psql -XqAt -U "${SCN5_PG_OPERATOR}" -d "${SCN5_SOURCE_DB}" -c "
    SELECT string_agg(
      format('SELECT %L::text AS t, count(*)::text AS n, coalesce(md5(string_agg(h, %L ORDER BY h)), %L) AS d FROM (SELECT md5(x.*::text) AS h FROM ${SCN5_SCHEMA}.%I x) s',
             c.relname, '', '<empty>', c.relname),
      ' UNION ALL ')
      FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace
     WHERE n.nspname = '${SCN5_SCHEMA}' AND c.relkind = 'r'
       AND c.relname <> 'schema_migrations';")
  docker exec "${SCN5_PG_CONTAINER}" psql -XqAt -F $'\t' -U "${SCN5_PG_OPERATOR}" -d "$1" \
    -c "SET TimeZone='UTC'; SET extra_float_digits = 3;" \
    -c "${gen} ORDER BY 1" > "$2"
}

scn5_assert_copy_fingerprint() { # $1 = fingerprint over the frozen 84-table copy set
  local input="$1" nonempty_summary
  test "$(wc -l < "${input}")" -eq 84
  awk -F'\t' '
    NF != 3 || $1 == "" || $2 !~ /^[0-9]+$/ || $3 == "" { exit 1 }
    END { if (NR != 84) exit 1 }
  ' "${input}"
  nonempty_summary="$(awk -F'\t' '$2 > 0 { rows += $2; tables++ }
    END { printf "%d|%d", tables + 0, rows + 0 }' "${input}")"
  test "${nonempty_summary}" = '14|135'
}

SCN5_SHAPE_PROJECTIONS=(
  relations columns indexes constraints sequences partitions
  types views functions triggers
)

scn5_capture_shape() { # $1 = database, $2 = receipt label
  local database="$1" label="$2" output_root
  [[ "${label}" =~ ^[a-z0-9-]+$ ]]
  output_root="${SCN5_ROOT}/shape/${label}"
  mkdir -p "${output_root}"

  docker exec "${SCN5_PG_CONTAINER}" psql -XqAt -F '|' -v ON_ERROR_STOP=1 \
    -U "${SCN5_PG_OPERATOR}" -d "${database}" -c \
    "SELECT c.relkind, c.relname
       FROM pg_class c
       JOIN pg_namespace n ON n.oid = c.relnamespace
      WHERE n.nspname = '${SCN5_SCHEMA}'
        AND c.relkind IN ('r','p','v','m','S','f')
      ORDER BY 1, 2;" \
    > "${output_root}/relations.tsv"

  docker exec "${SCN5_PG_CONTAINER}" psql -XqAt -F '|' -v ON_ERROR_STOP=1 \
    -U "${SCN5_PG_OPERATOR}" -d "${database}" -c \
    "SELECT cols.table_name,
            cols.column_name,
            cols.data_type,
            coalesce(cols.character_maximum_length::text, ''),
            coalesce(cols.numeric_precision::text, ''),
            coalesce(cols.numeric_scale::text, ''),
            cols.is_nullable,
            coalesce(cols.column_default, ''),
            cols.is_identity,
            coalesce(cols.identity_generation, ''),
            cols.udt_name
       FROM information_schema.columns cols
      WHERE cols.table_schema = '${SCN5_SCHEMA}'
      ORDER BY 1, 2;" \
    > "${output_root}/columns.tsv"

  docker exec "${SCN5_PG_CONTAINER}" psql -XqAt -F '|' -v ON_ERROR_STOP=1 \
    -U "${SCN5_PG_OPERATOR}" -d "${database}" -c \
    "SELECT tablename, indexname, indexdef
       FROM pg_indexes
      WHERE schemaname = '${SCN5_SCHEMA}'
      ORDER BY 1, 2, 3;" \
    > "${output_root}/indexes.tsv"

  docker exec "${SCN5_PG_CONTAINER}" psql -XqAt -F '|' -v ON_ERROR_STOP=1 \
    -U "${SCN5_PG_OPERATOR}" -d "${database}" -c \
    "SELECT c.relname,
            con.conname,
            con.contype,
            pg_get_constraintdef(con.oid)
       FROM pg_constraint con
       JOIN pg_class c ON c.oid = con.conrelid
       JOIN pg_namespace n ON n.oid = c.relnamespace
      WHERE n.nspname = '${SCN5_SCHEMA}'
      ORDER BY 1, 2, 3, 4;" \
    > "${output_root}/constraints.tsv"

  docker exec "${SCN5_PG_CONTAINER}" psql -XqAt -F '|' -v ON_ERROR_STOP=1 \
    -U "${SCN5_PG_OPERATOR}" -d "${database}" -c \
    "SELECT seq_class.relname,
            format_type(seq.seqtypid, NULL),
            seq.seqstart,
            seq.seqincrement,
            seq.seqmin,
            seq.seqmax,
            seq.seqcache,
            seq.seqcycle,
            coalesce(owner_class.relname, ''),
            coalesce(owner_attribute.attname, ''),
            coalesce(dependency.deptype::text, '')
       FROM pg_class seq_class
       JOIN pg_namespace n ON n.oid = seq_class.relnamespace
       JOIN pg_sequence seq ON seq.seqrelid = seq_class.oid
       LEFT JOIN pg_depend dependency
         ON dependency.classid = 'pg_class'::regclass
        AND dependency.objid = seq_class.oid
        AND dependency.refclassid = 'pg_class'::regclass
        AND dependency.deptype IN ('a','i')
       LEFT JOIN pg_class owner_class ON owner_class.oid = dependency.refobjid
       LEFT JOIN pg_attribute owner_attribute
         ON owner_attribute.attrelid = dependency.refobjid
        AND owner_attribute.attnum = dependency.refobjsubid
      WHERE n.nspname = '${SCN5_SCHEMA}'
      ORDER BY 1, 9, 10, 11;" \
    > "${output_root}/sequences.tsv"

  docker exec "${SCN5_PG_CONTAINER}" psql -XqAt -F '|' -v ON_ERROR_STOP=1 \
    -U "${SCN5_PG_OPERATOR}" -d "${database}" -c \
    "SELECT parent.relname,
            pg_get_partkeydef(parent.oid),
            child.relname,
            pg_get_expr(child.relpartbound, child.oid)
       FROM pg_class parent
       JOIN pg_namespace n ON n.oid = parent.relnamespace
       JOIN pg_inherits inheritance ON inheritance.inhparent = parent.oid
       JOIN pg_class child ON child.oid = inheritance.inhrelid
      WHERE n.nspname = '${SCN5_SCHEMA}'
        AND parent.relkind = 'p'
        AND child.relispartition
      ORDER BY 1, 3, 4;" \
    > "${output_root}/partitions.tsv"

  docker exec "${SCN5_PG_CONTAINER}" psql -XqAt -F '|' -v ON_ERROR_STOP=1 \
    -U "${SCN5_PG_OPERATOR}" -d "${database}" -c \
    "SELECT t.typname,
            t.typtype,
            CASE WHEN t.typbasetype = 0 THEN ''
                 ELSE format_type(t.typbasetype, t.typtypmod) END,
            CASE WHEN range_type.rngsubtype IS NULL THEN ''
                 ELSE format_type(range_type.rngsubtype, NULL) END,
            coalesce(string_agg(enum_type.enumlabel, ',' ORDER BY enum_type.enumsortorder), '')
       FROM pg_type t
       JOIN pg_namespace n ON n.oid = t.typnamespace
       LEFT JOIN pg_class composite ON composite.oid = t.typrelid
       LEFT JOIN pg_range range_type ON range_type.rngtypid = t.oid
       LEFT JOIN pg_enum enum_type ON enum_type.enumtypid = t.oid
      WHERE n.nspname = '${SCN5_SCHEMA}'
        AND t.typtype IN ('e','c','d','r')
        AND (t.typtype <> 'c' OR composite.relkind = 'c')
      GROUP BY t.typname, t.typtype, t.typbasetype, t.typtypmod, range_type.rngsubtype
      ORDER BY 1, 2;" \
    > "${output_root}/types.tsv"

  docker exec "${SCN5_PG_CONTAINER}" psql -XqAt -F '|' -v ON_ERROR_STOP=1 \
    -U "${SCN5_PG_OPERATOR}" -d "${database}" -c \
    "SELECT c.relkind, c.relname, md5(pg_get_viewdef(c.oid, true))
       FROM pg_class c
       JOIN pg_namespace n ON n.oid = c.relnamespace
      WHERE n.nspname = '${SCN5_SCHEMA}'
        AND c.relkind IN ('v','m')
      ORDER BY 1, 2;" \
    > "${output_root}/views.tsv"

  docker exec "${SCN5_PG_CONTAINER}" psql -XqAt -F '|' -v ON_ERROR_STOP=1 \
    -U "${SCN5_PG_OPERATOR}" -d "${database}" -c \
    "SELECT p.proname,
            p.prokind,
            pg_get_function_identity_arguments(p.oid),
            md5(pg_get_functiondef(p.oid))
       FROM pg_proc p
       JOIN pg_namespace n ON n.oid = p.pronamespace
      WHERE n.nspname = '${SCN5_SCHEMA}'
      ORDER BY 1, 2, 3, 4;" \
    > "${output_root}/functions.tsv"

  docker exec "${SCN5_PG_CONTAINER}" psql -XqAt -F '|' -v ON_ERROR_STOP=1 \
    -U "${SCN5_PG_OPERATOR}" -d "${database}" -c \
    "SELECT c.relname, trigger.tgname, pg_get_triggerdef(trigger.oid, true)
       FROM pg_trigger trigger
       JOIN pg_class c ON c.oid = trigger.tgrelid
       JOIN pg_namespace n ON n.oid = c.relnamespace
      WHERE n.nspname = '${SCN5_SCHEMA}'
        AND NOT trigger.tgisinternal
      ORDER BY 1, 2, 3;" \
    > "${output_root}/triggers.tsv"

  docker exec "${SCN5_PG_CONTAINER}" psql -XqAt -F '|' -v ON_ERROR_STOP=1 \
    -U "${SCN5_PG_OPERATOR}" -d "${database}" -c \
    "SELECT c.relname, c.relrowsecurity, c.relforcerowsecurity,
            policy.polname, pg_get_expr(policy.polqual, policy.polrelid)
       FROM pg_class c
       JOIN pg_namespace n ON n.oid = c.relnamespace
       LEFT JOIN pg_policy policy ON policy.polrelid = c.oid
      WHERE n.nspname = '${SCN5_SCHEMA}'
        AND c.relkind IN ('r','p')
        AND (c.relrowsecurity OR c.relforcerowsecurity)
      ORDER BY 1, 4, 5;" \
    > "${output_root}/rls.tsv"

  # PostgreSQL's database collation is not the authority for `comm`. Pin every
  # receipt to byte ordering so the set comparison is valid on every host.
  local projection
  for projection in "${SCN5_SHAPE_PROJECTIONS[@]}" rls; do
    LC_ALL=C sort -o "${output_root}/${projection}.tsv" \
      "${output_root}/${projection}.tsv"
  done
}

scn5_assert_shape_contains_source() { # $1 = pre-copy or post-restore receipt label
  local label="$1" projection source_rows target_rows missing_rows extra_rows failures=0
  case "${label}" in
    pre-copy|post-restore) ;;
    *) printf 'Unsupported shape-comparison label: %s\n' "${label}" >&2; return 2 ;;
  esac
  scn5_capture_shape "${SCN5_SOURCE_DB}" "${label}-source"
  scn5_capture_shape "${SCN5_TARGET_DB}" "${label}-target"

  printf 'projection|source|target|missing|extra\n' \
    > "${SCN5_ROOT}/shape/${label}-summary.txt"
  for projection in "${SCN5_SHAPE_PROJECTIONS[@]}"; do
    comm -23 \
      "${SCN5_ROOT}/shape/${label}-source/${projection}.tsv" \
      "${SCN5_ROOT}/shape/${label}-target/${projection}.tsv" \
      > "${SCN5_ROOT}/shape/${label}-${projection}.missing.tsv"
    comm -13 \
      "${SCN5_ROOT}/shape/${label}-source/${projection}.tsv" \
      "${SCN5_ROOT}/shape/${label}-target/${projection}.tsv" \
      > "${SCN5_ROOT}/shape/${label}-${projection}.extra.tsv"
    source_rows="$(wc -l < "${SCN5_ROOT}/shape/${label}-source/${projection}.tsv")"
    target_rows="$(wc -l < "${SCN5_ROOT}/shape/${label}-target/${projection}.tsv")"
    missing_rows="$(wc -l < "${SCN5_ROOT}/shape/${label}-${projection}.missing.tsv")"
    extra_rows="$(wc -l < "${SCN5_ROOT}/shape/${label}-${projection}.extra.tsv")"
    printf '%s|%s|%s|%s|%s\n' \
      "${projection}" "${source_rows}" "${target_rows}" "${missing_rows}" "${extra_rows}" \
      >> "${SCN5_ROOT}/shape/${label}-summary.txt"
    test "${missing_rows}" -eq 0 || failures=$((failures + 1))

    if test "${label}" = pre-copy; then
      # Every target-only object must be owned by migration 001's EPSS surface.
      # Check every identity/definition field: columns, indexes, constraints,
      # partitions, and triggers can be EPSS-owned while their parent table is not.
      awk -F'|' '
        {
          is_epss = 0
          for (field = 1; field <= NF; field++) {
            if ($field ~ /(^|[^[:alnum:]])epss([^[:alnum:]]|$)/) {
              is_epss = 1
              break
            }
          }
          if (!is_epss) print
        }
      ' "${SCN5_ROOT}/shape/${label}-${projection}.extra.tsv" \
        > "${SCN5_ROOT}/shape/${label}-${projection}.non-epss-extra.tsv"
      test ! -s "${SCN5_ROOT}/shape/${label}-${projection}.non-epss-extra.tsv" \
        || failures=$((failures + 1))
    else
      if ! cmp "${SCN5_ROOT}/shape/pre-copy-${projection}.extra.tsv" \
          "${SCN5_ROOT}/shape/${label}-${projection}.extra.tsv"; then
        comm -3 \
          "${SCN5_ROOT}/shape/pre-copy-${projection}.extra.tsv" \
          "${SCN5_ROOT}/shape/${label}-${projection}.extra.tsv" \
          > "${SCN5_ROOT}/shape/${label}-${projection}.extra.diff.tsv"
        failures=$((failures + 1))
      fi
    fi
  done

  comm -3 \
    "${SCN5_ROOT}/shape/${label}-source/rls.tsv" \
    "${SCN5_ROOT}/shape/${label}-target/rls.tsv" \
    > "${SCN5_ROOT}/shape/${label}-rls.diff.tsv"
  test ! -s "${SCN5_ROOT}/shape/${label}-rls.diff.tsv" \
    || failures=$((failures + 1))

  sed -n '1,20p' "${SCN5_ROOT}/shape/${label}-summary.txt"
  test "${failures}" -eq 0
}

Prove mechanically that the receipt directory exists and is excluded by the repository’s tmp/ rule. Persist only the ignore-rule receipt; no secret values are rendered:

test -d "${SCN5_ROOT}"
git check-ignore -v -- "${SCN5_ROOT}" \
  > "${SCN5_ROOT}/topology/receipt-path-ignore-rule.txt"
test "$(wc -l < "${SCN5_ROOT}/topology/receipt-path-ignore-rule.txt")" -eq 1
grep -Eq '^\.gitignore:[0-9]+:tmp/[[:space:]]+tmp/scn5-' \
  "${SCN5_ROOT}/topology/receipt-path-ignore-rule.txt"

Verified by: the receipt exists as one line and names .gitignore’s tmp/ rule plus this window’s tmp/scn5-... path. If any assertion fails, stop — receipts must not become commits.

Run H8’s FORCE/RLS survey now that every variable exists, against both databases. This assertion requires zero FORCE tables, exactly nine RLS-enabled leaf tables, a non-empty table census, and the two expected database names in order:

SCN5_FORCE_RLS_RECEIPT="${SCN5_ROOT}/shape/source-target-force-rls.txt"
: > "${SCN5_FORCE_RLS_RECEIPT}"
for database in "${SCN5_SOURCE_DB}" "${SCN5_TARGET_DB}"; do
  printf '%s|' "${database}" >> "${SCN5_FORCE_RLS_RECEIPT}"
  docker exec "${SCN5_PG_CONTAINER}" psql -XqAt -F '|' -v ON_ERROR_STOP=1 \
    -U "${SCN5_PG_OPERATOR}" -d "${database}" -c \
    "SELECT count(*) FILTER (WHERE relforcerowsecurity),
            count(*) FILTER (WHERE relrowsecurity),
            count(*)
       FROM pg_class
      WHERE relnamespace = to_regnamespace('${SCN5_SCHEMA}') AND relkind = 'r';" \
    >> "${SCN5_FORCE_RLS_RECEIPT}"
done
awk -F'|' -v source="${SCN5_SOURCE_DB}" -v target="${SCN5_TARGET_DB}" '
  NR == 1 { valid = ($1 == source && $2 == 0 && $3 == 9 && $4 > 0) }
  NR == 2 { valid = valid && ($1 == target && $2 == 0 && $3 == 9 && $4 > 0) }
  END { exit !(NR == 2 && valid) }
' "${SCN5_FORCE_RLS_RECEIPT}"

Verified by: the two-row receipt names stellaops_platform then stellaops_scanner; each row has FORCE=0, RLS=9, and a positive leaf-table census. Any mismatch is a NO-GO for H8.

Before any database fence, prove that the already-recreated Authority and Policy owner services are from this exact source and that Authority converged the Scanner owner scope. A clean image built at some other commit is not sufficient.

for service_module in stellaops-authority:authority stellaops-policy-engine:policy-engine; do
  service="${service_module%%:*}"
  module="${service_module##*:}"
  docker exec "${service}" cat /app/buildinfo.json \
    > "${SCN5_ROOT}/topology/${module}.running-buildinfo.json"
  jq -e --arg module "${module}" '
    (.gitSha | type == "string" and length == 40)
    and .module == $module
    and .worktreeState == "clean"
    and .dirtyFileCount == 0
  ' "${SCN5_ROOT}/topology/${module}.running-buildinfo.json" >/dev/null
  test "$(docker inspect "${service}" \
    --format '{{.State.Status}} {{.State.Health.Status}}')" = "running healthy"
done

docker exec "${SCN5_PG_CONTAINER}" psql -XqAt -F '|' -v ON_ERROR_STOP=1 \
  -U "${SCN5_PG_OPERATOR}" -d stellaops_authority -c \
  "SELECT
      (SELECT count(*) = 1 FROM authority.schema_migrations
        WHERE migration_name = '027_authority_policy_scanner_preview_scope_convergence.sql'),
      (SELECT count(*) = 1 FROM authority.schema_migrations
        WHERE migration_name = '028_authority_offlinekit_scanner_activation_scope_convergence.sql'),
      (SELECT count(*) = 1 FROM authority.permissions
        WHERE tenant_id = 'default' AND name = 'policy:preview:invoke'
          AND resource = 'policy' AND action = 'scanner-preview'),
      (SELECT count(*) = 1 FROM authority.clients
        WHERE client_id = 'stellaops-scanner-web' AND enabled AND require_client_secret
          AND is_first_party
          AND client_type = 'confidential'
          AND secret_hash IS NOT NULL AND btrim(secret_hash) <> ''
          AND sender_constraint IS NULL
          AND allowed_grant_types = ARRAY['client_credentials']::text[]
          AND allowed_scopes = ARRAY['advisory:read','aoc:verify','policy:preview:invoke']::text[]
          AND ARRAY(
            SELECT token FROM regexp_split_to_table(
              trim(coalesce(properties->>'allowedScopes','')), '[[:space:]]+') token
            WHERE token <> '' ORDER BY token
          ) = ARRAY['advisory:read','aoc:verify','policy:preview:invoke']::text[]
          AND ARRAY(
            SELECT token FROM regexp_split_to_table(
              trim(coalesce(properties->>'audiences','')), '[[:space:]]+') token
            WHERE token <> '' ORDER BY token
          ) = ARRAY['api://policy-engine','stellaops']::text[]
          AND properties->>'tenant' = 'default'
          AND ARRAY(
            SELECT token FROM regexp_split_to_table(
              trim(coalesce(properties->>'tenants','')), '[[:space:]]+') token
            WHERE token <> '' ORDER BY token
          ) = ARRAY['default']::text[]),
      (SELECT ARRAY(
          SELECT tenant_id FROM authority.client_tenants
          WHERE client_id = 'stellaops-scanner-web' ORDER BY tenant_id
        ) = ARRAY['default']::text[]),
      (SELECT count(*) = 1 FROM authority.permissions
        WHERE tenant_id = 'default' AND name = 'scanner:offlinekit:artifact-boms'
          AND resource = 'scanner' AND action = 'offlinekit-artifact-boms'),
      (SELECT count(*) = 1 FROM authority.clients
        WHERE client_id = 'stellaops-offlinekit-activator' AND enabled
          AND require_client_secret AND is_first_party
          AND client_type = 'confidential'
          AND secret_hash IS NOT NULL AND btrim(secret_hash) <> ''
          AND sender_constraint IS NULL
          AND allowed_grant_types = ARRAY['client_credentials']::text[]
          AND allowed_scopes = ARRAY['scanner:offlinekit:artifact-boms']::text[]
          AND ARRAY(
            SELECT token FROM regexp_split_to_table(
              trim(coalesce(properties->>'allowedScopes','')), '[[:space:]]+') token
            WHERE token <> '' ORDER BY token
          ) = ARRAY['scanner:offlinekit:artifact-boms']::text[]
          AND ARRAY(
            SELECT token FROM regexp_split_to_table(
              trim(coalesce(properties->>'audiences','')), '[[:space:]]+') token
            WHERE token <> '' ORDER BY token
          ) = ARRAY['api://scanner','stellaops']::text[]
          AND properties->>'tenant' = 'default'
          AND ARRAY(
            SELECT token FROM regexp_split_to_table(
              trim(coalesce(properties->>'tenants','')), '[[:space:]]+') token
            WHERE token <> '' ORDER BY token
          ) = ARRAY['default']::text[]
          AND NOT EXISTS (
            SELECT 1 FROM unnest(allowed_scopes) scope
            WHERE scope LIKE 'scanner:%'
              AND scope <> 'scanner:offlinekit:artifact-boms'
          )),
      (SELECT ARRAY(
          SELECT tenant_id FROM authority.client_tenants
          WHERE client_id = 'stellaops-offlinekit-activator' ORDER BY tenant_id
        ) = ARRAY['default']::text[]);" \
  > "${SCN5_ROOT}/topology/authority-scanner-owner-boundaries.txt"
grep -Fx 't|t|t|t|t|t|t|t' \
  "${SCN5_ROOT}/topology/authority-scanner-owner-boundaries.txt"

Verified by: both initial buildinfo receipts name their module, a 40-character source SHA, worktreeState=clean, and zero dirty files; both containers are healthy; the Authority receipt is exactly t|t|t|t|t|t|t|t. The exact candidate-SHA check is repeated after the stopped-writer build. That row proves startup companions 027/028, both exact permission records, confidential client type, non-empty secret hash, no undeclared sender constraint, Scanner Web’s exact three scopes/two audiences/default-only tenant boundary, and OfflineKit activator’s activation-only scope/two audiences/default-only tenant boundary. The explicit NOT EXISTS rejects every ordinary scanner:* grant on the activator. Any mismatch is a NO-GO because the later owner calls would otherwise exercise stale or overprivileged runtime configuration.

Confirm the operator connection and that the target is the one this runbook prepared:

docker exec "${SCN5_PG_CONTAINER}" psql -XqAt -U "${SCN5_PG_OPERATOR}" -d "${SCN5_TARGET_DB}" \
  -c "SELECT current_database();" > "${SCN5_ROOT}/topology/target-database.txt"
grep -Fx 'stellaops_scanner' "${SCN5_ROOT}/topology/target-database.txt"
test "$(wc -l < "${SCN5_ROOT}/topology/target-database.txt")" -eq 1

Verified by: the one-line receipt is exactly stellaops_scanner. Any other value means the container or the operator role is not what this procedure assumes; stop.

2. Capture rollback topology and pin the outgoing images

The inspections are read-only. The two docker tag calls add window-specific local aliases to the already-running image IDs; they do not rebuild, publish, recreate, or stop a container.

Before the fence, use apply_patch to create tmp/scn5-<window-id>/topology/rollback-images.override.yml with this exact LF-terminated template:

services:
  authority:
    image: ${SCN5_AUTHORITY_ROLLBACK_IMAGE:?SCN5_AUTHORITY_ROLLBACK_IMAGE is required}
    pull_policy: never
  policy-engine:
    image: ${SCN5_POLICY_ROLLBACK_IMAGE:?SCN5_POLICY_ROLLBACK_IMAGE is required}
    pull_policy: never
  # HISTORICAL (2026-09-14): the generic estate connection these rollback lines substitute was
  # retired by SPRINT_20260722_021 PLT-3 and is undefined on every estate, so this overlay no
  # longer renders. A Scanner rollback now means re-pinning the image on the family's OWN
  # STELLAOPS_POSTGRES_SCANNER_CONNECTION; the pre-move shared database is gone with it.
  scanner-web:
    image: ${SCN5_WEB_ROLLBACK_IMAGE:?SCN5_WEB_ROLLBACK_IMAGE is required}
    pull_policy: never
    environment:
      STELLAOPS_POSTGRES_SCANNER_CONNECTION: "${STELLAOPS_POSTGRES_CONNECTION:?STELLAOPS_POSTGRES_CONNECTION is required only for SCN-5 rollback}"
      SCANNER_SCANNER__STORAGE__DSN: "${STELLAOPS_POSTGRES_CONNECTION:?STELLAOPS_POSTGRES_CONNECTION is required only for SCN-5 rollback}"
    volumes:
      - "${SCN5_WEB_ROLLBACK_BUNDLE_SOURCE:?SCN5_WEB_ROLLBACK_BUNDLE_SOURCE is required}:/app/plugins/scanner/base:ro"
  scanner-worker:
    image: ${SCN5_WORKER_ROLLBACK_IMAGE:?SCN5_WORKER_ROLLBACK_IMAGE is required}
    pull_policy: never
    environment:
      STELLAOPS_POSTGRES_SCANNER_CONNECTION: "${STELLAOPS_POSTGRES_CONNECTION:?STELLAOPS_POSTGRES_CONNECTION is required only for SCN-5 rollback}"
      ScannerStorage__Postgres__ConnectionString: "${STELLAOPS_POSTGRES_CONNECTION:?STELLAOPS_POSTGRES_CONNECTION is required only for SCN-5 rollback}"
    volumes:
      - "${SCN5_WORKER_ROLLBACK_BUNDLE_SOURCE:?SCN5_WORKER_ROLLBACK_BUNDLE_SOURCE is required}:/app/plugins/scanner/base:ro"

SCN5_ENV_DELTA_EXCLUDE='(STELLAOPS_POSTGRES_SCANNER_CONNECTION|SCANNER_SCANNER__STORAGE__DSN|ScannerStorage__Postgres__ConnectionString|SCANNER_SCANNER__CUTOVER__INTAKEWINDOWID)'
for c in stellaops-scanner-web stellaops-scanner-worker; do
  docker inspect "$c" --format '{{index .Config.Labels "com.docker.compose.project.config_files"}}' \
    > "${SCN5_ROOT}/topology/${c}.config_files.txt"
  docker inspect "$c" --format '{{.Config.Image}} {{.Image}}' \
    > "${SCN5_ROOT}/topology/${c}.image.txt"
  scn5_capture_env_names "$c" "${SCN5_ROOT}/topology/${c}.env-names.txt"
  scn5_capture_env_hmac "$c" "${SCN5_ROOT}/topology/${c}.env.hmac"
  scn5_capture_env_hmac "$c" "${SCN5_ROOT}/topology/${c}.unchanged-env.hmac" \
    "${SCN5_ENV_DELTA_EXCLUDE}"
  scn5_capture_runtime_topology \
    "$c" "${SCN5_ROOT}/topology/${c}.runtime-topology.json"
  docker inspect "$c" --format '{{range .NetworkSettings.Networks}}{{println .IPAddress}}{{end}}' \
    | sed '/^[[:space:]]*$/d' | LC_ALL=C sort -u \
    > "${SCN5_ROOT}/topology/${c}.network-ips.txt"
done

for c in stellaops-authority stellaops-policy-engine; do
  docker inspect "$c" --format '{{.Config.Image}} {{.Image}}' \
    > "${SCN5_ROOT}/topology/${c}.image.txt"
  scn5_capture_env_names "$c" "${SCN5_ROOT}/topology/${c}.env-names.txt"
  scn5_capture_env_hmac "$c" "${SCN5_ROOT}/topology/${c}.env.hmac"
  scn5_capture_runtime_topology "$c" "${SCN5_ROOT}/topology/${c}.runtime-topology.json"
  docker exec "$c" cat /app/buildinfo.json \
    > "${SCN5_ROOT}/topology/${c}.buildinfo.json"
done

SCN5_SCANNER_IPS="$(LC_ALL=C sort -u "${SCN5_ROOT}"/topology/*.network-ips.txt | paste -sd, -)"
test -n "${SCN5_SCANNER_IPS}"

SCN5_WEB_ROLLBACK_IMAGE="stellaops/scanner-web:scn5-rollback-${SCN5_WINDOW_ID}"
SCN5_WORKER_ROLLBACK_IMAGE="stellaops/scanner-worker:scn5-rollback-${SCN5_WINDOW_ID}"
SCN5_AUTHORITY_ROLLBACK_IMAGE="stellaops/authority:scn5-rollback-${SCN5_WINDOW_ID}"
SCN5_POLICY_ROLLBACK_IMAGE="stellaops/policy-engine:scn5-rollback-${SCN5_WINDOW_ID}"
SCN5_WEB_OUTGOING_IMAGE_ID="$(awk '{print $2}' "${SCN5_ROOT}/topology/stellaops-scanner-web.image.txt")"
SCN5_WORKER_OUTGOING_IMAGE_ID="$(awk '{print $2}' "${SCN5_ROOT}/topology/stellaops-scanner-worker.image.txt")"
SCN5_AUTHORITY_OUTGOING_IMAGE_ID="$(awk '{print $2}' "${SCN5_ROOT}/topology/stellaops-authority.image.txt")"
SCN5_POLICY_OUTGOING_IMAGE_ID="$(awk '{print $2}' "${SCN5_ROOT}/topology/stellaops-policy-engine.image.txt")"
for outgoing_image_id in \
    "${SCN5_WEB_OUTGOING_IMAGE_ID}" "${SCN5_WORKER_OUTGOING_IMAGE_ID}" \
    "${SCN5_AUTHORITY_OUTGOING_IMAGE_ID}" "${SCN5_POLICY_OUTGOING_IMAGE_ID}"; do
  [[ "${outgoing_image_id}" =~ ^sha256:[0-9a-f]{64}$ ]]
done
docker tag "$(docker inspect stellaops-scanner-web --format '{{.Image}}')" "${SCN5_WEB_ROLLBACK_IMAGE}"
docker tag "$(docker inspect stellaops-scanner-worker --format '{{.Image}}')" "${SCN5_WORKER_ROLLBACK_IMAGE}"
docker tag "$(docker inspect stellaops-authority --format '{{.Image}}')" "${SCN5_AUTHORITY_ROLLBACK_IMAGE}"
docker tag "$(docker inspect stellaops-policy-engine --format '{{.Image}}')" "${SCN5_POLICY_ROLLBACK_IMAGE}"
test "$(docker image inspect "${SCN5_WEB_ROLLBACK_IMAGE}" --format '{{.Id}}')" = \
  "${SCN5_WEB_OUTGOING_IMAGE_ID}"
test "$(docker image inspect "${SCN5_WORKER_ROLLBACK_IMAGE}" --format '{{.Id}}')" = \
  "${SCN5_WORKER_OUTGOING_IMAGE_ID}"
test "$(docker image inspect "${SCN5_AUTHORITY_ROLLBACK_IMAGE}" --format '{{.Id}}')" = \
  "${SCN5_AUTHORITY_OUTGOING_IMAGE_ID}"
test "$(docker image inspect "${SCN5_POLICY_ROLLBACK_IMAGE}" --format '{{.Id}}')" = \
  "${SCN5_POLICY_OUTGOING_IMAGE_ID}"

scn5_bundle_mount_source() { # $1=container topology receipt
  jq -er '[.mounts[] | select(.destination == "/app/plugins/scanner/base")]
    | if length == 1 and .[0].type == "bind" and .[0].rw == false
      then .[0].source else error("expected one read-only Scanner base bind mount") end' "$1"
}
SCN5_WEB_OUTGOING_BUNDLE_SOURCE="$(scn5_bundle_mount_source \
  "${SCN5_ROOT}/topology/stellaops-scanner-web.runtime-topology.json")"
SCN5_WORKER_OUTGOING_BUNDLE_SOURCE="$(scn5_bundle_mount_source \
  "${SCN5_ROOT}/topology/stellaops-scanner-worker.runtime-topology.json")"
[[ "${SCN5_WEB_OUTGOING_BUNDLE_SOURCE//\\//}" == */devops/plugins/scanner/base ]]
[[ "${SCN5_WORKER_OUTGOING_BUNDLE_SOURCE//\\//}" == \
  */tmp/build/int6/scanner-worker-plugin-bundle/base ]]
test "${SCN5_WEB_OUTGOING_BUNDLE_SOURCE}" != "${SCN5_WORKER_OUTGOING_BUNDLE_SOURCE}"

SCN5_WEB_ROLLBACK_BUNDLE_SOURCE="${SCN5_REPO_ROOT}/${SCN5_ROOT}/backup/outgoing-scanner-web-base"
SCN5_WORKER_ROLLBACK_BUNDLE_SOURCE="${SCN5_REPO_ROOT}/${SCN5_ROOT}/backup/outgoing-scanner-worker-base"
SCN5_ROLLBACK_OVERLAY_FILE="${SCN5_ROOT}/topology/rollback-images.override.yml"
SCN5_ROLLBACK_OVERLAY="${SCN5_COMPOSE_ROOT_REL}/topology/rollback-images.override.yml"
test "$(sha256sum "${SCN5_ROLLBACK_OVERLAY_FILE}" | awk '{print $1}')" = \
  '07b3a9f436eab635c0e6253c85b8e5c7251005f99b80abf776540313885abd1d'
export SCN5_AUTHORITY_ROLLBACK_IMAGE SCN5_POLICY_ROLLBACK_IMAGE
export SCN5_WEB_ROLLBACK_IMAGE SCN5_WORKER_ROLLBACK_IMAGE
export SCN5_WEB_ROLLBACK_BUNDLE_SOURCE SCN5_WORKER_ROLLBACK_BUNDLE_SOURCE
[[ "${SCN5_ROLLBACK_OVERLAY}" != *[[:space:]]* ]]
test -f "devops/compose/${SCN5_ROLLBACK_OVERLAY}"

# Prove rollback is renderable while both old writers are still running. A
# rollback recipe first exercised after the fence cannot protect the window.
SCN5_ROLLBACK_FILES="${SCN5_ROLLBACK_OVERLAY}"
COMPOSE_EXTRA_FILES="${SCN5_ROLLBACK_FILES}" \
  scn5_render_topology "${SCN5_ROOT}/topology/rollback-rendered-names-only.json"
jq -e --arg web "${SCN5_WEB_ROLLBACK_IMAGE}" \
      --arg worker "${SCN5_WORKER_ROLLBACK_IMAGE}" \
      --arg webSource "${SCN5_WEB_ROLLBACK_BUNDLE_SOURCE}" \
      --arg workerSource "${SCN5_WORKER_ROLLBACK_BUNDLE_SOURCE}" '
  .services["scanner-web"].image == $web
  and .services["scanner-worker"].image == $worker
  and .services["scanner-web"].scannerBaseSource == $webSource
  and .services["scanner-worker"].scannerBaseSource == $workerSource
  and ([.services["scanner-web"].environmentKeys[]
        | select(test("STELLAOPS_POSTGRES_SCANNER_CONNECTION|SCANNER_SCANNER__STORAGE__DSN|ScannerStorage__Postgres__ConnectionString"))]
       == ["SCANNER_SCANNER__STORAGE__DSN", "STELLAOPS_POSTGRES_SCANNER_CONNECTION"])
  and ([.services["scanner-worker"].environmentKeys[]
        | select(test("STELLAOPS_POSTGRES_SCANNER_CONNECTION|SCANNER_SCANNER__STORAGE__DSN|ScannerStorage__Postgres__ConnectionString"))]
       == ["STELLAOPS_POSTGRES_SCANNER_CONNECTION", "ScannerStorage__Postgres__ConnectionString"])' \
  "${SCN5_ROOT}/topology/rollback-rendered-names-only.json" >/dev/null
scn5_assert_rollback_connection_contract

Verified by: each config_files.txt is non-empty; each names-only env-names.txt has the expected key count; full environment values exist only in process memory and the receipts contain keyed HMACs; the sanitized runtime-topology receipts omit environment values, dynamic container/network IDs, IPs, and Compose labels but cover the stable process, health, privilege, resource, mount, port, and network contract; both rollback tags resolve to the exact .Image ID captured in the adjacent receipt; and a pre-fence canonical-helper render with only ${SCN5_ROLLBACK_OVERLAY} plus the existing-networks override last sets both the canonical and host-specific legacy connection keys to the same non-empty source connection, and changes no other environment key. The reviewed Compose diff and in-memory equality assertion are the value-change authority; no rendered value is written to a receipt. The historical config_files.txt rows are drift evidence, not the rollback command.

Record which connection key each host resolves today, so the post-repoint check has a before:

grep -lE 'SCANNER_SCANNER__STORAGE__DSN|ScannerStorage__Postgres__ConnectionString' \
  "${SCN5_ROOT}"/topology/*.env-names.txt

Verified by: scanner-web matches on SCANNER_SCANNER__STORAGE__DSN, scanner-worker on ScannerStorage__Postgres__ConnectionString, and neither matches STELLAOPS_POSTGRES_SCANNER_CONNECTION (H5). If either already carries the canonical variable, the estate has moved since prep — re-take the A2 comparison before continuing.

3. Assert the migration-001 target and create the copy partitions

Do not create the target schema by copying the source and do not start a current Scanner candidate against it. The prepared target must still be at the old copy-compatible shape.

docker exec "${SCN5_PG_CONTAINER}" psql -XqAt -U "${SCN5_PG_OPERATOR}" -d "${SCN5_TARGET_DB}" -c \
  "SELECT migration_name || ' [' || category || ']' FROM ${SCN5_SCHEMA}.schema_migrations ORDER BY 1;" \
  > "${SCN5_ROOT}/shape/target-ledger.before.txt"
docker exec "${SCN5_PG_CONTAINER}" psql -XqAt -U "${SCN5_PG_OPERATOR}" -d "${SCN5_TARGET_DB}" -c \
  "SELECT nspname FROM pg_namespace
    WHERE nspname NOT LIKE 'pg_%' AND nspname <> 'information_schema' ORDER BY 1;" \
  > "${SCN5_ROOT}/shape/target-schemas.before.txt"
docker exec "${SCN5_PG_CONTAINER}" psql -XqAt -U "${SCN5_PG_OPERATOR}" -d "${SCN5_TARGET_DB}" -c \
  "WITH owners AS (
     SELECT pg_get_userbyid(n.nspowner) AS owner
       FROM pg_namespace n WHERE n.nspname = '${SCN5_SCHEMA}'
     UNION ALL
     SELECT pg_get_userbyid(c.relowner)
       FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace
      WHERE n.nspname = '${SCN5_SCHEMA}'
   ) SELECT DISTINCT owner FROM owners ORDER BY owner;" \
  > "${SCN5_ROOT}/shape/target-owners.before.txt"

grep -Fx '001_scanner_consolidated_baseline.sql [startup]' \
  "${SCN5_ROOT}/shape/target-ledger.before.txt"
test "$(wc -l < "${SCN5_ROOT}/shape/target-ledger.before.txt")" -eq 1
test "$(cat "${SCN5_ROOT}/shape/target-schemas.before.txt")" = $'public\nscanner'
grep -Fx 'scanner' "${SCN5_ROOT}/shape/target-owners.before.txt"
test "$(wc -l < "${SCN5_ROOT}/shape/target-owners.before.txt")" -eq 1

docker exec "${SCN5_PG_CONTAINER}" psql -XqAt -v ON_ERROR_STOP=1 \
  -U "${SCN5_PG_OPERATOR}" -d "${SCN5_SOURCE_DB}" -c \
  "SELECT c.relname
     FROM pg_class c
     JOIN pg_namespace n ON n.oid = c.relnamespace
    WHERE n.nspname = '${SCN5_SCHEMA}' AND c.relkind = 'r'
      AND c.relname <> 'schema_migrations'
    ORDER BY c.relname;" > "${SCN5_ROOT}/shape/source-copy-leaf-tables.txt"
test "$(wc -l < "${SCN5_ROOT}/shape/source-copy-leaf-tables.txt")" -eq 84

SCN5_COPY_LEAF_COUNT_SQL="$(docker exec "${SCN5_PG_CONTAINER}" \
  psql -XqAt -U "${SCN5_PG_OPERATOR}" -d "${SCN5_SOURCE_DB}" -c \
  "SELECT string_agg(
      format('SELECT %L::text AS table_name, count(*)::bigint AS row_count FROM ${SCN5_SCHEMA}.%I',
             c.relname, c.relname),
      ' UNION ALL ' ORDER BY c.relname)
     FROM pg_class c
     JOIN pg_namespace n ON n.oid = c.relnamespace
    WHERE n.nspname = '${SCN5_SCHEMA}' AND c.relkind = 'r'
      AND c.relname <> 'schema_migrations';")"
test -n "${SCN5_COPY_LEAF_COUNT_SQL}"
docker exec "${SCN5_PG_CONTAINER}" psql -XqAt -F $'\t' \
  -v ON_ERROR_STOP=1 -U "${SCN5_PG_OPERATOR}" -d "${SCN5_TARGET_DB}" \
  -c "${SCN5_COPY_LEAF_COUNT_SQL} ORDER BY 1" \
  > "${SCN5_ROOT}/shape/target-copy-leaf-counts.before.tsv"
test "$(wc -l < "${SCN5_ROOT}/shape/target-copy-leaf-counts.before.tsv")" -eq 84
awk -F'\t' '$2 != 0 { print $1 "\t" $2 }' \
  "${SCN5_ROOT}/shape/target-copy-leaf-counts.before.tsv" \
  > "${SCN5_ROOT}/shape/target-copy-leaf-nonempty.before.tsv"
test ! -s "${SCN5_ROOT}/shape/target-copy-leaf-nonempty.before.tsv"

Verified by: the executable receipts assert exactly one ledger row 001_scanner_consolidated_baseline.sql [startup]; schemas public then scanner only (this is where the signals orphan is proven absent, live rather than by grep); a single owner, scanner; exactly 84 source copy leaves; exactly 84 corresponding target counts; and zero rows in every target leaf table in that copy set. The historical epss_config row created by 001 and the target migration-ledger row are outside that copy set.

If any assertion fails, stop before fencing the writers. Do not “repair” the target by booting the candidate: that applies 002/003 before the copy. A retry target is recreated and brought to exactly 001 with the reviewed migration-001 preparation artefact, then this step is repeated.

Record the SERVICE ROLE identity that the later migration-only command must use (H8). Run this through a real login as the scanner role; SET ROLE is insufficient because it cannot prove session_user=scanner:

set +x # secrets below must remain silent even if the caller enabled xtrace
: "${SCANNER_POSTGRES_PASSWORD:?Inject the Scanner role password by name; do not print it}"
scn5_psql_as_scanner() { # $1 = database, $2 = SQL; password travels only on stdin
  local database="$1" sql="$2"
  local -a pipeline_rc
  set +e
  printf '%s\n' "${SCANNER_POSTGRES_PASSWORD}" \
    | docker exec -i "${SCN5_PG_CONTAINER}" sh -eu -c '
        IFS= read -r PGPASSWORD
        export PGPASSWORD
        exec psql -XqAt -v ON_ERROR_STOP=1 -h 127.0.0.1 \
          -U "$1" -d "$2" -c "$3"
      ' scn5-psql "${SCN5_SCANNER_ROLE}" "${database}" "${sql}"
  pipeline_rc=("${PIPESTATUS[@]}")
  set -e
  test "${#pipeline_rc[@]}" -eq 2
  test "${pipeline_rc[0]}" -eq 0
  test "${pipeline_rc[1]}" -eq 0
}

SCN5_SCANNER_IDENTITY_SQL="SELECT current_database(), session_user, current_user,
  (SELECT rolsuper FROM pg_roles WHERE rolname = session_user),
  (SELECT rolcreatedb FROM pg_roles WHERE rolname = session_user),
  (SELECT rolcreaterole FROM pg_roles WHERE rolname = session_user),
  (SELECT rolbypassrls FROM pg_roles WHERE rolname = session_user),
  current_setting('row_security');"
scn5_psql_as_scanner "${SCN5_TARGET_DB}" "${SCN5_SCANNER_IDENTITY_SQL}" \
  > "${SCN5_ROOT}/shape/scanner-role-identity.txt"
grep -Fx 'stellaops_scanner|scanner|scanner|f|f|f|f|on' \
  "${SCN5_ROOT}/shape/scanner-role-identity.txt"

Verified by: current_database = stellaops_scanner, session_user = current_user = scanner, rolsuper = rolcreatedb = rolcreaterole = rolbypassrls = false, and row_security = on. Anything else is a NO-GO. The supported command repeats these checks, including row_security, in-process and returns exit code 3 on a mismatch. Run the H8 FORCE-RLS survey against both source and target in the same sitting and record both results.

Every object the window creates by hand in the target follows the same rule. The partition below is created over a real scanner login, and its receipt is session_user = scanner, rolsuper = f.

Then close H1 — create every source partition the fresh baseline did not. Compute the set, do not copy the month from this page:

docker exec "${SCN5_PG_CONTAINER}" psql -XqAt -U "${SCN5_PG_OPERATOR}" -d "${SCN5_SOURCE_DB}" -c \
  "SELECT c.relname FROM pg_class p
     JOIN pg_namespace n ON n.oid = p.relnamespace
     JOIN pg_inherits i ON i.inhparent = p.oid
     JOIN pg_class c ON c.oid = i.inhrelid
    WHERE n.nspname = '${SCN5_SCHEMA}' AND p.relname = 'artifact_boms'
      AND p.relkind = 'p' AND c.relkind = 'r'
    ORDER BY 1;" > "${SCN5_ROOT}/shape/source-partitions.txt"

docker exec "${SCN5_PG_CONTAINER}" psql -XqAt -U "${SCN5_PG_OPERATOR}" -d "${SCN5_TARGET_DB}" -c \
  "SELECT c.relname FROM pg_class p
     JOIN pg_namespace n ON n.oid = p.relnamespace
     JOIN pg_inherits i ON i.inhparent = p.oid
     JOIN pg_class c ON c.oid = i.inhrelid
    WHERE n.nspname = '${SCN5_SCHEMA}' AND p.relname = 'artifact_boms'
      AND p.relkind = 'p' AND c.relkind = 'r'
    ORDER BY 1;" > "${SCN5_ROOT}/shape/target-partitions.before.txt"

comm -23 "${SCN5_ROOT}/shape/source-partitions.txt" \
  "${SCN5_ROOT}/shape/target-partitions.before.txt" \
  > "${SCN5_ROOT}/shape/missing-target-partitions.txt"
sed -n '1,120p' "${SCN5_ROOT}/shape/missing-target-partitions.txt"

For each missing artifact_boms_<YYYY>_<MM>, create it through the service’s own function, on the scanner connection so the objects are owned by scanner:

SCN5_PARTITION_CREATE_RECEIPT="${SCN5_ROOT}/shape/partition-create.txt"
: > "${SCN5_PARTITION_CREATE_RECEIPT}"
while IFS= read -r partition_name; do
  [[ "${partition_name}" =~ ^artifact_boms_([0-9]{4})_([0-9]{2})$ ]]
  partition_year="${BASH_REMATCH[1]}"
  partition_month=$((10#${BASH_REMATCH[2]}))
  test "${partition_month}" -ge 1
  test "${partition_month}" -le 12
  printf '%s|' "${partition_name}" >> "${SCN5_PARTITION_CREATE_RECEIPT}"
  scn5_psql_as_scanner "${SCN5_TARGET_DB}" \
    "SELECT ${SCN5_SCHEMA}.create_artifact_boms_partition(${partition_year}, ${partition_month});" \
    >> "${SCN5_PARTITION_CREATE_RECEIPT}"
done < "${SCN5_ROOT}/shape/missing-target-partitions.txt"

docker exec "${SCN5_PG_CONTAINER}" psql -XqAt -F $'\t' \
  -U "${SCN5_PG_OPERATOR}" -d "${SCN5_SOURCE_DB}" -c \
  "SELECT p.relname, c.relname, pg_get_expr(c.relpartbound, c.oid)
     FROM pg_class p
     JOIN pg_namespace n ON n.oid = p.relnamespace
     JOIN pg_inherits i ON i.inhparent = p.oid
     JOIN pg_class c ON c.oid = i.inhrelid
    WHERE n.nspname = '${SCN5_SCHEMA}' AND p.relname = 'artifact_boms'
      AND p.relkind = 'p' AND c.relkind = 'r'
    ORDER BY 1, 2;" > "${SCN5_ROOT}/shape/source-partition-bounds.tsv"

docker exec "${SCN5_PG_CONTAINER}" psql -XqAt -F $'\t' \
  -U "${SCN5_PG_OPERATOR}" -d "${SCN5_TARGET_DB}" -c \
  "SELECT p.relname, c.relname, pg_get_expr(c.relpartbound, c.oid)
     FROM pg_class p
     JOIN pg_namespace n ON n.oid = p.relnamespace
     JOIN pg_inherits i ON i.inhparent = p.oid
     JOIN pg_class c ON c.oid = i.inhrelid
    WHERE n.nspname = '${SCN5_SCHEMA}' AND p.relname = 'artifact_boms'
      AND p.relkind = 'p' AND c.relkind = 'r'
    ORDER BY 1, 2;" > "${SCN5_ROOT}/shape/target-partition-bounds.after.tsv"

docker exec "${SCN5_PG_CONTAINER}" psql -XqAt -U "${SCN5_PG_OPERATOR}" -d "${SCN5_TARGET_DB}" -c \
  "SELECT c.relname FROM pg_class p
     JOIN pg_namespace n ON n.oid = p.relnamespace
     JOIN pg_inherits i ON i.inhparent = p.oid
     JOIN pg_class c ON c.oid = i.inhrelid
    WHERE n.nspname = '${SCN5_SCHEMA}' AND p.relname = 'artifact_boms'
      AND p.relkind = 'p' AND c.relkind = 'r'
    ORDER BY 1;" > "${SCN5_ROOT}/shape/target-partitions.after.txt"

cmp "${SCN5_ROOT}/shape/source-partitions.txt" \
    "${SCN5_ROOT}/shape/target-partitions.after.txt"
cmp "${SCN5_ROOT}/shape/source-partition-bounds.tsv" \
    "${SCN5_ROOT}/shape/target-partition-bounds.after.tsv"

docker exec "${SCN5_PG_CONTAINER}" psql -XqAt -F $'\t' \
  -U "${SCN5_PG_OPERATOR}" -d "${SCN5_TARGET_DB}" -c \
  "SELECT c.relkind, c.relname, pg_get_userbyid(c.relowner)
     FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace
    WHERE n.nspname = '${SCN5_SCHEMA}'
      AND pg_get_userbyid(c.relowner) <> '${SCN5_SCANNER_ROLE}'
    ORDER BY 1, 2;" > "${SCN5_ROOT}/shape/target-nonscanner-owners.tsv"
test ! -s "${SCN5_ROOT}/shape/target-nonscanner-owners.tsv"

Verified by, in this order:

1. every persisted missing name passed the exact artifact_boms_<YYYY>_<MM> membership test and the loop created every row in that file rather than a month copied from this page;
2. the regenerated source/target artifact_boms child-name inventories compare byte-identically;
3. every artifact_boms child bound (pg_get_expr(c.relpartbound, c.oid)) compares byte-identically; target-only EPSS partitions are intentionally outside the copy contract;
4. target-nonscanner-owners.tsv is empty. A partition created as the superuser leaves the table plus its indexes owned by stellaops; that was measured in prep and is the reason for the direct scanner login.

Then re-take the full shape comparison (the ten projections of the A2 table) source vs target:

scn5_assert_shape_contains_source pre-copy

Verified by: pre-copy-summary.txt has ten projection rows and missing=0 on every row; every pre-copy-<projection>.non-epss-extra.tsv is empty; and pre-copy-rls.diff.tsv is empty. The individual missing, extra, and non-EPSS-extra receipts name the exact object if the gate fails. A non-zero MISSING, any target-only object whose owning name is not EPSS, or an RLS diff is the window’s blocking finding: stop, do not freeze anything, and report the projection and row.

4. Fence both writers

Both writers stop before anything is read for the copy. There is no dual-write and no shadow path — pre-release, this is a hard cutover (AGENTS.md §2.11).

docker stop stellaops-scanner-web stellaops-scanner-worker \
  > "${SCN5_ROOT}/logs/fence.txt"
sed -n '1,40p' "${SCN5_ROOT}/logs/fence.txt"

for c in stellaops-scanner-web stellaops-scanner-worker; do
  test "$(docker inspect -f '{{.State.Running}}' "$c")" = false
done

scn5_assert_writers_stopped() {
  local container
  for container in stellaops-scanner-web stellaops-scanner-worker; do
    test "$(docker inspect -f '{{.State.Running}}' "${container}")" = false
  done
}
scn5_assert_writers_stopped

SCN5_BUNDLE_EXPECTED_BASE="${SCN5_ROOT}/topology/scanner-base.expected.txt"
SCN5_BUNDLE_EXPECTED_FULL="${SCN5_ROOT}/topology/scanner-full.expected.txt"
printf '%s\n' \
  stellaops.scanner.plugin.dotnet stellaops.scanner.plugin.golang \
  stellaops.scanner.plugin.java stellaops.scanner.plugin.node \
  stellaops.scanner.plugin.os.apk stellaops.scanner.plugin.os.dpkg \
  stellaops.scanner.plugin.os.rpm stellaops.scanner.plugin.python \
  | tr ' ' '\n' | LC_ALL=C sort > "${SCN5_BUNDLE_EXPECTED_BASE}"
printf '%s\n' \
  stellaops.scanner.plugin.bun stellaops.scanner.plugin.ccpp \
  stellaops.scanner.plugin.dart stellaops.scanner.plugin.deno \
  stellaops.scanner.plugin.dotnet stellaops.scanner.plugin.elixir \
  stellaops.scanner.plugin.golang stellaops.scanner.plugin.java \
  stellaops.scanner.plugin.node stellaops.scanner.plugin.php \
  stellaops.scanner.plugin.python stellaops.scanner.plugin.ruby \
  stellaops.scanner.plugin.rust stellaops.scanner.plugin.swift \
  | tr ' ' '\n' | LC_ALL=C sort > "${SCN5_BUNDLE_EXPECTED_FULL}"

scn5_verify_bundle_root() { # $1=root, $2=expected directory list, $3=receipt prefix
  local root="$1" expected="$2" receipt="$3" bundle
  test -d "${root}"
  test -z "$(find "${root}" -mindepth 1 -maxdepth 1 ! -type d -print -quit)"
  find "${root}" -mindepth 1 -maxdepth 1 -type d -printf '%f\n' \
    | LC_ALL=C sort > "${receipt}.directories.txt"
  cmp "${expected}" "${receipt}.directories.txt"
  : > "${receipt}.checksums.txt"
  while IFS= read -r bundle; do
    test -f "${root}/${bundle}/manifest.json"
    test -f "${root}/${bundle}/checksums.sha256"
    (cd -- "${root}/${bundle}" && sed 's/\r$//' checksums.sha256 \
      | sha256sum --check --strict -) >> "${receipt}.checksums.txt"
  done < "${expected}"
}

test ! -e "${SCN5_WEB_ROLLBACK_BUNDLE_SOURCE}"
test ! -e "${SCN5_WORKER_ROLLBACK_BUNDLE_SOURCE}"
mkdir -p "${SCN5_WEB_ROLLBACK_BUNDLE_SOURCE}" "${SCN5_WORKER_ROLLBACK_BUNDLE_SOURCE}"
docker cp stellaops-scanner-web:/app/plugins/scanner/base/. \
  "${SCN5_WEB_ROLLBACK_BUNDLE_SOURCE}/"
docker cp stellaops-scanner-worker:/app/plugins/scanner/base/. \
  "${SCN5_WORKER_ROLLBACK_BUNDLE_SOURCE}/"
scn5_verify_bundle_root "${SCN5_WEB_ROLLBACK_BUNDLE_SOURCE}" "${SCN5_BUNDLE_EXPECTED_BASE}" \
  "${SCN5_ROOT}/topology/outgoing-web-base"
scn5_verify_bundle_root "${SCN5_WORKER_ROLLBACK_BUNDLE_SOURCE}" "${SCN5_BUNDLE_EXPECTED_BASE}" \
  "${SCN5_ROOT}/topology/outgoing-worker-base"

scn5_capture_full_tree_manifest() { # $1=root, $2=external receipt
  local root="$1" receipt="$2"
  (cd -- "${root}" && find . -type f -print0 | LC_ALL=C sort -z \
    | xargs -0 sha256sum) > "${receipt}"
}
SCN5_WEB_SNAPSHOT_MANIFEST="${SCN5_ROOT}/topology/outgoing-scanner-web-base.full.sha256"
SCN5_WORKER_SNAPSHOT_MANIFEST="${SCN5_ROOT}/topology/outgoing-scanner-worker-base.full.sha256"
scn5_capture_full_tree_manifest "${SCN5_WEB_ROLLBACK_BUNDLE_SOURCE}" \
  "${SCN5_WEB_SNAPSHOT_MANIFEST}"
scn5_capture_full_tree_manifest "${SCN5_WORKER_ROLLBACK_BUNDLE_SOURCE}" \
  "${SCN5_WORKER_SNAPSHOT_MANIFEST}"
(cd -- "${SCN5_ROOT}/topology" && sha256sum \
  "${SCN5_WEB_SNAPSHOT_MANIFEST##*/}" "${SCN5_WORKER_SNAPSHOT_MANIFEST##*/}" \
  > rollback-snapshot-manifests.sha256)

scn5_verify_snapshot_contract_matrix() { # $1=snapshot, $2=label, $3=host image or empty
  local snapshot="$1" label="$2" image="$3"
  local bundle kind contract expected key stamp copy_sha stamp_count required_contract
  local -a bundles required known copies pipeline_rc
  local -A host_shas=()

  known=(
    StellaOps.Scanner.Analyzers.Lang.dll
    StellaOps.Scanner.Analyzers.LangAdapter.dll
    StellaOps.Scanner.Analyzers.OS.dll
    StellaOps.Scanner.Plugin.Contracts.dll
  )
  mapfile -t bundles < <(find "${snapshot}" -mindepth 1 -maxdepth 1 -type d -print \
    | LC_ALL=C sort)
  test "${#bundles[@]}" -gt 0
  : > "${SCN5_ROOT}/topology/${label}-snapshot-contract-matrix.tsv"

  for bundle in "${bundles[@]}"; do
    test -f "${bundle}/manifest.json"
    kind="$(jq -er '.metadata["org.stellaops.analyzer.kind"]' \
      "${bundle}/manifest.json")"
    case "${kind}" in
      language)
        required=(
          StellaOps.Scanner.Analyzers.Lang.dll
          StellaOps.Scanner.Analyzers.LangAdapter.dll
          StellaOps.Scanner.Plugin.Contracts.dll
        )
        ;;
      os) required=(StellaOps.Scanner.Analyzers.OS.dll) ;;
      *) return 1 ;;
    esac

    stamp_count="$(jq -er '[.metadata | keys[] | select(startswith(
      "org.stellaops.contract-assembly."))] | length' "${bundle}/manifest.json")"
    test "${stamp_count}" -eq "${#required[@]}"

    for contract in "${known[@]}"; do
      key="org.stellaops.contract-assembly.${contract%.dll}"
      required_contract=false
      for expected in "${required[@]}"; do
        if test "${contract}" = "${expected}"; then
          required_contract=true
          break
        fi
      done
      mapfile -t copies < <(find "${bundle}" -type f -name "${contract}" -print \
        | LC_ALL=C sort)

      if test "${required_contract}" = false; then
        test "${#copies[@]}" -eq 0
        if jq -e --arg key "${key}" '.metadata | has($key)' \
            "${bundle}/manifest.json" >/dev/null; then
          return 1
        fi
        continue
      fi

      test "${#copies[@]}" -eq 1
      test "${copies[0]}" = "${bundle}/${contract}"
      copy_sha="$(sha256sum "${copies[0]}" | awk '{print $1}')"
      [[ "${copy_sha}" =~ ^[0-9a-f]{64}$ ]]
      stamp="$(jq -er --arg key "${key}" '.metadata[$key]' \
        "${bundle}/manifest.json")"
      [[ "${stamp}" =~ ^version=[^\;]+\;sha256=([0-9a-f]{64})$ ]]
      test "${BASH_REMATCH[1]}" = "${copy_sha}"

      if test -n "${image}"; then
        if test -z "${host_shas[${contract}]+x}"; then
          set +e
          MSYS_NO_PATHCONV=1 docker run --rm --network none --read-only \
            --entrypoint /bin/cat "${image}" "/app/${contract}" \
            | sha256sum > \
              "${SCN5_ROOT}/topology/${label}-image-${contract}.sha256"
          pipeline_rc=("${PIPESTATUS[@]}")
          set -e
          test "${#pipeline_rc[@]}" -eq 2
          test "${pipeline_rc[0]}" -eq 0
          test "${pipeline_rc[1]}" -eq 0
          host_shas[${contract}]="$(awk '{print $1}' \
            "${SCN5_ROOT}/topology/${label}-image-${contract}.sha256")"
          [[ "${host_shas[${contract}]}" =~ ^[0-9a-f]{64}$ ]]
        fi
        test "${copy_sha}" = "${host_shas[${contract}]}"
      fi

      printf '%s\t%s\t%s\t%s\n' "${bundle##*/}" "${kind}" \
        "${contract}" "${copy_sha}" \
        >> "${SCN5_ROOT}/topology/${label}-snapshot-contract-matrix.tsv"
    done
  done
}

scn5_assert_rollback_bundle_snapshots() {
  scn5_assert_writers_stopped
  scn5_verify_bundle_root "${SCN5_WEB_ROLLBACK_BUNDLE_SOURCE}" \
    "${SCN5_BUNDLE_EXPECTED_BASE}" "${SCN5_ROOT}/topology/rollback-gate-web-base"
  scn5_verify_bundle_root "${SCN5_WORKER_ROLLBACK_BUNDLE_SOURCE}" \
    "${SCN5_BUNDLE_EXPECTED_BASE}" "${SCN5_ROOT}/topology/rollback-gate-worker-base"
  (cd -- "${SCN5_ROOT}/topology" && sha256sum --check --strict \
    rollback-snapshot-manifests.sha256)
  scn5_capture_full_tree_manifest "${SCN5_WEB_ROLLBACK_BUNDLE_SOURCE}" \
    "${SCN5_ROOT}/topology/rollback-current-web-base.full.sha256"
  scn5_capture_full_tree_manifest "${SCN5_WORKER_ROLLBACK_BUNDLE_SOURCE}" \
    "${SCN5_ROOT}/topology/rollback-current-worker-base.full.sha256"
  cmp "${SCN5_WEB_SNAPSHOT_MANIFEST}" \
    "${SCN5_ROOT}/topology/rollback-current-web-base.full.sha256"
  cmp "${SCN5_WORKER_SNAPSHOT_MANIFEST}" \
    "${SCN5_ROOT}/topology/rollback-current-worker-base.full.sha256"
  # Web verifies immutable signed-bundle closure but does not activate analyzer
  # assemblies or enable host-bound contract-stamp admission.
  scn5_verify_snapshot_contract_matrix \
    "${SCN5_WEB_ROLLBACK_BUNDLE_SOURCE}" web-rollback ""
  # Worker activates each bundle, so every stamped contract stays bound to the
  # immutable outgoing Worker image.
  scn5_verify_snapshot_contract_matrix \
    "${SCN5_WORKER_ROLLBACK_BUNDLE_SOURCE}" worker-rollback \
    "${SCN5_WORKER_ROLLBACK_IMAGE}"
}
scn5_assert_rollback_bundle_snapshots

# No pre-window job may wake after the target worker starts. Check both the
# stream and its consumer-group delivery state while the writers are stopped.
SCN5_QUEUE_LENGTH="$(docker exec stellaops-valkey valkey-cli XLEN scanner:jobs)"
docker exec stellaops-valkey valkey-cli --json XINFO GROUPS scanner:jobs \
  > "${SCN5_ROOT}/parity/scanner-jobs-groups.json"
test "${SCN5_QUEUE_LENGTH}" -eq 0
jq -e '
  length == 1
  and .[0].name == "scanner-workers"
  and .[0].pending == 0
  and .[0].lag == 0
' "${SCN5_ROOT}/parity/scanner-jobs-groups.json" >/dev/null

Then prove every source-database connection authenticated as the Scanner service role is gone. Match the role, not application_name or last-query text: Scanner has both named and unnamed pools, and an idle writer’s last query may mention eventing or another schema rather than scanner. The role-wide check is intentionally stronger than the pre-stop container-IP inventory and remains valid after Docker has released a stopped container’s dynamic IP.

docker exec -i "${SCN5_PG_CONTAINER}" psql -XqAt -U "${SCN5_PG_OPERATOR}" \
  -d "${SCN5_SOURCE_DB}" -v scanner_role="${SCN5_SCANNER_ROLE}" \
  > "${SCN5_ROOT}/logs/source-activity.txt" <<'SQL'
SELECT pid, usename, application_name, client_addr, state, left(query, 120)
  FROM pg_stat_activity
 WHERE datname = current_database() AND pid <> pg_backend_pid()
   AND usename = :'scanner_role';
SQL
sed -n '1,80p' "${SCN5_ROOT}/logs/source-activity.txt"

SCN5_SOURCE_SCANNER_SESSIONS="$(docker exec -i "${SCN5_PG_CONTAINER}" \
  psql -XqAt -U "${SCN5_PG_OPERATOR}" -d "${SCN5_SOURCE_DB}" \
  -v scanner_role="${SCN5_SCANNER_ROLE}" <<'SQL'
SELECT count(*)
  FROM pg_stat_activity
 WHERE datname = current_database() AND pid <> pg_backend_pid()
   AND usename = :'scanner_role';
SQL
)"
test "${SCN5_SOURCE_SCANNER_SESSIONS}" -eq 0

Verified by: both containers report Running=false; both immutable snapshot manifests and every bundle checksum re-verify; Web’s manifest-derived matrix has exact internal copy/stamp closure; Worker’s matrix additionally binds every required copy to the outgoing Worker image; scanner:jobs has XLEN=0; the exact scanner-workers group has pending=0, lag=0; and the role-wide activity probe names no Scanner session. A non-empty queue, pending delivery, lag, or database session keeps both writers fenced until it is explained; do not delete a job or terminate a backend merely to make an assertion green.

5. Freeze the source fingerprint and capture the recovery backup

The fingerprint defined in step 1 derives the leaf-table list from the source and excludes only the migration ledger; the same function is used before copy, before repoint, after repoint, at soak completion, and during a pre-intake rollback.

scn5_fingerprint "${SCN5_SOURCE_DB}" "${SCN5_ROOT}/parity/source-frozen.tsv"
scn5_assert_copy_fingerprint "${SCN5_ROOT}/parity/source-frozen.tsv"
scn5_assert_writers_stopped

Then take the full, restorable source-schema backup while the writers remain fenced:

docker exec "${SCN5_PG_CONTAINER}" pg_dump -U "${SCN5_PG_OPERATOR}" -d "${SCN5_SOURCE_DB}" \
  -Fc --schema="${SCN5_SCHEMA}" \
  > "${SCN5_ROOT}/backup/source-scanner-full.dump"

test -s "${SCN5_ROOT}/backup/source-scanner-full.dump"
sha256sum "${SCN5_ROOT}/backup/source-scanner-full.dump" \
  > "${SCN5_ROOT}/backup/source-scanner-full.dump.sha256"
sed -n '1,4p' "${SCN5_ROOT}/backup/source-scanner-full.dump.sha256"
docker exec -i "${SCN5_PG_CONTAINER}" pg_restore --list \
  < "${SCN5_ROOT}/backup/source-scanner-full.dump" \
  > "${SCN5_ROOT}/backup/source-scanner-full.list"

scn5_dump_table_data_inventory() { # $1=pg_restore list, $2=sorted non-ledger table receipt
  local list="$1" output="$2"
  awk -v schema="${SCN5_SCHEMA}" '
    {
      for (field = 1; field <= NF - 3; field++) {
        if ($field == "TABLE" && $(field + 1) == "DATA" && $(field + 2) == schema) {
          print $(field + 3)
          next
        }
      }
    }
  ' "${list}" | awk '$0 != "schema_migrations"' | LC_ALL=C sort -u > "${output}"
}
scn5_dump_table_data_inventory \
  "${SCN5_ROOT}/backup/source-scanner-full.list" \
  "${SCN5_ROOT}/backup/source-scanner-full.tables.txt"
test "$(wc -l < "${SCN5_ROOT}/backup/source-scanner-full.tables.txt")" -eq 84
cmp "${SCN5_ROOT}/shape/source-copy-leaf-tables.txt" \
    "${SCN5_ROOT}/backup/source-scanner-full.tables.txt"
scn5_fingerprint "${SCN5_SOURCE_DB}" "${SCN5_ROOT}/parity/source-after-full-backup.tsv"
scn5_assert_copy_fingerprint "${SCN5_ROOT}/parity/source-after-full-backup.tsv"
cmp "${SCN5_ROOT}/parity/source-frozen.tsv" \
    "${SCN5_ROOT}/parity/source-after-full-backup.tsv"
scn5_assert_writers_stopped

Verified by: source-frozen.tsv has 84 rows; the full dump is non-empty and has a recorded digest; its table of contents lists TABLE DATA entries for all 84 source leaf tables; and a second source fingerprint after the dump is byte-identical to source-frozen.tsv. A pre-fence backup is not the rollback snapshot for this window.

6. Pin sequence and identity ownership before the dump

This step is a REQUIRED gate, not a note, and Scanner is not like its sibling windows. The scanner schema carries a MIXED identity model — two ordinary serials and one GENERATED ALWAYS AS IDENTITY — across exactly three columns, named here because a reader who only knows serials will not think to check the third:

Column	Model	Sequence	Source state at prep
classification_history.id	serial (nextval default)	classification_history_id_seq	last_value=1
execution_phases.id	serial (nextval default)	execution_phases_id_seq	never called
oci_referrers.id	GENERATED ALWAYS AS IDENTITY	oci_referrers_id_seq	last_value=16, 16 rows

oci_referrers is the dangerous one. A COPY-based restore populates its 16 ids correctly and does not advance the backing sequence (measured — hazard H3), so the first insert after the window collides on the primary key. That failure lands after the window closes and everyone has gone home. Steps 6 and 9 exist to make it impossible.

For contrast, the AAI-9 window measured zero sequences and zero identity columns on both sides and has no setval class at all. Do not read the two runbooks as symmetric here.

The chains agree on this model — it is not a target-only artefact. Verified three ways: stellaops_platform reports oci_referrers.id identity=YES/ALWAYS with the other two as nextval defaults; the freshly converged target reports the identical three rows; and the legacy 003_sprint014_oci_referrers.sql:13 and the consolidated baseline :6676 declare the column byte-identically (id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY), with no ALTER TABLE against oci_referrers anywhere in the baseline. ConsolidatedScannerSchemaTests could not have told you any of this: its columns projection omits is_identity/identity_generation.

scn5_sequence_inventory() {
  local database="$1" output="$2"
  docker exec "${SCN5_PG_CONTAINER}" psql -XqAt -F $'\t' -v ON_ERROR_STOP=1 \
    -U "${SCN5_PG_OPERATOR}" -d "${database}" -c \
    "SELECT format('%I.%I', sn.nspname, s.relname),
            format('%I.%I', tn.nspname, t.relname),
            a.attname,
            CASE a.attidentity WHEN 'a' THEN 'identity-always'
                               WHEN 'd' THEN 'identity-default'
                               ELSE 'owned-sequence' END
       FROM pg_class s
       JOIN pg_namespace sn ON sn.oid = s.relnamespace
       JOIN pg_depend d ON d.classid = 'pg_class'::regclass AND d.objid = s.oid
                       AND d.refclassid = 'pg_class'::regclass AND d.deptype IN ('a','i')
       JOIN pg_class t ON t.oid = d.refobjid
       JOIN pg_namespace tn ON tn.oid = t.relnamespace
       JOIN pg_attribute a ON a.attrelid = t.oid AND a.attnum = d.refobjsubid
      WHERE s.relkind = 'S' AND sn.nspname = '${SCN5_SCHEMA}' AND tn.nspname = '${SCN5_SCHEMA}'
        AND s.relname NOT LIKE 'epss%'
      ORDER BY 1,2,3;" > "${output}"
}

scn5_sequence_inventory "${SCN5_SOURCE_DB}" "${SCN5_ROOT}/parity/source-sequences.tsv"
scn5_sequence_inventory "${SCN5_TARGET_DB}" "${SCN5_ROOT}/parity/target-sequences.before.tsv"
cmp "${SCN5_ROOT}/parity/source-sequences.tsv" "${SCN5_ROOT}/parity/target-sequences.before.tsv"
for sequence_inventory in \
    "${SCN5_ROOT}/parity/source-sequences.tsv" \
    "${SCN5_ROOT}/parity/target-sequences.before.tsv"; do
  test "$(wc -l < "${sequence_inventory}")" -eq 3
  grep -Fx $'scanner.classification_history_id_seq\tscanner.classification_history\tid\towned-sequence' \
    "${sequence_inventory}"
  grep -Fx $'scanner.execution_phases_id_seq\tscanner.execution_phases\tid\towned-sequence' \
    "${sequence_inventory}"
  grep -Fx $'scanner.oci_referrers_id_seq\tscanner.oci_referrers\tid\tidentity-always' \
    "${sequence_inventory}"
done

Verified by: cmp exits 0 on a three-row inventory — classification_history_id_seq → classification_history.id owned-sequence, execution_phases_id_seq → execution_phases.id owned-sequence, oci_referrers_id_seq → oci_referrers.id identity-always. The NOT LIKE 'epss%' filter is the H4 exclusion; it must remove exactly two target-only rows and nothing from the source. A missing, extra, renamed or differently-owned sequence is NO-GO — do not accept row parity without resolving it.

Record the frozen source sequence state now, for step 9:

docker exec "${SCN5_PG_CONTAINER}" psql -XqAt -F $'\t' -U "${SCN5_PG_OPERATOR}" -d "${SCN5_SOURCE_DB}" -c \
  "SELECT sequencename, coalesce(last_value::text,'<never-called>')
     FROM pg_sequences WHERE schemaname = '${SCN5_SCHEMA}' ORDER BY 1;" \
  > "${SCN5_ROOT}/parity/source-sequence-state.tsv"
test "$(wc -l < "${SCN5_ROOT}/parity/source-sequence-state.tsv")" -eq 3
awk -F'\t' 'NF != 2 || $1 == "" || $2 == "" { exit 1 }
  END { if (NR != 3) exit 1 }' "${SCN5_ROOT}/parity/source-sequence-state.tsv"

Verified by: three rows, matching the H3 figures at prep time (classification_history_id_seq=1, execution_phases_id_seq=<never-called>, oci_referrers_id_seq=16) or their post-freeze successors. <never-called> is a meaningful state, not a missing value: it decides whether the next value is last_value or last_value + 1.

7. Create the fenced data-only dump

docker exec "${SCN5_PG_CONTAINER}" pg_dump \
  -U "${SCN5_PG_OPERATOR}" -d "${SCN5_SOURCE_DB}" -Fc \
  --data-only --schema="${SCN5_SCHEMA}" \
  --load-via-partition-root --no-owner --no-privileges \
  --exclude-table="${SCN5_SCHEMA}.schema_migrations" \
  > "${SCN5_ROOT}/backup/source-scanner-fenced-data.dump"

test -s "${SCN5_ROOT}/backup/source-scanner-fenced-data.dump"
sha256sum "${SCN5_ROOT}/backup/source-scanner-fenced-data.dump" \
  > "${SCN5_ROOT}/backup/source-scanner-fenced-data.dump.sha256"
sed -n '1,4p' "${SCN5_ROOT}/backup/source-scanner-fenced-data.dump.sha256"
docker exec -i "${SCN5_PG_CONTAINER}" pg_restore --list \
  < "${SCN5_ROOT}/backup/source-scanner-fenced-data.dump" \
  > "${SCN5_ROOT}/backup/source-scanner-fenced-data.list"

! grep -Eq "TABLE DATA ${SCN5_SCHEMA} schema_migrations" \
  "${SCN5_ROOT}/backup/source-scanner-fenced-data.list"
! grep -Eq 'TABLE DATA (signals|eventing|policy|public) ' \
  "${SCN5_ROOT}/backup/source-scanner-fenced-data.list"
test "$(grep -c 'SEQUENCE SET' "${SCN5_ROOT}/backup/source-scanner-fenced-data.list" || true)" -eq 3
scn5_dump_table_data_inventory \
  "${SCN5_ROOT}/backup/source-scanner-fenced-data.list" \
  "${SCN5_ROOT}/backup/source-scanner-fenced-data.tables.txt"
test "$(wc -l < "${SCN5_ROOT}/backup/source-scanner-fenced-data.tables.txt")" -eq 84
cmp "${SCN5_ROOT}/shape/source-copy-leaf-tables.txt" \
    "${SCN5_ROOT}/backup/source-scanner-fenced-data.tables.txt"
scn5_assert_writers_stopped

Verified by, all five:

1. ! grep -Eq 'TABLE DATA ${SCN5_SCHEMA} schema_migrations' "${SCN5_ROOT}/backup/source-scanner-fenced-data.list" — H2’s exclusion is asserted from the dump’s own table of contents, not from the flag being typed.
2. ! grep -Eq 'TABLE DATA (signals|eventing|policy|public) ' "${SCN5_ROOT}/backup/source-scanner-fenced-data.list" — the --schema filter held.
3. grep -c 'SEQUENCE SET' "${SCN5_ROOT}/backup/source-scanner-fenced-data.list" is 3 — the setvals H3 depends on are in the dump. If this is 0, stop: the restore will leave every sequence behind its data and the first post-cutover insert into oci_referrers will collide on the primary key.
4. the sorted TABLE DATA inventory has exactly 84 rows and compares byte-identically to the source-derived copy-leaf inventory from step 3;
5. scn5_assert_writers_stopped still finds both Scanner writers stopped after the dump.

--load-via-partition-root routes artifact_boms children through the parent, so the restore depends on a partition covering each row’s inserted_at rather than on a child of a matching name. Step 3 is what makes that true.

8. Restore once, atomically, into the migration-001 target

for c in stellaops-scanner-web stellaops-scanner-worker; do
  test "$(docker inspect -f '{{.State.Running}}' "$c")" = false
done

set +e
docker exec -i "${SCN5_PG_CONTAINER}" pg_restore \
  -U "${SCN5_PG_OPERATOR}" -d "${SCN5_TARGET_DB}" \
  --single-transaction --exit-on-error --data-only --no-owner --no-privileges \
  < "${SCN5_ROOT}/backup/source-scanner-fenced-data.dump" \
  > "${SCN5_ROOT}/logs/target-restore.txt" 2>&1
SCN5_RESTORE_RC=$?
set -e
sed -n '1,160p' "${SCN5_ROOT}/logs/target-restore.txt"
test "${SCN5_RESTORE_RC}" -eq 0
scn5_assert_writers_stopped

Verified by: exit status 0 and an empty target-restore.txt body (pg_restore is silent on success). Do not retry over a partially changed target: --single-transaction is precisely what makes one retry safe against a still-empty target after the failure is diagnosed. Do not truncate or merge a non-empty target.

9. Prove old-shape copy parity before migration

The parity harness derives its table list from the source, then runs the identical generated SQL against both databases — so the comparison cannot silently skip a table, and the target’s 24 EPSS-only tables cannot pollute it. scn5_fingerprint was defined before the freeze in step 5.

scn5_fingerprint "${SCN5_SOURCE_DB}" "${SCN5_ROOT}/parity/source-after.tsv"
scn5_assert_copy_fingerprint "${SCN5_ROOT}/parity/source-after.tsv"
cmp "${SCN5_ROOT}/parity/source-frozen.tsv" "${SCN5_ROOT}/parity/source-after.tsv"

scn5_fingerprint "${SCN5_TARGET_DB}" "${SCN5_ROOT}/parity/target-after.tsv"
scn5_assert_copy_fingerprint "${SCN5_ROOT}/parity/target-after.tsv"
set +e
diff -u "${SCN5_ROOT}/parity/source-after.tsv" "${SCN5_ROOT}/parity/target-after.tsv" \
  > "${SCN5_ROOT}/parity/table-parity.diff"
SCN5_PARITY_RC=$?
set -e
sed -n '1,160p' "${SCN5_ROOT}/parity/table-parity.diff"
test "${SCN5_PARITY_RC}" -eq 0
test ! -s "${SCN5_ROOT}/parity/table-parity.diff"

Verified by, all of:

1. cmp on the frozen vs re-read source exits 0 — the freeze held; nothing wrote to the source during the copy.
2. table-parity.diff is empty — 84 tables agree on both count and content digest. SET TimeZone='UTC' and SET extra_float_digits = 3 are load-bearing: without them the row text used for the digest is session-dependent and a clean diff would be luck.
3. Both files have 84 lines; the non-empty subset is the 14 tables of the copy contract summing to 135 rows: awk -F'\t' '$2>0 {s+=$2; n++} END {print n, s}' "${SCN5_ROOT}/parity/target-after.tsv" prints 14 135.

Then the separate sequence-state class (H3) — a mismatch here can coexist with a perfect digest diff:

docker exec "${SCN5_PG_CONTAINER}" psql -XqAt -F $'\t' -U "${SCN5_PG_OPERATOR}" -d "${SCN5_TARGET_DB}" -c \
  "SELECT sequencename, coalesce(last_value::text,'<never-called>')
     FROM pg_sequences WHERE schemaname = '${SCN5_SCHEMA}' AND sequencename NOT LIKE 'epss%'
    ORDER BY 1;" > "${SCN5_ROOT}/parity/target-sequence-state.tsv"

test "$(wc -l < "${SCN5_ROOT}/parity/target-sequence-state.tsv")" -eq 3
awk -F'\t' 'NF != 2 || $1 == "" || $2 == "" { exit 1 }
  END { if (NR != 3) exit 1 }' "${SCN5_ROOT}/parity/target-sequence-state.tsv"
diff -u "${SCN5_ROOT}/parity/source-sequence-state.tsv" \
        "${SCN5_ROOT}/parity/target-sequence-state.tsv"

Verified by: an empty diff after filtering the EPSS rows out of the source file the same way.

Then the REQUIRED per-column gate. Run it for all three columns of step 6’s table — not only the identity one — because a serial whose sequence lags is the same delayed primary-key collision. This query was executed against both databases in prep, so it runs as written:

docker exec -i "${SCN5_PG_CONTAINER}" psql -XqAt -F '|' \
  -U "${SCN5_PG_OPERATOR}" -d "${SCN5_TARGET_DB}" \
  -v ON_ERROR_STOP=1 -f - > "${SCN5_ROOT}/parity/identity-gate.txt" <<'SQL'
SELECT v.col,
       coalesce(v.maxid::text, '<no-rows>')                                      AS max_id,
       coalesce(pg_sequence_last_value(v.seq::regclass)::text, '<never-called>') AS seq_last_value,
       v.seq                                                                     AS sequence,
       CASE
         WHEN v.maxid IS NULL THEN true
         ELSE pg_sequence_last_value(v.seq::regclass) IS NOT NULL
          AND pg_sequence_last_value(v.seq::regclass) >= v.maxid
       END                                                                       AS valid
FROM (
  SELECT 'classification_history.id' AS col,
         (SELECT max(id) FROM scanner.classification_history) AS maxid,
         pg_get_serial_sequence('scanner.classification_history','id') AS seq
  UNION ALL
  SELECT 'execution_phases.id',
         (SELECT max(id) FROM scanner.execution_phases),
         pg_get_serial_sequence('scanner.execution_phases','id')
  UNION ALL
  SELECT 'oci_referrers.id',
         (SELECT max(id) FROM scanner.oci_referrers),
         pg_get_serial_sequence('scanner.oci_referrers','id')
) v
ORDER BY 1;
SQL
sed -n '1,40p' "${SCN5_ROOT}/parity/identity-gate.txt"
test "$(wc -l < "${SCN5_ROOT}/parity/identity-gate.txt")" -eq 3
SCN5_INVALID_SEQUENCE_COUNT="$(awk -F'|' '$5 != "t" { n++ } END { print n + 0 }' \
  "${SCN5_ROOT}/parity/identity-gate.txt")"
test "${SCN5_INVALID_SEQUENCE_COUNT}" -eq 0

Two properties make this the right lookup for a mixed model, both confirmed by running it: pg_get_serial_sequence resolves the identity column as well as the serial defaults (it returns scanner.oci_referrers_id_seq for the GENERATED ALWAYS column), and pg_sequence_last_value returns NULL exactly when is_called is false — so <never-called> in the output is the is_called edge, without a second query.

Verified by: exactly three rows and valid=t on every row; for non-empty tables this means seq_last_value >= max_id. The source values this must reproduce, read at prep and to be re-read after the freeze, are

classification_history.id  max_id=1          seq_last_value=1
execution_phases.id        max_id=<no-rows>  seq_last_value=<never-called>
oci_referrers.id           max_id=16         seq_last_value=16

execution_phases is the legitimate exception: <no-rows> with <never-called> is correct, and forcing a setval on it would be wrong. For the other two, <never-called> against a non-empty table is the failure — pg_dump emits setval(…, N, true), so a restored table whose sequence reads <never-called> did not receive it.

If any row reports valid=f, stop. Do not repoint, hand-patch a sequence, or restore over the loaded target. The restore populated rows without a safe next identity. Recreate the target at exact migration 001, restart at step 3, and repeat steps 3–9 from the same checksum-verified dump.

Finally, re-take the ten shape projections against the loaded target:

scn5_assert_shape_contains_source post-restore

Verified by: post-restore-summary.txt has missing=0 for all ten rows; every post-restore-<projection>.extra.tsv is byte-identical to its corresponding pre-copy extra receipt; and post-restore-rls.diff.tsv is empty. A restore can only add rows, but this catches one that quietly omitted or changed an object and one that introduced any new target-only object.

Forecast migration 003 before running it:

docker exec "${SCN5_PG_CONTAINER}" psql -XqAt -F '|' -U "${SCN5_PG_OPERATOR}" -d "${SCN5_TARGET_DB}" -c \
  "WITH exact_scan_tenants AS (
       SELECT scan_id, min(btrim(tenant_id)) AS tenant_id
         FROM scanner.scan_runtime_state
        WHERE tenant_id IS NOT NULL AND btrim(tenant_id) <> ''
        GROUP BY scan_id
       HAVING count(DISTINCT btrim(tenant_id)) = 1
   ), observed AS (
       SELECT DISTINCT btrim(tenant_id) AS tenant_id
         FROM scanner.scan_runtime_state
        WHERE tenant_id IS NOT NULL AND btrim(tenant_id) <> ''
   )
   SELECT count(*),
          count(*) FILTER (WHERE exact_scan_tenants.scan_id IS NOT NULL),
          count(*) FILTER (WHERE exact_scan_tenants.scan_id IS NULL),
          (SELECT count(*) FROM observed)
     FROM scanner.artifact_boms artifact
     LEFT JOIN exact_scan_tenants ON exact_scan_tenants.scan_id = artifact.build_id;" \
  > "${SCN5_ROOT}/parity/tenant-attribution-forecast.txt"
sed -n '1,20p' "${SCN5_ROOT}/parity/tenant-attribution-forecast.txt"
grep -Fx '8|7|1|1' "${SCN5_ROOT}/parity/tenant-attribution-forecast.txt"
test "$(wc -l < "${SCN5_ROOT}/parity/tenant-attribution-forecast.txt")" -eq 1

docker exec "${SCN5_PG_CONTAINER}" psql -XqAt -F $'\t' -U "${SCN5_PG_OPERATOR}" -d "${SCN5_SOURCE_DB}" -c \
  "SET TimeZone='UTC'; SET extra_float_digits=3;
   SELECT count(*)::text,
          coalesce(md5(string_agg(md5(ROW(build_id, canonical_bom_sha256, payload_digest, inserted_at)::text), '' ORDER BY build_id, inserted_at)), '<empty>')
     FROM scanner.artifact_boms;" \
  > "${SCN5_ROOT}/parity/source-artifact-boms-old-shape.tsv"

Verified by: the forecast is 8|7|1|1: eight rows, seven exact build_id = scan_id attributions, one unresolved row, and exactly one observed Scanner tenant. Any other result is a changed estate. In particular, more than one observed tenant makes the unresolved row ambiguous and must abort migration 003; stop and do not substitute an operator-chosen tenant.

Any failure in this step keeps the freeze. Do not repoint, do not delete a key, do not touch a grant. Go to Rollback.

10. Apply 002 and 003 through Scanner’s supported migration-only path

The candidate image is the migration executable. It must run as a one-off Compose service with the canonical Scanner connection, no dependencies, no published port, and the same exact image overlay that will be repointed. Do not run SQL files by hand and do not start scanner-web normally.

Before running the next fence, create tmp/scn5-<window-id>/topology/scanner-bundle-inventory.commit-message.txt with apply_patch, not a shell redirection. Its exact four lines are:

build(scanner): align SCN-5 mounted bundle inventories

Regenerate the Scanner base and full-analyzer bundle inventories from the
post-SCN-3 closed worker graph before the SCN-5 live cutover.

The executable hash check below rejects a missing, differently encoded, or edited message.

Also use apply_patch to create these exact LF-terminated static overlays under the same topology directory. Runtime values are Compose-expanded from the exported names; shell heredocs are forbidden.

candidate-images.override.yml:

services:
  scanner-web:
    image: ${SCN5_WEB_CANDIDATE_IMAGE:?SCN5_WEB_CANDIDATE_IMAGE is required}
    pull_policy: never
    environment:
      SCANNER_SCANNER__CUTOVER__INTAKEWINDOWID: "${SCN5_WINDOW_ID:?SCN5_WINDOW_ID is required}"
  scanner-worker:
    image: ${SCN5_WORKER_CANDIDATE_IMAGE:?SCN5_WORKER_CANDIDATE_IMAGE is required}
    pull_policy: never

owner-candidate-images.override.yml:

services:
  authority:
    image: ${SCN5_AUTHORITY_CANDIDATE_IMAGE:?SCN5_AUTHORITY_CANDIDATE_IMAGE is required}
    pull_policy: never
  policy-engine:
    image: ${SCN5_POLICY_CANDIDATE_IMAGE:?SCN5_POLICY_CANDIDATE_IMAGE is required}
    pull_policy: never

SCN5_BUNDLE_STAGE_ROOT="${SCN5_ROOT}/bundle-promotion"
test "${SCN5_BUNDLE_STAGE_ROOT}" = "tmp/scn5-${SCN5_WINDOW_ID}/bundle-promotion"
test ! -e "${SCN5_BUNDLE_STAGE_ROOT}"
scn5_verify_bundle_root "${SCN5_BUNDLE_SOURCE_BASE}" "${SCN5_BUNDLE_EXPECTED_BASE}" \
  "${SCN5_ROOT}/topology/prep-scanner-base"
scn5_verify_bundle_root "${SCN5_BUNDLE_SOURCE_FULL}" "${SCN5_BUNDLE_EXPECTED_FULL}" \
  "${SCN5_ROOT}/topology/prep-scanner-full"
mkdir -p "${SCN5_BUNDLE_STAGE_ROOT}"
cp -a -- "${SCN5_BUNDLE_SOURCE_BASE}" "${SCN5_BUNDLE_STAGE_ROOT}/base"
cp -a -- "${SCN5_BUNDLE_SOURCE_FULL}" "${SCN5_BUNDLE_STAGE_ROOT}/scanner-full-analyzers"

# Reassert the snapshots captured immediately after step 4.
scn5_assert_writers_stopped
scn5_assert_rollback_bundle_snapshots

# Promote the generated roots only while stopped. Bounds are exact absolute paths.
test "${SCN5_BUNDLE_TARGET_BASE}" = "${SCN5_REPO_ROOT}/devops/plugins/scanner/base"
test "${SCN5_BUNDLE_TARGET_FULL}" = \
  "${SCN5_REPO_ROOT}/devops/plugins/scanner/scanner-full-analyzers"
test "$(git branch --show-current)" = main
test "$(git rev-parse HEAD)" = "${SCN5_PRIMARY_PREP_SHA}"
test -z "$(git diff --cached --name-only)"
test -z "$(git status --porcelain=v1 --untracked-files=no -- \
  devops/plugins/scanner/base devops/plugins/scanner/scanner-full-analyzers)"
rm -rf -- "${SCN5_BUNDLE_TARGET_BASE}" "${SCN5_BUNDLE_TARGET_FULL}"
mv -- "${SCN5_BUNDLE_STAGE_ROOT}/base" "${SCN5_BUNDLE_TARGET_BASE}"
mv -- "${SCN5_BUNDLE_STAGE_ROOT}/scanner-full-analyzers" "${SCN5_BUNDLE_TARGET_FULL}"
scn5_verify_bundle_root "${SCN5_BUNDLE_TARGET_BASE}" "${SCN5_BUNDLE_EXPECTED_BASE}" \
  "${SCN5_ROOT}/topology/promoted-scanner-base"
scn5_verify_bundle_root "${SCN5_BUNDLE_TARGET_FULL}" "${SCN5_BUNDLE_EXPECTED_FULL}" \
  "${SCN5_ROOT}/topology/promoted-scanner-full"
diff -qr "${SCN5_BUNDLE_SOURCE_BASE}" "${SCN5_BUNDLE_TARGET_BASE}" \
  > "${SCN5_ROOT}/topology/prep-vs-promoted-base.diff"
diff -qr "${SCN5_BUNDLE_SOURCE_FULL}" "${SCN5_BUNDLE_TARGET_FULL}" \
  > "${SCN5_ROOT}/topology/prep-vs-promoted-full.diff"

# Review exact hunks before staging, then run this staging fence by itself.
git diff -- devops/plugins/scanner/base devops/plugins/scanner/scanner-full-analyzers
git diff --name-only -- devops/plugins/scanner/base \
  devops/plugins/scanner/scanner-full-analyzers \
  > "${SCN5_ROOT}/topology/expected-inventory-tracked-changes.txt"
git diff --check -- devops/plugins/scanner/base devops/plugins/scanner/scanner-full-analyzers
SCN5_INVENTORY_COMMIT_MESSAGE="${SCN5_ROOT}/topology/scanner-bundle-inventory.commit-message.txt"
test "$(sha256sum "${SCN5_INVENTORY_COMMIT_MESSAGE}" | awk '{print $1}')" = \
  'c2ce341190b76aa28efc92655ccd74525daad72ea0e32df44196fbb0de9db6aa'
SCN5_INVENTORY_PARENT_SHA="$(git rev-parse HEAD)"
git add -- devops/plugins/scanner/base devops/plugins/scanner/scanner-full-analyzers

Run the next fence in the immediately following tool invocation, with no intervening command. It re-reads the complete shared index and HEAD, then makes the required bare commit from the exact staged set. A primary-tree pathspec commit is forbidden because it would ship working-tree content instead of the reviewed index. If either the index or HEAD moved, the fence fails before git commit.

set -euo pipefail
test "$(git rev-parse HEAD)" = "${SCN5_INVENTORY_PARENT_SHA}"
git diff --cached --name-only > "${SCN5_ROOT}/topology/cached-paths.txt"
cmp "${SCN5_ROOT}/topology/expected-inventory-tracked-changes.txt" \
  "${SCN5_ROOT}/topology/cached-paths.txt"
git commit -F "${SCN5_INVENTORY_COMMIT_MESSAGE}"
SCN5_SOURCE_SHA="$(git rev-parse HEAD)"
test "$(git rev-parse HEAD^)" = "${SCN5_INVENTORY_PARENT_SHA}"
test "$(git log -1 --format=%H -- src)" = "${SCN5_SCANNER_SOURCE_REVISION}"
git show --stat --oneline HEAD
git diff-tree --no-commit-id --name-only -r HEAD | LC_ALL=C sort \
  > "${SCN5_ROOT}/topology/committed-inventory-paths.txt"
LC_ALL=C sort "${SCN5_ROOT}/topology/expected-inventory-tracked-changes.txt" \
  > "${SCN5_ROOT}/topology/expected-inventory-committed-paths.txt"
cmp "${SCN5_ROOT}/topology/expected-inventory-committed-paths.txt" \
  "${SCN5_ROOT}/topology/committed-inventory-paths.txt"
test -z "$(git diff --cached --name-only)"
git diff --quiet --exit-code -- \
  devops/plugins/scanner/base devops/plugins/scanner/scanner-full-analyzers
scn5_assert_writers_stopped

# Build all four images serially from a fresh detached worktree at the exact commit.
SCN5_BUNDLE_BUILD_ROOT="${SCN5_REPO_ROOT}/tmp/worktrees/scn-live-build-${SCN5_SOURCE_SHA:0:12}"
test ! -e "${SCN5_BUNDLE_BUILD_ROOT}"
GIT_LFS_SKIP_SMUDGE=1 git worktree add --detach "${SCN5_BUNDLE_BUILD_ROOT}" "${SCN5_SOURCE_SHA}"
test -z "$(git -C "${SCN5_BUNDLE_BUILD_ROOT}" branch --show-current)"
for build_service in authority policy-engine scanner-web scanner-worker; do
  (cd -- "${SCN5_BUNDLE_BUILD_ROOT}" && CI=true IMAGE_TAG=dev \
    bash devops/docker/build-service-publish.sh "${build_service}")
done
test "$(git -C "${SCN5_BUNDLE_BUILD_ROOT}" rev-parse HEAD)" = "${SCN5_SOURCE_SHA}"
test "$(git -C "${SCN5_BUNDLE_BUILD_ROOT}" log -1 --format=%H -- src)" = \
  "${SCN5_SCANNER_SOURCE_REVISION}"
# `status --porcelain` can report CRLF-normalized tracked files as dirty in a Windows worktree.
# Retain it as diagnostic evidence; use Git's canonical diff as the fail-closed content gate.
git -C "${SCN5_BUNDLE_BUILD_ROOT}" status --porcelain=v1 --untracked-files=no \
  > "${SCN5_ROOT}/topology/final-build-raw-status.txt"
git -C "${SCN5_BUNDLE_BUILD_ROOT}" diff --name-only \
  > "${SCN5_ROOT}/topology/final-build-tracked-changes.txt"
test ! -s "${SCN5_ROOT}/topology/final-build-tracked-changes.txt"
git -C "${SCN5_BUNDLE_BUILD_ROOT}" diff --quiet --exit-code
SCN5_BUNDLE_SOURCE_BASE="${SCN5_BUNDLE_BUILD_ROOT}/devops/plugins/scanner/base"
SCN5_BUNDLE_SOURCE_FULL="${SCN5_BUNDLE_BUILD_ROOT}/devops/plugins/scanner/scanner-full-analyzers"
SCN5_BUNDLE_SOURCE_HOST="${SCN5_BUNDLE_BUILD_ROOT}/tmp/build/stella-publish/scanner-worker"
diff -qr "${SCN5_BUNDLE_SOURCE_BASE}" "${SCN5_BUNDLE_TARGET_BASE}" \
  > "${SCN5_ROOT}/topology/final-build-vs-promoted-base.diff"
diff -qr "${SCN5_BUNDLE_SOURCE_FULL}" "${SCN5_BUNDLE_TARGET_FULL}" \
  > "${SCN5_ROOT}/topology/final-build-vs-promoted-full.diff"

SCN5_WEB_CANDIDATE_IMAGE="stellaops/scanner-web:scn5-${SCN5_SOURCE_SHA:0:12}-${SCN5_WINDOW_ID}"
SCN5_WORKER_CANDIDATE_IMAGE="stellaops/scanner-worker:scn5-${SCN5_SOURCE_SHA:0:12}-${SCN5_WINDOW_ID}"
SCN5_AUTHORITY_CANDIDATE_IMAGE="stellaops/authority:scn5-${SCN5_SOURCE_SHA:0:12}-${SCN5_WINDOW_ID}"
SCN5_POLICY_CANDIDATE_IMAGE="stellaops/policy-engine:scn5-${SCN5_SOURCE_SHA:0:12}-${SCN5_WINDOW_ID}"
docker tag stellaops/scanner-web:dev "${SCN5_WEB_CANDIDATE_IMAGE}"
docker tag stellaops/scanner-worker:dev "${SCN5_WORKER_CANDIDATE_IMAGE}"
docker tag stellaops/authority:dev "${SCN5_AUTHORITY_CANDIDATE_IMAGE}"
docker tag stellaops/policy-engine:dev "${SCN5_POLICY_CANDIDATE_IMAGE}"
export SCN5_WEB_CANDIDATE_IMAGE SCN5_WORKER_CANDIDATE_IMAGE
export SCN5_AUTHORITY_CANDIDATE_IMAGE SCN5_POLICY_CANDIDATE_IMAGE SCN5_WINDOW_ID
SCN5_WEB_CANDIDATE_IMAGE_ID="$(docker image inspect "${SCN5_WEB_CANDIDATE_IMAGE}" --format '{{.Id}}')"
SCN5_WORKER_CANDIDATE_IMAGE_ID="$(docker image inspect "${SCN5_WORKER_CANDIDATE_IMAGE}" --format '{{.Id}}')"
SCN5_AUTHORITY_CANDIDATE_IMAGE_ID="$(docker image inspect "${SCN5_AUTHORITY_CANDIDATE_IMAGE}" --format '{{.Id}}')"
SCN5_POLICY_CANDIDATE_IMAGE_ID="$(docker image inspect "${SCN5_POLICY_CANDIDATE_IMAGE}" --format '{{.Id}}')"
scn5_assert_candidate_image_ids() {
  test "$(docker image inspect "${SCN5_WEB_CANDIDATE_IMAGE}" --format '{{.Id}}')" = \
    "${SCN5_WEB_CANDIDATE_IMAGE_ID}"
  test "$(docker image inspect "${SCN5_WORKER_CANDIDATE_IMAGE}" --format '{{.Id}}')" = \
    "${SCN5_WORKER_CANDIDATE_IMAGE_ID}"
  test "$(docker image inspect "${SCN5_AUTHORITY_CANDIDATE_IMAGE}" --format '{{.Id}}')" = \
    "${SCN5_AUTHORITY_CANDIDATE_IMAGE_ID}"
  test "$(docker image inspect "${SCN5_POLICY_CANDIDATE_IMAGE}" --format '{{.Id}}')" = \
    "${SCN5_POLICY_CANDIDATE_IMAGE_ID}"
}
printf 'authority=%s\npolicy-engine=%s\nscanner-web=%s\nscanner-worker=%s\n' \
  "${SCN5_AUTHORITY_CANDIDATE_IMAGE_ID}" "${SCN5_POLICY_CANDIDATE_IMAGE_ID}" \
  "${SCN5_WEB_CANDIDATE_IMAGE_ID}" "${SCN5_WORKER_CANDIDATE_IMAGE_ID}"

Pause here. Use apply_patch to create devops/compose/docker-compose.local-estate-image-pins.override.yml. Write exactly four service entries (authority, policy-engine, scanner-web, scanner-worker), each with only image and pull_policy. Copy the four IDs printed above as literal image: sha256:<64 lowercase hex> values and set every pull_policy to never. Do not put a tag, shell variable, Compose interpolation, comment, anchor, or placeholder in this file. The next fence rejects additional content and proves a fresh canonical-helper invocation selects the four immutable IDs before any owner recreate.

scn5_assert_local_estate_pin \
  "${SCN5_AUTHORITY_CANDIDATE_IMAGE_ID}" "${SCN5_POLICY_CANDIDATE_IMAGE_ID}" \
  "${SCN5_WEB_CANDIDATE_IMAGE_ID}" "${SCN5_WORKER_CANDIDATE_IMAGE_ID}"
SCN5_CANDIDATE_LOCAL_PIN_SHA="$(sha256sum "${SCN5_LOCAL_ESTATE_PIN}" | awk '{print $1}')"
[[ "${SCN5_CANDIDATE_LOCAL_PIN_SHA}" =~ ^[0-9a-f]{64}$ ]]

scn5_verify_buildinfo() { # $1=image, $2=module, $3=receipt
  local image="$1" module="$2" receipt="$3"
  MSYS_NO_PATHCONV=1 docker run --rm --network none --read-only \
    --entrypoint /bin/cat "${image}" /app/buildinfo.json > "${receipt}"
  jq -e --arg sha "${SCN5_SOURCE_SHA}" --arg module "${module}" \
    '.gitSha == $sha and .module == $module and .worktreeState == "clean" and .dirtyFileCount == 0' \
    "${receipt}" >/dev/null
}
scn5_verify_buildinfo "${SCN5_WEB_CANDIDATE_IMAGE}" scanner-web \
  "${SCN5_ROOT}/topology/scanner-web.buildinfo.json"
scn5_verify_buildinfo "${SCN5_WORKER_CANDIDATE_IMAGE}" scanner-worker \
  "${SCN5_ROOT}/topology/scanner-worker.buildinfo.json"
scn5_capture_image_env_names "${SCN5_WEB_CANDIDATE_IMAGE}" \
  "${SCN5_ROOT}/topology/scanner-web.candidate-image-env-names.txt"
scn5_capture_image_env_names "${SCN5_WORKER_CANDIDATE_IMAGE}" \
  "${SCN5_ROOT}/topology/scanner-worker.candidate-image-env-names.txt"

# Authority and Policy move to the same exact source SHA before Scanner forcing calls.
SCN5_OWNER_PREP_ROOT="${SCN5_ROOT}/owner-repoint"
mkdir -p "${SCN5_OWNER_PREP_ROOT}"
scn5_verify_buildinfo "${SCN5_AUTHORITY_CANDIDATE_IMAGE}" authority \
  "${SCN5_OWNER_PREP_ROOT}/authority.candidate-buildinfo.json"
scn5_verify_buildinfo "${SCN5_POLICY_CANDIDATE_IMAGE}" policy-engine \
  "${SCN5_OWNER_PREP_ROOT}/policy-engine.candidate-buildinfo.json"
SCN5_OWNER_CANDIDATE_OVERLAY_FILE="${SCN5_ROOT}/topology/owner-candidate-images.override.yml"
SCN5_OWNER_CANDIDATE_OVERLAY="${SCN5_COMPOSE_ROOT_REL}/topology/owner-candidate-images.override.yml"
test "$(sha256sum "${SCN5_OWNER_CANDIDATE_OVERLAY_FILE}" | awk '{print $1}')" = \
  '5b8aa8471e73b4bc1c4a536e3e30043d9163b38b7b5e3255325a508c664b5056'
SCN5_OWNER_CANDIDATE_FILES="${SCN5_OWNER_CANDIDATE_OVERLAY}"
set +e
COMPOSE_EXTRA_FILES="${SCN5_OWNER_CANDIDATE_FILES}" \
  ./devops/compose/scripts/compose-cli.sh config --format json \
  | jq -e --arg authority "${SCN5_AUTHORITY_CANDIDATE_IMAGE}" \
      --arg policy "${SCN5_POLICY_CANDIDATE_IMAGE}" '
      .services.authority.image == $authority
      and .services["policy-engine"].image == $policy' >/dev/null
SCN5_OWNER_RENDER_RC=("${PIPESTATUS[@]}")
set -e
test "${#SCN5_OWNER_RENDER_RC[@]}" -eq 2
test "${SCN5_OWNER_RENDER_RC[0]}" -eq 0
test "${SCN5_OWNER_RENDER_RC[1]}" -eq 0
SCN5_LIVE_RECREATE_SINCE="$(date -u +'%Y-%m-%dT%H:%M:%SZ')"
COMPOSE_EXTRA_FILES="${SCN5_OWNER_CANDIDATE_FILES}" \
  ./devops/compose/scripts/compose-cli.sh recreate authority policy-engine \
  > "${SCN5_OWNER_PREP_ROOT}/recreate.log" 2>&1
for owner_container in stellaops-authority stellaops-policy-engine; do
  for _ in $(seq 1 60); do
    test "$(docker inspect "${owner_container}" --format '{{.State.Health.Status}}')" = healthy && break
    sleep 2
  done
  test "$(docker inspect "${owner_container}" --format '{{.State.Running}} {{.State.Health.Status}}')" = \
    "true healthy"
done
for service_module in stellaops-authority:authority stellaops-policy-engine:policy-engine; do
  service="${service_module%%:*}"
  module="${service_module##*:}"
  case "${module}" in
    authority) expected_image_id="${SCN5_AUTHORITY_CANDIDATE_IMAGE_ID}" ;;
    policy-engine) expected_image_id="${SCN5_POLICY_CANDIDATE_IMAGE_ID}" ;;
    *) exit 1 ;;
  esac
  test "$(docker inspect "${service}" --format '{{.Image}}')" = "${expected_image_id}"
  docker exec "${service}" cat /app/buildinfo.json \
    > "${SCN5_OWNER_PREP_ROOT}/${module}.running-buildinfo.json"
  jq -e --arg sha "${SCN5_SOURCE_SHA}" --arg module "${module}" \
    '.gitSha == $sha and .module == $module and .worktreeState == "clean" and .dirtyFileCount == 0' \
    "${SCN5_OWNER_PREP_ROOT}/${module}.running-buildinfo.json" >/dev/null
done
scn5_assert_writers_stopped

SCN5_CANDIDATE_OVERLAY_FILE="${SCN5_ROOT}/topology/candidate-images.override.yml"
SCN5_CANDIDATE_OVERLAY="${SCN5_COMPOSE_ROOT_REL}/topology/candidate-images.override.yml"
test "$(sha256sum "${SCN5_CANDIDATE_OVERLAY_FILE}" | awk '{print $1}')" = \
  '4e0c5147f9b926c3d1ef2791bb3fd5660a4b84c3747ebbdda8be9817402c4ae8'
test -f "devops/compose/${SCN5_CANDIDATE_OVERLAY}"

export COMPOSE_EXTRA_FILES="${SCN5_CANDIDATE_OVERLAY}"
scn5_assert_candidate_image_ids
scn5_render_topology "${SCN5_ROOT}/topology/rendered-names-only.json"
jq -e --arg web "${SCN5_WEB_CANDIDATE_IMAGE}" \
      --arg worker "${SCN5_WORKER_CANDIDATE_IMAGE}" \
      --arg source "${SCN5_BUNDLE_TARGET_BASE}" '
  .services["scanner-web"].image == $web
  and .services["scanner-worker"].image == $worker
  and .services["scanner-web"].scannerBaseSource == $source
  and .services["scanner-worker"].scannerBaseSource == $source
' "${SCN5_ROOT}/topology/rendered-names-only.json" >/dev/null

scn5_assert_candidate_image_ids
set +e
./devops/compose/scripts/compose-cli.sh raw -- run --rm --no-deps \
  --entrypoint dotnet scanner-web /app/StellaOps.Scanner.WebService.dll --migrate-only \
  > "${SCN5_ROOT}/logs/migrate-only-first.txt" 2>&1
SCN5_MIGRATE_FIRST_RC=$?
set -e
sed -n '1,240p' "${SCN5_ROOT}/logs/migrate-only-first.txt"
test "${SCN5_MIGRATE_FIRST_RC}" -eq 0
grep -F "applied=2, already-applied=1" "${SCN5_ROOT}/logs/migrate-only-first.txt"
scn5_assert_candidate_image_ids
scn5_assert_writers_stopped

docker exec "${SCN5_PG_CONTAINER}" psql -XqAt -U "${SCN5_PG_OPERATOR}" -d "${SCN5_TARGET_DB}" -c \
  "SELECT migration_name || ' [' || category || ']'
     FROM scanner.schema_migrations ORDER BY migration_name;" \
  > "${SCN5_ROOT}/parity/migration-ledger.txt"
sed -n '1,20p' "${SCN5_ROOT}/parity/migration-ledger.txt"
test "$(cat "${SCN5_ROOT}/parity/migration-ledger.txt")" = \
  $'001_scanner_consolidated_baseline.sql [startup]\n002_remove_mishomed_epss.sql [startup]\n003_tenant_qualify_artifact_boms.sql [startup]'
test "$(wc -l < "${SCN5_ROOT}/parity/migration-ledger.txt")" -eq 3

scn5_assert_candidate_image_ids
set +e
./devops/compose/scripts/compose-cli.sh raw -- run --rm --no-deps \
  --entrypoint dotnet scanner-web /app/StellaOps.Scanner.WebService.dll --migrate-only \
  > "${SCN5_ROOT}/logs/migrate-only-second.txt" 2>&1
SCN5_MIGRATE_SECOND_RC=$?
set -e
sed -n '1,240p' "${SCN5_ROOT}/logs/migrate-only-second.txt"
test "${SCN5_MIGRATE_SECOND_RC}" -eq 0
grep -F "applied=0, already-applied=3" "${SCN5_ROOT}/logs/migrate-only-second.txt"
scn5_assert_candidate_image_ids
scn5_assert_writers_stopped

Verified by: the first command exits 0 and applies exactly two migrations; the ledger contains exactly 001_scanner_consolidated_baseline.sql, 002_remove_mishomed_epss.sql, and 003_tenant_qualify_artifact_boms.sql, all [startup]; the second command exits 0 and applies zero; both candidate tags stayed on the captured image IDs before and after each invocation; and both live Scanner writer containers remain stopped. Exit 0 also proves the in-process target check accepted only database stellaops_scanner and role scanner with no superuser, CREATEDB, CREATEROLE, or BYPASSRLS attribute.

If 003 fails after 002 commits, preserve both logs and keep the freeze. The target is no longer the copy-compatible 001 shape: do not rerun, truncate, hand-edit the ledger, or patch the tenant column. Roll the containers back to the untouched source, then recreate the retry target at exact 001, restart at step 3, and repeat the downstream copy gates.

11. Prove the post-migration contract before any repoint

docker exec "${SCN5_PG_CONTAINER}" psql -XqAt -F '|' -U "${SCN5_PG_OPERATOR}" -d "${SCN5_TARGET_DB}" -c \
  "SELECT
     (SELECT count(*) FROM pg_class c JOIN pg_namespace n ON n.oid=c.relnamespace
       WHERE n.nspname='scanner' AND c.relname LIKE 'epss%'),
     (SELECT count(*) FROM pg_proc p JOIN pg_namespace n ON n.oid=p.pronamespace
       WHERE n.nspname='scanner' AND p.proname LIKE '%epss%'),
     (SELECT count(*) FROM pg_class c JOIN pg_namespace n ON n.oid=c.relnamespace
       WHERE n.nspname='scanner' AND c.relkind IN ('r','p')
         AND NOT c.relispartition AND c.relname <> 'schema_migrations');" \
  > "${SCN5_ROOT}/parity/post-migration-census.txt"
sed -n '1,20p' "${SCN5_ROOT}/parity/post-migration-census.txt"
grep -Fx '0|0|82' "${SCN5_ROOT}/parity/post-migration-census.txt"
test "$(wc -l < "${SCN5_ROOT}/parity/post-migration-census.txt")" -eq 1

docker exec "${SCN5_PG_CONTAINER}" psql -XqAt -F '|' -U "${SCN5_PG_OPERATOR}" -d "${SCN5_TARGET_DB}" -c \
  "WITH exact AS (
       SELECT artifact.build_id, artifact.inserted_at
         FROM scanner.artifact_boms artifact
         JOIN scanner.scan_runtime_state runtime
           ON runtime.scan_id = artifact.build_id
          AND btrim(runtime.tenant_id) = artifact.tenant_id
   ), sole AS (
       SELECT min(btrim(tenant_id)) AS tenant_id
         FROM scanner.scan_runtime_state
        WHERE tenant_id IS NOT NULL AND btrim(tenant_id) <> ''
       HAVING count(DISTINCT btrim(tenant_id)) = 1
   )
   SELECT count(*),
          count(*) FILTER (WHERE artifact.tenant_id IS NULL OR btrim(artifact.tenant_id) = ''),
          count(*) FILTER (WHERE exact.build_id IS NOT NULL),
          count(*) FILTER (WHERE exact.build_id IS NULL AND artifact.tenant_id = sole.tenant_id)
     FROM scanner.artifact_boms artifact
     LEFT JOIN exact USING (build_id, inserted_at)
     CROSS JOIN sole;" \
  > "${SCN5_ROOT}/parity/post-migration-tenant-attribution.txt"
sed -n '1,20p' "${SCN5_ROOT}/parity/post-migration-tenant-attribution.txt"
grep -Fx '8|0|7|1' "${SCN5_ROOT}/parity/post-migration-tenant-attribution.txt"
test "$(wc -l < "${SCN5_ROOT}/parity/post-migration-tenant-attribution.txt")" -eq 1

docker exec "${SCN5_PG_CONTAINER}" psql -XqAt -F '|' -U "${SCN5_PG_OPERATOR}" -d "${SCN5_TARGET_DB}" -c \
  "SELECT c.is_nullable, coalesce(c.column_default, '<null>'),
          (SELECT pg_get_constraintdef(oid) FROM pg_constraint
            WHERE conrelid='scanner.artifact_boms'::regclass AND contype='p')
     FROM information_schema.columns c
    WHERE c.table_schema='scanner' AND c.table_name='artifact_boms' AND c.column_name='tenant_id';" \
  > "${SCN5_ROOT}/parity/post-migration-tenant-contract.txt"
sed -n '1,20p' "${SCN5_ROOT}/parity/post-migration-tenant-contract.txt"
grep -Fx 'NO|<null>|PRIMARY KEY (tenant_id, build_id, inserted_at)' \
  "${SCN5_ROOT}/parity/post-migration-tenant-contract.txt"
test "$(wc -l < "${SCN5_ROOT}/parity/post-migration-tenant-contract.txt")" -eq 1

scn5_fingerprint "${SCN5_TARGET_DB}" "${SCN5_ROOT}/parity/target-post-migration.tsv"
scn5_assert_copy_fingerprint "${SCN5_ROOT}/parity/target-post-migration.tsv"
grep -v '^artifact_boms_' "${SCN5_ROOT}/parity/target-after.tsv" \
  > "${SCN5_ROOT}/parity/target-before-without-artifact-boms.tsv"
grep -v '^artifact_boms_' "${SCN5_ROOT}/parity/target-post-migration.tsv" \
  > "${SCN5_ROOT}/parity/target-after-without-artifact-boms.tsv"
cmp "${SCN5_ROOT}/parity/target-before-without-artifact-boms.tsv" \
    "${SCN5_ROOT}/parity/target-after-without-artifact-boms.tsv"

docker exec "${SCN5_PG_CONTAINER}" psql -XqAt -F $'\t' -U "${SCN5_PG_OPERATOR}" -d "${SCN5_TARGET_DB}" -c \
  "SET TimeZone='UTC'; SET extra_float_digits=3;
   SELECT count(*)::text,
          coalesce(md5(string_agg(md5(ROW(build_id, canonical_bom_sha256, payload_digest, inserted_at)::text), '' ORDER BY build_id, inserted_at)), '<empty>')
     FROM scanner.artifact_boms;" \
  > "${SCN5_ROOT}/parity/target-artifact-boms-old-shape.tsv"
cmp "${SCN5_ROOT}/parity/source-artifact-boms-old-shape.tsv" \
    "${SCN5_ROOT}/parity/target-artifact-boms-old-shape.tsv"

scn5_fingerprint "${SCN5_SOURCE_DB}" "${SCN5_ROOT}/parity/source-before-repoint.tsv"
scn5_assert_copy_fingerprint "${SCN5_ROOT}/parity/source-before-repoint.tsv"
cmp "${SCN5_ROOT}/parity/source-frozen.tsv" "${SCN5_ROOT}/parity/source-before-repoint.tsv"
scn5_assert_writers_stopped

Verified by: executable one-line assertions require the census 0|0|82, tenant result 8|0|7|1, and tenant contract NO|<null>|PRIMARY KEY (tenant_id, build_id, inserted_at); the full target copy fingerprint still has 84 non-empty digest fields and the expected 14|135 non-empty summary; the non-artifact copy fingerprints are unchanged; artifact_boms has identical count and old-column digest when tenant_id is ignored; the frozen source is still byte-identical; and both writers are still stopped. Re-run the sequence and identity gate from step 9; it must remain green after both migrations.

12. Repoint onto the consolidated overlay

devops/compose/docker-compose.stella-services.yml now carries the canonical variable for both Scanner hosts and fails closed when it is absent. That durable base change, the removal of both resolver fallbacks, and their conformance test must be committed and green before this step. A later ordinary recreate must not be able to return Scanner to stellaops_platform.

Use the canonical Compose helper (H6), not either container’s stale label chain. Use only the exact candidate image set built in step 10. Keep docker-compose.existing-networks.override.yml last so Compose never tries to adopt or recreate live-estate networks.

export COMPOSE_EXTRA_FILES="${SCN5_CANDIDATE_OVERLAY}"
scn5_assert_candidate_image_ids
scn5_render_topology "${SCN5_ROOT}/topology/rendered-names-only.json"
jq -e --arg web "${SCN5_WEB_CANDIDATE_IMAGE}" \
      --arg worker "${SCN5_WORKER_CANDIDATE_IMAGE}" \
      --arg source "${SCN5_BUNDLE_TARGET_BASE}" '
  .services["scanner-web"].image == $web
  and .services["scanner-worker"].image == $worker
  and .services["scanner-web"].scannerBaseSource == $source
  and .services["scanner-worker"].scannerBaseSource == $source
' "${SCN5_ROOT}/topology/rendered-names-only.json" >/dev/null

scn5_prepare_candidate_env_contract() { # $1=container, $2=service, $3=image, $4=image names
  local container="$1" service="$2" image="$3" image_names="$4"
  local expected_names compose_names rendered_names expected_hmac unchanged_hmac
  expected_names="${SCN5_ROOT}/topology/${container}.candidate-expected-env-names.txt"
  compose_names="${SCN5_ROOT}/topology/${container}.candidate-compose-env-names.txt"
  rendered_names="${SCN5_ROOT}/topology/${container}.candidate-rendered-env-names.txt"
  expected_hmac="${SCN5_ROOT}/topology/${container}.candidate-env.hmac"
  unchanged_hmac="${SCN5_ROOT}/topology/${container}.candidate-unchanged-env.hmac"

  awk '$0 !~ /^(STELLAOPS_POSTGRES_SCANNER_CONNECTION|SCANNER_SCANNER__STORAGE__DSN|ScannerStorage__Postgres__ConnectionString|SCANNER_SCANNER__CUTOVER__INTAKEWINDOWID)$/' \
    "${SCN5_ROOT}/topology/${container}.env-names.txt" > "${expected_names}"
  printf '%s\n' STELLAOPS_POSTGRES_SCANNER_CONNECTION >> "${expected_names}"
  if test "${service}" = scanner-web; then
    printf '%s\n' SCANNER_SCANNER__CUTOVER__INTAKEWINDOWID >> "${expected_names}"
  fi
  LC_ALL=C sort -u -o "${expected_names}" "${expected_names}"

  jq -r --arg service "${service}" '.services[$service].environmentKeys[]' \
    "${SCN5_ROOT}/topology/rendered-names-only.json" \
    | LC_ALL=C sort -u > "${compose_names}"
  LC_ALL=C sort -u "${image_names}" "${compose_names}" > "${rendered_names}"
  cmp "${expected_names}" "${rendered_names}"
  scn5_capture_rendered_env_hmac "${service}" "${image}" "${expected_hmac}"
  scn5_capture_rendered_env_hmac "${service}" "${image}" "${unchanged_hmac}" \
    "${SCN5_ENV_DELTA_EXCLUDE}"
  cmp "${SCN5_ROOT}/topology/${container}.unchanged-env.hmac" "${unchanged_hmac}"
}

scn5_prepare_candidate_env_contract \
  stellaops-scanner-web scanner-web "${SCN5_WEB_CANDIDATE_IMAGE}" \
  "${SCN5_ROOT}/topology/scanner-web.candidate-image-env-names.txt"
scn5_prepare_candidate_env_contract \
  stellaops-scanner-worker scanner-worker "${SCN5_WORKER_CANDIDATE_IMAGE}" \
  "${SCN5_ROOT}/topology/scanner-worker.candidate-image-env-names.txt"

# Assert the exact intended value changes without writing the rendered values.
# The two connection strings must be identical and must name only Scanner's
# database and role. The full keyed HMAC pins the candidate render, while the
# excluded-name keyed HMAC proves every unrelated outgoing value is unchanged.
set +e
./devops/compose/scripts/compose-cli.sh config --format json \
  | jq -e --arg window "${SCN5_WINDOW_ID}" '
      def scanner_connection:
        type == "string"
        and test("(?i)(^|;)[[:space:]]*database[[:space:]]*=[[:space:]]*stellaops_scanner[[:space:]]*(;|$)")
        and test("(?i)(^|;)[[:space:]]*(username|user[[:space:]]*id)[[:space:]]*=[[:space:]]*scanner[[:space:]]*(;|$)");
      .services["scanner-web"].environment as $web
      | .services["scanner-worker"].environment as $worker
      | ($web.STELLAOPS_POSTGRES_SCANNER_CONNECTION | scanner_connection)
        and ($worker.STELLAOPS_POSTGRES_SCANNER_CONNECTION | scanner_connection)
        and $web.STELLAOPS_POSTGRES_SCANNER_CONNECTION
          == $worker.STELLAOPS_POSTGRES_SCANNER_CONNECTION
        and $web.SCANNER_SCANNER__CUTOVER__INTAKEWINDOWID == $window
        and (($worker | has("SCANNER_SCANNER__CUTOVER__INTAKEWINDOWID")) | not)
        and (($web | has("SCANNER_SCANNER__STORAGE__DSN")) | not)
        and (($web | has("ScannerStorage__Postgres__ConnectionString")) | not)
        and (($worker | has("SCANNER_SCANNER__STORAGE__DSN")) | not)
        and (($worker | has("ScannerStorage__Postgres__ConnectionString")) | not)
        and $worker.ScannerStorage__Postgres__SchemaName == "scanner"
        and (($web | has("VulnSurfaces__FeedTrigger__ScannerConnectionString")) | not)
        and (($worker | has("VulnSurfaces__FeedTrigger__ScannerConnectionString")) | not)
    ' >/dev/null
SCN5_CANDIDATE_VALUE_RENDER_RC=("${PIPESTATUS[@]}")
set -e
test "${#SCN5_CANDIDATE_VALUE_RENDER_RC[@]}" -eq 2
test "${SCN5_CANDIDATE_VALUE_RENDER_RC[0]}" -eq 0
test "${SCN5_CANDIDATE_VALUE_RENDER_RC[1]}" -eq 0

scn5_assert_live_candidate_contract() { # $1=container, $2=image ID, $3=expected names, $4=expected HMAC, $5=label
  local container="$1" image_id="$2" expected_names="$3" expected_hmac="$4" label="$5"
  local actual_names actual_hmac actual_topology expected_topology actual_bundle_source
  actual_names="${SCN5_ROOT}/topology/${container}.${label}-env-names.txt"
  actual_hmac="${SCN5_ROOT}/topology/${container}.${label}-env.hmac"
  actual_topology="${SCN5_ROOT}/topology/${container}.${label}-runtime-topology.json"

  scn5_assert_candidate_image_ids
  test "$(docker inspect "${container}" --format '{{.Image}}')" = "${image_id}"
  scn5_capture_env_names "${container}" "${actual_names}"
  cmp "${expected_names}" "${actual_names}"
  scn5_capture_env_hmac "${container}" "${actual_hmac}"
  cmp "${expected_hmac}" "${actual_hmac}"
  scn5_capture_runtime_topology "${container}" "${actual_topology}"
  actual_bundle_source="$(scn5_bundle_mount_source "${actual_topology}")"
  test "$(cd -- "${actual_bundle_source}" && pwd -P)" = \
    "$(cd -- "${SCN5_BUNDLE_TARGET_BASE}" && pwd -P)"
  expected_topology="${SCN5_ROOT}/topology/${container}.${label}-expected-runtime-topology.json"
  jq --arg source "${actual_bundle_source}" '
    .mounts |= map(if .destination == "/app/plugins/scanner/base"
      then .source = $source else . end)' \
    "${SCN5_ROOT}/topology/${container}.runtime-topology.json" > "${expected_topology}"
  cmp "${expected_topology}" "${actual_topology}"
}

scn5_wait_container_healthy() { # $1=container; bounded to 120 seconds
  local container="$1" attempt state inspect_rc
  for ((attempt = 1; attempt <= 60; attempt++)); do
    if state="$(docker inspect "${container}" \
        --format '{{if .State.Health}}{{.State.Health.Status}}{{else}}{{.State.Status}}{{end}}')"; then
      inspect_rc=0
    else
      inspect_rc=$?
    fi
    test "${inspect_rc}" -eq 0 || return "${inspect_rc}"
    case "${state}" in
      healthy) return 0 ;;
      unhealthy|exited|dead)
        printf '%s entered terminal state %s while waiting for health\n' \
          "${container}" "${state}" >&2
        return 1
        ;;
    esac
    sleep 2
  done
  printf '%s did not become healthy within 120 seconds\n' "${container}" >&2
  return 1
}

Verified by: the executable names-only comparisons report only the intended delta:

• ADDED STELLAOPS_POSTGRES_SCANNER_CONNECTION on web and worker;
• ADDED on web only SCANNER_SCANNER__CUTOVER__INTAKEWINDOWID, equal to the current non-secret SCN5_WINDOW_ID; this keeps every ordinary submission outside the owner-controlled forcing call;
• DROPPED SCANNER_SCANNER__STORAGE__DSN (web) and ScannerStorage__Postgres__ConnectionString (worker) — this deletion is mandatory, not cosmetic (H5);
• KEPT ScannerStorage__Postgres__SchemaName=scanner;
• ABSENT throughout VulnSurfaces__FeedTrigger__ScannerConnectionString.

Image-supplied and Compose environment names are merged with Compose override semantics. The keyed HMAC is over the complete sorted candidate environment after the same merge, so every value — not a secret denylist — is covered without persisting it. The render assertion also proves the canonical connection names stellaops_scanner as role scanner, both hosts receive the same DSN, and the web fence is the exact window ID. Any other delta is behaviour smuggled into a database move: stop and remove it. CanonicalConnectionKey_IsTheOnlyLiveComposeAndResolverSource is the durable pin.

Verify the promoted candidate bundles while both hosts remain stopped

The stopped-writer gate above promoted and committed the inventories, then produced the exact-SHA images and complete ignored payload from a fresh detached build. Recheck each bundle’s manifest- derived contract matrix immediately before recreate. Worker admission remains host-bound. Web is not an analyzer host and therefore carries no byte-equality requirement against the mounted bundle contracts.

scn5_verify_candidate_host_contracts() {
  local contract host_path host_sha image_sha

  scn5_verify_snapshot_contract_matrix \
    "${SCN5_BUNDLE_TARGET_BASE}" candidate-base \
    "${SCN5_WORKER_CANDIDATE_IMAGE}"
  scn5_verify_snapshot_contract_matrix \
    "${SCN5_BUNDLE_TARGET_FULL}" candidate-full \
    "${SCN5_WORKER_CANDIDATE_IMAGE}"

  for contract in \
      StellaOps.Scanner.Analyzers.Lang.dll \
      StellaOps.Scanner.Analyzers.LangAdapter.dll \
      StellaOps.Scanner.Analyzers.OS.dll \
      StellaOps.Scanner.Plugin.Contracts.dll; do
    host_path="${SCN5_BUNDLE_SOURCE_HOST}/${contract}"
    test -f "${host_path}"
    host_sha="$(sha256sum "${host_path}" | awk '{print $1}')"
    [[ "${host_sha}" =~ ^[0-9a-f]{64}$ ]]
    image_sha="$(awk '{print $1}' \
      "${SCN5_ROOT}/topology/candidate-base-image-${contract}.sha256")"
    test "${image_sha}" = "${host_sha}"
  done
}

scn5_assert_bundle_promotion_gate() {
  scn5_assert_writers_stopped
  test "$(git rev-parse HEAD)" = "${SCN5_SOURCE_SHA}"
  git diff --quiet --exit-code -- \
    devops/plugins/scanner/base \
    devops/plugins/scanner/scanner-full-analyzers
  scn5_verify_bundle_root \
    "${SCN5_BUNDLE_TARGET_BASE}" "${SCN5_BUNDLE_EXPECTED_BASE}" \
    "${SCN5_ROOT}/topology/recreate-gate-scanner-base"
  scn5_verify_bundle_root \
    "${SCN5_BUNDLE_TARGET_FULL}" "${SCN5_BUNDLE_EXPECTED_FULL}" \
    "${SCN5_ROOT}/topology/recreate-gate-scanner-full"
  diff -qr "${SCN5_BUNDLE_SOURCE_BASE}" "${SCN5_BUNDLE_TARGET_BASE}" \
    > "${SCN5_ROOT}/topology/recreate-source-vs-base.diff"
  diff -qr "${SCN5_BUNDLE_SOURCE_FULL}" "${SCN5_BUNDLE_TARGET_FULL}" \
    > "${SCN5_ROOT}/topology/recreate-source-vs-full.diff"
  scn5_verify_candidate_host_contracts
  scn5_assert_candidate_image_ids
}

scn5_assert_bundle_promotion_gate

Verified by: primary main receives one exact-path inventory commit only after both containers are stopped; exact-path staging is the immediately preceding invocation, and the complete unscoped index and unchanged parent are asserted in the commit invocation; Web and Worker’s distinct outgoing roots are checksum-valid and byte-identical to their own immutable snapshots; the complete source, staged, and durable candidate roots each contain exactly 8 base and 14 full bundles; every checksums.sha256 entry passes; the durable trees are byte-identical to the detached build output; the final exact-commit rebuild retains the same real latest-src/ assembly revision and has zero canonical tracked diff (raw Windows status is diagnostic, not the content authority); each manifest kind requires exactly its defined language or OS contract set with no extra stamp or copy, every required copy matches its stamp and the worker publish output plus /app in the exact worker image. Web is not an analyzer host and is deliberately excluded from that host-contract equality gate. Any failed assertion leaves both Scanner hosts fenced and prevents recreate. Keep the detached build worktree until the post-repoint forcing function is green; it is evidence and the source of the ignored payload, not a runtime mount.

Then recreate by Compose service key. The helper supplies --no-deps --force-recreate; do not use container names and do not pipe the command because a pipe can hide the Compose exit code.

scn5_assert_bundle_promotion_gate
scn5_assert_candidate_image_ids
set +e
./devops/compose/scripts/compose-cli.sh recreate scanner-web scanner-worker \
  > "${SCN5_ROOT}/logs/recreate.txt" 2>&1
SCN5_RECREATE_RC=$?
set -e
sed -n '1,240p' "${SCN5_ROOT}/logs/recreate.txt"
test "${SCN5_RECREATE_RC}" -eq 0
scn5_assert_candidate_image_ids
scn5_wait_container_healthy stellaops-scanner-web
scn5_wait_container_healthy stellaops-scanner-worker

Verified by: both containers reach healthy, and each resolves the owner connection:

SCN5_EXPECTED_CONNECTION_KEYS="${SCN5_ROOT}/topology/expected-scanner-connection-keys.txt"
printf '%s\n' STELLAOPS_POSTGRES_SCANNER_CONNECTION > "${SCN5_EXPECTED_CONNECTION_KEYS}"
for c in stellaops-scanner-web stellaops-scanner-worker; do
  test "$(docker inspect "$c" --format '{{.State.Status}} {{.State.Health.Status}}')" = \
    "running healthy"
  scn5_capture_env_names "$c" "${SCN5_ROOT}/topology/${c}.post-repoint-env-names.txt"
  grep -E 'STELLAOPS_POSTGRES_SCANNER_CONNECTION|SCANNER_SCANNER__STORAGE__DSN|ScannerStorage__Postgres__ConnectionString' \
    "${SCN5_ROOT}/topology/${c}.post-repoint-env-names.txt" \
    | LC_ALL=C sort -u > "${SCN5_ROOT}/topology/${c}.connection-keys.after.txt"
  cmp "${SCN5_EXPECTED_CONNECTION_KEYS}" \
      "${SCN5_ROOT}/topology/${c}.connection-keys.after.txt"
done

scn5_assert_live_candidate_contract \
  stellaops-scanner-web "${SCN5_WEB_CANDIDATE_IMAGE_ID}" \
  "${SCN5_ROOT}/topology/stellaops-scanner-web.candidate-expected-env-names.txt" \
  "${SCN5_ROOT}/topology/stellaops-scanner-web.candidate-env.hmac" candidate
scn5_assert_live_candidate_contract \
  stellaops-scanner-worker "${SCN5_WORKER_CANDIDATE_IMAGE_ID}" \
  "${SCN5_ROOT}/topology/stellaops-scanner-worker.candidate-expected-env-names.txt" \
  "${SCN5_ROOT}/topology/stellaops-scanner-worker.candidate-env.hmac" candidate

scn5_assert_live_logs() { # $1=receipt label; logs only, never environment values
  local label="$1" container receipt
  for container in \
      stellaops-authority stellaops-policy-engine \
      stellaops-scanner-web stellaops-scanner-worker; do
    receipt="${SCN5_ROOT}/logs/${label}-${container}.log"
    docker logs --timestamps --since "${SCN5_LIVE_RECREATE_SINCE}" --tail 2000 \
      "${container}" > "${receipt}" 2>&1
    sed -n '1,240p' "${receipt}"
    if grep -Ein \
      '(^|[[:space:]\[])(fatal|critical)([[:space:]\]:]|$)|unhandled exception|host terminated unexpectedly|startup migration.*(fail(ed|ure)?|error)|plugin[- ]?(load|initializ)[^[:cntrl:]]*(fail(ed|ure)?|error)' \
      "${receipt}"; then
      return 1
    fi
  done
}
scn5_assert_live_logs post-recreate

The canonical variable must be the only connection-key match on each. The complete live environment names and keyed HMAC must equal the expected candidate render; a second keyed HMAC, excluding only the four named connection/fence keys, must equal the outgoing live receipt so unrelated values cannot drift under cover of the repoint. The stable runtime-topology projection must remain byte-identical to the pre-window receipt, both containers must use the captured candidate image IDs, and bounded logs from all four recreated services must contain none of the asserted fatal, unhandled, startup-migration, or plugin-load failure patterns. doctor.scanner.database.own-database-cutover (src/Scanner/__Libraries/StellaOps.Scanner.Operations/Doctor/ScannerDoctorChecks.cs:364) reports which source answered the resolver — a host sitting on stellaops_scanner but still resolving through a transitional key is a half-done repoint that every functional probe calls healthy. Run it through Scanner’s deterministic loopback-only publish in step 13 and record its machine-asserted verdict before submitting the forcing scan. Do not use the aggregate /doctor/scanner-web/checks path: multiple services publish that path and the gateway does not select a service by name when paths collide.

13. Live verification — the forcing function

Container-up is not the gate. The gate is that one real, digest-pinned scan persists in the Scanner-owned database, reads back through Scanner’s owner API, emits the tenant event, and keeps the old source byte-for-byte frozen. Mint two exact-scope, session-only client-credentials bearers: scanner:scan submits the controlled scan; scanner:read polls and reads the evidence surfaces. Bearers and the client secret are consumed through stdin or shell memory only — never printed, passed as an operating-system process argument, or written to a receipt. The Windows operator host must use the repository CA bundle; -k is forbidden.

Before the fence, use apply_patch to create tmp/scn5-<window-id>/topology/policy-owner-outage.override.yml with this exact LF-terminated template:

services:
  scanner-web:
    environment:
      SCANNER_SCANNER__POLICYOWNER__BASEADDRESS: "http://127.0.0.1:9"
      SCANNER_SCANNER__POLICYOWNER__REQUESTTIMEOUTSECONDS: "2"

scn5_assert_live_logs pre-forcing-function
set +x # tokens and both client secrets must remain silent even if xtrace was inherited
: "${STELLAOPS_BACKEND_URL:?Set the accepted live backend URL}"
: "${STELLAOPS_RELEASE_DISPATCH_CLIENT_SECRET:?Load by name from the ignored operator secret source}"
: "${SCN5_SCAN_IMAGE:?Set SCN5_SCAN_IMAGE to a reachable, previously unscanned digest-pinned registry reference}"
: "${SCN5_WINDOW_EPOCH_MS:?Step 1 must have captured SCN5_WINDOW_EPOCH_MS}"
[[ "${SCN5_WINDOW_EPOCH_MS}" =~ ^[0-9]{13}$ ]]
[[ "${SCN5_SCAN_IMAGE}" =~ ^[^@[:space:]]+@sha256:[0-9a-f]{64}$ ]]
SCN5_SCAN_DIGEST="${SCN5_SCAN_IMAGE##*@}"
[[ "${SCN5_SCAN_DIGEST}" =~ ^sha256:[0-9a-f]{64}$ ]]
test -s devops/compose/combined-ca-bundle.crt
SCN5_CURL=(
  env MSYS2_ARG_CONV_EXCL='*' curl.exe --silent --show-error --ssl-no-revoke
  --cacert "${SCN5_REPO_ROOT}/devops/compose/combined-ca-bundle.crt"
  --noproxy '*'
  --connect-timeout 5 --max-time 30
)

scn5_mint_token() { # $1 = one exact scope; token is returned only in shell memory
  local scope="$1" raw body http_code token mint_rc
  set +e
  raw="$(
    printf '%s' "${STELLAOPS_RELEASE_DISPATCH_CLIENT_SECRET}" \
      | "${SCN5_CURL[@]}" --request POST \
          --header 'Content-Type: application/x-www-form-urlencoded' \
          --data-urlencode 'grant_type=client_credentials' \
          --data-urlencode 'client_id=stellaops-release-dispatch' \
          --data-urlencode "scope=${scope}" \
          --data-urlencode "tenant=${SCN5_TENANT}" \
          --data-urlencode 'client_secret@-' \
          --write-out $'\n%{http_code}' \
          "${STELLAOPS_BACKEND_URL%/}/connect/token"
  )"
  mint_rc=$?
  set -e
  test "${mint_rc}" -eq 0
  http_code="${raw##*$'\n'}"
  body="${raw%$'\n'*}"
  test "${http_code}" = 200
  token="$(printf '%s' "${body}" | jq -er \
    'select(.token_type == "Bearer") | .access_token | select(type == "string" and length > 20)')"
  printf '%s' "${token}"
}

SCN5_SCAN_TOKEN="$(scn5_mint_token scanner:scan)"
SCN5_READ_TOKEN="$(scn5_mint_token scanner:read)"
export SCN5_SCAN_TOKEN SCN5_READ_TOKEN SCN5_TENANT

# Verify both bearers without echoing or persisting either. Scanner itself obtains
# policy:preview:invoke on the separate Policy-owner hop.
python - <<'PY'
import base64
import json
import os

for variable, expected_scope in (
    ("SCN5_SCAN_TOKEN", "scanner:scan"),
    ("SCN5_READ_TOKEN", "scanner:read"),
):
    token = os.environ[variable]
    parts = token.split(".")
    if len(parts) < 2:
        raise SystemExit(f"{variable} is not a JWT")

    encoded = parts[1] + ("=" * (-len(parts[1]) % 4))
    claims = json.loads(base64.urlsafe_b64decode(encoded))
    scope_claim = claims.get("scope", "")
    scopes = set(scope_claim if isinstance(scope_claim, list) else scope_claim.split())
    aud_claim = claims.get("aud", [])
    audiences = set(aud_claim if isinstance(aud_claim, list) else [aud_claim])

    if claims.get("stellaops:tenant") != os.environ["SCN5_TENANT"]:
        raise SystemExit(f"{variable} has the wrong tenant")
    if scopes != {expected_scope}:
        raise SystemExit(f"{variable} does not carry exactly {expected_scope}")
    if "stellaops" not in audiences:
        raise SystemExit(f"{variable} lacks the deployed stellaops audience")

print("SCN5 tokens: exact tenant, exact split scopes, stellaops audience")
PY
export -n SCN5_SCAN_TOKEN SCN5_READ_TOKEN

# RAR-5 owns the external prefix. Discover every Scanner path from the running
# gateway aggregate instead of assuming a committed prefix.
set +e
"${SCN5_CURL[@]}" --fail-with-body \
  "${STELLAOPS_BACKEND_URL%/}/api/openapi/aggregate" \
  --output "${SCN5_ROOT}/topology/live-gateway-openapi-aggregate.json"
SCN5_OPENAPI_RC=$?
set -e
test "${SCN5_OPENAPI_RC}" -eq 0
jq -e '
  [.services[]? | select(.service == "scanner" and .status == "ok")]
  | length == 1 and (.[0].spec | type) == "object"
' "${SCN5_ROOT}/topology/live-gateway-openapi-aggregate.json" >/dev/null
jq -e '
  .services[] | select(.service == "scanner" and .status == "ok") | .spec
' "${SCN5_ROOT}/topology/live-gateway-openapi-aggregate.json" \
  > "${SCN5_ROOT}/topology/live-scanner-openapi.json"

scn5_unique_openapi_path() {
  local suffix="$1"
  jq -er --arg suffix "${suffix}" '
    [.paths | keys[] | select(endswith($suffix))]
    | if length == 1 then .[0]
      else error("expected exactly one live route ending in " + $suffix)
      end
  ' "${SCN5_ROOT}/topology/live-scanner-openapi.json"
}

SCN5_SUBMIT_PATH="$(MSYS2_ARG_CONV_EXCL='*' scn5_unique_openapi_path '/api/v1/scans')"
SCN5_STATUS_PATH_TEMPLATE="$(MSYS2_ARG_CONV_EXCL='*' scn5_unique_openapi_path '/api/v1/scans/{scanId}')"
SCN5_RUNTIME_PATH="$(MSYS2_ARG_CONV_EXCL='*' scn5_unique_openapi_path '/api/v1/policy/runtime')"
SCN5_REPORT_PATH="$(MSYS2_ARG_CONV_EXCL='*' scn5_unique_openapi_path '/api/v1/reports')"
SCN5_HOT_LOOKUP_PATH_TEMPLATE="$(MSYS2_ARG_CONV_EXCL='*' scn5_unique_openapi_path '/api/v1/sbom/hot-lookup/payload/{payloadDigest}/latest')"
SCN5_DETAIL_PATH_TEMPLATE="$(MSYS2_ARG_CONV_EXCL='*' scn5_unique_openapi_path '/api/v1/scans/by-digest/{imageDigest}/detail')"

jq -e \
  --arg submit "${SCN5_SUBMIT_PATH}" \
  --arg status "${SCN5_STATUS_PATH_TEMPLATE}" \
  --arg runtime "${SCN5_RUNTIME_PATH}" \
  --arg report "${SCN5_REPORT_PATH}" \
  --arg hotLookup "${SCN5_HOT_LOOKUP_PATH_TEMPLATE}" \
  --arg detail "${SCN5_DETAIL_PATH_TEMPLATE}" '
    .paths[$submit].post != null
    and .paths[$status].get != null
    and .paths[$runtime].post != null
    and .paths[$report].post != null
    and .paths[$hotLookup].get != null
    and .paths[$detail].get != null
  ' "${SCN5_ROOT}/topology/live-scanner-openapi.json" >/dev/null

SCN5_SUBMIT_URL="${STELLAOPS_BACKEND_URL%/}${SCN5_SUBMIT_PATH}"
SCN5_RUNTIME_URL="${STELLAOPS_BACKEND_URL%/}${SCN5_RUNTIME_PATH}"
SCN5_REPORT_URL="${STELLAOPS_BACKEND_URL%/}${SCN5_REPORT_PATH}"

# `/doctor/scanner-web/checks` is intentionally common to service hosts. The aggregate router
# resolves colliding paths by connection health/recency rather than by service name,
# so it cannot prove that Scanner answered. Derive Scanner's single loopback-only
# compose publish instead. The direct path is inside the configured Authority bypass
# network. Source/config conformance proves the host's `ops.health` policy; this
# request proves the Scanner process and database boundary deterministically.
SCN5_DOCTOR_BINDING="$(
  docker inspect stellaops-scanner-web \
    | jq -er '
      [.[0].NetworkSettings.Ports["80/tcp"][]?
        | select(.HostIp | test("^127[.][0-9]+[.][0-9]+[.][0-9]+$"))]
      | if length == 1 and (.[0].HostPort | test("^[0-9]+$"))
        then .[0].HostIp + ":" + .[0].HostPort
        else error("expected one loopback-only Scanner Web 80/tcp publish")
        end'
)"
[[ "${SCN5_DOCTOR_BINDING}" =~ ^127\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}:[0-9]+$ ]]
SCN5_DOCTOR_URL="http://${SCN5_DOCTOR_BINDING}/doctor/scanner-web/checks"

scn5_curl_bearer() { # $1 = token, remaining args are curl args; bearer enters curl on stdin
  local token="$1" had_errexit=0 result=0
  local -a pipeline_rc
  shift
  [[ $- == *e* ]] && had_errexit=1
  set +e
  printf 'Authorization: Bearer %s\n' "${token}" \
    | "${SCN5_CURL[@]}" --header @- "$@"
  pipeline_rc=("${PIPESTATUS[@]}")
  test "${#pipeline_rc[@]}" -eq 2 || result=1
  test "${pipeline_rc[0]}" -eq 0 || result=1
  test "${pipeline_rc[1]}" -eq 0 || result=1
  test "${had_errexit}" -eq 0 || set -e
  return "${result}"
}

# Doctor's exact host route is mapped by ServiceDoctorHosting.cs as GET
# /doctor/scanner-web/checks. Use the derived loopback-only Scanner publish so an identically
# named route from another healthy service cannot satisfy this cutover gate.
SCN5_DOCTOR_HTTP="$(
  "${SCN5_CURL[@]}" \
    --header "X-StellaOps-TenantId: ${SCN5_TENANT}" \
    --header 'Accept: application/json' \
    --output "${SCN5_ROOT}/logs/scanner-doctor-checks.json" \
    --write-out '%{http_code}' \
    "${SCN5_DOCTOR_URL}"
)"
test "${SCN5_DOCTOR_HTTP}" = 200
jq -e '
  [.checks[]? | select(.checkId == "doctor.scanner.database.own-database-cutover")] as $cutover
  | .service == "scanner-web"
    and .contractVersion == "doctor-check/v1"
    and ($cutover | length) == 1
    and ($cutover[0]
      | .category == "config"
        and .severity == "info"
        and .healthy == true
        and .evidence["config.canonical-variable"] == "STELLAOPS_POSTGRES_SCANNER_CONNECTION"
        and .evidence["config.host-plane.source"] == "STELLAOPS_POSTGRES_SCANNER_CONNECTION"
        and .evidence["config.host-plane.canonical"] == "yes"
        and .evidence["config.role"] == "scanner-web"
        and .evidence["database.expected-after-window"] == "stellaops_scanner"
        and .evidence["database.host-plane.declared"] == "stellaops_scanner"
        and .evidence["database.connected"] == "stellaops_scanner"
        and .evidence["ledger.rows"] == "3"
        and .evidence["ledger.lineage"] == "consolidated"
        and .evidence["ledger.owned-rows-present"] == "3"
        and .evidence["ledger.legacy-rows-present"] == "0"
        and .evidence["ledger.unrecognised"] == "none")
' "${SCN5_ROOT}/logs/scanner-doctor-checks.json" >/dev/null

jq -n --arg reference "${SCN5_SCAN_IMAGE}" --arg digest "${SCN5_SCAN_DIGEST}" '
  {image: {reference: $reference, digest: $digest}, force: true}
' > "${SCN5_ROOT}/logs/intake-fence-missing-request.json"
jq -n --arg reference "${SCN5_SCAN_IMAGE}" --arg digest "${SCN5_SCAN_DIGEST}" \
  --arg marker "${SCN5_WINDOW_ID}-wrong" '
  {
    image: {reference: $reference, digest: $digest},
    force: true,
    metadata: {"stellaops.cutover.window": $marker}
  }
' > "${SCN5_ROOT}/logs/intake-fence-wrong-request.json"
jq -n --arg reference "${SCN5_SCAN_IMAGE}" --arg digest "${SCN5_SCAN_DIGEST}" \
  --arg marker "${SCN5_WINDOW_ID}" '
  {
    image: {reference: $reference, digest: $digest},
    force: true,
    metadata: {"stellaops.cutover.window": $marker}
  }
' > "${SCN5_ROOT}/logs/forcing-scan-request.json"

SCN5_NEGATIVE_DB_BEFORE="$(docker exec "${SCN5_PG_CONTAINER}" psql -XqAt \
  -U "${SCN5_PG_OPERATOR}" -d "${SCN5_TARGET_DB}" -v digest="${SCN5_SCAN_DIGEST}" -c \
  "SELECT count(*) FROM scanner.scan_runtime_state
    WHERE lower(target_digest) = lower(:'digest');")"
SCN5_NEGATIVE_QUEUE_BEFORE="$(
  docker exec stellaops-valkey valkey-cli XLEN scanner:jobs
)"
[[ "${SCN5_NEGATIVE_DB_BEFORE}" =~ ^[0-9]+$ ]]
[[ "${SCN5_NEGATIVE_QUEUE_BEFORE}" =~ ^[0-9]+$ ]]
test "${SCN5_NEGATIVE_DB_BEFORE}" -eq 0

for negative_case in missing wrong; do
  SCN5_FENCE_HTTP="$(
    scn5_curl_bearer "${SCN5_SCAN_TOKEN}" \
      --request POST \
      --header "X-StellaOps-TenantId: ${SCN5_TENANT}" \
      --header 'Accept: application/problem+json' \
      --header 'Content-Type: application/json' \
      --data-binary "@${SCN5_ROOT}/logs/intake-fence-${negative_case}-request.json" \
      --output "${SCN5_ROOT}/logs/intake-fence-${negative_case}-response.json" \
      --write-out '%{http_code}' \
      "${SCN5_SUBMIT_URL}"
  )"
  test "${SCN5_FENCE_HTTP}" = 503
  jq -e '
    .status == 503
    and .detail == "scan_intake_fenced"
    and (.instance | endswith("/api/v1/scans"))
  ' "${SCN5_ROOT}/logs/intake-fence-${negative_case}-response.json" >/dev/null
  ! grep -F -- "${SCN5_WINDOW_ID}" \
    "${SCN5_ROOT}/logs/intake-fence-${negative_case}-response.json"
done

SCN5_NEGATIVE_DB_AFTER="$(docker exec "${SCN5_PG_CONTAINER}" psql -XqAt \
  -U "${SCN5_PG_OPERATOR}" -d "${SCN5_TARGET_DB}" -v digest="${SCN5_SCAN_DIGEST}" -c \
  "SELECT count(*) FROM scanner.scan_runtime_state
    WHERE lower(target_digest) = lower(:'digest');")"
SCN5_NEGATIVE_QUEUE_AFTER="$(
  docker exec stellaops-valkey valkey-cli XLEN scanner:jobs
)"
test "${SCN5_NEGATIVE_DB_AFTER}" = "${SCN5_NEGATIVE_DB_BEFORE}"
test "${SCN5_NEGATIVE_QUEUE_AFTER}" = "${SCN5_NEGATIVE_QUEUE_BEFORE}"

SCN5_SUBMIT_HTTP="$(
  scn5_curl_bearer "${SCN5_SCAN_TOKEN}" \
    --request POST \
    --header "X-StellaOps-TenantId: ${SCN5_TENANT}" \
    --header 'Accept: application/json' \
    --header 'Content-Type: application/json' \
    --data-binary "@${SCN5_ROOT}/logs/forcing-scan-request.json" \
    --output "${SCN5_ROOT}/logs/forcing-scan-submit.json" \
    --write-out '%{http_code}' \
    "${SCN5_SUBMIT_URL}"
)"
test "${SCN5_SUBMIT_HTTP}" = 202
jq -e '
  (.scanId | type) == "string"
  and (.scanId | length) > 0
  and .created == true
' "${SCN5_ROOT}/logs/forcing-scan-submit.json" >/dev/null
SCN5_SCAN_ID="$(jq -er '.scanId' "${SCN5_ROOT}/logs/forcing-scan-submit.json")"
SCN5_STATUS_PATH="${SCN5_STATUS_PATH_TEMPLATE/\{scanId\}/${SCN5_SCAN_ID}}"
[[ "${SCN5_STATUS_PATH}" != *'{'* ]]
SCN5_STATUS_URL="${STELLAOPS_BACKEND_URL%/}${SCN5_STATUS_PATH}"

SCN5_SCAN_STATUS=unknown
for _ in $(seq 1 180); do
  SCN5_STATUS_HTTP="$(
    scn5_curl_bearer "${SCN5_READ_TOKEN}" \
      --header "X-StellaOps-TenantId: ${SCN5_TENANT}" \
      --header 'Accept: application/json' \
      --output "${SCN5_ROOT}/logs/forcing-scan-status.json" \
      --write-out '%{http_code}' \
      "${SCN5_STATUS_URL}"
  )"
  test "${SCN5_STATUS_HTTP}" = 200
  SCN5_SCAN_STATUS="$(jq -er '.status | ascii_downcase' \
    "${SCN5_ROOT}/logs/forcing-scan-status.json")"
  case "${SCN5_SCAN_STATUS}" in
    succeeded) break ;;
    failed|cancelled) exit 1 ;;
    pending|queued|running|processing) sleep 2 ;;
    *) exit 1 ;;
  esac
done
test "${SCN5_SCAN_STATUS}" = succeeded
jq -e --arg scan "${SCN5_SCAN_ID}" --arg digest "${SCN5_SCAN_DIGEST}" \
  --arg reference "${SCN5_SCAN_IMAGE}" '
  .scanId == $scan
  and (.status | ascii_downcase) == "succeeded"
  and .image.reference == $reference
  and .image.digest == $digest
' "${SCN5_ROOT}/logs/forcing-scan-status.json" >/dev/null

# The source fingerprint must not move, while the exact succeeded row must exist
# in the Scanner-owned target.
scn5_fingerprint "${SCN5_SOURCE_DB}" "${SCN5_ROOT}/parity/source-after-forcing-scan.tsv"
cmp "${SCN5_ROOT}/parity/source-frozen.tsv" \
    "${SCN5_ROOT}/parity/source-after-forcing-scan.tsv"

set +e
docker exec "${SCN5_PG_CONTAINER}" psql -XqAt -v ON_ERROR_STOP=1 -F '|' \
  -U "${SCN5_PG_OPERATOR}" -d "${SCN5_TARGET_DB}" \
  -v scan_id="${SCN5_SCAN_ID}" \
  -v tenant_id="${SCN5_TENANT}" \
  -v digest="${SCN5_SCAN_DIGEST}" -c \
  "SELECT scan_id, tenant_id, lower(status), lower(target_digest)
     FROM scanner.scan_runtime_state
    WHERE scan_id = :'scan_id'
      AND tenant_id = :'tenant_id'
      AND lower(target_digest) = lower(:'digest');" \
  > "${SCN5_ROOT}/parity/forcing-scan-target-row.txt"
SCN5_TARGET_ROW_RC=$?
set -e
test "${SCN5_TARGET_ROW_RC}" -eq 0
grep -Fqx -- \
  "${SCN5_SCAN_ID}|${SCN5_TENANT}|succeeded|${SCN5_SCAN_DIGEST}" \
  "${SCN5_ROOT}/parity/forcing-scan-target-row.txt"

# Read the persisted SBOM identity through Scanner's authenticated owner
# hot-lookup surface. The per-scan export route is a renderer and may have no
# export row even when artifact_boms persistence succeeded; it is not this
# database-cutover forcing function. Use Scanner's derived loopback listener so
# an ingress SPA fallback cannot satisfy the persistence proof.
SCN5_HOT_LOOKUP_PATH="${SCN5_HOT_LOOKUP_PATH_TEMPLATE/\{payloadDigest\}/${SCN5_SCAN_DIGEST}}"
[[ "${SCN5_HOT_LOOKUP_PATH}" != *'{'* ]]
SCN5_HOT_LOOKUP_URL="http://${SCN5_DOCTOR_BINDING}${SCN5_HOT_LOOKUP_PATH}"

set +e
SCN5_HOT_LOOKUP_HTTP="$(
  scn5_curl_bearer "${SCN5_READ_TOKEN}" \
    --header "X-StellaOps-TenantId: ${SCN5_TENANT}" \
    --header 'Accept: application/json' \
    --dump-header "${SCN5_ROOT}/logs/forcing-scan-sbom-hot-lookup.headers.txt" \
    --output "${SCN5_ROOT}/logs/forcing-scan-sbom-hot-lookup.json" \
    --write-out '%{http_code}' \
    "${SCN5_HOT_LOOKUP_URL}"
)"
SCN5_HOT_LOOKUP_RC=$?
set -e
test "${SCN5_HOT_LOOKUP_RC}" -eq 0
test "${SCN5_HOT_LOOKUP_HTTP}" = 200
grep -Eiq \
  '^Content-Type:[[:space:]]*application/json([;[:space:]]|$)' \
  "${SCN5_ROOT}/logs/forcing-scan-sbom-hot-lookup.headers.txt"
jq -e --arg build "${SCN5_SCAN_ID}" --arg digest "${SCN5_SCAN_DIGEST}" '
  .buildId == $build
  and .payloadDigest == $digest
  and (.canonicalBomSha256 | test("^sha256:[0-9a-f]{64}$"))
  and (.insertedAtUtc | type) == "string"
  and (.evidenceScore | type) == "number"
' "${SCN5_ROOT}/logs/forcing-scan-sbom-hot-lookup.json" >/dev/null

# The target owns one v2 completion event on the exact tenant stream. There is
# no separate envelope-hash column: the content proofs are payloadDigest,
# canonicalBomSha256, purlSetSha256, and findingSetSha256 inside payload.
set +e
docker exec "${SCN5_PG_CONTAINER}" psql -XqAt -v ON_ERROR_STOP=1 \
  -U "${SCN5_PG_OPERATOR}" -d "${SCN5_TARGET_DB}" \
  -v tenant_stream="${SCN5_TENANT_STREAM}" \
  -v tenant_id="${SCN5_TENANT}" \
  -v scan_id="${SCN5_SCAN_ID}" \
  -v digest="${SCN5_SCAN_DIGEST}" -c \
  "SELECT jsonb_build_object(
      'stream', o.stream,
      'streamEpoch', o.stream_epoch,
      'seq', o.seq,
      'eventId', o.event_id,
      'type', o.event_type,
      'v', o.v,
      'occurredAt', o.occurred_at,
      'createdAt', o.created_at,
      'payload', o.payload,
      'retiredGlobalRows', (
        SELECT count(*)
          FROM eventing.outbox g
         WHERE g.stream = 'scanner.scans'
           AND g.payload->>'scanId' = :'scan_id'
           AND lower(g.payload->>'imageDigest') = lower(:'digest')
      ))
     FROM eventing.outbox o
    WHERE o.stream = :'tenant_stream'
      AND o.event_type = 'scanner.scan.completed'
      AND o.payload->>'tenant' = :'tenant_id'
      AND o.payload->>'scanId' = :'scan_id'
      AND lower(o.payload->>'imageDigest') = lower(:'digest')
    ORDER BY o.seq;" \
  > "${SCN5_ROOT}/parity/forcing-scan-outbox.json"
SCN5_OUTBOX_RC=$?
set -e
test "${SCN5_OUTBOX_RC}" -eq 0
test "$(wc -l < "${SCN5_ROOT}/parity/forcing-scan-outbox.json")" -eq 1
jq -e \
  --arg stream "${SCN5_TENANT_STREAM}" \
  --arg tenant "${SCN5_TENANT}" \
  --arg scan "${SCN5_SCAN_ID}" \
  --arg digest "${SCN5_SCAN_DIGEST}" '
    .stream == $stream
    and .type == "scanner.scan.completed"
    and .v == 2
    and .seq == 1
    and (.streamEpoch | type) == "number"
    and .payload.tenant == $tenant
    and .payload.scanId == $scan
    and .payload.imageDigest == $digest
    and (.payload.status | ascii_downcase) == "succeeded"
    and (.payload.sbom | type) == "object"
    and .payload.sbom.payloadDigest == $digest
    and (.payload.sbom.canonicalBomSha256 | test("^sha256:[0-9a-f]{64}$"))
    and (.payload.componentPurls | type) == "array"
    and (.payload.componentStats.purlSetSha256 | test("^[0-9a-f]{64}$"))
    and (.payload.reachabilityFindings | type) == "array"
    and (.payload.reachability.findingSetSha256 | test("^[0-9a-f]{64}$"))
    and .retiredGlobalRows == 0
  ' "${SCN5_ROOT}/parity/forcing-scan-outbox.json" >/dev/null
SCN5_EVENT_EPOCH="$(jq -er '.streamEpoch' "${SCN5_ROOT}/parity/forcing-scan-outbox.json")"
SCN5_EVENT_CANONICAL_BOM_SHA="$(
  jq -er '.payload.sbom.canonicalBomSha256' \
    "${SCN5_ROOT}/parity/forcing-scan-outbox.json"
)"
SCN5_EVENT_PURL_SHA="$(
  jq -er '.payload.componentStats.purlSetSha256' \
    "${SCN5_ROOT}/parity/forcing-scan-outbox.json"
)"
SCN5_EVENT_FINDING_SHA="$(
  jq -er '.payload.reachability.findingSetSha256' \
    "${SCN5_ROOT}/parity/forcing-scan-outbox.json"
)"
jq -e \
  --arg build "${SCN5_SCAN_ID}" \
  --arg digest "${SCN5_SCAN_DIGEST}" \
  --arg canonical "${SCN5_EVENT_CANONICAL_BOM_SHA}" '
    .buildId == $build
    and .payloadDigest == $digest
    and .canonicalBomSha256 == $canonical
  ' "${SCN5_ROOT}/logs/forcing-scan-sbom-hot-lookup.json" >/dev/null

# A fresh stream returns event seq=1 and leaves stream_state.next_seq=2.
# stream_epoch is Unix epoch milliseconds minted on first append.
set +e
docker exec "${SCN5_PG_CONTAINER}" psql -XqAt -v ON_ERROR_STOP=1 \
  -U "${SCN5_PG_OPERATOR}" -d "${SCN5_TARGET_DB}" \
  -v tenant_stream="${SCN5_TENANT_STREAM}" \
  -v scan_id="${SCN5_SCAN_ID}" \
  -v window_epoch_ms="${SCN5_WINDOW_EPOCH_MS}" -c \
  "SELECT jsonb_build_object(
      'stream', s.stream,
      'streamEpoch', s.stream_epoch,
      'nextSeq', s.next_seq,
      'updatedAt', s.updated_at,
      'eventCount', count(o.*),
      'matchingEventCount', count(o.*) FILTER (
        WHERE o.payload->>'scanId' = :'scan_id'
          AND o.event_type = 'scanner.scan.completed'
      ),
      'minSeq', min(o.seq),
      'maxSeq', max(o.seq),
      'sameEpoch', bool_and(o.stream_epoch = s.stream_epoch),
      'epochAtOrAfterWindow',
        s.stream_epoch >= :'window_epoch_ms'::bigint,
      'retiredGlobalStateRows', (
        SELECT count(*)
          FROM eventing.stream_state g
         WHERE g.stream = 'scanner.scans'
      ))
     FROM eventing.stream_state s
     JOIN eventing.outbox o ON o.stream = s.stream
    WHERE s.stream = :'tenant_stream'
    GROUP BY s.stream, s.stream_epoch, s.next_seq, s.updated_at;" \
  > "${SCN5_ROOT}/parity/forcing-scan-stream-state.json"
SCN5_STREAM_STATE_RC=$?
set -e
test "${SCN5_STREAM_STATE_RC}" -eq 0
test "$(wc -l < "${SCN5_ROOT}/parity/forcing-scan-stream-state.json")" -eq 1
jq -e \
  --arg stream "${SCN5_TENANT_STREAM}" \
  --argjson event_epoch "${SCN5_EVENT_EPOCH}" '
    .stream == $stream
    and .streamEpoch == $event_epoch
    and .nextSeq == 2
    and .eventCount == 1
    and .matchingEventCount == 1
    and .minSeq == 1
    and .maxSeq == 1
    and .sameEpoch == true
    and .epochAtOrAfterWindow == true
    and .retiredGlobalStateRows == 0
  ' "${SCN5_ROOT}/parity/forcing-scan-stream-state.json" >/dev/null

# Read the digest projection and prove that its content hashes and carried
# collections are the same generation as the completion event.
SCN5_DETAIL_PATH="${SCN5_DETAIL_PATH_TEMPLATE/\{imageDigest\}/${SCN5_SCAN_DIGEST}}"
[[ "${SCN5_DETAIL_PATH}" != *'{'* ]]
SCN5_DETAIL_URL="${STELLAOPS_BACKEND_URL%/}${SCN5_DETAIL_PATH}"

set +e
SCN5_DETAIL_HTTP="$(
  scn5_curl_bearer "${SCN5_READ_TOKEN}" \
    --header "X-StellaOps-TenantId: ${SCN5_TENANT}" \
    --header 'Accept: application/json' \
    --output "${SCN5_ROOT}/logs/forcing-scan-detail.json" \
    --write-out '%{http_code}' \
    "${SCN5_DETAIL_URL}"
)"
SCN5_DETAIL_CURL_RC=$?
set -e
test "${SCN5_DETAIL_CURL_RC}" -eq 0
test "${SCN5_DETAIL_HTTP}" = 200
jq -e \
  --arg tenant "${SCN5_TENANT}" \
  --arg digest "${SCN5_SCAN_DIGEST}" \
  --arg scan "${SCN5_SCAN_ID}" \
  --arg purl_sha "${SCN5_EVENT_PURL_SHA}" \
  --arg finding_sha "${SCN5_EVENT_FINDING_SHA}" \
  --slurpfile event "${SCN5_ROOT}/parity/forcing-scan-outbox.json" '
    .tenant == $tenant
    and .imageDigest == $digest
    and .latestAttempt.scanId == $scan
    and (.latestAttempt.status | ascii_downcase) == "succeeded"
    and .latestSuccessfulEvidence.scanId == $scan
    and .latestSuccessfulEvidence.purlSetSha256 == $purl_sha
    and .latestSuccessfulEvidence.findingSetSha256 == $finding_sha
    and .latestSuccessfulEvidence.componentPurls
        == $event[0].payload.componentPurls
    and .latestSuccessfulEvidence.reachabilityFindings
        == $event[0].payload.reachabilityFindings
  ' "${SCN5_ROOT}/logs/forcing-scan-detail.json" >/dev/null

# X-new-3: the operator calls Scanner with scanner:read; Scanner calls Policy
# with its own exact-tenant policy:preview:invoke token.
jq -n --arg digest "${SCN5_SCAN_DIGEST}" '
  {
    namespace: "scn5-x-new-3",
    labels: {"stellaops.scn5": "x-new-3"},
    images: [$digest]
  }
' > "${SCN5_ROOT}/logs/xnew3-runtime-request.json"
jq -n --arg digest "${SCN5_SCAN_DIGEST}" \
  '{imageDigest: $digest, findings: [], baseline: []}' \
  > "${SCN5_ROOT}/logs/xnew3-report-request.json"

scn5_post_json() {
  local url="$1"
  local request_file="$2"
  local response_file="$3"
  local http_code
  local curl_rc

  if http_code="$(
    scn5_curl_bearer "${SCN5_READ_TOKEN}" \
      --max-time 25 \
      --header "X-StellaOps-TenantId: ${SCN5_TENANT}" \
      --header 'Accept: application/json' \
      --header 'Content-Type: application/json' \
      --data-binary "@${request_file}" \
      --output "${response_file}" \
      --write-out '%{http_code}' \
      "${url}"
  )"; then
    curl_rc=0
  else
    curl_rc=$?
  fi

  test "${curl_rc}" -eq 0 || return "${curl_rc}"
  printf '%s' "${http_code}"
}

# The route must not be anonymous.
set +e
SCN5_RUNTIME_NOAUTH_HTTP="$(
  "${SCN5_CURL[@]}" --max-time 20 \
    --header 'Accept: application/json' \
    --header 'Content-Type: application/json' \
    --data-binary "@${SCN5_ROOT}/logs/xnew3-runtime-request.json" \
    --output "${SCN5_ROOT}/logs/xnew3-runtime-noauth.json" \
    --write-out '%{http_code}' \
    "${SCN5_RUNTIME_URL}"
)"
SCN5_RUNTIME_NOAUTH_RC=$?
set -e
test "${SCN5_RUNTIME_NOAUTH_RC}" -eq 0
test "${SCN5_RUNTIME_NOAUTH_HTTP}" = 401

# Scanner's database must not contain Policy's schema.
set +e
docker exec "${SCN5_PG_CONTAINER}" \
  psql -XqAt -v ON_ERROR_STOP=1 \
  -U "${SCN5_PG_OPERATOR}" -d "${SCN5_TARGET_DB}" \
  -c "SELECT to_regnamespace('policy') IS NULL;" \
  > "${SCN5_ROOT}/logs/xnew3-policy-schema-absent.txt"
SCN5_POLICY_SCHEMA_RC=$?
set -e
test "${SCN5_POLICY_SCHEMA_RC}" -eq 0
grep -qx 't' "${SCN5_ROOT}/logs/xnew3-policy-schema-absent.txt"

# Bind the state-aware owner acceptance branch to Policy's exact tenant and compatibility row.
docker exec -i "${SCN5_PG_CONTAINER}" \
  psql -XqAt -F $'\t' -v ON_ERROR_STOP=1 \
  -v scn5_tenant="${SCN5_TENANT}" \
  -v scn5_policy_id="${SCN5_POLICY_COMPATIBILITY_ID}" \
  -U "${SCN5_PG_OPERATOR}" -d "${SCN5_POLICY_DB}" \
  > "${SCN5_ROOT}/logs/xnew3-policy-compatibility-count.tsv" <<'SQL'
SELECT :'scn5_tenant', :'scn5_policy_id', count(*)
FROM policy.snapshots
WHERE tenant_id = :'scn5_tenant'
  AND policy_id = :'scn5_policy_id'::uuid;
SQL
IFS=$'\t' read -r SCN5_POLICY_RECEIPT_TENANT SCN5_POLICY_RECEIPT_ID \
  SCN5_POLICY_COMPAT_COUNT \
  < "${SCN5_ROOT}/logs/xnew3-policy-compatibility-count.tsv"
test "${SCN5_POLICY_RECEIPT_TENANT}" = "${SCN5_TENANT}"
test "${SCN5_POLICY_RECEIPT_ID}" = "${SCN5_POLICY_COMPATIBILITY_ID}"
[[ "${SCN5_POLICY_COMPAT_COUNT}" =~ ^[0-9]+$ ]]

# Healthy owner-path baselines. Signing and SBOM facts are observations, not seam preconditions.
set +e
SCN5_RUNTIME_HEALTHY_HTTP="$(
  scn5_post_json \
    "${SCN5_RUNTIME_URL}" \
    "${SCN5_ROOT}/logs/xnew3-runtime-request.json" \
    "${SCN5_ROOT}/logs/xnew3-runtime-healthy.json"
)"
SCN5_RUNTIME_HEALTHY_RC=$?
set -e
test "${SCN5_RUNTIME_HEALTHY_RC}" -eq 0
test "${SCN5_RUNTIME_HEALTHY_HTTP}" = 200
jq -e --arg digest "${SCN5_SCAN_DIGEST}" '
  (.ttlSeconds | type) == "number"
  and .ttlSeconds > 0
  and (.expiresAtUtc | type) == "string"
  and (.results[$digest].signed | type) == "boolean"
  and (.results[$digest].hasSbomReferrers | type) == "boolean"
  and (.results[$digest].hasSbom | type) == "boolean"
  and ([.results[$digest].reasons[]?
        | select(startswith("policy.owner."))] | length) == 0
  and (.results[$digest].policyVerdict as $verdict
       | ["pass", "warn", "fail"] | index($verdict) != null)
' "${SCN5_ROOT}/logs/xnew3-runtime-healthy.json" >/dev/null

set +e
SCN5_REPORT_HEALTHY_HTTP="$(
  scn5_post_json \
    "${SCN5_REPORT_URL}" \
    "${SCN5_ROOT}/logs/xnew3-report-request.json" \
    "${SCN5_ROOT}/logs/xnew3-report-healthy.json"
)"
SCN5_REPORT_HEALTHY_RC=$?
set -e
test "${SCN5_REPORT_HEALTHY_RC}" -eq 0
if test "${SCN5_POLICY_COMPAT_COUNT}" -eq 0; then
  jq -e --arg digest "${SCN5_SCAN_DIGEST}" '
    .policyRevision == null
    and (.results[$digest].reasons | index("policy.snapshot.missing") != null)
  ' "${SCN5_ROOT}/logs/xnew3-runtime-healthy.json" >/dev/null
  test "${SCN5_REPORT_HEALTHY_HTTP}" = 503
  jq -e '
    .status == 503
    and .detail == "scanner.report.no_policy_snapshot"
    and (.extensions.issues | any(.code == "policy.preview.snapshot_missing"))
    and (has("report") | not)
  ' "${SCN5_ROOT}/logs/xnew3-report-healthy.json" >/dev/null
else
  test "${SCN5_POLICY_COMPAT_COUNT}" -gt 0
  jq -e --arg digest "${SCN5_SCAN_DIGEST}" '
    (.policyRevision | type) == "string"
    and (.policyRevision | length) > 0
    and (.results[$digest].reasons | index("policy.snapshot.missing") == null)
  ' "${SCN5_ROOT}/logs/xnew3-runtime-healthy.json" >/dev/null
  test "${SCN5_REPORT_HEALTHY_HTTP}" = 200
  jq -e --arg digest "${SCN5_SCAN_DIGEST}" '
    .report.imageDigest == $digest
    and (.report.reportId | type) == "string"
    and (.report.reportId | length) > 0
    and (.report.policy.revisionId | type) == "string"
    and (.report.policy.revisionId | length) > 0
    and (.report.policy.digest | type) == "string"
    and (.report.policy.digest | length) > 0
    and (.report.verdict | type) == "string"
    and (.report.verdict | length) > 0
    and (.report.verdicts | type) == "array"
    and (.report.issues | type) == "array"
  ' "${SCN5_ROOT}/logs/xnew3-report-healthy.json" >/dev/null
fi

SCN5_NORMAL_COMPOSE_FILES="${SCN5_CANDIDATE_OVERLAY}"
SCN5_OWNER_OUTAGE_OVERLAY_FILE="${SCN5_ROOT}/topology/policy-owner-outage.override.yml"
SCN5_OWNER_OUTAGE_OVERLAY="${SCN5_COMPOSE_ROOT_REL}/topology/policy-owner-outage.override.yml"
SCN5_OWNER_OUTAGE_COMPOSE_FILES="${SCN5_CANDIDATE_OVERLAY} ${SCN5_OWNER_OUTAGE_OVERLAY}"
test "$(sha256sum "${SCN5_OWNER_OUTAGE_OVERLAY_FILE}" | awk '{print $1}')" = \
  '84029e2be9fe4c610da2255f6a1aba1ec7e8a8ed86de65c28a3e13070771bcda'
test -f "devops/compose/${SCN5_OWNER_OUTAGE_OVERLAY}"

# Normal render relies on source defaults. The outage overlay changes web only.
set +e
COMPOSE_EXTRA_FILES="${SCN5_NORMAL_COMPOSE_FILES}" \
  ./devops/compose/scripts/compose-cli.sh config --format json \
  | jq -e --arg window "${SCN5_WINDOW_ID}" '
      (((.services["scanner-web"].environment // {})
        | has("SCANNER_SCANNER__POLICYOWNER__BASEADDRESS")) | not)
      and (((.services["scanner-web"].environment // {})
        | has("SCANNER_SCANNER__POLICYOWNER__REQUESTTIMEOUTSECONDS")) | not)
      and .services["scanner-web"].environment.SCANNER_SCANNER__CUTOVER__INTAKEWINDOWID
        == $window
      and (((.services["scanner-worker"].environment // {})
        | has("SCANNER_SCANNER__CUTOVER__INTAKEWINDOWID")) | not)
    ' >/dev/null
SCN5_NORMAL_RENDER_RC=("${PIPESTATUS[@]}")
set -e
test "${#SCN5_NORMAL_RENDER_RC[@]}" -eq 2
test "${SCN5_NORMAL_RENDER_RC[0]}" -eq 0
test "${SCN5_NORMAL_RENDER_RC[1]}" -eq 0

set +e
COMPOSE_EXTRA_FILES="${SCN5_OWNER_OUTAGE_COMPOSE_FILES}" \
  ./devops/compose/scripts/compose-cli.sh config --format json \
  | jq -e --arg window "${SCN5_WINDOW_ID}" '
      .services["scanner-web"].environment.SCANNER_SCANNER__POLICYOWNER__BASEADDRESS
        == "http://127.0.0.1:9"
      and .services["scanner-web"].environment.SCANNER_SCANNER__POLICYOWNER__REQUESTTIMEOUTSECONDS
        == "2"
      and .services["scanner-web"].environment.SCANNER_SCANNER__CUTOVER__INTAKEWINDOWID
        == $window
      and (((.services["scanner-worker"].environment // {})
        | has("SCANNER_SCANNER__POLICYOWNER__BASEADDRESS")) | not)
      and (((.services["scanner-worker"].environment // {})
        | has("SCANNER_SCANNER__POLICYOWNER__REQUESTTIMEOUTSECONDS")) | not)
    ' >/dev/null
SCN5_OUTAGE_RENDER_RC=("${PIPESTATUS[@]}")
set -e
test "${#SCN5_OUTAGE_RENDER_RC[@]}" -eq 2
test "${SCN5_OUTAGE_RENDER_RC[0]}" -eq 0
test "${SCN5_OUTAGE_RENDER_RC[1]}" -eq 0

SCN5_OWNER_OUTAGE_EXPECTED_ENV_NAMES="${SCN5_ROOT}/topology/stellaops-scanner-web.owner-outage-expected-env-names.txt"
cp "${SCN5_ROOT}/topology/stellaops-scanner-web.candidate-expected-env-names.txt" \
  "${SCN5_OWNER_OUTAGE_EXPECTED_ENV_NAMES}"
printf '%s\n' SCANNER_SCANNER__POLICYOWNER__BASEADDRESS \
  SCANNER_SCANNER__POLICYOWNER__REQUESTTIMEOUTSECONDS \
  >> "${SCN5_OWNER_OUTAGE_EXPECTED_ENV_NAMES}"
LC_ALL=C sort -u -o "${SCN5_OWNER_OUTAGE_EXPECTED_ENV_NAMES}" \
  "${SCN5_OWNER_OUTAGE_EXPECTED_ENV_NAMES}"
SCN5_OWNER_OUTAGE_EXPECTED_ENV_HMAC="${SCN5_ROOT}/topology/stellaops-scanner-web.owner-outage-env.hmac"
COMPOSE_EXTRA_FILES="${SCN5_OWNER_OUTAGE_COMPOSE_FILES}" \
  scn5_capture_rendered_env_hmac scanner-web "${SCN5_WEB_CANDIDATE_IMAGE}" \
    "${SCN5_OWNER_OUTAGE_EXPECTED_ENV_HMAC}"

scn5_assert_candidate_image_ids
test "$(docker inspect stellaops-scanner-web --format '{{.Image}}')" = \
  "${SCN5_WEB_CANDIDATE_IMAGE_ID}"
test "$(docker inspect stellaops-scanner-worker --format '{{.Image}}')" = \
  "${SCN5_WORKER_CANDIDATE_IMAGE_ID}"
SCN5_WORKER_CONTAINER_ID="$(docker inspect stellaops-scanner-worker --format '{{.Id}}')"
SCN5_POLICY_CONTAINER_ID="$(docker inspect stellaops-policy-engine --format '{{.Id}}')"
SCN5_POLICY_RESTARTS="$(docker inspect stellaops-policy-engine --format '{{.RestartCount}}')"

SCN5_WEB_RESTORE_NEEDED=0
scn5_restore_web() {
  local restore_rc

  test "${SCN5_WEB_RESTORE_NEEDED:-0}" = 1 || return 0
  scn5_assert_candidate_image_ids || return $?
  if COMPOSE_EXTRA_FILES="${SCN5_NORMAL_COMPOSE_FILES}" \
      ./devops/compose/scripts/compose-cli.sh recreate scanner-web \
      >> "${SCN5_ROOT}/logs/xnew3-scanner-web-restore.txt" 2>&1; then
    restore_rc=0
  else
    restore_rc=$?
  fi
  test "${restore_rc}" -eq 0 || return "${restore_rc}"
  scn5_wait_container_healthy stellaops-scanner-web || return $?
  docker exec stellaops-scanner-web sh -eu -c '
    test -z "${SCANNER_SCANNER__POLICYOWNER__BASEADDRESS+x}"
    test -z "${SCANNER_SCANNER__POLICYOWNER__REQUESTTIMEOUTSECONDS+x}"
    test "${SCANNER_SCANNER__CUTOVER__INTAKEWINDOWID}" = "$1"
  ' scn5-restore "${SCN5_WINDOW_ID}" || return $?
  scn5_assert_live_candidate_contract \
    stellaops-scanner-web "${SCN5_WEB_CANDIDATE_IMAGE_ID}" \
    "${SCN5_ROOT}/topology/stellaops-scanner-web.candidate-expected-env-names.txt" \
    "${SCN5_ROOT}/topology/stellaops-scanner-web.candidate-env.hmac" \
    owner-restored || return $?

  SCN5_WEB_RESTORE_NEEDED=0
  return 0
}
trap scn5_restore_web EXIT

# Recreate scanner-web only onto a bounded black-hole owner address. Policy
# remains running and no Policy DML occurs.
SCN5_WEB_RESTORE_NEEDED=1
scn5_assert_candidate_image_ids
set +e
COMPOSE_EXTRA_FILES="${SCN5_OWNER_OUTAGE_COMPOSE_FILES}" \
  ./devops/compose/scripts/compose-cli.sh recreate scanner-web \
  > "${SCN5_ROOT}/logs/xnew3-scanner-web-owner-outage.txt" 2>&1
SCN5_OWNER_OUTAGE_RECREATE_RC=$?
set -e
sed -n '1,160p' "${SCN5_ROOT}/logs/xnew3-scanner-web-owner-outage.txt"
test "${SCN5_OWNER_OUTAGE_RECREATE_RC}" -eq 0
scn5_wait_container_healthy stellaops-scanner-web
docker exec stellaops-scanner-web sh -eu -c '
  test "${SCANNER_SCANNER__POLICYOWNER__BASEADDRESS}" = "http://127.0.0.1:9"
  test "${SCANNER_SCANNER__POLICYOWNER__REQUESTTIMEOUTSECONDS}" = "2"
  test "${SCANNER_SCANNER__CUTOVER__INTAKEWINDOWID}" = "$1"
' scn5-outage "${SCN5_WINDOW_ID}"
scn5_assert_live_candidate_contract \
  stellaops-scanner-web "${SCN5_WEB_CANDIDATE_IMAGE_ID}" \
  "${SCN5_OWNER_OUTAGE_EXPECTED_ENV_NAMES}" \
  "${SCN5_OWNER_OUTAGE_EXPECTED_ENV_HMAC}" owner-outage
test "$(docker inspect stellaops-scanner-worker --format '{{.Id}}')" = \
  "${SCN5_WORKER_CONTAINER_ID}"
test "$(docker inspect stellaops-scanner-worker --format '{{.Image}}')" = \
  "${SCN5_WORKER_CANDIDATE_IMAGE_ID}"
test "$(docker inspect stellaops-policy-engine --format '{{.Id}}')" = \
  "${SCN5_POLICY_CONTAINER_ID}"
test "$(docker inspect stellaops-policy-engine --format '{{.RestartCount}}')" = \
  "${SCN5_POLICY_RESTARTS}"

# Runtime-policy remains an HTTP 200 operation result but must fail closed for
# the exact owner reason. Signing/SBOM facts must remain byte-for-byte equal to
# the healthy owner-path observation; the outage may not erase them.
set +e
SCN5_RUNTIME_OUTAGE_HTTP="$(
  scn5_post_json \
    "${SCN5_RUNTIME_URL}" \
    "${SCN5_ROOT}/logs/xnew3-runtime-request.json" \
    "${SCN5_ROOT}/logs/xnew3-runtime-owner-outage.json"
)"
SCN5_RUNTIME_OUTAGE_RC=$?
set -e
test "${SCN5_RUNTIME_OUTAGE_RC}" -eq 0
test "${SCN5_RUNTIME_OUTAGE_HTTP}" = 200
jq -e --arg digest "${SCN5_SCAN_DIGEST}" \
  --slurpfile healthy "${SCN5_ROOT}/logs/xnew3-runtime-healthy.json" '
  .results[$digest].policyVerdict == "fail"
  and .results[$digest].signed == $healthy[0].results[$digest].signed
  and .results[$digest].hasSbomReferrers
      == $healthy[0].results[$digest].hasSbomReferrers
  and .results[$digest].hasSbom == $healthy[0].results[$digest].hasSbom
  and (.results[$digest].reasons
       | index("policy.owner.owner_unavailable") != null)
' "${SCN5_ROOT}/logs/xnew3-runtime-owner-outage.json" >/dev/null

# Report transport failure is HTTP 503 and returns no report.
set +e
SCN5_REPORT_OUTAGE_HTTP="$(
  scn5_post_json \
    "${SCN5_REPORT_URL}" \
    "${SCN5_ROOT}/logs/xnew3-report-request.json" \
    "${SCN5_ROOT}/logs/xnew3-report-owner-outage.json"
)"
SCN5_REPORT_OUTAGE_RC=$?
set -e
test "${SCN5_REPORT_OUTAGE_RC}" -eq 0
test "${SCN5_REPORT_OUTAGE_HTTP}" = 503
jq -e '
  .status == 503
  and .detail == "owner_unavailable"
  and (.instance | endswith("/api/v1/reports"))
  and (.extensions.traceId | type) == "string"
  and (.extensions.traceId | length) > 0
  and (has("report") | not)
' "${SCN5_ROOT}/logs/xnew3-report-owner-outage.json" >/dev/null

# This estate has report-event publication disabled. A live Redis no-growth
# check would therefore be vacuous; the non-vacuous proof is the required green
# ReportsEndpointFailsClosedWithoutPublishingWhenPolicyOwnerIsUnavailable test,
# which installs a recording publisher and requires zero calls.
test "$(docker exec stellaops-scanner-web \
  printenv SCANNER_SCANNER__EVENTS__ENABLED)" = false

# Restore immediately through the exact normal Compose chain.
set +e
scn5_restore_web
SCN5_WEB_RESTORE_RC=$?
set -e
test "${SCN5_WEB_RESTORE_RC}" -eq 0
trap - EXIT
test "$(docker inspect stellaops-scanner-worker --format '{{.Id}}')" = \
  "${SCN5_WORKER_CONTAINER_ID}"
test "$(docker inspect stellaops-policy-engine --format '{{.Id}}')" = \
  "${SCN5_POLICY_CONTAINER_ID}"
test "$(docker inspect stellaops-policy-engine --format '{{.RestartCount}}')" = \
  "${SCN5_POLICY_RESTARTS}"

# Prove functional restoration; container health alone is not sufficient.
set +e
SCN5_RUNTIME_RESTORED_HTTP="$(
  scn5_post_json \
    "${SCN5_RUNTIME_URL}" \
    "${SCN5_ROOT}/logs/xnew3-runtime-request.json" \
    "${SCN5_ROOT}/logs/xnew3-runtime-restored.json"
)"
SCN5_RUNTIME_RESTORED_RC=$?
set -e
test "${SCN5_RUNTIME_RESTORED_RC}" -eq 0
test "${SCN5_RUNTIME_RESTORED_HTTP}" = 200
jq -e --arg digest "${SCN5_SCAN_DIGEST}" '
  ([.results[$digest].reasons[]?
        | select(startswith("policy.owner."))] | length) == 0
' "${SCN5_ROOT}/logs/xnew3-runtime-restored.json" >/dev/null

set +e
SCN5_REPORT_RESTORED_HTTP="$(
  scn5_post_json \
    "${SCN5_REPORT_URL}" \
    "${SCN5_ROOT}/logs/xnew3-report-request.json" \
    "${SCN5_ROOT}/logs/xnew3-report-restored.json"
)"
SCN5_REPORT_RESTORED_RC=$?
set -e
test "${SCN5_REPORT_RESTORED_RC}" -eq 0
test "${SCN5_REPORT_RESTORED_HTTP}" = "${SCN5_REPORT_HEALTHY_HTTP}"
if test "${SCN5_REPORT_HEALTHY_HTTP}" = 503; then
  test "${SCN5_POLICY_COMPAT_COUNT}" -eq 0
  jq -e --arg digest "${SCN5_SCAN_DIGEST}" '
    .policyRevision == null
    and (.results[$digest].reasons | index("policy.snapshot.missing") != null)
  ' "${SCN5_ROOT}/logs/xnew3-runtime-restored.json" >/dev/null
  jq -e '
    .status == 503
    and .detail == "scanner.report.no_policy_snapshot"
    and (.extensions.issues | any(.code == "policy.preview.snapshot_missing"))
    and (has("report") | not)
  ' "${SCN5_ROOT}/logs/xnew3-report-restored.json" >/dev/null
else
  test "${SCN5_POLICY_COMPAT_COUNT}" -gt 0
  jq -e --arg digest "${SCN5_SCAN_DIGEST}" '
    (.policyRevision | type) == "string"
    and (.policyRevision | length) > 0
    and (.results[$digest].reasons | index("policy.snapshot.missing") == null)
  ' "${SCN5_ROOT}/logs/xnew3-runtime-restored.json" >/dev/null
  jq -e --arg digest "${SCN5_SCAN_DIGEST}" '
    .report.imageDigest == $digest
    and (.report.policy.revisionId | type) == "string"
    and (.report.policy.revisionId | length) > 0
  ' "${SCN5_ROOT}/logs/xnew3-report-restored.json" >/dev/null
fi

The eventing schema has no separate envelope-hash column. The four content-addressed values above are the event payload’s imageDigest, sbom.payloadDigest / sbom.canonicalBomSha256, componentStats.purlSetSha256, and reachability.findingSetSha256. A fresh first append writes outbox seq=1 and advances stream_state.next_seq to 2. A pre-window epoch, any global-stream row for this scan, or any source fingerprint change is a NO-GO. Do not delete or seal the old global stream during SCN-5; SCN-8 owns its later retention-gated cleanup.

14. Revoke source access — by removing the credential, not the grant

Read the boundary honestly before running anything: the live hosts connect to stellaops_platform as the cluster superuser stellaops (H5). No REVOKE can fence a superuser. The revocation that actually holds is step 12’s deletion of both transitional keys, and the check is that no Scanner container carries any DSN naming the source database.

for c in stellaops-scanner-web stellaops-scanner-worker; do
  if docker exec "$c" sh -lc '
    for name in STELLAOPS_POSTGRES_SCANNER_CONNECTION \
                SCANNER_SCANNER__STORAGE__DSN \
                ScannerStorage__Postgres__ConnectionString; do
      value="$(printenv "$name" 2>/dev/null || true)"
      case "$value" in *Database=stellaops_platform*) exit 1 ;; esac
    done
  '; then
    printf '%s source-dsn=absent\n' "$c"
  else
    exit 1
  fi
done

Verified by: both sanitized receipts say source-dsn=absent; no environment value is printed.

Then assert the role boundary is already what ADR-039 requires:

docker exec "${SCN5_PG_CONTAINER}" psql -XqAt -U "${SCN5_PG_OPERATOR}" -d postgres -c \
  "SELECT has_database_privilege('scanner','stellaops_platform','CONNECT'),
          has_database_privilege('scanner','stellaops_scanner','CONNECT'),
          has_database_privilege('scanner','stellaops_findings','CONNECT');"
SCN5_ISOLATION_ENV="$(mktemp "${SCN5_REPO_ROOT}/.local-secrets/scn5-db-probe.XXXXXX")"
case "${SCN5_ISOLATION_ENV}" in
  "${SCN5_REPO_ROOT}"/.local-secrets/scn5-db-probe.*) ;;
  *) exit 1 ;;
esac
cp devops/compose/.env "${SCN5_ISOLATION_ENV}"
chmod 600 "${SCN5_ISOLATION_ENV}"
set +e
docker inspect stellaops-vulnerabilities-web \
  --format '{{range .Config.Env}}{{println .}}{{end}}' \
  | grep '^STELLAOPS_POSTGRES_VULN_CONNECTION=' >> "${SCN5_ISOLATION_ENV}"
SCN5_VULN_ENV_RC=("${PIPESTATUS[@]}")
bash tools/scripts/deploy/postgres/probe-database-isolation.sh \
  --env-file "${SCN5_ISOLATION_ENV}" \
  > "${SCN5_ROOT}/logs/post-cutover-database-isolation.txt"
SCN5_ISOLATION_RC=$?
set -e
rm -f -- "${SCN5_ISOLATION_ENV}"
test ! -e "${SCN5_ISOLATION_ENV}"
test "${#SCN5_VULN_ENV_RC[@]}" -eq 2
test "${SCN5_VULN_ENV_RC[0]}" -eq 0
test "${SCN5_VULN_ENV_RC[1]}" -eq 0
test "${SCN5_ISOLATION_RC}" -eq 0
grep -Fq 'PASS: every role reaches exactly its own database' \
  "${SCN5_ROOT}/logs/post-cutover-database-isolation.txt"

Verified by: f|t|f, and the probe reporting PASS: every role reaches exactly its own database. Both were already true at prep on 2026-08-28; this step confirms the window did not weaken them.

Do not drop the source schema here. It is SCN-6’s work, it is rollback’s only safety net, and it is a separate destructive approval.

15. Declare the gate, then soak

Record in the sprint’s SCN-5 Status and Decisions & Risks: the parity receipt paths, the digest of both dumps, the doctor verdict, the forcing-function result, the epoch reading, and the X-new-1 (H7) and X-new-3 dispositions. Until that receipt is complete, admit no traffic beyond the controlled forcing scan. Deliberately open external intake only after every gate is green; that is the point at which automatic rollback to the frozen source ends. Soak with both writers live, then re-run the step-9 source fingerprint and require it still equals source-frozen.tsv — a source that moved during the soak means a writer was never repointed.

The candidate overlay deliberately carries the intake fence, so success requires an explicit web-only recreate that sets it empty. ScanIntakeFenceCoordinator treats the empty string as no fence. Keep the worker untouched and prove the already-completed scan remains readable after the web swap.

Before the fence, use apply_patch to create tmp/scn5-<window-id>/topology/intake-unfence.override.yml with this exact LF-terminated template:

services:
  scanner-web:
    environment:
      SCANNER_SCANNER__CUTOVER__INTAKEWINDOWID: ""

scn5_assert_live_logs pre-unfreeze
SCN5_UNFENCE_OVERLAY_FILE="${SCN5_ROOT}/topology/intake-unfence.override.yml"
SCN5_UNFENCE_OVERLAY="${SCN5_COMPOSE_ROOT_REL}/topology/intake-unfence.override.yml"
SCN5_SUCCESS_COMPOSE_FILES="${SCN5_CANDIDATE_OVERLAY} ${SCN5_UNFENCE_OVERLAY}"
test "$(sha256sum "${SCN5_UNFENCE_OVERLAY_FILE}" | awk '{print $1}')" = \
  '85b34ebf4cc90e865c987e7bdaa8c4406c3e31943c6a3566247177a617ff337a'
test -f "devops/compose/${SCN5_UNFENCE_OVERLAY}"

set +e
COMPOSE_EXTRA_FILES="${SCN5_SUCCESS_COMPOSE_FILES}" \
  ./devops/compose/scripts/compose-cli.sh config --format json \
  | jq -e '
      .services["scanner-web"].environment.SCANNER_SCANNER__CUTOVER__INTAKEWINDOWID == ""
      and ((.services["scanner-worker"].environment // {})
        | has("SCANNER_SCANNER__CUTOVER__INTAKEWINDOWID") | not)
    ' >/dev/null
SCN5_UNFENCE_RENDER_RC=("${PIPESTATUS[@]}")
set -e
test "${#SCN5_UNFENCE_RENDER_RC[@]}" -eq 2
test "${SCN5_UNFENCE_RENDER_RC[0]}" -eq 0
test "${SCN5_UNFENCE_RENDER_RC[1]}" -eq 0

SCN5_UNFENCED_EXPECTED_ENV_NAMES="${SCN5_ROOT}/topology/stellaops-scanner-web.unfenced-expected-env-names.txt"
cp \
  "${SCN5_ROOT}/topology/stellaops-scanner-web.candidate-expected-env-names.txt" \
  "${SCN5_UNFENCED_EXPECTED_ENV_NAMES}"
SCN5_UNFENCED_EXPECTED_ENV_HMAC="${SCN5_ROOT}/topology/stellaops-scanner-web.unfenced-env.hmac"
COMPOSE_EXTRA_FILES="${SCN5_SUCCESS_COMPOSE_FILES}" \
  scn5_capture_rendered_env_hmac scanner-web "${SCN5_WEB_CANDIDATE_IMAGE}" \
    "${SCN5_UNFENCED_EXPECTED_ENV_HMAC}"

scn5_assert_candidate_image_ids
SCN5_WORKER_BEFORE_UNFENCE="$(docker inspect stellaops-scanner-worker --format '{{.Id}}')"
set +e
COMPOSE_EXTRA_FILES="${SCN5_SUCCESS_COMPOSE_FILES}" \
  ./devops/compose/scripts/compose-cli.sh recreate scanner-web \
  > "${SCN5_ROOT}/logs/intake-unfence-recreate.txt" 2>&1
SCN5_UNFENCE_RECREATE_RC=$?
set -e
sed -n '1,160p' "${SCN5_ROOT}/logs/intake-unfence-recreate.txt"
test "${SCN5_UNFENCE_RECREATE_RC}" -eq 0
scn5_wait_container_healthy stellaops-scanner-web
docker exec stellaops-scanner-web sh -eu -c '
  test "${SCANNER_SCANNER__CUTOVER__INTAKEWINDOWID+x}" = x
  test -z "${SCANNER_SCANNER__CUTOVER__INTAKEWINDOWID}"
'
scn5_assert_live_candidate_contract \
  stellaops-scanner-web "${SCN5_WEB_CANDIDATE_IMAGE_ID}" \
  "${SCN5_UNFENCED_EXPECTED_ENV_NAMES}" \
  "${SCN5_UNFENCED_EXPECTED_ENV_HMAC}" unfenced
test "$(docker inspect stellaops-scanner-worker --format '{{.Id}}')" = \
  "${SCN5_WORKER_BEFORE_UNFENCE}"
test "$(docker inspect stellaops-scanner-worker --format '{{.Image}}')" = \
  "${SCN5_WORKER_CANDIDATE_IMAGE_ID}"

SCN5_POST_UNFENCE_HTTP="$(
  scn5_curl_bearer "${SCN5_READ_TOKEN}" \
    --header "X-StellaOps-TenantId: ${SCN5_TENANT}" \
    --header 'Accept: application/json' \
    --output "${SCN5_ROOT}/logs/post-unfence-scan-status.json" \
    --write-out '%{http_code}' \
    "${SCN5_STATUS_URL}"
)"
test "${SCN5_POST_UNFENCE_HTTP}" = 200
jq -e --arg scan "${SCN5_SCAN_ID}" '
  .scanId == $scan and (.status | ascii_downcase) == "succeeded"
' "${SCN5_ROOT}/logs/post-unfence-scan-status.json" >/dev/null

sleep 60
scn5_fingerprint "${SCN5_SOURCE_DB}" "${SCN5_ROOT}/parity/source-after-soak.tsv"
cmp "${SCN5_ROOT}/parity/source-frozen.tsv" \
    "${SCN5_ROOT}/parity/source-after-soak.tsv"
test "$(sha256sum "${SCN5_LOCAL_ESTATE_PIN}" | awk '{print $1}')" = \
  "${SCN5_CANDIDATE_LOCAL_PIN_SHA}"
scn5_assert_local_estate_pin \
  "${SCN5_AUTHORITY_CANDIDATE_IMAGE_ID}" "${SCN5_POLICY_CANDIDATE_IMAGE_ID}" \
  "${SCN5_WEB_CANDIDATE_IMAGE_ID}" "${SCN5_WORKER_CANDIDATE_IMAGE_ID}"
env -u COMPOSE_EXTRA_FILES bash -c '
  set -euo pipefail
  ./devops/compose/scripts/compose-cli.sh config --format json \
    | tee "$5" \
    | jq -e --arg authority "$1" --arg policy "$2" --arg web "$3" --arg worker "$4" \
      '\'' .services.authority.image == $authority
          and .services["policy-engine"].image == $policy
          and .services["scanner-web"].image == $web
          and .services["scanner-worker"].image == $worker '\'' >/dev/null
' bash "${SCN5_AUTHORITY_CANDIDATE_IMAGE_ID}" "${SCN5_POLICY_CANDIDATE_IMAGE_ID}" \
  "${SCN5_WEB_CANDIDATE_IMAGE_ID}" "${SCN5_WORKER_CANDIDATE_IMAGE_ID}" \
  "${SCN5_ROOT}/topology/canonical-post-soak-compose.json"
for running_pair in \
    "stellaops-authority:${SCN5_AUTHORITY_CANDIDATE_IMAGE_ID}" \
    "stellaops-policy-engine:${SCN5_POLICY_CANDIDATE_IMAGE_ID}" \
    "stellaops-scanner-web:${SCN5_WEB_CANDIDATE_IMAGE_ID}" \
    "stellaops-scanner-worker:${SCN5_WORKER_CANDIDATE_IMAGE_ID}"; do
  test "$(docker inspect "${running_pair%%:*}" --format '{{.Image}}')" = \
    "${running_pair#*:}"
done
scn5_assert_live_logs post-soak
for cleanup_worktree in "${SCN5_INVENTORY_WORKTREE}" "${SCN5_BUNDLE_BUILD_ROOT}"; do
  case "${cleanup_worktree}" in
    "${SCN5_REPO_ROOT}"/tmp/worktrees/scn5-*|"${SCN5_REPO_ROOT}"/tmp/worktrees/scn-live-build-*) ;;
    *) exit 1 ;;
  esac
  git -C "${SCN5_REPO_ROOT}" worktree remove --force "${cleanup_worktree}"
  test ! -e "${cleanup_worktree}"
done
test "${SCN5_CANDIDATE_ENV}" = "${SCN5_REPO_ROOT}/tmp/scn5-bundle-candidate.env"
rm -f -- "${SCN5_CANDIDATE_ENV}"
unset SCN5_SCAN_TOKEN SCN5_READ_TOKEN \
  STELLAOPS_RELEASE_DISPATCH_CLIENT_SECRET \
  SCN5_TOPOLOGY_HMAC_KEY SCANNER_POSTGRES_PASSWORD

Verified by: the unfenced render and recreated web container carry the intake-window key with an explicit empty value, which the coordinator treats as disabled; only scanner-web is recreated; the worker container ID is unchanged; both candidate tags remain on the captured immutable IDs; the web container’s complete names and keyed environment HMAC equal the unfenced render; its stable runtime topology matches the pre-window projection except for the explicitly normalized Scanner base mount source, which resolves to the durable primary root; it is healthy on the exact candidate image; the controlled scan still reads as succeeded; the source fingerprint is unchanged after the bounded soak; the post-soak logs remain free of the fatal/startup/plugin-load patterns; and a fresh shell with no extra overlays records canonical Compose output carrying all four durable literal image IDs, which equal the four running container image IDs. The successful web recreate is the explicit external-intake open point. Keep the git-ignored local-estate pin file; do not remove it during cleanup.

Rollback — automatic only before external intake is unfenced

The preserved source makes an automatic rollback possible only while the target contains no real post-cutover traffic. The one forcing scan in step 13 is controlled synthetic data and is explicitly disposable on rollback. Once external intake is unfenced, the two databases can diverge and returning to the source is a reconciliation operation, not this runbook’s rollback.

Failures before step 4 require no rollback: the old Scanner containers are still live on the source, so retain the receipts and stop. Enter this rollback only after step 4 has fenced the writers. The process freeze remains in force through rollback step 5. The outgoing image predates the source-level intake fence, and this estate has no independent gateway/network fence. Therefore rollback cannot truthfully start an old web container while claiming intake is still closed: rollback step 6 is the explicit source-intake unfreeze point, not an automatic controlled-probe phase.

1. Stop both writers again if they were recreated: docker stop stellaops-scanner-web stellaops-scanner-worker, then assert Running=false on both.

   printf 'authority=%s\npolicy-engine=%s\nscanner-web=%s\nscanner-worker=%s\n' \
     "${SCN5_AUTHORITY_OUTGOING_IMAGE_ID}" "${SCN5_POLICY_OUTGOING_IMAGE_ID}" \
     "${SCN5_WEB_OUTGOING_IMAGE_ID}" "${SCN5_WORKER_OUTGOING_IMAGE_ID}"
   

   Before step 2, use apply_patch to create or completely replace devops/compose/docker-compose.local-estate-image-pins.override.yml, including when rollback is entered before step 10. Write the four printed outgoing IDs as literal image values for exactly authority, policy-engine, scanner-web, and scanner-worker; each service has only image: sha256:<64 lowercase hex> and pull_policy: never. Do not retain candidate IDs, use a tag/variable/placeholder, or delete the file after recovery. The executable assertion below proves a canonical fresh-helper render with no extra overlay selects these outgoing IDs before the window-specific rollback aliases are applied.

2. Render and assert the self-contained rollback overlay before creating anything. It pins both outgoing images and explicitly sets both the canonical key and each host’s legacy key to the same source connection. Docker Compose v5.3.1 retains the canonical key through this repository’s include chain for !reset null, !reset "", plain null, and !override null; rollback therefore does not rely on key removal or omission from canonical Compose. Do not replay the stale label chain (H6).

   scn5_assert_local_estate_pin \
     "${SCN5_AUTHORITY_OUTGOING_IMAGE_ID}" "${SCN5_POLICY_OUTGOING_IMAGE_ID}" \
     "${SCN5_WEB_OUTGOING_IMAGE_ID}" "${SCN5_WORKER_OUTGOING_IMAGE_ID}"
   SCN5_ROLLBACK_LOCAL_PIN_SHA="$(sha256sum "${SCN5_LOCAL_ESTATE_PIN}" | awk '{print $1}')"
   [[ "${SCN5_ROLLBACK_LOCAL_PIN_SHA}" =~ ^[0-9a-f]{64}$ ]]
   test "$(docker image inspect "${SCN5_AUTHORITY_ROLLBACK_IMAGE}" --format '{{.Id}}')" = \
     "${SCN5_AUTHORITY_OUTGOING_IMAGE_ID}"
   test "$(docker image inspect "${SCN5_POLICY_ROLLBACK_IMAGE}" --format '{{.Id}}')" = \
     "${SCN5_POLICY_OUTGOING_IMAGE_ID}"
   test "$(docker image inspect "${SCN5_WEB_ROLLBACK_IMAGE}" --format '{{.Id}}')" = \
     "${SCN5_WEB_OUTGOING_IMAGE_ID}"
   test "$(docker image inspect "${SCN5_WORKER_ROLLBACK_IMAGE}" --format '{{.Id}}')" = \
     "${SCN5_WORKER_OUTGOING_IMAGE_ID}"
   SCN5_ROLLBACK_FILES="${SCN5_ROLLBACK_OVERLAY}"
   COMPOSE_EXTRA_FILES="${SCN5_ROLLBACK_FILES}" \
     scn5_render_topology "${SCN5_ROOT}/topology/rollback-rendered-names-only.json"
   
   jq -e --arg web "${SCN5_WEB_ROLLBACK_IMAGE}" \
         --arg worker "${SCN5_WORKER_ROLLBACK_IMAGE}" \
         --arg webSource "${SCN5_WEB_ROLLBACK_BUNDLE_SOURCE}" \
         --arg workerSource "${SCN5_WORKER_ROLLBACK_BUNDLE_SOURCE}" '
     .services["scanner-web"].image == $web
     and .services["scanner-worker"].image == $worker
     and .services["scanner-web"].scannerBaseSource == $webSource
     and .services["scanner-worker"].scannerBaseSource == $workerSource
     and ([.services["scanner-web"].environmentKeys[]
           | select(test("STELLAOPS_POSTGRES_SCANNER_CONNECTION|SCANNER_SCANNER__STORAGE__DSN|ScannerStorage__Postgres__ConnectionString"))]
          == ["SCANNER_SCANNER__STORAGE__DSN", "STELLAOPS_POSTGRES_SCANNER_CONNECTION"])
     and ([.services["scanner-worker"].environmentKeys[]
           | select(test("STELLAOPS_POSTGRES_SCANNER_CONNECTION|SCANNER_SCANNER__STORAGE__DSN|ScannerStorage__Postgres__ConnectionString"))]
          == ["STELLAOPS_POSTGRES_SCANNER_CONNECTION", "ScannerStorage__Postgres__ConnectionString"])' \
     "${SCN5_ROOT}/topology/rollback-rendered-names-only.json" >/dev/null
   scn5_assert_rollback_connection_contract
   set +e
   COMPOSE_EXTRA_FILES="${SCN5_ROLLBACK_FILES}" \
     ./devops/compose/scripts/compose-cli.sh config --format json \
     | jq -e --arg authority "${SCN5_AUTHORITY_ROLLBACK_IMAGE}" \
         --arg policy "${SCN5_POLICY_ROLLBACK_IMAGE}" '
         .services.authority.image == $authority
         and .services["policy-engine"].image == $policy' >/dev/null
   SCN5_ROLLBACK_OWNER_RENDER_RC=("${PIPESTATUS[@]}")
   set -e
   test "${#SCN5_ROLLBACK_OWNER_RENDER_RC[@]}" -eq 2
   test "${SCN5_ROLLBACK_OWNER_RENDER_RC[0]}" -eq 0
   test "${SCN5_ROLLBACK_OWNER_RENDER_RC[1]}" -eq 0
   

   Verified by: the canonical no-extra render selects four literal outgoing image IDs that still resolve in the local daemon; render pipeline statuses are both zero; the temporary overlay images are window-specific aliases resolving to those same IDs; each service names its own immutable outgoing bundle snapshot; web has the canonical key plus SCANNER_SCANNER__STORAGE__DSN, worker has the canonical key plus ScannerStorage__Postgres__ConnectionString, and all four entries are the same non-empty source connection in process memory.

3. Create the rollback containers stopped, then inspect them while no writer can run. Keep the file list command-scoped so an unexported shell variable cannot silently omit the rollback overlay.

   scn5_assert_rollback_bundle_snapshots
   COMPOSE_EXTRA_FILES="${SCN5_ROLLBACK_FILES}" \
     ./devops/compose/scripts/compose-cli.sh recreate authority policy-engine \
     > "${SCN5_ROOT}/logs/rollback-owner-recreate.txt" 2>&1
   for owner_container in stellaops-authority stellaops-policy-engine; do
     for _ in $(seq 1 60); do
       test "$(docker inspect "${owner_container}" --format '{{.State.Health.Status}}')" = healthy && break
       sleep 2
     done
     test "$(docker inspect "${owner_container}" --format '{{.State.Running}} {{.State.Health.Status}}')" = \
       "true healthy"
     test "$(docker inspect "${owner_container}" --format '{{.Image}}')" = \
       "$(awk '{print $2}' "${SCN5_ROOT}/topology/${owner_container}.image.txt")"
     scn5_capture_env_names "${owner_container}" \
       "${SCN5_ROOT}/topology/${owner_container}.rollback-env-names.txt"
     scn5_capture_env_hmac "${owner_container}" \
       "${SCN5_ROOT}/topology/${owner_container}.rollback-env.hmac"
     scn5_capture_runtime_topology "${owner_container}" \
       "${SCN5_ROOT}/topology/${owner_container}.rollback-runtime-topology.json"
     cmp "${SCN5_ROOT}/topology/${owner_container}.env-names.txt" \
       "${SCN5_ROOT}/topology/${owner_container}.rollback-env-names.txt"
     cmp "${SCN5_ROOT}/topology/${owner_container}.env.hmac" \
       "${SCN5_ROOT}/topology/${owner_container}.rollback-env.hmac"
     cmp "${SCN5_ROOT}/topology/${owner_container}.runtime-topology.json" \
       "${SCN5_ROOT}/topology/${owner_container}.rollback-runtime-topology.json"
     docker exec "${owner_container}" cat /app/buildinfo.json \
       > "${SCN5_ROOT}/topology/${owner_container}.rollback-buildinfo.json"
     cmp "${SCN5_ROOT}/topology/${owner_container}.buildinfo.json" \
       "${SCN5_ROOT}/topology/${owner_container}.rollback-buildinfo.json"
   done
   for rollback_contract in \
       "stellaops-scanner-web|scanner-web|${SCN5_WEB_ROLLBACK_IMAGE}" \
       "stellaops-scanner-worker|scanner-worker|${SCN5_WORKER_ROLLBACK_IMAGE}"; do
     IFS='|' read -r rollback_container rollback_service rollback_image \
       <<< "${rollback_contract}"
     rollback_expected_names="${SCN5_ROOT}/topology/${rollback_container}.rollback-expected-env-names.txt"
     rollback_expected_hmac="${SCN5_ROOT}/topology/${rollback_container}.rollback-expected-env.hmac"
     rollback_expected_unchanged_hmac="${SCN5_ROOT}/topology/${rollback_container}.rollback-expected-unchanged-env.hmac"
     cp "${SCN5_ROOT}/topology/${rollback_container}.env-names.txt" \
       "${rollback_expected_names}"
     printf '%s\n' STELLAOPS_POSTGRES_SCANNER_CONNECTION >> "${rollback_expected_names}"
     LC_ALL=C sort -u -o "${rollback_expected_names}" "${rollback_expected_names}"
     COMPOSE_EXTRA_FILES="${SCN5_ROLLBACK_FILES}" \
       scn5_capture_rendered_env_hmac "${rollback_service}" "${rollback_image}" \
         "${rollback_expected_hmac}"
     COMPOSE_EXTRA_FILES="${SCN5_ROLLBACK_FILES}" \
       scn5_capture_rendered_env_hmac "${rollback_service}" "${rollback_image}" \
         "${rollback_expected_unchanged_hmac}" "${SCN5_ENV_DELTA_EXCLUDE}"
     cmp "${SCN5_ROOT}/topology/${rollback_container}.unchanged-env.hmac" \
       "${rollback_expected_unchanged_hmac}"
   done
   set +e
   COMPOSE_EXTRA_FILES="${SCN5_ROLLBACK_FILES}" \
     ./devops/compose/scripts/compose-cli.sh raw -- \
       up --no-start --no-deps --force-recreate scanner-web scanner-worker \
       > "${SCN5_ROOT}/logs/rollback-recreate-stopped.txt" 2>&1
   SCN5_ROLLBACK_RECREATE_RC=$?
   set -e
   sed -n '1,240p' "${SCN5_ROOT}/logs/rollback-recreate-stopped.txt"
   test "${SCN5_ROLLBACK_RECREATE_RC}" -eq 0
   
   test "$(docker inspect stellaops-scanner-web --format '{{.State.Running}}')" = false
   test "$(docker inspect stellaops-scanner-worker --format '{{.State.Running}}')" = false
   test "$(docker inspect stellaops-scanner-web --format '{{.Image}}')" = \
     "$(awk '{print $2}' "${SCN5_ROOT}/topology/stellaops-scanner-web.image.txt")"
   test "$(docker inspect stellaops-scanner-worker --format '{{.Image}}')" = \
     "$(awk '{print $2}' "${SCN5_ROOT}/topology/stellaops-scanner-worker.image.txt")"
   
   for c in stellaops-scanner-web stellaops-scanner-worker; do
     scn5_capture_env_names "$c" \
       "${SCN5_ROOT}/topology/${c}.rollback-env-names.txt"
     cmp "${SCN5_ROOT}/topology/${c}.rollback-expected-env-names.txt" \
         "${SCN5_ROOT}/topology/${c}.rollback-env-names.txt"
     scn5_capture_env_hmac "$c" \
       "${SCN5_ROOT}/topology/${c}.rollback-env.hmac"
     cmp "${SCN5_ROOT}/topology/${c}.rollback-expected-env.hmac" \
         "${SCN5_ROOT}/topology/${c}.rollback-env.hmac"
     scn5_capture_env_hmac "$c" \
       "${SCN5_ROOT}/topology/${c}.rollback-unchanged-env.hmac" \
       "${SCN5_ENV_DELTA_EXCLUDE}"
     cmp "${SCN5_ROOT}/topology/${c}.unchanged-env.hmac" \
         "${SCN5_ROOT}/topology/${c}.rollback-unchanged-env.hmac"
     rollback_topology="${SCN5_ROOT}/topology/${c}.rollback-runtime-topology.json"
     expected_topology="${SCN5_ROOT}/topology/${c}.rollback-expected-runtime-topology.json"
     scn5_capture_runtime_topology "$c" "${rollback_topology}"
     rollback_bundle_source="$(scn5_bundle_mount_source "${rollback_topology}")"
     case "$c" in
       stellaops-scanner-web) expected_bundle_source="${SCN5_WEB_ROLLBACK_BUNDLE_SOURCE}" ;;
       stellaops-scanner-worker) expected_bundle_source="${SCN5_WORKER_ROLLBACK_BUNDLE_SOURCE}" ;;
       *) exit 1 ;;
     esac
     test "$(cd -- "${rollback_bundle_source}" && pwd -P)" = \
       "$(cd -- "${expected_bundle_source}" && pwd -P)"
     jq --arg source "${rollback_bundle_source}" '
       .mounts |= map(if .destination == "/app/plugins/scanner/base"
         then .source = $source else . end)' \
       "${SCN5_ROOT}/topology/${c}.runtime-topology.json" > "${expected_topology}"
     cmp "${expected_topology}" "${rollback_topology}"
   done
   

   Verified by: each old-image service is created with its own immutable outgoing bundle snapshot; both snapshots have exact directory sets and passing bundle checksums before Compose creates either container. Primary main remains at the candidate descendant. Both processes are stopped, both .Image IDs equal the exact pre-window IDs, and the complete environment-name inventory is exactly the pre-window set plus the canonical connection key. Each full keyed value HMAC matches the pre-create rollback render, while the excluded-name HMAC proves every unrelated pre-window value is unchanged. The sanitized process/health/privilege/resource/mount/port/network topology projections differ only in the named /app/plugins/scanner/base source, which resolves to that service’s exact snapshot. Dynamic IDs, IPs, Compose labels, and environment values are intentionally absent from that projection; the image-ID and environment-HMAC checks cover the two material fields excluded from it.

4. Do not clean the target. Leaving stellaops_scanner loaded costs nothing (no writer points at it once step 2 is undone) and preserves the evidence of what failed. If the window will be retried, the retry needs an empty target: recreate it at exact migration 001, then restart at step 3 and repeat the downstream copy gates rather than truncating, so the ledger and partition set are minted rather than patched.

5. If the source was damaged, stop. Every source access in this procedure is read-only, so damage is an inversion. Verify the full dump’s checksum, keep both writers stopped, and invoke a separately reviewed destructive recovery procedure that recreates an explicit target before restore. Do not restore over either live database from this runbook.

6. Validate the stopped connection target, then deliberately reopen source intake. This is the rollback unfreeze point. Before starting anything, parse only the database name from each stopped container’s connection value; never print or persist the DSN. Freeze the failed target so a later comparison proves the restored old processes did not write there.

   scn5_container_env_value() { # $1=container, $2=environment key
     local container="$1" key="$2"
     docker inspect "${container}" --format '{{range .Config.Env}}{{println .}}{{end}}' \
       | awk -v prefix="${key}=" 'index($0, prefix) == 1 { sub(prefix, ""); print; exit }'
   }
   
   scn5_connection_database() { # $1=connection value
     printf '%s' "$1" | tr ';' '\n' \
       | awk -F= 'tolower($1) == "database" { print $2; exit }'
   }
   
   SCN5_ROLLBACK_WEB_CANONICAL="$(scn5_container_env_value \
     stellaops-scanner-web STELLAOPS_POSTGRES_SCANNER_CONNECTION)"
   SCN5_ROLLBACK_WEB_LEGACY="$(scn5_container_env_value \
     stellaops-scanner-web SCANNER_SCANNER__STORAGE__DSN)"
   SCN5_ROLLBACK_WORKER_CANONICAL="$(scn5_container_env_value \
     stellaops-scanner-worker STELLAOPS_POSTGRES_SCANNER_CONNECTION)"
   SCN5_ROLLBACK_WORKER_LEGACY="$(scn5_container_env_value \
     stellaops-scanner-worker ScannerStorage__Postgres__ConnectionString)"
   test -n "${SCN5_ROLLBACK_WEB_CANONICAL}"
   test "${SCN5_ROLLBACK_WEB_CANONICAL}" = "${SCN5_ROLLBACK_WEB_LEGACY}"
   test "${SCN5_ROLLBACK_WORKER_CANONICAL}" = "${SCN5_ROLLBACK_WORKER_LEGACY}"
   test "${SCN5_ROLLBACK_WEB_CANONICAL}" = "${SCN5_ROLLBACK_WORKER_CANONICAL}"
   test "$(scn5_connection_database "${SCN5_ROLLBACK_WEB_CANONICAL}")" = \
     "${SCN5_SOURCE_DB}"
   unset SCN5_ROLLBACK_WEB_CANONICAL SCN5_ROLLBACK_WEB_LEGACY \
     SCN5_ROLLBACK_WORKER_CANONICAL SCN5_ROLLBACK_WORKER_LEGACY
   scn5_fingerprint "${SCN5_TARGET_DB}" \
     "${SCN5_ROOT}/parity/target-before-rollback-unfreeze.tsv"
   
   # There is deliberately no "still fenced" claim after this command. The old
   # web image has no intake-fence implementation; starting it reopens source intake.
   docker start stellaops-scanner-web stellaops-scanner-worker \
     > "${SCN5_ROOT}/logs/rollback-start.txt"
   for c in stellaops-scanner-web stellaops-scanner-worker; do
     for _ in $(seq 1 60); do
       test "$(docker inspect "$c" --format '{{.State.Health.Status}}')" = healthy && break
       sleep 2
     done
     test "$(docker inspect "$c" --format '{{.State.Running}} {{.State.Health.Status}}')" = \
       "true healthy"
   done
   
   # Read one existing succeeded source scan through the restored public API. This
   # is non-mutating and remains valid even if normal intake arrives after unfreeze.
   : "${STELLAOPS_BACKEND_URL:?Set the accepted live backend URL}"
   : "${STELLAOPS_RELEASE_DISPATCH_CLIENT_SECRET:?Load by name from the ignored operator secret source}"
   set +x
   test -s "${SCN5_REPO_ROOT}/devops/compose/combined-ca-bundle.crt"
   SCN5_ROLLBACK_CURL=(
     curl.exe --silent --show-error --ssl-no-revoke
     --cacert "${SCN5_REPO_ROOT}/devops/compose/combined-ca-bundle.crt"
     --connect-timeout 5 --max-time 30
   )
   SCN5_ROLLBACK_READ_SCAN_ID="$(docker exec "${SCN5_PG_CONTAINER}" psql -XqAt \
     -U "${SCN5_PG_OPERATOR}" -d "${SCN5_SOURCE_DB}" -v tenant="${SCN5_TENANT}" -c \
     "SELECT scan_id FROM scanner.scan_runtime_state
       WHERE tenant_id = :'tenant' AND lower(status) = 'succeeded'
       ORDER BY updated_at DESC, scan_id DESC LIMIT 1;")"
   test -n "${SCN5_ROLLBACK_READ_SCAN_ID}"
   
   # Stage 13 normally discovers the status route from the running aggregate.
   # A failure before Stage 13 leaves no shell variable, so discover it here
   # without depending on any function that Stage 13 might not have defined.
   SCN5_ROLLBACK_STATUS_PATH_TEMPLATE="${SCN5_STATUS_PATH_TEMPLATE:-}"
   if test -z "${SCN5_ROLLBACK_STATUS_PATH_TEMPLATE}"; then
     "${SCN5_ROLLBACK_CURL[@]}" --fail-with-body \
       "${STELLAOPS_BACKEND_URL%/}/api/openapi/aggregate" \
       --output "${SCN5_ROOT}/topology/rollback-live-gateway-openapi-aggregate.json"
     SCN5_ROLLBACK_STATUS_PATH_TEMPLATE="$(jq -er '
       [.services[]? | select(.service == "scanner" and .status == "ok")]
       | if length == 1 and (.[0].spec | type) == "object" then .[0].spec
         else error("expected exactly one healthy Scanner OpenAPI document")
         end
       | [.paths | keys[] | select(endswith("/api/v1/scans/{scanId}"))]
       | if length == 1 then .[0]
         else error("expected exactly one live Scanner status route")
         end
     ' "${SCN5_ROOT}/topology/rollback-live-gateway-openapi-aggregate.json")"
   fi
   [[ "${SCN5_ROLLBACK_STATUS_PATH_TEMPLATE}" == /* ]]
   [[ "${SCN5_ROLLBACK_STATUS_PATH_TEMPLATE}" == *'{scanId}'* ]]
   SCN5_ROLLBACK_STATUS_PATH="${SCN5_ROLLBACK_STATUS_PATH_TEMPLATE/\{scanId\}/${SCN5_ROLLBACK_READ_SCAN_ID}}"
   [[ "${SCN5_ROLLBACK_STATUS_PATH}" != *'{'* ]]
   [[ "${SCN5_ROLLBACK_STATUS_PATH}" != *'}'* ]]
   
   SCN5_ROLLBACK_TOKEN_RAW="$(
     printf '%s' "${STELLAOPS_RELEASE_DISPATCH_CLIENT_SECRET}" \
       | "${SCN5_ROLLBACK_CURL[@]}" --request POST \
           --header 'Content-Type: application/x-www-form-urlencoded' \
           --data-urlencode 'grant_type=client_credentials' \
           --data-urlencode 'client_id=stellaops-release-dispatch' \
           --data-urlencode 'scope=scanner:read' \
           --data-urlencode "tenant=${SCN5_TENANT}" \
           --data-urlencode 'client_secret@-' \
           --write-out $'\n%{http_code}' \
           "${STELLAOPS_BACKEND_URL%/}/connect/token"
   )"
   test "${SCN5_ROLLBACK_TOKEN_RAW##*$'\n'}" = 200
   SCN5_ROLLBACK_READ_TOKEN="$(
     printf '%s' "${SCN5_ROLLBACK_TOKEN_RAW%$'\n'*}" \
       | jq -er '.access_token | select(type == "string" and length > 20)'
   )"
   unset SCN5_ROLLBACK_TOKEN_RAW
   
   set +e
   SCN5_ROLLBACK_READ_HTTP="$(
     printf 'Authorization: Bearer %s\n' "${SCN5_ROLLBACK_READ_TOKEN}" \
       | "${SCN5_ROLLBACK_CURL[@]}" \
           --header @- \
           --header "X-StellaOps-TenantId: ${SCN5_TENANT}" \
            --header 'Accept: application/json' \
            --output "${SCN5_ROOT}/logs/rollback-source-existing-scan.json" \
            --write-out '%{http_code}' \
            "${STELLAOPS_BACKEND_URL%/}${SCN5_ROLLBACK_STATUS_PATH}"
   )"
   SCN5_ROLLBACK_READ_RC=$?
   set -e
   test "${SCN5_ROLLBACK_READ_RC}" -eq 0
   test "${SCN5_ROLLBACK_READ_HTTP}" = 200
   jq -e --arg scan "${SCN5_ROLLBACK_READ_SCAN_ID}" '
     .scanId == $scan and (.status | ascii_downcase) == "succeeded"
   ' "${SCN5_ROOT}/logs/rollback-source-existing-scan.json" >/dev/null
   
   scn5_fingerprint "${SCN5_TARGET_DB}" \
     "${SCN5_ROOT}/parity/target-after-rollback-unfreeze.tsv"
   cmp "${SCN5_ROOT}/parity/target-before-rollback-unfreeze.tsv" \
       "${SCN5_ROOT}/parity/target-after-rollback-unfreeze.tsv"
   test "$(sha256sum "${SCN5_LOCAL_ESTATE_PIN}" | awk '{print $1}')" = \
     "${SCN5_ROLLBACK_LOCAL_PIN_SHA}"
   scn5_assert_local_estate_pin \
     "${SCN5_AUTHORITY_OUTGOING_IMAGE_ID}" "${SCN5_POLICY_OUTGOING_IMAGE_ID}" \
     "${SCN5_WEB_OUTGOING_IMAGE_ID}" "${SCN5_WORKER_OUTGOING_IMAGE_ID}"
   SCN5_ROLLBACK_CLEANUP_WORKTREES=("${SCN5_INVENTORY_WORKTREE}")
   if test -n "${SCN5_BUNDLE_BUILD_ROOT:-}"; then
     SCN5_ROLLBACK_CLEANUP_WORKTREES+=("${SCN5_BUNDLE_BUILD_ROOT}")
   fi
   for cleanup_worktree in "${SCN5_ROLLBACK_CLEANUP_WORKTREES[@]}"; do
     case "${cleanup_worktree}" in
       "${SCN5_REPO_ROOT}"/tmp/worktrees/scn5-*|"${SCN5_REPO_ROOT}"/tmp/worktrees/scn-live-build-*) ;;
       *) exit 1 ;;
     esac
     git -C "${SCN5_REPO_ROOT}" worktree remove --force "${cleanup_worktree}"
     test ! -e "${cleanup_worktree}"
   done
   unset SCN5_ROLLBACK_CLEANUP_WORKTREES
   test "${SCN5_CANDIDATE_ENV}" = "${SCN5_REPO_ROOT}/tmp/scn5-bundle-candidate.env"
   rm -f -- "${SCN5_CANDIDATE_ENV}"
   # Keep ${SCN5_WEB_ROLLBACK_BUNDLE_SOURCE} and
   # ${SCN5_WORKER_ROLLBACK_BUNDLE_SOURCE}: the restored containers mount them.
   # Keep ${SCN5_LOCAL_ESTATE_PIN}: it is the durable canonical rollback image-ID set.
   unset SCN5_ROLLBACK_READ_TOKEN \
     STELLAOPS_RELEASE_DISPATCH_CLIENT_SECRET \
     SCN5_TOPOLOGY_HMAC_KEY SCANNER_POSTGRES_PASSWORD
   

   Verified by: while stopped, both outgoing hosts carry canonical and legacy connection keys that are byte-identical, non-empty, mutually equal, and name only stellaops_platform; after the deliberate unfreeze, both old images are healthy; an authenticated non-mutating read returns an existing succeeded source scan; the failed target fingerprint is unchanged; and the retained no-extra canonical pin render still selects the four outgoing image IDs. No claim is made that intake stayed fenced after the docker start command.

If the source fingerprint changes unexpectedly at any point before rollback completes, stop both writers and prohibit automatic rollback. If any real target traffic was admitted after step 15, capture both fingerprints and require a reviewed reverse-delta reconciliation or a forward fix.

UNFREEZE POINT

The freeze lifts only when either:

• the full window succeeded — steps 9–14 are green, receipts recorded in the sprint, and the operator deliberately opens external intake; from that moment this automatic rollback is no longer valid; or
• rollback steps 1–5 are green and step 6 deliberately starts the old source-bound images; that start is the rollback external-intake open point, followed by the authenticated source read.

Until one of those holds, both Scanner writers stay stopped. A partially repointed estate with a live writer is the one state this procedure exists to make impossible.