fix(cdk): explicit Lambda LogGroups replace log_retention (INFRA-114) (#126)

First seahaven-prod deploy failed CREATE on the sender-auth MetricFilter:
it imported /aws/lambda/<fn> by name, which pre-existed in mgmt but not in
a fresh account. All 5 functions now get an explicit logs.LogGroup
(TWO_MONTHS, RETAIN) via common.make_function_log_group, and the metric
filter takes the construct so CFN orders it after the group exists.

Also removes the deprecated LogRetention custom resource and its
wildcard logs:PutRetentionPolicy role (CKV_AWS_111). Function roles keep
AWSLambdaBasicExecutionRole (verified in the synthesized template), so
log-write permissions are unchanged; cross-review's grant_write FIX was
a false positive on that basis. Supersedes PR #84, which hardcoded the
mgmt logs-CMK ARN and predates the common.py refactor.

mgmt collision note: these CREATEs would collide with the pre-existing
groups in mgmt; acceptable because the deploy secret now targets prod
and mgmt is frozen pending decommission.

.github/workflows/ci.yaml

@@ -8,7 +8,7 @@ permissions:
 
 jobs:
   ci:
-    uses: Sea-Haven-Industries/.github/.github/workflows/ci-python-sam.yaml@fd60e4c9041784f666ac0fdefb9bec3c7fbf5143 # main
+    uses: Sea-Haven-Industries/.github/.github/workflows/ci-python-sam.yaml@f71002a9ed2938730b683249b28059c92a081af6
     with:
       source-dirs: "lambdas cdk tests scripts"
       run-tests: true

README.md

@@ -88,7 +88,7 @@ Amazon APM work order emails (from Hexagon EAM / HxGN SmartCloud) are received a
 
 ## Architecture
 
-**IaC:** AWS CDK (Python), two stacks in one app, region `us-east-1`. The `cdk.Environment` is deliberately **account-agnostic** (region-only, no `account=`): the stacks deploy to whichever account the deploy credentials target (`328440206208` today), and every account-derived template value — bucket names, `Lambda::Permission` source account, the site-alerts SNS action ARN, the Bedrock ARN below — renders as the CloudFormation `AWS::AccountId` pseudo-parameter rather than a literal. Pinning `account=` was evaluated in Phase 4 and rejected: it would resolve those tokens to literals, and against the deployed (account-agnostic) templates CloudFormation flags the `RemovalPolicy.RETAIN` email buckets as requiring replacement — a data-loss risk — for no functional gain.
+**IaC:** AWS CDK (Python), two stacks in one app, region `us-east-1`. The `cdk.Environment` is deliberately **account-agnostic** (region-only, no `account=`): the stacks deploy to whichever account the deploy credentials target (`011934824531` seahaven-prod since the 2026-07 account migration; formerly mgmt `328440206208`), and every account-derived template value — bucket names, `Lambda::Permission` source account, the site-alerts SNS action ARN, the Bedrock ARN below — renders as the CloudFormation `AWS::AccountId` pseudo-parameter rather than a literal. Pinning `account=` was evaluated in Phase 4 and rejected: it would resolve those tokens to literals, and against the deployed (account-agnostic) templates CloudFormation flags the `RemovalPolicy.RETAIN` email buckets as requiring replacement — a data-loss risk — for no functional gain.
 
 All Lambdas: Python 3.12, ARM64, 60-day log retention.
 
@@ -313,6 +313,20 @@ Branch protection on `main` — all changes through PR.
 
 ## Setup
 
+**Account prerequisites** — the stacks import four dependencies by name, so all
+of these must exist in the target account BEFORE the first deploy (in
+seahaven-prod they are provisioned by the seahaven-org-baseline repo and the
+migration Phase 0 runbook):
+
+- SES receipt rule set `INBOUND_MAIL` (may be inactive; the stacks attach
+  their receipt rules to it) + a verified `int.seahaven.com` domain identity
+- SNS topic `site-alerts` (with the `alias/seahaven-alarm-topics` CMK)
+- SSM param `/seahaven/dynamodb/cmk-arn` -> KMS `alias/seahaven-dynamodb`
+- Secrets Manager secret `procurement-ingest/web-ui-auth-token` (step 3)
+- OIDC deploy role `githubdeploy-procurement-ingest` (see `infra/deploy-role/`)
+  with the `smoke-invoke-lambda` policy, or the post-deploy smoke gate fails
+  closed
+
 1. Bootstrap CDK: `cdk bootstrap aws://{AccountId}/us-east-1`
 2. Ensure the Bedrock inference profile `us.anthropic.claude-haiku-4-5-20251001-v1:0` is enabled in `us-east-1` (it is; the CDK grants cover cross-region routing to `us-east-2`/`us-west-2`). No API key or secret to set — the processors authenticate to Bedrock via their IAM roles.
 3. Create the web UI auth-gate shared secret. This secret is **imported by name**

cdk/common.py

@@ -73,7 +73,36 @@ def add_ddb_alarms(scope, id_prefix, table, alarm_name_prefix, alarm_topic):
     ).add_alarm_action(cw_actions.SnsAction(alarm_topic))
 
 
-def add_sender_auth_rejected_alarm(scope, id_prefix, function_name, alarm_topic):
+def make_function_log_group(scope, id_prefix, function_name):
+    """Explicit log group for a Lambda, replacing the deprecated
+    ``log_retention`` prop (INFRA-114).
+
+    Making the group a real stack resource (a) drops the LogRetention custom
+    resource whose role carried wildcard ``logs:PutRetentionPolicy`` (checkov
+    CKV_AWS_111), and (b) makes the group an orderable CFN dependency:
+    consumers like the sender-auth metric filter now deploy AFTER the group
+    exists. The previous by-name import raced group creation in a fresh
+    account and failed the first seahaven-prod deploy (the mgmt account
+    masked this because its groups predated the filter).
+
+    RETAIN matches the repo convention for stateful resources and mirrors the
+    old behavior (LogRetention never deleted groups on stack delete). NOTE:
+    in an account where ``/aws/lambda/<fn>`` already exists out-of-band
+    (mgmt), deploying this CREATE would collide -- acceptable because mgmt is
+    frozen post-migration and never redeployed from main.
+    """
+    return logs.LogGroup(
+        scope,
+        f"{id_prefix}LogGroup",
+        log_group_name=f"/aws/lambda/{function_name}",
+        retention=logs.RetentionDays.TWO_MONTHS,
+        removal_policy=RemovalPolicy.RETAIN,
+    )
+
+
+def add_sender_auth_rejected_alarm(
+    scope, id_prefix, function_name, alarm_topic, log_group
+):
     """Metric-filter + alarm on ``sender_auth_rejected`` warnings (INFRA-107).
 
     A rejected inbound email is skipped without erroring the invocation, so it
@@ -84,21 +113,18 @@ def add_sender_auth_rejected_alarm(scope, id_prefix, function_name, alarm_topic)
     mail while the pipeline reports healthy.
 
     ALARM-only SnsAction to site-alerts; no OK action. The metric filter reads
-    the function's own log group (imported by the deterministic
-    ``/aws/lambda/<fn>`` name, created by the function's log_retention). A plain
-    substring pattern is used because Lambda prefixes each line with its own
-    level/timestamp/request-id, so the JSON payload is not a standalone JSON
-    log event a `{$.event=...}` pattern could match.
+    the function's own log group, passed in as the EXPLICIT LogGroup construct
+    (from ``make_function_log_group``) so CFN orders the filter after the
+    group exists -- a by-name import here failed the first fresh-account
+    deploy. A plain substring pattern is used because Lambda prefixes each
+    line with its own level/timestamp/request-id, so the JSON payload is not
+    a standalone JSON log event a `{$.event=...}` pattern could match.
     """
     metric_name = f"{function_name}-sender-auth-rejected"
     logs.MetricFilter(
         scope,
         f"{id_prefix}SenderAuthRejectedFilter",
-        log_group=logs.LogGroup.from_log_group_name(
-            scope,
-            f"{id_prefix}LogGroup",
-            f"/aws/lambda/{function_name}",
-        ),
+        log_group=log_group,
         filter_pattern=logs.FilterPattern.literal('"sender_auth_rejected"'),
         metric_namespace=_SENDER_AUTH_METRIC_NAMESPACE,
         metric_name=metric_name,

cdk/po_stack.py

@@ -11,7 +11,6 @@ from aws_cdk import (
     aws_kms as kms,
     aws_lambda as lambda_,
     aws_lambda_event_sources as lambda_event_sources,
-    aws_logs as logs,
     aws_s3 as s3,
     aws_s3_notifications as s3n,
     aws_ses as ses,
@@ -91,6 +90,9 @@ class PoIngestStack(Stack):
         email_processor_dlq = common.make_processor_dlq(self, "EmailProcessorDlq")
 
         # --- Lambda function ---
+        email_processor_log_group = common.make_function_log_group(
+            self, "EmailProcessor", "po-email-processor"
+        )
         email_processor = lambda_.Function(
             self,
             "EmailProcessor",
@@ -134,7 +136,7 @@ class PoIngestStack(Stack):
             ),
             timeout=Duration.seconds(60),
             memory_size=256,
-            log_retention=logs.RetentionDays.TWO_MONTHS,
+            log_group=email_processor_log_group,
             dead_letter_queue=email_processor_dlq,
             environment={
                 "PO_TABLE": "purchase-orders",
@@ -196,7 +198,11 @@ class PoIngestStack(Stack):
         # false-reject storm pages instead of vanishing. default_value=0 keeps the
         # series populated (alarm stays OK, never INSUFFICIENT_DATA) between events.
         common.add_sender_auth_rejected_alarm(
-            self, "EmailProcessor", "po-email-processor", alarm_topic
+            self,
+            "EmailProcessor",
+            "po-email-processor",
+            alarm_topic,
+            email_processor_log_group,
         )
 
         # --- Template fallback-rate alarm: po-email-processor ---
@@ -372,6 +378,7 @@ class PoIngestStack(Stack):
         )
 
         # --- Web UI Lambda ---
+        web_ui_log_group = common.make_function_log_group(self, "WebUI", "po-web-ui")
         web_ui = lambda_.Function(
             self,
             "WebUI",
@@ -409,7 +416,7 @@ class PoIngestStack(Stack):
             ),
             timeout=Duration.seconds(60),
             memory_size=256,
-            log_retention=logs.RetentionDays.TWO_MONTHS,
+            log_group=web_ui_log_group,
             environment={
                 "PO_TABLE": "purchase-orders",
                 # Defense-in-depth shared secret for the web UI handler. The
@@ -464,6 +471,9 @@ class PoIngestStack(Stack):
         # 518 WCU of write amplification. Re-add if a state-level query path ships.
 
         # --- Site extractor Lambda (DynamoDB Streams → verified-sites) ---
+        site_extractor_log_group = common.make_function_log_group(
+            self, "SiteExtractor", "po-ingest-site-extractor"
+        )
         site_extractor = lambda_.Function(
             self,
             "SiteExtractor",
@@ -483,7 +493,7 @@ class PoIngestStack(Stack):
             ),
             timeout=Duration.seconds(60),
             memory_size=256,
-            log_retention=logs.RetentionDays.TWO_MONTHS,
+            log_group=site_extractor_log_group,
             environment={
                 "VERIFIED_SITES_TABLE": verified_sites_table.table_name,
                 "PENDING_REVIEW_TABLE": "pending-site-review",

cdk/wo_stack.py

@@ -9,7 +9,6 @@ from aws_cdk import (
     aws_cloudwatch_actions as cw_actions,
     aws_dynamodb as dynamodb,
     aws_lambda as lambda_,
-    aws_logs as logs,
     aws_s3 as s3,
     aws_s3_notifications as s3n,
     aws_ses as ses,
@@ -89,6 +88,9 @@ class WorkorderIngestStack(Stack):
         email_processor_dlq = common.make_processor_dlq(self, "EmailProcessorDlq")
 
         # --- Lambda function ---
+        email_processor_log_group = common.make_function_log_group(
+            self, "EmailProcessor", "workorder-email-processor"
+        )
         email_processor = lambda_.Function(
             self,
             "EmailProcessor",
@@ -119,7 +121,7 @@ class WorkorderIngestStack(Stack):
             ),
             timeout=Duration.seconds(60),
             memory_size=256,
-            log_retention=logs.RetentionDays.TWO_MONTHS,
+            log_group=email_processor_log_group,
             dead_letter_queue=email_processor_dlq,
             environment={
                 "WORK_ORDERS_TABLE": work_orders_table.table_name,
@@ -194,7 +196,11 @@ class WorkorderIngestStack(Stack):
         # dropped. This metric filter + alarm turns those warnings into a paging
         # signal so a false-reject storm surfaces instead of a silent outage.
         common.add_sender_auth_rejected_alarm(
-            self, "EmailProcessor", "workorder-email-processor", alarm_topic
+            self,
+            "EmailProcessor",
+            "workorder-email-processor",
+            alarm_topic,
+            email_processor_log_group,
         )
 
         # --- Template fallback-rate alarm: workorder-email-processor ---
@@ -304,6 +310,9 @@ class WorkorderIngestStack(Stack):
         )
 
         # --- Web UI Lambda ---
+        web_ui_log_group = common.make_function_log_group(
+            self, "WebUI", "workorder-web-ui"
+        )
         web_ui = lambda_.Function(
             self,
             "WebUI",
@@ -340,7 +349,7 @@ class WorkorderIngestStack(Stack):
             ),
             timeout=Duration.seconds(15),
             memory_size=128,
-            log_retention=logs.RetentionDays.TWO_MONTHS,
+            log_group=web_ui_log_group,
             environment={
                 "WORK_ORDERS_TABLE": work_orders_table.table_name,
                 "COMMENTS_TABLE": comments_table.table_name,

docs/po-template-parser.md

@@ -29,6 +29,10 @@ The PO email processor (`lambdas/po/email_processor/handler.py`) sends **every**
 Two phases: a read-only multi-agent (ultracode) feasibility study over a 120-email sample, then a full-bucket triage over **all 3,448** inbound emails to get real distribution numbers.
 
 ### 2.1 Data access
+> **Migration note (2026-07):** this section is historical — the harvest ran
+> against the management account. Post-migration the live bucket is
+> `s3://po-ingest-emails-011934824531/inbound/` (seahaven-prod, profile
+> `seahaven-prod`); the mgmt bucket persists only until decommission.
 - PO email bucket: `s3://po-ingest-emails-328440206208/inbound/` (AWS account **328440206208**, us-east-1).
 - ~~Reached via AWS profile `amoussa-mgmt` (SSO); default CLI creds are the personal account `681986854588`~~ **Stale (corrected 2026-07-16 during the fixture harvest):** the default CLI session is now authenticated to **328440206208** directly — no `--profile` flag needed.
 - ⚠️ **Corpus is aging out:** `inbound/` objects carry an S3 lifecycle expiration (~90-day rolling window; oldest object 2026-04-17 at harvest time, 3,422 objects vs 3,448 at triage). Any further harvesting should not be deferred long.
@@ -279,4 +283,4 @@ Plus `missing_required_field` for the required labeled fields (per-item metadata
 - PO handler + `EXTRACTION_PROMPT`: `lambdas/po/email_processor/handler.py`.
 - PO stack (alarms, IAM, Bedrock): `cdk/po_stack.py`.
 - Bedrock model: inference profile `us.anthropic.claude-haiku-4-5-20251001-v1:0`.
-- PO email bucket: `s3://po-ingest-emails-328440206208/inbound/` (account 328440206208, profile `amoussa-mgmt`).
+- PO email bucket (historical, pre-migration): `s3://po-ingest-emails-328440206208/inbound/`; post-migration `s3://po-ingest-emails-011934824531/inbound/` (seahaven-prod).

docs/runbook-dlq-recovery.md

@@ -18,14 +18,18 @@ There is **no console redrive-to-source**: the SQS console's "redrive to source"
 applies only to SQS-to-SQS DLQ relationships, not to a Lambda `DeadLetterConfig`
 target. Recovery is manual, via targeted re-invoke.
 
-## Resource inventory (acct 328440206208, us-east-1)
+## Resource inventory (acct 011934824531 seahaven-prod, us-east-1)
 
 | Pipeline | Function | DLQ queue name | Raw-email bucket | DLQ alarm |
 |---|---|---|---|---|
-| PO | `po-email-processor` | `po-ingest-EmailProcessorDlqA753DED5-az8LUZE3ubtz` | `po-ingest-emails-328440206208` | `po-email-processor-dlq-messages` |
-| WO | `workorder-email-processor` | `WorkorderIngestStack-EmailProcessorDlqA753DED5-Q8H555LrqSU1` | `workorder-ingest-emails-328440206208` | `workorder-email-processor-dlq-messages` |
+| PO | `po-email-processor` | _CDK-generated; fill from stack resources after the first prod deploy_ | `po-ingest-emails-011934824531` | `po-email-processor-dlq-messages` |
+| WO | `workorder-email-processor` | _CDK-generated; fill from stack resources after the first prod deploy_ | `workorder-ingest-emails-011934824531` | `workorder-email-processor-dlq-messages` |
 
-DLQ URLs are `https://sqs.us-east-1.amazonaws.com/328440206208/<queue-name>`.
+DLQ URLs are `https://sqs.us-east-1.amazonaws.com/011934824531/<queue-name>`.
+(Until mgmt decommission completes, the pre-migration mgmt-account queues
+`po-ingest-EmailProcessorDlqA753DED5-az8LUZE3ubtz` and
+`WorkorderIngestStack-EmailProcessorDlqA753DED5-Q8H555LrqSU1` still exist in
+328440206208 with any pre-cutover dead letters.)
 Both DLQs: 14-day retention, SSE, TLS-enforced, `VisibilityTimeout` 30s.
 
 ## Recovery procedure (no redrive — receive → extract key → targeted re-invoke → verify → purge)

infra/deploy-role/README.md

@@ -0,0 +1,34 @@
+# Deploy role: githubdeploy-procurement-ingest (seahaven-prod)
+
+OIDC deploy role for this repo's GitHub Actions pipeline in AWS account
+`011934824531` (seahaven-prod), us-east-1. Created as part of the migration
+from the management account (328440206208); the mgmt role of the same name
+stays untouched until decommission as the emergency mgmt deploy path.
+
+## Files
+
+| File | Purpose |
+|---|---|
+| `trust-policy.json` | OIDC trust: `repo:Sea-Haven-Industries/procurement-ingest:ref:refs/heads/main` only |
+| `permissions-policy.json` | `sts:AssumeRole` on the four `cdk-hnb659fds-*` bootstrap roles, `cloudformation:DescribeStacks` scoped to this repo's stacks + `CDKToolkit` (cd-cdk health check), and `lambda:InvokeFunction` on exactly the two email-processor function ARNs (post-deploy smoke gate) |
+| `create-deploy-role.sh` | Idempotent create-or-update from the two JSON files, profile `seahaven-prod` |
+
+> **Maintenance note:** `DescribeStacks` is scoped to `stack/po-ingest/*`, `stack/WorkorderIngestStack/*`, and `stack/CDKToolkit/*`. If a third stack is ever added to this CDK app, add its ARN pattern here and re-run the review-then-apply flow — otherwise the cd-cdk health check on the new stack will `AccessDenied`.
+
+## Why the SmokeInvokeLambda statement exists
+
+`deploy.yaml` runs `scripts/post-deploy-smoke.sh` under the deploy role's own
+session, not the assumed `cdk-*` roles. Without `lambda:InvokeFunction` on the
+two function ARNs the smoke gate hits AccessDenied and every deploy fails
+closed. The mgmt-era grant was applied out-of-band and undocumented; keeping
+it in these reviewed artifacts closes that gap. Scope it to exactly the two
+ARNs, never `Resource: "*"`.
+
+## Change process
+
+1. Edit the JSON artifacts on a branch; both gates must pass on the exact
+   files before anything is applied: GPT-4.1 cross-family review
+   (cross_review.py) and /sh-security-review.
+2. Run `./create-deploy-role.sh` (idempotent) with the seahaven-prod profile.
+3. Verify: `aws iam simulate-principal-policy` for the bootstrap-role
+   AssumeRole and both InvokeFunction ARNs, then a real pipeline run.

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

@@ -0,0 +1,96 @@
+#!/usr/bin/env bash
+###############################################################################
+# create-deploy-role.sh
+#
+# Creates / updates the GitHub Actions OIDC deploy role
+# `githubdeploy-procurement-ingest` in AWS account 011934824531 (seahaven-prod),
+# us-east-1, for the Sea-Haven-Industries/procurement-ingest repo (main branch).
+#
+# Part of the mgmt -> seahaven-prod migration. The mgmt-account role of the
+# same name is left untouched until decommission (emergency mgmt deploy path).
+#
+# GATE - DO NOT EXECUTE until BOTH of the following have passed on these
+# exact artifact files:
+#   1. GPT-4.1 cross-family review (IAM policy / trust-policy change) via
+#      cross_review.py in the security-review repo.
+#   2. /sh-security-review (deep agentic pass - IaC/IAM is a gated surface).
+#
+# The SmokeInvokeLambda statement is REQUIRED, not optional: deploy.yaml runs
+# scripts/post-deploy-smoke.sh under this role's session (not the assumed
+# cdk-* roles), and the smoke gate fails closed without lambda:InvokeFunction
+# on exactly the two email-processor function ARNs. The mgmt-era equivalent
+# of this policy was applied out-of-band and documented nowhere; these
+# artifacts are the fix for that debt.
+#
+# Resolved facts (verified 2026-07-23, read-only):
+#   Account   : 011934824531 (seahaven-prod)
+#   Region    : us-east-1
+#   Qualifier : hnb659fds  (AWS CDK DEFAULT - matches cdk/cdk.json)
+#   OIDC prov : arn:aws:iam::011934824531:oidc-provider/token.actions.githubusercontent.com
+###############################################################################
+
+set -euo pipefail
+
+PROFILE="seahaven-prod"
+ROLE_NAME="githubdeploy-procurement-ingest"
+POLICY_NAME="procurement-ingest-deploy"
+ACCOUNT_ID="011934824531"
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+TRUST_POLICY="file://${SCRIPT_DIR}/trust-policy.json"
+PERMS_POLICY="file://${SCRIPT_DIR}/permissions-policy.json"
+
+echo "==> Verifying active account for profile '${PROFILE}'..."
+CALLER_ACCOUNT="$(aws --profile "${PROFILE}" sts get-caller-identity --query Account --output text)"
+if [[ "${CALLER_ACCOUNT}" != "${ACCOUNT_ID}" ]]; then
+  echo "ERROR: profile '${PROFILE}' resolves to account ${CALLER_ACCOUNT}, expected ${ACCOUNT_ID}. Aborting." >&2
+  exit 1
+fi
+
+echo "==> Ensuring role '${ROLE_NAME}' exists with the correct trust policy..."
+if aws --profile "${PROFILE}" iam get-role --role-name "${ROLE_NAME}" >/dev/null 2>&1; then
+  echo "    Role exists - updating assume-role (trust) policy."
+  aws --profile "${PROFILE}" iam update-assume-role-policy \
+    --role-name "${ROLE_NAME}" \
+    --policy-document "${TRUST_POLICY}"
+else
+  echo "    Role absent - creating."
+  aws --profile "${PROFILE}" iam create-role \
+    --role-name "${ROLE_NAME}" \
+    --assume-role-policy-document "${TRUST_POLICY}" \
+    --description "GitHub Actions OIDC deploy role for Sea-Haven-Industries/procurement-ingest (main)" \
+    --max-session-duration 3600 \
+    --tags Key=project,Value=procurement-ingest Key=managed-by,Value=create-deploy-role.sh
+fi
+
+echo "==> Putting inline permissions policy '${POLICY_NAME}' (create-or-replace)..."
+aws --profile "${PROFILE}" iam put-role-policy \
+  --role-name "${ROLE_NAME}" \
+  --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-only (never auto-delete) is
+# deliberate: this is a shared prod-tenant account and an extra policy might be
+# a legitimate out-of-band grant an operator must adjudicate, not blindly strip.
+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)"
+
+echo "==> Done. Deploy role ready:"
+echo "    ${ROLE_ARN}"
+echo "    Set the GitHub repo secret AWS_DEPLOY_ROLE_ARN to this ARN (Phase 1 merge sequencing:"
+echo "    flip the secret only when Phase 0 is complete and the migration PR is about to merge)."

infra/deploy-role/permissions-policy.json

@@ -0,0 +1,40 @@
+{
+    "Version": "2012-10-17",
+    "Statement": [
+        {
+            "Sid": "AssumeCdkBootstrapRoles",
+            "Effect": "Allow",
+            "Action": "sts:AssumeRole",
+            "Resource": [
+                "arn:aws:iam::011934824531:role/cdk-hnb659fds-deploy-role-011934824531-us-east-1",
+                "arn:aws:iam::011934824531:role/cdk-hnb659fds-file-publishing-role-011934824531-us-east-1",
+                "arn:aws:iam::011934824531:role/cdk-hnb659fds-lookup-role-011934824531-us-east-1",
+                "arn:aws:iam::011934824531:role/cdk-hnb659fds-image-publishing-role-011934824531-us-east-1"
+            ]
+        },
+        {
+            "Sid": "CdkDeployHealthCheck",
+            "Effect": "Allow",
+            "Action": "cloudformation:DescribeStacks",
+            "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"
+                }
+            }
+        },
+        {
+            "Sid": "SmokeInvokeLambda",
+            "Effect": "Allow",
+            "Action": "lambda:InvokeFunction",
+            "Resource": [
+                "arn:aws:lambda:us-east-1:011934824531:function:po-email-processor",
+                "arn:aws:lambda:us-east-1:011934824531:function:workorder-email-processor"
+            ]
+        }
+    ]
+}

infra/deploy-role/trust-policy.json

@@ -0,0 +1,18 @@
+{
+    "Version": "2012-10-17",
+    "Statement": [
+        {
+            "Effect": "Allow",
+            "Principal": {
+                "Federated": "arn:aws:iam::011934824531:oidc-provider/token.actions.githubusercontent.com"
+            },
+            "Action": "sts:AssumeRoleWithWebIdentity",
+            "Condition": {
+                "StringEquals": {
+                    "token.actions.githubusercontent.com:aud": "sts.amazonaws.com",
+                    "token.actions.githubusercontent.com:sub": "repo:Sea-Haven-Industries/procurement-ingest:ref:refs/heads/main"
+                }
+            }
+        }
+    ]
+}

scripts/migrate_tables.py

@@ -0,0 +1,416 @@
+"""
+One-off cross-account DynamoDB backfill for the mgmt -> seahaven-prod
+migration (Phase 4 of the migration plan). Scans each table in the source
+(mgmt) account and writes into the same-named, already-CDK-created table in
+the destination (prod) account.
+
+Dry-run is the DEFAULT in every mode; nothing is written or deleted unless
+--execute is passed.
+
+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). 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 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
+                                replays the prod site-extractor stream, which
+                                creates junk `pending` rows and non-idempotent
+                                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 a fresh --backup-arn
+                                of THIS table, on DRY-RUN as well as --execute:
+                                the rehearsal validates the same preconditions
+                                the real run will).
+
+Usage:
+    # 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
+    python scripts/migrate_tables.py copy --table verified-sites \
+        --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 --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
+resolved account ids before doing anything.
+"""
+
+import argparse
+import json
+import sys
+from datetime import datetime, timedelta, timezone
+
+import boto3
+
+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"]},
+    "WorkOrderComments": {
+        "mode": "cutoff",
+        "keys": ["work_order_id", "comment_id"],
+        "cutoff_attr": "ingested_at",
+    },
+    "verified-sites": {"mode": "truncate_load", "keys": ["siteCode"]},
+    "pending-site-review": {"mode": "truncate_load", "keys": ["po_number"]},
+}
+
+
+def session_table(profile, expected_account, table_name):
+    session = boto3.Session(profile_name=profile, region_name=REGION)
+    acct = session.client("sts").get_caller_identity()["Account"]
+    if acct != expected_account:
+        sys.exit(
+            f"ERROR: profile {profile!r} resolves to account {acct}, "
+            f"expected {expected_account}. Aborting."
+        )
+    return session.resource("dynamodb").Table(table_name)
+
+
+def check_key_schema(table, name, keys):
+    """Abort with a clear message if the configured keys drift from the live
+    table's key schema. Every DynamoDB item necessarily carries its table's
+    key attributes, so once this passes the `item[k]` key projections below
+    cannot KeyError; a mismatch here is the only way they could. Set (not
+    ordered) comparison is deliberate: keys are only ever used to build dict
+    Key= projections, which are order-insensitive."""
+    live = {k["AttributeName"] for k in table.key_schema}
+    if live != set(keys):
+        sys.exit(
+            f"ERROR: {name} live key schema {sorted(live)} != configured "
+            f"{sorted(keys)}; refusing to build keys from a stale TABLES entry."
+        )
+
+
+def scan_items(table, filter_kwargs=None):
+    kwargs = dict(filter_kwargs or {})
+    while True:
+        page = table.scan(**kwargs)
+        yield from page.get("Items", [])
+        lek = page.get("LastEvaluatedKey")
+        if not lek:
+            return
+        kwargs["ExclusiveStartKey"] = lek
+
+
+def scan_count(table, filter_kwargs=None):
+    kwargs = dict(filter_kwargs or {}, Select="COUNT")
+    total = 0
+    while True:
+        page = table.scan(**kwargs)
+        total += page["Count"]
+        lek = page.get("LastEvaluatedKey")
+        if not lek:
+            return total
+        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"),
+        "ExpressionAttributeNames": {"#ia": "ingested_at"},
+        "ExpressionAttributeValues": {":cutoff": cutoff},
+    }
+
+
+def _dest_cancelled_po_numbers(dst):
+    """po_numbers currently Cancelled in the destination purchase-orders table.
+
+    Read once before an overwrite copy so a stale non-Cancelled source item
+    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 _validate_truncate_backup(name, dst, args):
+    """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):
+    # Validate in dry-run too: a stale/missing --backup-arn should surface on
+    # the rehearsal run, not first appear when the operator adds --execute.
+    _validate_truncate_backup(name, dst, args)
+    dest_keys = [{k: item[k] for k in 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:
+            for key in dest_keys:
+                batch.delete_item(Key=key)
+        print(f"[{name}] truncate: done")
+    else:
+        print(f"[{name}] DRY-RUN: no deletes performed")
+
+
+def copy_table(name, cfg, args):
+    src = session_table(args.source_profile, SOURCE_ACCOUNT, name)
+    dst = session_table(args.dest_profile, DEST_ACCOUNT, name)
+    check_key_schema(src, name, cfg["keys"])
+    check_key_schema(dst, name, cfg["keys"])
+
+    filter_kwargs = None
+    if cfg["mode"] == "cutoff":
+        # --cutoff presence is enforced up front in main(); normalize here.
+        filter_kwargs = comments_cutoff_filter(normalize_cutoff(args.cutoff))
+
+    if cfg["mode"] == "truncate_load":
+        _truncate_destination(name, dst, args, cfg["keys"])
+
+    # 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 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):
+    src = session_table(args.source_profile, SOURCE_ACCOUNT, name)
+    dst = session_table(args.dest_profile, DEST_ACCOUNT, name)
+    check_key_schema(src, name, cfg["keys"])
+    check_key_schema(dst, name, cfg["keys"])
+
+    src_filter = None
+    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/overwrite tables; exact parity is expected
+    # only for truncate_load immediately after the load.
+    relation = {
+        "overwrite": dst_count >= src_count,
+        "cutoff": dst_count >= src_count,
+        "truncate_load": dst_count == src_count,
+    }[cfg["mode"]]
+    status = "OK" if relation else "MISMATCH"
+    print(f"[{name}] source={src_count} dest={dst_count} -> {status}")
+
+    ok = relation
+    # Spot-check every mode. Key-existence lookups are sound for ALL modes
+    # because the copy batch-Puts source items VERBATIM (source comment_id
+    # included) -- per-account comment_id divergence only affects rows each
+    # pipeline ingests itself, and those dual-window rows are excluded here by
+    # the same cutoff filter the copy used. For truncate_load keys are likewise
+    # copied verbatim.
+    if args.sample:
+        sampled = mismatched = 0
+        for item in scan_items(src, src_filter):
+            if sampled >= args.sample:
+                break
+            sampled += 1
+            key = {k: item[k] for k in cfg["keys"]}
+            got = dst.get_item(Key=key).get("Item")
+            if got is None:
+                mismatched += 1
+                print(f"[{name}] MISSING in dest: {json.dumps(key, default=str)}")
+        print(f"[{name}] spot-check: {sampled} sampled, {mismatched} missing")
+        ok = ok and mismatched == 0
+
+    cancelled_check_limit = 25
+    if name == "purchase-orders":
+        # Propagation check: a PO Cancelled in mgmt must be Cancelled in prod.
+        # (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"},
+        }
+        checked = wrong = 0
+        for item in scan_items(src, cancelled_filter):
+            if checked >= cancelled_check_limit:
+                break
+            checked += 1
+            got = dst.get_item(Key={"po_number": item["po_number"]}).get("Item")
+            if not got or got.get("po_status") != "Cancelled":
+                wrong += 1
+                print(f"[{name}] CANCELLED-DRIFT: {item['po_number']}")
+        print(f"[{name}] cancelled spot-check: {checked} checked, {wrong} drifted")
+        ok = ok and wrong == 0
+
+    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 (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 for WorkOrderComments",
+        )
+        if cmd == "copy":
+            p.add_argument("--execute", action="store_true")
+            p.add_argument(
+                "--backup-arn",
+                help="required for truncate-and-load tables (validated on "
+                "dry-run as well as --execute)",
+            )
+        else:
+            p.add_argument(
+                "--sample",
+                type=int,
+                default=10,
+                help="per-table spot-check sample size (0 disables)",
+            )
+
+    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":
+        copy_table(args.table, TABLES[args.table], args)
+    else:
+        results = [verify_table(n, TABLES[n], args) for n in names]
+        if not all(results):
+            sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()

tests/test_reprocess_contract.py

@@ -44,7 +44,9 @@ _spec.loader.exec_module(reprocess)
 
 
 def test_build_s3_event_shape_and_raw_key():
-    bucket = "po-ingest-emails-328440206208"
+    # Account-agnostic fixture (the suffix is never parsed); the real bucket
+    # name is account-derived at deploy time.
+    bucket = "po-ingest-emails-000000000000"
     # Deliberately contains a SPACE, a literal '+', and a literal '%41'.
     # A real S3 event notification would deliver this key encoded as
     #   "inbound/2026/AB+12%2B34+%2541.eml"