diff --git a/pipeline/__init__.py b/pipeline/__init__.py
new file mode 100644
index 0000000..7abe59c
--- /dev/null
+++ b/pipeline/__init__.py
@@ -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",
+]
diff --git a/pipeline/confidence_ladder.py b/pipeline/confidence_ladder.py
new file mode 100644
index 0000000..d68f41b
--- /dev/null
+++ b/pipeline/confidence_ladder.py
@@ -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
diff --git a/pipeline/demo.py b/pipeline/demo.py
new file mode 100644
index 0000000..41dfb34
--- /dev/null
+++ b/pipeline/demo.py
@@ -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}")
diff --git a/pipeline/expertise.py b/pipeline/expertise.py
new file mode 100644
index 0000000..0eed593
--- /dev/null
+++ b/pipeline/expertise.py
@@ -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()
diff --git a/pipeline/expertise_cron.sh b/pipeline/expertise_cron.sh
new file mode 100644
index 0000000..fbe2fc8
--- /dev/null
+++ b/pipeline/expertise_cron.sh
@@ -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
diff --git a/pipeline/rejection_envelope.py b/pipeline/rejection_envelope.py
new file mode 100644
index 0000000..b8c4f00
--- /dev/null
+++ b/pipeline/rejection_envelope.py
@@ -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
diff --git a/pipeline/state_gate.py b/pipeline/state_gate.py
new file mode 100644
index 0000000..55307a4
--- /dev/null
+++ b/pipeline/state_gate.py
@@ -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,
+        )
diff --git a/pipeline/test_all.py b/pipeline/test_all.py
new file mode 100644
index 0000000..cc580b6
--- /dev/null
+++ b/pipeline/test_all.py
@@ -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)
diff --git a/pipeline/test_integration.py b/pipeline/test_integration.py
new file mode 100644
index 0000000..a50e818
--- /dev/null
+++ b/pipeline/test_integration.py
@@ -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)