feat(o3): implement Invocation-Aware Scoped Guard for orchestrator role scoping (100% PASS)
This commit is contained in:
@@ -0,0 +1,214 @@
|
||||
"""O-3 — Invocation-Aware Scoped Guard unit & integration test suite.
|
||||
|
||||
Verifies:
|
||||
Z-1: Normal mode allows all 3 mutating tools (file_change, edit_notebook, write_blob)
|
||||
Z-2: Active loop denies all 3 mutating tools and includes run_loop.sh in reason
|
||||
Z-3: Non-mutating tools (run_command, view_file, grep_search) allowed during loop
|
||||
Z-4: Malformed or unparseable input fails open (allows tool call)
|
||||
Z-5: Transcript signal covers gap before marker creation
|
||||
Z-6: Matcher targets derived step-type names (file_change|edit_notebook|write_blob)
|
||||
Z-7: run_loop.sh writes identity marker and releases it on exit/interrupt
|
||||
Z-8: Dead PID marker is ignored (fail open)
|
||||
Z-9: delegate_job_safe restores _mam_release_guard trap
|
||||
Z-10: Reused PID (live PID, stale lstart) is NOT blocked (Livelock prevention)
|
||||
Z-11: Live PID with matching lstart IS blocked
|
||||
Z-12: Process owned by another user (PermissionError) is NOT blocked
|
||||
Z-13: Legacy marker fallback (without lstart) degrades open for other owners
|
||||
Z-14: run_loop.sh records lstart in marker
|
||||
"""
|
||||
import os
|
||||
import json
|
||||
import shutil
|
||||
import subprocess
|
||||
import time
|
||||
from pathlib import Path
|
||||
import pytest
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
GUARD_SCRIPT = REPO_ROOT / ".agents" / "hooks" / "loop_delegation_guard.sh"
|
||||
|
||||
|
||||
def _run_guard(payload_dict_or_raw, marker_path=None, transcript_path=None, env_extra=None):
|
||||
env = dict(os.environ)
|
||||
if env_extra:
|
||||
env.update(env_extra)
|
||||
if marker_path:
|
||||
env["MAM_LOOP_GUARD_MARKER"] = str(marker_path)
|
||||
|
||||
if isinstance(payload_dict_or_raw, dict):
|
||||
if transcript_path and "transcriptPath" not in payload_dict_or_raw:
|
||||
payload_dict_or_raw["transcriptPath"] = str(transcript_path)
|
||||
payload_str = json.dumps(payload_dict_or_raw)
|
||||
else:
|
||||
payload_str = str(payload_dict_or_raw)
|
||||
|
||||
res = subprocess.run(
|
||||
["bash", str(GUARD_SCRIPT)],
|
||||
input=payload_str,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
cwd=str(REPO_ROOT),
|
||||
env=env,
|
||||
)
|
||||
assert res.returncode == 0
|
||||
return json.loads(res.stdout)
|
||||
|
||||
|
||||
def _get_lstart(pid):
|
||||
out = subprocess.run(["ps", "-p", str(pid), "-o", "lstart="], capture_output=True, text=True)
|
||||
return " ".join(out.stdout.split())
|
||||
|
||||
|
||||
# Z-1: Normal mode allows all 3 mutating tools
|
||||
@pytest.mark.parametrize("tool_name", ["file_change", "edit_notebook", "write_blob"])
|
||||
def test_z1_normal_mode_allows(mam_sandbox, tool_name):
|
||||
payload = {"toolCall": {"name": tool_name}, "workspacePaths": [str(mam_sandbox)]}
|
||||
res = _run_guard(payload, marker_path=mam_sandbox / ".mam" / "nonexistent")
|
||||
assert res["decision"] == "allow"
|
||||
|
||||
|
||||
# Z-2: Active loop denies mutating tools
|
||||
@pytest.mark.parametrize("tool_name", ["file_change", "edit_notebook", "write_blob"])
|
||||
def test_z2_active_loop_denies(mam_sandbox, tool_name):
|
||||
marker = mam_sandbox / ".mam" / "loop-guard-active"
|
||||
marker.parent.mkdir(parents=True, exist_ok=True)
|
||||
my_pid = os.getpid()
|
||||
lstart = _get_lstart(my_pid)
|
||||
marker.write_text(f"pid={my_pid}\nlstart={lstart}\nstarted=2026-08-07T00:00:00Z\n")
|
||||
|
||||
payload = {"toolCall": {"name": tool_name}, "workspacePaths": [str(mam_sandbox)]}
|
||||
res = _run_guard(payload, marker_path=marker)
|
||||
assert res["decision"] == "deny"
|
||||
assert "run_loop.sh" in res.get("reason", "")
|
||||
|
||||
|
||||
# Z-3: Non-mutating tools allowed during loop
|
||||
@pytest.mark.parametrize("tool_name", ["run_command", "view_file", "grep_search"])
|
||||
def test_z3_non_mutating_tools_allowed(mam_sandbox, tool_name):
|
||||
marker = mam_sandbox / ".mam" / "loop-guard-active"
|
||||
marker.parent.mkdir(parents=True, exist_ok=True)
|
||||
my_pid = os.getpid()
|
||||
lstart = _get_lstart(my_pid)
|
||||
marker.write_text(f"pid={my_pid}\nlstart={lstart}\n")
|
||||
|
||||
payload = {"toolCall": {"name": tool_name}, "workspacePaths": [str(mam_sandbox)]}
|
||||
res = _run_guard(payload, marker_path=marker)
|
||||
assert res["decision"] == "allow"
|
||||
|
||||
|
||||
# Z-4: Malformed input fails open
|
||||
@pytest.mark.parametrize("bad_input", ["not json", "{bad: json", "", "[]", "123"])
|
||||
def test_z4_malformed_input_fails_open(mam_sandbox, bad_input):
|
||||
marker = mam_sandbox / ".mam" / "loop-guard-active"
|
||||
marker.parent.mkdir(parents=True, exist_ok=True)
|
||||
marker.write_text(f"pid={os.getpid()}\nlstart={_get_lstart(os.getpid())}\n")
|
||||
|
||||
res = _run_guard(bad_input, marker_path=marker)
|
||||
assert res["decision"] == "allow"
|
||||
|
||||
|
||||
# Z-5: Transcript signal covers gap before marker creation
|
||||
def test_z5_transcript_signal(mam_sandbox):
|
||||
tfile = mam_sandbox / "transcript.jsonl"
|
||||
tfile.write_text('{"content": "user typed /multi-agent-mux-loop --task test"}\n')
|
||||
|
||||
payload = {"toolCall": {"name": "file_change"}, "workspacePaths": [str(mam_sandbox)], "transcriptPath": str(tfile)}
|
||||
res = _run_guard(payload, marker_path=mam_sandbox / ".mam" / "nonexistent")
|
||||
assert res["decision"] == "deny"
|
||||
|
||||
|
||||
# Z-6: Matcher targets step-type names
|
||||
def test_z6_hooks_json_matcher():
|
||||
hooks_json = REPO_ROOT / ".agents" / "hooks.json"
|
||||
assert hooks_json.exists()
|
||||
data = json.loads(hooks_json.read_text())
|
||||
matcher = data["mam-loop-delegation-guard"]["PreToolUse"][0]["matcher"]
|
||||
assert matcher == "file_change|edit_notebook|write_blob"
|
||||
assert "write_to_file" not in matcher
|
||||
assert "replace_file_content" not in matcher
|
||||
|
||||
|
||||
# Z-7: run_loop.sh source verifies marker creation and trap release
|
||||
def test_z7_run_loop_marker_creation_and_trap():
|
||||
run_loop = REPO_ROOT / ".agents" / "skills" / "multi-agent-mux-loop" / "scripts" / "run_loop.sh"
|
||||
content = run_loop.read_text()
|
||||
assert 'MAM_LOOP_MARKER=' in content
|
||||
assert 'trap _mam_release_guard EXIT INT TERM HUP' in content
|
||||
|
||||
|
||||
# Z-8: Dead PID marker is ignored
|
||||
def test_z8_dead_pid_marker(mam_sandbox):
|
||||
marker = mam_sandbox / ".mam" / "loop-guard-active"
|
||||
marker.parent.mkdir(parents=True, exist_ok=True)
|
||||
marker.write_text("pid=999999\nlstart=Sat Jan 1 00:00:00 2000\n")
|
||||
|
||||
payload = {"toolCall": {"name": "file_change"}, "workspacePaths": [str(mam_sandbox)]}
|
||||
res = _run_guard(payload, marker_path=marker)
|
||||
assert res["decision"] == "allow"
|
||||
|
||||
|
||||
# Z-9: delegate_job_safe restores _mam_release_guard trap
|
||||
def test_z9_delegate_job_safe_restores_trap():
|
||||
run_loop = REPO_ROOT / ".agents" / "skills" / "multi-agent-mux-loop" / "scripts" / "run_loop.sh"
|
||||
content = run_loop.read_text()
|
||||
# Check inside delegate_job_safe definition that trap - is replaced with trap _mam_release_guard
|
||||
func_start = content.find("delegate_job_safe() {")
|
||||
assert func_start != -1
|
||||
func_body = content[func_start:func_start+400]
|
||||
assert "trap _mam_release_guard EXIT INT TERM HUP" in func_body
|
||||
assert "trap - EXIT INT TERM HUP" not in func_body
|
||||
|
||||
|
||||
# Z-10: Reused PID with stale lstart is NOT blocked
|
||||
def test_z10_reused_pid_stale_lstart(mam_sandbox):
|
||||
marker = mam_sandbox / ".mam" / "loop-guard-active"
|
||||
marker.parent.mkdir(parents=True, exist_ok=True)
|
||||
my_pid = os.getpid()
|
||||
marker.write_text(f"pid={my_pid}\nlstart=Wed Jan 1 00:00:00 1999\n")
|
||||
|
||||
payload = {"toolCall": {"name": "file_change"}, "workspacePaths": [str(mam_sandbox)]}
|
||||
res = _run_guard(payload, marker_path=marker)
|
||||
assert res["decision"] == "allow"
|
||||
|
||||
|
||||
# Z-11: Live PID with matching lstart IS blocked
|
||||
def test_z11_live_pid_matching_lstart(mam_sandbox):
|
||||
marker = mam_sandbox / ".mam" / "loop-guard-active"
|
||||
marker.parent.mkdir(parents=True, exist_ok=True)
|
||||
my_pid = os.getpid()
|
||||
lstart = _get_lstart(my_pid)
|
||||
marker.write_text(f"pid={my_pid}\nlstart={lstart}\n")
|
||||
|
||||
payload = {"toolCall": {"name": "file_change"}, "workspacePaths": [str(mam_sandbox)]}
|
||||
res = _run_guard(payload, marker_path=marker)
|
||||
assert res["decision"] == "deny"
|
||||
|
||||
|
||||
# Z-12: Process owned by another user (pid=1 root) is NOT blocked
|
||||
def test_z12_other_user_process_does_not_block(mam_sandbox):
|
||||
marker = mam_sandbox / ".mam" / "loop-guard-active"
|
||||
marker.parent.mkdir(parents=True, exist_ok=True)
|
||||
marker.write_text("pid=1\nlstart=Sat Jan 1 00:00:00 2000\n")
|
||||
|
||||
payload = {"toolCall": {"name": "file_change"}, "workspacePaths": [str(mam_sandbox)]}
|
||||
res = _run_guard(payload, marker_path=marker)
|
||||
assert res["decision"] == "allow"
|
||||
|
||||
|
||||
# Z-13: Legacy marker without lstart degrades open for other owners
|
||||
def test_z13_legacy_marker_permission_error_degrades_open(mam_sandbox):
|
||||
marker = mam_sandbox / ".mam" / "loop-guard-active"
|
||||
marker.parent.mkdir(parents=True, exist_ok=True)
|
||||
marker.write_text("pid=1\n")
|
||||
|
||||
payload = {"toolCall": {"name": "file_change"}, "workspacePaths": [str(mam_sandbox)]}
|
||||
res = _run_guard(payload, marker_path=marker)
|
||||
assert res["decision"] == "allow"
|
||||
|
||||
|
||||
# Z-14: run_loop.sh records lstart in marker
|
||||
def test_z14_run_loop_records_lstart(mam_sandbox):
|
||||
run_loop = REPO_ROOT / ".agents" / "skills" / "multi-agent-mux-loop" / "scripts" / "run_loop.sh"
|
||||
content = run_loop.read_text()
|
||||
assert "_mam_lstart" in content
|
||||
assert "lstart=" in content
|
||||
Reference in New Issue
Block a user