feat: harden benchmark evaluation and guard runtime

plugins/violin_guard/bootstrap.py

@@ -14,7 +14,7 @@ from pathlib import Path
 import yaml
 
 from .results import GuardResult
-from .state import record_session_id, resolve_eng_dir
+from .state import ensure_dir, record_session_id, resolve_eng_dir
 
 __all__ = [
     "init_engagement",
@@ -167,10 +167,10 @@ def init_engagement(
     result = BootstrapResult()
     host = (host or "").strip() or _derive_host(eng_dir)
 
-    eng_dir.mkdir(parents=True, exist_ok=True)
+    ensure_dir(eng_dir)
     record_session_id(eng_dir, session_id)
     for rel in _ARTIFACT_DIRECTORIES:
-        (eng_dir / rel).mkdir(parents=True, exist_ok=True)
+        ensure_dir(eng_dir / rel)
     for rel, (template_rel, placeholder) in _REPAIR_TEMPLATES.items():
         target = eng_dir / rel
         if target.exists():
@@ -285,14 +285,14 @@ def _auto_repair_corrupt_artifacts(eng_dir: Path, result: BootstrapResult) -> Bo
 
     if not eng_dir.exists():
         try:
-            eng_dir.mkdir(parents=True, exist_ok=True)
+            ensure_dir(eng_dir)
             new_infos.append(f"AUTO-REPAIR: created missing engagement directory {eng_dir}")
         except Exception as exc:
             new_errors.append(f"AUTO-REPAIR FAILED creating {eng_dir}: {exc}")
 
     for rel in _ARTIFACT_DIRECTORIES:
         try:
-            (eng_dir / rel).mkdir(parents=True, exist_ok=True)
+            ensure_dir(eng_dir / rel)
         except OSError as exc:
             new_errors.append(f"AUTO-REPAIR FAILED creating {eng_dir / rel}: {exc}")
 

plugins/violin_guard/code_execution_audit.py

@@ -80,7 +80,7 @@ def record_completion(source: object, result: object, duration_ms: object = 0) -
     stamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ")
     phase_dir = metadata["phase"].lower().replace("_", "-")
     receipt = eng_dir / "evidence" / phase_dir / f"execute-code-{stamp}-{digest[:12]}.py"
-    receipt.parent.mkdir(parents=True, exist_ok=True)
+    state.ensure_dir(receipt.parent)
     receipt.write_text(str(source), encoding="utf-8")
 
     summary = _result_summary(result, duration_ms)

plugins/violin_guard/execution.py

@@ -397,7 +397,7 @@ def execute(
     rel_stdout = stdout_path.relative_to(engagement).as_posix()
     rel_stderr = stderr_path.relative_to(engagement).as_posix()
 
-    evidence_dir.mkdir(parents=True, exist_ok=True)
+    state.ensure_dir(evidence_dir)
 
     record: dict[str, Any] = {
         "schema_version": SCHEMA_VERSION,

plugins/violin_guard/findings.py

@@ -146,7 +146,7 @@ def _create_from_pending_batch(
     )
 
     directory = engagement / "evidence" / "findings"
-    directory.mkdir(parents=True, exist_ok=True)
+    state.ensure_dir(directory)
     batch_id = str(pending.get("batch_id") or "")
     existing = _existing_batch_finding(directory, batch_id)
     if existing:

plugins/violin_guard/history.py

@@ -10,7 +10,7 @@ from __future__ import annotations
 from datetime import UTC, datetime
 from pathlib import Path
 
-from .state import lock_file, resolve_eng_dir
+from .state import ensure_dir, lock_file, resolve_eng_dir
 
 _COMMAND_MARKER = " | command="
 _COMMAND_LENGTH_MARKER = " | command_length="
@@ -39,7 +39,7 @@ def append_history(
 ) -> None:
     """Append one execution record to history.md under an advisory lock."""
     path = _history_path(eng_dir)
-    path.parent.mkdir(parents=True, exist_ok=True)
+    ensure_dir(path.parent)
     stamp = datetime.now(UTC).isoformat().replace("+00:00", "Z")
     clean_command = normalize_command(command) if "\n" in command else command
     line = (

plugins/violin_guard/hypotheses.py

@@ -15,6 +15,7 @@ from pathlib import Path
 from typing import Any
 
 from .phases import normalize_phase
+from .state import ensure_dir
 from .targets import normalise_target
 
 __all__ = [
@@ -414,7 +415,7 @@ def update_hypothesis(
 
 def _rewrite_hypotheses(path: Path, hypotheses_list: list[Hypothesis]) -> None:
     """Rewrite the hypotheses file while preserving structural sections (Decoy Trail, Observations, etc.)."""
-    path.parent.mkdir(parents=True, exist_ok=True)
+    ensure_dir(path.parent)
     template = path.read_text(encoding="utf-8") if path.exists() else "# Hypothesis Board\n\n"
 
     # Remove template instruction HTML comment if present

plugins/violin_guard/runtime_backend.py

@@ -35,15 +35,18 @@ def _docker_container_ready(
 ) -> tuple[bool, str]:
     if shutil.which("docker") is None:
         return False, "docker executable not found"
-    result = run(
-        ["docker", "inspect", "--format", "{{.State.Running}}|{{json .Mounts}}", container],
-        capture_output=True,
-        text=True,
-        encoding="utf-8",
-        errors="replace",
-        timeout=10,
-        check=False,
-    )
+    try:
+        result = run(
+            ["docker", "inspect", "--format", "{{.State.Running}}|{{json .Mounts}}", container],
+            capture_output=True,
+            text=True,
+            encoding="utf-8",
+            errors="replace",
+            timeout=10,
+            check=False,
+        )
+    except subprocess.TimeoutExpired:
+        return False, f"Docker container {container!r} probe timed out"
     if result.returncode != 0:
         return False, f"Docker container {container!r} is unavailable"
     running, _, mounts = result.stdout.strip().partition("|")

plugins/violin_guard/skill_receipts.py

@@ -111,7 +111,7 @@ def _load(path: Path) -> tuple[dict[str, Any], bool]:
 
 def _mutate(eng_dir: str | Path, mutation: Callable[[dict[str, Any]], Any]) -> Any:
     path = _path(eng_dir)
-    path.parent.mkdir(parents=True, exist_ok=True)
+    state.ensure_dir(path.parent)
     with state.lock_file(path):
         data, recovered = _load(path)
         if recovered:

plugins/violin_guard/state.py

@@ -95,9 +95,19 @@ def record_session_id(eng_dir: str | Path, session_id: str | None) -> None:
             atomic_json(path, {"session_id": session_id.strip()})
 
 
+def ensure_dir(path: Path) -> Path:
+    """Ensure directory exists fail-safe against symlinks and cross-platform FileExistsError [Errno 17]."""
+    try:
+        path.mkdir(parents=True, exist_ok=True)
+    except FileExistsError:
+        if not path.exists():
+            raise
+    return path
+
+
 def _state_dir(eng_dir: str | Path) -> Path:
     p = resolve_eng_dir(eng_dir) / _STATE_DIR
-    p.mkdir(parents=True, exist_ok=True)
+    ensure_dir(p)
     return p
 
 
@@ -108,7 +118,7 @@ def _state_dir(eng_dir: str | Path) -> Path:
 def lock_file(path: Path):
     """Acquire an exclusive advisory lock on ``path`` for the duration of a ``with`` block."""
     lock_path = path.with_suffix(path.suffix + ".lock")
-    lock_path.parent.mkdir(parents=True, exist_ok=True)
+    ensure_dir(lock_path.parent)
     with FileLock(str(lock_path), timeout=20):
         yield
 
@@ -138,7 +148,7 @@ 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)
+    ensure_dir(path.parent)
     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")
     try:

plugins/violin_guard/targets.py

@@ -68,14 +68,40 @@ class _TargetPolicy:
     excluded_paths: set[str]
 
     def check_command_payload(self, command: str, result: TargetCheckResult) -> None:
-        for url in self.excluded_urls:
-            if url in command:
-                result.errors.append(f"strict block: command payload contains excluded URL '{url}'")
-        for path in self.excluded_paths:
-            if path in command:
-                result.errors.append(
-                    f"strict block: command payload contains excluded path '{path}'"
-                )
+        if not self.excluded_urls and not self.excluded_paths:
+            return
+        for token in _command_tokens(command):
+            candidate = token.strip("'\"(),;")
+            if not candidate:
+                continue
+
+            for url in self.excluded_urls:
+                if url and url in candidate:
+                    result.errors.append(
+                        f"strict block: command payload contains excluded URL '{url}'"
+                    )
+                    break
+
+            for path in self.excluded_paths:
+                if not path:
+                    continue
+                if candidate == path:
+                    result.errors.append(
+                        f"strict block: command payload contains excluded path '{path}'"
+                    )
+                    break
+                if "://" in candidate or candidate.startswith("/"):
+                    with contextlib.suppress(Exception):
+                        cand_url = URL(
+                            candidate if "://" in candidate else f"http://dummy.local{candidate}"
+                        )
+                        cpath = cand_url.path
+                        norm_ex = path.rstrip("/")
+                        if cpath in (path, norm_ex) or cpath.startswith(norm_ex + "/"):
+                            result.errors.append(
+                                f"strict block: command payload contains excluded path '{path}'"
+                            )
+                            break
 
     def is_excluded(self, candidate: str) -> bool:
         return _matches_host(candidate, self.excluded) or _matches_ip_set(