Jon Chery
2397336cbb
Multi-persona review of the contract surface redesign (031887e+10b87a6). P0-1 (auto-fixed): scripts/run_platform.sh:437 read the uptime_enabled feature flag from the OLD top-level contract.inputs.uptime_enabled path, which P57 removed. With the new contract shape c.get('inputs',{}) returns {} so the flag silently always defaulted to True — a consumer setting uptime_enabled:false under infrastructure.<module>.inputs could NOT disable uptime monitoring. Fixed to scan infrastructure.<module>.inputs.uptime_enabled (any module false wins). P0-2 (auto-fixed): docs/consumer-guide.md:417,472 documented the ${contract.module} interpolation token, but P57 dropped the `module` field. _expand_vars fails loud (D-081) on unknown tokens, so a consumer following the documented bucket_name example (acdl-${env.environment}-${contract.module}-...) hit a hard ValueError at resolve time. Replaced with ${contract.id} (the surviving short acronym field) in both the example and the interpolation reference table. P0-3 (auto-fixed): core/regression_verify.py CAP-006 and tests/test_consumer_guide_per_env_section.py both asserted the dropped ${contract.module} token. Updated CAP-006 to use ${contract.id} and the doc test to assert ${contract.id} present / ${contract.module} absent. P1+ flags (post-hoc): - P1: _namespace_resources does not rewrite ref: targets in stack.outputs[].from for cross-module refs (within-module is handled; multi-module refs across fragments are not wired today, but no contract uses them yet). - P1: _latest_version raises ValueError (not a clear message) on a malformed semver string in the registry; the schema pins version to ^\d+\.\d+\.\d+$ so this is unreachable from a contract, but registry authors have no guardrail. - P2: docs/consumer-guide.md:407 example path uses .yaml extension while the repo-wide rename standardized on .yml (consumer-repo paths, not platform, so non-blocking). ---ci--- project: acdl phase: 57 milestone: v1.10.2 status: verify lessons: - P0 fix applied: uptime_enabled read path migrated to infrastructure.<module>.inputs (was stale top-level contract.inputs) - P0 fix applied: docs + tests migrated off dropped ${contract.module} interpolation token to ${contract.id} ---/ci---
536 lines
21 KiB
Python
Executable File
536 lines
21 KiB
Python
Executable File
"""Regression-class VERIFY (D-091).
|
|
|
|
The standard VERIFY stage is diff-scoped: it checks the phase diff only
|
|
and never re-runs underlying platform capability. That structural defect
|
|
(let 8 NFR-patch phases pass while the platform decayed) is recorded as
|
|
D-091. This module provides the regression-class VERIFY that re-runs
|
|
capability checks against the current codebase and tags each capability
|
|
Verified / Decayed / Broken.
|
|
|
|
A capability check is a function that takes no args and returns
|
|
(status, detail) where status is one of:
|
|
- "Verified" : the capability runs as advertised
|
|
- "Decayed" : the capability runs partially / with errors but the
|
|
core path is intact (e.g. needs revival work)
|
|
- "Broken" : the capability does not run at all
|
|
|
|
The regression run fails closed: any non-Verified capability blocks
|
|
milestone completion. The result is written to
|
|
`.ciagent/REGRESSION_REPORT.md` and a machine-readable JSON file.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
from dataclasses import dataclass, field, asdict
|
|
from pathlib import Path
|
|
from typing import Callable, Dict, List, Optional, Tuple
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
CIAgent = ROOT / ".ciagent"
|
|
|
|
Status = str # "Verified" | "Decayed" | "Broken"
|
|
|
|
|
|
@dataclass
|
|
class CapabilityResult:
|
|
capability_id: str
|
|
name: str
|
|
status: Status
|
|
detail: str
|
|
tier: str # "local" | "live-aws"
|
|
duration_ms: int
|
|
|
|
|
|
@dataclass
|
|
class RegressionReport:
|
|
run_id: str
|
|
run_at_utc: str
|
|
milestone: str
|
|
phase: int
|
|
results: List[CapabilityResult] = field(default_factory=list)
|
|
|
|
@property
|
|
def summary(self) -> Dict[str, int]:
|
|
counts = {"Verified": 0, "Decayed": 0, "Broken": 0}
|
|
for r in self.results:
|
|
counts[r.status] = counts.get(r.status, 0) + 1
|
|
return counts
|
|
|
|
@property
|
|
def passed(self) -> bool:
|
|
return all(r.status == "Verified" for r in self.results)
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"run_id": self.run_id,
|
|
"run_at_utc": self.run_at_utc,
|
|
"milestone": self.milestone,
|
|
"phase": self.phase,
|
|
"summary": self.summary,
|
|
"passed": self.passed,
|
|
"results": [asdict(r) for r in self.results],
|
|
}
|
|
|
|
|
|
def _run_subprocess(cmd: List[str], cwd: Optional[str] = None,
|
|
timeout: int = 120,
|
|
env: Optional[Dict[str, str]] = None) -> Tuple[int, str, str]:
|
|
"""Run a subprocess, return (returncode, stdout, stderr)."""
|
|
try:
|
|
p = subprocess.run(
|
|
cmd, cwd=cwd or str(ROOT), capture_output=True,
|
|
text=True, timeout=timeout, env=env,
|
|
)
|
|
return p.returncode, p.stdout, p.stderr
|
|
except subprocess.TimeoutExpired as e:
|
|
return 124, e.stdout or "", e.stderr or ""
|
|
except FileNotFoundError as e:
|
|
return 127, "", str(e)
|
|
|
|
|
|
def _check_subprocess(cmd: List[str], cwd: Optional[str] = None,
|
|
timeout: int = 120,
|
|
env: Optional[Dict[str, str]] = None) -> Tuple[Status, str]:
|
|
"""Run a subprocess; map returncode to a status."""
|
|
rc, out, err = _run_subprocess(cmd, cwd=cwd, timeout=timeout, env=env)
|
|
if rc == 0:
|
|
return "Verified", f"exit 0; {out.strip()[-200:]}"
|
|
if rc == 124:
|
|
return "Decayed", f"timeout after {timeout}s; {err.strip()[-200:]}"
|
|
return "Broken", f"exit {rc}; {err.strip()[-200:]}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Capability checks (seeded for Phase 52; Phase 54 expands the registry).
|
|
# Each check is local-only at this stage (Phase 53 adds the local emulators;
|
|
# Phase 54 adds the live-AWS tier for the headline E2E).
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _check_contract_schema_validation() -> Tuple[Status, str]:
|
|
"""CAP-001: contract.schema.json validates sample contracts."""
|
|
return _check_subprocess([
|
|
"python3", "-c",
|
|
"import json, yaml, jsonschema; "
|
|
"s=json.load(open('schemas/contract.schema.json')); "
|
|
"[jsonschema.validate(yaml.safe_load(open(f)), s) "
|
|
" for f in ['contracts/static-assets.yml','contracts/microservice.yml']]; "
|
|
"print('2 sample contracts validate')",
|
|
])
|
|
|
|
|
|
def _check_environment_schema_validation() -> Tuple[Status, str]:
|
|
"""CAP-002: environment.schema.json validates the env files."""
|
|
return _check_subprocess([
|
|
"python3", "-c",
|
|
"import json, jsonschema; "
|
|
"s=json.load(open('schemas/environment.schema.json')); "
|
|
"[jsonschema.validate(json.load(open(f)), s) "
|
|
" for f in ['core/environments/dev.json']]; "
|
|
"print('env schema validates')",
|
|
])
|
|
|
|
|
|
def _check_resolver_static_assets() -> Tuple[Status, str]:
|
|
"""CAP-003: contract_resolver resolves static-assets to a Target Stack."""
|
|
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,
|
|
])
|
|
finally:
|
|
try:
|
|
os.unlink(out)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def _check_resolver_microservice() -> Tuple[Status, str]:
|
|
"""CAP-004: contract_resolver resolves the microservice contract."""
|
|
with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as t:
|
|
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]:
|
|
"""CAP-005: terraform adapter compiles a resolved stack to .tf files."""
|
|
work = tempfile.mkdtemp(prefix="acdl_regr_")
|
|
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:]}"
|
|
status, detail = _check_subprocess([
|
|
"python3", "adapters/terraform/adapter.py", stack_path, tf_dir,
|
|
])
|
|
if status == "Verified":
|
|
main_tf = os.path.join(tf_dir, "main.tf")
|
|
if not os.path.isfile(main_tf) or os.path.getsize(main_tf) == 0:
|
|
return "Broken", "adapter exited 0 but main.tf missing/empty"
|
|
return status, detail
|
|
|
|
|
|
def _check_interpolation() -> Tuple[Status, str]:
|
|
"""CAP-006: contract interpolation expands ${env.*} / ${contract.*}.
|
|
|
|
P57: the contract's `module` field was dropped in favor of `id`
|
|
(short acronym) + `infrastructure` map; the interpolation check uses
|
|
`contract.id` (the surviving field)."""
|
|
return _check_subprocess([
|
|
"python3", "-c",
|
|
"import sys; sys.path.insert(0,'.'); "
|
|
"from core.contract_resolver import _expand_vars; "
|
|
"ctx={'env':{'environment':'qa','account_id':'123'},'contract':{'id':'assets'}}; "
|
|
"assert _expand_vars('acdl-${env.environment}-${contract.id}', ctx)=='acdl-qa-assets'; "
|
|
"print('interpolation ok')",
|
|
])
|
|
|
|
|
|
def _check_confidence_signal() -> Tuple[Status, str]:
|
|
"""CAP-007: confidence_signal.compute returns a band for a pass/fail input."""
|
|
return _check_subprocess([
|
|
"python3", "-c",
|
|
"import sys, json; sys.path.insert(0,'.'); "
|
|
"import core.confidence_signal as c; "
|
|
"inputs={'policy':[],'validation':{'schema':True,'stack_resolved':True,'tf_validated':True,'tf_planned':True},'freshness':{'age_days':0,'max_age_days':7},'source':{'submitter':'consumer','commit_sha':'x','signed':False},'history':{'prior_rollbacks':0,'prior_policy_fails':0},'nfrs':{'conformance':None}}; "
|
|
"sig=c.compute('cid','dev',inputs); "
|
|
"assert sig.band in ('pass','warn','fail'); "
|
|
"print(f'confidence band={sig.band}')",
|
|
])
|
|
|
|
|
|
def _check_outbox_writer() -> Tuple[Status, str]:
|
|
"""CAP-008: outbox_writer writes a hash-chained event to a temp file."""
|
|
work = tempfile.mkdtemp(prefix="acdl_outbox_")
|
|
event_path = os.path.join(work, "event.json")
|
|
event = {
|
|
"contractId": "regression-test", "eventType": "CONFIDENCE_COMPUTED",
|
|
"ts": "2026-07-27T00:00:00Z", "environment": "dev",
|
|
"stack": "regression", "score": 0.9, "band": "pass",
|
|
"prev_event_hash": "GENESIS",
|
|
}
|
|
with open(event_path, "w") as f:
|
|
json.dump(event, f)
|
|
# The outbox writer writes to DynamoDB in prod; for the regression we
|
|
# verify the hash-chain logic (the testable core) without AWS. The
|
|
# actual DynamoDB write is a live-AWS concern, deferred to Phase 54.
|
|
return _check_subprocess([
|
|
"python3", "-c",
|
|
f"import sys, json; sys.path.insert(0,'.'); "
|
|
f"import core.outbox_writer as w; "
|
|
f"ev=json.load(open('{event_path}')); "
|
|
f"h=w._canonical_hash(ev); "
|
|
f"assert len(h)==64; "
|
|
f"assert w._canonical_hash(ev)==h; "
|
|
f"print('outbox hash chain ok')",
|
|
])
|
|
|
|
|
|
def _check_pytest_offline() -> Tuple[Status, str]:
|
|
"""CAP-009: the offline pytest suite passes (the regression baseline).
|
|
|
|
Excludes slow tests (which invoke the full pipeline) and the
|
|
regression test itself (to avoid recursion: this check runs inside
|
|
the regression run)."""
|
|
return _check_subprocess(
|
|
["python3", "-m", "pytest", "tests/", "-q", "--tb=line",
|
|
"-m", "not slow",
|
|
"--ignore=tests/test_contract_ingestor.py",
|
|
"--ignore=tests/test_verify_regression_mode.py"],
|
|
timeout=180,
|
|
)
|
|
|
|
|
|
def _check_run_ci_check_only() -> Tuple[Status, str]:
|
|
"""CAP-010: run_ci.sh reproduces the CI pipeline locally (offline).
|
|
|
|
Excluded from the regression's own pytest invocation to avoid
|
|
recursion; invoked directly here."""
|
|
return _check_subprocess(
|
|
["bash", "scripts/run_ci.sh", "--quiet"], timeout=240,
|
|
)
|
|
|
|
|
|
def _check_local_e2e_microservice() -> Tuple[Status, str]:
|
|
"""CAP-011: headline E2E runs against the local emulating tier (D-092).
|
|
|
|
The local tier emulates ECS, the DynamoDB outbox, S3 state, and the
|
|
contract-ingestor Lambda in-process. No AWS credentials required.
|
|
This is the local-tier half of the headline E2E; the live-AWS half
|
|
lands in Phase 54 (D-093)."""
|
|
return _check_subprocess(
|
|
["python3", "core/local_emulators.py", "contracts/microservice.yml"],
|
|
timeout=60,
|
|
)
|
|
|
|
|
|
def _check_local_e2e_static_assets() -> Tuple[Status, str]:
|
|
"""CAP-012: local E2E on the static-assets stack (no ECS service)."""
|
|
return _check_subprocess(
|
|
["python3", "core/local_emulators.py", "contracts/static-assets.yml"],
|
|
timeout=60,
|
|
)
|
|
|
|
|
|
def _load_aws_env() -> Dict[str, str]:
|
|
"""Load AWS credentials from .env.secrets and return an env dict
|
|
with AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY / AWS_DEFAULT_REGION set."""
|
|
env = os.environ.copy()
|
|
secrets_path = os.path.join(str(ROOT), ".env.secrets")
|
|
if os.path.isfile(secrets_path):
|
|
with open(secrets_path) as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if not line or line.startswith("#"):
|
|
continue
|
|
if "=" in line:
|
|
k, v = line.split("=", 1)
|
|
if k == "ACDL_AWS_ACCESS_KEY_ID":
|
|
env["AWS_ACCESS_KEY_ID"] = v
|
|
elif k == "ACDL_AWS_SECRET_ACCESS_KEY":
|
|
env["AWS_SECRET_ACCESS_KEY"] = v
|
|
elif k == "AWS_DEFAULT_REGION":
|
|
env["AWS_DEFAULT_REGION"] = v
|
|
return env
|
|
|
|
|
|
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).
|
|
|
|
Requires AWS credentials (ACDL_AWS_ACCESS_KEY_ID etc. in .env.secrets).
|
|
Runs in a temp dir; does NOT apply (plan only)."""
|
|
import tempfile, os
|
|
work = tempfile.mkdtemp(prefix="acdl_regr_live_")
|
|
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,
|
|
])
|
|
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, 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
|
|
work = tempfile.mkdtemp(prefix="acdl_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]:
|
|
"""CAP-015: DynamoDB outbox table exists + is describable (live AWS)."""
|
|
import boto3
|
|
env = _load_aws_env()
|
|
try:
|
|
dyn = boto3.client("dynamodb", region_name=env.get("AWS_DEFAULT_REGION", "us-east-1"),
|
|
aws_access_key_id=env.get("AWS_ACCESS_KEY_ID"),
|
|
aws_secret_access_key=env.get("AWS_SECRET_ACCESS_KEY"))
|
|
r = dyn.describe_table(TableName="acdl-outbox")
|
|
count = r["Table"].get("ItemCount", "unknown")
|
|
return "Verified", f"acdl-outbox exists, item_count={count}"
|
|
except Exception as e:
|
|
return "Decayed", f"describe_table failed: {type(e).__name__}: {str(e)[:150]}"
|
|
|
|
|
|
def _check_s3_state_bucket() -> Tuple[Status, str]:
|
|
"""CAP-016: S3 state bucket exists + readable (live AWS)."""
|
|
import boto3
|
|
env = _load_aws_env()
|
|
try:
|
|
s3 = boto3.client("s3", region_name=env.get("AWS_DEFAULT_REGION", "us-east-1"),
|
|
aws_access_key_id=env.get("AWS_ACCESS_KEY_ID"),
|
|
aws_secret_access_key=env.get("AWS_SECRET_ACCESS_KEY"))
|
|
s3.head_bucket(Bucket="acdl-tfstate-581513795199-us-east-1")
|
|
r = s3.list_objects_v2(Bucket="acdl-tfstate-581513795199-us-east-1", MaxKeys=5)
|
|
keys = [o["Key"] for o in r.get("Contents", [])]
|
|
return "Verified", f"state bucket exists, keys={keys}"
|
|
except Exception as e:
|
|
return "Decayed", f"head_bucket failed: {type(e).__name__}: {str(e)[:150]}"
|
|
|
|
|
|
# Registry: ordered, each entry is (capability_id, name, tier, check_fn).
|
|
# Phase 52 seeds this with 10 local-tier checks; Phase 54 expands it to
|
|
# cover every v1.1->v1.8 advertised capability and adds the live-AWS tier
|
|
# for the headline E2E.
|
|
CAPABILITY_REGISTRY: List[Tuple[str, str, str, Callable[[], Tuple[Status, str]]]] = [
|
|
("CAP-001", "contract.schema.json validates sample contracts", "local",
|
|
_check_contract_schema_validation),
|
|
("CAP-002", "environment.schema.json validates env files", "local",
|
|
_check_environment_schema_validation),
|
|
("CAP-003", "contract_resolver resolves static-assets", "local",
|
|
_check_resolver_static_assets),
|
|
("CAP-004", "contract_resolver resolves microservice", "local",
|
|
_check_resolver_microservice),
|
|
("CAP-005", "terraform adapter emits .tf files", "local",
|
|
_check_adapter_emits_terraform),
|
|
("CAP-006", "contract interpolation expands env/contract tokens", "local",
|
|
_check_interpolation),
|
|
("CAP-007", "confidence_signal.compute returns a band", "local",
|
|
_check_confidence_signal),
|
|
("CAP-008", "outbox_writer builds a hash-chained item", "local",
|
|
_check_outbox_writer),
|
|
("CAP-009", "offline pytest suite passes", "local",
|
|
_check_pytest_offline),
|
|
("CAP-010", "run_ci.sh reproduces CI pipeline locally", "local",
|
|
_check_run_ci_check_only),
|
|
("CAP-011", "headline E2E runs against the local emulating tier (microservice)", "local",
|
|
_check_local_e2e_microservice),
|
|
("CAP-012", "local E2E on the static-assets stack (no ECS)", "local",
|
|
_check_local_e2e_static_assets),
|
|
("CAP-013", "terraform init+validate+plan live AWS (microservice)", "live-aws",
|
|
_check_live_terraform_plan_microservice),
|
|
("CAP-014", "terraform init+validate+plan live AWS (static-assets)", "live-aws",
|
|
_check_live_terraform_plan_static_assets),
|
|
("CAP-015", "DynamoDB outbox table exists (live AWS)", "live-aws",
|
|
_check_dynamodb_outbox_table),
|
|
("CAP-016", "S3 state bucket exists + readable (live AWS)", "live-aws",
|
|
_check_s3_state_bucket),
|
|
]
|
|
|
|
|
|
def run_regression(milestone: str = "v1.10", phase: int = 52,
|
|
registry: Optional[List] = None) -> RegressionReport:
|
|
"""Run every capability check in the registry; return a RegressionReport."""
|
|
reg = registry if registry is not None else CAPABILITY_REGISTRY
|
|
run_id = f"regr-{int(time.time())}"
|
|
run_at = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
|
|
report = RegressionReport(run_id=run_id, run_at_utc=run_at,
|
|
milestone=milestone, phase=phase)
|
|
for cap_id, name, tier, fn in reg:
|
|
t0 = time.monotonic()
|
|
try:
|
|
status, detail = fn()
|
|
except Exception as e: # noqa: BLE001
|
|
status, detail = "Broken", f"check raised: {type(e).__name__}: {e}"[:300]
|
|
dur = int((time.monotonic() - t0) * 1000)
|
|
report.results.append(CapabilityResult(
|
|
capability_id=cap_id, name=name, status=status,
|
|
detail=detail, tier=tier, duration_ms=dur,
|
|
))
|
|
return report
|
|
|
|
|
|
def write_report(report: RegressionReport,
|
|
md_path: Optional[Path] = None,
|
|
json_path: Optional[Path] = None) -> Tuple[Path, Path]:
|
|
"""Write the report to .ciagent/REGRESSION_REPORT.md + .json."""
|
|
md_path = md_path or (CIAgent / "REGRESSION_REPORT.md")
|
|
json_path = json_path or (CIAgent / "REGRESSION_REPORT.json")
|
|
json_path.write_text(json.dumps(report.to_dict(), indent=2))
|
|
lines = [
|
|
f"# Regression Report — {report.milestone} Phase {report.phase}",
|
|
"",
|
|
f"- **Run ID:** `{report.run_id}`",
|
|
f"- **Run at (UTC):** {report.run_at_utc}",
|
|
f"- **Summary:** {report.summary}",
|
|
f"- **Passed (milestone gate):** {report.passed}",
|
|
"",
|
|
"| Capability | Name | Tier | Status | Duration (ms) | Detail |",
|
|
"|-----------|------|------|--------|--------------|--------|",
|
|
]
|
|
for r in report.results:
|
|
lines.append(
|
|
f"| {r.capability_id} | {r.name} | {r.tier} | "
|
|
f"**{r.status}** | {r.duration_ms} | {r.detail[:160]} |"
|
|
)
|
|
md_path.write_text("\n".join(lines) + "\n")
|
|
return md_path, json_path
|
|
|
|
|
|
def main() -> int:
|
|
milestone = os.environ.get("ACDL_REGRESSION_MILESTONE", "v1.10")
|
|
phase = int(os.environ.get("ACDL_REGRESSION_PHASE", "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 capabilities "
|
|
"(milestone gate blocks)", file=sys.stderr)
|
|
return 1
|
|
print("regression: all capabilities Verified (milestone gate passes)")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main()) |