aboutsummaryrefslogtreecommitdiff
path: root/.ai/scripts/tests/test_cross_agent_halt.py
diff options
context:
space:
mode:
authorCraig Jennings <c@cjennings.net>2026-05-06 21:59:52 -0500
committerCraig Jennings <c@cjennings.net>2026-05-06 21:59:52 -0500
commitd81b23ad6b6e437dfe3c338a00a4be39bc555146 (patch)
tree2d4b0d7890fd1fc70d81282b81fed2808c28a106 /.ai/scripts/tests/test_cross_agent_halt.py
parent201377f57430ef28d02e703a2191434bbee55c75 (diff)
downloadrulesets-d81b23ad6b6e437dfe3c338a00a4be39bc555146.tar.gz
rulesets-d81b23ad6b6e437dfe3c338a00a4be39bc555146.zip
chore(ai): initialize project notes and Claude tooling surfaces
Replace the seed notes.org with project-specific context (layout, install modes, task tracker location, recent inflection point). Bring in the synced template surfaces (protocols, workflows, scripts, references, retrospectives, someday-maybe) as tracked content for this content/documentation project.
Diffstat (limited to '.ai/scripts/tests/test_cross_agent_halt.py')
-rw-r--r--.ai/scripts/tests/test_cross_agent_halt.py204
1 files changed, 204 insertions, 0 deletions
diff --git a/.ai/scripts/tests/test_cross_agent_halt.py b/.ai/scripts/tests/test_cross_agent_halt.py
new file mode 100644
index 0000000..f8bf0b3
--- /dev/null
+++ b/.ai/scripts/tests/test_cross_agent_halt.py
@@ -0,0 +1,204 @@
+"""Tests for cross-agent-halt and cross-agent-resume (TDD)."""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import textwrap
+from pathlib import Path
+
+import pytest
+
+HALT_SCRIPT = Path(__file__).resolve().parent.parent / "cross-agent-comms" / "cross-agent-halt"
+RESUME_SCRIPT = Path(__file__).resolve().parent.parent / "cross-agent-comms" / "cross-agent-resume"
+
+
+def _run(script: Path, args: list[str], env: dict | None = None) -> subprocess.CompletedProcess:
+ return subprocess.run([str(script), *args], capture_output=True, text=True, env=env)
+
+
+@pytest.fixture
+def isolated_env(tmp_path, monkeypatch):
+ """Isolated HOME + a fake systemctl that records calls without acting."""
+ fake_home = tmp_path / "home"
+ fake_home.mkdir()
+ fake_bin = tmp_path / "bin"
+ fake_bin.mkdir()
+ # Fake systemctl: no-op, exit 0.
+ fake_systemctl = fake_bin / "systemctl"
+ fake_systemctl.write_text("#!/usr/bin/env bash\nexit 0\n")
+ fake_systemctl.chmod(0o755)
+ # Fake ssh: succeed only for known-good host.
+ fake_ssh = fake_bin / "ssh"
+ fake_ssh.write_text(textwrap.dedent("""\
+ #!/usr/bin/env bash
+ # Find the destination arg (skip flags).
+ target=""
+ for arg in "$@"; do
+ case "$arg" in
+ -*|*=*) ;;
+ *@*|localhost|*.local|*.invalid) target="$arg"; break ;;
+ *) target="$arg"; break ;;
+ esac
+ done
+ case "$target" in
+ *invalid*|*unreachable*) exit 255 ;;
+ *) exit 0 ;;
+ esac
+ """))
+ fake_ssh.chmod(0o755)
+
+ monkeypatch.setenv("HOME", str(fake_home))
+ # Prepend our fake bin so systemctl + ssh are intercepted, but keep real /bin etc.
+ monkeypatch.setenv("PATH", f"{fake_bin}:{os.environ.get('PATH', '')}")
+ return fake_home
+
+
+# ---- cross-agent-halt ----
+
+
+def test_halt_help(isolated_env):
+ result = _run(HALT_SCRIPT, ["--help"], env={**os.environ, "HOME": str(isolated_env),
+ "PATH": os.environ["PATH"]})
+ assert result.returncode == 0
+ assert "halt" in result.stdout.lower()
+
+
+def test_halt_creates_halt_file(isolated_env):
+ halt_file = isolated_env / ".config" / "cross-agent-comms" / "HALT"
+ assert not halt_file.exists()
+ result = _run(HALT_SCRIPT, [], env={**os.environ, "HOME": str(isolated_env),
+ "PATH": os.environ["PATH"]})
+ assert result.returncode == 0
+ assert halt_file.exists()
+
+
+def test_halt_with_reason_writes_body(isolated_env):
+ result = _run(HALT_SCRIPT, ["pausing for incident review"],
+ env={**os.environ, "HOME": str(isolated_env), "PATH": os.environ["PATH"]})
+ assert result.returncode == 0
+ halt_file = isolated_env / ".config" / "cross-agent-comms" / "HALT"
+ assert halt_file.exists()
+ assert "pausing for incident review" in halt_file.read_text()
+
+
+def test_halt_idempotent(isolated_env):
+ """Running halt twice doesn't error."""
+ halt_file = isolated_env / ".config" / "cross-agent-comms" / "HALT"
+ r1 = _run(HALT_SCRIPT, [], env={**os.environ, "HOME": str(isolated_env), "PATH": os.environ["PATH"]})
+ assert r1.returncode == 0
+ assert halt_file.exists()
+ r2 = _run(HALT_SCRIPT, [], env={**os.environ, "HOME": str(isolated_env), "PATH": os.environ["PATH"]})
+ assert r2.returncode == 0
+ assert halt_file.exists()
+
+
+def test_halt_does_not_pkill(isolated_env):
+ """Per design: halt does NOT call pkill. Verify by checking no pkill process gets launched."""
+ # Replace pkill in PATH with something that fails loudly so we'd see if halt invoked it.
+ fake_bin = isolated_env.parent / "bin"
+ pkill = fake_bin / "pkill"
+ pkill.write_text("#!/usr/bin/env bash\necho 'PKILL CALLED' >&2\nexit 99\n")
+ pkill.chmod(0o755)
+ result = _run(HALT_SCRIPT, [], env={**os.environ, "HOME": str(isolated_env), "PATH": os.environ["PATH"]})
+ assert result.returncode == 0
+ assert "PKILL CALLED" not in result.stderr
+
+
+def test_halt_tailnet_reports_per_peer(isolated_env):
+ """--tailnet iterates peers.toml and reports per-peer status."""
+ cfg = isolated_env / ".config" / "cross-agent-comms"
+ cfg.mkdir(parents=True)
+ (cfg / "peers.toml").write_text(textwrap.dedent("""\
+ [peers.velox]
+ host = "velox"
+ ssh_user = "cjennings"
+
+ [peers.bogus]
+ host = "definitely-unreachable.invalid"
+ ssh_user = "cjennings"
+ """))
+ result = _run(HALT_SCRIPT, ["--tailnet"],
+ env={**os.environ, "HOME": str(isolated_env), "PATH": os.environ["PATH"]})
+ # Partial halt → exit 1.
+ assert result.returncode == 1
+ assert "velox" in result.stdout
+ assert "bogus" in result.stdout
+ # ✓ marker for velox, ✗ for bogus.
+ assert "✓" in result.stdout
+ assert "✗" in result.stdout
+ assert "PARTIAL" in result.stdout or "partial" in result.stdout.lower()
+
+
+def test_halt_tailnet_all_reachable_exits_zero(isolated_env):
+ cfg = isolated_env / ".config" / "cross-agent-comms"
+ cfg.mkdir(parents=True)
+ (cfg / "peers.toml").write_text(textwrap.dedent("""\
+ [peers.velox]
+ host = "velox"
+ ssh_user = "cjennings"
+ """))
+ result = _run(HALT_SCRIPT, ["--tailnet"],
+ env={**os.environ, "HOME": str(isolated_env), "PATH": os.environ["PATH"]})
+ assert result.returncode == 0
+ assert "velox" in result.stdout
+
+
+# ---- cross-agent-resume ----
+
+
+def test_resume_help(isolated_env):
+ result = _run(RESUME_SCRIPT, ["--help"],
+ env={**os.environ, "HOME": str(isolated_env), "PATH": os.environ["PATH"]})
+ assert result.returncode == 0
+ assert "resume" in result.stdout.lower()
+
+
+def test_resume_removes_halt_file(isolated_env):
+ halt_file = isolated_env / ".config" / "cross-agent-comms" / "HALT"
+ halt_file.parent.mkdir(parents=True)
+ halt_file.write_text("halted")
+ assert halt_file.exists()
+ result = _run(RESUME_SCRIPT, [],
+ env={**os.environ, "HOME": str(isolated_env), "PATH": os.environ["PATH"]})
+ assert result.returncode == 0
+ assert not halt_file.exists()
+
+
+def test_resume_when_no_halt_active_succeeds(isolated_env):
+ """No HALT to clear is not an error."""
+ result = _run(RESUME_SCRIPT, [],
+ env={**os.environ, "HOME": str(isolated_env), "PATH": os.environ["PATH"]})
+ assert result.returncode == 0
+
+
+def test_resume_prints_per_session_instructions(isolated_env):
+ """Resume must surface that polling does NOT auto-resume."""
+ halt_file = isolated_env / ".config" / "cross-agent-comms" / "HALT"
+ halt_file.parent.mkdir(parents=True)
+ halt_file.write_text("halted")
+ result = _run(RESUME_SCRIPT, [],
+ env={**os.environ, "HOME": str(isolated_env), "PATH": os.environ["PATH"]})
+ assert result.returncode == 0
+ out = result.stdout.lower()
+ assert "polling" in out
+ assert "auto" in out or "explicit" in out or "session" in out
+
+
+def test_resume_tailnet_partial_failure_exit_1(isolated_env):
+ cfg = isolated_env / ".config" / "cross-agent-comms"
+ cfg.mkdir(parents=True)
+ (cfg / "peers.toml").write_text(textwrap.dedent("""\
+ [peers.velox]
+ host = "velox"
+
+ [peers.bogus]
+ host = "unreachable-host.invalid"
+ """))
+ halt_file = cfg / "HALT"
+ halt_file.write_text("halted")
+ result = _run(RESUME_SCRIPT, ["--tailnet"],
+ env={**os.environ, "HOME": str(isolated_env), "PATH": os.environ["PATH"]})
+ assert result.returncode == 1
+ assert "velox" in result.stdout
+ assert "bogus" in result.stdout