acdl/core/submission_readiness.py

"""core/submission_readiness.py — Nova submission-readiness validator (REQ-218).

Defines what is acceptable to start — a superset gate ABOVE
contract.schema.json validity. Invoked as
``contract_ingestor.py --check-readiness`` (D-133). Returns a structured
ReadinessResult (pass/fail per check, with reason codes). On fail → the
ingestor rejects with a citizen-developer-facing error (not a stack
trace). On pass → proceeds to existing contract ingestion.

The validator calls contract.schema.json validation first (the shape),
then the readiness checks (the gate): tags, env mandatory, policy
preconditions, profile:agentic markers, appSource.

Reason codes:
  MISSING_TAGS — one or more required Nova tags are absent
  ENV_MISSING_MANDATORY:<env>:<field> — a per-env mandatory field is missing
  AGENTIC_MISSING_INTENT — profile=agentic but naturalLanguageIntent absent
  MISSING_APP_SOURCE — appSource (repo + ref) is missing
  POLICY_PRECONDITION_MISSING — a declared policy precondition is absent
"""
from __future__ import annotations

import json
import os
import sys
from dataclasses import dataclass, field
from typing import Any

_SCHEMA_DIR = os.path.join(
    os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "schemas"
)

REQUIRED_TAGS = [
    "nova:owner",
    "nova:contract",
    "nova:environment",
    "nova:cost-center",
    "nova:ref",
]

ENV_MANDATORY: dict[str, list[str]] = {
    "dev": [],  # dev requires only the base contract shape (id+environment+infrastructure)
    "qa": ["validation.e2eSuite", "validation.loadTest"],
    "prod": ["runbook", "dashboard", "oncall"],
    "dr": ["drDrillRef"],
}

AGENTIC_REQUIRED = ["naturalLanguageIntent", "confidenceAtSubmission", "agentTrace"]


@dataclass
class ReadinessResult:
    """Structured result of the submission-readiness gate."""

    ready: bool
    reason_codes: list[str] = field(default_factory=list)
    contract_id: str | None = None

    def to_dict(self) -> dict[str, Any]:
        return {
            "ready": self.ready,
            "reason_codes": self.reason_codes,
            "contractId": self.contract_id,
        }

    def __str__(self) -> str:
        if self.ready:
            return f"READY — contract {self.contract_id} passes submission-readiness gate"
        codes = "; ".join(self.reason_codes) if self.reason_codes else "unknown"
        return f"NOT READY — contract {self.contract_id}: {codes}"


def _validate_contract_schema(contract: dict[str, Any]) -> list[str]:
    """Validate the contract against contract.schema.json (the shape).
    Returns a list of reason codes (empty if valid). Falls back to no-op
    if jsonschema or the schema file is unavailable (the contract is
    validated upstream by run_platform.sh in the normal path).
    """
    codes: list[str] = []
    try:
        import jsonschema

        schema_path = os.path.join(_SCHEMA_DIR, "contract.schema.json")
        with open(schema_path) as f:
            schema = json.load(f)
        jsonschema.validate(instance=contract, schema=schema)
    except (OSError, ImportError):
        pass
    except jsonschema.ValidationError as e:
        codes.append(f"CONTRACT_SCHEMA_INVALID:{e.message}")
    return codes


def _get_nested(data: dict[str, Any], dotted_key: str) -> Any:
    parts = dotted_key.split(".")
    val: Any = data
    for p in parts:
        if not isinstance(val, dict) or p not in val:
            return None
        val = val[p]
    return val


def check_readiness(submission: dict[str, Any]) -> ReadinessResult:
    """Run the full submission-readiness gate.

    1. Validate the contract shape (contract.schema.json).
    2. Validate the readiness schema (submission-readiness.schema.json).
    3. Run the semantic readiness checks (tags, env mandatory, agentic, appSource, policy).

    Returns a ReadinessResult. Never raises — all failures are reason codes.
    """
    contract_id = submission.get("contractId") or submission.get("id", "unknown")
    codes: list[str] = []

    # Step 1: contract shape validation
    contract_shape = {k: v for k, v in submission.items() if k in ("id", "name", "environment", "infrastructure")}
    if contract_shape:
        codes.extend(_validate_contract_schema(contract_shape))

    # Step 2: readiness schema validation
    try:
        import jsonschema

        schema_path = os.path.join(_SCHEMA_DIR, "submission-readiness.schema.json")
        with open(schema_path) as f:
            readiness_schema = json.load(f)
        jsonschema.validate(instance=submission, schema=readiness_schema)
    except (OSError, ImportError):
        pass
    except jsonschema.ValidationError as e:
        codes.append(f"READINESS_SCHEMA_INVALID:{e.message}")

    # Step 3: semantic checks (reason codes for citizen-developer-facing errors)

    # 3a: tags
    tags = submission.get("tags", {})
    missing_tags = [t for t in REQUIRED_TAGS if t not in tags or not tags[t]]
    if missing_tags:
        codes.append(f"MISSING_TAGS:{','.join(missing_tags)}")

    # 3b: env mandatory (W3.E per-env table)
    env = submission.get("environment")
    if env and env in ENV_MANDATORY:
        for field_key in ENV_MANDATORY[env]:
            val = _get_nested(submission, field_key)
            if val is None:
                codes.append(f"ENV_MISSING_MANDATORY:{env}:{field_key}")

    # 3c: agentic profile markers
    if submission.get("profile") == "agentic":
        for marker in AGENTIC_REQUIRED:
            if not submission.get(marker):
                codes.append(f"AGENTIC_MISSING_INTENT:{marker}")

    # 3d: appSource
    app_source = submission.get("appSource")
    if not app_source or not app_source.get("repo") or not app_source.get("ref"):
        codes.append("MISSING_APP_SOURCE")

    # 3e: policy preconditions (warn if declared but not enforced this milestone)
    policy = submission.get("policyPreconditions", {})
    if not policy:
        codes.append("POLICY_PRECONDITION_MISSING")

    ready = len(codes) == 0
    return ReadinessResult(ready=ready, reason_codes=codes, contract_id=contract_id)


def cli_main(argv: list[str]) -> int:
    """CLI entry: python3 -m core.submission_readiness <contract.json>

    Also invoked via contract_ingestor.py --check-readiness (D-133).
    Prints the ReadinessResult to stdout; exits 0 if ready, 1 if not.
    """
    if len(argv) < 2:
        print("Usage: submission_readiness <contract.json>", file=sys.stderr)
        return 2
    path = argv[1]
    try:
        with open(path) as f:
            submission = json.load(f)
    except (OSError, json.JSONDecodeError) as e:
        print(f"ERROR: cannot read {path}: {e}", file=sys.stderr)
        return 2
    result = check_readiness(submission)
    print(result)
    print(json.dumps(result.to_dict(), indent=2))
    return 0 if result.ready else 1


if __name__ == "__main__":
    sys.exit(cli_main(sys.argv))