refactor(migration): drop cross-account read grants (slack-bot decommissioned); harden backfill + deploy role

seahaven-slack-bot was decommissioned 2026-07-23 (stack DELETE_IN_PROGRESS,
consumer Lambdas gone); its successor sh-mcp is undeployed and uses
same-account DynamoDB access. So no live consumer reads these tables
cross-account. Per Adam's call, drop the cross-account grants entirely and
re-add correctly-scoped ones if/when sh-mcp deploys to a different account.

- Remove the four table resource policies + make_slack_bot_read_policy helper
  + its constants (cdk/common.py, po_stack.py, wo_stack.py) and the helper's
  unit test. Both stacks synth with zero table ResourcePolicy.
- scripts/migrate_tables.py hardening (fixes from the sh-security-review
  fan-out on the destructive backfill tool):
  * validate --cutoff strictly (parse ISO-8601, require aware UTC, re-emit
    canonical second-precision form) so a malformed cutoff can't silently
    copy dual-window rows or drop history;
  * reject `copy --all` up front (must run tables individually, in order,
    with the stream-drain wait) instead of writing three tables then erroring;
  * truncate backup gate now also checks recency (<1h) and TableId, not just
    status+name;
  * verify requires --cutoff whenever a cutoff table is in scope (else it
    false-flags dual-window rows as MISSING);
  * sticky-cancel is now PREVENTED copy-side (a non-Cancelled source item
    never overwrites a dest-Cancelled PO), and the verify comment no longer
    overstates what its source-side scan covers;
  * spot-check all modes (truncate_load keys are verbatim, so key-existence
    is sound there too).
- Deploy role: scope cloudformation:DescribeStacks to this repo's stacks +
  CDKToolkit (was Resource:*, disclosed all tenant stacks in the shared prod
  account); add a drift check warning on unexpected role policies and drop the
  dead SMOKE_POLICY_NAME var; document the shared-account bootstrap-role
  accepted risk in the deploy-role README.

cdk/common.py

@@ -35,68 +35,6 @@ _DDB_ALARM_OPERATIONS = [
 # CloudWatch namespace for the log-derived sender-authentication metrics.
 _SENDER_AUTH_METRIC_NAMESPACE = "Seahaven/ProcurementIngest"
 
-# Deliberate account literal: after the seahaven-prod migration, the mgmt
-# account (328440206208) is the FOREIGN principal here - the seahaven-slack-bot
-# stays deployed there and reads four of these tables cross-account.
-_SLACK_BOT_READER_ACCOUNT = "328440206208"
-
-# Read-only data-plane actions the bot consumers use (wo-po-lookup, po-sync,
-# workorder-sync). DescribeTable is included because ARN-as-TableName SDK
-# clients call it for validation paths.
-_SLACK_BOT_READ_ACTIONS = [
-    "dynamodb:GetItem",
-    "dynamodb:BatchGetItem",
-    "dynamodb:Query",
-    "dynamodb:Scan",
-    "dynamodb:DescribeTable",
-]
-
-
-def make_slack_bot_read_policy(scope, table_name):
-    """Resource-based policy allowing the mgmt slack-bot roles to read a table.
-
-    Cross-account pair, resource-side half: the bot's identity policies (in the
-    seahaven-slack-bot repo) grant the same actions on this table's ARN, and
-    its SDK calls pass the full ARN as TableName. Allow-only, so it cannot
-    affect in-account callers. PrincipalArn is wildcarded by the bot's CFN
-    stack prefix because role-name hash suffixes rotate on replacement;
-    account + prefix + read-only actions bound the grant.
-
-    Accepted risk (GPT-4.1 cross-review 2026-07-23): a mgmt principal with
-    iam:CreateRole could mint a prefix-matching role, but that capability
-    already implies near-org-admin blast radius; exact-ARN pinning was
-    rejected because it breaks on CFN role replacement. Assumption: the bot
-    Lambdas call DynamoDB DIRECTLY under their execution roles - if a
-    consumer ever role-chains first, aws:PrincipalArn will not match and
-    reads fail closed.
-    """
-    stack = Stack.of(scope)
-    return iam.PolicyDocument(
-        statements=[
-            iam.PolicyStatement(
-                sid="AllowMgmtSlackBotRead",
-                effect=iam.Effect.ALLOW,
-                principals=[iam.AccountPrincipal(_SLACK_BOT_READER_ACCOUNT)],
-                actions=list(_SLACK_BOT_READ_ACTIONS),
-                resources=[
-                    stack.format_arn(
-                        service="dynamodb",
-                        resource="table",
-                        resource_name=table_name,
-                    )
-                ],
-                conditions={
-                    "ArnLike": {
-                        "aws:PrincipalArn": (
-                            f"arn:aws:iam::{_SLACK_BOT_READER_ACCOUNT}:role/"
-                            "seahaven-slack-bot-*"
-                        )
-                    }
-                },
-            )
-        ]
-    )
-
 
 def add_ddb_alarms(scope, id_prefix, table, alarm_name_prefix, alarm_topic):
     """Add throttle + system-error alarms for a DynamoDB table.

cdk/po_stack.py

@@ -73,7 +73,6 @@ class PoIngestStack(Stack):
             stream=dynamodb.StreamViewType.NEW_IMAGE,
             encryption=dynamodb.TableEncryption.CUSTOMER_MANAGED,
             encryption_key=dynamodb_cmk,
-            resource_policy=common.make_slack_bot_read_policy(self, "purchase-orders"),
         )
 
         # --- Anthropic API key secret removed (Bedrock migration) ---
@@ -460,7 +459,6 @@ class PoIngestStack(Stack):
             ),
             billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
             removal_policy=RemovalPolicy.RETAIN,
-            resource_policy=common.make_slack_bot_read_policy(self, "verified-sites"),
         )
         # by-state GSI removed 2026-06-03 (audit M-20): 0 reads in 30d against
         # 518 WCU of write amplification. Re-add if a state-level query path ships.

cdk/wo_stack.py

@@ -53,7 +53,6 @@ class WorkorderIngestStack(Stack):
             ),
             billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
             removal_policy=RemovalPolicy.RETAIN,
-            resource_policy=common.make_slack_bot_read_policy(self, "WorkOrders"),
         )
         # site-code-index and status-index GSIs removed 2026-06-03 (audit M-20):
         # 0 reads in 30d against ~50k WCU each of write amplification. Re-add if
@@ -73,9 +72,6 @@ class WorkorderIngestStack(Stack):
             ),
             billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
             removal_policy=RemovalPolicy.RETAIN,
-            resource_policy=common.make_slack_bot_read_policy(
-                self, "WorkOrderComments"
-            ),
         )
 
         # --- Anthropic API key secret removed (Bedrock migration) ---

infra/deploy-role/create-deploy-role.sh

@@ -34,7 +34,6 @@ set -euo pipefail
 PROFILE="seahaven-prod"
 ROLE_NAME="githubdeploy-procurement-ingest"
 POLICY_NAME="procurement-ingest-deploy"
-SMOKE_POLICY_NAME="smoke-invoke-lambda"
 ACCOUNT_ID="011934824531"
 SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
 TRUST_POLICY="file://${SCRIPT_DIR}/trust-policy.json"
@@ -69,6 +68,24 @@ aws --profile "${PROFILE}" iam put-role-policy \
   --policy-name "${POLICY_NAME}" \
   --policy-document "${PERMS_POLICY}"
 
+# Drift check: this script only reconciles the single inline policy above, so
+# any OTHER inline or attached policy on the role is an unreviewed grant that
+# diverges the live role from these artifacts. Warn loudly (do not auto-delete
+# -- an operator decides).
+echo "==> Checking for unexpected policies on the role..."
+EXTRA_INLINE="$(aws --profile "${PROFILE}" iam list-role-policies \
+  --role-name "${ROLE_NAME}" --query "PolicyNames[?@!='${POLICY_NAME}']" --output text)"
+EXTRA_ATTACHED="$(aws --profile "${PROFILE}" iam list-attached-role-policies \
+  --role-name "${ROLE_NAME}" --query 'AttachedPolicies[].PolicyName' --output text)"
+if [[ -n "${EXTRA_INLINE}" || -n "${EXTRA_ATTACHED}" ]]; then
+  echo "    WARNING: unreviewed policies present on ${ROLE_NAME} (not in these artifacts):" >&2
+  [[ -n "${EXTRA_INLINE}" ]] && echo "      inline:   ${EXTRA_INLINE}" >&2
+  [[ -n "${EXTRA_ATTACHED}" ]] && echo "      attached: ${EXTRA_ATTACHED}" >&2
+  echo "    Review and remove them (delete-role-policy / detach-role-policy) if unexpected." >&2
+else
+  echo "    OK - only ${POLICY_NAME} present."
+fi
+
 ROLE_ARN="$(aws --profile "${PROFILE}" iam get-role --role-name "${ROLE_NAME}" \
   --query Role.Arn --output text)"
 

infra/deploy-role/permissions-policy.json

@@ -16,7 +16,11 @@
             "Sid": "CdkDeployHealthCheck",
             "Effect": "Allow",
             "Action": "cloudformation:DescribeStacks",
-            "Resource": "*",
+            "Resource": [
+                "arn:aws:cloudformation:us-east-1:011934824531:stack/po-ingest/*",
+                "arn:aws:cloudformation:us-east-1:011934824531:stack/WorkorderIngestStack/*",
+                "arn:aws:cloudformation:us-east-1:011934824531:stack/CDKToolkit/*"
+            ],
             "Condition": {
                 "StringEquals": {
                     "aws:RequestedRegion": "us-east-1"

scripts/migrate_tables.py

@@ -12,14 +12,18 @@ Per-table modes (encode the plan's dual-delivery-safe semantics):
   WorkOrders, purchase-orders   overwrite Put. Safe because during dual
                                 delivery the mgmt item is a strict SUPERSET of
                                 the prod item (both pipelines apply SET-only
-                                merges to the same inbound mail).
+                                merges to the same inbound mail). purchase-orders
+                                additionally PRESERVES a dest-Cancelled PO: a
+                                non-Cancelled source item never overwrites a
+                                prod row already Cancelled (sticky-cancel guard).
   WorkOrderComments             cutoff-filtered Put: only rows with
                                 attribute_not_exists(ingested_at) OR
                                 ingested_at < --cutoff (T_activate). The same
                                 email yields DIFFERENT comment_ids per account
                                 (S3-object-key hash suffix), so copying
                                 dual-window rows would create duplicate
-                                history. --cutoff is REQUIRED for this table.
+                                history. --cutoff is REQUIRED for this table on
+                                BOTH copy and verify.
   verified-sites,               truncate-and-load: delete ALL destination rows
   pending-site-review           first, then copy the source scan verbatim.
                                 Required because the purchase-orders backfill
@@ -28,10 +32,11 @@ Per-table modes (encode the plan's dual-delivery-safe semantics):
                                 poCount ADDs; mgmt values are ground truth.
                                 Take an on-demand backup of the dest table and
                                 wait for it to be AVAILABLE before running
-                                (the script refuses without --backup-arn).
+                                (the script refuses without a fresh --backup-arn
+                                of THIS table).
 
 Usage:
-    # copy (dry-run first, then --execute)
+    # copy (dry-run first, then --execute) -- run tables INDIVIDUALLY, in order
     python scripts/migrate_tables.py copy --table WorkOrders
     python scripts/migrate_tables.py copy --table WorkOrderComments \
         --cutoff 2026-07-24T02:00:00Z --execute
@@ -39,7 +44,11 @@ Usage:
         --backup-arn arn:aws:dynamodb:...:backup/... --execute
     # verify (count parity + spot checks; always read-only)
     python scripts/migrate_tables.py verify --table purchase-orders
-    python scripts/migrate_tables.py verify --all
+    python scripts/migrate_tables.py verify --all --cutoff 2026-07-24T02:00:00Z
+
+`copy --all` is deliberately rejected: the plan runs each table as a separate
+step with a stream-drain wait before the truncate-and-load tables, so a single
+sweep would either write in the wrong order or truncate before the drain.
 
 Profiles: --source-profile (default: "default", the mgmt SSO session) and
 --dest-profile (default: "seahaven-prod"). The script hard-verifies both
@@ -49,6 +58,7 @@ resolved account ids before doing anything.
 import argparse
 import json
 import sys
+from datetime import datetime, timedelta, timezone
 
 import boto3
 
@@ -56,6 +66,10 @@ SOURCE_ACCOUNT = "328440206208"
 DEST_ACCOUNT = "011934824531"
 REGION = "us-east-1"
 
+# A backup passed to a truncate-and-load must be fresh -- the whole point is a
+# just-taken restore point, not a week-old one that predates prod-native rows.
+BACKUP_MAX_AGE = timedelta(hours=1)
+
 TABLES = {
     "purchase-orders": {"mode": "overwrite", "keys": ["po_number"]},
     "WorkOrders": {"mode": "overwrite", "keys": ["work_order_id"]},
@@ -103,6 +117,30 @@ def scan_count(table, filter_kwargs=None):
         kwargs["ExclusiveStartKey"] = lek
 
 
+def normalize_cutoff(value):
+    """Parse an operator-supplied cutoff and re-emit the canonical form that
+    ``ingested_at`` is lexicographically comparable against.
+
+    ``ingested_at`` is ``datetime.now(timezone.utc).isoformat()`` -> a
+    zero-padded ``...+00:00`` string. A raw operator string ("2026-7-24...",
+    a space for the 'T', a fractional part) compares WRONG under DynamoDB's
+    byte-wise ``<`` and would silently copy dual-window rows or drop history.
+    So parse strictly, require an aware UTC instant, and re-emit second
+    precision ``+00:00`` with no fraction. A row stamped exactly at T_activate
+    then sorts >= the cutoff and is excluded -- the safe direction.
+    """
+    raw = value[:-1] + "+00:00" if value.endswith("Z") else value
+    try:
+        dt = datetime.fromisoformat(raw)
+    except ValueError:
+        sys.exit(
+            f"ERROR: --cutoff {value!r} is not ISO-8601 (e.g. 2026-07-24T02:00:00Z)."
+        )
+    if dt.tzinfo is None or dt.utcoffset() != timedelta(0):
+        sys.exit(f"ERROR: --cutoff {value!r} must be UTC (trailing 'Z' or '+00:00').")
+    return dt.replace(microsecond=0).isoformat()
+
+
 def comments_cutoff_filter(cutoff):
     return {
         "FilterExpression": ("attribute_not_exists(#ia) OR #ia < :cutoff"),
@@ -111,33 +149,57 @@ def comments_cutoff_filter(cutoff):
     }
 
 
-def normalize_cutoff(value):
+def _dest_cancelled_po_numbers(dst):
-    # ingested_at is datetime.now(timezone.utc).isoformat() -> "+00:00"
+    """po_numbers currently Cancelled in the destination purchase-orders table.
-    # offset form. Normalize a trailing Z so string comparison is apples
+
-    # to apples.
+    Read once before an overwrite copy so a stale non-Cancelled source item
-    return value.replace("Z", "+00:00")
+    can never revive a PO the prod pipeline already cancelled (sticky-cancel).
+    """
+    flt = {
+        "FilterExpression": "po_status = :c",
+        "ExpressionAttributeValues": {":c": "Cancelled"},
+        "ProjectionExpression": "po_number",
+    }
+    return {item["po_number"] for item in scan_items(dst, flt)}
 
 
-def _truncate_destination(name, cfg, dst, args):
+def _validate_truncate_backup(name, dst, args):
-    """Backup-gated destination truncate for the truncate-and-load tables."""
+    """Fail closed unless --backup-arn is an AVAILABLE, fresh backup of THIS
+    destination table incarnation (status + name + recency + TableId)."""
+    if not args.backup_arn:
+        sys.exit(
+            f"ERROR: {name} is truncate-and-load; pass --backup-arn of an "
+            "AVAILABLE on-demand backup of the DESTINATION table taken "
+            "just now (aws dynamodb create-backup / describe-backup)."
+        )
+    desc = dst.meta.client.describe_backup(BackupArn=args.backup_arn)
+    details = desc["BackupDescription"]["BackupDetails"]
+    src_table = desc["BackupDescription"]["SourceTableDetails"]
+    if details["BackupStatus"] != "AVAILABLE" or src_table["TableName"] != name:
+        sys.exit(
+            f"ERROR: backup {args.backup_arn} is {details['BackupStatus']} for "
+            f"table {src_table['TableName']!r}; need an AVAILABLE backup of "
+            f"{name!r}. Aborting before truncate."
+        )
+    live_id = dst.meta.client.describe_table(TableName=name)["Table"]["TableId"]
+    if src_table.get("TableId") != live_id:
+        sys.exit(
+            f"ERROR: backup {args.backup_arn} is of a DIFFERENT incarnation of "
+            f"{name!r} (TableId {src_table.get('TableId')} != live {live_id}); "
+            "a RETAIN+recreate cycle produces a same-named table. Aborting."
+        )
+    age = datetime.now(timezone.utc) - details["BackupCreationDateTime"]
+    if age > BACKUP_MAX_AGE:
+        sys.exit(
+            f"ERROR: backup {args.backup_arn} is {age} old (> {BACKUP_MAX_AGE}); "
+            "take a fresh backup immediately before truncating. Aborting."
+        )
+
+
+def _truncate_destination(name, dst, args, keys):
     if args.execute:
-        if not args.backup_arn:
+        _validate_truncate_backup(name, dst, args)
-            sys.exit(
+    dest_keys = [{k: item[k] for k in keys} for item in scan_items(dst)]
-                f"ERROR: {name} is truncate-and-load; pass --backup-arn of an "
-                "AVAILABLE on-demand backup of the DESTINATION table taken "
-                "just now (aws dynamodb create-backup / describe-backup)."
-            )
-        desc = dst.meta.client.describe_backup(BackupArn=args.backup_arn)
-        details = desc["BackupDescription"]["BackupDetails"]
-        src_table = desc["BackupDescription"]["SourceTableDetails"]
-        if details["BackupStatus"] != "AVAILABLE" or src_table["TableName"] != name:
-            sys.exit(
-                f"ERROR: backup {args.backup_arn} is "
-                f"{details['BackupStatus']} for table "
-                f"{src_table['TableName']!r}; need an AVAILABLE backup "
-                f"of {name!r}. Aborting before truncate."
-            )
-    dest_keys = [{k: item[k] for k in cfg["keys"]} for item in scan_items(dst)]
     print(f"[{name}] truncate: {len(dest_keys)} destination rows to delete")
     if args.execute:
         with dst.batch_writer() as batch:
@@ -154,24 +216,44 @@ def copy_table(name, cfg, args):
 
     filter_kwargs = None
     if cfg["mode"] == "cutoff":
-        if not args.cutoff:
+        # --cutoff presence is enforced up front in main(); normalize here.
-            sys.exit(f"ERROR: --cutoff (T_activate) is required for {name}.")
         filter_kwargs = comments_cutoff_filter(normalize_cutoff(args.cutoff))
 
     if cfg["mode"] == "truncate_load":
-        _truncate_destination(name, cfg, dst, args)
+        _truncate_destination(name, dst, args, cfg["keys"])
 
-    copied = 0
+    # Sticky-cancel guard: never let a stale non-Cancelled source item revive a
+    # PO already Cancelled in prod (the raw batch Put bypasses the handler's
+    # ConditionExpression that normally enforces this).
+    dest_cancelled = (
+        _dest_cancelled_po_numbers(dst) if name == "purchase-orders" else set()
+    )
+
+    copied = preserved = 0
     if args.execute:
         with dst.batch_writer() as batch:
             for item in scan_items(src, filter_kwargs):
+                if (
+                    item.get("po_number") in dest_cancelled
+                    and item.get("po_status") != "Cancelled"
+                ):
+                    preserved += 1
+                    continue
                 batch.put_item(Item=item)
                 copied += 1
     else:
-        for _ in scan_items(src, filter_kwargs):
+        for item in scan_items(src, filter_kwargs):
+            if (
+                item.get("po_number") in dest_cancelled
+                and item.get("po_status") != "Cancelled"
+            ):
+                preserved += 1
+                continue
             copied += 1
     verb = "copied" if args.execute else "would copy (DRY-RUN)"
     print(f"[{name}] {verb} {copied} items {SOURCE_ACCOUNT} -> {DEST_ACCOUNT}")
+    if preserved:
+        print(f"[{name}] preserved {preserved} dest-Cancelled PO(s) (not overwritten)")
 
 
 def verify_table(name, cfg, args):
@@ -179,15 +261,15 @@ def verify_table(name, cfg, args):
     dst = session_table(args.dest_profile, DEST_ACCOUNT, name)
 
     src_filter = None
-    if cfg["mode"] == "cutoff" and args.cutoff:
+    if cfg["mode"] == "cutoff":
+        # main() guarantees --cutoff is present when a cutoff table is in scope.
         src_filter = comments_cutoff_filter(normalize_cutoff(args.cutoff))
 
     src_count = scan_count(src, src_filter)
     dst_count = scan_count(dst)
     # Destination may legitimately EXCEED source-side-of-cutoff (prod-native
-    # dual-delivery rows) for cutoff tables, and match-or-exceed for overwrite
+    # dual-delivery rows) for cutoff/overwrite tables; exact parity is expected
-    # tables written after the copy; exact parity is expected only for
+    # only for truncate_load immediately after the load.
-    # truncate_load immediately after the load.
     relation = {
         "overwrite": dst_count >= src_count,
         "cutoff": dst_count >= src_count,
@@ -197,7 +279,10 @@ def verify_table(name, cfg, args):
     print(f"[{name}] source={src_count} dest={dst_count} -> {status}")
 
     ok = relation
-    if args.sample and cfg["mode"] != "truncate_load":
+    # Spot-check every mode. For cutoff tables the source scan is cutoff-filtered
+    # (so no dual-window rows are sampled); for truncate_load keys are copied
+    # verbatim, so key-existence is a sound check there too.
+    if args.sample:
         sampled = mismatched = 0
         for item in scan_items(src, src_filter):
             if sampled >= args.sample:
@@ -213,8 +298,10 @@ def verify_table(name, cfg, args):
 
     cancelled_check_limit = 25
     if name == "purchase-orders":
-        # Sticky-Cancelled spot check: every source-Cancelled PO must still be
+        # Propagation check: a PO Cancelled in mgmt must be Cancelled in prod.
-        # Cancelled in dest (the overwrite Put must not have raced a revival).
+        # (Revival in the OTHER direction -- a prod-Cancelled PO overwritten by
+        # a non-Cancelled mgmt item -- is prevented copy-side by the sticky-
+        # cancel guard, not detectable here after the fact.)
         cancelled_filter = {
             "FilterExpression": "po_status = :c",
             "ExpressionAttributeValues": {":c": "Cancelled"},
@@ -234,18 +321,31 @@ def verify_table(name, cfg, args):
     return ok
 
 
+def _require_cutoff_if_needed(names, args):
+    cutoff_tables = [n for n in names if TABLES[n]["mode"] == "cutoff"]
+    if cutoff_tables and not args.cutoff:
+        sys.exit(
+            f"ERROR: {', '.join(cutoff_tables)} is cutoff-mode; --cutoff "
+            "(T_activate) is required so dual-window rows are excluded."
+        )
+
+
 def main():
     ap = argparse.ArgumentParser(description=__doc__)
     sub = ap.add_subparsers(dest="cmd", required=True)
     for cmd in ("copy", "verify"):
         p = sub.add_parser(cmd)
         p.add_argument("--table", choices=sorted(TABLES))
-        p.add_argument("--all", action="store_true", help="all five tables")
+        p.add_argument(
+            "--all",
+            action="store_true",
+            help="all five tables (verify only; copy runs tables individually)",
+        )
         p.add_argument("--source-profile", default="default")
         p.add_argument("--dest-profile", default="seahaven-prod")
         p.add_argument(
             "--cutoff",
-            help="T_activate ISO-8601 UTC; required to copy WorkOrderComments",
+            help="T_activate ISO-8601 UTC; required for WorkOrderComments",
         )
         if cmd == "copy":
             p.add_argument("--execute", action="store_true")
@@ -264,27 +364,22 @@ def main():
     args = ap.parse_args()
     if bool(args.table) == bool(args.all):
         sys.exit("ERROR: pass exactly one of --table or --all.")
+
+    if args.cmd == "copy" and args.all:
+        # Validation precedes any side effect: --all would copy in the wrong
+        # order and truncate before the stream drain. Run tables individually.
+        sys.exit(
+            "ERROR: `copy --all` is not supported. Run each table individually "
+            "in plan order (WorkOrders, WorkOrderComments, purchase-orders, "
+            "wait for the site-extractor stream to drain, then verified-sites "
+            "and pending-site-review)."
+        )
+
     names = sorted(TABLES) if args.all else [args.table]
+    _require_cutoff_if_needed(names, args)
 
     if args.cmd == "copy":
-        # Plan-mandated order: WorkOrders, WorkOrderComments, purchase-orders
+        copy_table(args.table, TABLES[args.table], args)
-        # (stream replay), drain, then the two truncate-and-load tables. With
-        # --all we enforce it; single --table trusts the operator.
-        order = [
-            "WorkOrders",
-            "WorkOrderComments",
-            "purchase-orders",
-            "verified-sites",
-            "pending-site-review",
-        ]
-        for name in sorted(names, key=order.index):
-            if TABLES[name]["mode"] == "truncate_load" and args.all:
-                sys.exit(
-                    "ERROR: run verified-sites / pending-site-review "
-                    "individually AFTER the purchase-orders stream drain "
-                    "(ESM IteratorAge ~ 0), not as part of --all."
-                )
-            copy_table(name, TABLES[name], args)
     else:
         results = [verify_table(n, TABLES[n], args) for n in names]
         if not all(results):

tests/test_resource_policy_helper.py

@@ -1,146 +0,0 @@
-"""Unit tests for common.make_slack_bot_read_policy (prod-migration PR).
-
-Two layers, no full-app synth (PoIngestStack synth would trigger Docker
-bundling in CI):
-
-1. Synthesizes a minimal throwaway stack with one table using the helper and
-   asserts the exact ResourcePolicy statement shape (sid, read-only actions,
-   mgmt-account principal, PrincipalArn prefix condition, per-table ARN).
-2. Static pins that the four bot-read tables in the real stack files pass
-   resource_policy= and pending-site-review does NOT (no bot consumer).
-"""
-
-import importlib.util
-import re
-import sys
-from pathlib import Path
-
-import aws_cdk as cdk
-import pytest
-from aws_cdk import assertions
-from aws_cdk import aws_dynamodb as dynamodb
-
-REPO_ROOT = Path(__file__).resolve().parents[1]
-
-_spec = importlib.util.spec_from_file_location(
-    "cdk_common", REPO_ROOT / "cdk" / "common.py"
-)
-common = importlib.util.module_from_spec(_spec)
-sys.modules["cdk_common"] = common
-_spec.loader.exec_module(common)
-
-MGMT_ACCOUNT = "328440206208"
-PROD_ACCOUNT = "011934824531"
-
-
-def _synth_policy(table_name):
-    app = cdk.App()
-    stack = cdk.Stack(
-        app,
-        "T",
-        env=cdk.Environment(account=PROD_ACCOUNT, region="us-east-1"),
-    )
-    dynamodb.Table(
-        stack,
-        "Tbl",
-        table_name=table_name,
-        partition_key=dynamodb.Attribute(name="pk", type=dynamodb.AttributeType.STRING),
-        billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
-        resource_policy=common.make_slack_bot_read_policy(stack, table_name),
-    )
-    template = assertions.Template.from_stack(stack)
-    tables = template.find_resources("AWS::DynamoDB::Table")
-    (props,) = [r["Properties"] for r in tables.values()]
-    statements = props["ResourcePolicy"]["PolicyDocument"]["Statement"]
-    assert len(statements) == 1
-    return statements[0]
-
-
-def test_statement_shape_and_actions():
-    s = _synth_policy("purchase-orders")
-    assert s["Sid"] == "AllowMgmtSlackBotRead"
-    assert s["Effect"] == "Allow"
-    assert sorted(s["Action"]) == [
-        "dynamodb:BatchGetItem",
-        "dynamodb:DescribeTable",
-        "dynamodb:GetItem",
-        "dynamodb:Query",
-        "dynamodb:Scan",
-    ]
-
-
-def test_no_write_actions():
-    s = _synth_policy("purchase-orders")
-    writes = {
-        a
-        for a in s["Action"]
-        if re.search(r"Put|Update|Delete|Write|Create|Restore|Export|Import", a)
-    }
-    assert not writes
-
-
-def test_principal_is_mgmt_account_root():
-    s = _synth_policy("WorkOrders")
-    principal = s["Principal"]["AWS"]
-    # Token-joined root-principal ARN: {"Fn::Join": ["", ["arn:", {...}, ":iam::328440206208:root"]]}
-    assert MGMT_ACCOUNT in str(principal)
-    assert ":root" in str(principal)
-
-
-def test_principal_arn_condition_pins_bot_stack_prefix():
-    s = _synth_policy("WorkOrders")
-    assert s["Condition"]["ArnLike"]["aws:PrincipalArn"] == (
-        f"arn:aws:iam::{MGMT_ACCOUNT}:role/seahaven-slack-bot-*"
-    )
-
-
-@pytest.mark.parametrize("name", ["purchase-orders", "WorkOrderComments"])
-def test_resource_is_exactly_the_named_table_arn(name):
-    import json
-
-    s = _synth_policy(name)
-    # format_arn leaves the partition as {"Ref": "AWS::Partition"}, so the
-    # resource is an Fn::Join; assert on the deterministic joined tail.
-    dumped = json.dumps(s["Resource"])
-    assert f"dynamodb:us-east-1:{PROD_ACCOUNT}:table/{name}" in dumped
-    # Exactly one resource entry (no index wildcard, no extra ARNs).
-    assert not isinstance(s["Resource"], list) or len(s["Resource"]) == 1
-
-
-# --- static pins against the real stack files ---
-
-_PO_STACK = (REPO_ROOT / "cdk" / "po_stack.py").read_text()
-_WO_STACK = (REPO_ROOT / "cdk" / "wo_stack.py").read_text()
-
-
-def _table_blocks(source):
-    """Map table_name -> the dynamodb.Table(...) call block containing it."""
-    blocks = {}
-    for m in re.finditer(r"dynamodb\.Table\(", source):
-        depth, i = 0, m.end() - 1
-        while i < len(source):
-            depth += {"(": 1, ")": -1}.get(source[i], 0)
-            if depth == 0:
-                break
-            i += 1
-        block = source[m.start() : i + 1]
-        nm = re.search(r'table_name="([^"]+)"', block)
-        if nm:
-            blocks[nm.group(1)] = block
-    return blocks
-
-
-def test_bot_read_tables_carry_resource_policy():
-    blocks = {**_table_blocks(_PO_STACK), **_table_blocks(_WO_STACK)}
-    for name in [
-        "purchase-orders",
-        "verified-sites",
-        "WorkOrders",
-        "WorkOrderComments",
-    ]:
-        assert "make_slack_bot_read_policy" in blocks[name], name
-
-
-def test_pending_site_review_has_no_resource_policy():
-    blocks = _table_blocks(_PO_STACK)
-    assert "resource_policy" not in blocks["pending-site-review"]