318 lines
11 KiB
Python
318 lines
11 KiB
Python
"""M16 v1 technical release gate.
|
|
|
|
A deterministic aggregation of the evidence M16 produces. It reads results rather than generating
|
|
them: the chaos, soak and security runs happen first and this decides what they add up to.
|
|
|
|
The verdict is mechanical on purpose. "Looks fine" is not a release decision, and a gate that can
|
|
be argued with is a gate that will be.
|
|
|
|
python scripts/m16_release_gate.py --chaos chaos.json --soak soak.json --report gate.json
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
from dataclasses import dataclass, field
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from sqlalchemy import create_engine, text
|
|
from sqlalchemy.orm import Session
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / "backend" / "src"))
|
|
|
|
from modelforge_api.services.invariants import ( # noqa: E402
|
|
InvariantStatus,
|
|
check_invariants,
|
|
)
|
|
|
|
DEFAULT_DATABASE_URL = os.getenv("MODELFORGE_RUNTIME_DATABASE_URL")
|
|
|
|
# A finding in any of these classes blocks the release outright. Everything else is a documented
|
|
# limitation at most.
|
|
HARD_BLOCKERS = (
|
|
"data corruption",
|
|
"privilege escalation",
|
|
"secret leakage",
|
|
"unbounded resource exhaustion",
|
|
"unrecoverable control-plane crash",
|
|
"duplicate production state",
|
|
"unsafe credential reuse",
|
|
"artifact integrity bypass",
|
|
"production alias or recognizer mutation",
|
|
"critical exploitable vulnerability",
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class Check:
|
|
name: str
|
|
passed: bool
|
|
detail: str
|
|
blocking: bool = True
|
|
evidence: dict[str, Any] = field(default_factory=dict)
|
|
|
|
|
|
def _load(path: str | None) -> dict[str, Any] | None:
|
|
if not path:
|
|
return None
|
|
candidate = Path(path)
|
|
if not candidate.is_file():
|
|
return None
|
|
try:
|
|
return dict(json.loads(candidate.read_text("utf-8")))
|
|
except (json.JSONDecodeError, ValueError):
|
|
return None
|
|
|
|
|
|
def invariant_check(database_url: str) -> Check:
|
|
engine = create_engine(database_url, pool_pre_ping=True)
|
|
try:
|
|
with Session(engine) as session:
|
|
report = check_invariants(session)
|
|
finally:
|
|
engine.dispose()
|
|
violations = [
|
|
item.key for item in report.results if item.status is InvariantStatus.VIOLATED
|
|
]
|
|
return Check(
|
|
name="platform invariants",
|
|
passed=report.violated == 0,
|
|
detail=f"{report.holding}/{report.checked} hold",
|
|
evidence={"violations": violations},
|
|
)
|
|
|
|
|
|
def chaos_check(report: dict[str, Any] | None) -> Check:
|
|
if report is None:
|
|
return Check("chaos scenarios", False, "no chaos report supplied")
|
|
scenarios = report.get("scenarios", [])
|
|
failed = [item["scenario"] for item in scenarios if item.get("outcome") != "PASSED"]
|
|
unclean = [
|
|
item["scenario"]
|
|
for item in scenarios
|
|
if item.get("cleanup", "NOT_REQUIRED") not in ("DONE", "NOT_REQUIRED")
|
|
]
|
|
return Check(
|
|
name="chaos scenarios",
|
|
passed=not failed and not unclean,
|
|
detail=f"{len(scenarios) - len(failed)}/{len(scenarios)} passed",
|
|
evidence={"failed": failed, "unclean_cleanup": unclean, "seed": report.get("seed")},
|
|
)
|
|
|
|
|
|
def soak_check(report: dict[str, Any] | None, minimum_minutes: float) -> Check:
|
|
if report is None:
|
|
return Check("soak", False, "no soak report supplied")
|
|
duration = float(report.get("actual_duration_seconds", 0)) / 60
|
|
after = report.get("resources_after", {})
|
|
leases = int(after.get("active_leases", -1))
|
|
in_flight = int(after.get("serving_jobs_in_flight", -1))
|
|
violated = int(report.get("invariants_final", {}).get("violated", 1))
|
|
internal = sum(
|
|
int(entry.get("internal_errors", 0)) for entry in report.get("by_kind", {}).values()
|
|
)
|
|
passed = (
|
|
duration >= minimum_minutes
|
|
and leases == 0
|
|
and in_flight == 0
|
|
and violated == 0
|
|
)
|
|
return Check(
|
|
name="soak",
|
|
passed=passed,
|
|
detail=(
|
|
f"{duration:.1f} min, {report.get('total_requests', 0)} requests, "
|
|
f"{leases} active leases, {in_flight} jobs in flight, {internal} internal errors"
|
|
),
|
|
evidence={
|
|
"duration_minutes": round(duration, 2),
|
|
"minimum_minutes": minimum_minutes,
|
|
"total_requests": report.get("total_requests"),
|
|
"internal_errors": internal,
|
|
"invariants_violated": violated,
|
|
"row_growth": report.get("row_growth"),
|
|
},
|
|
)
|
|
|
|
|
|
def orphan_check(database_url: str) -> Check:
|
|
engine = create_engine(database_url, pool_pre_ping=True)
|
|
try:
|
|
with Session(engine) as session:
|
|
leases = int(
|
|
session.execute(
|
|
text(
|
|
"select count(*) from serving_gpu_leases where released_at is null "
|
|
"and state in ('pending','granted','active','held')"
|
|
)
|
|
).scalar_one()
|
|
)
|
|
jobs = int(
|
|
session.execute(
|
|
text(
|
|
"select count(*) from serving_jobs "
|
|
"where status in ('queued','leased','running')"
|
|
)
|
|
).scalar_one()
|
|
)
|
|
finally:
|
|
engine.dispose()
|
|
return Check(
|
|
name="no orphaned work",
|
|
passed=leases == 0 and jobs == 0,
|
|
detail=f"{leases} active leases, {jobs} jobs in flight",
|
|
evidence={"active_leases": leases, "serving_jobs_in_flight": jobs},
|
|
)
|
|
|
|
|
|
def alert_check(database_url: str) -> Check:
|
|
"""Artificial alerts raised during the gate must not be left firing."""
|
|
|
|
engine = create_engine(database_url, pool_pre_ping=True)
|
|
try:
|
|
with Session(engine) as session:
|
|
rows = session.execute(
|
|
text(
|
|
"select alert_type, state, count(*) from operational_alerts "
|
|
"where state in ('PENDING','FIRING') group by alert_type, state"
|
|
)
|
|
).all()
|
|
finally:
|
|
engine.dispose()
|
|
active = [
|
|
{"alert_type": alert_type, "state": state, "count": int(count)}
|
|
for alert_type, state, count in rows
|
|
]
|
|
return Check(
|
|
name="no unresolved artificial alerts",
|
|
passed=not active,
|
|
detail=f"{sum(item['count'] for item in active)} alerts pending or firing",
|
|
blocking=False,
|
|
evidence={"active": active},
|
|
)
|
|
|
|
|
|
def production_identity_check(database_url: str, expected: Path | None) -> Check:
|
|
engine = create_engine(database_url, pool_pre_ping=True)
|
|
try:
|
|
with Session(engine) as session:
|
|
rows = session.execute(
|
|
text(
|
|
"select d.id::text, c.key, d.status, d.artifact_set_id::text, "
|
|
"d.runtime_profile_id::text, d.capability_contract_id::text "
|
|
"from capability_deployments d "
|
|
"join capability_contracts cc on cc.id = d.capability_contract_id "
|
|
"join capabilities c on c.id = cc.capability_id "
|
|
"where d.production order by c.key, d.id"
|
|
)
|
|
).all()
|
|
finally:
|
|
engine.dispose()
|
|
current = ["|".join(str(value) for value in row) for row in rows]
|
|
if expected is None or not expected.is_file():
|
|
return Check(
|
|
name="production identities unchanged",
|
|
passed=False,
|
|
detail="no baseline supplied to compare against",
|
|
evidence={"current": current},
|
|
)
|
|
baseline = [
|
|
line.strip() for line in expected.read_text("utf-8").splitlines() if line.strip()
|
|
]
|
|
added = sorted(set(current) - set(baseline))
|
|
removed = sorted(set(baseline) - set(current))
|
|
return Check(
|
|
name="production identities unchanged",
|
|
passed=not added and not removed,
|
|
detail=f"{len(current)} production deployments",
|
|
evidence={"added": added, "removed": removed},
|
|
)
|
|
|
|
|
|
def sbom_check() -> Check:
|
|
sbom = ROOT / "docs" / "security" / "sbom" / "modelforge-cyclonedx.json"
|
|
provenance = ROOT / "docs" / "security" / "sbom" / "image-provenance.json"
|
|
if not sbom.is_file() or not provenance.is_file():
|
|
return Check("supply-chain inventory", False, "SBOM or provenance is missing")
|
|
document = json.loads(sbom.read_text("utf-8"))
|
|
images = json.loads(provenance.read_text("utf-8"))
|
|
components = len(document.get("components", []))
|
|
return Check(
|
|
name="supply-chain inventory",
|
|
passed=components > 0 and bool(images.get("images")),
|
|
detail=f"{components} components across {len(images.get('images', []))} images",
|
|
evidence={"source_commit": images.get("source_commit")},
|
|
)
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--database-url", default=DEFAULT_DATABASE_URL)
|
|
parser.add_argument("--chaos", default=None)
|
|
parser.add_argument("--soak", default=None)
|
|
parser.add_argument("--production-baseline", default=None)
|
|
parser.add_argument("--minimum-soak-minutes", type=float, default=30.0)
|
|
parser.add_argument("--report", default=None)
|
|
args = parser.parse_args(argv)
|
|
if not args.database_url:
|
|
parser.error(
|
|
"--database-url or MODELFORGE_RUNTIME_DATABASE_URL is required "
|
|
"(use the non-owner runtime role)"
|
|
)
|
|
|
|
checks = [
|
|
invariant_check(args.database_url),
|
|
chaos_check(_load(args.chaos)),
|
|
soak_check(_load(args.soak), args.minimum_soak_minutes),
|
|
orphan_check(args.database_url),
|
|
alert_check(args.database_url),
|
|
production_identity_check(
|
|
args.database_url,
|
|
Path(args.production_baseline) if args.production_baseline else None,
|
|
),
|
|
sbom_check(),
|
|
]
|
|
|
|
blocking_failures = [item for item in checks if item.blocking and not item.passed]
|
|
advisory_failures = [item for item in checks if not item.blocking and not item.passed]
|
|
readiness = "READY_FOR_PACKAGING" if not blocking_failures else "NOT_READY"
|
|
|
|
report = {
|
|
"generated_at": datetime.now(UTC).isoformat(),
|
|
"hard_blocker_classes": list(HARD_BLOCKERS),
|
|
"checks": [
|
|
{
|
|
"name": item.name,
|
|
"passed": item.passed,
|
|
"blocking": item.blocking,
|
|
"detail": item.detail,
|
|
"evidence": item.evidence,
|
|
}
|
|
for item in checks
|
|
],
|
|
"blocking_failures": [item.name for item in blocking_failures],
|
|
"advisory_failures": [item.name for item in advisory_failures],
|
|
"v1_technical_release_readiness": readiness,
|
|
}
|
|
|
|
for item in checks:
|
|
mark = "PASS" if item.passed else ("FAIL" if item.blocking else "WARN")
|
|
print(f" {mark} {item.name:36} {item.detail}")
|
|
print()
|
|
print(f"V1_TECHNICAL_RELEASE_READINESS = {readiness}")
|
|
|
|
if args.report:
|
|
Path(args.report).write_text(json.dumps(report, indent=2), encoding="utf-8")
|
|
print(f"report written to {args.report}")
|
|
return 0 if readiness == "READY_FOR_PACKAGING" else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|