feat(P4): Nova rebrand — AWS resource migration (REQ-163)

Rename all acdl-* AWS resources → nova-* across terraform (DynamoDB,
Secrets Manager, Lambda, SNS, SG, KMS alias, ECS, ECR, IAM user/policy,
state bucket, ALB, VPC/subnet names). Lambda default table names → nova-*
(D-111). State bucket backend → nova-tfstate (-migrate-state documented).
New docs/NOVA_AWS_MIGRATION.md runbook (staged migration + rollback).
New scripts/migrate_dynamodb_data.py (scan+copy, dry-run default).
acdl-deploy- → nova-deploy- role ARN in deploy workflows. Test fixtures
updated; terraform validate + pytest + run_ci.sh PASS.

---ci---
project: acdl
phase: 4
milestone: v1.15
status: execute
---/ci---

tests/test_contract_ingestor.py

@@ -66,7 +66,7 @@ def moto_contracts_table(monkeypatch):
     with mock_aws():
         dyn = boto3.client("dynamodb", region_name="us-east-1")
         dyn.create_table(
-            TableName="acdl-contracts",
+            TableName="nova-contracts",
             KeySchema=[
                 {"AttributeName": "consumerRepo", "KeyType": "HASH"},
                 {"AttributeName": "contractId#submittedAt", "KeyType": "RANGE"},
@@ -84,7 +84,7 @@ def moto_contracts_table(monkeypatch):
         saved_secrets = ingestor._secrets_client
         ingestor._dynamodb = None
         ingestor._secrets_client = None
-        monkeypatch.setattr(ingestor, "TABLE_NAME", "acdl-contracts")
+        monkeypatch.setattr(ingestor, "TABLE_NAME", "nova-contracts")
 
         yield dyn
 
@@ -108,7 +108,7 @@ class TestSubmitContract:
         # Verify what landed in DynamoDB.
         sk = f"contract-001#{result['submittedAt']}"
         resp = moto_contracts_table.get_item(
-            TableName="acdl-contracts",
+            TableName="nova-contracts",
             Key={
                 "consumerRepo": {"S": "acdl/consumer-a"},
                 "contractId#submittedAt": {"S": sk},
@@ -369,7 +369,7 @@ class TestCallerIdentityValidation:
     def test_invalid_consumer_repo_format_rejected(self, moto_contracts_table, sample_payload):
         # A consumerRepo without "/" is invalid (not org/repo format).
         sample_payload["consumerRepo"] = "not-a-repo-format"
-        event = {"body": json.dumps(sample_payload), "requestContext": {"identity": {"userArn": "arn:aws:sts::000:assumed-role/acdl-deploy/session"}}}
+        event = {"body": json.dumps(sample_payload), "requestContext": {"identity": {"userArn": "arn:aws:sts::000:assumed-role/nova-deploy/session"}}}
         resp = ingestor.lambda_handler(event, None)
         assert resp["statusCode"] == 400
         assert "invalid consumerRepo" in json.loads(resp["body"])["error"]
@@ -378,7 +378,7 @@ class TestCallerIdentityValidation:
         # A valid org/repo consumerRepo with an identity present — passes.
         event = {
             "body": json.dumps(sample_payload),
-            "requestContext": {"identity": {"userArn": "arn:aws:sts::000:assumed-role/acdl-deploy/acdl-consumer-a"}},
+            "requestContext": {"identity": {"userArn": "arn:aws:sts::000:assumed-role/nova-deploy/nova-consumer-a"}},
         }
         resp = ingestor.lambda_handler(event, None)
         assert resp["statusCode"] == 200
@@ -431,12 +431,12 @@ class TestValidateChangeRequest:
         monkeypatch.setenv("AWS_DEFAULT_REGION", "us-east-1")
         monkeypatch.setenv("AWS_ACCESS_KEY_ID", "testing")
         monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "testing")
-        monkeypatch.setenv("CHANGE_REQUESTS_TABLE", "acdl-change-requests")
+        monkeypatch.setenv("CHANGE_REQUESTS_TABLE", "nova-change-requests")
 
         with mock_aws():
             dynamodb = boto3.resource("dynamodb", region_name="us-east-1")
             table = dynamodb.create_table(
-                TableName="acdl-change-requests",
+                TableName="nova-change-requests",
                 KeySchema=[
                     {"AttributeName": "changeRequestId", "KeyType": "HASH"},
                     {"AttributeName": "submittedAt", "KeyType": "RANGE"},

tests/test_contract_resolver.py

@@ -67,7 +67,7 @@ class TestResolveMicroservice:
                 "microservice": {
                     "version": "1.0.0",
                     "inputs": {
-                        "image": "581513795199.dkr.ecr.us-east-1.amazonaws.com/acdl-microservice:latest",
+                        "image": "581513795199.dkr.ecr.us-east-1.amazonaws.com/nova-microservice:latest",
                         "port": 8080,
                         "region": "us-east-1",
                     },

tests/test_iam_policy_baseline.py

@@ -134,13 +134,13 @@ class TestIAMPolicyBaseline:
     def test_dynamodb_contracts_table_in_resource(self, policy):
         contracts_stmts = [
             s for s in policy["Statement"]
-            if any("acdl-contracts" in r for r in (
+            if any("nova-contracts" in r for r in (
                 s.get("Resource") if isinstance(s.get("Resource"), list) else [s.get("Resource", "")]
             ))
         ]
-        assert contracts_stmts, "no statement references the acdl-contracts table"
+        assert contracts_stmts, "no statement references the nova-contracts table"
 
-    def test_lambda_scoped_to_acdl_functions(self, policy):
+    def test_lambda_scoped_to_nova_functions(self, policy):
         lambda_stmts = [s for s in policy["Statement"] if any(
             a.startswith("lambda:") for a in (
                 s.get("Action") if isinstance(s.get("Action"), list) else [s.get("Action", "")]
@@ -151,8 +151,8 @@ class TestIAMPolicyBaseline:
             res = s.get("Resource", "")
             if isinstance(res, list):
                 res = " ".join(res)
-            assert "function:acdl-*" in res or res == "*", \
-                "lambda actions not scoped to acdl-* functions"
+            assert "function:nova-*" in res or res == "*", \
+                "lambda actions not scoped to nova-* functions"
 
     def test_cost_explorer_is_read_only(self, policy):
         ce_actions = set()
@@ -178,8 +178,8 @@ class TestIAMPolicyBaseline:
                     res = " ".join(res)
                 assert res != "*", "iam:PassRole must not be granted to Resource: *"
 
-    def test_iam_role_creation_scoped_to_acdl_prefix(self, policy):
-        """G-104: iam:CreateRole must be scoped to role/acdl-* (not Resource: *)."""
+    def test_iam_role_creation_scoped_to_nova_prefix(self, policy):
+        """G-104: iam:CreateRole must be scoped to role/nova-* (not Resource: *)."""
         for s in policy["Statement"]:
             acts = s.get("Action", [])
             if isinstance(acts, str):
@@ -188,10 +188,10 @@ class TestIAMPolicyBaseline:
                 res = s.get("Resource", "")
                 if isinstance(res, list):
                     res = " ".join(res)
-                assert "acdl-*" in res, f"iam:CreateRole must be scoped to acdl-* (got: {res})"
+                assert "nova-*" in res, f"iam:CreateRole must be scoped to nova-* (got: {res})"
 
-    def test_kms_scoped_to_acdl_alias(self, policy):
-        """G-104: kms:CreateKey etc. must be scoped to alias/acdl-* (not Resource: *)."""
+    def test_kms_scoped_to_nova_alias(self, policy):
+        """G-104: kms:CreateKey etc. must be scoped to alias/nova-* (not Resource: *)."""
         for s in policy["Statement"]:
             acts = s.get("Action", [])
             if isinstance(acts, str):
@@ -200,7 +200,7 @@ class TestIAMPolicyBaseline:
                 res = s.get("Resource", "")
                 if isinstance(res, list):
                     res = " ".join(res)
-                assert "acdl-*" in res, f"kms actions must be scoped to acdl-* (got: {res})"
+                assert "nova-*" in res, f"kms actions must be scoped to nova-* (got: {res})"
 
     def test_cloudfront_waf_remain_global(self, policy):
         """G-104: CloudFront + WAFv2 (CloudFront scope) ARNs are global;
@@ -215,4 +215,4 @@ class TestIAMPolicyBaseline:
                 if isinstance(res, list):
                     res = res[0] if res else ""
                 # CloudFront/WAFv2 are allowed to be * (global ARNs)
-                assert res == "*" or "acdl" in res
+                assert res == "*" or "nova" in res

tests/test_local_emulating_adapters.py

@@ -121,7 +121,7 @@ def test_local_s3_backend_rewrites_s3_to_local(tmp_path):
     tf = tmp_path / "terraform.tf"
     tf.write_text(
         'terraform {\n  required_version = ">= 1.9"\n  backend "s3" {\n'
-        '    bucket = "acdl-tfstate-x"\n    key = "spike/s.tfstate"\n'
+        '    bucket = "nova-tfstate-x"\n    key = "spike/s.tfstate"\n'
         '    region = "us-east-1"\n  }\n}\n'
     )
     backend.rewrite_terraform_tf(tf, "test-stack")

tests/test_migrate_dynamodb_data.py

@@ -0,0 +1,148 @@
+"""Unit tests for scripts/migrate_dynamodb_data.py (REQ-163, P4).
+
+Tests the pure item-mapping logic + table-pair resolution. The AWS I/O
+(scan_all/copy_items) is thin boto3 glue, not unit-tested here (covered
+by the dry-run path + the runbook's live verification).
+"""
+import sys
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(ROOT / "scripts"))
+
+import migrate_dynamodb_data as mig
+
+
+class TestMapItem:
+    def test_map_item_preserves_typed_attributes(self):
+        item = {
+            "consumerRepo": {"S": "acdl/consumer-a"},
+            "contractId#submittedAt": {"S": "c-1#2026-01-01T00:00:00Z"},
+            "contract": {"S": "name: foo\n"},
+            "count": {"N": "42"},
+        }
+        result = mig.map_item(item)
+        assert result == item
+
+    def test_map_item_returns_independent_copy(self):
+        """The mapped item must not alias the scanned item (callers may mutate)."""
+        item = {"k": {"S": "v"}}
+        result = mig.map_item(item)
+        result["k"]["S"] = "mutated"
+        assert item["k"]["S"] == "v", "map_item returned an alias, not a copy"
+
+    def test_map_item_empty(self):
+        assert mig.map_item({}) == {}
+
+    def test_map_item_preserves_binary_and_nested(self):
+        item = {
+            "pk": {"B": b"\x01\x02"},
+            "nested": {"M": {"a": {"S": "x"}}},
+            "list": {"L": [{"S": "1"}, {"S": "2"}]},
+        }
+        assert mig.map_item(item) == item
+
+
+class TestTablePair:
+    def test_contracts_alias(self):
+        assert mig.table_pair_for("contracts") == ("acdl-contracts", "nova-contracts")
+
+    def test_change_requests_alias(self):
+        assert mig.table_pair_for("change-requests") == (
+            "acdl-change-requests", "nova-change-requests"
+        )
+
+    def test_literal_source_name(self):
+        assert mig.table_pair_for("acdl-contracts") == ("acdl-contracts", "nova-contracts")
+
+    def test_literal_dest_name(self):
+        assert mig.table_pair_for("nova-contracts") == ("acdl-contracts", "nova-contracts")
+
+    def test_unknown_name_raises(self):
+        with pytest.raises(ValueError, match="unknown table"):
+            mig.table_pair_for("nope")
+
+    def test_custom_pairs(self):
+        pairs = [("old-x", "new-x")]
+        assert mig.table_pair_for("old-x", pairs=pairs) == ("old-x", "new-x")
+
+
+class TestDefaultPairs:
+    def test_default_pairs_cover_both_tables(self):
+        sources = [s for s, _ in mig.DEFAULT_TABLE_PAIRS]
+        dests = [d for _, d in mig.DEFAULT_TABLE_PAIRS]
+        assert sources == ["acdl-contracts", "acdl-change-requests"]
+        assert dests == ["nova-contracts", "nova-change-requests"]
+
+
+class TestArgparser:
+    def test_dry_run_default(self):
+        args = mig.build_parser().parse_args([])
+        assert args.apply is False
+        assert args.region == "us-east-1"
+        assert args.table is None
+
+    def test_apply_flag(self):
+        args = mig.build_parser().parse_args(["--apply"])
+        assert args.apply is True
+
+    def test_table_filter(self):
+        args = mig.build_parser().parse_args(["--table", "contracts"])
+        assert args.table == "contracts"
+
+    def test_source_dest_override(self):
+        args = mig.build_parser().parse_args(["--source", "old", "--dest", "new"])
+        assert args.source == "old"
+        assert args.dest == "new"
+
+
+class TestRunDryRun:
+    """The dry-run path exercises the table-pair resolution + describes both
+    tables without writing. We stub the boto3 client so no AWS access occurs."""
+
+    def _fake_client(self, describable=True):
+        client = type("FakeClient", (), {})()
+        def describe_table(TableName):
+            if not describable:
+                raise Exception("ResourceNotFoundException")
+            return {"Table": {"ItemCount": 0}}
+        client.describe_table = describe_table
+        client.scan = lambda **k: {"Items": []}
+        client.put_item = lambda **k: None
+        return client
+
+    def test_run_dry_run_reports_planned_copy(self, monkeypatch, capsys):
+        # Build args with both default pairs.
+        args = mig.build_parser().parse_args([])
+        # Stub the client constructor so no real boto3 client is built.
+        monkeypatch.setattr(mig.boto3, "client", lambda *a, **k: self._fake_client())
+        rc = mig.run(args)
+        out = capsys.readouterr().out
+        assert rc == 0
+        assert "DRY-RUN" in out
+        assert "acdl-contracts" in out and "nova-contracts" in out
+        assert "acdl-change-requests" in out and "nova-change-requests" in out
+        assert "would PutItem" in out
+        assert "NOT deleted" in out
+
+    def test_run_source_table_not_describable_fails(self, monkeypatch, capsys):
+        args = mig.build_parser().parse_args([])
+        # First describe_table (source) raises, second (dest) is fine — emulate by
+        # raising on the first call only.
+        calls = {"n": 0}
+        client = type("FakeClient", (), {})()
+        def describe_table(TableName):
+            calls["n"] += 1
+            if calls["n"] % 2 == 1:  # source (odd calls)
+                raise Exception("ResourceNotFoundException")
+            return {"Table": {"ItemCount": 0}}
+        client.describe_table = describe_table
+        client.scan = lambda **k: {"Items": []}
+        client.put_item = lambda **k: None
+        monkeypatch.setattr(mig.boto3, "client", lambda *a, **k: client)
+        rc = mig.run(args)
+        err = capsys.readouterr().err
+        assert rc == 1
+        assert "not describable" in err

tests/test_outbox_writer.py

@@ -55,7 +55,7 @@ class TestWriteEvent:
         with mock_aws():
             dyn = boto3.client("dynamodb", region_name="us-east-1")
             dyn.create_table(
-                TableName="acdl-outbox",
+                TableName="nova-outbox",
                 KeySchema=[
                     {"AttributeName": "contractId", "KeyType": "HASH"},
                     {"AttributeName": "eventType#eventTs", "KeyType": "RANGE"},
@@ -68,7 +68,7 @@ class TestWriteEvent:
             )
 
             event = self._sample_event()
-            item = write_event(event, outbox_table="acdl-outbox", region="us-east-1")
+            item = write_event(event, outbox_table="nova-outbox", region="us-east-1")
 
             assert item["contractId"]["S"] == "test-contract-001"
             assert item["prev_event_hash"]["S"] == "GENESIS"
@@ -84,7 +84,7 @@ class TestWriteEvent:
         with mock_aws():
             dyn = boto3.client("dynamodb", region_name="us-east-1")
             dyn.create_table(
-                TableName="acdl-outbox",
+                TableName="nova-outbox",
                 KeySchema=[
                     {"AttributeName": "contractId", "KeyType": "HASH"},
                     {"AttributeName": "eventType#eventTs", "KeyType": "RANGE"},
@@ -97,7 +97,7 @@ class TestWriteEvent:
             )
 
             event = self._sample_event()
-            item = write_event(event, outbox_table="acdl-outbox", region="us-east-1")
+            item = write_event(event, outbox_table="nova-outbox", region="us-east-1")
             expected_hash = _canonical_hash(event)
             assert item["hash"]["S"] == expected_hash
 
@@ -109,7 +109,7 @@ class TestWriteEvent:
         with mock_aws():
             dyn = boto3.client("dynamodb", region_name="us-east-1")
             dyn.create_table(
-                TableName="acdl-outbox",
+                TableName="nova-outbox",
                 KeySchema=[
                     {"AttributeName": "contractId", "KeyType": "HASH"},
                     {"AttributeName": "eventType#eventTs", "KeyType": "RANGE"},
@@ -122,10 +122,10 @@ class TestWriteEvent:
             )
 
             event = self._sample_event()
-            write_event(event, outbox_table="acdl-outbox", region="us-east-1")
+            write_event(event, outbox_table="nova-outbox", region="us-east-1")
 
             resp = dyn.get_item(
-                TableName="acdl-outbox",
+                TableName="nova-outbox",
                 Key={
                     "contractId": {"S": "test-contract-001"},
                     "eventType#eventTs": {"S": "CONFIDENCE_COMPUTED#2026-07-22T00:00:00Z"},

tests/test_output_publisher.py

@@ -455,7 +455,7 @@ class TestInvokePolicyTemplate:
         rendered = template.replace("${account_id}", "123456789012").replace("${region}", "us-east-1")
         policy = json.loads(rendered)
         resource_arn = policy["Statement"][0]["Resource"]
-        assert resource_arn == "arn:aws:lambda:us-east-1:123456789012:function:acdl-contract-ingestor"
+        assert resource_arn == "arn:aws:lambda:us-east-1:123456789012:function:nova-contract-ingestor"
         assert "000000000000" not in resource_arn
         # ${consumerRepo} is a runtime placeholder (not a Terraform variable) — it stays.
         assert "${account_id}" not in rendered

tests/test_route_halt_artifact.py

@@ -15,13 +15,13 @@ from core.separation_of_duties import route_halt_artifact
 
 def test_route_halt_publishes_to_sns_when_arn_set(monkeypatch):
     """With ACDL_SOD_HALT_TOPIC_ARN set, the SNS client receives the publish."""
-    monkeypatch.setenv("ACDL_SOD_HALT_TOPIC_ARN", "arn:aws:sns:us-east-1:000000000000:acdl-sod-halt")
+    monkeypatch.setenv("ACDL_SOD_HALT_TOPIC_ARN", "arn:aws:sns:us-east-1:000000000000:nova-sod-halt")
     sns_client = mock.MagicMock()
     route_halt_artifact("contract-123", "SEPARATION_OF_DUTIES_VIOLATION: x==y",
                         oncall_client=sns_client)
     sns_client.publish.assert_called_once()
     call = sns_client.publish.call_args
-    assert call.kwargs["TopicArn"] == "arn:aws:sns:us-east-1:000000000000:acdl-sod-halt"
+    assert call.kwargs["TopicArn"] == "arn:aws:sns:us-east-1:000000000000:nova-sod-halt"
     assert "contract-123" in call.kwargs["Message"]
     assert "SEPARATION_OF_DUTIES_VIOLATION" in call.kwargs["Message"]
     assert call.kwargs["Subject"] == "ACDL SoD halt"
@@ -54,7 +54,7 @@ def test_route_halt_outbox_fallback_writes_event(monkeypatch):
 
 def test_route_halt_sns_failure_falls_back_to_outbox(monkeypatch):
     """If SNS publish raises, the outbox fallback is used."""
-    monkeypatch.setenv("ACDL_SOD_HALT_TOPIC_ARN", "arn:aws:sns:us-east-1:000000000000:acdl-sod-halt")
+    monkeypatch.setenv("ACDL_SOD_HALT_TOPIC_ARN", "arn:aws:sns:us-east-1:000000000000:nova-sod-halt")
     sns_client = mock.MagicMock()
     sns_client.publish.side_effect = Exception("SNS down")
     with mock.patch("core.outbox_writer.write_event") as mock_write:
@@ -63,8 +63,8 @@ def test_route_halt_sns_failure_falls_back_to_outbox(monkeypatch):
 
 
 def test_sns_topic_defined_in_terraform():
-    """terraform/platform/main.tf defines the acdl-sod-halt SNS topic."""
+    """terraform/platform/main.tf defines the nova-sod-halt SNS topic."""
     tf = (ROOT / "terraform" / "platform" / "main.tf").read_text()
     assert "aws_sns_topic" in tf
-    assert "acdl-sod-halt" in tf
-    assert "acdl_sod_halt_topic_arn" in tf
+    assert "nova-sod-halt" in tf
+    assert "nova_sod_halt_topic_arn" in tf

tests/test_untested_scripts.py

@@ -107,12 +107,11 @@ class TestCreateStateBackend:
 
     def test_state_bucket_name_construction(self, monkeypatch):
         """The state bucket name is derived from NOVA_AWS_ACCOUNT_ID
-        (P2 renamed from ACDL_AWS_ACCOUNT_ID; the bucket name acdl-tfstate-*
-        stays until P4, REQ-163)."""
+        (P4, REQ-163: bucket renamed acdl-tfstate-* → nova-tfstate-*)."""
         monkeypatch.setenv("NOVA_AWS_ACCOUNT_ID", "123456789012")
         account_id = os.environ.get("NOVA_AWS_ACCOUNT_ID", "581513795199")
-        state_bucket = f"acdl-tfstate-{account_id}-us-east-1"
-        assert state_bucket == "acdl-tfstate-123456789012-us-east-1"
+        state_bucket = f"nova-tfstate-{account_id}-us-east-1"
+        assert state_bucket == "nova-tfstate-123456789012-us-east-1"
 
     def test_idempotent_bucket_creation(self, monkeypatch):
         """head_bucket success -> no create_bucket called."""
@@ -141,11 +140,11 @@ class TestCreateIamUser:
         from unittest import mock
 
         mock_iam = mock.MagicMock()
-        mock_iam.get_user.return_value = {"User": {"UserName": "acdl-spike-runner"}}
+        mock_iam.get_user.return_value = {"User": {"UserName": "nova-spike-runner"}}
         monkeypatch.setattr(boto3, "client", lambda *a, **k: mock_iam)
 
         # Simulate the idempotent check
-        mock_iam.get_user(UserName="acdl-spike-runner")
+        mock_iam.get_user(UserName="nova-spike-runner")
         mock_iam.create_user.assert_not_called()
 
     def test_policy_overwrite_is_idempotent(self, monkeypatch):
@@ -157,5 +156,5 @@ class TestCreateIamUser:
         monkeypatch.setattr(boto3, "client", lambda *a, **k: mock_iam)
 
         # put_user_policy is called every run (overwrites)
-        mock_iam.put_user_policy(UserName="acdl-spike-runner", PolicyName="p", PolicyDocument="{}")
+        mock_iam.put_user_policy(UserName="nova-spike-runner", PolicyName="p", PolicyDocument="{}")
         mock_iam.put_user_policy.assert_called_once()