fix: wire verification hooks into completion gate (fixes #4053)

Make the dormant verification primitive enforceable:
- Wire _verification_gate() into run_autonomous(_async): blocking hook
  failures block a success outcome and feed captured stdout/stderr back as
  feedback; never-passing checks end at max_iterations, not success.
- Fix lossy _run_verification_hooks re-wrap (preserve details/error/duration).
- Add declarative FileCheckHook (exists/non_empty/contains/json_field==equals),
  no shell; CommandVerificationHook now accepts a string command.
- Feed the goal judge executable evidence; a failing blocking hook
  short-circuits to continue without spending a judge call.
- add_verification()/remove_verification() for runtime registration.
- Add 'verification' run-journal event kind and YAML `verification:` config
  parity mapping command:/file: shapes onto the two hook classes.

Opt-in only: agents without verification_hooks are unchanged.

Co-authored-by: MervinPraison <[email protected]>

examples/python/autonomy/03_verification_hooks.py

@@ -23,6 +23,7 @@ from praisonaiagents.hooks.verification import (
     VerificationResult,
     BaseVerificationHook,
     CommandVerificationHook,
+    FileCheckHook,
 )
 
 
@@ -100,7 +101,19 @@ def main():
     print(f"   Name: {cmd_hook.name}")
     print(f"   Command: {cmd_hook.command}")
     print(f"   Timeout: {cmd_hook.timeout_seconds}s")
-    
+
+    # 6. FileCheckHook: declarative, no-shell completion gate
+    print("\n6. FileCheckHook (declarative completion gate):")
+    file_hook = FileCheckHook(
+        name="changelog", path="CHANGELOG.md", non_empty=True
+    )
+    file_result = file_hook.run()
+    print(f"   {file_hook.name}: success={file_result.success} — {file_result.output}")
+    print(
+        "   Gate wiring: when verification_hooks are configured, the autonomous "
+        "loop cannot finalise with a success outcome while a blocking hook fails."
+    )
+
     print("\n" + "=" * 60)
     print("✓ Verification hooks example completed successfully!")
     print("=" * 60)

src/praisonai-agents/praisonaiagents/agent/agent.py

@@ -4068,6 +4068,18 @@ Summary:"""
                     prompt = self._goal_continuation_prompt(self._goal_state)
                     continue
 
+                # ─────────────────────────────────────────────────────────────
+                # VERIFICATION GATE (opt-in): before any success signal below
+                # can finalise the run, all blocking verification hooks must
+                # pass. On failure, feed captured stdout/stderr back and keep
+                # iterating (never-passing checks end at max_iterations).
+                # No-op when no verification hooks are configured.
+                # ─────────────────────────────────────────────────────────────
+                _gate_feedback = self._verification_gate(response_str, iterations)
+                if _gate_feedback is not None:
+                    prompt = _gate_feedback
+                    continue
+
                 # ─────────────────────────────────────────────────────────────
                 # TOOL-CALL COMPLETION: If the model used tools this turn AND
                 # produced a substantive response, the inner loop completed
@@ -4521,6 +4533,16 @@ Summary:"""
                     await asyncio.sleep(0)
                     continue
 
+                # ─────────────────────────────────────────────────────────────
+                # VERIFICATION GATE (opt-in): blocking hooks must pass before a
+                # success signal below can finalise the run. No-op when unset.
+                # ─────────────────────────────────────────────────────────────
+                _gate_feedback = self._verification_gate(response_str, iterations)
+                if _gate_feedback is not None:
+                    prompt = _gate_feedback
+                    await asyncio.sleep(0)
+                    continue
+
                 # ─────────────────────────────────────────────────────────────
                 # TOOL-CALL COMPLETION: If the model used tools this turn AND
                 # produced a substantive response, the inner loop completed
@@ -4859,28 +4881,138 @@ Summary:"""
     
     def _run_verification_hooks(self) -> List[Dict[str, Any]]:
         """Run all registered verification hooks.
-        
+
+        Preserves the full result payload (``details``/``error``/
+        ``duration_seconds`` — including ``returncode``/``stdout``/``stderr``
+        captured by :class:`CommandVerificationHook`) so callers and the goal
+        judge can gate on executable evidence, not just a boolean.
+
         Returns:
-            List of verification results
+            List of verification result dicts.
         """
-        results = []
-        if hasattr(self, '_verification_hooks') and self._verification_hooks:
-            for hook in self._verification_hooks:
-                try:
-                    result = hook.run()
-                    results.append({
-                        "hook": hook.name,
-                        "success": result.get("success", False) if isinstance(result, dict) else getattr(result, 'success', False),
-                        "output": result.get("output", "") if isinstance(result, dict) else getattr(result, 'output', ""),
-                    })
-                except Exception as e:
-                    results.append({
-                        "hook": getattr(hook, 'name', 'unknown'),
-                        "success": False,
-                        "error": str(e),
-                    })
+        results: List[Dict[str, Any]] = []
+        for hook in getattr(self, '_verification_hooks', None) or []:
+            try:
+                result = hook.run()
+                if isinstance(result, dict):
+                    record = dict(result)
+                    record.setdefault("success", False)
+                elif hasattr(result, "to_dict"):
+                    record = result.to_dict()
+                else:
+                    record = {
+                        "success": getattr(result, "success", False),
+                        "output": getattr(result, "output", ""),
+                        "details": getattr(result, "details", {}),
+                        "error": getattr(result, "error", None),
+                        "duration_seconds": getattr(result, "duration_seconds", 0.0),
+                    }
+                record["hook"] = getattr(hook, "name", "unknown")
+                record["blocking"] = getattr(hook, "blocking", True)
+                results.append(record)
+            except Exception as e:
+                results.append({
+                    "hook": getattr(hook, 'name', 'unknown'),
+                    "blocking": getattr(hook, "blocking", True),
+                    "success": False,
+                    "error": str(e),
+                })
         return results
 
+    def add_verification(self, hook: Any) -> None:
+        """Register a verification hook for the current run.
+
+        Lets an autonomous agent (or caller) add a task-specific completion
+        gate discovered mid-run, e.g. the regression test it just wrote.
+        """
+        if not hasattr(self, "_verification_hooks") or self._verification_hooks is None:
+            self._verification_hooks = []
+        self._verification_hooks.append(hook)
+
+    def remove_verification(self, name: str) -> bool:
+        """Remove a verification hook by name. Returns True if one was removed."""
+        hooks = getattr(self, "_verification_hooks", None) or []
+        remaining = [h for h in hooks if getattr(h, "name", None) != name]
+        removed = len(remaining) != len(hooks)
+        self._verification_hooks = remaining
+        return removed
+
+    def _verification_summary_tail(self, results: List[Dict[str, Any]], limit: int = 800) -> str:
+        """Build a compact human-readable summary of gate results for feedback."""
+        lines = []
+        for r in results:
+            status = "PASS" if r.get("success") else "FAIL"
+            details = r.get("details") or {}
+            rc = details.get("returncode")
+            head = f"[{status}] {r.get('hook')}"
+            if rc is not None:
+                head += f" (exit={rc})"
+            lines.append(head)
+            out = r.get("output") or r.get("error") or ""
+            if out and not r.get("success"):
+                lines.append(str(out)[-limit:])
+        return "\n".join(lines)
+
+    def _verification_gate(self, response_str: str, iterations: int) -> Optional[str]:
+        """Run verification hooks and gate completion.
+
+        Returns ``None`` when no hooks are configured or all blocking hooks
+        pass (loop may finalise with success). Otherwise returns a feedback
+        prompt containing the failing hooks' captured output so the loop can
+        continue and the agent can fix the failure.
+        """
+        if not getattr(self, "_verification_hooks", None):
+            return None
+        results = self._run_verification_hooks()
+        self._last_verification_results = results
+        self._record_verification_journal(results)
+        failed = [r for r in results if r.get("blocking", True) and not r.get("success")]
+        if not failed:
+            return None
+        summary = self._verification_summary_tail(failed)
+        return (
+            "Completion is blocked: the following verification checks failed. "
+            "Fix the underlying problem, then continue.\n" + summary
+        )
+
+    def _verification_journal_payloads(
+        self, results: List[Dict[str, Any]]
+    ) -> List[Dict[str, Any]]:
+        """Build durable ``verification``-kind payloads for the run journal."""
+        import hashlib
+
+        payloads: List[Dict[str, Any]] = []
+        for r in results:
+            details = r.get("details") or {}
+            output = r.get("output") or ""
+            payloads.append({
+                "name": r.get("hook"),
+                "success": bool(r.get("success")),
+                "returncode": details.get("returncode"),
+                "command": details.get("command"),
+                "path": details.get("path"),
+                "duration_seconds": r.get("duration_seconds"),
+                "output_hash": hashlib.sha256(str(output).encode()).hexdigest(),
+            })
+        return payloads
+
+    def _record_verification_journal(self, results: List[Dict[str, Any]]) -> None:
+        """Best-effort durable record of gate results to the run journal.
+
+        Only runs when a run journal exposing ``record(kind, data)`` is
+        attached; otherwise this is a no-op so the default (journal-off) path
+        stays zero-overhead.
+        """
+        journal = getattr(self, "_run_journal", None)
+        record = getattr(journal, "record", None)
+        if journal is None or not callable(record):
+            return
+        try:
+            for payload in self._verification_journal_payloads(results):
+                record(kind="verification", data=payload)
+        except Exception:  # pragma: no cover - journalling is best-effort
+            pass
+
     def get_available_tools(self) -> List[Any]:
         """
         Get tools available to this agent, filtered by plan_mode if enabled.

src/praisonai-agents/praisonaiagents/agent/autonomy.py

@@ -217,6 +217,13 @@ class AutonomyConfig:
                 f"Invalid autonomy level: {level!r}. "
                 f"Must be one of {sorted(VALID_AUTONOMY_LEVELS)}"
             )
+        verification_hooks = data.get("verification_hooks")
+        # Config parity: accept declarative `verification:` entries and map
+        # `command:`/`file:` shapes onto the two hook classes.
+        declarative = data.get("verification")
+        if declarative:
+            built = _build_verification_hooks(declarative)
+            verification_hooks = (verification_hooks or []) + built
         return cls(
             enabled=data.get("enabled", True),
             level=level,
@@ -227,7 +234,7 @@ class AutonomyConfig:
             observe=data.get("observe", False),
             completion_promise=data.get("completion_promise"),
             clear_context=data.get("clear_context", False),
-            verification_hooks=data.get("verification_hooks"),
+            verification_hooks=verification_hooks,
             track_changes=data.get("track_changes"),
             sandbox=data.get("sandbox"),
             snapshot_dir=data.get("snapshot_dir"),
@@ -237,6 +244,49 @@ class AutonomyConfig:
             budget_action=data.get("budget_action", "pause"),
         )
 
+
+def _build_verification_hooks(entries: List[Any]) -> List[Any]:
+    """Map declarative ``verification:`` config entries onto hook instances.
+
+    Each entry is a dict with either a ``command`` key (→
+    :class:`CommandVerificationHook`) or a ``file`` key (→
+    :class:`FileCheckHook`). Unknown shapes are skipped.
+
+    Example::
+
+        verification:
+          - {name: tests, command: "python -m pytest -q"}
+          - {name: changelog, file: CHANGELOG.md, non_empty: true}
+    """
+    from ..hooks.verification import CommandVerificationHook, FileCheckHook
+
+    hooks: List[Any] = []
+    for entry in entries or []:
+        if not isinstance(entry, dict):
+            continue
+        blocking = entry.get("blocking", True)
+        if "command" in entry:
+            hooks.append(CommandVerificationHook(
+                name=entry.get("name", "command"),
+                command=entry["command"],
+                cwd=entry.get("cwd"),
+                timeout=entry.get("timeout", 60.0),
+                blocking=blocking,
+            ))
+        elif "file" in entry:
+            hooks.append(FileCheckHook(
+                name=entry.get("name", "file"),
+                path=entry["file"],
+                exists=entry.get("exists", True),
+                non_empty=entry.get("non_empty", False),
+                contains=entry.get("contains"),
+                json_field=entry.get("json_field"),
+                equals=entry.get("equals"),
+                blocking=blocking,
+            ))
+    return hooks
+
+
 class AutonomySignal(str, Enum):
     """Signals detected from prompts for autonomy decisions.
     

src/praisonai-agents/praisonaiagents/goal/judge.py

@@ -47,7 +47,11 @@ def _tail(text: str, limit: int = 4000) -> str:
     return text[-limit:]
 
 
-def _build_goal_judge_prompt(state: "GoalState", transcript_tail: str) -> str:
+def _build_goal_judge_prompt(
+    state: "GoalState",
+    transcript_tail: str,
+    verification_block: Optional[str] = None,
+) -> str:
     """Build a strict, criteria-anchored, JSON-only judge prompt."""
     lines = [
         "You are an independent completion judge for an autonomous agent.",
@@ -72,6 +76,13 @@ def _build_goal_judge_prompt(state: "GoalState", transcript_tail: str) -> str:
             lines.append(
                 "If ANY constraint is violated, you MUST answer 'continue'."
             )
+    if verification_block:
+        lines += [
+            "",
+            "EXECUTABLE VERIFICATION RESULTS (authoritative — command exit "
+            "codes and file checks; prefer this over prose claims):",
+            verification_block,
+        ]
     lines += [
         "",
         "AGENT OUTPUT (most recent):",
@@ -165,6 +176,7 @@ def judge_goal(
     transcript_tail: str,
     *,
     judge_model: Optional[str] = None,
+    verification_block: Optional[str] = None,
 ) -> Tuple[str, str]:
     """Judge whether ``state.goal`` is met given the latest agent output.
 
@@ -180,7 +192,7 @@ def judge_goal(
     """
     from ..eval.judge import Judge
 
-    prompt = _build_goal_judge_prompt(state, transcript_tail)
+    prompt = _build_goal_judge_prompt(state, transcript_tail, verification_block)
     try:
         judge = Judge(model=judge_model or _default_judge_model(), temperature=0.0)
         litellm = judge._get_litellm()

src/praisonai-agents/praisonaiagents/goal/loop.py

@@ -128,11 +128,37 @@ class GoalLoopMixin:
         if state is None or state.status != "active":
             return None
 
+        # Executable verification: run configured hooks and gate on the result.
+        # A failing BLOCKING hook short-circuits to "continue" WITHOUT spending
+        # a judge call. Passing/non-blocking results become authoritative
+        # evidence in the judge prompt.
+        verification_block: Optional[str] = None
+        if getattr(self, "_verification_hooks", None):
+            results = self._run_verification_hooks()
+            self._last_verification_results = results
+            self._record_verification_journal(results)
+            verification_block = self._verification_summary_tail(results)
+            blocking_failed = [
+                r for r in results
+                if r.get("blocking", True) and not r.get("success")
+            ]
+            if blocking_failed:
+                state.turns_used += 1
+                state.last_verdict = "continue"
+                state.last_reason = "verification gate failed"
+                if state.turns_used >= state.max_turns:
+                    state.status = "paused"
+                    self._persist_goal_state()
+                    return "budget_paused", "verification gate failed"
+                self._persist_goal_state()
+                return "continue", "verification gate failed"
+
         state.turns_used += 1
         verdict, reason = judge_goal(
             state,
             _tail(response),
             judge_model=getattr(self, "_goal_judge_model", None),
+            verification_block=verification_block,
         )
         state.last_verdict, state.last_reason = verdict, reason
 

src/praisonai-agents/praisonaiagents/hooks/__init__.py

@@ -104,6 +104,9 @@ __all__ = [
     # Verification hooks (protocols)
     "VerificationHook",
     "VerificationResult",
+    "BaseVerificationHook",
+    "CommandVerificationHook",
+    "FileCheckHook",
     # Simplified API (beginner-friendly)
     "add_hook",
     "remove_hook",
@@ -178,6 +181,9 @@ _LAZY_GROUPS = {
     'verification': {
         'VerificationHook': ('praisonaiagents.hooks.verification', 'VerificationHook'),
         'VerificationResult': ('praisonaiagents.hooks.verification', 'VerificationResult'),
+        'BaseVerificationHook': ('praisonaiagents.hooks.verification', 'BaseVerificationHook'),
+        'CommandVerificationHook': ('praisonaiagents.hooks.verification', 'CommandVerificationHook'),
+        'FileCheckHook': ('praisonaiagents.hooks.verification', 'FileCheckHook'),
     },
     # Simplified API (beginner-friendly aliases)
     'simplified_api': {

src/praisonai-agents/praisonaiagents/hooks/verification.py

@@ -114,17 +114,26 @@ class BaseVerificationHook:
     
     name: str = "base"
     timeout_seconds: float = 60.0
-    
-    def __init__(self, name: Optional[str] = None, timeout: float = 60.0):
+    blocking: bool = True
+
+    def __init__(
+        self,
+        name: Optional[str] = None,
+        timeout: float = 60.0,
+        blocking: bool = True,
+    ):
         """Initialize the hook.
         
         Args:
             name: Override the hook name
             timeout: Timeout in seconds for execution
+            blocking: When True (default), a failure of this hook blocks the
+                autonomous loop from finalising with a success outcome.
         """
         if name:
             self.name = name
         self.timeout_seconds = timeout
+        self.blocking = blocking
     
     def run(self, context: Optional[Dict[str, Any]] = None) -> VerificationResult:
         """Run the verification hook.
@@ -177,19 +186,25 @@ class CommandVerificationHook(BaseVerificationHook):
     def __init__(
         self,
         name: str,
-        command: list,
+        command,
         cwd: Optional[str] = None,
         timeout: float = 60.0,
+        blocking: bool = True,
     ):
         """Initialize command hook.
         
         Args:
             name: Hook name
-            command: Command to run as list (e.g., ["pytest", "-v"])
+            command: Command to run as a list (e.g., ["pytest", "-v"]) or a
+                string (e.g., "python -m pytest -q"), which is shell-split.
             cwd: Working directory for command
             timeout: Timeout in seconds
+            blocking: When True (default), a failure blocks completion.
         """
-        super().__init__(name=name, timeout=timeout)
+        super().__init__(name=name, timeout=timeout, blocking=blocking)
+        if isinstance(command, str):
+            import shlex
+            command = shlex.split(command)
         self.command = command
         self.cwd = cwd
     
@@ -235,3 +250,143 @@ class CommandVerificationHook(BaseVerificationHook):
                 output=str(e),
                 error=str(e),
             )
+
+
+def _get_dotted(data: Any, path: str) -> Any:
+    """Access a dotted path (e.g. ``"a.b.c"``) in nested dicts/lists.
+
+    Returns ``None`` if any segment is missing. List indices are supported
+    via integer-looking segments.
+    """
+    value = data
+    for part in path.split("."):
+        if isinstance(value, dict):
+            value = value.get(part)
+        elif isinstance(value, (list, tuple)):
+            try:
+                value = value[int(part)]
+            except (ValueError, IndexError):
+                return None
+        else:
+            return None
+    return value
+
+
+class FileCheckHook(BaseVerificationHook):
+    """Declarative file-assertion verification hook (no shell involved).
+
+    Checks portable filesystem facts:
+      - ``exists``    : the file exists (default True).
+      - ``non_empty`` : the file has non-zero size.
+      - ``contains``  : the file text contains a substring.
+      - ``json_field``/``equals`` : the JSON value at a dotted path equals a value.
+
+    Example:
+        hook = FileCheckHook(name="changelog", path="CHANGELOG.md", non_empty=True)
+        hook = FileCheckHook(name="cfg", path="out.json",
+                             json_field="status.code", equals=0)
+    """
+
+    def __init__(
+        self,
+        name: str,
+        path: str,
+        exists: bool = True,
+        non_empty: bool = False,
+        contains: Optional[str] = None,
+        json_field: Optional[str] = None,
+        equals: Any = None,
+        blocking: bool = True,
+    ):
+        super().__init__(name=name, blocking=blocking)
+        self.path = path
+        self.exists = exists
+        self.non_empty = non_empty
+        self.contains = contains
+        self.json_field = json_field
+        self.equals = equals
+
+    def _execute(self, context: Optional[Dict[str, Any]] = None) -> VerificationResult:
+        import os
+
+        details: Dict[str, Any] = {"path": self.path}
+        file_exists = os.path.isfile(self.path)
+        details["exists"] = file_exists
+
+        if self.exists and not file_exists:
+            return VerificationResult(
+                success=False,
+                output=f"File not found: {self.path}",
+                error="not_found",
+                details=details,
+            )
+        if not self.exists:
+            success = not file_exists
+            return VerificationResult(
+                success=success,
+                output=(
+                    f"File exists but was expected absent: {self.path}"
+                    if not success else f"File absent as expected: {self.path}"
+                ),
+                details=details,
+            )
+
+        if self.non_empty:
+            size = os.path.getsize(self.path)
+            details["size"] = size
+            if size == 0:
+                return VerificationResult(
+                    success=False,
+                    output=f"File is empty: {self.path}",
+                    error="empty",
+                    details=details,
+                )
+
+        if self.contains is not None or self.json_field is not None:
+            try:
+                with open(self.path, "r", encoding="utf-8") as fh:
+                    text = fh.read()
+            except Exception as e:
+                return VerificationResult(
+                    success=False, output=str(e), error=str(e), details=details
+                )
+
+            if self.contains is not None and self.contains not in text:
+                return VerificationResult(
+                    success=False,
+                    output=f"File {self.path} does not contain {self.contains!r}",
+                    error="missing_substring",
+                    details=details,
+                )
+
+            if self.json_field is not None:
+                import json as _json
+
+                try:
+                    data = _json.loads(text)
+                except Exception as e:
+                    return VerificationResult(
+                        success=False,
+                        output=f"File {self.path} is not valid JSON: {e}",
+                        error="invalid_json",
+                        details=details,
+                    )
+                actual = _get_dotted(data, self.json_field)
+                details["json_field"] = self.json_field
+                details["actual"] = actual
+                if actual != self.equals:
+                    return VerificationResult(
+                        success=False,
+                        output=(
+                            f"{self.path}:{self.json_field} = {actual!r}, "
+                            f"expected {self.equals!r}"
+                        ),
+                        error="mismatch",
+                        details=details,
+                    )
+
+        return VerificationResult(
+            success=True,
+            output=f"File check passed: {self.path}",
+            details=details,
+        )

src/praisonai-agents/praisonaiagents/runtime/journal.py

@@ -78,6 +78,9 @@ KIND_TOOL_RESULT = "tool_result"
 KIND_APPROVAL = "approval"
 #: The loop iteration index, journalled so the cursor is restored exactly.
 KIND_ITERATION = "iteration"
+#: A verification/completion-gate check result: ``{"name", "success",
+#: "returncode", "command"/"path", "duration_seconds", "output_hash"}``.
+KIND_VERIFICATION = "verification"
 
 VALID_KINDS = frozenset(
     {
@@ -86,6 +89,7 @@ VALID_KINDS = frozenset(
         KIND_TOOL_RESULT,
         KIND_APPROVAL,
         KIND_ITERATION,
+        KIND_VERIFICATION,
     }
 )
 

src/praisonai-agents/tests/unit/hooks/test_verification_gate.py

@@ -0,0 +1,260 @@
+"""
+Unit tests for executable completion gates (issue #4053).
+
+Covers:
+- FileCheckHook matrix (exists / non_empty / json_field equals / contains).
+- CommandVerificationHook string-command parsing.
+- The verification gate blocks completion until a blocking hook passes; a
+  never-passing check ends at the budget cap (max_iterations), never success.
+- The gate releases once a hook flips to passing (temp-file flag).
+- add_verification mid-run is enforced on the same run.
+- The goal loop short-circuits to 'continue' without a judge call when a
+  blocking hook fails, and injects executable evidence into the judge prompt.
+"""
+
+import json
+import os
+
+from unittest.mock import patch
+
+from praisonaiagents.hooks.verification import (
+    CommandVerificationHook,
+    FileCheckHook,
+)
+from praisonaiagents.goal.judge import _build_goal_judge_prompt
+from praisonaiagents.goal.models import GoalState
+
+
+def _make_agent(verification_hooks=None):
+    from praisonaiagents import Agent
+    return Agent(
+        instructions="Test agent",
+        autonomy=True,
+        output="silent",
+        verification_hooks=verification_hooks,
+    )
+
+
+# =============================================================================
+# FileCheckHook matrix
+# =============================================================================
+
+class TestFileCheckHook:
+    def test_exists_pass(self, tmp_path):
+        p = tmp_path / "f.txt"
+        p.write_text("x")
+        r = FileCheckHook(name="f", path=str(p)).run()
+        assert r.success is True
+
+    def test_exists_fail(self, tmp_path):
+        r = FileCheckHook(name="f", path=str(tmp_path / "missing.txt")).run()
+        assert r.success is False
+        assert "not found" in r.output.lower()
+
+    def test_absent_pass(self, tmp_path):
+        r = FileCheckHook(name="f", path=str(tmp_path / "missing.txt"),
+                          exists=False).run()
+        assert r.success is True
+
+    def test_non_empty_fail(self, tmp_path):
+        p = tmp_path / "empty.txt"
+        p.write_text("")
+        r = FileCheckHook(name="f", path=str(p), non_empty=True).run()
+        assert r.success is False
+        assert "empty" in r.output.lower()
+
+    def test_non_empty_pass(self, tmp_path):
+        p = tmp_path / "full.txt"
+        p.write_text("data")
+        r = FileCheckHook(name="f", path=str(p), non_empty=True).run()
+        assert r.success is True
+
+    def test_contains_pass_and_fail(self, tmp_path):
+        p = tmp_path / "c.txt"
+        p.write_text("hello world")
+        assert FileCheckHook(name="c", path=str(p), contains="world").run().success
+        r = FileCheckHook(name="c", path=str(p), contains="absent").run()
+        assert r.success is False
+
+    def test_json_field_equals(self, tmp_path):
+        p = tmp_path / "d.json"
+        p.write_text(json.dumps({"status": {"code": 0}}))
+        ok = FileCheckHook(name="j", path=str(p),
+                           json_field="status.code", equals=0).run()
+        assert ok.success is True
+        bad = FileCheckHook(name="j", path=str(p),
+                            json_field="status.code", equals=1).run()
+        assert bad.success is False
+        assert "expected" in bad.output.lower()
+
+    def test_json_invalid(self, tmp_path):
+        p = tmp_path / "bad.json"
+        p.write_text("not json")
+        r = FileCheckHook(name="j", path=str(p),
+                          json_field="a", equals=1).run()
+        assert r.success is False
+        assert "json" in r.output.lower()
+
+
+class TestCommandHookParsing:
+    def test_string_command_is_split(self):
+        hook = CommandVerificationHook(name="t", command="python -c pass")
+        assert hook.command == ["python", "-c", "pass"]
+
+    def test_list_command_preserved(self):
+        hook = CommandVerificationHook(name="t", command=["echo", "hi"])
+        assert hook.command == ["echo", "hi"]
+
+
+# =============================================================================
+# _run_verification_hooks preserves the full payload
+# =============================================================================
+
+class TestRunnerFidelity:
+    def test_details_survive_rewrap(self, tmp_path):
+        p = tmp_path / "x.txt"
+        p.write_text("hi")
+        agent = _make_agent([FileCheckHook(name="f", path=str(p))])
+        results = agent._run_verification_hooks()
+        assert len(results) == 1
+        rec = results[0]
+        assert rec["hook"] == "f"
+        assert rec["success"] is True
+        assert "details" in rec
+        assert "duration_seconds" in rec
+        assert rec["blocking"] is True
+
+
+# =============================================================================
+# Gate blocks / releases completion in the autonomous loop
+# =============================================================================
+
+class TestGateBlocksCompletion:
+    def test_failing_hook_blocks_success_until_budget(self, tmp_path):
+        # A file that never exists → blocking hook never passes.
+        hook = FileCheckHook(name="need", path=str(tmp_path / "never.txt"))
+        agent = _make_agent([hook])
+
+        counter = {"n": 0}
+
+        def fake_chat(prompt):
+            counter["n"] += 1
+            return f"used a tool and did work step {counter['n']} " + "x" * 200
+
+        with patch.object(agent, "chat", side_effect=fake_chat):
+            result = agent.run_autonomous("do it", max_iterations=3)
+
+        assert result.success is False
+        assert result.completion_reason == "max_iterations"
+        # last feedback prompt carries the failing hook's name
+        assert any(
+            r.get("hook") == "need" and not r.get("success")
+            for r in getattr(agent, "_last_verification_results", [])
+        )
+
+    def test_gate_releases_when_hook_flips(self, tmp_path):
+        flag = tmp_path / "done.flag"
+        hook = FileCheckHook(name="flag", path=str(flag))
+        agent = _make_agent([hook])
+
+        counter = {"n": 0}
+
+        def fake_chat(prompt):
+            counter["n"] += 1
+            # a tool was "used" this turn so the tool_completion signal fires
+            agent._autonomy_turn_tool_count = 1
+            # create the flag on the 2nd turn so the gate releases afterwards
+            if counter["n"] == 2:
+                flag.write_text("ok")
+            return f"progress {counter['n']} " + "y" * 200
+
+        with patch.object(agent, "chat", side_effect=fake_chat), \
+             patch.object(agent, "get_recommended_stage", return_value="autonomous"):
+            result = agent.run_autonomous("do it", max_iterations=10)
+
+        assert result.success is True
+        # released only after the flag appeared (2+ iterations)
+        assert result.iterations >= 2
+
+    def test_add_verification_midrun_is_enforced(self, tmp_path):
+        agent = _make_agent()  # no hooks initially
+        missing = tmp_path / "missing.txt"
+
+        counter = {"n": 0}
+
+        def fake_chat(prompt):
+            counter["n"] += 1
+            if counter["n"] == 1:
+                # discover a completion condition mid-run
+                agent.add_verification(FileCheckHook(name="late", path=str(missing)))
+            return f"work {counter['n']} " + "z" * 200
+
+        with patch.object(agent, "chat", side_effect=fake_chat):
+            result = agent.run_autonomous("do it", max_iterations=3)
+
+        # Now blocked by the late-added hook → cannot finish successfully.
+        assert result.success is False
+        assert result.completion_reason == "max_iterations"
+
+    def test_no_hooks_unchanged(self):
+        agent = _make_agent()
+        with patch.object(agent, "chat",
+                          side_effect=lambda p: "used tool " + "q" * 200):
+            result = agent.run_autonomous("do it", max_iterations=5)
+        # Behaviour is unchanged for agents without verification hooks: the
+        # existing heuristics still terminate successfully (not blocked).
+        assert result.success is True
+        assert result.completion_reason != "max_iterations"
+
+
+# =============================================================================
+# Goal loop: executable evidence + short-circuit
+# =============================================================================
+
+class TestGoalLoopEvidence:
+    def test_failing_blocking_hook_short_circuits_without_judge(self, tmp_path):
+        hook = FileCheckHook(name="need", path=str(tmp_path / "never.txt"))
+        agent = _make_agent([hook])
+        judge_calls = {"n": 0}
+
+        def fake_judge(*a, **k):
+            judge_calls["n"] += 1
+            return ("done", "met")
+
+        counter = {"n": 0}
+
+        with patch.object(agent, "chat",
+                          side_effect=lambda p: f"step {counter.__setitem__('n', counter['n'] + 1) or counter['n']}"), \
+             patch("praisonaiagents.goal.loop.judge_goal", side_effect=fake_judge):
+            result = agent.run_goal("task", goal="g", max_turns=2)
+
+        # Blocking hook fails every turn → judge never consulted, ends paused.
+        assert judge_calls["n"] == 0
+        assert result.completion_reason == "budget_paused"
+
+    def test_passing_hook_evidence_reaches_judge(self, tmp_path):
+        p = tmp_path / "ok.txt"
+        p.write_text("data")
+        hook = FileCheckHook(name="ok", path=str(p))
+        agent = _make_agent([hook])
+        seen = {}
+
+        def fake_judge(state, tail, **k):
+            seen["block"] = k.get("verification_block")
+            return ("done", "met")
+
+        with patch.object(agent, "chat", side_effect=lambda pr: "did it"), \
+             patch("praisonaiagents.goal.loop.judge_goal", side_effect=fake_judge):
+            result = agent.run_goal("task", goal="g", max_turns=5)
+
+        assert result.completion_reason == "goal_met"
+        assert seen.get("block") is not None
+        assert "ok" in seen["block"]
+
+    def test_judge_prompt_includes_verification_block(self):
+        state = GoalState(goal="G")
+        prompt = _build_goal_judge_prompt(
+            state, "output", verification_block="[PASS] tests (exit=0)"
+        )
+        assert "[PASS] tests (exit=0)" in prompt
+        assert "EXECUTABLE VERIFICATION RESULTS" in prompt