Merge phase/03-submission-readiness — v1.17.3 (v1.18 P3 submission-readiness complete)

This commit is contained in:
Jon Chery
2026-08-06 15:09:32 +00:00
5 changed files with 656 additions and 1 deletions
+19
View File
@@ -500,3 +500,22 @@ def lambda_handler(event, context):
return {"statusCode": 400, "body": json.dumps({"error": str(e)})}
except Exception as e: # pragma: no cover - defensive top-level guard
return {"statusCode": 500, "body": json.dumps({"error": str(e)})}
# --- CLI: --check-readiness (D-133, REQ-218) ---------------------------
# Invoked as: python3 -m core.lambda.contract_ingestor --check-readiness <submission.json>
# Delegates to core.submission_readiness.check_readiness() and prints the
# structured ReadinessResult. Exits 0 if ready, 1 if not.
if __name__ == "__main__": # pragma: no cover - CLI entry
import sys
if "--check-readiness" in sys.argv:
sys.path.insert(
0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
)
from core.submission_readiness import cli_main
# Strip the --check-readiness flag; pass the file path.
rest = [a for a in sys.argv[1:] if a != "--check-readiness"]
sys.exit(cli_main(["check-readiness"] + rest))
else:
print("Usage: python3 -m core.lambda.contract_ingestor --check-readiness <submission.json>")
+193
View File
@@ -0,0 +1,193 @@
"""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))
+150
View File
@@ -0,0 +1,150 @@
# Submission Readiness — What is Acceptable to Start
> **Source of truth:** `schemas/submission-readiness.schema.json` (v1.18,
> REQ-217). The validator is `core/submission_readiness.py` (REQ-218),
> invoked as `python3 -m core.lambda.contract_ingestor --check-readiness
> <submission.json>` (D-133).
Nova's submission-readiness gate defines what is **acceptable to start**.
It is a superset gate *above* contract-schema validity: the contract schema
(`schemas/contract.schema.json`) defines the **shape** (id / name /
environment / infrastructure); the readiness schema defines the **gate**
(tags, per-env mandatory metadata, policy preconditions, profile markers,
appSource). Both must pass before ingestion proceeds.
## How It Works
```
citizen developer submits
contract.schema.json validation (shape) ← the existing check
submission-readiness.schema.json (gate) ← the new check
├── contractId present (non-empty)
├── environment valid (dev/qa/prod/dr)
├── tags: all 5 Nova tags present (D-054)
├── policyPreconditions declared
├── profile: developer or agentic
│ └── if agentic: naturalLanguageIntent + confidenceAtSubmission + agentTrace
├── appSource: repo + ref (for runtime fetch)
└── per-env mandatory (W3.E):
dev → stack + environment
qa → + validation.e2eSuite + validation.loadTest
prod → + runbook + dashboard + oncall
dr → + drDrillRef
ready → proceed to contract ingestion
not ready → reject with citizen-developer-facing error (reason code)
```
## Reason Codes
When a submission is not ready, the validator returns one or more reason
codes. These are citizen-developer-facing — no stack traces.
| Code | Meaning |
|---|---|
| `MISSING_TAGS:<tag1>,<tag2>` | One or more required Nova tags are absent |
| `ENV_MISSING_MANDATORY:<env>:<field>` | A per-env mandatory field (W3.E) is missing |
| `AGENTIC_MISSING_INTENT:<marker>` | profile=agentic but a required marker is absent |
| `MISSING_APP_SOURCE` | appSource (repo + ref) is missing |
| `POLICY_PRECONDITION_MISSING` | No policy preconditions declared |
| `CONTRACT_SCHEMA_INVALID:<detail>` | The contract shape failed contract.schema.json |
| `READINESS_SCHEMA_INVALID:<detail>` | The submission failed the readiness schema |
## Good Example
```json
{
"contractId": "uuid-1234",
"id": "webapi",
"name": "Customer Web API",
"environment": "dev",
"tags": {
"nova:owner": "consumer-repo",
"nova:contract": "uuid-1234",
"nova:environment": "dev",
"nova:cost-center": "nova-default",
"nova:ref": "CHG0678912"
},
"policyPreconditions": {
"public-ingress": false,
"encryption_enabled": true,
"deletion_protection": true
},
"profile": "developer",
"appSource": {
"repo": "consumer/web-api",
"ref": "main"
},
"infrastructure": {
"static-assets": {
"inputs": {
"bucket_name": "webapi-assets"
}
}
}
}
```
Result: **READY** — passes the shape + the gate.
## Rejected Examples
### Missing Tags
```json
{
"contractId": "uuid-1234",
"environment": "dev",
"tags": {
"nova:owner": "consumer-repo"
},
"policyPreconditions": {"public-ingress": false},
"profile": "developer",
"appSource": {"repo": "consumer/repo", "ref": "main"}
}
```
Result: `NOT READY — MISSING_TAGS:nova:contract,nova:environment,nova:cost-center,nova:ref`
### Agentic Missing Intent
```json
{
"contractId": "uuid-1234",
"environment": "qa",
"tags": { "nova:owner": "x", "nova:contract": "x", "nova:environment": "qa", "nova:cost-center": "x", "nova:ref": "x" },
"policyPreconditions": {"public-ingress": false},
"profile": "agentic",
"appSource": {"repo": "x", "ref": "x"},
"validation": {"e2eSuite": true, "loadTest": true}
}
```
Result: `NOT READY — AGENTIC_MISSING_INTENT:naturalLanguageIntent; AGENTIC_MISSING_INTENT:confidenceAtSubmission; AGENTIC_MISSING_INTENT:agentTrace`
### Env Missing Mandatory (prod without runbook)
```json
{
"contractId": "uuid-1234",
"environment": "prod",
"tags": { "nova:owner": "x", "nova:contract": "x", "nova:environment": "prod", "nova:cost-center": "x", "nova:ref": "x" },
"policyPreconditions": {"public-ingress": false},
"profile": "developer",
"appSource": {"repo": "x", "ref": "x"}
}
```
Result: `NOT READY — ENV_MISSING_MANDATORY:prod:runbook; ENV_MISSING_MANDATORY:prod:dashboard; ENV_MISSING_MANDATORY:prod:oncall`
## Compliance-Standard Equivalence
The submission-readiness gate applies **equally** to all upstream sources.
Whether the citizen developer's submission originated from an AI coding
agent, an agentic SDLC platform, or a traditional development platform —
the same tags, the same env mandatory, the same policy preconditions, the
same profile markers are required. The source does not matter; the
submission does. This is the RACI compliance-standard equivalence note
(`docs/raci.md`) made machine-checkable.
+104
View File
@@ -0,0 +1,104 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://nova.dev/schemas/submission-readiness.schema.json",
"title": "Nova Submission-Readiness Gate",
"description": "Defines what is acceptable to start — a superset gate ABOVE contract.schema.json validity. The contract schema defines the SHAPE (id/name/environment/infrastructure); this schema defines the READINESS gate: required Nova tags, per-env mandatory metadata (W3.E), declared policy preconditions, profile:agentic markers, and the appSource pointer. The validator (core/submission_readiness.py, REQ-218) calls contract.schema.json validation first, then these readiness checks. On fail → citizen-developer-facing error (not a stack trace); on pass → proceeds to existing contract ingestion.",
"type": "object",
"required": ["contractId", "environment", "tags", "policyPreconditions", "profile", "appSource"],
"properties": {
"contractId": {
"type": "string",
"minLength": 1,
"description": "The contract identifier (UUID or operational id). Non-empty."
},
"environment": {
"type": "string",
"enum": ["dev", "qa", "prod", "dr"],
"description": "Target environment. Determines the per-env mandatory fields (allOf below)."
},
"tags": {
"type": "object",
"description": "The 5 required Nova tags (D-054). References schemas/tagging-standard.json.",
"required": ["nova:owner", "nova:contract", "nova:environment", "nova:cost-center", "nova:ref"],
"properties": {
"nova:owner": {"type": "string", "minLength": 1},
"nova:contract": {"type": "string", "minLength": 1},
"nova:environment": {"type": "string", "enum": ["dev", "qa", "prod", "dr"]},
"nova:cost-center": {"type": "string", "minLength": 1},
"nova:ref": {"type": "string", "minLength": 1}
},
"additionalProperties": false
},
"policyPreconditions": {
"type": "object",
"description": "Declared policy expectations the platform will enforce. The citizen developer states what the platform should check; the platform enforces it at apply time. Missing a declared precondition is POLICY_PRECONDITION_MISSING.",
"properties": {
"public-ingress": {"type": "boolean", "default": false},
"encryption_enabled": {"type": "boolean", "default": true},
"deletion_protection": {"type": "boolean", "default": true}
},
"additionalProperties": true
},
"profile": {
"type": "string",
"enum": ["developer", "agentic"],
"description": "developer = L3A (technical); agentic = L3B (non-technical, requires naturalLanguageIntent + confidenceAtSubmission + agentTrace per REQ-22 / W3.E)."
},
"appSource": {
"type": "object",
"description": "Pointer to the consumer application code so the platform can fetch at run time.",
"required": ["repo", "ref"],
"properties": {
"repo": {"type": "string", "minLength": 1, "description": "Repository URL or owner/repo shorthand."},
"ref": {"type": "string", "minLength": 1, "description": "Git ref (branch, tag, or SHA)."}
},
"additionalProperties": false
},
"naturalLanguageIntent": {
"type": "string",
"description": "Required when profile=agentic (L3B). The citizen developer's plain-language intent."
},
"confidenceAtSubmission": {
"type": "number",
"minimum": 0,
"maximum": 1,
"description": "Required when profile=agentic (L3B). The submitter's self-assessed confidence."
},
"agentTrace": {
"type": "string",
"description": "Required when profile=agentic (L3B). The agent's trace/reasoning for the submission."
},
"validation": {
"type": "object",
"description": "Per-env mandatory metadata (W3.E). qa requires e2eSuite + loadTest; prod requires runbook + dashboard + oncall; dr requires drDrillRef.",
"properties": {
"e2eSuite": {"type": "boolean"},
"loadTest": {"type": "boolean"}
},
"additionalProperties": true
},
"runbook": {"type": "string", "description": "Required when environment=prod (W3.E)."},
"dashboard": {"type": "string", "description": "Required when environment=prod (W3.E)."},
"oncall": {"type": "string", "description": "Required when environment=prod (W3.E)."},
"drDrillRef": {"type": "string", "description": "Required when environment=dr (W3.E)."}
},
"allOf": [
{
"if": {"properties": {"environment": {"const": "qa"}}},
"then": {"required": ["validation"], "properties": {"validation": {"required": ["e2eSuite", "loadTest"]}}}
},
{
"if": {"properties": {"environment": {"const": "prod"}}},
"then": {"required": ["runbook", "dashboard", "oncall"]}
},
{
"if": {"properties": {"environment": {"const": "dr"}}},
"then": {"required": ["drDrillRef"]}
},
{
"if": {"properties": {"profile": {"const": "agentic"}}},
"then": {"required": ["naturalLanguageIntent", "confidenceAtSubmission", "agentTrace"]}
}
],
"additionalProperties": true
}
+189
View File
@@ -0,0 +1,189 @@
"""tests/test_submission_readiness.py — REQ-220.
Covers: good contract passes; missing tags fail with MISSING_TAGS;
env-missing-mandatory fails with ENV_MISSING_MANDATORY:<env>:<field>;
agentic profile missing intent fails with AGENTIC_MISSING_INTENT;
missing appSource fails with MISSING_APP_SOURCE.
"""
import json
import os
import sys
import unittest
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from core.submission_readiness import check_readiness, ReadinessResult
GOOD_TAGS = {
"nova:owner": "consumer-repo",
"nova:contract": "uuid-1234",
"nova:environment": "dev",
"nova:cost-center": "nova-default",
"nova:ref": "CHG0678912",
}
def _base(**overrides):
submission = {
"contractId": "uuid-1234",
"id": "webapi",
"name": "Customer Web API",
"environment": "dev",
"tags": dict(GOOD_TAGS),
"policyPreconditions": {"public-ingress": False, "encryption_enabled": True},
"profile": "developer",
"appSource": {"repo": "consumer/repo", "ref": "main"},
"infrastructure": {
"static-assets": {"inputs": {"bucket_name": "webapi-assets"}}
},
}
submission.update(overrides)
return submission
class TestGoodContract(unittest.TestCase):
def test_good_contract_passes(self):
result = check_readiness(_base())
self.assertTrue(result.ready, f"Expected ready, got: {result.reason_codes}")
self.assertEqual(result.contract_id, "uuid-1234")
def test_good_agentic_contract_passes(self):
submission = _base(
profile="agentic",
naturalLanguageIntent="A web API for customer data",
confidenceAtSubmission=0.85,
agentTrace="LLM generated contract from issue #42",
)
result = check_readiness(submission)
self.assertTrue(result.ready, f"Expected ready, got: {result.reason_codes}")
class TestMissingTags(unittest.TestCase):
def test_missing_tags_fail(self):
submission = _base()
submission["tags"] = {"nova:owner": "consumer-repo"}
result = check_readiness(submission)
self.assertFalse(result.ready)
codes = " ".join(result.reason_codes)
self.assertIn("MISSING_TAGS", codes)
self.assertIn("nova:contract", codes)
self.assertIn("nova:environment", codes)
self.assertIn("nova:cost-center", codes)
self.assertIn("nova:ref", codes)
def test_empty_tag_value_fails(self):
submission = _base()
submission["tags"]["nova:owner"] = ""
result = check_readiness(submission)
self.assertFalse(result.ready)
self.assertTrue(any("MISSING_TAGS" in c for c in result.reason_codes))
class TestEnvMissingMandatory(unittest.TestCase):
def test_qa_missing_e2e_suite_fails(self):
submission = _base(environment="qa")
submission["tags"]["nova:environment"] = "qa"
# No validation.e2eSuite
result = check_readiness(submission)
self.assertFalse(result.ready)
codes = " ".join(result.reason_codes)
self.assertIn("ENV_MISSING_MANDATORY:qa:validation.e2eSuite", codes)
def test_prod_missing_runbook_fails(self):
submission = _base(environment="prod")
submission["tags"]["nova:environment"] = "prod"
# No runbook/dashboard/oncall
result = check_readiness(submission)
self.assertFalse(result.ready)
codes = " ".join(result.reason_codes)
self.assertIn("ENV_MISSING_MANDATORY:prod:runbook", codes)
self.assertIn("ENV_MISSING_MANDATORY:prod:dashboard", codes)
self.assertIn("ENV_MISSING_MANDATORY:prod:oncall", codes)
def test_dr_missing_drdrillref_fails(self):
submission = _base(environment="dr")
submission["tags"]["nova:environment"] = "dr"
result = check_readiness(submission)
self.assertFalse(result.ready)
codes = " ".join(result.reason_codes)
self.assertIn("ENV_MISSING_MANDATORY:dr:drDrillRef", codes)
def test_prod_with_all_mandatory_passes(self):
submission = _base(
environment="prod",
runbook="docs/runbooks/webapi.md",
dashboard="https://grafana/nova/webapi",
oncall="oncall@company.com",
)
submission["tags"]["nova:environment"] = "prod"
result = check_readiness(submission)
self.assertTrue(result.ready, f"Expected ready, got: {result.reason_codes}")
class TestAgenticMissingIntent(unittest.TestCase):
def test_agentic_missing_all_markers_fails(self):
submission = _base(profile="agentic")
result = check_readiness(submission)
self.assertFalse(result.ready)
codes = " ".join(result.reason_codes)
self.assertIn("AGENTIC_MISSING_INTENT:naturalLanguageIntent", codes)
self.assertIn("AGENTIC_MISSING_INTENT:confidenceAtSubmission", codes)
self.assertIn("AGENTIC_MISSING_INTENT:agentTrace", codes)
def test_agentic_missing_one_marker_fails(self):
submission = _base(
profile="agentic",
naturalLanguageIntent="A web API",
confidenceAtSubmission=0.85,
# agentTrace missing
)
result = check_readiness(submission)
self.assertFalse(result.ready)
self.assertTrue(any("agentTrace" in c for c in result.reason_codes))
class TestMissingAppSource(unittest.TestCase):
def test_missing_appsource_fails(self):
submission = _base()
del submission["appSource"]
result = check_readiness(submission)
self.assertFalse(result.ready)
self.assertTrue(any("MISSING_APP_SOURCE" in c for c in result.reason_codes))
def test_appsource_missing_ref_fails(self):
submission = _base()
submission["appSource"] = {"repo": "consumer/repo"}
result = check_readiness(submission)
self.assertFalse(result.ready)
self.assertTrue(any("MISSING_APP_SOURCE" in c for c in result.reason_codes))
class TestPolicyPreconditionMissing(unittest.TestCase):
def test_empty_policy_fails(self):
submission = _base()
submission["policyPreconditions"] = {}
result = check_readiness(submission)
self.assertFalse(result.ready)
self.assertTrue(any("POLICY_PRECONDITION_MISSING" in c for c in result.reason_codes))
class TestReadinessResultStructure(unittest.TestCase):
def test_result_to_dict(self):
result = check_readiness(_base())
d = result.to_dict()
self.assertIn("ready", d)
self.assertIn("reason_codes", d)
self.assertIn("contractId", d)
def test_result_str_ready(self):
result = check_readiness(_base())
self.assertIn("READY", str(result))
def test_result_str_not_ready(self):
submission = _base()
del submission["appSource"]
result = check_readiness(submission)
self.assertIn("NOT READY", str(result))
if __name__ == "__main__":
unittest.main()