verify(P14): schema-driven-outputs-and-cache — 4-layer verify PASS + ship

VERIFY: structural — schema-driven outputs + cache; behavioral — 47 tests + CI PASS; quality — mid-milestone checkpoint clean.

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

core/contract_resolver.py

@@ -64,6 +64,21 @@ def _load_json(path):
         return json.load(fh)
 
 
+# P14 (REQ-178): cache loaded JSON schemas so resolve() doesn't re-read
+# from disk on every call.
+_SCHEMA_CACHE: dict = {}
+
+
+def _load_schema(path):
+    """Load a JSON schema with caching (P14, REQ-178)."""
+    cached = _SCHEMA_CACHE.get(path)
+    if cached is not None:
+        return cached
+    schema = _load_json(path)
+    _SCHEMA_CACHE[path] = schema
+    return schema
+
+
 def _load_yaml(path):
     with open(path, "r") as fh:
         return yaml.safe_load(fh)
@@ -437,24 +452,9 @@ def _namespace_resources(resources, module_name):
 
 
 def decommission_transform(stack_instance):
-    """REQ-92: Transform a resolved stack instance for decommission.
+    """REQ-92: re-export from core.decommission_transform (P12, REQ-176)."""
-
+    from core.decommission_transform import decommission_transform as _dt
-    Sets all scalable counts to 0 and deletion_protection to false on
+    return _dt(stack_instance)
-    every resource. Used by the decommission pipeline mode after the
-    first step (disable deletion protection) has been applied.
-    """
-    for res in stack_instance.get("resources", []):
-        if "nfrs" not in res:
-            res["nfrs"] = {}
-        res["nfrs"]["deletion_protection"] = False
-        inputs = res.get("inputs", {})
-        if "desired_count" in inputs:
-            inputs["desired_count"] = 0
-        if "min_capacity" in inputs:
-            inputs["min_capacity"] = 0
-        if "max_capacity" in inputs:
-            inputs["max_capacity"] = 0
-    return stack_instance
 
 
 def resolve(contract_path, repo_root=None, environment_override=None):
@@ -483,7 +483,7 @@ def resolve(contract_path, repo_root=None, environment_override=None):
         contract["environment"] = environment_override
 
     # Load schemas
-    contract_schema = _load_json(os.path.join(repo_root, "schemas", "contract.schema.json"))
+    contract_schema = _load_schema(os.path.join(repo_root, "schemas", "contract.schema.json"))
 
     # Validate contract against schema
     jsonschema.validate(contract, contract_schema)
@@ -603,27 +603,13 @@ def resolve(contract_path, repo_root=None, environment_override=None):
         stack_instance["outputs"] = merged_outputs
 
     # Validate against stack schema
-    stack_schema = _load_json(os.path.join(repo_root, "schemas", "stack.schema.json"))
+    stack_schema = _load_schema(os.path.join(repo_root, "schemas", "stack.schema.json"))
     jsonschema.validate(stack_instance, stack_schema)
 
     return stack_instance
 
 
 if __name__ == "__main__":
-    if len(sys.argv) < 3:
+    # P12 (REQ-176): CLI extracted to core/contract_resolver_cli.py.
-        print("usage: contract_resolver.py <contract.yml> <out.json> [--environment <name>]", file=sys.stderr)
+    from core.contract_resolver_cli import main
-        sys.exit(2)
+    sys.exit(main())
-    contract_path = sys.argv[1]
-    out_path = sys.argv[2]
-    env_override = None
-    if "--environment" in sys.argv:
-        idx = sys.argv.index("--environment")
-        if idx + 1 < len(sys.argv):
-            env_override = sys.argv[idx + 1]
-    # Also honor the NOVA_ENVIRONMENT_OVERRIDE env var (used by run_platform.sh).
-    # Dual-read via core/env.py: NOVA_* preferred, ACDL_* fallback until P5.
-    if env_override is None and env.get_env("ENVIRONMENT_OVERRIDE"):
-        env_override = env.get_env("ENVIRONMENT_OVERRIDE")
-    result = resolve(contract_path, environment_override=env_override)
-    with open(out_path, "w") as fh:
-        json.dump(result, fh, indent=2)

core/contract_resolver_cli.py

@@ -0,0 +1,41 @@
+"""Nova Contract Resolver CLI — command-line entry point.
+
+Extracted from core/contract_resolver.py (P12, REQ-176).
+
+G-113 import direction: this module imports core.contract_resolver (the
+re-export shim) for the resolve function. The shim imports the split
+modules. Nothing imports this CLI module except direct invocation.
+"""
+from __future__ import annotations
+
+import json
+import sys
+
+from core.contract_resolver import resolve
+from core import env
+
+
+def main(argv=None):
+    """CLI: resolve a contract YAML to a Target Stack JSON."""
+    argv = argv if argv is not None else sys.argv[1:]
+    if len(argv) < 2:
+        print("usage: contract_resolver.py <contract.yml> <out.json> [--environment <name>", file=sys.stderr)
+        return 2
+    contract_path = argv[0]
+    out_path = argv[1]
+    env_override = None
+    if "--environment" in argv:
+        idx = argv.index("--environment")
+        if idx + 1 < len(argv):
+            env_override = argv[idx + 1]
+    # Also honor the NOVA_ENVIRONMENT_OVERRIDE env var (used by run_platform.sh).
+    if env_override is None and env.get_env("ENVIRONMENT_OVERRIDE"):
+        env_override = env.get_env("ENVIRONMENT_OVERRIDE")
+    result = resolve(contract_path, environment_override=env_override)
+    with open(out_path, "w") as fh:
+        json.dump(result, fh, indent=2)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

core/decommission_transform.py

@@ -0,0 +1,31 @@
+"""Nova Decommission Transform — zero counts + disable deletion protection (REQ-92).
+
+Extracted from core/contract_resolver.py (P12, REQ-176).
+
+G-113 import direction: this module imports only stdlib. The re-export
+shim core/contract_resolver.py imports this module. Nothing imports the
+shim except external callers.
+"""
+
+from __future__ import annotations
+
+
+def decommission_transform(stack_instance):
+    """REQ-92: Transform a resolved stack instance for decommission.
+
+    Sets all scalable counts to 0 and deletion_protection to false on
+    every resource. Used by the decommission pipeline mode after the
+    first step (disable deletion protection) has been applied.
+    """
+    for res in stack_instance.get("resources", []):
+        if "nfrs" not in res:
+            res["nfrs"] = {}
+        res["nfrs"]["deletion_protection"] = False
+        inputs = res.get("inputs", {})
+        if "desired_count" in inputs:
+            inputs["desired_count"] = 0
+        if "min_capacity" in inputs:
+            inputs["min_capacity"] = 0
+        if "max_capacity" in inputs:
+            inputs["max_capacity"] = 0
+    return stack_instance

core/output_publisher.py

@@ -38,8 +38,10 @@ from core import env as _envhelper
 SSM_PREFIX = "/nova"
 KMS_KEY_ID_ENV = "NOVA_KMS_KEY_ID"
 
-# Outputs that are safe to display in a PR comment (no secrets).
+# P14 (REQ-178): SAFE_OUTPUT_NAMES is schema-driven (derived from
-SAFE_OUTPUT_NAMES = {
+# modules/l1/*/interface.json outputs that don't have sensitive:true).
+# Falls back to the hardcoded set if the interfaces can't be read.
+_HARDCODED_SAFE_OUTPUTS = {
     "distribution_domain_name",
     "bucket_arn",
     "bucket_name",
@@ -59,6 +61,37 @@ SAFE_OUTPUT_NAMES = {
 }
 
 
+def _load_safe_output_names():
+    """Derive the safe-output allowlist from interface.json outputs.
+
+    P14 (REQ-178): scan modules/l1/*/interface.json; an output is safe if
+    its spec does not set sensitive:true. Falls back to the hardcoded set
+    if no interfaces are readable.
+    """
+    import json
+    from pathlib import Path
+    root = Path(__file__).resolve().parent.parent
+    safe = set()
+    try:
+        for iface in (root / "modules" / "l1").glob("*/interface.json"):
+            d = json.loads(iface.read_text())
+            outs = d.get("outputs", {})
+            if isinstance(outs, dict):
+                for name, spec in outs.items():
+                    if not (isinstance(spec, dict) and spec.get("sensitive")):
+                        safe.add(name)
+            elif isinstance(outs, list):
+                for out in outs:
+                    if isinstance(out, dict) and not out.get("sensitive"):
+                        safe.add(out.get("name", ""))
+    except (OSError, ValueError):
+        pass
+    return safe or _HARDCODED_SAFE_OUTPUTS
+
+
+SAFE_OUTPUT_NAMES = _load_safe_output_names()
+
+
 def _ssm_client():
     if boto3 is None:
         raise RuntimeError("boto3 is required for SSM publishing")

core/regression_verify.py

@@ -668,17 +668,9 @@ def write_report(report: RegressionReport,
 
 
 def main() -> int:
-    milestone = _envhelper.get_env("REGRESSION_MILESTONE", "v1.10") or "v1.10"
+    """P13 (REQ-177): re-export from core.regression_verify_cli."""
-    phase = int(_envhelper.get_env("REGRESSION_PHASE", "52") or "52")
+    from core.regression_verify_cli import main as _cli_main
-    report = run_regression(milestone=milestone, phase=phase)
+    return _cli_main()
-    md, js = write_report(report)
-    print(f"regression: {report.summary} -> {md}")
-    if not report.passed:
-        print("FAIL: regression surfaced non-Verified/non-Skipped capabilities "
-              "(milestone gate blocks)", file=sys.stderr)
-        return 1
-    print(f"regression: gate passes (summary={report.summary})")
-    return 0
 
 
 if __name__ == "__main__":

core/regression_verify_cli.py

@@ -0,0 +1,33 @@
+"""Nova Regression Verify CLI — command-line entry point.
+
+Extracted from core/regression_verify.py (P13, REQ-177).
+
+G-113 import direction: this module imports core.regression_verify (the
+library) for run_regression + write_report. The library does not import
+this CLI module. Nothing imports this CLI except direct invocation.
+"""
+from __future__ import annotations
+
+import sys
+
+from core import env as _envhelper
+from core.regression_verify import run_regression, write_report
+
+
+def main(argv=None):
+    """CLI: run the regression gate and write the report."""
+    milestone = _envhelper.get_env("REGRESSION_MILESTONE", "v1.10") or "v1.10"
+    phase = int(_envhelper.get_env("REGRESSION_PHASE", "52") or "52")
+    report = run_regression(milestone=milestone, phase=phase)
+    md, js = write_report(report)
+    print(f"regression: {report.summary} -> {md}")
+    if not report.passed:
+        print("FAIL: regression surfaced non-Verified/non-Skipped capabilities "
+              "(milestone gate blocks)", file=sys.stderr)
+        return 1
+    print(f"regression: gate passes (summary={report.summary})")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())