Skip to content

Commit 36680d5

Browse files
Trecekclaude
andcommitted
test(execution/backends): add parametrized deny-mechanism guard probe harness
Adds TestHookDenyEfficacyProbe with parametrization over the full (tool_class x session_mode x backend) matrix for every PreToolUse+deny hook script in HOOK_REGISTRY. Each probe: - Skips inert codex rows after verifying the script is excluded from generate_codex_hooks_config() output - Invokes the guard as a subprocess via sys.executable with a clean env - Classifies the outcome (deny -> hard, else -> soft) - Records to the probe matrix via record_probe_row Includes TOOL_CLASS_PAYLOADS (matching TOOL_CLASSES.keys() exactly), SESSION_MODE_ENV / BACKEND_ENV env override dicts, helper functions (_clean_env, _invoke_guard, _build_payload, _collect_probe_params), and the standalone test_probe_collection_count meta-test that guards against silent drift between the registry and the expected matrix size. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
1 parent 58d5b87 commit 36680d5

1 file changed

Lines changed: 306 additions & 0 deletions

File tree

Lines changed: 306 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,306 @@
1+
"""PreToolUse deny-mechanism guard efficacy probe harness.
2+
3+
These probes measure guard decision outcomes conditional on hook firing. Whether the
4+
backend fires PreToolUse for MCP-class tools is an open question (B4v) requiring live
5+
CLI probes (CODEX_SMOKE_TEST-gated). The strength matrix emitted by P1-A1-WP3 consumes
6+
these results.
7+
"""
8+
9+
from __future__ import annotations
10+
11+
import copy
12+
import json
13+
import os
14+
import re
15+
import subprocess
16+
import sys
17+
from pathlib import Path
18+
from typing import Any
19+
20+
import pytest
21+
22+
from autoskillit.execution.backends._codex_hooks import generate_codex_hooks_config
23+
from autoskillit.hook_registry import HOOK_REGISTRY, HOOKS_DIR
24+
from tests.execution.backends.conftest import (
25+
_SKIP_CODEX,
26+
BACKENDS,
27+
EXPECTED_TOTAL_PROBE_COUNT,
28+
SESSION_MODES,
29+
TOOL_CLASSES,
30+
record_probe_row,
31+
)
32+
33+
pytestmark = [pytest.mark.layer("execution"), pytest.mark.medium]
34+
35+
36+
SESSION_MODE_ENV: dict[str, dict[str, str]] = {
37+
"interactive": {},
38+
"headless-p": {
39+
"AUTOSKILLIT_HEADLESS": "1",
40+
"AUTOSKILLIT_SESSION_TYPE": "skill",
41+
},
42+
"subagent": {
43+
"AUTOSKILLIT_HEADLESS": "1",
44+
"AUTOSKILLIT_SESSION_TYPE": "skill",
45+
},
46+
}
47+
48+
BACKEND_ENV: dict[str, dict[str, str]] = {
49+
"claude-code": {"AUTOSKILLIT_AGENT_BACKEND": "claude-code"},
50+
"codex": {"AUTOSKILLIT_AGENT_BACKEND": "codex"},
51+
}
52+
53+
# Contract: TOOL_CLASS_PAYLOADS.keys() == TOOL_CLASSES.keys()
54+
TOOL_CLASS_PAYLOADS: dict[str, dict] = {
55+
"Bash": {
56+
"tool_name": TOOL_CLASSES["Bash"],
57+
"tool_input": {
58+
"command": "pip install -e . && pytest tests/ && gh pr create && gh run download && git commit --amend",
59+
"run_in_background": True,
60+
},
61+
},
62+
"Write": {
63+
"tool_name": TOOL_CLASSES["Write"],
64+
"tool_input": {
65+
"file_path": "/worktree/.autoskillit/phases/non-canonical-name.md",
66+
},
67+
},
68+
"Edit": {
69+
"tool_name": TOOL_CLASSES["Edit"],
70+
"tool_input": {
71+
"file_path": "/worktree/.autoskillit/phases/non-canonical-name.md",
72+
},
73+
},
74+
"Grep": {
75+
"tool_name": TOOL_CLASSES["Grep"],
76+
"tool_input": {
77+
"pattern": r"foo\|bar",
78+
"path": "/worktree/src",
79+
},
80+
},
81+
"AskUserQuestion": {
82+
"tool_name": TOOL_CLASSES["AskUserQuestion"],
83+
"tool_input": {
84+
"question": "probe",
85+
},
86+
},
87+
"mcp_run_skill": {
88+
"tool_name": TOOL_CLASSES["mcp_run_skill"],
89+
"tool_input": {
90+
"skill_command": "non-slash-command",
91+
"step_name": "probe-step",
92+
},
93+
},
94+
"mcp_run_cmd": {
95+
"tool_name": TOOL_CLASSES["mcp_run_cmd"],
96+
"tool_input": {
97+
"command": "pip install -e . && pytest tests/ && gh pr create && gh run download && git commit --amend",
98+
"run_in_background": True,
99+
},
100+
},
101+
"mcp_run_python": {
102+
"tool_name": TOOL_CLASSES["mcp_run_python"],
103+
"tool_input": {
104+
"callable": "autoskillit.recipe.run_thing",
105+
},
106+
},
107+
"mcp_open_kitchen": {
108+
"tool_name": TOOL_CLASSES["mcp_open_kitchen"],
109+
"tool_input": {},
110+
},
111+
"mcp_remove_clone": {
112+
"tool_name": TOOL_CLASSES["mcp_remove_clone"],
113+
"tool_input": {
114+
"clone_path": "/nonexistent/path/that/does/not/exist",
115+
},
116+
},
117+
"mcp_merge_worktree": {
118+
"tool_name": TOOL_CLASSES["mcp_merge_worktree"],
119+
"tool_input": {
120+
"base_branch": "main",
121+
},
122+
},
123+
"mcp_push_to_remote": {
124+
"tool_name": TOOL_CLASSES["mcp_push_to_remote"],
125+
"tool_input": {
126+
"branch": "main",
127+
},
128+
},
129+
"mcp_dispatch_food_truck": {
130+
"tool_name": TOOL_CLASSES["mcp_dispatch_food_truck"],
131+
"tool_input": {},
132+
},
133+
"mcp_wait_for_ci": {
134+
"tool_name": TOOL_CLASSES["mcp_wait_for_ci"],
135+
"tool_input": {},
136+
},
137+
"mcp_reset_dispatch": {
138+
"tool_name": TOOL_CLASSES["mcp_reset_dispatch"],
139+
"tool_input": {},
140+
},
141+
"apply_patch": {
142+
"tool_name": TOOL_CLASSES["apply_patch"],
143+
"tool_input": {
144+
"patch": "--- a/foo\n+++ b/foo\n@@ -1 +1 @@\n-old\n+new\n",
145+
},
146+
},
147+
}
148+
149+
# Verify the contract: TOOL_CLASS_PAYLOADS.keys() == TOOL_CLASSES.keys()
150+
assert set(TOOL_CLASS_PAYLOADS.keys()) == set(TOOL_CLASSES.keys()), (
151+
"TOOL_CLASS_PAYLOADS keys must match TOOL_CLASSES keys exactly"
152+
)
153+
154+
155+
def _clean_env() -> dict[str, str]:
156+
"""Strip AUTOSKILLIT_* keys from the test runner's environment."""
157+
return {k: v for k, v in os.environ.items() if not k.startswith("AUTOSKILLIT_")}
158+
159+
160+
def _invoke_guard(
161+
script_path: Path,
162+
stdin_payload: dict,
163+
env_overrides: dict[str, str],
164+
tmp_path: Path,
165+
) -> dict:
166+
"""Invoke a guard script as a subprocess and parse its JSON output.
167+
168+
Asserts exit code 0 (fail-open contract — guards must exit 0 on either decision).
169+
Returns the parsed stdout JSON dict, or empty dict on empty stdout.
170+
"""
171+
merged_env = {**_clean_env(), **env_overrides}
172+
result = subprocess.run( # noqa: S603 (intentional subprocess probe)
173+
[sys.executable, str(script_path)],
174+
input=json.dumps(stdin_payload),
175+
env=merged_env,
176+
capture_output=True,
177+
text=True,
178+
cwd=tmp_path,
179+
timeout=5,
180+
)
181+
assert result.returncode == 0, (
182+
f"guard {script_path.name} exited {result.returncode}: {result.stderr}"
183+
)
184+
stdout = result.stdout.strip()
185+
if not stdout:
186+
return {}
187+
return json.loads(stdout)
188+
189+
190+
def _build_payload(tool_class: str, session_mode: str) -> dict:
191+
"""Deep-copy the deny-triggering payload for a tool class, plus subagent fields."""
192+
payload = copy.deepcopy(TOOL_CLASS_PAYLOADS[tool_class])
193+
if session_mode == "subagent":
194+
payload["agent_id"] = "probe-subagent-001"
195+
payload["agent_type"] = "general-purpose"
196+
return payload
197+
198+
199+
def _collect_probe_params() -> list[Any]:
200+
"""Yield pytest.param entries for every PreToolUse+deny combination.
201+
202+
Cross-product: (script) × (matching tool classes) × SESSION_MODES × BACKENDS.
203+
Co-derivation contract: every yielded ``script`` satisfies
204+
``script in HOOK_REGISTRY[hookdef_idx].scripts``.
205+
Yields ALL combinations including inert codex rows (inert check is in test body).
206+
"""
207+
params: list[Any] = []
208+
for hookdef_idx, hd in enumerate(HOOK_REGISTRY):
209+
if hd.event_type != "PreToolUse" or hd.mechanism != "deny":
210+
continue
211+
for tc_key, tc_value in TOOL_CLASSES.items():
212+
if not re.fullmatch(hd.matcher, tc_value):
213+
continue
214+
for script in hd.scripts:
215+
for session_mode in SESSION_MODES:
216+
for backend in BACKENDS:
217+
params.append(
218+
pytest.param(
219+
hookdef_idx,
220+
script,
221+
tc_key,
222+
session_mode,
223+
backend,
224+
id=(f"{Path(script).stem}__{tc_key}__{session_mode}__{backend}"),
225+
)
226+
)
227+
return params
228+
229+
230+
class TestHookDenyEfficacyProbe:
231+
"""Probe each deny-mechanism guard script as a subprocess across the matrix."""
232+
233+
@pytest.mark.parametrize(
234+
"hookdef_idx, script, tool_class, session_mode, backend",
235+
_collect_probe_params(),
236+
)
237+
def test_probe(
238+
self,
239+
hookdef_idx: int,
240+
script: str,
241+
tool_class: str,
242+
session_mode: str,
243+
backend: str,
244+
tmp_path: Path,
245+
) -> None:
246+
hookdef = HOOK_REGISTRY[hookdef_idx]
247+
248+
# --- Inert check for codex backend ---
249+
# Mirrors generate_codex_hooks_config() at _codex_hooks.py:52-55: skip
250+
# session_scope=="interactive_only" AND skip codex_status in {"fix-required", "not-applicable"}.
251+
if backend == "codex":
252+
if hookdef.session_scope == "interactive_only" or hookdef.codex_status in _SKIP_CODEX:
253+
codex_config = generate_codex_hooks_config()
254+
script_stem = Path(script).stem
255+
for entries in codex_config.values():
256+
for entry in entries:
257+
for hook in entry.get("hooks", []):
258+
assert script_stem not in hook.get("command", ""), (
259+
f"{script_stem} found in codex config but should be excluded"
260+
)
261+
return # Inert — do not record to matrix
262+
263+
# --- Build env and payload ---
264+
env = {**SESSION_MODE_ENV[session_mode], **BACKEND_ENV[backend]}
265+
env["AUTOSKILLIT_ALLOWED_WRITE_PREFIXES"] = str(tmp_path)
266+
env["AUTOSKILLIT_CWD"] = str(tmp_path)
267+
payload = _build_payload(tool_class, session_mode)
268+
269+
# --- Set up minimal filesystem state where needed ---
270+
(tmp_path / ".autoskillit" / "temp").mkdir(parents=True, exist_ok=True)
271+
272+
# --- Invoke guard ---
273+
script_path = HOOKS_DIR / script
274+
result = _invoke_guard(script_path, payload, env, tmp_path)
275+
276+
# --- Classify outcome ---
277+
hook_output = result.get("hookSpecificOutput", {})
278+
decision = hook_output.get("permissionDecision")
279+
280+
if decision == "deny":
281+
strength = "hard"
282+
else:
283+
strength = "soft"
284+
285+
# --- Record to matrix ---
286+
record_probe_row(
287+
{
288+
"tool_class": tool_class,
289+
"session_mode": session_mode,
290+
"backend": backend,
291+
"hook": Path(script).stem,
292+
"observed_decision": decision or "allow",
293+
"codex_status_claimed": hookdef.codex_status,
294+
"strength": strength,
295+
}
296+
)
297+
298+
299+
def test_probe_collection_count() -> None:
300+
"""Meta-test: parametrized probe count must match the registry-derived total.
301+
302+
Uses ``EXPECTED_TOTAL_PROBE_COUNT`` (total including inert), since
303+
``_collect_probe_params()`` yields all combinations. Adding a new
304+
deny-mechanism hook automatically fails this test until the matrix adapts.
305+
"""
306+
assert len(_collect_probe_params()) == EXPECTED_TOTAL_PROBE_COUNT

0 commit comments

Comments
 (0)