403 lines
19 KiB
Python
403 lines
19 KiB
Python
# #region Tooling.ReconcileContractsModule [C:3] [TYPE Module]
|
|
# @LAYER Infra
|
|
# @BRIEF Machine-contract reconciliation gate for the 036-047 scenario specs.
|
|
# @RELATION DEPENDS_ON -> [Tooling.Merge.Spec]
|
|
# @RATIONALE Cross-spec identity/contract drift is the dominant defect class after prose rewrites:
|
|
# a spec can read well while its executable contracts (JSON Schema, OpenAPI, fixtures, state machines)
|
|
# remain on the old model. This gate parses and cross-checks the machine-readable source of truth.
|
|
# @REJECTED Trusting prose-only reviews — the 038 package previously validated "PASS" while its JSON Schema
|
|
# still required scenario_id/revision_hash and 3 OpenAPI files failed YAML parsing.
|
|
|
|
from __future__ import annotations
|
|
|
|
import fnmatch
|
|
import json
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
SPECS_DIR = Path("specs")
|
|
|
|
# Content/identity drift tokens: forbidden in machine-readable contracts.
|
|
# NOTE: `scenario_id` is the legitimate entity id (UUID) in the registry/run APIs and
|
|
# OpenAPI path params; it is forbidden ONLY in the 038 compiled-output schema/fixtures
|
|
# (the compiler must not emit it). Handled separately below.
|
|
FORBIDDEN_IDENTITY = [
|
|
"revision_hash", "parent_revision_hash", "scenario_revision_hash",
|
|
"base_revision_hash", "new_revision_hash", "step_ids", "affected_step_ids",
|
|
"dismiss", "flaky_test",
|
|
]
|
|
# Compiled-output schema/fixtures where the compiler must not emit entity identity.
|
|
COMPILED_OUTPUT_SCHEMA = "dashboard-test-scenario.schema.json"
|
|
COMPILED_OUTPUT_DIRS = {"fixtures"}
|
|
|
|
# Rejected architectural alternatives that have previously resurfaced in prose
|
|
# after the data model changed. These are deliberately narrow, high-signal
|
|
# phrases; a hit is a review task, not a substitute for architectural review.
|
|
REJECTED_DECISION_PATTERNS = {
|
|
"major revision analytics key": r"same major scenario revision",
|
|
"triage classification fingerprint": r"step\s*\+\s*error_code\s*\+\s*classification",
|
|
"runner plan pack source": r"RunnerPlan.*loaded from the saved scenario pack",
|
|
"materialized runner plan source": r"runner\.plan\.json.*(real materialized contract|runner consumes)",
|
|
"ordinal step identity": r"step_key\s*=.*ordinal",
|
|
"derived step uuid identity": r"logical_step_id.*derived from scenario_key\s*\+\s*step_key",
|
|
"ambiguous dependency step id": r"target_step_id",
|
|
"resolved recurrence suppressed": r"(resolved|known-issue|accepted).*not re-alerted",
|
|
"save silently activates revision": r"(?:save|saved|saving).*current_revision\s+(?:advanced|advances)",
|
|
"current pointer advances on edit": r"current_revision\s+pointer\s+advances\s+on\s+edit",
|
|
"runtime agent blanket exclusion": r"agent is not in the hot path",
|
|
"required authoring runtime parameter": r"unresolved required parameter.*preview_only",
|
|
}
|
|
|
|
# Final closure invariants are presence checks for the canonical, normative
|
|
# clauses. They deliberately complement (not replace) schema/OpenAPI parsing.
|
|
REQUIRED_CONTRACT_FRAGMENTS = {
|
|
"038-dashboard-scenario-model/data-model.md": (
|
|
"AuthoringValidation", "RunPreflight", "ActionRegistry(version)", "VerificationProgram", "SqlEvidenceSpec", "TransformSpec", "AgentEvaluationSpec", "mutating browser steps in PROD are prohibited",
|
|
),
|
|
"038-dashboard-scenario-model/contracts/verification-program.md": (
|
|
"SqlEvidenceSpec", "TransformSpec", "AgentEvaluationSpec", "DecisionPolicy", "Superset SQL Lab adapter", "SQL compilation gate",
|
|
),
|
|
"036-agent-test-stabilization/data-model.md": (
|
|
"owner_type", "scenario_run", "load_run", "DelegatedAuthorityPolicy", "InvestigationSignal",
|
|
),
|
|
"036-agent-test-stabilization/contracts/agent-runs.openapi.yaml": (
|
|
"/action-approval-gates/{gateId}/decision", "ScenarioExecutionApprovalRequest", "/investigation-cases/{caseId}/actions", "AgentActionRequest", "DelegatedAuthorityPolicy",
|
|
),
|
|
"042-dashboard-scenario-registry/data-model.md": (
|
|
"042 does not derive health", "MVP is archive-only", "intersection of scenario permission", "Revision save and activation are separate operations", "may_activate_current_revision",
|
|
),
|
|
"043-dashboard-scenario-editor/contracts/openapi.yaml": (
|
|
"/metadata", "If-Match", "agent_action_id", "/migration-proposals/{proposal_id}/resolve",
|
|
),
|
|
"044-dashboard-scenario-execution/data-model.md": (
|
|
"pending_approval", "Idempotency uses canonical execution-request hash", "BrowserExecutor", "SqlEvidenceExecutor", "AgentEvaluation", "DecisionPolicy", "checkpoint_type", "manual_run_only=true", "InvestigationSignal", "AnalyticsContextKey", "Mutating browser steps in PROD are prohibited",
|
|
),
|
|
"044-dashboard-scenario-execution/contracts/openapi.yaml": (
|
|
"ScenarioExecutionResult", "RunComparison", "ScenarioRunEvent", "requested_target_reference", "StepOutcome", "AgentEvaluationSummary", "agent_evaluation_completed",
|
|
),
|
|
"046-dashboard-scenario-automation/data-model.md": (
|
|
"revision_id (required when pinned)", "canonical_execution_request_hash", "Concurrency bucket", "AUTOMATION_INELIGIBLE_HUMAN_STEP", "candidate", "atomic activation",
|
|
),
|
|
"047-dashboard-scenario-analytics/data-model.md": (
|
|
"InvestigationQueueItem", "InvestigationCase", "compatibility_family", "FailureEpisode", "product_health", "agent_evaluation_health", "does not start AgentRun", "Closure policy", "AnalyticsContextKey", "InvestigationSignal",
|
|
),
|
|
"036-agent-test-stabilization/contracts/investigation-cases.md": (
|
|
"InvestigationQueueItem", "InvestigationCase", "AgentAction", "MUST NOT use a modal", "DelegatedAuthorityPolicy", "InvestigationSignal", "Process Boundary Matrix",
|
|
),
|
|
}
|
|
|
|
# Mutable vs derived: a `revision_hash`-style field present in an OpenAPI/JSON Schema path is a drift flag.
|
|
SKIP_DIRS = {"__pycache__"}
|
|
MERGEABLE_REVIEW = {".md", ".yaml", ".yml", ".json", ".html"}
|
|
|
|
|
|
def _resolve_dir(token: str) -> Path | None:
|
|
"""Resolve a CLI token to a spec dir: number, range, full name, or 'all'."""
|
|
if not token or token.lower() == "all":
|
|
return None
|
|
# exact directory name
|
|
cand = SPECS_DIR / token
|
|
if cand.is_dir():
|
|
return cand
|
|
# number range "036-041" (inclusive)
|
|
m = re.fullmatch(r"(\d{3})-(\d{3})", token)
|
|
if m:
|
|
lo, hi = int(m.group(1)), int(m.group(2))
|
|
if lo > hi:
|
|
lo, hi = hi, lo
|
|
return [d for d in sorted(SPECS_DIR.iterdir()) if d.is_dir()
|
|
and re.match(r"^(\d{3})-", d.name)
|
|
and lo <= int(d.name[:3]) <= hi]
|
|
# single number "038"
|
|
if re.fullmatch(r"\d{3}", token):
|
|
for d in SPECS_DIR.iterdir():
|
|
if d.is_dir() and d.name.startswith(f"{token}-"):
|
|
return d
|
|
print(f" [warn] no spec matched '{token}'")
|
|
return None
|
|
|
|
|
|
def _walk(spec_id: str) -> list[Path]:
|
|
d = SPECS_DIR / spec_id
|
|
if not d.is_dir():
|
|
return []
|
|
return [p for p in d.rglob("*") if p.is_file() and p.suffix in MERGEABLE_REVIEW
|
|
and not any(part in SKIP_DIRS for part in p.parts)]
|
|
|
|
|
|
def check_yaml(paths: list[Path]) -> list[str]:
|
|
out = []
|
|
for p in paths:
|
|
if p.suffix in {".yaml", ".yml"}:
|
|
try:
|
|
yaml.safe_load(p.read_text(encoding="utf-8"))
|
|
except Exception as e: # noqa: BLE001
|
|
out.append(f"[YAML] {p.relative_to(SPECS_DIR)}: {e}")
|
|
return out
|
|
|
|
|
|
def check_json(paths: list[Path]) -> list[str]:
|
|
out = []
|
|
for p in paths:
|
|
if p.suffix == ".json":
|
|
try:
|
|
json.loads(p.read_text(encoding="utf-8"))
|
|
except Exception as e: # noqa: BLE001
|
|
out.append(f"[JSON] {p.relative_to(SPECS_DIR)}: {e}")
|
|
return out
|
|
|
|
|
|
def check_forbidden_identity(paths: list[Path]) -> list[str]:
|
|
"""Forbidden old-identity token in a machine-readable file is a drift flag.
|
|
|
|
`scenario_id` is only flagged in the 038 compiled-output schema/fixtures (compiler
|
|
must not emit entity identity); it is legitimate in registry/run OpenAPI path params.
|
|
"""
|
|
out = []
|
|
for p in paths:
|
|
if p.suffix not in {".yaml", ".yml", ".json", ".html"}:
|
|
continue
|
|
tokens = list(FORBIDDEN_IDENTITY)
|
|
is_compiled_output = (
|
|
p.name == COMPILED_OUTPUT_SCHEMA
|
|
or any(part in COMPILED_OUTPUT_DIRS for part in p.parts)
|
|
)
|
|
if is_compiled_output:
|
|
tokens.append("scenario_id")
|
|
text = p.read_text(encoding="utf-8", errors="ignore")
|
|
for token in tokens:
|
|
hits = []
|
|
for i, line in enumerate(text.splitlines(), 1):
|
|
stripped = line.strip()
|
|
if stripped.startswith(("#", "//", "*")):
|
|
continue
|
|
# skip explanatory descriptions/comments mentioning the token (e.g. "assigned by 042")
|
|
if '"description"' in line or '"title"' in line:
|
|
continue
|
|
if re.search(rf"\b{token}\b", line):
|
|
hits.append(i)
|
|
if hits:
|
|
out.append(f"[IDENT] {p.relative_to(SPECS_DIR)}: '{token}' at lines {hits[:6]}")
|
|
return out
|
|
|
|
|
|
def check_fixtures_vs_schema(spec_id: str = "038-dashboard-scenario-model") -> list[str]:
|
|
"""Validate fixtures/api/*.json against dashboard-test-scenario.schema.json (if jsonschema present)."""
|
|
out = []
|
|
schema_path = SPECS_DIR / spec_id / "contracts" / "dashboard-test-scenario.schema.json"
|
|
if not schema_path.exists():
|
|
return out
|
|
try:
|
|
import jsonschema # type: ignore
|
|
except ImportError:
|
|
out.append(f"[FIXTURE] jsonschema not installed; skip schema validation for {spec_id}")
|
|
return out
|
|
schema = json.loads(schema_path.read_text(encoding="utf-8"))
|
|
for f in sorted((SPECS_DIR / spec_id / "fixtures" / "api").glob("*.json")):
|
|
data = json.loads(f.read_text(encoding="utf-8"))
|
|
try:
|
|
jsonschema.validate(data, schema)
|
|
except jsonschema.ValidationError as e: # type: ignore
|
|
out.append(f"[FIXTURE] {f.relative_to(SPECS_DIR)}: {list(e.absolute_path)[:5]} {e.message[:90]}")
|
|
return out
|
|
|
|
|
|
def check_verification_program_boundaries(paths: list[Path]) -> list[str]:
|
|
"""Assert the agent-authored / immutable-runtime Verification Program boundary.
|
|
|
|
These checks intentionally inspect canonical machine contracts rather than prose:
|
|
038 must model the program and keep runtime values out; 044 must expose typed
|
|
outcome/evaluation events without a request surface that can mutate SQL/code.
|
|
"""
|
|
out = []
|
|
by_name = {p.relative_to(SPECS_DIR).as_posix(): p for p in paths}
|
|
schema_path = by_name.get("038-dashboard-scenario-model/contracts/dashboard-test-scenario.schema.json")
|
|
if schema_path:
|
|
schema = json.loads(schema_path.read_text(encoding="utf-8"))
|
|
required = set(schema.get("required") or [])
|
|
props = schema.get("properties") or {}
|
|
defs = schema.get("$defs") or {}
|
|
if "verification_program" not in required or "verification_program" not in props:
|
|
out.append("[PROGRAM] 038 schema: verification_program must be required canonical content")
|
|
for name in ("verificationProgram", "sqlEvidenceSpec", "transformSpec", "assertionSpec", "agentEvaluationSpec"):
|
|
if name not in defs:
|
|
out.append(f"[PROGRAM] 038 schema: missing $defs/{name}")
|
|
parameter_props = (defs.get("parameter") or {}).get("properties") or {}
|
|
forbidden = sorted({"value", "status"} & set(parameter_props))
|
|
if forbidden:
|
|
out.append(f"[PROGRAM] 038 schema: ParameterDefinition contains runtime field(s) {', '.join(forbidden)}")
|
|
registry_path = by_name.get("038-dashboard-scenario-model/contracts/action-registry.yaml")
|
|
valid_fixture = by_name.get("038-dashboard-scenario-model/fixtures/api/scenario_valid.json")
|
|
if not registry_path:
|
|
out.append("[PROGRAM] 038 contract: missing versioned action-registry.yaml")
|
|
elif valid_fixture:
|
|
registry = yaml.safe_load(registry_path.read_text(encoding="utf-8")) or {}
|
|
registered = {(item.get("tool"), item.get("action")) for item in registry.get("actions", []) if isinstance(item, dict)}
|
|
fixture = json.loads(valid_fixture.read_text(encoding="utf-8"))
|
|
for step in fixture.get("steps", []):
|
|
pair = (step.get("tool"), step.get("action"))
|
|
if pair not in registered:
|
|
out.append(f"[PROGRAM] 038 registry: valid fixture action not registered: {pair[0]}/{pair[1]}")
|
|
run_api = by_name.get("044-dashboard-scenario-execution/contracts/openapi.yaml")
|
|
if run_api:
|
|
document = yaml.safe_load(run_api.read_text(encoding="utf-8")) or {}
|
|
schemas = ((document.get("components") or {}).get("schemas") or {})
|
|
for name in ("ScenarioExecutionResult", "RunComparison", "ScenarioRunEvent", "StepOutcome", "AgentEvaluationSummary"):
|
|
if name not in schemas:
|
|
out.append(f"[PROGRAM] 044 OpenAPI: missing typed schema {name}")
|
|
text = run_api.read_text(encoding="utf-8")
|
|
forbidden_runtime_sql = ("sql_template" in text and "/api/scenario-runs" in text)
|
|
if forbidden_runtime_sql:
|
|
out.append("[PROGRAM] 044 OpenAPI: runtime API must not accept sql_template mutation")
|
|
return out
|
|
|
|
|
|
def check_revision_vs_content_hash(paths: list[Path]) -> list[str]:
|
|
"""flag 'revision_id' where the field is used to mean content (hash) vs identity (uuid)."""
|
|
out = []
|
|
for p in paths:
|
|
if p.suffix != ".yaml":
|
|
continue
|
|
text = p.read_text(encoding="utf-8", errors="ignore")
|
|
for line in text.splitlines():
|
|
if "revision_id" in line and "pattern" in line:
|
|
if '"uuid"' not in line:
|
|
out.append(f"[REV] {p.relative_to(SPECS_DIR)}: revision_id must be uuid format: {line.strip()}")
|
|
return out
|
|
|
|
|
|
def check_openapi_path_parameters(paths: list[Path]) -> list[str]:
|
|
"""Validate OpenAPI path-template variables against operation parameters.
|
|
|
|
YAML parsing proves only syntax. OpenAPI requires every `{name}` in a path
|
|
template to be represented by an `in: path`, `required: true` parameter on
|
|
either the path item or every operation that exposes the path.
|
|
"""
|
|
out = []
|
|
methods = {"get", "post", "put", "patch", "delete", "head", "options", "trace"}
|
|
for p in paths:
|
|
if p.name != "openapi.yaml":
|
|
continue
|
|
try:
|
|
document = yaml.safe_load(p.read_text(encoding="utf-8")) or {}
|
|
except Exception: # check_yaml reports the parse failure with detail.
|
|
continue
|
|
for path, path_item in (document.get("paths") or {}).items():
|
|
if not isinstance(path_item, dict):
|
|
continue
|
|
placeholders = set(re.findall(r"\{([^}/]+)\}", path))
|
|
shared = path_item.get("parameters") or []
|
|
for method, operation in path_item.items():
|
|
if method.lower() not in methods or not isinstance(operation, dict):
|
|
continue
|
|
declared: dict[str, dict] = {}
|
|
for parameter in [*shared, *(operation.get("parameters") or [])]:
|
|
if not isinstance(parameter, dict) or "$ref" in parameter:
|
|
continue
|
|
if parameter.get("in") == "path" and isinstance(parameter.get("name"), str):
|
|
declared[parameter["name"]] = parameter
|
|
missing = sorted(
|
|
name for name in placeholders
|
|
if name not in declared or declared[name].get("required") is not True
|
|
)
|
|
extra = sorted(name for name in declared if name not in placeholders)
|
|
op_name = operation.get("operationId", method.upper())
|
|
if missing:
|
|
out.append(
|
|
f"[OPENAPI] {p.relative_to(SPECS_DIR)}: {op_name} {path} "
|
|
f"missing required path parameter(s): {', '.join(missing)}"
|
|
)
|
|
if extra:
|
|
out.append(
|
|
f"[OPENAPI] {p.relative_to(SPECS_DIR)}: {op_name} {path} "
|
|
f"declares path parameter(s) absent from template: {', '.join(extra)}"
|
|
)
|
|
return out
|
|
|
|
|
|
def check_rejected_decision_drift(paths: list[Path]) -> list[str]:
|
|
"""Catch known superseded design phrases in normative prose/contracts."""
|
|
out = []
|
|
for p in paths:
|
|
if p.suffix not in {".md", ".yaml", ".yml", ".json"}:
|
|
continue
|
|
for line_number, line in enumerate(p.read_text(encoding="utf-8", errors="ignore").splitlines(), 1):
|
|
for decision, pattern in REJECTED_DECISION_PATTERNS.items():
|
|
if re.search(pattern, line, flags=re.IGNORECASE):
|
|
out.append(
|
|
f"[DECISION] {p.relative_to(SPECS_DIR)}:{line_number}: "
|
|
f"superseded '{decision}'"
|
|
)
|
|
return out
|
|
|
|
|
|
def check_final_closure_invariants(paths: list[Path]) -> list[str]:
|
|
"""Keep the final cross-spec execution decisions from silently eroding."""
|
|
out = []
|
|
available = {p.relative_to(SPECS_DIR).as_posix(): p for p in paths}
|
|
for relative_path, fragments in REQUIRED_CONTRACT_FRAGMENTS.items():
|
|
p = available.get(relative_path)
|
|
if p is None:
|
|
continue
|
|
content = p.read_text(encoding="utf-8", errors="ignore")
|
|
for fragment in fragments:
|
|
if fragment not in content:
|
|
out.append(
|
|
f"[INVARIANT] {relative_path}: missing final-closure clause '{fragment}'"
|
|
)
|
|
return out
|
|
|
|
|
|
def gate(spec_ids: list[str]) -> int:
|
|
total = 0
|
|
for token in spec_ids:
|
|
dirs = _resolve_dir(token)
|
|
if dirs is None:
|
|
continue
|
|
if not isinstance(dirs, list):
|
|
dirs = [dirs]
|
|
for d in dirs:
|
|
paths = _walk(d.name)
|
|
print(f"\n=== {d.name} ({len(paths)} files) ===")
|
|
findings = []
|
|
findings += check_yaml(paths)
|
|
findings += check_json(paths)
|
|
findings += check_forbidden_identity(paths)
|
|
findings += check_revision_vs_content_hash(paths)
|
|
findings += check_openapi_path_parameters(paths)
|
|
findings += check_rejected_decision_drift(paths)
|
|
findings += check_final_closure_invariants(paths)
|
|
findings += check_fixtures_vs_schema(d.name)
|
|
findings += check_verification_program_boundaries(paths)
|
|
if not findings:
|
|
print(" PASS — machine contracts consistent")
|
|
else:
|
|
total += len(findings)
|
|
for f in findings:
|
|
print(f" FAIL {f}")
|
|
print(f"\nGate result: {'FAIL' if total else 'PASS'} ({total} findings)")
|
|
return 1 if total else 0
|
|
|
|
|
|
def main(argv: list[str]) -> int:
|
|
if len(argv) < 2:
|
|
print("Usage: python reconcile_contracts.py <spec-id> [<spec-id> ...] | all")
|
|
return 1
|
|
if "all" in argv:
|
|
ids = sorted(
|
|
p.name for p in SPECS_DIR.iterdir()
|
|
if p.is_dir() and re.match(r"^\d{3}-", p.name)
|
|
)
|
|
else:
|
|
ids = argv[1:]
|
|
return gate(ids)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main(sys.argv))
|
|
|
|
# #endregion Tooling.ReconcileContractsModule
|