Enforce skill receipts in hooks and browsers

plugins/violin_guard/__init__.py

@@ -6,13 +6,22 @@ Hermes plugin registration entry point.
 from __future__ import annotations
 
 import contextlib
+import json
+from pathlib import Path
 
 from . import (
     code_execution_audit,
+    ptt,
     schemas,  # noqa: E402
     service,  # noqa: E402
     state,  # noqa: E402
 )
+from .skill_receipts import (
+    advance_context_generation,
+    binding_readiness,
+    record_binding_turn,
+    record_delivery_turn,
+)
 from .terminal_policy import block_terminal_command
 
 __all__ = ["register", "TOOLS", "REGISTERED_TOOLS", "tools"]
@@ -20,6 +29,23 @@ __all__ = ["register", "TOOLS", "REGISTERED_TOOLS", "tools"]
 TOOLS = service
 tools = service
 
+_SESSION_ENGAGEMENTS: dict[str, str] = {}
+_TARGET_TOOLS = {
+    "violin_exec",
+    "violin_exec_burst",
+    "violin_httpx",
+    "violin_nuclei",
+    "violin_ffuf",
+    "violin_listener",
+}
+_BROWSER_TARGET_TOOLS = {
+    "browser_navigate",
+    "browser_click",
+    "browser_type",
+    "browser_select",
+    "browser_scroll",
+}
+
 # Tool names registered with the Hermes plugin loader. Kept in sync with the
 # registration tuple below; the release gate compares these against
 # plugin.yaml's provides_tools.
@@ -119,6 +145,14 @@ def _pre_tool_call_hook(tool_name=None, args=None, **kwargs):
     commands must use ``violin_exec`` or ``violin_exec_burst``.
     """
     args = args if isinstance(args, dict) else {}
+    session_id = str(kwargs.get("session_id") or args.get("session_id") or "")
+    eng_dir = str(args.get("eng_dir") or "")
+    if session_id and eng_dir:
+        _SESSION_ENGAGEMENTS[session_id] = eng_dir
+    if tool_name in _TARGET_TOOLS or tool_name in _BROWSER_TARGET_TOOLS:
+        blocked = _check_turn_binding(tool_name, args, kwargs)
+        if blocked:
+            return {"action": "block", "message": blocked}
     if tool_name == "execute_code":
         _metadata, message = code_execution_audit.validate_source(args.get("code"))
         return None if message is None else {"action": "block", "message": message}
@@ -133,6 +167,26 @@ def _post_tool_call_hook(tool_name=None, args=None, result=None, duration_ms=0,
     if tool_name == "execute_code" and isinstance(args, dict):
         with contextlib.suppress(Exception):
             code_execution_audit.record_completion(args.get("code"), result, duration_ms)
+    if tool_name not in {"violin_record_ptt", "violin_review_batch"}:
+        return
+    args = args if isinstance(args, dict) else {}
+    eng_dir = str(args.get("eng_dir") or "")
+    turn_id = str(kwargs.get("turn_id") or "")
+    if not eng_dir or not turn_id:
+        return
+    try:
+        payload = json.loads(result) if isinstance(result, str) else result
+        if not isinstance(payload, dict):
+            return
+        skill = payload.get("skill") or {}
+        delivery_id = str(skill.get("delivery_id") or "")
+        if payload.get("status") == "skill_prepared" and delivery_id:
+            record_delivery_turn(eng_dir, delivery_id=delivery_id, turn_id=turn_id)
+        task_id = str(payload.get("task_id") or payload.get("binding_task_id") or "")
+        if payload.get("status") == "ok" and task_id:
+            record_binding_turn(eng_dir, task_id=task_id, turn_id=turn_id)
+    except Exception:
+        return
 
 
 # ---------------------------------------------------------------------------
@@ -152,14 +206,46 @@ def _pre_llm_call_hook(session_id=None, eng_dir=None, **kwargs):
         with contextlib.suppress(Exception):
             state.tick_message(str(eng_dir))
             state.record_session_id(str(eng_dir), session_id)
+            if session_id:
+                _SESSION_ENGAGEMENTS[str(session_id)] = str(eng_dir)
     return None
 
 
 def _on_session_reset_hook(session_id=None, eng_dir=None, **kwargs) -> None:
     """Hook: session reset (context compression, /goal set, etc.)."""
+    eng_dir = eng_dir or _SESSION_ENGAGEMENTS.get(str(session_id or ""))
     if eng_dir:
         with contextlib.suppress(Exception):
             state.tick_message(str(eng_dir))
+            advance_context_generation(str(eng_dir), str(session_id or "reset"))
+
+
+def _check_turn_binding(tool_name: str, args: dict, hook: dict) -> str | None:
+    """Stop target/browser activity until an earlier-turn receipt is bound."""
+
+    session_id = str(hook.get("session_id") or args.get("session_id") or "")
+    eng_dir = str(args.get("eng_dir") or _SESSION_ENGAGEMENTS.get(session_id) or "")
+    if not eng_dir:
+        return (
+            f"{tool_name} needs an engagement associated through violin_status or a Violin tool "
+            "before browser or target activity is allowed"
+        )
+    tasks = ptt.parse_ptt(Path(eng_dir) / "state" / "ptt.md")
+    active = ptt.find_active_task(tasks)
+    if not active:
+        return "target activity requires exactly one active [~] PTT task"
+    binding, reason = binding_readiness(eng_dir, task_id=active.id, session_id=session_id)
+    if reason:
+        return f"target activity blocked: {reason}; select and prepare a routed skill first"
+    turn_id = str(hook.get("turn_id") or "")
+    if turn_id and turn_id in {
+        str(binding.get("bound_turn_id") or ""),
+        str((binding or {}).get("delivered_turn_id") or ""),
+    }:
+        return (
+            "target activity is blocked in the skill delivery/binding turn; retry on the next turn"
+        )
+    return None
 
 
 def _on_session_finalize_hook(session_id=None, eng_dir=None, **kwargs) -> None:

plugins/violin_guard/service.py

@@ -22,7 +22,14 @@ from .adapters import (
 from .command import CheckCommandArgs
 from .history import history_contains
 from .phases import Phase, requires_hypothesis, suppresses_heartbeat
-from .skill_receipts import HermesSkillViewAdapter, bind_task, complete_delivery, prepare_delivery
+from .skill_policy import resolve_skill_route
+from .skill_receipts import (
+    HermesSkillViewAdapter,
+    bind_task,
+    binding_readiness,
+    complete_delivery,
+    prepare_delivery,
+)
 from .targets import resolve_target
 
 # ---------------------------------------------------------------------------
@@ -146,6 +153,7 @@ def handle_record_ptt(a, **kwargs):
                 "digest": digest,
                 "content": viewed.content,
                 "error": viewed.error,
+                "delivery_id": reservation.id,
             },
         )
     if reservation.status == "preparing":
@@ -378,6 +386,7 @@ def handle_review_batch(a, **kwargs):
                             "digest": digest,
                             "content": viewed.content,
                             "error": viewed.error,
+                            "delivery_id": reservation.id,
                         },
                     )
                 if reservation.status == "preparing":
@@ -418,6 +427,7 @@ def handle_review_batch(a, **kwargs):
                 released=True,
                 finding=finding_result,
                 finding_path=finding_result.get("path") if finding_result else None,
+                binding_task_id=(context["task_id"] if skill else None),
             )
     except (OSError, ValueError) as exc:
         return _json(
@@ -728,7 +738,13 @@ def handle_status(a, **kwargs):
     counts = state.read_counts(eng_dir)
     session_id = state.resolve_session_id(eng_dir)
     marker = eng_dir / "state" / f".skill-loaded-{session_id}" if session_id else None
-    skill_loaded = bool(marker and marker.is_file())
+    legacy_marker = str(marker) if marker and marker.is_file() else None
+    binding, binding_reason = (
+        binding_readiness(eng_dir, task_id=active.id, session_id=session_id)
+        if active and session_id
+        else (None, "no active task or session")
+    )
+    route = resolve_skill_route(current_phase or "RECON")
 
     blockers = [
         {
@@ -742,16 +758,16 @@ def handle_status(a, **kwargs):
         blockers.append(
             {
                 "code": "skill_session_unknown",
-                "reason": "No session id is recorded for the skill-load gate",
-                "next_action": "Load pentest, then create its marker for the current session",
+                "reason": "No session id is recorded for receipt-backed skill delivery",
+                "next_action": "Use violin_record_ptt to select and prepare a routed skill",
             }
         )
-    elif not skill_loaded:
+    elif active and binding_reason:
         blockers.append(
             {
-                "code": "skill_not_loaded",
-                "reason": f"Pentest skill marker is missing for session {session_id}",
-                "next_action": f"Load pentest, then create {marker}",
+                "code": "skill_binding_required",
+                "reason": binding_reason,
+                "next_action": "Prepare the routed skill, then repeat the PTT update next turn",
             }
         )
     blockers.extend(
@@ -822,10 +838,14 @@ def handle_status(a, **kwargs):
         command_count=counts["commands"],
         message_count=counts["messages"],
         skill={
-            "name": "pentest",
             "session_id": session_id or None,
-            "loaded": skill_loaded,
-            "marker": str(marker) if marker else None,
+            "binding": binding,
+            "binding_ready": binding_reason is None,
+            "binding_reason": binding_reason,
+            "route_candidates": list(route.allowed),
+            "legacy_marker": legacy_marker,
+            "legacy_marker_status": "obsolete" if legacy_marker else "absent",
+            "recovery": "Select a route candidate with violin_record_ptt and retry next turn",
         },
         runtime=runtime_backend.runtime_readiness(eng_dir),
     )

plugins/violin_guard/skill_receipts.py

@@ -26,7 +26,10 @@ __all__ = [
     "bind_task",
     "complete_delivery",
     "get_binding",
+    "binding_readiness",
     "get_delivery",
+    "record_binding_turn",
+    "record_delivery_turn",
     "prepare_delivery",
     "prepare_review_readiness",
 ]
@@ -302,6 +305,60 @@ def get_binding(eng_dir: str | Path, task_id: str) -> dict[str, Any] | None:
     return data["bindings"].get(task_id)
 
 
+def binding_readiness(
+    eng_dir: str | Path, *, task_id: str, session_id: str
+) -> tuple[dict[str, Any] | None, str | None]:
+    """Return the active receipt binding, or a precise fail-closed reason."""
+
+    data, recovered = _load(_path(eng_dir))
+    if recovered:
+        return None, "skill receipt state is unavailable; prepare the selected skill again"
+    context = data.get("context") or {}
+    binding = (data.get("bindings") or {}).get(task_id)
+    if not binding:
+        return None, "the active PTT task has no delivered skill binding"
+    if str(context.get("session_id") or "") != session_id:
+        return None, "the skill binding belongs to a different session"
+    if int(binding.get("context_generation", -1)) != int(context.get("generation") or 0):
+        return None, "the skill binding is stale after a context reset"
+    delivery = (data.get("deliveries") or {}).get(binding.get("delivery_id"))
+    if not delivery or delivery.get("status") != "delivered":
+        return None, "the bound skill delivery is not ready"
+    if delivery.get("bundle_digest") != binding.get("bundle_digest"):
+        return None, "the bound skill digest no longer matches its delivery"
+    return {**binding, "delivered_turn_id": delivery.get("delivered_turn_id")}, None
+
+
+def record_delivery_turn(eng_dir: str | Path, *, delivery_id: str, turn_id: str) -> None:
+    """Associate a successful ``skill_view`` result with its Hermes turn."""
+
+    if not turn_id:
+        return
+
+    def record(data: dict[str, Any]) -> None:
+        entry = data["deliveries"].get(delivery_id)
+        if entry and entry.get("status") == "delivered":
+            entry["delivered_turn_id"] = turn_id
+            entry["updated_at"] = _now()
+
+    _mutate(eng_dir, record)
+
+
+def record_binding_turn(eng_dir: str | Path, *, task_id: str, turn_id: str) -> None:
+    """Associate a binding commit with its Hermes turn."""
+
+    if not turn_id:
+        return
+
+    def record(data: dict[str, Any]) -> None:
+        binding = data["bindings"].get(task_id)
+        if binding:
+            binding["bound_turn_id"] = turn_id
+            binding["bound_at"] = _now()
+
+    _mutate(eng_dir, record)
+
+
 def prepare_review_readiness(
     eng_dir: str | Path,
     *,

scripts/violin_guard.py

@@ -134,7 +134,7 @@ def cmd_status(args: argparse.Namespace) -> int:
     if args.section == "skill":
         skill = out.get("skill", {})
         print(json.dumps(skill, indent=2))
-        return 0 if skill.get("loaded") else 1
+        return 0 if skill.get("binding_ready") else 1
     print(json.dumps(out, indent=2))
     return 0 if out["status"] == "ok" else 1
 

tests/guard/state/test_burst_and_target.py

@@ -361,17 +361,20 @@ def test_plugin_exposes_new_tools():
 
 def test_status_skill_section_reports_load_state_and_exit_code(eng):
     state.record_session_id(eng, "ts")
+    bind_active_task(eng, "ts")
     loaded = _run("status", "--eng-dir", str(eng), "--section", "skill")
     loaded_data = json.loads(loaded.stdout)
     assert loaded.returncode == 0
-    assert loaded_data["loaded"] is True
+    assert loaded_data["binding_ready"] is True
+    assert loaded_data["legacy_marker_status"] == "obsolete"
 
-    marker = Path(loaded_data["marker"])
+    marker = Path(loaded_data["legacy_marker"])
     marker.unlink()
     missing = _run("status", "--eng-dir", str(eng), "--section", "skill")
     missing_data = json.loads(missing.stdout)
-    assert missing.returncode == 1
-    assert missing_data["loaded"] is False
+    assert missing.returncode == 0
+    assert missing_data["binding_ready"] is True
+    assert missing_data["legacy_marker_status"] == "absent"
 
 
 @pytest.mark.parametrize(

tests/guard/state/test_collaboration_ux.py

@@ -9,6 +9,7 @@ from pathlib import Path
 import pytest
 
 from plugins.violin_guard import bootstrap, history, ptt, service, state
+from tests.guard.receipt_fixture import bind_active_task
 
 ROOT = Path(__file__).resolve().parents[3]
 
@@ -30,6 +31,7 @@ def _engagement(tmp_path: Path) -> Path:
     (eng / "state" / ".skill-loaded-test-session").write_text(
         "skill-loaded: pentest\n", encoding="utf-8"
     )
+    bind_active_task(eng, "test-session")
     return eng
 
 
@@ -89,7 +91,8 @@ def test_status_explains_current_phase_pending_commands_and_skill(tmp_path: Path
     assert result["current_phase"] == "RECON"
     assert result["pending_batch"]["commands"][0]["required_phase"] == "RECON"
     assert result["phase_requirements"]["EXPLOITATION"]["sync_window"] == 20
-    assert result["skill"]["loaded"] is True
+    assert result["skill"]["binding_ready"] is True
+    assert result["skill"]["legacy_marker_status"] in {"absent", "obsolete"}
 
 
 @pytest.mark.parametrize("task_status", ["[~]", "[x]", "[!]", "[-]"])

tests/guard/test_terminal_policy.py

@@ -5,7 +5,9 @@ from __future__ import annotations
 import pytest
 
 from plugins.violin_guard import (
+    _on_session_reset_hook,
     _post_tool_call_hook,
+    _pre_llm_call_hook,
     _pre_tool_call_hook,
     bootstrap,
     register,
@@ -172,14 +174,68 @@ def test_local_file_path_containing_an_ip_is_not_treated_as_a_socket() -> None:
     )
 
 
-def test_non_terminal_tools_are_not_affected() -> None:
+def test_target_tools_require_an_engagement_binding() -> None:
     result = _pre_tool_call_hook(
         tool_name="violin_exec",
         args={"command": "nmap -sV 10.10.10.10"},
         session_id="test-session",
     )
 
-    assert result is None
+    assert result["action"] == "block"
+    assert "engagement associated" in result["message"]
+
+
+def test_skill_delivery_turn_blocks_target_and_browser_until_next_turn(tmp_path) -> None:
+    eng = _engagement(tmp_path)
+    _pre_llm_call_hook(session_id="test", eng_dir=str(eng))
+    _post_tool_call_hook(
+        tool_name="violin_record_ptt",
+        args={"eng_dir": str(eng), "id": "PT-010"},
+        result='{"status":"ok","task_id":"PT-010"}',
+        turn_id="turn-bind",
+    )
+
+    blocked = _pre_tool_call_hook(
+        tool_name="violin_exec",
+        args={"eng_dir": str(eng), "session_id": "test"},
+        session_id="test",
+        turn_id="turn-bind",
+    )
+    assert blocked["action"] == "block"
+    assert "next turn" in blocked["message"]
+
+    browser_blocked = _pre_tool_call_hook(
+        tool_name="browser_navigate",
+        args={"url": "https://10.10.10.10"},
+        session_id="test",
+        turn_id="turn-bind",
+    )
+    assert browser_blocked["action"] == "block"
+
+    assert (
+        _pre_tool_call_hook(
+            tool_name="browser_navigate",
+            args={"url": "https://10.10.10.10"},
+            session_id="test",
+            turn_id="turn-next",
+        )
+        is None
+    )
+
+
+def test_session_reset_invalidates_active_skill_binding(tmp_path) -> None:
+    eng = _engagement(tmp_path)
+    _pre_llm_call_hook(session_id="test", eng_dir=str(eng))
+    _on_session_reset_hook(session_id="test")
+
+    blocked = _pre_tool_call_hook(
+        tool_name="violin_exec",
+        args={"eng_dir": str(eng), "session_id": "test"},
+        session_id="test",
+        turn_id="after-reset",
+    )
+    assert blocked["action"] == "block"
+    assert "stale after a context reset" in blocked["message"]
 
 
 def _engagement(tmp_path):