feat: add zte pipeline primitives
This commit is contained in:
artale
parent
82a04879ea
commit
92b75b9629
|
|
@ -0,0 +1,29 @@
|
|||
"""
|
||||
pipeline — Portable, Pi-agnostic pipeline infrastructure.
|
||||
|
||||
Inspired by IndyDevDan's ADW/ZTE architecture, stripped of Pi dependencies.
|
||||
Each module is a standalone building block usable from cron jobs, CLI scripts,
|
||||
or delegator subagents.
|
||||
|
||||
Modules:
|
||||
confidence_ladder — 5-level Verdict ladder (PERFECT/VERIFIED/PARTIAL/FEEDBACK/FAILED)
|
||||
rejection_envelope — Structured rejection {received, reason, allowed, hint}
|
||||
state_gate — Pre-merge state completeness validation (ADW-style)
|
||||
"""
|
||||
|
||||
from .confidence_ladder import Verdict, VerdictLevel, VerdictReport
|
||||
from .rejection_envelope import RejectionEnvelope, Blocker
|
||||
from .state_gate import PipelineState, StateGate, FieldSpec
|
||||
from .expertise import (
|
||||
load as load_expertise,
|
||||
append_entry as append_expertise,
|
||||
render_context as render_expertise,
|
||||
purge_domain as purge_expertise,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
"Verdict", "VerdictLevel", "VerdictReport",
|
||||
"RejectionEnvelope", "Blocker",
|
||||
"PipelineState", "StateGate", "FieldSpec",
|
||||
"load_expertise", "append_expertise", "render_expertise", "purge_expertise",
|
||||
]
|
||||
|
|
@ -0,0 +1,163 @@
|
|||
"""
|
||||
Confidence Ladder — 5-level Verdict system.
|
||||
|
||||
Replaces binary pass/fail gates with a graduated confidence scale.
|
||||
Each verdict carries a human-readable summary and optional evidence.
|
||||
|
||||
Levels:
|
||||
PERFECT — All assertions passed, all paths verified
|
||||
VERIFIED — Core assertions passed, edge cases known
|
||||
PARTIAL — Nothing failed, but significant gaps were unverifiable
|
||||
FEEDBACK — Assertions failed OR verification revealed blocking issues
|
||||
FAILED — Assertions failed AND blocker will require rework
|
||||
|
||||
PARTIAL is the critical addition: it catches "green but dead" — cases where
|
||||
the system ran without errors but couldn't prove correctness.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from dataclasses import dataclass, field, asdict
|
||||
from datetime import datetime, timezone
|
||||
from enum import Enum
|
||||
from typing import Any, Optional
|
||||
|
||||
|
||||
class VerdictLevel(Enum):
|
||||
"""Five-level confidence ladder.
|
||||
|
||||
Ordered from most to least confident. Comparisons are meaningful:
|
||||
``PERFECT > VERIFIED > PARTIAL > FEEDBACK > FAILED``.
|
||||
"""
|
||||
|
||||
PERFECT = 5
|
||||
VERIFIED = 4
|
||||
PARTIAL = 3
|
||||
FEEDBACK = 2
|
||||
FAILED = 1
|
||||
|
||||
def __lt__(self, other: Any) -> bool:
|
||||
if not isinstance(other, VerdictLevel):
|
||||
return NotImplemented
|
||||
return self.value < other.value
|
||||
|
||||
def __le__(self, other: Any) -> bool:
|
||||
if not isinstance(other, VerdictLevel):
|
||||
return NotImplemented
|
||||
return self.value <= other.value
|
||||
|
||||
def __gt__(self, other: Any) -> bool:
|
||||
if not isinstance(other, VerdictLevel):
|
||||
return NotImplemented
|
||||
return self.value > other.value
|
||||
|
||||
def __ge__(self, other: Any) -> bool:
|
||||
if not isinstance(other, VerdictLevel):
|
||||
return NotImplemented
|
||||
return self.value >= other.value
|
||||
|
||||
@classmethod
|
||||
def from_str(cls, label: str) -> VerdictLevel:
|
||||
"""Parse a case-insensitive string to a VerdictLevel."""
|
||||
return cls[label.upper()]
|
||||
|
||||
@property
|
||||
def label(self) -> str:
|
||||
return self.name
|
||||
|
||||
@property
|
||||
def is_pass(self) -> bool:
|
||||
"""``True`` for PERFECT, VERIFIED, PARTIAL — the system should continue."""
|
||||
return self.value >= VerdictLevel.PARTIAL.value
|
||||
|
||||
@property
|
||||
def is_blocking(self) -> bool:
|
||||
"""``True`` for PERFECT, VERIFIED — may proceed without intervention."""
|
||||
return self.value >= VerdictLevel.VERIFIED.value
|
||||
|
||||
|
||||
@dataclass
|
||||
class Verdict:
|
||||
"""A single verdict from one verifier check."""
|
||||
|
||||
level: VerdictLevel
|
||||
summary: str
|
||||
evidence: list[str] = field(default_factory=list)
|
||||
source: Optional[str] = None # e.g. "critic:style", "test:unit"
|
||||
|
||||
def dict(self) -> dict[str, Any]:
|
||||
return {
|
||||
"level": self.level.name,
|
||||
"summary": self.summary,
|
||||
"evidence": self.evidence,
|
||||
"source": self.source,
|
||||
}
|
||||
|
||||
|
||||
@dataclass
|
||||
class VerdictReport:
|
||||
"""Aggregate report from one or more verifier checks."""
|
||||
|
||||
verdicts: list[Verdict] = field(default_factory=list)
|
||||
pipeline_id: Optional[str] = None # e.g. ADW ID or cron run ID
|
||||
timestamp: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
|
||||
|
||||
@property
|
||||
def aggregate(self) -> VerdictLevel:
|
||||
"""The lowest individual verdict sets the aggregate level.
|
||||
|
||||
The chain is only as strong as its weakest gate.
|
||||
"""
|
||||
if not self.verdicts:
|
||||
return VerdictLevel.FAILED
|
||||
return min(v.level for v in self.verdicts)
|
||||
|
||||
@property
|
||||
def aggregate_label(self) -> str:
|
||||
return self.aggregate.label
|
||||
|
||||
@property
|
||||
def passed(self) -> bool:
|
||||
"""``True`` if aggregate is PERFECT, VERIFIED, or PARTIAL."""
|
||||
return self.aggregate.is_pass
|
||||
|
||||
@property
|
||||
def blocked(self) -> bool:
|
||||
"""``True`` if aggregate is FEEDBACK or FAILED."""
|
||||
return not self.aggregate.is_pass
|
||||
|
||||
def add(self, verdict: Verdict) -> None:
|
||||
self.verdicts.append(verdict)
|
||||
|
||||
def dict(self) -> dict[str, Any]:
|
||||
return {
|
||||
"pipeline_id": self.pipeline_id,
|
||||
"timestamp": self.timestamp,
|
||||
"aggregate": self.aggregate_label,
|
||||
"passed": self.passed,
|
||||
"blocked": self.blocked,
|
||||
"verdicts": [v.dict() for v in self.verdicts],
|
||||
}
|
||||
|
||||
def to_json(self, indent: int = 2) -> str:
|
||||
return json.dumps(self.dict(), indent=indent)
|
||||
|
||||
@classmethod
|
||||
def from_json(cls, raw: str) -> VerdictReport:
|
||||
data = json.loads(raw)
|
||||
report = cls(pipeline_id=data.get("pipeline_id"), timestamp=data.get("timestamp", ""))
|
||||
for vd in data.get("verdicts", []):
|
||||
report.add(
|
||||
Verdict(
|
||||
level=VerdictLevel.from_str(vd["level"]),
|
||||
summary=vd["summary"],
|
||||
evidence=vd.get("evidence", []),
|
||||
source=vd.get("source"),
|
||||
)
|
||||
)
|
||||
return report
|
||||
|
||||
def threshhold_met(self, minimum: VerdictLevel = VerdictLevel.PARTIAL) -> bool:
|
||||
"""Check if aggregate meets a minimum confidence threshold."""
|
||||
return self.aggregate >= minimum
|
||||
|
|
@ -0,0 +1,221 @@
|
|||
#!/usr/bin/env python3
|
||||
"""
|
||||
Pipeline Demo — end-to-end walkthrough of all pipeline modules.
|
||||
|
||||
Demonstrates:
|
||||
1. Confidence Ladder (5-level Verdict)
|
||||
2. Rejection Envelope (4-field blocked action)
|
||||
3. State Gate (pre-merge completeness check)
|
||||
4. Expertise file (per-cron compounding memory)
|
||||
|
||||
Run: python -m pipeline.demo
|
||||
"""
|
||||
import json
|
||||
import sys
|
||||
import os
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
from pipeline.confidence_ladder import Verdict, VerdictLevel, VerdictReport
|
||||
from pipeline.rejection_envelope import RejectionEnvelope, Blocker
|
||||
from pipeline.state_gate import StateGate, FieldSpec, PipelineState
|
||||
from pipeline.expertise import append_entry, render_context, purge_domain
|
||||
|
||||
|
||||
def hr(title: str):
|
||||
print(f"\n{'=' * 60}")
|
||||
print(f" {title}")
|
||||
print(f"{'=' * 60}")
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════════════════════════════════════
|
||||
# 1. STATE INIT
|
||||
# ═══════════════════════════════════════════════════════════════════════════════
|
||||
hr("1. STATE INIT — PipelineState with core_fields")
|
||||
|
||||
state = PipelineState(
|
||||
{
|
||||
"plan_file": "plans/feat-123.md",
|
||||
"branch_name": "feat/instant-polls",
|
||||
"issue_number": 42,
|
||||
"rogue_field": "should be silently dropped",
|
||||
},
|
||||
core_fields=["plan_file", "branch_name", "issue_number", "backend_port", "frontend_port"],
|
||||
)
|
||||
print(f" plan_file: {state.get('plan_file')}")
|
||||
print(f" branch_name: {state.get('branch_name')}")
|
||||
print(f" issue_number: {state.get('issue_number')}")
|
||||
print(f" rogue dropped: {state.get('rogue_field')}") # should be None
|
||||
assert state.get("rogue_field") is None, "rogue fields are silently dropped"
|
||||
|
||||
# ═══════════════════════════════════════════════════════════════════════════════
|
||||
# 2. PLAN PHASE — PERFECT
|
||||
# ═══════════════════════════════════════════════════════════════════════════════
|
||||
hr("2. PLAN PHASE — Verdict: PERFECT")
|
||||
|
||||
report = VerdictReport(pipeline_id="demo-2026-06-11")
|
||||
report.add(Verdict(
|
||||
VerdictLevel.PERFECT,
|
||||
"Plan covers all acceptance criteria with test coverage map",
|
||||
evidence=["AC1 coverage: tests/test_instant_polls.py", "AC2 coverage: tests/test_instant_polls.py"],
|
||||
source="plan-review",
|
||||
))
|
||||
print(f" Ve rdict: {report.aggregate_label}")
|
||||
print(f" Pass? {report.passed}")
|
||||
|
||||
# ═══════════════════════════════════════════════════════════════════════════════
|
||||
# 3. BUILD PHASE — VERIFIED
|
||||
# ═══════════════════════════════════════════════════════════════════════════════
|
||||
hr("3. BUILD PHASE — Verdict: VERIFIED")
|
||||
|
||||
state.update(backend_port=9105, frontend_port=9205)
|
||||
report.add(Verdict(
|
||||
VerdictLevel.VERIFIED,
|
||||
"Build completed, CI confirmed all 14 tests pass",
|
||||
evidence=["exit code 0", "CI run #4185: all tests green"],
|
||||
source="ci",
|
||||
))
|
||||
print(f" Verdict: {report.aggregate_label}")
|
||||
print(f" Pass? {report.passed}")
|
||||
print(f" Verdicts: {len(report.verdicts)} phases so far")
|
||||
|
||||
# ═══════════════════════════════════════════════════════════════════════════════
|
||||
# 4. SECURITY REVIEW — PARTIAL (the critical gate)
|
||||
# ═══════════════════════════════════════════════════════════════════════════════
|
||||
hr("4. SECURITY REVIEW — Verdict: PARTIAL (green-but-dead check)")
|
||||
|
||||
report.add(Verdict(
|
||||
VerdictLevel.PARTIAL,
|
||||
"Static analysis clean, dependency scan clean — but no live penetration test was run",
|
||||
evidence=["semgrep: 0 findings", "pip audit: 0 known vulns", "NOTE: no live pen test — needs staging env"],
|
||||
source="security-review",
|
||||
))
|
||||
print(f" Verdict: {report.aggregate_label}")
|
||||
print(f" Pass? {report.passed} (PARTIAL = green + documented gap)")
|
||||
print(f" Blocked for merge? {report.aggregate.is_blocking}")
|
||||
print()
|
||||
print(" ▶ THIS is the green-but-dead prevention:")
|
||||
print(" The pipeline passes (green) but the security gap is")
|
||||
print(" documented, not forgotten. The reviewer knows what's missing.")
|
||||
assert report.aggregate == VerdictLevel.PARTIAL
|
||||
assert report.passed is True
|
||||
assert report.aggregate.is_blocking is False
|
||||
|
||||
# ═══════════════════════════════════════════════════════════════════════════════
|
||||
# 5. REJECTION ENVELOPE — leader tries to write a file
|
||||
# ═══════════════════════════════════════════════════════════════════════════════
|
||||
hr("5. REJECTION ENVELOPE — leader blocked from writing files")
|
||||
|
||||
blocker = Blocker(
|
||||
name="no_direct_write",
|
||||
match=lambda a: a.get("tool") in ("write_file", "patch", "execute_code"),
|
||||
reject=lambda a: RejectionEnvelope.block(
|
||||
received=a.get("tool", "unknown"),
|
||||
reason="LEADER_MUST_NOT_WRITE_FILES",
|
||||
allowed="Use delegate_task with a writer subagent",
|
||||
hint="Leaders read, delegate, synthesize. Delegate file writes.",
|
||||
),
|
||||
)
|
||||
|
||||
# This gets blocked
|
||||
blocked = blocker.evaluate({"tool": "write_file", "path": "src/main.py"})
|
||||
print(f" Action: write_file('src/main.py')")
|
||||
print(f" Result: BLOCKED")
|
||||
print(f" Reason: {blocked.reason}")
|
||||
print(f" Allowed: {blocked.allowed}")
|
||||
print(f" Hint: {blocked.hint}")
|
||||
print(f" Timestamp: {blocked.timestamp}")
|
||||
assert blocked is not None
|
||||
|
||||
# This passes through
|
||||
allowed = blocker.evaluate({"tool": "read_file", "path": "state.json"})
|
||||
print(f"\n Action: read_file('state.json')")
|
||||
print(f" Result: {'PASSES THROUGH (not blocked)' if allowed is None else 'BLOCKED'}")
|
||||
|
||||
# ═══════════════════════════════════════════════════════════════════════════════
|
||||
# 6. STATE GATE — pre-merge completeness check
|
||||
# ═══════════════════════════════════════════════════════════════════════════════
|
||||
hr("6. STATE GATE — pre-merge completeness check")
|
||||
|
||||
ship_gate = StateGate(
|
||||
FieldSpec("plan_file", required=True, predicate=lambda v: isinstance(v, str) and v.startswith("plans/")),
|
||||
FieldSpec("branch_name", required=True),
|
||||
FieldSpec("issue_number", required=True, type_hint=int),
|
||||
FieldSpec("backend_port", required=True, type_hint=int),
|
||||
FieldSpec("frontend_port", required=True, type_hint=int),
|
||||
label="pre-ship",
|
||||
)
|
||||
|
||||
# Complete state should pass
|
||||
result = ship_gate.validate(state)
|
||||
print(f" Fields checked: {result.fields_checked}")
|
||||
print(f" Pass: {result.passed}")
|
||||
if result.missing:
|
||||
print(f" Missing: {result.missing}")
|
||||
assert result.passed, "Complete state should pass ship gate"
|
||||
|
||||
# Incomplete state should fail
|
||||
incomplete = PipelineState(
|
||||
{"plan_file": "plans/feat-123.md", "branch_name": "feat/instant-polls"},
|
||||
core_fields=["plan_file", "branch_name", "issue_number", "backend_port", "frontend_port"],
|
||||
)
|
||||
fail_result = ship_gate.validate(incomplete)
|
||||
print(f"\n --- With missing fields ---")
|
||||
print(f" Pass: {fail_result.passed}")
|
||||
print(f" Missing: {fail_result.missing}")
|
||||
assert not fail_result.passed
|
||||
assert "issue_number" in fail_result.missing
|
||||
|
||||
# ═══════════════════════════════════════════════════════════════════════════════
|
||||
# 7. EXPERTISE — store what was learned
|
||||
# ═══════════════════════════════════════════════════════════════════════════════
|
||||
hr("7. EXPERTISE — store what the run learned")
|
||||
|
||||
append_entry(
|
||||
"demo-pipeline",
|
||||
domain="pipeline",
|
||||
insight="PARTIAL verdict is the key innovation — it lets documented gaps pass without blocking, preventing green-but-dead",
|
||||
source="cron:pipeline-demo",
|
||||
verified=True,
|
||||
)
|
||||
append_entry(
|
||||
"demo-pipeline",
|
||||
domain="confidence-ladder",
|
||||
insight="VERIFIED level requires independent confirmation (CI, review) — the agent trusting itself is only PERFECT",
|
||||
source="cron:pipeline-demo",
|
||||
verified=True,
|
||||
)
|
||||
append_entry(
|
||||
"demo-pipeline",
|
||||
domain="state-gate",
|
||||
insight="Core_fields filter ensures no rogue keys leak into pipeline state — strict struct, not a dict",
|
||||
source="cron:pipeline-demo",
|
||||
verified=False,
|
||||
)
|
||||
|
||||
print(render_context("demo-pipeline"))
|
||||
|
||||
# ═══════════════════════════════════════════════════════════════════════════════
|
||||
# 8. JSON PORTABILITY
|
||||
# ═══════════════════════════════════════════════════════════════════════════════
|
||||
hr("8. JSON PORTABILITY — serializable for cron/API delivery")
|
||||
|
||||
serialized = report.to_json()
|
||||
restored = VerdictReport.from_json(serialized)
|
||||
print(f" Round-trip aggregate matches: {restored.aggregate == report.aggregate}")
|
||||
print(f" Verdict count: {len(restored.verdicts)}")
|
||||
print(f"\n Final JSON payload (compact):")
|
||||
print(f" {json.dumps(json.loads(serialized), indent=2)}")
|
||||
|
||||
# ═══════════════════════════════════════════════════════════════════════════════
|
||||
# CLEANUP
|
||||
# ═══════════════════════════════════════════════════════════════════════════════
|
||||
purge_domain("demo-pipeline", "pipeline")
|
||||
purge_domain("demo-pipeline", "confidence-ladder")
|
||||
purge_domain("demo-pipeline", "state-gate")
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
print(f"\n{'=' * 60}")
|
||||
print(f" ✅ DEMO COMPLETE — all 8 stages verified")
|
||||
print(f"{'=' * 60}")
|
||||
|
|
@ -0,0 +1,198 @@
|
|||
#!/usr/bin/env python3
|
||||
"""
|
||||
Expertise File Manager — per-cron compounding memory.
|
||||
|
||||
Each recurring cron job gets its own expertise file at
|
||||
~/.hermes/expertise/<job-name>.yaml
|
||||
|
||||
The file is loaded at job start and updated at job end, making every run
|
||||
smarter than the last.
|
||||
|
||||
Usage:
|
||||
# Load expertise into context (inline at top of cron prompt):
|
||||
python -m pipeline.expertise --load --job triage
|
||||
|
||||
# Append a new insight after work is done:
|
||||
python -m pipeline.expertise --append --job triage --domain github-issues \
|
||||
--insight "Issues with label 'bug' most often miss reproduction steps"
|
||||
|
||||
# Read all expertise entries (JSON for tool consumption):
|
||||
python -m pipeline.expertise --read --job triage
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any, Optional
|
||||
|
||||
import yaml
|
||||
|
||||
|
||||
# ── Paths ─────────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def expertise_dir() -> str:
|
||||
"""Return ~/.hermes/expertise/ directory, creating it if needed."""
|
||||
base = os.path.join(os.path.expanduser("~"), ".hermes", "expertise")
|
||||
os.makedirs(base, exist_ok=True)
|
||||
return base
|
||||
|
||||
|
||||
def expertise_path(job_name: str) -> str:
|
||||
return os.path.join(expertise_dir(), f"{job_name}.yaml")
|
||||
|
||||
|
||||
# ── Data Model ────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
ENTRY_SCHEMA = {
|
||||
"domain": str,
|
||||
"insight": str,
|
||||
"date": str,
|
||||
"source": str,
|
||||
"verified": bool,
|
||||
}
|
||||
|
||||
|
||||
def default_expertise() -> dict:
|
||||
return {
|
||||
"updatable": True,
|
||||
"version": 1,
|
||||
"entries": [],
|
||||
}
|
||||
|
||||
|
||||
# ── Load / Save ───────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def load(job_name: str) -> dict:
|
||||
path = expertise_path(job_name)
|
||||
if os.path.exists(path):
|
||||
with open(path, "r", encoding="utf-8") as f:
|
||||
data = yaml.safe_load(f) or default_expertise()
|
||||
data.setdefault("updatable", True)
|
||||
data.setdefault("version", 1)
|
||||
data.setdefault("entries", [])
|
||||
return data
|
||||
return default_expertise()
|
||||
|
||||
|
||||
def save(job_name: str, data: dict) -> None:
|
||||
path = expertise_path(job_name)
|
||||
with open(path, "w", encoding="utf-8") as f:
|
||||
yaml.dump(
|
||||
data,
|
||||
f,
|
||||
default_flow_style=False,
|
||||
sort_keys=False,
|
||||
allow_unicode=True,
|
||||
)
|
||||
|
||||
|
||||
# ── Operations ────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def append_entry(
|
||||
job_name: str,
|
||||
domain: str,
|
||||
insight: str,
|
||||
source: Optional[str] = None,
|
||||
verified: bool = False,
|
||||
) -> dict:
|
||||
"""Append a new insight entry to the job's expertise file."""
|
||||
data = load(job_name)
|
||||
if not data.get("updatable", True):
|
||||
print(f"Expertise file for '{job_name}' is not updatable.", file=sys.stderr)
|
||||
return data
|
||||
|
||||
entry = {
|
||||
"domain": domain,
|
||||
"insight": insight,
|
||||
"date": datetime.now(timezone.utc).strftime("%Y-%m-%d"),
|
||||
"source": source or f"cron:{job_name}",
|
||||
"verified": verified,
|
||||
}
|
||||
data["entries"].append(entry)
|
||||
save(job_name, data)
|
||||
return data
|
||||
|
||||
|
||||
def render_context(job_name: str) -> str:
|
||||
"""Render expertise entries as a context string for agent prompts."""
|
||||
data = load(job_name)
|
||||
if not data["entries"]:
|
||||
return ""
|
||||
lines = [f"# Expertise ({job_name})"]
|
||||
for i, e in enumerate(data["entries"], 1):
|
||||
tag = "✓" if e.get("verified") else "·"
|
||||
lines.append(f"{tag} [{e['domain']}] {e['insight']} ({e['date']})")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def purge_domain(job_name: str, domain: str) -> dict:
|
||||
"""Remove all entries for a specific domain."""
|
||||
data = load(job_name)
|
||||
before = len(data["entries"])
|
||||
data["entries"] = [e for e in data["entries"] if e.get("domain") != domain]
|
||||
removed = before - len(data["entries"])
|
||||
if removed:
|
||||
save(job_name, data)
|
||||
print(f"Removed {removed} entries for domain '{domain}'.")
|
||||
else:
|
||||
print(f"No entries found for domain '{domain}'.")
|
||||
return data
|
||||
|
||||
|
||||
# ── CLI ───────────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="Expertise file manager")
|
||||
parser.add_argument("--job", required=True, help="Job name (corresponds to expertise filename)")
|
||||
parser.add_argument("--load", action="store_true", help="Print expertise as context (for injection at cron start)")
|
||||
parser.add_argument("--read", action="store_true", help="Dump entire expertise file as JSON")
|
||||
parser.add_argument("--append", action="store_true", help="Append a new insight entry")
|
||||
parser.add_argument("--domain", default="general", help="Domain tag for the insight")
|
||||
parser.add_argument("--insight", help="The insight text to store")
|
||||
parser.add_argument("--source", help="Source identifier (default: cron:<job>)")
|
||||
parser.add_argument("--verified", action="store_true", default=False, help="Mark insight as verified")
|
||||
parser.add_argument("--purge", help="Remove all entries for a domain")
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
if args.load:
|
||||
print(render_context(args.job))
|
||||
return
|
||||
|
||||
if args.read:
|
||||
data = load(args.job)
|
||||
print(json.dumps(data, indent=2))
|
||||
return
|
||||
|
||||
if args.purge:
|
||||
purge_domain(args.job, args.purge)
|
||||
return
|
||||
|
||||
if args.append:
|
||||
if not args.insight:
|
||||
print("ERROR: --insight is required with --append", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
append_entry(
|
||||
args.job,
|
||||
domain=args.domain,
|
||||
insight=args.insight,
|
||||
source=args.source,
|
||||
verified=args.verified,
|
||||
)
|
||||
print(f"Appended insight to {args.job}.yaml")
|
||||
return
|
||||
|
||||
parser.print_help()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
|
@ -0,0 +1,66 @@
|
|||
#!/bin/bash
|
||||
"""":"
|
||||
|
||||
# ── Cron Expertise Integrator ─────────────────────────────────────────────────
|
||||
#
|
||||
# Injects expertise load at cron job start and stores new insights at end.
|
||||
#
|
||||
# Usage — add to top of cron prompt:
|
||||
# ```
|
||||
# Load your expertise file:
|
||||
# python -m pipeline.expertise --job <cron-job-name> --load
|
||||
# ```
|
||||
#
|
||||
# Usage — add to end of cron prompt:
|
||||
# ```
|
||||
# Before finishing, store any new learnings:
|
||||
# python -m pipeline.expertise --job <cron-job-name> --append \
|
||||
# --domain "<domain>" --insight "<insight>"
|
||||
# ```
|
||||
#
|
||||
# Or wrap a shell job:
|
||||
# pipeline/expertise_cron.sh --job triage --domain github-issues -- bash ./triage.sh
|
||||
# ────────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
# Shell wrapper for bash-based cron jobs that want to store expertise.
|
||||
# Calls python -m pipeline.expertise under the hood.
|
||||
|
||||
set -euo pipefail
|
||||
|
||||
JOB=""
|
||||
DOMAIN=""
|
||||
COMMAND=()
|
||||
|
||||
while [[ $# -gt 0 ]]; do
|
||||
case "$1" in
|
||||
--job) JOB="$2"; shift 2 ;;
|
||||
--domain) DOMAIN="$2"; shift 2 ;;
|
||||
--) shift; COMMAND=("$@"); break ;;
|
||||
*) echo "Unknown: $1"; exit 1 ;;
|
||||
esac
|
||||
done
|
||||
|
||||
if [[ -z "$JOB" || ${#COMMAND[@]} -eq 0 ]]; then
|
||||
echo "Usage: $0 --job <name> --domain <domain> -- <command...>"
|
||||
exit 1
|
||||
fi
|
||||
|
||||
# Load expertise context
|
||||
echo "=== Expertise ($JOB) ==="
|
||||
python -m pipeline.expertise --job "$JOB" --load || true
|
||||
echo "=== End Expertise ==="
|
||||
|
||||
# Run the actual command
|
||||
"${COMMAND[@]}"
|
||||
|
||||
# Store the exit code
|
||||
RC=$?
|
||||
|
||||
# On success, prompt the user/agent to store what they learned
|
||||
if [[ $RC -eq 0 ]]; then
|
||||
echo ""
|
||||
echo "Job succeeded. If you learned something worth remembering, store it:"
|
||||
echo " python -m pipeline.expertise --job $JOB --append --domain \"$DOMAIN\" --insight \"<what you learned>\""
|
||||
fi
|
||||
|
||||
exit $RC
|
||||
|
|
@ -0,0 +1,107 @@
|
|||
"""
|
||||
Structured Rejection Envelope — {received, reason, allowed, hint}.
|
||||
|
||||
Every blocked action returns a standard 4-field envelope so the verifier is
|
||||
self-documenting: the error message IS the documentation.
|
||||
|
||||
Inspired by pi-agents's blocked-action format, ported to be Agno-agnostic.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from dataclasses import dataclass, field, asdict
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any, Optional
|
||||
|
||||
|
||||
@dataclass
|
||||
class RejectionEnvelope:
|
||||
"""Universal envelope for blocked actions.
|
||||
|
||||
Fields:
|
||||
received: What the agent tried to do (verbatim action or intent)
|
||||
reason: Why it was blocked (machine-readable short code + human explanation)
|
||||
allowed: What the agent CAN do instead (alternative action, or None if completely denied)
|
||||
hint: How to resolve the rejection (docs reference, fix instruction, or None)
|
||||
"""
|
||||
|
||||
received: str
|
||||
reason: str
|
||||
allowed: Optional[str] = None
|
||||
hint: Optional[str] = None
|
||||
timestamp: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
|
||||
|
||||
def dict(self) -> dict[str, Any]:
|
||||
return {
|
||||
"received": self.received,
|
||||
"reason": self.reason,
|
||||
"allowed": self.allowed,
|
||||
"hint": self.hint,
|
||||
"timestamp": self.timestamp,
|
||||
}
|
||||
|
||||
def to_json(self, indent: int = 2) -> str:
|
||||
return json.dumps(self.dict(), indent=indent)
|
||||
|
||||
def __str__(self) -> str:
|
||||
parts = [f"BLOCKED: {self.reason}"]
|
||||
if self.allowed:
|
||||
parts.append(f" Allowed: {self.allowed}")
|
||||
if self.hint:
|
||||
parts.append(f" Hint: {self.hint}")
|
||||
return "\n".join(parts)
|
||||
|
||||
@classmethod
|
||||
def from_json(cls, raw: str) -> RejectionEnvelope:
|
||||
data = json.loads(raw)
|
||||
return cls(
|
||||
received=data["received"],
|
||||
reason=data["reason"],
|
||||
allowed=data.get("allowed"),
|
||||
hint=data.get("hint"),
|
||||
timestamp=data.get("timestamp", ""),
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def block(
|
||||
cls,
|
||||
received: str,
|
||||
reason: str,
|
||||
allowed: Optional[str] = None,
|
||||
hint: Optional[str] = None,
|
||||
) -> RejectionEnvelope:
|
||||
"""Convenience constructor for a blocking rejection."""
|
||||
return cls(received=received, reason=reason, allowed=allowed, hint=hint)
|
||||
|
||||
|
||||
# ── Blocker: a rule that generates RejectionEnvelopes ──────────────────────────
|
||||
|
||||
|
||||
@dataclass
|
||||
class Blocker:
|
||||
"""A named rule that evaluates an action and returns a rejection or None.
|
||||
|
||||
Usage::
|
||||
|
||||
no_write_file = Blocker(
|
||||
name="no_direct_write",
|
||||
match=lambda action: "write_file" in action.get("tool", ""),
|
||||
reject=lambda action: RejectionEnvelope.block(
|
||||
received=str(action),
|
||||
reason="LEADER_MUST_NOT_WRITE_FILES",
|
||||
allowed="Use delegate_task to spawn a writer subagent",
|
||||
hint="Leaders read, delegate, and synthesize — they never write files directly.",
|
||||
),
|
||||
)
|
||||
"""
|
||||
|
||||
name: str
|
||||
match: Any # Callable[[dict], bool]
|
||||
reject: Any # Callable[[dict], RejectionEnvelope]
|
||||
|
||||
def evaluate(self, action: dict) -> Optional[RejectionEnvelope]:
|
||||
"""Return a RejectionEnvelope if this action is blocked, None otherwise."""
|
||||
if self.match(action):
|
||||
return self.reject(action)
|
||||
return None
|
||||
|
|
@ -0,0 +1,199 @@
|
|||
"""
|
||||
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,
|
||||
)
|
||||
|
|
@ -0,0 +1,116 @@
|
|||
"""Quick smoke test for all pipeline modules."""
|
||||
import json
|
||||
import sys
|
||||
import os
|
||||
|
||||
# Add pipeline dir to path
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
from pipeline.confidence_ladder import Verdict, VerdictLevel, VerdictReport
|
||||
from pipeline.rejection_envelope import RejectionEnvelope, Blocker
|
||||
from pipeline.state_gate import StateGate, FieldSpec, PipelineState, GateResult
|
||||
|
||||
errors = 0
|
||||
|
||||
# ── Test 1: VerdictLevel ordering ──
|
||||
assert VerdictLevel.PERFECT > VerdictLevel.VERIFIED, "PERFECT should be > VERIFIED"
|
||||
assert VerdictLevel.PARTIAL > VerdictLevel.FEEDBACK, "PARTIAL should be > FEEDBACK"
|
||||
assert VerdictLevel.FAILED < VerdictLevel.FEEDBACK, "FAILED should be < FEEDBACK"
|
||||
assert VerdictLevel.PARTIAL.is_pass, "PARTIAL should be pass"
|
||||
assert VerdictLevel.FEEDBACK.is_pass == False, "FEEDBACK should NOT be pass"
|
||||
assert VerdictLevel.VERIFIED.is_blocking, "VERIFIED should be blocking"
|
||||
assert VerdictLevel.PARTIAL.is_blocking == False, "PARTIAL should NOT be blocking"
|
||||
print("✓ VerdictLevel ordering and predicates")
|
||||
|
||||
# ── Test 2: VerdictReport aggregation ──
|
||||
report = VerdictReport(pipeline_id="test-001")
|
||||
report.add(Verdict(VerdictLevel.PERFECT, "style check passed"))
|
||||
report.add(Verdict(VerdictLevel.PARTIAL, "unit tests: can't test DB layer offline", ["no DB in CI"]))
|
||||
report.add(Verdict(VerdictLevel.VERIFIED, "integration tests passed"))
|
||||
assert report.aggregate == VerdictLevel.PARTIAL, f"Aggregate should be PARTIAL (lowest), got {report.aggregate}"
|
||||
assert report.passed == True, "PARTIAL should be a pass"
|
||||
print("✓ VerdictReport aggregation (PARTIAL is min, but still pass)")
|
||||
|
||||
# ── Test 3: VerdictReport JSON round-trip ──
|
||||
serialized = report.to_json()
|
||||
restored = VerdictReport.from_json(serialized)
|
||||
assert restored.aggregate == report.aggregate, "JSON round-trip should preserve aggregate"
|
||||
assert len(restored.verdicts) == len(report.verdicts), "JSON round-trip should preserve verdict count"
|
||||
print("✓ JSON serialization round-trip")
|
||||
|
||||
# ── Test 4: RejectionEnvelope ──
|
||||
reject = RejectionEnvelope.block(
|
||||
received="write_file('foo.py')",
|
||||
reason="LEADER_MUST_NOT_WRITE_FILES",
|
||||
allowed="Delegate write to a writer subagent",
|
||||
hint="Leaders read, delegate, synthesize. Never write files directly.",
|
||||
)
|
||||
assert "BLOCKED" in str(reject)
|
||||
assert reject.received == "write_file('foo.py')"
|
||||
reject_json = reject.to_json()
|
||||
reject2 = RejectionEnvelope.from_json(reject_json)
|
||||
assert reject2.reason == "LEADER_MUST_NOT_WRITE_FILES"
|
||||
print("✓ RejectionEnvelope with 4-field schema")
|
||||
|
||||
# ── Test 5: Blocker rule ──
|
||||
no_write = Blocker(
|
||||
name="no_direct_write",
|
||||
match=lambda a: "write_file" in str(a.get("tool", "")),
|
||||
reject=lambda a: RejectionEnvelope.block(
|
||||
received=str(a),
|
||||
reason="LEADER_MUST_NOT_WRITE_FILES",
|
||||
),
|
||||
)
|
||||
result = no_write.evaluate({"tool": "write_file", "path": "test.py"})
|
||||
assert result is not None, "write_file should be blocked"
|
||||
assert result.reason == "LEADER_MUST_NOT_WRITE_FILES"
|
||||
result = no_write.evaluate({"tool": "read_file", "path": "test.py"})
|
||||
assert result is None, "read_file should NOT be blocked"
|
||||
print("✓ Blocker rule evaluation")
|
||||
|
||||
# ── Test 6: PipelineState with core_fields ──
|
||||
state = PipelineState(
|
||||
{"plan_file": "plan.md", "branch_name": "feat/x", "rogue_key": "should_be_dropped"},
|
||||
core_fields=["plan_file", "branch_name", "backend_port"],
|
||||
)
|
||||
assert state.get("plan_file") == "plan.md"
|
||||
assert state.get("branch_name") == "feat/x"
|
||||
assert state.get("rogue_key") is None, "rogue_key should be silently dropped"
|
||||
state.update(backend_port=9100, another_rogue="dropped")
|
||||
assert state.get("backend_port") == 9100
|
||||
assert state.get("another_rogue") is None
|
||||
print("✓ PipelineState with core_fields filter")
|
||||
|
||||
# ── Test 7: StateGate validation ──
|
||||
gate = StateGate(
|
||||
FieldSpec("plan_file", required=True),
|
||||
FieldSpec("branch_name", required=True),
|
||||
FieldSpec("backend_port", required=True, type_hint=int),
|
||||
)
|
||||
incomplete = PipelineState({"plan_file": "plan.md"}, core_fields=["plan_file", "branch_name", "backend_port"])
|
||||
result = gate.validate(incomplete)
|
||||
assert not result.passed, "Missing fields should fail"
|
||||
assert "branch_name" in result.missing
|
||||
assert "backend_port" in result.missing
|
||||
print("✓ StateGate catches missing fields")
|
||||
|
||||
complete = PipelineState(
|
||||
{"plan_file": "plan.md", "branch_name": "feat/x", "backend_port": 9100},
|
||||
core_fields=["plan_file", "branch_name", "backend_port"],
|
||||
)
|
||||
result = gate.validate(complete)
|
||||
assert result.passed, "Complete state should pass"
|
||||
print("✓ StateGate passes complete state")
|
||||
|
||||
# ── Test 8: Type validation ──
|
||||
wrong_type = PipelineState(
|
||||
{"plan_file": "plan.md", "branch_name": "feat/x", "backend_port": "not-an-int"},
|
||||
core_fields=["plan_file", "branch_name", "backend_port"],
|
||||
)
|
||||
result = gate.validate(wrong_type)
|
||||
assert not result.passed, "Type mismatch should fail"
|
||||
assert any("backend_port" in e[0] for e in result.type_errors), "Should report type error for backend_port"
|
||||
print("✓ StateGate type validation")
|
||||
|
||||
print(f"\n✅ All {8 - errors} tests passed!" if errors == 0 else f"\n❌ {errors} test(s) failed")
|
||||
sys.exit(0 if errors == 0 else 1)
|
||||
|
|
@ -0,0 +1,134 @@
|
|||
"""
|
||||
Integration test: full pipeline flow.
|
||||
|
||||
Simulates a complete ADW-style pipeline run with:
|
||||
1. State initialization + core_fields filter
|
||||
2. Multiple phases, each producing a Verdict
|
||||
3. Rejection envelope for blocked actions
|
||||
4. State gate at the end
|
||||
5. Expertise entry for what was learned
|
||||
"""
|
||||
import json
|
||||
import sys
|
||||
import os
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
# Also add the parent of pipeline to ensure direct imports work
|
||||
_parent = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
if _parent not in sys.path:
|
||||
sys.path.insert(0, _parent)
|
||||
|
||||
from pipeline.confidence_ladder import Verdict, VerdictLevel, VerdictReport
|
||||
from pipeline.rejection_envelope import RejectionEnvelope, Blocker
|
||||
from pipeline.state_gate import StateGate, FieldSpec, PipelineState
|
||||
from pipeline.expertise import append_entry, render_context, purge_domain
|
||||
|
||||
errors = 0
|
||||
|
||||
# ── Step 1: Pipeline state with core_fields ──
|
||||
state = PipelineState(
|
||||
{"plan_file": "plans/feat-123.md", "branch_name": "feat/instant-polls", "issue_number": 42},
|
||||
core_fields=["plan_file", "branch_name", "issue_number", "backend_port", "frontend_port", "worktree_path"],
|
||||
)
|
||||
assert state.get("plan_file") == "plans/feat-123.md"
|
||||
print("✓ Step 1: State init with core_fields")
|
||||
|
||||
# ── Step 2: Plan phase verdict ──
|
||||
report = VerdictReport(pipeline_id="adw-abc12345")
|
||||
report.add(Verdict(VerdictLevel.VERIFIED, "Plan covers all acceptance criteria", ["AC1 covered", "AC2 covered"], source="plan-review"))
|
||||
print(f" Plan verdict: {report.aggregate_label}")
|
||||
assert report.aggregate == VerdictLevel.VERIFIED
|
||||
|
||||
# ── Step 3: Build phase produces artifacts ──
|
||||
state.update(backend_port=9105, frontend_port=9205, worktree_path="/tmp/worktrees/adw-abc12345")
|
||||
report.add(Verdict(VerdictLevel.PERFECT, "Build completed, tests pass in worktree", ["exit code 0", "all 14 tests pass"], source="build"))
|
||||
print(f" Build verdict: PERFECT")
|
||||
|
||||
# ── Step 4: A blocked action (leader tries to write a file) ──
|
||||
no_write_blocker = Blocker(
|
||||
name="no_direct_write",
|
||||
match=lambda a: a.get("tool") in ("write_file", "patch", "execute_code"),
|
||||
reject=lambda a: RejectionEnvelope.block(
|
||||
received=a.get("tool", "unknown"),
|
||||
reason="LEADER_MUST_NOT_WRITE_FILES",
|
||||
allowed="Use delegate_task with a writer subagent",
|
||||
hint="Leaders read, delegate, synthesize. Delegate file writes.",
|
||||
),
|
||||
)
|
||||
blocked = no_write_blocker.evaluate({"tool": "write_file", "path": "src/main.py"})
|
||||
assert blocked is not None, "write_file should be blocked"
|
||||
print(f"✓ Step 4: Action blocked — {blocked.reason}")
|
||||
print(f" Allowed: {blocked.allowed}")
|
||||
|
||||
# Verify a non-blocked action passes
|
||||
allowed_action = no_write_blocker.evaluate({"tool": "read_file", "path": "state.json"})
|
||||
assert allowed_action is None, "read_file should NOT be blocked"
|
||||
print("✓ Step 4b: read_file passes through")
|
||||
|
||||
# ── Step 5: Security review finds gaps (PARTIAL) ──
|
||||
report.add(Verdict(VerdictLevel.PARTIAL, "No vulnerabilities found, but no live penetration test was run", ["static analysis clean", "dependency scan clean", "NOTE: no live pen test"], source="security-review"))
|
||||
print(f" Security verdict: PARTIAL (important: nothing failed, but gap is documented)")
|
||||
|
||||
# ── Step 6: Check aggregate after all phases ──
|
||||
print(f" Aggregate: {report.aggregate_label}")
|
||||
assert report.aggregate == VerdictLevel.PARTIAL, f"Aggregate should be PARTIAL (lowest of VERIFIED+PERFECT+PARTIAL), got {report.aggregate_label}"
|
||||
assert report.passed == True, "PARTIAL should pass (green but documented gap)"
|
||||
print("✓ Step 6: Aggregate correctly identifies PARTIAL gate")
|
||||
|
||||
# ── Step 7: State completeness gate before ship ──
|
||||
ship_gate = StateGate(
|
||||
FieldSpec("plan_file", required=True, predicate=lambda v: isinstance(v, str) and v.startswith("plans/")),
|
||||
FieldSpec("branch_name", required=True),
|
||||
FieldSpec("issue_number", required=True, type_hint=int),
|
||||
FieldSpec("backend_port", required=True, type_hint=int),
|
||||
FieldSpec("frontend_port", required=True, type_hint=int),
|
||||
FieldSpec("worktree_path", required=True),
|
||||
label="pre-ship",
|
||||
)
|
||||
gate_result = ship_gate.validate(state)
|
||||
assert gate_result.passed, f"Complete state should pass ship gate: {gate_result.summary()}"
|
||||
print(f"✓ Step 7: Ship gate passed ({gate_result.fields_checked} fields validated)")
|
||||
|
||||
# ── Step 8: State gate catches missing fields ──
|
||||
incomplete = PipelineState(
|
||||
{"plan_file": "plans/feat-123.md", "branch_name": "feat/instant-polls"},
|
||||
core_fields=state._core_fields,
|
||||
)
|
||||
fail_result = ship_gate.validate(incomplete)
|
||||
assert not fail_result.passed
|
||||
assert "issue_number" in fail_result.missing
|
||||
assert "backend_port" in fail_result.missing
|
||||
print(f"✓ Step 8: Ship gate correctly catches missing: {fail_result.missing}")
|
||||
|
||||
# ── Step 9: Store expertise about what was learned ──
|
||||
append_entry(
|
||||
"integration-test",
|
||||
domain="pipeline",
|
||||
insight="PARTIAL verdict allows documented gaps to pass without blocking — critical for green-but-dead prevention",
|
||||
source="cron:pipeline-integration-test",
|
||||
verified=True,
|
||||
)
|
||||
append_entry(
|
||||
"integration-test",
|
||||
domain="state-gate",
|
||||
insight="Required fields must include worktree_path and ports, not just plan_file and branch_name",
|
||||
source="cron:pipeline-integration-test",
|
||||
)
|
||||
context = render_context("integration-test")
|
||||
assert "PARTIAL" in context
|
||||
assert "worktree_path" in context
|
||||
print("✓ Step 9: Expertise entries stored and loadable")
|
||||
|
||||
# ── Step 10: JSON serialization of final report ──
|
||||
serialized = report.to_json()
|
||||
restored = VerdictReport.from_json(serialized)
|
||||
assert restored.aggregate == report.aggregate
|
||||
assert len(restored.verdicts) == len(report.verdicts)
|
||||
print("✓ Step 10: Full VerdictReport JSON round-trip")
|
||||
|
||||
# Cleanup test expertise file
|
||||
purge_domain("integration-test", "pipeline")
|
||||
purge_domain("integration-test", "state-gate")
|
||||
|
||||
print(f"\n✅ Integration test passed ({10 - errors}/10)")
|
||||
sys.exit(0 if errors == 0 else 1)
|
||||
Loading…
Reference in New Issue