feat(P2): env-transition detect-and-destroy — REQ-282..287

- core/env_transition.py: detect_prior_env() + record_applied_env() via DynamoDB nova-contracts table (REQ-282,283)
- scripts/run_platform.sh Step 0b: detect env change, destroy prior env (deletion_protection=false, terraform init -reconfigure + destroy), emit ENV_DESTROYED evidence event, fail closed on destroy failure (REQ-284)
- scripts/run_platform.sh: record applied env after successful apply (REQ-285)
- .github/workflows/deploy.yml: pass NOVA_CONSUMER_REPO to run_platform.sh (REQ-286)
- adapters/terraform/adapter.py: doc comment on env-scoped state key (REQ-287)

No orphan path: if destroy fails, pipeline exits non-zero (no apply runs).

---ci---
project: acdl
phase: 2
milestone: v1.24
status: execute
requirements: [REQ-282,REQ-283,REQ-284,REQ-285,REQ-286,REQ-287]
---/ci---

core/env_transition.py

@@ -0,0 +1,159 @@
+"""Nova Environment Transition — detect prior env + record applied env.
+
+When a consumer edits the `environment:` field on a stable contract `id`
+(Shape A promotion), the platform must destroy the prior environment's
+resources before building the new environment. This module provides the
+DynamoDB query logic to detect the prior environment and record the
+applied environment after a successful apply.
+
+Source of truth: the `nova-contracts` DynamoDB table (PK `consumerRepo`,
+SK `contractId#submittedAt`), written by `core/lambda/contract_ingestor.py`.
+
+detect_prior_env() queries the table for the last-applied environment for
+a given consumerRepo + contractId. If it differs from the new env, the
+prior env name is returned (so the pipeline can destroy it). If no record
+exists (first deploy or Shape B per-env caller), returns None.
+
+record_applied_env() writes a `#LAST_APPLIED` record after a successful
+apply, so the next run's detect step has a source of truth.
+
+Failures to reach DynamoDB (local/CI mode without the table) log a warning
+and return None (conservative — no false-positive destroys). This is the
+no-orphan-path guarantee: if we can't confirm a prior env, we don't
+destroy, but we also don't silently proceed in a way that orphans — the
+record step ensures future runs have the data.
+
+CLI:
+    python3 core/env_transition.py detect --contract-id <id> --consumer-repo <repo> --new-env <env>
+    python3 core/env_transition.py record --contract-id <id> --consumer-repo <repo> --env <env>
+"""
+
+import datetime
+import json
+import os
+import sys
+from typing import Optional
+
+try:
+    import boto3
+except ImportError:
+    boto3 = None
+
+TABLE_NAME = os.environ.get("CONTRACTS_TABLE", "nova-contracts")
+REGION = os.environ.get("AWS_DEFAULT_REGION", "us-east-1")
+LAST_APPLIED_SUFFIX = "#LAST_APPLIED"
+
+
+def _get_table():
+    """Return the DynamoDB table resource, or raise if boto3 unavailable."""
+    if boto3 is None:
+        raise RuntimeError("boto3 is required for env_transition")
+    session = boto3.Session(region_name=REGION)
+    dyn = session.resource("dynamodb")
+    return dyn.Table(TABLE_NAME)
+
+
+def detect_prior_env(contract_id: str, consumer_repo: str, new_env: str) -> Optional[str]:
+    """Query the nova-contracts table for the last-applied env.
+
+    Returns the prior env name if it differs from new_env, else None.
+    Failures to reach DynamoDB log a warning and return None (conservative).
+    """
+    try:
+        table = _get_table()
+        sk_prefix = f"{contract_id}{LAST_APPLIED_SUFFIX}#"
+        resp = table.query(
+            KeyConditionExpression="consumerRepo = :repo AND begins_with(#sk, :prefix)",
+            FilterExpression="#status = :status",
+            ExpressionAttributeNames={
+                "#sk": "contractId#submittedAt",
+                "#status": "status",
+            },
+            ExpressionAttributeValues={
+                ":repo": consumer_repo,
+                ":prefix": sk_prefix,
+                ":status": "applied",
+            },
+            ScanIndexForward=False,
+            Limit=1,
+        )
+        items = resp.get("Items", [])
+        if not items:
+            return None
+        prior_env = items[0].get("environment")
+        if prior_env and prior_env != new_env:
+            return prior_env
+        return None
+    except Exception as exc:
+        sys.stderr.write(
+            f"WARNING: env_transition.detect_prior_env: could not query "
+            f"DynamoDB table {TABLE_NAME} — {type(exc).__name__}: {exc}. "
+            f"Assuming no prior env (conservative). This is expected in "
+            f"local/CI mode without the nova-contracts table.\n"
+        )
+        return None
+
+
+def record_applied_env(contract_id: str, consumer_repo: str, env: str) -> bool:
+    """Write a LAST_APPLIED record to the nova-contracts table.
+
+    Called after a successful apply. Idempotent (writes a new timestamped
+    record each time; the detect step reads the latest by ScanIndexForward).
+    Returns True on success, False on failure (non-fatal — the pipeline
+    should not halt if the record write fails).
+    """
+    try:
+        table = _get_table()
+        ts = datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+        sk = f"{contract_id}{LAST_APPLIED_SUFFIX}#{ts}"
+        table.put_item(
+            Item={
+                "consumerRepo": consumer_repo,
+                "contractId#submittedAt": sk,
+                "contractId": contract_id,
+                "environment": env,
+                "status": "applied",
+                "appliedAt": ts,
+            }
+        )
+        return True
+    except Exception as exc:
+        sys.stderr.write(
+            f"WARNING: env_transition.record_applied_env: could not write to "
+            f"DynamoDB table {TABLE_NAME} — {type(exc).__name__}: {exc}. "
+            f"The apply succeeded but the last-applied env record was not "
+            f"persisted. Future env-transition detection may not work.\n"
+        )
+        return False
+
+
+def main(argv):
+    import argparse
+
+    parser = argparse.ArgumentParser(description="Nova env-transition detect/record")
+    sub = parser.add_subparsers(dest="command", required=True)
+
+    p_detect = sub.add_parser("detect", help="Detect prior env for a contract")
+    p_detect.add_argument("--contract-id", required=True)
+    p_detect.add_argument("--consumer-repo", required=True)
+    p_detect.add_argument("--new-env", required=True)
+
+    p_record = sub.add_parser("record", help="Record the applied env for a contract")
+    p_record.add_argument("--contract-id", required=True)
+    p_record.add_argument("--consumer-repo", required=True)
+    p_record.add_argument("--env", required=True)
+
+    args = parser.parse_args(argv[1:])
+
+    if args.command == "detect":
+        prior = detect_prior_env(args.contract_id, args.consumer_repo, args.new_env)
+        print(json.dumps({"prior_env": prior}))
+        return 0 if prior is None else 0
+    elif args.command == "record":
+        ok = record_applied_env(args.contract_id, args.consumer_repo, args.env)
+        print(json.dumps({"recorded": ok}))
+        return 0 if ok else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv))