200 lines
7.2 KiB
Python
200 lines
7.2 KiB
Python
"""
|
|
State Gate — Pre-merge / pre-ship state completeness validation.
|
|
|
|
Inspired by ADW's ``validate_state_completeness()`` which checks every core
|
|
field is non-None before allowing a merge. Architecture-agnostic: works with
|
|
JSON state files, dicts, or any key-value store.
|
|
|
|
Usage::
|
|
|
|
gate = StateGate(
|
|
FieldSpec("plan_file", required=True, predicate=lambda v: os.path.exists(v)),
|
|
FieldSpec("branch_name", required=True),
|
|
FieldSpec("backend_port", required=True, type_hint=int),
|
|
)
|
|
state = PipelineState({"plan_file": "/tmp/plan.md", "branch_name": "feat/x"})
|
|
result = gate.validate(state)
|
|
# result.passed == False (missing backend_port)
|
|
# result.missing == ["backend_port"]
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from dataclasses import dataclass, field, asdict
|
|
from datetime import datetime, timezone
|
|
from typing import Any, Callable, Optional
|
|
|
|
|
|
# ── FieldSpec ──────────────────────────────────────────────────────────────────
|
|
|
|
|
|
@dataclass
|
|
class FieldSpec:
|
|
"""Specification for a single pipeline state field.
|
|
|
|
Args:
|
|
name: Field key in the state dict.
|
|
required: If True, field must be present and non-None.
|
|
type_hint: Optional Python type for runtime type checking.
|
|
predicate: Optional custom predicate that receives the field value
|
|
and returns True if valid. Useful for file-exists checks, port
|
|
availability, URL validity, etc.
|
|
description: Human-readable description of what this field represents.
|
|
"""
|
|
|
|
name: str
|
|
required: bool = True
|
|
type_hint: Optional[type] = None
|
|
predicate: Optional[Callable[[Any], bool]] = None
|
|
description: str = ""
|
|
|
|
|
|
# ── PipelineState ──────────────────────────────────────────────────────────────
|
|
|
|
|
|
class PipelineState:
|
|
"""Immutable-ish state container with field filtering (ADW-style).
|
|
|
|
The ``core_fields`` filter prevents state bloat: only known keys are
|
|
accepted on update. Unknown keys are silently dropped with a warning.
|
|
"""
|
|
|
|
def __init__(self, initial: Optional[dict[str, Any]] = None, core_fields: Optional[list[str]] = None):
|
|
self._core_fields = core_fields or []
|
|
self._data: dict[str, Any] = {}
|
|
if initial:
|
|
self._apply(initial)
|
|
|
|
def _apply(self, data: dict[str, Any]) -> None:
|
|
if self._core_fields:
|
|
for key in list(data.keys()):
|
|
if key not in self._core_fields:
|
|
continue # silently drop unknown fields (ADW pattern)
|
|
self._data[key] = data[key]
|
|
else:
|
|
self._data.update(data)
|
|
|
|
def get(self, key: str, default: Any = None) -> Any:
|
|
return self._data.get(key, default)
|
|
|
|
def set(self, key: str, value: Any) -> None:
|
|
if self._core_fields and key not in self._core_fields:
|
|
return
|
|
self._data[key] = value
|
|
|
|
def update(self, **kwargs) -> None:
|
|
self._apply(kwargs)
|
|
|
|
@property
|
|
def data(self) -> dict[str, Any]:
|
|
return dict(self._data)
|
|
|
|
@property
|
|
def keys(self) -> list[str]:
|
|
return list(self._data.keys())
|
|
|
|
def __contains__(self, key: str) -> bool:
|
|
return key in self._data
|
|
|
|
def __repr__(self) -> str:
|
|
return f"PipelineState({self._data})"
|
|
|
|
|
|
# ── StateGate ──────────────────────────────────────────────────────────────────
|
|
|
|
|
|
@dataclass
|
|
class GateResult:
|
|
"""Result of a StateGate validation check."""
|
|
|
|
passed: bool
|
|
fields_checked: int = 0
|
|
missing: list[str] = field(default_factory=list)
|
|
type_errors: list[tuple[str, str]] = field(default_factory=list)
|
|
predicate_failures: list[tuple[str, str]] = field(default_factory=list)
|
|
timestamp: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
|
|
|
|
def dict(self) -> dict[str, Any]:
|
|
return {
|
|
"passed": self.passed,
|
|
"fields_checked": self.fields_checked,
|
|
"missing": self.missing,
|
|
"type_errors": self.type_errors,
|
|
"predicate_failures": self.predicate_failures,
|
|
"timestamp": self.timestamp,
|
|
}
|
|
|
|
def summary(self) -> str:
|
|
if self.passed:
|
|
return f"Gate PASSED: all {self.fields_checked} fields validated."
|
|
parts = [f"Gate FAILED ({self.fields_checked} fields checked):"]
|
|
if self.missing:
|
|
parts.append(f" Missing: {', '.join(self.missing)}")
|
|
if self.type_errors:
|
|
for name, msg in self.type_errors:
|
|
parts.append(f" Type error [{name}]: {msg}")
|
|
if self.predicate_failures:
|
|
for name, msg in self.predicate_failures:
|
|
parts.append(f" Predicate fail [{name}]: {msg}")
|
|
return "\n".join(parts)
|
|
|
|
|
|
class StateGate:
|
|
"""Validates that pipeline state is complete and consistent before allowing
|
|
a merge, ship, or transition to the next phase.
|
|
|
|
Analogous to ADW's ``validate_state_completeness()`` in adw_ship_iso.py.
|
|
"""
|
|
|
|
def __init__(self, *field_specs: FieldSpec, label: str = "pre-ship"):
|
|
self.field_specs = list(field_specs)
|
|
self.label = label
|
|
|
|
def validate(self, state: PipelineState) -> GateResult:
|
|
"""Run all field specs against a PipelineState.
|
|
|
|
Returns a GateResult with detailed failure information.
|
|
The caller decides whether to block based on ``result.passed``.
|
|
"""
|
|
missing: list[str] = []
|
|
type_errors: list[tuple[str, str]] = []
|
|
predicate_failures: list[tuple[str, str]] = []
|
|
|
|
for spec in self.field_specs:
|
|
value = state.get(spec.name)
|
|
|
|
# Required check
|
|
if spec.required and value is None:
|
|
missing.append(spec.name)
|
|
continue
|
|
|
|
# Type check (skip if value is None and not required — already caught above)
|
|
if spec.type_hint and value is not None:
|
|
if not isinstance(value, spec.type_hint):
|
|
type_errors.append(
|
|
(spec.name, f"expected {spec.type_hint.__name__}, got {type(value).__name__}")
|
|
)
|
|
|
|
# Predicate check
|
|
if spec.predicate and value is not None:
|
|
try:
|
|
if not spec.predicate(value):
|
|
predicate_failures.append(
|
|
(spec.name, f"predicate failed for value {value!r}")
|
|
)
|
|
except Exception as exc:
|
|
predicate_failures.append(
|
|
(spec.name, f"predicate raised {type(exc).__name__}: {exc}")
|
|
)
|
|
|
|
total = len(self.field_specs)
|
|
passed = not (missing or type_errors or predicate_failures)
|
|
return GateResult(
|
|
passed=passed,
|
|
fields_checked=total,
|
|
missing=missing,
|
|
type_errors=type_errors,
|
|
predicate_failures=predicate_failures,
|
|
)
|