Gate execution on skill receipts

Gate execution on skill receipts

plugins/violin_guard/bootstrap.py

@@ -180,10 +180,6 @@ def init_engagement(
         scope_path = eng_dir / "scope" / "scope.yaml"
         scope_path.write_text(yaml.safe_dump(_ctf_scope(host), sort_keys=False), encoding="utf-8")
         result.add_info("wrote CTF scope")
-        if session_id:
-            marker = eng_dir / "state" / f".skill-loaded-{session_id}"
-            marker.write_text("skill-loaded: ctf bootstrap\n", encoding="utf-8")
-            result.add_info(f"marked skill loaded for session {session_id}")
 
     if result.errors or result.warnings:
         result.add_error("init-engagement produced an incomplete or non-compliant engagement")

plugins/violin_guard/command.py

@@ -16,6 +16,7 @@ from . import bootstrap, hypotheses, ptt, state
 from . import history as history_mod
 from .phases import Phase, normalize_phase, requires_hypothesis, suppresses_heartbeat
 from .results import GuardResult
+from .skill_receipts import get_binding
 from .targets import (
     check_scope_targets,
     extract_target_candidates,
@@ -33,6 +34,7 @@ __all__ = [
     "validate_scope",
     "check_scope_authorization",
     "check_skill_load",
+    "check_skill_binding",
     "check_hypothesis_freshness",
 ]
 
@@ -259,6 +261,25 @@ def check_skill_load(eng_dir: Path, session_id: str, mandatory: bool = True) ->
     return result
 
 
+def check_skill_binding(
+    eng_dir: Path, task_id: str, session_id: str, phase: Phase
+) -> SkillLoadResult:
+    """Require a delivered, current-context receipt binding for target work."""
+    result = SkillLoadResult()
+    binding = get_binding(eng_dir, task_id)
+    if not binding:
+        result.add_error(f"skill receipt binding missing for active task {task_id}")
+        return result
+    if binding.get("session_id") != session_id:
+        result.add_error("skill receipt binding belongs to a different session")
+    current = state.read_json(eng_dir / "state" / "skills.json").get("context", {})
+    if binding.get("context_generation") != current.get("generation"):
+        result.add_error("skill receipt binding is stale after context reset")
+    if not result.errors:
+        result.add_info(f"skill receipt binding verified: {binding.get('skill')}")
+    return result
+
+
 # ---------------------------------------------------------------------------
 # Hypothesis freshness gate
 # ---------------------------------------------------------------------------
@@ -405,15 +426,10 @@ def check_command(args: CheckCommandArgs) -> CheckResult:
     artifact_result = check_local_artifact_paths(args.command)
     result.infos.extend(artifact_result.infos)
 
-    # 3. Skill-load gate (mandatory)
+    # 3. Session identity (legacy markers may infer identity, but never authorize work).
     session_id = state.resolve_session_id(eng_dir, args.session_id)
     if not session_id:
-        result.add_error("session_id is required for the skill-load gate")
-    if session_id:
-        skill_result = check_skill_load(eng_dir, session_id, mandatory=True)
-        result.errors.extend(skill_result.errors)
-        result.warnings.extend(skill_result.warnings)
-        result.infos.extend(skill_result.infos)
+        result.add_error("session_id is required for the skill receipt gate")
 
     # 4. PTT active task
     ptt_path = eng_dir / "state" / "ptt.md"
@@ -430,6 +446,11 @@ def check_command(args: CheckCommandArgs) -> CheckResult:
                 "pause the current task and start one under the requested Phase heading with "
                 "violin_record_ptt"
             )
+        if active_task and session_id:
+            binding_result = check_skill_binding(eng_dir, active_task.id, session_id, phase)
+            result.errors.extend(binding_result.errors)
+            result.warnings.extend(binding_result.warnings)
+            result.infos.extend(binding_result.infos)
 
     # 5. History staleness (duplicate detection)
     pending = state.get_pending_sync(str(eng_dir)) or {}

plugins/violin_guard/schemas.py

@@ -18,7 +18,6 @@ CHECK_COMMAND_SCHEMA = {
             "command": {"type": "string"},
             "target": {"type": "string", "description": "Explicit primary target host/IP/URL"},
             "session_id": {"type": "string"},
-            "skill_loaded_file": {"type": "string"},
         },
         "required": ["eng_dir", "phase", "command", "target"],
         "additionalProperties": False,
@@ -111,7 +110,6 @@ EXEC_SCHEMA = {
             },
             "target": {"type": "string", "description": "Explicit primary target host/IP/URL"},
             "session_id": {"type": "string"},
-            "skill_loaded_file": {"type": "string"},
             "backend": {"type": "string", "enum": ["auto", "local", "docker"], "default": "auto"},
             "timeout_seconds": {"type": "integer", "minimum": 1, "maximum": 1800},
             "cwd": {"type": "string", "description": "Engagement-relative working directory"},
@@ -235,7 +233,6 @@ EXEC_BURST_SCHEMA = {
                 "type": "string",
                 "description": "session/goal label for skill-load gating",
             },
-            "skill_loaded_file": {"type": "string", "description": "skill-load marker path"},
             "label": {"type": "string", "description": "optional batch label for logging"},
             "backend": {"type": "string", "enum": ["auto", "local", "docker"], "default": "auto"},
             "timeout_seconds": {"type": "integer", "minimum": 1, "maximum": 1800},
@@ -292,7 +289,6 @@ _ADAPTER_COMMON = {
     "phase": {"type": "string"},
     "target": {"type": "string"},
     "session_id": {"type": "string"},
-    "skill_loaded_file": {"type": "string"},
     "backend": {"type": "string", "enum": ["auto", "local", "docker"], "default": "auto"},
     "timeout_seconds": {"type": "integer", "minimum": 1, "maximum": 1800},
     "cwd": {"type": "string"},
@@ -350,7 +346,6 @@ LISTENER_SCHEMA = {
             "phase": {"type": "string"},
             "target": {"type": "string", "description": "In-scope assessment target"},
             "session_id": {"type": "string"},
-            "skill_loaded_file": {"type": "string"},
             "port": {"type": "integer", "minimum": 1, "maximum": 65535},
             "bind_host": {"type": "string"},
             "keep_open": {"type": "boolean", "default": False},
@@ -408,10 +403,6 @@ STATUS_SCHEMA = {
                 "type": "string",
                 "description": "engagement dir ($ENG_DIR / $VIOLIN_ENG_ROOT env also honoured)",
             },
-            "skill_loaded_file": {
-                "type": "string",
-                "description": "explicit skill-load marker path (else $ENG_DIR/.skill-loaded)",
-            },
         },
         "required": ["eng_dir"],
         "additionalProperties": False,

plugins/violin_guard/service.py

@@ -52,7 +52,6 @@ def _check_command_internal(a) -> cmd_module.CheckResult:
             scope=a.get("scope", ""),
             target=a.get("target"),
             session_id=a.get("session_id"),
-            skill_loaded_file=a.get("skill_loaded_file"),
         )
     )
 
@@ -593,7 +592,6 @@ def handle_exec_burst(a, **kwargs):
     phase = a.get("phase", "")
     scope = a.get("scope", "")
     session_id = a.get("session_id", "")
-    skill_loaded_file = a.get("skill_loaded_file", "")
     label = a.get("label", "")
     backend = a.get("backend", "auto")
     timeout_seconds = a.get("timeout_seconds", 180)
@@ -625,7 +623,6 @@ def handle_exec_burst(a, **kwargs):
             "eng_dir": eng_dir,
             "scope": scope,
             "session_id": session_id,
-            "skill_loaded_file": skill_loaded_file,
             "target": a.get("target"),
         }
         r = _check_command_internal(cmd_args)

tests/guard/docs/test_correctness_roadmap_1_1_1.py

@@ -22,6 +22,8 @@ from pathlib import Path
 
 import pytest
 
+from tests.guard.receipt_fixture import bind_active_task
+
 ROOT = Path(__file__).resolve().parents[3]
 sys.path.insert(0, str(ROOT / "scripts"))
 
@@ -83,6 +85,7 @@ def _init_e2e(tmp_path, skill_file, allowed=("recon", "vuln-research", "exploita
         ptt.read_text(encoding="utf-8").replace("| PT-010 | [ ] |", "| PT-010 | [~] |"),
         encoding="utf-8",
     )
+    bind_active_task(eng, "ts")
     return eng
 
 
@@ -215,6 +218,7 @@ def test_post_exploitation_requires_scope_and_skill_load(tmp_path):
         .replace("| PT-042 | [ ] |", "| PT-042 | [~] |"),
         encoding="utf-8",
     )
+    bind_active_task(eng, "ts")
 
     res = command.check_command(
         command.CheckCommandArgs(

tests/guard/integration/test_plugin_guard.py

@@ -7,6 +7,8 @@ from pathlib import Path
 
 import pytest
 
+from tests.guard.receipt_fixture import bind_active_task
+
 ROOT = Path(__file__).resolve().parents[3]
 
 # Make `violin_guard` resolvable
@@ -149,6 +151,7 @@ def _init_e2e(tmp_path, skill_file):
         ptt_path.read_text(encoding="utf-8").replace("| PT-010 | [ ] |", "| PT-010 | [~] |"),
         encoding="utf-8",
     )
+    bind_active_task(eng, "ts")
     return eng
 
 
@@ -216,6 +219,7 @@ def test_recon_does_not_require_hypothesis(tmp_path):
         .replace("| PT-030 | [ ] |", "| PT-030 | [~] |"),
         encoding="utf-8",
     )
+    bind_active_task(eng, "ts")
     research2 = command.check_command(
         command.CheckCommandArgs(
             command="nmap -sV 10.10.10.10",
@@ -655,6 +659,7 @@ def test_exploitation_gets_bounded_window_then_requires_ptt_review(monkeypatch,
         .replace("| PT-042 | [ ] |", "| PT-042 | [~] |"),
         encoding="utf-8",
     )
+    bind_active_task(eng, "ts")
     # Create a real hypothesis (not in comment) for exploitation phase
     recorded = json.loads(
         TOOLS.handle_record_hypothesis(

tests/guard/receipt_fixture.py

@@ -0,0 +1,28 @@
+"""Test-only setup for receipt-authoritative target execution."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from plugins.violin_guard import ptt, state
+from plugins.violin_guard.skill_receipts import (
+    SkillViewResult,
+    bind_task,
+    complete_delivery,
+    prepare_delivery,
+)
+
+
+def bind_active_task(engagement: Path, session_id: str = "test") -> None:
+    state.record_session_id(engagement, session_id)
+    active = ptt.find_active_task(ptt.parse_ptt(engagement / "state" / "ptt.md"))
+    assert active is not None
+    digest = "sha256:" + "a" * 64
+    reserved = prepare_delivery(
+        engagement, session_id=session_id, skill="pentest", bundle_digest=digest, phase=active.phase
+    )
+    if reserved.owner:
+        reserved = complete_delivery(
+            engagement, reserved, SkillViewResult(True, content="test skill")
+        )
+    bind_task(engagement, task_id=active.id, delivery_id=reserved.id, technique="test")

tests/guard/state/test_burst_and_target.py

@@ -23,6 +23,7 @@ from plugins.violin_guard import (  # noqa: E402
 )
 from plugins.violin_guard import service as tools  # noqa: E402
 from plugins.violin_guard.targets import resolve_target  # noqa: E402
+from tests.guard.receipt_fixture import bind_active_task  # noqa: E402
 
 _SCOPE = """targets:
   ip_addresses: ["10.10.10.10"]
@@ -102,6 +103,7 @@ def eng(tmp_path):
         ptt.read_text(encoding="utf-8").replace("| PT-010 | [ ] |", "| PT-010 | [~] |"),
         encoding="utf-8",
     )
+    bind_active_task(d, "ts")
     return d
 
 
@@ -302,6 +304,7 @@ def test_plugin_exec_burst_accepts_inline_commands(monkeypatch, tmp_path):
         ptt.read_text(encoding="utf-8").replace("| PT-010 | [ ] |", "| PT-010 | [~] |"),
         encoding="utf-8",
     )
+    bind_active_task(d, "ts")
     _patch_burst(monkeypatch, str(d))
     raw = service.handle_exec_burst(
         {

tests/guard/state/test_sync_credit_window.py

@@ -6,6 +6,7 @@ import json
 from pathlib import Path
 
 from plugins.violin_guard import bootstrap, execution, ptt, service, state
+from tests.guard.receipt_fixture import bind_active_task
 
 
 def _engagement(tmp_path: Path) -> Path:
@@ -22,6 +23,7 @@ def _engagement(tmp_path: Path) -> Path:
         ptt_path.read_text(encoding="utf-8").replace("| PT-010 | [ ] |", "| PT-010 | [~] |"),
         encoding="utf-8",
     )
+    bind_active_task(eng, "test")
     return eng
 
 

tests/guard/test_terminal_policy.py

@@ -11,6 +11,7 @@ from plugins.violin_guard import (
     register,
 )
 from plugins.violin_guard import command as guard_command
+from tests.guard.receipt_fixture import bind_active_task
 
 _SCOPE = """targets:
   ip_addresses: ["10.10.10.10"]
@@ -191,6 +192,7 @@ def _engagement(tmp_path):
         ptt.read_text(encoding="utf-8").replace("| PT-010 | [ ] |", "| PT-010 | [~] |"),
         encoding="utf-8",
     )
+    bind_active_task(eng, "test")
     return eng