verify(P6): run-platform-deadcode-and-hitl-fn — 4-layer verify PASS + ship

VERIFY: structural — HITL fn extracted + deadcode/config; behavioral — syntax clean + CI PASS; quality — ~14 lines saved.

---ci---
project: acdl
phase: 6
milestone: v1.16
status: complete
phase_role: execution
requirements:
  covered: [REQ-170]
  partial: []
---/ci---

core/regression_verify.py

@@ -146,14 +146,18 @@ def _check_environment_schema_validation() -> Tuple[Status, str]:
     ])
 
 
-def _check_resolver_static_assets() -> Tuple[Status, str]:
+def _check_resolver(contract_path: str) -> Tuple[Status, str]:
-    """CAP-003: contract_resolver resolves static-assets to a Target Stack."""
+    """Shared helper: contract_resolver resolves a contract to a Target Stack.
+
+    Used by CAP-003 (static-assets) and CAP-004 (microservice) — the two
+    were ~95% identical except the contract path (P5 dedup, REQ-169).
+    """
     with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as t:
         out = t.name
     try:
         return _check_subprocess([
             "python3", "core/contract_resolver.py",
-            "contracts/static-assets.yml", out,
+            contract_path, out,
         ])
     finally:
         try:
@@ -162,20 +166,14 @@ def _check_resolver_static_assets() -> Tuple[Status, str]:
             pass
 
 
+def _check_resolver_static_assets() -> Tuple[Status, str]:
+    """CAP-003: contract_resolver resolves static-assets to a Target Stack."""
+    return _check_resolver("contracts/static-assets.yml")
+
+
 def _check_resolver_microservice() -> Tuple[Status, str]:
     """CAP-004: contract_resolver resolves the microservice contract."""
-    with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as t:
+    return _check_resolver("contracts/microservice.yml")
-        out = t.name
-    try:
-        return _check_subprocess([
-            "python3", "core/contract_resolver.py",
-            "contracts/microservice.yml", out,
-        ])
-    finally:
-        try:
-            os.unlink(out)
-        except OSError:
-            pass
 
 
 def _check_adapter_emits_terraform() -> Tuple[Status, str]:
@@ -325,21 +323,24 @@ def _load_aws_env() -> Dict[str, str]:
     return env
 
 
-def _check_live_terraform_plan_microservice() -> Tuple[Status, str]:
+def _check_live_terraform_plan(contract_path: str, label: str) -> Tuple[Status, str]:
-    """CAP-013: terraform init+validate+plan against live AWS for the
+    """Shared helper: terraform init+validate+plan against live AWS for a
-    microservice stack (D-093 live-AWS tier of the headline E2E).
+    contract (D-093 live-AWS tier of the headline E2E).
 
-    Requires AWS credentials (NOVA_AWS_ACCESS_KEY_ID etc. in .env.secrets;
+    Used by CAP-013 (microservice) and CAP-014 (static-assets) — the two
-    NOVA_* only — the ACDL_* fallback was removed in v1.15 P5, REQ-164).
+    were ~95% identical except the contract path + label (P5 dedup,
-    Runs in a temp dir; does NOT apply (plan only)."""
+    REQ-169). Requires AWS credentials (NOVA_AWS_ACCESS_KEY_ID etc. in
+    .env.secrets; NOVA_* only — the ACDL_* fallback was removed in v1.15
+    P5, REQ-164). Runs in a temp dir; does NOT apply (plan only).
+    """
     import tempfile, os
-    work = tempfile.mkdtemp(prefix="nova_regr_live_")
+    work = tempfile.mkdtemp(prefix=f"nova_regr_live_{label}_")
     stack_path = os.path.join(work, "stack.json")
     tf_dir = os.path.join(work, "tf")
     os.makedirs(tf_dir, exist_ok=True)
     rc, out, err = _run_subprocess([
         "python3", "core/contract_resolver.py",
-        "contracts/microservice.yml", stack_path,
+        contract_path, stack_path,
     ])
     if rc != 0:
         return "Broken", f"resolver failed: {err.strip()[-200:]}"
@@ -366,47 +367,19 @@ def _check_live_terraform_plan_microservice() -> Tuple[Status, str]:
     )
     if rc != 0:
         return "Decayed", f"terraform plan failed: {err.strip()[-200:]}"
-    return "Verified", "terraform init+validate+plan OK (live AWS, microservice)"
+    return "Verified", f"terraform init+validate+plan OK (live AWS, {label})"
+
+
+def _check_live_terraform_plan_microservice() -> Tuple[Status, str]:
+    """CAP-013: terraform init+validate+plan against live AWS for the
+    microservice stack (D-093 live-AWS tier of the headline E2E)."""
+    return _check_live_terraform_plan("contracts/microservice.yml", "microservice")
 
 
 def _check_live_terraform_plan_static_assets() -> Tuple[Status, str]:
     """CAP-014: terraform init+validate+plan against live AWS for the
     static-assets stack (CloudFront + WAF + S3)."""
-    import tempfile, os
+    return _check_live_terraform_plan("contracts/static-assets.yml", "static-assets")
-    work = tempfile.mkdtemp(prefix="nova_regr_live_sa_")
-    stack_path = os.path.join(work, "stack.json")
-    tf_dir = os.path.join(work, "tf")
-    os.makedirs(tf_dir, exist_ok=True)
-    rc, out, err = _run_subprocess([
-        "python3", "core/contract_resolver.py",
-        "contracts/static-assets.yml", stack_path,
-    ])
-    if rc != 0:
-        return "Broken", f"resolver failed: {err.strip()[-200:]}"
-    rc, out, err = _run_subprocess([
-        "python3", "adapters/terraform/adapter.py", stack_path, tf_dir,
-    ])
-    if rc != 0:
-        return "Broken", f"adapter failed: {err.strip()[-200:]}"
-    env = _load_aws_env()
-    rc, out, err = _run_subprocess(
-        ["terraform", "init", "-reconfigure", "-lock=false", "-input=false"],
-        cwd=tf_dir, timeout=120, env=env,
-    )
-    if rc != 0:
-        return "Broken", f"terraform init failed: {err.strip()[-200:]}"
-    rc, out, err = _run_subprocess(
-        ["terraform", "validate"], cwd=tf_dir, timeout=60, env=env,
-    )
-    if rc != 0:
-        return "Broken", f"terraform validate failed: {err.strip()[-200:]}"
-    rc, out, err = _run_subprocess(
-        ["terraform", "plan", "-lock=false", "-input=false", "-out=tfplan"],
-        cwd=tf_dir, timeout=180, env=env,
-    )
-    if rc != 0:
-        return "Decayed", f"terraform plan failed: {err.strip()[-200:]}"
-    return "Verified", "terraform init+validate+plan OK (live AWS, static-assets)"
 
 
 def _check_dynamodb_outbox_table() -> Tuple[Status, str]:
@@ -474,16 +447,30 @@ def _check_lifecycle_module_terraform(module: str) -> Tuple[Status, str]:
         ["terraform", "fmt", "-check", "-diff", str(tf_dir)], timeout=30)
     if rc != 0:
         return "Broken", f"terraform fmt -check failed: {err.strip()[-200:]}"
+    status, detail = _assert_contracts_resolve(ROOT / "modules" / "l1" / module, "l1")
+    if status != "Verified":
+        return status, detail
+    return "Verified", f"terraform files present + fmt -check passes + simple/complex contracts resolve"
+
+
+def _assert_contracts_resolve(module_dir: Path, level: str) -> Tuple[Status, str]:
+    """Shared helper: assert an L1/L2 module's example contracts resolve.
+
+    Used by _check_lifecycle_module_terraform (L1) and
+    _check_lifecycle_l2_module (L2) — the two had a duplicated
+    for-ex-in-simple-complex-resolve block (P5 dedup, REQ-169).
+    ``level`` is "l1" or "l2" (selects the examples dir parent).
+    """
     for ex in ["simple", "complex"]:
-        contract = ROOT / "modules" / "l1" / module / "examples" / f"{ex}.yml"
+        contract = module_dir / "examples" / f"{ex}.yml"
         if not contract.is_file():
-            return "Broken", f"modules/l1/{module}/examples/{ex}.yml missing"
+            return "Broken", f"{module_dir.relative_to(ROOT)}/examples/{ex}.yml missing"
         rc, out, err = _run_subprocess([
             "python3", "core/contract_resolver.py", str(contract), "/dev/null",
         ], timeout=30)
         if rc != 0:
             return "Broken", f"{ex}.yml resolver failed: {err.strip()[-200:]}"
-    return "Verified", f"terraform files present + fmt -check passes + simple/complex contracts resolve"
+    return "Verified", ""
 
 
 def _check_lifecycle_l2_module(module: str) -> Tuple[Status, str]:
@@ -492,16 +479,11 @@ def _check_lifecycle_l2_module(module: str) -> Tuple[Status, str]:
     This is an offline proxy, not live pipeline evidence; the live
     apply/modify/destroy is verified by the modules-lifecycle workflow
     run, not by this gate."""
-    for ex in ["simple", "complex"]:
+    module_dir = ROOT / "modules" / "l2" / module
-        contract = ROOT / "modules" / "l2" / module / "examples" / f"{ex}.yml"
+    status, detail = _assert_contracts_resolve(module_dir, "l2")
-        if not contract.is_file():
+    if status != "Verified":
-            return "Broken", f"modules/l2/{module}/examples/{ex}.yml missing"
+        return status, detail
-        rc, out, err = _run_subprocess([
+    return "Verified", "L2 composition resolves (simple + complex contracts; offline proxy)"
-            "python3", "core/contract_resolver.py", str(contract), "/dev/null",
-        ], timeout=30)
-        if rc != 0:
-            return "Broken", f"{ex}.yml resolver failed: {err.strip()[-200:]}"
-    return "Verified", f"L2 composition resolves (simple + complex contracts; offline proxy)"
 
 
 def _check_cap_017_dynamodb() -> Tuple[Status, str]:

scripts/run_platform.sh

@@ -113,6 +113,38 @@ fi
 
 fail() { echo "FAIL: $*" >&2; exit 1; }
 
+# run_hitl_gate <contract_id> <resolved_env> <context>
+# REQ-108: for qa/prod/dr, call hitl_gates.attest before apply. Dev skips.
+# Extracted from the two duplicated inline blocks (P6, REQ-170).
+run_hitl_gate() {
+  local _cid="$1" _env="$2" _ctx="$3"
+  if [ "$_env" = "dev" ]; then
+    echo "Environment is $_env — autonomous (no HITL gate)."
+    return 0
+  fi
+  echo "Environment is $_env — HITL attestation gate required$_ctx."
+  local _approver="${GITHUB_ACTOR:-${GITEA_ACTOR:-}}"
+  if [ -z "$_approver" ]; then
+    echo "WARNING: no approver identity (GITHUB_ACTOR/GITEA_ACTOR unset)" >&2
+    echo "         the gate would block in a real CI run. Passing for local." >&2
+  fi
+  python3 -c "
+import os, sys
+sys.path.insert(0, '.')
+from core.hitl_gates import attest
+from core import env as _envhelper
+contract_id = _envhelper.get_env('HITL_CONTRACT_ID') or os.environ['NOVA_HITL_CONTRACT_ID']
+env = _envhelper.get_env('HITL_ENV') or os.environ['NOVA_HITL_ENV']
+approver = _envhelper.get_env('HITL_APPROVER', '') or 'local-test'
+ok, reason = attest(contract_id, env, approver)
+if ok:
+    print(f'HITL PASS: {reason}')
+else:
+    print(f'HITL BLOCK: {reason}', file=sys.stderr)
+    sys.exit(1)
+" NOVA_HITL_CONTRACT_ID="$_cid" NOVA_HITL_ENV="$_env" NOVA_HITL_APPROVER="$_approver"
+}
+
 # --local: run the headline E2E against the local emulating tier (D-092).
 # No AWS credentials, no Checkov, no DynamoDB. Emulates ECS, outbox, S3
 # state, and the contract-ingestor Lambda in-process. Exits 0 on success.
@@ -142,8 +174,8 @@ stream() {
   fi
 }
 
-CONTRACT_ID="11111111-1111-1111-1111-111111111111"  # spike fixed UUID
+CONTRACT_ID="${NOVA_CONTRACT_ID:-11111111-1111-1111-1111-111111111111}"  # spike UUID (override via NOVA_CONTRACT_ID)
-WORK="/tmp/acdl_platform_run_v18"
+WORK="${NOVA_WORK_DIR:-/tmp/nova_platform_run}"
 TF_DIR="$WORK/tf"
 rm -rf "$WORK"; mkdir -p "$TF_DIR"
 
@@ -325,31 +357,7 @@ if [ "$APPLY_ONLY" = "1" ]; then
   if [ -n "$ENVIRONMENT_OVERRIDE" ]; then
     RESOLVED_ENV="$ENVIRONMENT_OVERRIDE"
   fi
-  if [ "$RESOLVED_ENV" != "dev" ]; then
+  run_hitl_gate "$CONTRACT_ID" "$RESOLVED_ENV" " before apply" || { echo "FAIL: HITL attestation gate blocked the apply" >&2; exit 1; }
-    echo "Environment is $RESOLVED_ENV — HITL attestation gate required before apply."
-    APPROVER="${GITHUB_ACTOR:-${GITEA_ACTOR:-}}"
-    if [ -z "$APPROVER" ]; then
-      echo "WARNING: no approver identity (GITHUB_ACTOR/GITEA_ACTOR unset)" >&2
-      echo "         the gate would block in a real CI run. Passing for local." >&2
-    fi
-    python3 -c "
-import os, sys
-sys.path.insert(0, '.')
-from core.hitl_gates import attest
-from core import env as _envhelper
-contract_id = _envhelper.get_env('HITL_CONTRACT_ID') or os.environ['NOVA_HITL_CONTRACT_ID']
-env = _envhelper.get_env('HITL_ENV') or os.environ['NOVA_HITL_ENV']
-approver = _envhelper.get_env('HITL_APPROVER', '') or 'local-test'
-ok, reason = attest(contract_id, env, approver)
-if ok:
-    print(f'HITL PASS: {reason}')
-else:
-    print(f'HITL BLOCK: {reason}', file=sys.stderr)
-    sys.exit(1)
-" NOVA_HITL_CONTRACT_ID="$CONTRACT_ID" NOVA_HITL_ENV="$RESOLVED_ENV" NOVA_HITL_APPROVER="$APPROVER" || { echo "FAIL: HITL attestation gate blocked the apply" >&2; exit 1; }
-  else
-    echo "Environment is dev — autonomous (no HITL gate)."
-  fi
 
   echo ""
   echo "=== Step 5: terraform apply -auto-approve ==="
@@ -441,31 +449,7 @@ RESOLVED_ENV=$(python3 -c "import yaml; print(yaml.safe_load(open('$CONTRACT')).
 if [ -n "$ENVIRONMENT_OVERRIDE" ]; then
   RESOLVED_ENV="$ENVIRONMENT_OVERRIDE"
 fi
-if [ "$RESOLVED_ENV" != "dev" ]; then
+run_hitl_gate "$CONTRACT_ID" "$RESOLVED_ENV" "" || { echo "FAIL: HITL attestation gate blocked the promotion" >&2; exit 1; }
-  echo "Environment is $RESOLVED_ENV — HITL attestation gate required."
-  APPROVER="${GITHUB_ACTOR:-${GITEA_ACTOR:-}}"
-  if [ -z "$APPROVER" ]; then
-    echo "WARNING: no approver identity (GITHUB_ACTOR/GITEA_ACTOR unset); " >&2
-    echo "         the gate would block in a real CI run. Passing for local." >&2
-  fi
-  python3 -c "
-import os, sys
-sys.path.insert(0, '.')
-from core.hitl_gates import attest
-from core import env as _envhelper
-contract_id = _envhelper.get_env('HITL_CONTRACT_ID') or os.environ['NOVA_HITL_CONTRACT_ID']
-env = _envhelper.get_env('HITL_ENV') or os.environ['NOVA_HITL_ENV']
-approver = _envhelper.get_env('HITL_APPROVER', '') or 'local-test'
-ok, reason = attest(contract_id, env, approver)
-if ok:
-    print(f'HITL PASS: {reason}')
-else:
-    print(f'HITL BLOCK: {reason}', file=sys.stderr)
-    sys.exit(1)
-" NOVA_HITL_CONTRACT_ID="$CONTRACT_ID" NOVA_HITL_ENV="$RESOLVED_ENV" NOVA_HITL_APPROVER="$APPROVER" || { echo "FAIL: HITL attestation gate blocked the promotion" >&2; exit 1; }
-else
-  echo "Environment is dev — autonomous (no HITL gate)."
-fi
 
 echo ""
 echo "=== Step 8: write evidence event to DynamoDB outbox ==="