agentic-ai-engineering/pipeline/state_gate.py

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,
)