Improvement Gate

[
  {
    "type": "coverage",
    "succeeded": true,
    "context": {
      "dimension": "coverage",
      "scores": {
        "coverage": 1.0,
        "accuracy": 0.8,
        "efficiency": 1.0,
        "reliability": 1.0,
        "security": 0.8
      }
    },
    "timestamp": "2026-04-02T12:14:25Z",
    "hit_count": 1
  },
  {
    "type": "instruction",
    "succeeded": true,
    "context": {
      "dimension": "accuracy",
      "scores": {
        "coverage": 1.0,
        "accuracy": 0.8,
        "efficiency": 1.0,
        "reliability": 1.0,
        "security": 0.8
      }
    },
    "timestamp": "2026-04-02T12:14:26Z",
    "hit_count": 1
  }
]

#!/usr/bin/env python3
"""Gate for the first runnable generic-skill lane.

Includes a 6-layer mechanical validation system that runs BEFORE the
keep/pending_promote/revert/reject decision logic.
"""

from __future__ import annotations

import argparse
import sys
from pathlib import Path

_REPO_ROOT = Path(__file__).resolve().parents[3]
if str(_REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(_REPO_ROOT))

from lib.common import (
    KEEP_CATEGORIES,
    SCHEMA_VERSION,
    protected_target,
    read_json,
    utc_now_iso,
    write_json,
)
from lib.state_machine import (
    DEFAULT_STATE_ROOT,
    append_pending_promote,
    append_veto,
    ensure_tree,
    make_receipt_path,
    restore_backup,
    update_state,
)


# ---------------------------------------------------------------------------
# 6-layer mechanical validation
# ---------------------------------------------------------------------------


class GateLayer:
    """Base class for gate validation layers."""

    def __init__(self, name: str, required: bool = True):
        self.name = name
        self.required = required

    def validate(self, candidate: dict, execution: dict | None = None) -> dict:
        """Returns {passed: bool, details: str, layer: str}"""
        raise NotImplementedError


class SchemaGate(GateLayer):
    """Layer 0: Validate JSON artifact structure."""

    def __init__(self):
        super().__init__("schema", required=True)

    VALID_CATEGORIES = {"docs", "reference", "guardrail", "prompt", "workflow", "tests"}
    VALID_RISK_LEVELS = {"low", "medium", "high"}

    def validate(self, candidate, execution=None):
        required_fields = ["id", "category", "risk_level", "execution_plan"]
        missing = [f for f in required_fields if f not in candidate]
        if missing:
            return {"passed": False, "details": f"Missing: {missing}", "layer": self.name}
        errors = []
        if not isinstance(candidate.get("id"), str) or not candidate["id"]:
            errors.append("id must be a non-empty string")
        if candidate.get("category") not in self.VALID_CATEGORIES:
            errors.append(f"category '{candidate.get('category')}' not in {self.VALID_CATEGORIES}")
        if candidate.get("risk_level") not in self.VALID_RISK_LEVELS:
            errors.append(f"risk_level '{candidate.get('risk_level')}' not in {self.VALID_RISK_LEVELS}")
        if not isinstance(candidate.get("execution_plan"), dict):
            errors.append("execution_plan must be a dict")
        return {
            "passed": len(errors) == 0,
            "details": "; ".join(errors) if errors else "OK",
            "layer": self.name,
        }


class CompileGate(GateLayer):
    """Layer 1: Verify modified files are valid (py_compile for Python, basic checks for Markdown)."""

    def __init__(self):
        super().__init__("compile", required=True)

    def validate(self, candidate, execution=None):
        if not execution or not execution.get("result", {}).get("modified"):
            return {"passed": True, "details": "No file modified", "layer": self.name}
        target = execution.get("result", {}).get("rollback_pointer", {}).get("target_path", "")
        if target.endswith(".py"):
            import py_compile

            try:
                py_compile.compile(target, doraise=True)
                return {"passed": True, "details": "Python compile OK", "layer": self.name}
            except py_compile.PyCompileError as e:
                return {"passed": False, "details": str(e), "layer": self.name}
        # For markdown and other files, basic validation
        return {"passed": True, "details": "Non-Python file, skip compile check", "layer": self.name}


class LintGate(GateLayer):
    """Layer 2: Basic lint checks on modified content."""

    def __init__(self):
        super().__init__("lint", required=False)  # Advisory, not blocking

    def validate(self, candidate, execution=None):
        diff = execution.get("result", {}).get("diff", "") if execution else ""
        warnings = []
        # Check for common issues in the diff
        for line in diff.split("\n"):
            if line.startswith("+") and not line.startswith("+++"):
                if len(line) > 120:
                    warnings.append(f"Line too long ({len(line)} chars)")
                if "\t" in line and "    " in line:
                    warnings.append("Mixed tabs and spaces")
        passed = len(warnings) == 0
        return {
            "passed": passed,
            "details": "; ".join(warnings) if warnings else "OK",
            "layer": self.name,
        }


class RegressionGate(GateLayer):
    """Layer 3: Check that the change doesn't regress any dimension."""

    def __init__(self):
        super().__init__("regression", required=True)

    def validate(self, candidate, execution=None):
        # Check evaluator evidence
        evidence = candidate.get("evaluator_evidence", {})
        if evidence and evidence.get("enabled"):
            verdict = evidence.get("verdict", "")
            if verdict == "reject":
                return {"passed": False, "details": "Evaluator verdict: reject", "layer": self.name}
            overall = evidence.get("overall_score_10", 10.0)
            if overall < 4.0:
                return {"passed": False, "details": f"Evaluator overall score too low: {overall}/10", "layer": self.name}

        # Check execution result for regression signals
        if execution:
            exec_result = execution.get("result", {})
            if exec_result.get("status") == "error":
                return {"passed": False, "details": f"Execution error: {exec_result.get('reason', 'unknown')}", "layer": self.name}

        return {"passed": True, "details": "No regression detected", "layer": self.name}


class ReviewGate(GateLayer):
    """Layer 4: Check discriminator panel consensus + LLM judge verdict."""

    def __init__(self):
        super().__init__("review", required=True)

    def validate(self, candidate, execution=None):
        recommendation = candidate.get("recommendation", "hold")
        panel = candidate.get("panel", candidate.get("panel_result", {}))
        label = panel.get("cognitive_label", "")
        llm_verdict = candidate.get("llm_verdict", {})
        llm_decision = llm_verdict.get("decision", "")
        llm_confidence = llm_verdict.get("confidence", 0.0)

        # LLM judge hard reject
        if llm_decision == "reject":
            return {"passed": False, "details": "LLM judge: reject", "layer": self.name}

        # DISPUTED + LLM reject = definite fail (redundant guard)
        if label == "DISPUTED" and llm_decision == "reject":
            return {"passed": False, "details": "Panel disputed + LLM judge reject", "layer": self.name}

        if recommendation == "reject":
            return {"passed": False, "details": "Recommendation: reject", "layer": self.name}
        if label == "DISPUTED":
            return {"passed": False, "details": "Panel disputed — needs human review", "layer": self.name}

        # LLM conditional with high confidence: warn but pass
        if llm_decision == "conditional" and llm_confidence > 0.7:
            if recommendation == "accept_for_execution":
                return {
                    "passed": True,
                    "details": f"Accepted [{label}] (LLM conditional, confidence={llm_confidence:.2f})",
                    "layer": self.name,
                }

        if recommendation == "accept_for_execution":
            return {"passed": True, "details": f"Accepted [{label}]", "layer": self.name}
        return {"passed": False, "details": f"Recommendation: {recommendation}", "layer": self.name}


class HumanReviewGate(GateLayer):
    """Layer 5 (optional): Request human review for non-trivial changes.

    When a candidate reaches this layer, it means all mechanical gates passed
    but the change is too risky for auto-keep. This layer:
    1. Creates a review request (writes a review receipt JSON)
    2. Returns pending=True so the orchestrator knows to wait
    3. Can be checked later for completion
    """

    def __init__(self):
        super().__init__("human_review", required=False)  # Advisory — doesn't block, but signals

    def validate(self, candidate, execution=None):
        # Check if this candidate needs human review
        category = candidate.get("category", "")
        risk = candidate.get("risk_level", "low")
        recommendation = candidate.get("recommendation", "")

        needs_review = (
            risk in ("medium", "high")
            or category in ("prompt", "workflow", "tests", "code")
            or recommendation == "hold"
        )

        if not needs_review:
            return {"passed": True, "details": "Auto-keep eligible", "layer": self.name, "needs_human": False}

        # Create a review request
        review_request = self._create_review_request(candidate, execution)
        return {
            "passed": True,  # Don't block — but signal that human review is needed
            "details": f"Human review requested: {review_request['request_id']}",
            "layer": self.name,
            "needs_human": True,
            "review_request": review_request,
        }

    def _create_review_request(self, candidate, execution=None):
        """Create a structured review request."""
        return {
            "request_id": f"review-{candidate.get('id', 'unknown')}",
            "candidate_id": candidate.get("id", ""),
            "category": candidate.get("category", ""),
            "risk_level": candidate.get("risk_level", ""),
            "title": candidate.get("title", ""),
            "description": candidate.get("proposed_change_summary", ""),
            "diff": execution.get("result", {}).get("diff", "") if execution else "",
            "status": "pending",
            "requested_at": None,  # Will be set by caller with utc_now_iso()
        }

    def check_review_status(self, review_request_path: Path) -> dict:
        """Check if a pending review has been completed."""
        if not review_request_path.exists():
            return {"completed": False, "reason": "request_not_found"}

        data = read_json(review_request_path)
        if data.get("status") == "completed":
            return {
                "completed": True,
                "decision": data.get("decision", "pending"),
                "reviewer": data.get("reviewer", "unknown"),
                "comments": data.get("comments", ""),
            }
        return {"completed": False, "reason": "still_pending"}


DEFAULT_GATE_LAYERS = [SchemaGate(), CompileGate(), LintGate(), RegressionGate(), ReviewGate(), HumanReviewGate()]


def run_gate_layers(
    candidate: dict,
    execution: dict | None,
    layers: list[GateLayer] | None = None,
) -> dict:
    """Run all gate layers sequentially. Stop on first required failure."""
    if layers is None:
        layers = DEFAULT_GATE_LAYERS
    results = []
    all_passed = True
    failed_at = None

    for layer in layers:
        result = layer.validate(candidate, execution)
        results.append(result)
        if not result["passed"]:
            if layer.required:
                all_passed = False
                failed_at = layer.name
                break
            # Non-required layers just warn

    return {
        "all_passed": all_passed,
        "failed_at": failed_at,
        "layer_results": results,
        "layers_run": len(results),
        "layers_total": len(layers),
    }


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Apply gate decision for generic-skill lane")
    parser.add_argument("--ranking", required=True, help="Ranking artifact JSON")
    parser.add_argument("--execution", required=True, help="Execution artifact JSON")
    parser.add_argument("--state-root", default=str(DEFAULT_STATE_ROOT))
    parser.add_argument("--output", default=None)
    parser.add_argument(
        "--layers",
        default=None,
        help="Comma-separated list of layer names to run (default: all). "
        "Available: schema,compile,lint,regression,review,human_review",
    )
    return parser.parse_args()


def select_layers(layer_names: str | None) -> list[GateLayer]:
    """Filter DEFAULT_GATE_LAYERS to only those whose name is in the comma-separated list."""
    if layer_names is None:
        return list(DEFAULT_GATE_LAYERS)
    requested = [n.strip() for n in layer_names.split(",") if n.strip()]
    available = {layer.name: layer for layer in DEFAULT_GATE_LAYERS}
    selected = []
    for name in requested:
        if name not in available:
            raise ValueError(f"Unknown gate layer: {name!r}. Available: {sorted(available)}")
        selected.append(available[name])
    return selected


def load_candidate(ranking_artifact: dict, candidate_id: str) -> dict:
    for candidate in ranking_artifact.get("scored_candidates", []):
        if candidate["id"] == candidate_id:
            return candidate
    raise KeyError(f"candidate not found in ranking artifact: {candidate_id}")


def maybe_restore(execution_artifact: dict) -> dict:
    result = execution_artifact.get("result", {})
    rollback = result.get("rollback_pointer")
    if not rollback or not result.get("modified"):
        return {"attempted": False, "success": False, "reason": "no modified file to restore"}
    backup_path = Path(rollback["backup_path"]).expanduser().resolve()
    target_path = Path(rollback["target_path"]).expanduser().resolve()
    restore_backup(backup_path, target_path)
    return {
        "attempted": True,
        "success": True,
        "backup_path": str(backup_path),
        "target_path": str(target_path),
    }


def _layer_failure_detail(verdict: dict) -> str:
    """Extract the failure detail string from the first failed required layer."""
    for r in verdict["layer_results"]:
        if not r["passed"]:
            return r["details"]
    return "unknown"


def _extract_human_review_signal(verdict: dict) -> dict | None:
    """If any layer returned needs_human=True, return the review request."""
    for r in verdict["layer_results"]:
        if r.get("needs_human"):
            return r.get("review_request")
    return None


def main() -> int:
    args = parse_args()
    state_root = Path(args.state_root).expanduser().resolve()
    ensure_tree(state_root)

    ranking_artifact = read_json(Path(args.ranking).expanduser().resolve())
    execution_artifact = read_json(Path(args.execution).expanduser().resolve())
    run_id = ranking_artifact["run_id"]
    candidate_id = execution_artifact["candidate_id"]
    candidate = load_candidate(ranking_artifact, candidate_id)
    execution_result = execution_artifact.get("result", {})

    # --- 6-layer mechanical validation (runs BEFORE decision logic) ---
    selected_layers = select_layers(args.layers)
    layer_verdict = run_gate_layers(candidate, execution_artifact, layers=selected_layers)

    recommendation = candidate.get("recommendation")

    # If any required layer fails, override to revert/reject immediately
    if not layer_verdict["all_passed"]:
        rollback_outcome = {"attempted": False, "success": False, "reason": "not_needed"}
        if execution_result.get("modified"):
            decision = "revert"
            rollback_outcome = maybe_restore(execution_artifact)
        else:
            decision = "reject"
        reason = f"gate layer '{layer_verdict['failed_at']}' failed: {_layer_failure_detail(layer_verdict)}"
    else:
        # --- Original 4-way decision logic (only reached if all required layers pass) ---
        category = candidate.get("category")
        target_path = candidate.get("target_path")
        keep_eligible = (
            recommendation == "accept_for_execution"
            and category in KEEP_CATEGORIES
            and candidate.get("risk_level") == "low"
            and not protected_target(target_path)
            and execution_result.get("status") in {"success", "no_change"}
        )

        rollback_outcome = {"attempted": False, "success": False, "reason": "not_needed"}
        reason = ""
        if keep_eligible:
            decision = "keep"
            reason = "low-risk docs/reference/guardrail candidate executed successfully"
        elif recommendation == "reject":
            decision = "revert" if execution_result.get("modified") else "reject"
            rollback_outcome = maybe_restore(execution_artifact)
            reason = "critic rejected candidate under conservative gate"
        elif recommendation == "hold":
            decision = "pending_promote"
            rollback_outcome = maybe_restore(execution_artifact)
            reason = "critic marked candidate as hold; persisted to pending_promote for later human/control-plane review"
        elif execution_result.get("status") == "unsupported":
            decision = "reject"
            reason = execution_result.get("reason", "executor returned unsupported")
        elif execution_result.get("status") not in {"success", "no_change"}:
            decision = "revert" if execution_result.get("modified") else "reject"
            rollback_outcome = maybe_restore(execution_artifact)
            reason = f"execution status `{execution_result.get('status')}` did not pass gate"
        else:
            decision = "pending_promote"
            rollback_outcome = maybe_restore(execution_artifact)
            reason = "candidate is not auto-keep eligible in first runnable version; escalated to pending_promote"

    # --- Human review escalation ---
    # If mechanical gates passed but human review layer flagged needs_human,
    # escalate keep → pending_promote so a human can approve.
    human_review_request = _extract_human_review_signal(layer_verdict) if layer_verdict["all_passed"] else None
    if human_review_request and decision == "keep":
        decision = "pending_promote"
        human_review_request["requested_at"] = utc_now_iso()
        reason = f"human review requested ({human_review_request['request_id']}); escalated from keep to pending_promote"
        rollback_outcome = maybe_restore(execution_artifact)

    # Attach category/target_path for receipt even when layers failed early
    category = candidate.get("category")
    target_path = candidate.get("target_path")

    # Collect blockers from all failed layers
    blockers = [
        {"layer": r["layer"], "details": r["details"]}
        for r in layer_verdict.get("layer_results", [])
        if not r["passed"]
    ]

    output_path = Path(args.output).expanduser().resolve() if args.output else make_receipt_path(state_root, "gate", run_id, candidate_id)
    receipt = {
        "schema_version": SCHEMA_VERSION,
        "lane": ranking_artifact.get("lane", "generic-skill"),
        "run_id": run_id,
        "stage": "gated",
        "status": "success",
        "created_at": utc_now_iso(),
        "decision": decision,
        "reason": reason,
        "candidate_id": candidate_id,
        "candidate_category": category,
        "source_ranking_artifact": args.ranking,
        "source_execution_artifact": args.execution,
        "rollback": rollback_outcome,
        "blockers": blockers,
        "gate_layers": layer_verdict,
        "next_step": "propose_candidates" if decision == "keep" else "human_promote_review" if decision == "pending_promote" else "re-propose_or_manual_override",
        "next_owner": "proposer" if decision == "keep" else "human" if decision == "pending_promote" else "proposer",
        "truth_anchor": str(output_path),
    }
    if human_review_request:
        receipt["human_review_request"] = human_review_request
    write_json(output_path, receipt)

    if decision == "pending_promote":
        pending_entry = {
            "run_id": run_id,
            "candidate_id": candidate_id,
            "category": category,
            "target_path": target_path,
            "recommendation": recommendation,
            "receipt_path": str(output_path),
            "created_at": utc_now_iso(),
        }
        if human_review_request:
            pending_entry["human_review_request"] = human_review_request
            # Also write the review request to its own file for the review CLI
            review_dir = state_root / "state" / "reviews"
            review_dir.mkdir(parents=True, exist_ok=True)
            review_path = review_dir / f"{human_review_request['request_id']}.json"
            write_json(review_path, human_review_request)
        append_pending_promote(state_root, pending_entry)
    elif decision in {"reject", "revert"}:
        append_veto(
            state_root,
            {
                "run_id": run_id,
                "candidate_id": candidate_id,
                "category": category,
                "target_path": target_path,
                "decision": decision,
                "reason": reason,
                "receipt_path": str(output_path),
                "created_at": utc_now_iso(),
            },
        )

    update_state(
        state_root,
        run_id=run_id,
        stage={
            "keep": "gated_keep",
            "pending_promote": "gated_pending",
            "revert": "gated_revert",
            "reject": "gated_reject",
        }[decision],
        status=decision,
        target_path=ranking_artifact["target"]["path"],
        truth_anchor=str(output_path),
        extra={
            "candidate_id": candidate_id,
            "decision": decision,
            "receipt_path": str(output_path),
        },
    )
    print(str(output_path))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""CLI for completing pending human reviews.

Usage:
  python3 review.py --state-root /path/to/state --list
  python3 review.py --state-root /path/to/state --complete REQ_ID --decision approve
  python3 review.py --state-root /path/to/state --complete REQ_ID --decision reject --reason "..."
"""

from __future__ import annotations

import argparse
import sys
from pathlib import Path

_REPO_ROOT = Path(__file__).resolve().parents[3]
if str(_REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(_REPO_ROOT))

from lib.common import read_json, utc_now_iso, write_json
from lib.state_machine import DEFAULT_STATE_ROOT, ensure_tree


def _reviews_dir(state_root: Path) -> Path:
    return state_root / "state" / "reviews"


def list_pending(state_root: Path) -> list[dict]:
    """Return all pending review requests from the state directory."""
    reviews_dir = _reviews_dir(state_root)
    if not reviews_dir.exists():
        return []
    pending = []
    for path in sorted(reviews_dir.glob("review-*.json")):
        data = read_json(path)
        if data.get("status") == "pending":
            data["_path"] = str(path)
            pending.append(data)
    return pending


def complete_review(
    state_root: Path,
    request_id: str,
    decision: str,
    reviewer: str = "cli-user",
    reason: str = "",
) -> dict:
    """Mark a pending review as completed.

    Args:
        state_root: Root of the state directory.
        request_id: The review request ID (e.g. review-cand-001).
        decision: One of 'approve' or 'reject'.
        reviewer: Who completed the review.
        reason: Optional reason (especially for rejections).

    Returns:
        The updated review request dict.

    Raises:
        FileNotFoundError: If the review request file doesn't exist.
        ValueError: If the review is already completed or decision is invalid.
    """
    if decision not in ("approve", "reject"):
        raise ValueError(f"Decision must be 'approve' or 'reject', got: {decision!r}")

    reviews_dir = _reviews_dir(state_root)
    review_path = reviews_dir / f"{request_id}.json"
    if not review_path.exists():
        raise FileNotFoundError(f"Review request not found: {review_path}")

    data = read_json(review_path)
    if data.get("status") == "completed":
        raise ValueError(f"Review {request_id} is already completed (decision: {data.get('decision')})")

    data["status"] = "completed"
    data["decision"] = decision
    data["reviewer"] = reviewer
    data["comments"] = reason
    data["completed_at"] = utc_now_iso()
    write_json(review_path, data)
    return data


def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="CLI for completing pending human reviews",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=__doc__,
    )
    parser.add_argument("--state-root", default=str(DEFAULT_STATE_ROOT), help="State directory root")

    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument("--list", action="store_true", dest="list_pending", help="List pending reviews")
    group.add_argument("--complete", metavar="REQ_ID", help="Complete a pending review")

    parser.add_argument("--decision", choices=["approve", "reject"], help="Review decision (required with --complete)")
    parser.add_argument("--reason", default="", help="Reason for decision")
    parser.add_argument("--reviewer", default="cli-user", help="Reviewer name/id")
    return parser.parse_args(argv)


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv)
    state_root = Path(args.state_root).expanduser().resolve()
    ensure_tree(state_root)

    if args.list_pending:
        pending = list_pending(state_root)
        if not pending:
            print("No pending reviews.")
            return 0
        print(f"Pending reviews ({len(pending)}):\n")
        for review in pending:
            print(f"  ID:       {review.get('request_id', '?')}")
            print(f"  Category: {review.get('category', '?')}")
            print(f"  Risk:     {review.get('risk_level', '?')}")
            print(f"  Title:    {review.get('title', '(no title)')}")
            if review.get("description"):
                print(f"  Desc:     {review['description'][:80]}")
            print(f"  Since:    {review.get('requested_at', '?')}")
            print()
        return 0

    if args.complete:
        if not args.decision:
            print("ERROR: --decision is required with --complete", file=sys.stderr)
            return 1
        try:
            result = complete_review(
                state_root,
                args.complete,
                args.decision,
                reviewer=args.reviewer,
                reason=args.reason,
            )
            print(f"Review {args.complete} completed: {result['decision']}")
            return 0
        except (FileNotFoundError, ValueError) as e:
            print(f"ERROR: {e}", file=sys.stderr)
            return 1

    return 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""Tests for the 6-layer mechanical gate validation system (including human review)."""

from __future__ import annotations

import importlib.util
import os
import sys
import tempfile
from pathlib import Path

import pytest

# The directory name has a hyphen, so we can't use normal Python imports.
# Load gate.py by file path instead.
_GATE_PY = Path(__file__).resolve().parents[1] / "scripts" / "gate.py"
_spec = importlib.util.spec_from_file_location("gate", _GATE_PY)
_gate = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(_gate)

CompileGate = _gate.CompileGate
GateLayer = _gate.GateLayer
HumanReviewGate = _gate.HumanReviewGate
LintGate = _gate.LintGate
RegressionGate = _gate.RegressionGate
ReviewGate = _gate.ReviewGate
SchemaGate = _gate.SchemaGate
run_gate_layers = _gate.run_gate_layers
select_layers = _gate.select_layers
_extract_human_review_signal = _gate._extract_human_review_signal


# ---------------------------------------------------------------------------
# Fixtures — reusable candidate / execution dicts
# ---------------------------------------------------------------------------


def _full_candidate(**overrides) -> dict:
    """A candidate dict that passes SchemaGate and ReviewGate by default."""
    base = {
        "id": "cand-001",
        "category": "docs",
        "risk_level": "low",
        "execution_plan": {"steps": []},
        "recommendation": "accept_for_execution",
        "panel_result": {"cognitive_label": "CONSENSUS"},
        "target_path": "skills/example/SKILL.md",
    }
    base.update(overrides)
    return base


def _execution(modified: bool = False, diff: str = "", target_path: str = "", status: str = "success") -> dict:
    result = {"modified": modified, "diff": diff, "status": status}
    if target_path:
        result["rollback_pointer"] = {"target_path": target_path, "backup_path": "", "method": "restore_backup_file"}
    return {"candidate_id": "cand-001", "result": result}


# ===================================================================
# Layer 0 — SchemaGate
# ===================================================================


class TestSchemaGate:
    def test_all_fields_present(self):
        gate = SchemaGate()
        result = gate.validate(_full_candidate())
        assert result["passed"] is True
        assert result["layer"] == "schema"
        assert result["details"] == "OK"

    def test_missing_one_field(self):
        candidate = _full_candidate()
        del candidate["execution_plan"]
        result = SchemaGate().validate(candidate)
        assert result["passed"] is False
        assert "execution_plan" in result["details"]

    def test_missing_multiple_fields(self):
        candidate = {"recommendation": "accept_for_execution"}
        result = SchemaGate().validate(candidate)
        assert result["passed"] is False
        assert "id" in result["details"]
        assert "category" in result["details"]

    def test_extra_fields_ignored(self):
        candidate = _full_candidate(extra_field="hello")
        result = SchemaGate().validate(candidate)
        assert result["passed"] is True


# ===================================================================
# Layer 1 — CompileGate
# ===================================================================


class TestCompileGate:
    def test_no_execution(self):
        result = CompileGate().validate(_full_candidate(), execution=None)
        assert result["passed"] is True
        assert "No file modified" in result["details"]

    def test_no_modification(self):
        exe = _execution(modified=False)
        result = CompileGate().validate(_full_candidate(), exe)
        assert result["passed"] is True

    def test_non_python_file(self):
        exe = _execution(modified=True, target_path="README.md")
        result = CompileGate().validate(_full_candidate(), exe)
        assert result["passed"] is True
        assert "Non-Python" in result["details"]

    def test_valid_python_file(self):
        with tempfile.NamedTemporaryFile(suffix=".py", mode="w", delete=False) as f:
            f.write("x = 1 + 2\n")
            f.flush()
            tmp_path = f.name
        try:
            exe = _execution(modified=True, target_path=tmp_path)
            result = CompileGate().validate(_full_candidate(), exe)
            assert result["passed"] is True
            assert "compile OK" in result["details"]
        finally:
            os.unlink(tmp_path)

    def test_invalid_python_file(self):
        with tempfile.NamedTemporaryFile(suffix=".py", mode="w", delete=False) as f:
            f.write("def broken(\n")  # syntax error
            f.flush()
            tmp_path = f.name
        try:
            exe = _execution(modified=True, target_path=tmp_path)
            result = CompileGate().validate(_full_candidate(), exe)
            assert result["passed"] is False
            assert result["layer"] == "compile"
        finally:
            os.unlink(tmp_path)


# ===================================================================
# Layer 2 — LintGate
# ===================================================================


class TestLintGate:
    def test_clean_diff(self):
        exe = _execution(diff="+short line\n+another")
        result = LintGate().validate(_full_candidate(), exe)
        assert result["passed"] is True

    def test_no_execution(self):
        result = LintGate().validate(_full_candidate(), execution=None)
        assert result["passed"] is True

    def test_long_line(self):
        long_line = "+" + "x" * 130
        exe = _execution(diff=long_line)
        result = LintGate().validate(_full_candidate(), exe)
        assert result["passed"] is False
        assert "too long" in result["details"]

    def test_mixed_tabs_spaces(self):
        exe = _execution(diff="+\tsome    mixed indentation")
        result = LintGate().validate(_full_candidate(), exe)
        assert result["passed"] is False
        assert "Mixed tabs" in result["details"]

    def test_header_lines_ignored(self):
        # Lines starting with +++ should be ignored (diff header)
        long_header = "+++" + "x" * 200
        exe = _execution(diff=long_header)
        result = LintGate().validate(_full_candidate(), exe)
        assert result["passed"] is True

    def test_lint_is_advisory(self):
        gate = LintGate()
        assert gate.required is False


# ===================================================================
# Layer 3 — RegressionGate
# ===================================================================


class TestRegressionGate:
    def test_no_evidence(self):
        result = RegressionGate().validate(_full_candidate())
        assert result["passed"] is True
        assert "No regression" in result["details"]

    def test_evidence_disabled(self):
        candidate = _full_candidate(evaluator_evidence={"enabled": False})
        result = RegressionGate().validate(candidate)
        assert result["passed"] is True

    def test_evidence_accept(self):
        candidate = _full_candidate(evaluator_evidence={"enabled": True, "verdict": "accept"})
        result = RegressionGate().validate(candidate)
        assert result["passed"] is True

    def test_evidence_reject(self):
        candidate = _full_candidate(evaluator_evidence={"enabled": True, "verdict": "reject"})
        result = RegressionGate().validate(candidate)
        assert result["passed"] is False
        assert "reject" in result["details"]


# ===================================================================
# Layer 4 — ReviewGate
# ===================================================================


class TestReviewGate:
    def test_accept_consensus(self):
        candidate = _full_candidate(
            recommendation="accept_for_execution",
            panel_result={"cognitive_label": "CONSENSUS"},
        )
        result = ReviewGate().validate(candidate)
        assert result["passed"] is True
        assert "CONSENSUS" in result["details"]

    def test_reject(self):
        candidate = _full_candidate(recommendation="reject")
        result = ReviewGate().validate(candidate)
        assert result["passed"] is False

    def test_disputed(self):
        candidate = _full_candidate(
            recommendation="accept_for_execution",
            panel_result={"cognitive_label": "DISPUTED"},
        )
        result = ReviewGate().validate(candidate)
        assert result["passed"] is False
        assert "disputed" in result["details"].lower()

    def test_hold_defaults_to_fail(self):
        candidate = _full_candidate(recommendation="hold")
        result = ReviewGate().validate(candidate)
        assert result["passed"] is False
        assert "hold" in result["details"]

    def test_no_recommendation_defaults_hold(self):
        candidate = _full_candidate()
        del candidate["recommendation"]
        result = ReviewGate().validate(candidate)
        assert result["passed"] is False


# ===================================================================
# run_gate_layers()
# ===================================================================


class TestRunGateLayers:
    def test_all_pass(self):
        candidate = _full_candidate()
        verdict = run_gate_layers(candidate, _execution())
        assert verdict["all_passed"] is True
        assert verdict["failed_at"] is None
        assert verdict["layers_run"] == 6
        assert verdict["layers_total"] == 6

    def test_first_required_failure_stops(self):
        # Schema fails (missing fields) -> stops immediately
        candidate = {"recommendation": "accept_for_execution"}
        verdict = run_gate_layers(candidate, _execution())
        assert verdict["all_passed"] is False
        assert verdict["failed_at"] == "schema"
        # Only schema ran before stopping
        assert verdict["layers_run"] == 1

    def test_non_required_failure_continues(self):
        """LintGate is not required — failure should not stop the pipeline."""
        long_line = "+" + "x" * 200
        exe = _execution(diff=long_line)
        candidate = _full_candidate()
        verdict = run_gate_layers(candidate, exe)
        # Lint fails but is not required, so pipeline continues
        assert verdict["all_passed"] is True
        lint_result = [r for r in verdict["layer_results"] if r["layer"] == "lint"][0]
        assert lint_result["passed"] is False

    def test_required_failure_after_non_required(self):
        """If a non-required layer fails AND a later required layer also fails, report the required one."""
        long_line = "+" + "x" * 200
        exe = _execution(diff=long_line)
        # Candidate that will fail ReviewGate (recommendation=reject)
        candidate = _full_candidate(recommendation="reject")
        verdict = run_gate_layers(candidate, exe)
        assert verdict["all_passed"] is False
        assert verdict["failed_at"] == "review"

    def test_custom_layers_subset(self):
        candidate = _full_candidate()
        layers = [SchemaGate(), LintGate()]
        verdict = run_gate_layers(candidate, _execution(), layers=layers)
        assert verdict["layers_total"] == 2
        assert verdict["layers_run"] == 2
        assert verdict["all_passed"] is True

    def test_empty_layers_list(self):
        verdict = run_gate_layers({}, None, layers=[])
        assert verdict["all_passed"] is True
        assert verdict["layers_run"] == 0


# ===================================================================
# select_layers()
# ===================================================================


class TestSelectLayers:
    def test_none_returns_all(self):
        layers = select_layers(None)
        assert len(layers) == 6

    def test_single_layer(self):
        layers = select_layers("schema")
        assert len(layers) == 1
        assert layers[0].name == "schema"

    def test_multiple_layers(self):
        layers = select_layers("schema,lint,review")
        assert [l.name for l in layers] == ["schema", "lint", "review"]

    def test_unknown_layer_raises(self):
        with pytest.raises(ValueError, match="Unknown gate layer"):
            select_layers("schema,bogus")

    def test_whitespace_handling(self):
        layers = select_layers(" schema , lint ")
        assert [l.name for l in layers] == ["schema", "lint"]


# ===================================================================
# Integration: layer failures override decision logic
# ===================================================================


class TestIntegrationLayerOverridesDecision:
    """Test that layer failures correctly override the 4-way decision to revert/reject."""

    def test_schema_failure_with_modified_file_gives_revert(self):
        """When schema fails and file was modified, decision should be revert."""
        candidate = {"recommendation": "accept_for_execution"}  # missing required fields
        exe = _execution(modified=True, status="success")
        # Run just schema layer
        verdict = run_gate_layers(candidate, exe, layers=[SchemaGate()])
        assert not verdict["all_passed"]
        # Simulate the decision logic from main()
        if not verdict["all_passed"]:
            if exe["result"].get("modified"):
                decision = "revert"
            else:
                decision = "reject"
        assert decision == "revert"

    def test_schema_failure_without_modification_gives_reject(self):
        candidate = {"recommendation": "accept_for_execution"}
        exe = _execution(modified=False, status="success")
        verdict = run_gate_layers(candidate, exe, layers=[SchemaGate()])
        assert not verdict["all_passed"]
        if not verdict["all_passed"]:
            if exe["result"].get("modified"):
                decision = "revert"
            else:
                decision = "reject"
        assert decision == "reject"

    def test_all_layers_pass_allows_keep(self):
        """A fully valid candidate with all layers passing should reach the keep decision."""
        candidate = _full_candidate()
        exe = _execution(modified=False, status="success")
        verdict = run_gate_layers(candidate, exe)
        assert verdict["all_passed"]

    def test_regression_rejection_overrides_accept(self):
        """Even if recommendation is accept, a regression gate failure should block."""
        candidate = _full_candidate(evaluator_evidence={"enabled": True, "verdict": "reject"})
        exe = _execution(modified=True, status="success")
        verdict = run_gate_layers(candidate, exe)
        assert not verdict["all_passed"]
        assert verdict["failed_at"] == "regression"


# ===================================================================
# GateLayer base class
# ===================================================================


class TestGateLayerBase:
    def test_validate_not_implemented(self):
        layer = GateLayer("test_layer")
        with pytest.raises(NotImplementedError):
            layer.validate({})

    def test_default_required(self):
        layer = GateLayer("test")
        assert layer.required is True

    def test_optional_layer(self):
        layer = GateLayer("test", required=False)
        assert layer.required is False


# ===================================================================
# Layer 5 — HumanReviewGate
# ===================================================================


class TestHumanReviewGate:
    def test_low_risk_docs_no_review_needed(self):
        """Low risk docs category should not need human review."""
        candidate = _full_candidate(category="docs", risk_level="low")
        gate = HumanReviewGate()
        result = gate.validate(candidate)
        assert result["passed"] is True
        assert result["needs_human"] is False
        assert "Auto-keep" in result["details"]

    def test_medium_risk_prompt_review_requested(self):
        """Medium risk prompt category should trigger human review."""
        candidate = _full_candidate(category="prompt", risk_level="medium")
        gate = HumanReviewGate()
        result = gate.validate(candidate)
        assert result["passed"] is True
        assert result["needs_human"] is True
        assert "review_request" in result
        assert result["review_request"]["request_id"] == "review-cand-001"

    def test_hold_recommendation_review_requested(self):
        """Hold recommendation should trigger human review regardless of risk."""
        candidate = _full_candidate(recommendation="hold", risk_level="low", category="docs")
        gate = HumanReviewGate()
        result = gate.validate(candidate)
        assert result["passed"] is True
        assert result["needs_human"] is True

    def test_high_risk_triggers_review(self):
        """High risk should trigger review even for docs category."""
        candidate = _full_candidate(risk_level="high", category="docs")
        gate = HumanReviewGate()
        result = gate.validate(candidate)
        assert result["needs_human"] is True

    def test_code_category_triggers_review(self):
        candidate = _full_candidate(category="code", risk_level="low")
        gate = HumanReviewGate()
        result = gate.validate(candidate)
        assert result["needs_human"] is True

    def test_workflow_category_triggers_review(self):
        candidate = _full_candidate(category="workflow", risk_level="low")
        gate = HumanReviewGate()
        result = gate.validate(candidate)
        assert result["needs_human"] is True

    def test_tests_category_triggers_review(self):
        candidate = _full_candidate(category="tests", risk_level="low")
        gate = HumanReviewGate()
        result = gate.validate(candidate)
        assert result["needs_human"] is True

    def test_is_advisory_not_required(self):
        gate = HumanReviewGate()
        assert gate.required is False

    def test_create_review_request_structure(self):
        """Verify _create_review_request returns all expected fields."""
        candidate = _full_candidate(
            id="cand-99",
            category="prompt",
            risk_level="medium",
            title="Improve greeting",
            proposed_change_summary="Add morning greeting variant",
        )
        exe = _execution(diff="+ hello world")
        gate = HumanReviewGate()
        req = gate._create_review_request(candidate, exe)
        assert req["request_id"] == "review-cand-99"
        assert req["candidate_id"] == "cand-99"
        assert req["category"] == "prompt"
        assert req["risk_level"] == "medium"
        assert req["title"] == "Improve greeting"
        assert req["description"] == "Add morning greeting variant"
        assert req["diff"] == "+ hello world"
        assert req["status"] == "pending"
        assert req["requested_at"] is None  # Set by caller

    def test_create_review_request_no_execution(self):
        """Review request with no execution should have empty diff."""
        candidate = _full_candidate(id="cand-42")
        gate = HumanReviewGate()
        req = gate._create_review_request(candidate, execution=None)
        assert req["diff"] == ""

    def test_check_review_status_not_found(self):
        gate = HumanReviewGate()
        result = gate.check_review_status(Path("/nonexistent/path.json"))
        assert result["completed"] is False
        assert result["reason"] == "request_not_found"

    def test_check_review_status_completed(self):
        import json
        import tempfile

        with tempfile.NamedTemporaryFile(suffix=".json", mode="w", delete=False) as f:
            json.dump(
                {"status": "completed", "decision": "approve", "reviewer": "alice", "comments": "LGTM"},
                f,
            )
            tmp_path = f.name
        try:
            gate = HumanReviewGate()
            result = gate.check_review_status(Path(tmp_path))
            assert result["completed"] is True
            assert result["decision"] == "approve"
            assert result["reviewer"] == "alice"
            assert result["comments"] == "LGTM"
        finally:
            os.unlink(tmp_path)

    def test_check_review_status_still_pending(self):
        import json
        import tempfile

        with tempfile.NamedTemporaryFile(suffix=".json", mode="w", delete=False) as f:
            json.dump({"status": "pending"}, f)
            tmp_path = f.name
        try:
            gate = HumanReviewGate()
            result = gate.check_review_status(Path(tmp_path))
            assert result["completed"] is False
            assert result["reason"] == "still_pending"
        finally:
            os.unlink(tmp_path)


# ===================================================================
# _extract_human_review_signal()
# ===================================================================


class TestExtractHumanReviewSignal:
    def test_no_human_review_layer(self):
        verdict = {"layer_results": [{"passed": True, "layer": "schema"}]}
        assert _extract_human_review_signal(verdict) is None

    def test_human_review_present_but_not_needed(self):
        verdict = {
            "layer_results": [
                {"passed": True, "layer": "schema"},
                {"passed": True, "layer": "human_review", "needs_human": False},
            ]
        }
        assert _extract_human_review_signal(verdict) is None

    def test_human_review_needed(self):
        req = {"request_id": "review-x"}
        verdict = {
            "layer_results": [
                {"passed": True, "layer": "schema"},
                {"passed": True, "layer": "human_review", "needs_human": True, "review_request": req},
            ]
        }
        assert _extract_human_review_signal(verdict) == req


# ===================================================================
# Integration: human review gate escalates keep → pending_promote
# ===================================================================


class TestIntegrationHumanReviewEscalation:
    """Test that when HumanReviewGate signals needs_human, keep→pending_promote."""

    def test_keep_escalated_when_human_review_needed(self):
        """A medium-risk prompt candidate that would otherwise keep should be escalated."""
        # Candidate that would pass all required gates
        candidate = _full_candidate(
            category="prompt",  # triggers human review
            risk_level="medium",  # also triggers human review
        )
        exe = _execution(modified=False, status="success")
        # Run all layers
        verdict = run_gate_layers(candidate, exe)
        assert verdict["all_passed"] is True
        # The human review layer should signal needs_human
        human_req = _extract_human_review_signal(verdict)
        assert human_req is not None
        assert human_req["request_id"] == "review-cand-001"

    def test_low_risk_docs_no_escalation(self):
        """Low-risk docs should not trigger human review signal."""
        candidate = _full_candidate(category="docs", risk_level="low")
        exe = _execution(modified=False, status="success")
        verdict = run_gate_layers(candidate, exe)
        assert verdict["all_passed"] is True
        human_req = _extract_human_review_signal(verdict)
        assert human_req is None

#!/usr/bin/env python3
"""Tests for the human review CLI (review.py)."""

from __future__ import annotations

import importlib.util
import json
import sys
import tempfile
from pathlib import Path

import pytest

# Load review.py by file path (hyphenated directory name).
_REVIEW_PY = Path(__file__).resolve().parents[1] / "scripts" / "review.py"
_spec = importlib.util.spec_from_file_location("review", _REVIEW_PY)
_review = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(_review)

list_pending = _review.list_pending
complete_review = _review.complete_review
main = _review.main

# Also load lib helpers for setup
_REPO_ROOT = Path(__file__).resolve().parents[3]
if str(_REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(_REPO_ROOT))

from lib.common import write_json
from lib.state_machine import ensure_tree


@pytest.fixture
def state_root(tmp_path):
    """Create an initialized state tree in a temp directory."""
    root = tmp_path / "state"
    ensure_tree(root)
    return root


def _write_review(state_root: Path, request_id: str, **overrides) -> Path:
    """Write a review request file to the state directory."""
    reviews_dir = state_root / "state" / "reviews"
    reviews_dir.mkdir(parents=True, exist_ok=True)
    data = {
        "request_id": request_id,
        "candidate_id": "cand-001",
        "category": "prompt",
        "risk_level": "medium",
        "title": "Test improvement",
        "description": "A test change",
        "diff": "+ hello",
        "status": "pending",
        "requested_at": "2026-01-01T00:00:00Z",
    }
    data.update(overrides)
    path = reviews_dir / f"{request_id}.json"
    write_json(path, data)
    return path


# ===================================================================
# list_pending
# ===================================================================


class TestListPending:
    def test_empty_state(self, state_root):
        result = list_pending(state_root)
        assert result == []

    def test_one_pending(self, state_root):
        _write_review(state_root, "review-cand-001")
        result = list_pending(state_root)
        assert len(result) == 1
        assert result[0]["request_id"] == "review-cand-001"

    def test_multiple_pending(self, state_root):
        _write_review(state_root, "review-cand-001")
        _write_review(state_root, "review-cand-002", candidate_id="cand-002")
        result = list_pending(state_root)
        assert len(result) == 2

    def test_completed_not_listed(self, state_root):
        _write_review(state_root, "review-cand-001", status="completed")
        _write_review(state_root, "review-cand-002")
        result = list_pending(state_root)
        assert len(result) == 1
        assert result[0]["request_id"] == "review-cand-002"

    def test_non_review_files_ignored(self, state_root):
        """Files that don't match review-*.json pattern should be ignored."""
        reviews_dir = state_root / "state" / "reviews"
        reviews_dir.mkdir(parents=True, exist_ok=True)
        write_json(reviews_dir / "other.json", {"status": "pending"})
        _write_review(state_root, "review-cand-001")
        result = list_pending(state_root)
        assert len(result) == 1


# ===================================================================
# complete_review
# ===================================================================


class TestCompleteReview:
    def test_approve(self, state_root):
        _write_review(state_root, "review-cand-001")
        result = complete_review(state_root, "review-cand-001", "approve", reviewer="bob")
        assert result["status"] == "completed"
        assert result["decision"] == "approve"
        assert result["reviewer"] == "bob"
        assert "completed_at" in result

    def test_reject_with_reason(self, state_root):
        _write_review(state_root, "review-cand-001")
        result = complete_review(
            state_root, "review-cand-001", "reject", reason="Too risky"
        )
        assert result["decision"] == "reject"
        assert result["comments"] == "Too risky"

    def test_not_found(self, state_root):
        with pytest.raises(FileNotFoundError):
            complete_review(state_root, "review-nonexistent", "approve")

    def test_already_completed(self, state_root):
        _write_review(state_root, "review-cand-001", status="completed", decision="approve")
        with pytest.raises(ValueError, match="already completed"):
            complete_review(state_root, "review-cand-001", "approve")

    def test_invalid_decision(self, state_root):
        _write_review(state_root, "review-cand-001")
        with pytest.raises(ValueError, match="must be 'approve' or 'reject'"):
            complete_review(state_root, "review-cand-001", "maybe")

    def test_persists_to_disk(self, state_root):
        path = _write_review(state_root, "review-cand-001")
        complete_review(state_root, "review-cand-001", "approve")
        # Re-read from disk
        from lib.common import read_json

        data = read_json(path)
        assert data["status"] == "completed"
        assert data["decision"] == "approve"


# ===================================================================
# CLI main()
# ===================================================================


class TestMainCLI:
    def test_list_empty(self, state_root, capsys):
        rc = main(["--state-root", str(state_root), "--list"])
        assert rc == 0
        assert "No pending reviews" in capsys.readouterr().out

    def test_list_with_items(self, state_root, capsys):
        _write_review(state_root, "review-cand-001", title="Fix greeting")
        rc = main(["--state-root", str(state_root), "--list"])
        assert rc == 0
        out = capsys.readouterr().out
        assert "review-cand-001" in out
        assert "Fix greeting" in out

    def test_complete_approve(self, state_root, capsys):
        _write_review(state_root, "review-cand-001")
        rc = main([
            "--state-root", str(state_root),
            "--complete", "review-cand-001",
            "--decision", "approve",
        ])
        assert rc == 0
        assert "completed" in capsys.readouterr().out.lower()

    def test_complete_reject(self, state_root, capsys):
        _write_review(state_root, "review-cand-001")
        rc = main([
            "--state-root", str(state_root),
            "--complete", "review-cand-001",
            "--decision", "reject",
            "--reason", "Not ready",
        ])
        assert rc == 0

    def test_complete_missing_decision(self, state_root, capsys):
        _write_review(state_root, "review-cand-001")
        rc = main([
            "--state-root", str(state_root),
            "--complete", "review-cand-001",
        ])
        assert rc == 1
        assert "required" in capsys.readouterr().err.lower()

    def test_complete_not_found(self, state_root, capsys):
        rc = main([
            "--state-root", str(state_root),
            "--complete", "review-nonexistent",
            "--decision", "approve",
        ])
        assert rc == 1
        assert "not found" in capsys.readouterr().err.lower()

    def test_help(self):
        """--help should print usage and exit without error."""
        with pytest.raises(SystemExit) as exc_info:
            main(["--help"])
        assert exc_info.value.code == 0