Godopu
87b4501bbf
- Replace '--target-agent' with mandatory '--creator <session>' flag
- Add explicit rejection error for legacy '--target-agent'
- Add '--planner <session>' flag with fail-fast check when '--plan' is missing
- Implement 2-branch session validation ('is not registered' vs 'is not running')
- Update in-repo references in hooks, SKILL.md, INSTALL.md, and tests
- Add tests/test_loop_cli.py with 9 unit/contract tests
472 lines
18 KiB
Python
472 lines
18 KiB
Python
"""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"
|
|
|
|
|
|
def _extract_delegate_job_safe():
|
|
run_loop = REPO_ROOT / ".agents" / "skills" / "multi-agent-mux-loop" / "scripts" / "run_loop.sh"
|
|
content = run_loop.read_text()
|
|
func_start = content.find("delegate_job_safe() {")
|
|
assert func_start != -1
|
|
func_end = content.find("\n}\n", func_start)
|
|
assert func_end != -1
|
|
return content[func_start:func_end + 3]
|
|
|
|
|
|
# Z-9: delegate_job_safe preserves loop lock across delegations without tmp copies/traps (B-6, D1)
|
|
def test_z9_loop_lock_survives_delegation(tmp_path):
|
|
loop_lock_sh = REPO_ROOT / ".agents" / "skills" / "multi-agent-mux-loop" / "scripts" / "loop_lock.sh"
|
|
skill_dir = tmp_path / ".agents" / "skills" / "multi-agent-mux-delegate-job"
|
|
skill_dir.mkdir(parents=True)
|
|
stub_wrapper = skill_dir / "multi-agent-mux-delegate-job"
|
|
stub_wrapper.write_text("#!/usr/bin/env bash\necho 'JOB_ID: 12345678'\n")
|
|
stub_wrapper.chmod(0o755)
|
|
|
|
extracted_func = _extract_delegate_job_safe()
|
|
script = f"""#!/usr/bin/env bash
|
|
set -euo pipefail
|
|
REPO_ROOT="{tmp_path}"
|
|
MAM_LOOP_MARKER="$REPO_ROOT/.mam/loop-guard-active"
|
|
source "{loop_lock_sh}"
|
|
|
|
log_error() {{ echo "ERROR: $@" >&2; }}
|
|
log_info() {{ echo "INFO: $@" >&2; }}
|
|
|
|
_mam_release_guard() {{ mam_release_loop_lock "$MAM_LOOP_MARKER" || true; }}
|
|
|
|
mam_acquire_loop_lock "$MAM_LOOP_MARKER"
|
|
trap _mam_release_guard EXIT INT TERM HUP
|
|
|
|
{extracted_func}
|
|
|
|
PLAN_JOB_OUTPUT=$(delegate_job_safe submit --task test)
|
|
|
|
if [ -f "$MAM_LOOP_MARKER" ]; then
|
|
echo "MARKER: HELD"
|
|
else
|
|
echo "MARKER: RELEASED"
|
|
fi
|
|
"""
|
|
res = subprocess.run(["bash", "-c", script], capture_output=True, text=True)
|
|
assert res.returncode == 0
|
|
assert "MARKER: HELD" in res.stdout
|
|
|
|
|
|
def test_z9_probe_detects_the_defect(tmp_path):
|
|
loop_lock_sh = REPO_ROOT / ".agents" / "skills" / "multi-agent-mux-loop" / "scripts" / "loop_lock.sh"
|
|
skill_dir = tmp_path / ".agents" / "skills" / "multi-agent-mux-delegate-job"
|
|
skill_dir.mkdir(parents=True)
|
|
stub_wrapper = skill_dir / "multi-agent-mux-delegate-job"
|
|
stub_wrapper.write_text("#!/usr/bin/env bash\necho 'JOB_ID: 12345678'\n")
|
|
stub_wrapper.chmod(0o755)
|
|
|
|
defective_func = """
|
|
delegate_job_safe() {
|
|
local orig_script="$REPO_ROOT/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job"
|
|
local tmp_script
|
|
tmp_script="${orig_script}.${RANDOM}_$$.tmp"
|
|
cp "$orig_script" "$tmp_script"
|
|
trap 'rm -f "$tmp_script"' EXIT INT TERM HUP
|
|
local rc=0
|
|
bash "$tmp_script" "$@" || rc=$?
|
|
rm -f "$tmp_script"
|
|
trap _mam_release_guard EXIT INT TERM HUP
|
|
return $rc
|
|
}
|
|
"""
|
|
script = f"""#!/usr/bin/env bash
|
|
set -euo pipefail
|
|
REPO_ROOT="{tmp_path}"
|
|
MAM_LOOP_MARKER="$REPO_ROOT/.mam/loop-guard-active"
|
|
source "{loop_lock_sh}"
|
|
|
|
log_error() {{ echo "ERROR: $@" >&2; }}
|
|
log_info() {{ echo "INFO: $@" >&2; }}
|
|
|
|
_mam_release_guard() {{ mam_release_loop_lock "$MAM_LOOP_MARKER" || true; }}
|
|
|
|
mam_acquire_loop_lock "$MAM_LOOP_MARKER"
|
|
trap _mam_release_guard EXIT INT TERM HUP
|
|
|
|
{defective_func}
|
|
|
|
PLAN_JOB_OUTPUT=$(delegate_job_safe submit --task test)
|
|
|
|
if [ -f "$MAM_LOOP_MARKER" ]; then
|
|
echo "MARKER: HELD"
|
|
else
|
|
echo "MARKER: RELEASED"
|
|
fi
|
|
"""
|
|
res = subprocess.run(["bash", "-c", script], capture_output=True, text=True)
|
|
assert res.returncode == 0
|
|
assert "MARKER: RELEASED" in res.stdout
|
|
|
|
|
|
def test_z9_no_tmp_copy_left_in_skill_tree(tmp_path):
|
|
loop_lock_sh = REPO_ROOT / ".agents" / "skills" / "multi-agent-mux-loop" / "scripts" / "loop_lock.sh"
|
|
skill_dir = tmp_path / ".agents" / "skills" / "multi-agent-mux-delegate-job"
|
|
skill_dir.mkdir(parents=True)
|
|
stub_wrapper = skill_dir / "multi-agent-mux-delegate-job"
|
|
stub_wrapper.write_text("#!/usr/bin/env bash\necho 'JOB_ID: 12345678'\n")
|
|
stub_wrapper.chmod(0o755)
|
|
|
|
extracted_func = _extract_delegate_job_safe()
|
|
script = f"""#!/usr/bin/env bash
|
|
set -euo pipefail
|
|
REPO_ROOT="{tmp_path}"
|
|
MAM_LOOP_MARKER="$REPO_ROOT/.mam/loop-guard-active"
|
|
source "{loop_lock_sh}"
|
|
|
|
log_error() {{ echo "ERROR: $@" >&2; }}
|
|
log_info() {{ echo "INFO: $@" >&2; }}
|
|
|
|
_mam_release_guard() {{ mam_release_loop_lock "$MAM_LOOP_MARKER" || true; }}
|
|
|
|
mam_acquire_loop_lock "$MAM_LOOP_MARKER"
|
|
trap _mam_release_guard EXIT INT TERM HUP
|
|
|
|
{extracted_func}
|
|
|
|
PLAN_JOB_OUTPUT=$(delegate_job_safe submit --task test)
|
|
"""
|
|
res = subprocess.run(["bash", "-c", script], capture_output=True, text=True)
|
|
assert res.returncode == 0
|
|
tmp_files = list(skill_dir.glob("*.tmp"))
|
|
assert tmp_files == []
|
|
|
|
|
|
def test_z9_exit_code_and_diagnostics_propagation(tmp_path):
|
|
loop_lock_sh = REPO_ROOT / ".agents" / "skills" / "multi-agent-mux-loop" / "scripts" / "loop_lock.sh"
|
|
skill_dir = tmp_path / ".agents" / "skills" / "multi-agent-mux-delegate-job"
|
|
skill_dir.mkdir(parents=True)
|
|
stub_wrapper = skill_dir / "multi-agent-mux-delegate-job"
|
|
stub_wrapper.write_text("#!/usr/bin/env bash\necho 'syntax error' >&2\nexit 7\n")
|
|
stub_wrapper.chmod(0o755)
|
|
|
|
extracted_func = _extract_delegate_job_safe()
|
|
script = f"""#!/usr/bin/env bash
|
|
REPO_ROOT="{tmp_path}"
|
|
MAM_LOOP_MARKER="$REPO_ROOT/.mam/loop-guard-active"
|
|
source "{loop_lock_sh}"
|
|
|
|
log_error() {{ echo "ERROR: $@" >&2; }}
|
|
log_info() {{ echo "INFO: $@" >&2; }}
|
|
|
|
_mam_release_guard() {{ mam_release_loop_lock "$MAM_LOOP_MARKER" || true; }}
|
|
|
|
mam_acquire_loop_lock "$MAM_LOOP_MARKER"
|
|
trap _mam_release_guard EXIT INT TERM HUP
|
|
|
|
{extracted_func}
|
|
|
|
rc=0
|
|
delegate_job_safe submit --task test || rc=$?
|
|
echo "DELEGATE_RC: $rc"
|
|
"""
|
|
res = subprocess.run(["bash", "-c", script], capture_output=True, text=True)
|
|
assert "DELEGATE_RC: 7" in res.stdout
|
|
assert "delegate_job_safe failed (exit 7):" in res.stderr
|
|
assert "bash -n" in res.stderr
|
|
|
|
|
|
# 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: loop_lock.sh records lstart in marker
|
|
def test_z14_run_loop_records_lstart(mam_sandbox):
|
|
lock_script = REPO_ROOT / ".agents" / "skills" / "multi-agent-mux-loop" / "scripts" / "loop_lock.sh"
|
|
content = lock_script.read_text()
|
|
assert "mam_lstart" in content
|
|
assert "lstart=" in content
|
|
|
|
|
|
# ===========================================================================
|
|
# B-13 Stage 2: Self-hosting loop runtime freeze snapshot regression guards
|
|
# ===========================================================================
|
|
|
|
def test_b13_reexec_preserves_original_argv(tmp_path):
|
|
"""B-13/P1: the freeze re-exec must forward the original arguments.
|
|
The arg parser consumes "$@" (shift), so capturing argv beforehand is mandatory.
|
|
"""
|
|
run_loop_sh = REPO_ROOT / ".agents" / "skills" / "multi-agent-mux-loop" / "scripts" / "run_loop.sh"
|
|
res = subprocess.run(
|
|
["bash", str(run_loop_sh), "--creator", "nonexistent-agent", "--task", "test goal with spaces"],
|
|
capture_output=True,
|
|
text=True,
|
|
cwd=str(REPO_ROOT),
|
|
)
|
|
combined = res.stdout + res.stderr
|
|
assert "mandatory fields" not in combined
|
|
|
|
|
|
def test_b13_freeze_survives_broken_wrapper(tmp_path):
|
|
"""B-13: a wrapper broken mid-loop must not break an already-frozen runtime."""
|
|
skills_src = REPO_ROOT / ".agents" / "skills"
|
|
skills_dst = tmp_path / ".agents" / "skills"
|
|
shutil.copytree(skills_src, skills_dst)
|
|
|
|
freeze_dir = tmp_path / "freeze"
|
|
freeze_skills = freeze_dir / ".agents" / "skills"
|
|
shutil.copytree(skills_dst, freeze_skills)
|
|
|
|
wrapper_orig = skills_dst / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job"
|
|
wrapper_orig.write_text('echo "broken wrapper syntax" "\nunexpected EOF\n')
|
|
|
|
r_orig = subprocess.run(["bash", str(wrapper_orig)], capture_output=True, text=True)
|
|
assert r_orig.returncode != 0
|
|
|
|
wrapper_frozen = freeze_skills / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job"
|
|
r_frozen = subprocess.run(["bash", str(wrapper_frozen), "--help"], capture_output=True, text=True)
|
|
assert r_frozen.returncode == 0
|
|
|
|
|
|
def test_b13_freeze_dir_is_outside_the_skill_tree(tmp_path):
|
|
"""B-13 must not reintroduce B-6: nothing may be written under .agents/skills/."""
|
|
run_loop_sh = REPO_ROOT / ".agents" / "skills" / "multi-agent-mux-loop" / "scripts" / "run_loop.sh"
|
|
subprocess.run(["bash", str(run_loop_sh), "--help"], capture_output=True, text=True, cwd=str(REPO_ROOT))
|
|
assert list((REPO_ROOT / ".agents" / "skills").rglob("*.tmp")) == []
|
|
assert "mam-loop-freeze" not in str(list((REPO_ROOT / ".agents").rglob("*")))
|
|
|
|
|
|
def test_b13_release_guard_cleans_up_and_releases_lock(tmp_path):
|
|
"""B-13 cleanup must extend _mam_release_guard, releasing lock and removing snapshot."""
|
|
loop_lock_sh = REPO_ROOT / ".agents" / "skills" / "multi-agent-mux-loop" / "scripts" / "loop_lock.sh"
|
|
marker = tmp_path / ".mam" / "loop-guard-active"
|
|
freeze_dir = tmp_path / "mam-loop-freeze.TEST1234"
|
|
freeze_dir.mkdir(parents=True)
|
|
(freeze_dir / "dummy").write_text("test")
|
|
|
|
script = f"""#!/usr/bin/env bash
|
|
set -euo pipefail
|
|
MAM_REAL_ROOT="{tmp_path}"
|
|
MAM_LOOP_MARKER="{marker}"
|
|
MAM_LOOP_FREEZE_DIR="{freeze_dir}"
|
|
MAM_LOOP_FREEZE_OWNED="1"
|
|
source "{loop_lock_sh}"
|
|
|
|
_mam_release_guard() {{
|
|
mam_release_loop_lock "$MAM_LOOP_MARKER" || true
|
|
if [ -n "${{MAM_LOOP_FREEZE_DIR:-}}" ] && [ "${{MAM_LOOP_FREEZE_OWNED:-0}}" = "1" ]; then
|
|
case "$MAM_LOOP_FREEZE_DIR" in
|
|
*/mam-loop-freeze.*) rm -rf "$MAM_LOOP_FREEZE_DIR" ;;
|
|
*) : ;;
|
|
esac
|
|
fi
|
|
}}
|
|
|
|
mam_acquire_loop_lock "$MAM_LOOP_MARKER"
|
|
trap _mam_release_guard EXIT INT TERM HUP
|
|
|
|
[ -f "$MAM_LOOP_MARKER" ]
|
|
[ -d "$MAM_LOOP_FREEZE_DIR" ]
|
|
"""
|
|
res = subprocess.run(["bash", "-c", script], capture_output=True, text=True)
|
|
assert res.returncode == 0
|
|
assert not marker.exists(), "loop lock marker was not released"
|
|
assert not freeze_dir.exists(), "freeze snapshot dir was not cleaned up"
|
|
|
|
|
|
def test_b13_no_freeze_switch_disables_reexec(tmp_path):
|
|
"""MAM_LOOP_NO_FREEZE=1 must skip the snapshot and re-exec entirely."""
|
|
run_loop_sh = REPO_ROOT / ".agents" / "skills" / "multi-agent-mux-loop" / "scripts" / "run_loop.sh"
|
|
env = os.environ.copy()
|
|
env["MAM_LOOP_NO_FREEZE"] = "1"
|
|
res = subprocess.run(
|
|
["bash", str(run_loop_sh), "--creator", "nonexistent-agent", "--task", "test goal"],
|
|
capture_output=True,
|
|
text=True,
|
|
env=env,
|
|
cwd=str(REPO_ROOT),
|
|
)
|
|
assert "mam-loop-freeze" not in res.stderr
|
|
|