Require skill delivery for PTT updates

Require skill delivery for PTT updates

plugins/violin_guard/schemas.py

@@ -34,6 +34,15 @@ RECORD_PTT_SCHEMA = {
             "id": {"type": "string"},
             "status": {"type": "string"},
             "note": {"type": "string"},
+            "skill": {"type": "string", "description": "Required selected Violin skill"},
+            "technique": {
+                "type": "string",
+                "description": "Required concrete technique for this task",
+            },
+            "hypothesis_id": {
+                "type": "string",
+                "description": "Required for hypothesis-driven phases",
+            },
             "title": {
                 "type": "string",
                 "description": "Required when explicitly creating a new PTT task",
@@ -130,6 +139,11 @@ REVIEW_BATCH_SCHEMA = {
                 "enum": ["[~]", "[x]", "[!]", "[-]"],
             },
             "note": {"type": "string", "description": "Truthful result/evidence review"},
+            "skill": {"type": "string", "description": "Required selected review skill"},
+            "hypothesis_id": {
+                "type": "string",
+                "description": "Required for hypothesis-driven phases",
+            },
             "finding": {
                 "type": "object",
                 "description": "Optional structured finding derived only from this batch",

plugins/violin_guard/service.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import hashlib
 import json
 import os
 import re
@@ -21,6 +22,7 @@ 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 .targets import resolve_target
 
 # ---------------------------------------------------------------------------
@@ -93,13 +95,72 @@ def handle_record_ptt(a, **kwargs):
     task = a.get("id")
     note = (a.get("note") or "").strip()
     status = a.get("status", "[~]")
+    skill = str(a.get("skill") or "").strip()
+    technique = str(a.get("technique") or "").strip()
 
     if not task or not note:
         raise ValueError("task id and non-empty lifecycle note required")
+    if not skill or not technique:
+        raise ValueError("skill and technique are required before a PTT update")
     if pending:
         raise ValueError(
             "a target batch is pending; use violin_review_batch instead of violin_record_ptt"
         )
+    selected = next((item for item in doc if item.id == task), None)
+    selected_phase = selected.phase if selected else str(a.get("phase") or "RECON")
+    try:
+        phase = ptt.normalize_phase(selected_phase)
+    except ValueError as exc:
+        raise ValueError(f"PTT task {task!r} must sit below a valid Phase heading") from exc
+    hypothesis_id = str(a.get("hypothesis_id") or "").strip()
+    if requires_hypothesis(phase) and not hypothesis_id:
+        raise ValueError(f"hypothesis_id is required for {phase.value} PTT work")
+    vulnerability_class = ""
+    if hypothesis_id:
+        normalized = hypothesis_id.removeprefix("H-").lstrip("0") or "0"
+        matched = next(
+            (
+                h
+                for h in hypotheses.parse_hypotheses(_eng_path(eng_dir) / "hypotheses.md")
+                if h.id.lstrip("0") == normalized
+            ),
+            None,
+        )
+        vulnerability_class = matched.vuln_class if matched else ""
+    digest = "sha256:" + hashlib.sha256(f"policy:{skill}".encode()).hexdigest()
+    reservation = prepare_delivery(
+        eng_dir,
+        session_id=state.resolve_session_id(eng_dir) or "ptt",
+        skill=skill,
+        bundle_digest=digest,
+        phase=phase.value,
+        vulnerability_class=vulnerability_class or None,
+    )
+    if reservation.owner:
+        viewed = HermesSkillViewAdapter().view(skill, task_id=task)
+        completed = complete_delivery(eng_dir, reservation, viewed)
+        return _json(
+            "skill_prepared" if completed.status == "delivered" else "skill_unavailable",
+            transition_applied=False,
+            skill={
+                "name": skill,
+                "digest": digest,
+                "content": viewed.content,
+                "error": viewed.error,
+            },
+        )
+    if reservation.status == "preparing":
+        return _json(
+            "skill_preparing", transition_applied=False, skill={"name": skill, "digest": digest}
+        )
+    binding = bind_task(
+        eng_dir,
+        task_id=task,
+        delivery_id=reservation.id,
+        hypothesis_id=hypothesis_id,
+        technique=technique,
+    )
+    note = _with_skill_token(note, skill, digest)
     if not any(item.id == task for item in doc):
         created = ptt.create_task(
             _eng_path(eng_dir) / "state" / "ptt.md",
@@ -112,6 +173,9 @@ def handle_record_ptt(a, **kwargs):
         if status == "[ ]":
             return _json("ok", task_id=created.id, task_created=True)
     existing = next((item for item in doc if item.id == task), None)
+    if existing and existing.status == "[~]" and status == "[~]":
+        ptt.update_task(_eng_path(eng_dir) / "state" / "ptt.md", task, "[~]", note)
+        return _json("ok", task_id=task, task_refreshed=True, binding=binding)
     if existing and status in {"[x]", "[-]"}:
         if existing.status != "[~]":
             raise ValueError("only the active [~] task may be closed outside a batch")
@@ -120,6 +184,13 @@ def handle_record_ptt(a, **kwargs):
     return _start_ptt_task(_eng_path(eng_dir) / "state" / "ptt.md", doc, task, status, note)
 
 
+def _with_skill_token(note: str, skill: str, digest: str) -> str:
+    """Keep exactly one replaceable selection token in a PTT note."""
+    token = f"[skill:{skill}@{digest}]"
+    stripped = re.sub(r"\s*\[skill:[^\]]+\]", "", note).strip()
+    return f"{stripped} {token}".strip()
+
+
 def _start_ptt_task(ptt_path: Path, tasks, task_id: str, status: str, note: str) -> str:
     """Arm one untouched, phase-bound task before the first target command."""
 
@@ -275,6 +346,51 @@ def handle_review_batch(a, **kwargs):
                     finding_path=None,
                     message="nothing pending",
                 )
+            skill = str(a.get("skill") or "").strip()
+            if skill:
+                tasks = ptt.parse_ptt(engagement / "state" / "ptt.md")
+                task_id = str(a.get("id") or "").strip()
+                task = next((item for item in tasks if item.id == task_id), None)
+                if task is None:
+                    raise ValueError(f"batch task {task_id!r} is missing from the PTT")
+                hypothesis_id = str(a.get("hypothesis_id") or "").strip()
+                phase = ptt.normalize_phase(task.phase)
+                if requires_hypothesis(phase) and not hypothesis_id:
+                    raise ValueError(f"hypothesis_id is required for {phase.value} batch review")
+                digest = "sha256:" + hashlib.sha256(f"policy:{skill}".encode()).hexdigest()
+                reservation = prepare_delivery(
+                    engagement,
+                    session_id=state.resolve_session_id(engagement) or "review",
+                    skill=skill,
+                    bundle_digest=digest,
+                    phase=phase.value,
+                )
+                if reservation.owner:
+                    viewed = HermesSkillViewAdapter().view(skill, task_id=task_id)
+                    completed = complete_delivery(engagement, reservation, viewed)
+                    return _json(
+                        "skill_prepared"
+                        if completed.status == "delivered"
+                        else "skill_unavailable",
+                        transition_applied=False,
+                        released=False,
+                        skill={
+                            "name": skill,
+                            "digest": digest,
+                            "content": viewed.content,
+                            "error": viewed.error,
+                        },
+                    )
+                if reservation.status == "preparing":
+                    return _json("skill_preparing", transition_applied=False, released=False)
+                bind_task(
+                    engagement,
+                    task_id=task_id,
+                    delivery_id=reservation.id,
+                    hypothesis_id=hypothesis_id,
+                    technique="batch-review",
+                )
+                a = {**a, "note": _with_skill_token(str(a.get("note") or ""), skill, digest)}
             context = _validate_review_batch(a, pending)
             finding_result = None
             finding = context["finding"]

plugins/violin_guard/state.py

@@ -4,6 +4,8 @@ from __future__ import annotations
 
 import json
 import os
+import time
+import uuid
 from contextlib import contextmanager
 from datetime import UTC, datetime
 from pathlib import Path
@@ -114,9 +116,16 @@ def read_json(path: Path) -> dict[str, Any]:
 def atomic_json(path: Path, data: dict[str, Any]) -> None:
     """Write JSON atomically by replacing a temporary swap file."""
     path.parent.mkdir(parents=True, exist_ok=True)
-    tmp = path.with_suffix(path.suffix + ".tmp")
+    tmp = path.with_name(f"{path.name}.{uuid.uuid4().hex}.tmp")
     tmp.write_text(json.dumps(data, indent=2, sort_keys=True), encoding="utf-8")
-    tmp.replace(path)
+    for attempt in range(5):
+        try:
+            tmp.replace(path)
+            return
+        except PermissionError:
+            if attempt == 4:
+                raise
+            time.sleep(0.02 * (attempt + 1))
 
 
 def mutate_json(path: Path, mutation) -> Any:
@@ -357,7 +366,17 @@ def tick_message(eng_dir: str | Path) -> int:
         data["messages"] = data.get("messages", 0) + 1
         return data["messages"]
 
-    return mutate_json(path, tick)
+    # Windows can briefly deny the replace/read sequence immediately after a
+    # prior hook writes this same file.  Lifecycle hooks intentionally do not
+    # fail the model turn, so retry here rather than silently dropping a tick.
+    for attempt in range(3):
+        try:
+            return mutate_json(path, tick)
+        except OSError:
+            if attempt == 2:
+                raise
+            time.sleep(0.02 * (attempt + 1))
+    raise RuntimeError("unreachable")
 
 
 def record_ok_check(eng_dir: str | Path, command: str, phase: str) -> None:

tests/guard/integration/test_plugin_guard.py

@@ -397,7 +397,14 @@ def test_exploitation_requires_cve_and_exploit_research_attempts(tmp_path):
     assert not allowed.errors, allowed.errors
 
 
-def test_record_ptt_can_start_pristine_task(tmp_path):
+def test_record_ptt_can_start_pristine_task(tmp_path, monkeypatch):
+    from plugins.violin_guard.skill_receipts import SkillViewResult
+
+    monkeypatch.setattr(
+        TOOLS,
+        "HermesSkillViewAdapter",
+        lambda: type("Ready", (), {"view": lambda *_a, **_k: SkillViewResult(True, "skill")})(),
+    )
     skill_file = tmp_path / ".skill-loaded-ts"
     eng = _init_e2e(tmp_path, skill_file)
     ptt_path = eng / "state" / "ptt.md"
@@ -408,7 +415,27 @@ def test_record_ptt_can_start_pristine_task(tmp_path):
 
     result = json.loads(
         TOOLS.handle_record_ptt(
-            {"eng_dir": str(eng), "id": "PT-010", "status": "[~]", "note": "Start recon"}
+            {
+                "eng_dir": str(eng),
+                "id": "PT-010",
+                "status": "[~]",
+                "note": "Start recon",
+                "skill": "pentest",
+                "technique": "recon",
+            }
+        )
+    )
+    assert result["status"] == "skill_prepared", result
+    result = json.loads(
+        TOOLS.handle_record_ptt(
+            {
+                "eng_dir": str(eng),
+                "id": "PT-010",
+                "status": "[~]",
+                "note": "Start recon",
+                "skill": "pentest",
+                "technique": "recon",
+            }
         )
     )
     assert result["status"] == "ok", result

tests/guard/state/test_a1_a15_regressions.py

@@ -12,10 +12,12 @@ from plugins.violin_guard import (
     hypotheses,
     ptt,
     service,
+    state,
 )
 from plugins.violin_guard.command import CheckCommandArgs, CheckResult, check_scope_authorization
 from plugins.violin_guard.history import append_history, check_history_staleness
 from plugins.violin_guard.phases import Phase
+from plugins.violin_guard.skill_receipts import SkillViewResult
 from plugins.violin_guard.targets import check_scope_targets
 
 
@@ -56,7 +58,14 @@ def test_wildcard_scope_allows_subdomains(tmp_path: Path) -> None:
     ).errors
 
 
-def test_ptt_heading_parenthetical_and_explicit_task_create_close(tmp_path: Path) -> None:
+def test_ptt_heading_parenthetical_and_explicit_task_create_close(
+    tmp_path: Path, monkeypatch
+) -> None:
+    monkeypatch.setattr(
+        service,
+        "HermesSkillViewAdapter",
+        lambda: type("Ready", (), {"view": lambda *_a, **_k: SkillViewResult(True, "skill")})(),
+    )
     eng = _engagement(tmp_path)
     ptt_path = eng / "state" / "ptt.md"
     ptt_path.write_text(
@@ -75,15 +84,40 @@ def test_ptt_heading_parenthetical_and_explicit_task_create_close(tmp_path: Path
                 "title": "extra check",
                 "phase": "recon",
                 "note": "planned",
+                "skill": "pentest",
+                "technique": "recon",
+            }
+        )
+    )
+    assert created["status"] == "skill_prepared"
+    created = json.loads(
+        service.handle_record_ptt(
+            {
+                "eng_dir": str(eng),
+                "id": "PT-900",
+                "status": "[ ]",
+                "title": "extra check",
+                "phase": "recon",
+                "note": "planned",
+                "skill": "pentest",
+                "technique": "recon",
             }
         )
     )
     assert created["task_created"] is True
-    closed = json.loads(
-        service.handle_record_ptt(
-            {"eng_dir": str(eng), "id": "PT-010", "status": "[x]", "note": "done"}
-        )
-    )
+    # A fresh session must prepare again; do not rely on cross-task reuse here.
+    state.record_session_id(eng, "close-session")
+    close_args = {
+        "eng_dir": str(eng),
+        "id": "PT-010",
+        "status": "[x]",
+        "note": "done",
+        "skill": "pentest",
+        "technique": "recon",
+    }
+    closed = json.loads(service.handle_record_ptt(close_args))
+    assert closed["status"] == "skill_prepared"
+    closed = json.loads(service.handle_record_ptt(close_args))
     assert closed["task_closed"] is True
 
 

tests/guard/state/test_batch_integrity.py

@@ -33,7 +33,14 @@ def test_record_ptt_refuses_to_reconcile_a_pending_batch(tmp_path: Path) -> None
     )
     result = json.loads(
         service.handle_record_ptt(
-            {"eng_dir": str(eng), "id": "PT-011", "status": "[~]", "note": "review"}
+            {
+                "eng_dir": str(eng),
+                "id": "PT-011",
+                "status": "[~]",
+                "note": "review",
+                "skill": "pentest",
+                "technique": "recon",
+            }
         )
     )
     assert result["status"] == "error"