import hashlib import json import os import signal import subprocess import sys import tempfile import time import unittest from pathlib import Path SCRIPT = Path(__file__).resolve().parents[2] / "agents/impl/skills/gitea-issue-rework/scripts/evidence_manifest.py" SCRATCH = Path("/opt/data/cache/scratch/evidence-deadline-regressions") def git(cwd, *args): return subprocess.run(["git", *args], cwd=cwd, check=True, capture_output=True, text=True) class EvidenceCommandDeadlineTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory(dir="/opt/data/cache/scratch") root = Path(self.tmp.name) self.contract = root / "harness/operations/repository-cache.md" self.contract.parent.mkdir(parents=True) self.contract.write_text("contract\n") self.repo = root / "repositories/o/p/worktrees/issue-1/checkout" self.artifacts = self.repo.parent / "verification/run-a" self.repo.mkdir(parents=True) self.artifacts.mkdir(parents=True) for p in (root / "repositories", root / "repositories/o", root / "repositories/o/p", root / "repositories/o/p/worktrees", root / "repositories/o/p/worktrees/issue-1", self.repo, self.artifacts.parent, self.artifacts): if p.exists(): p.chmod(0o700) git(self.repo, "init", "-q") git(self.repo, "config", "user.email", "test@example.invalid") git(self.repo, "config", "user.name", "Test") (self.repo / "input.txt").write_text("input\n") git(self.repo, "add", "input.txt") git(self.repo, "commit", "-qm", "initial") self.known_pid_files = set() self.outer_processes = set() def tearDown(self): for pid_file in self.known_pid_files: try: pid = int(pid_file.read_text().strip()) except (FileNotFoundError, ValueError): continue self._kill_pid_group(pid) for proc in self.outer_processes: if proc.poll() is None: try: os.killpg(proc.pid, signal.SIGKILL) except ProcessLookupError: pass proc.wait(timeout=1) self.tmp.cleanup() @staticmethod def _kill_pid_group(pid): try: pgid = os.getpgid(pid) if pgid == os.getpgrp(): raise AssertionError("refusing cleanup of test runner group") os.killpg(pgid, signal.SIGTERM) time.sleep(0.05) except ProcessLookupError: return try: os.killpg(pgid, signal.SIGKILL) except ProcessLookupError: pass def invoke(self, plan, output="manifest.json", op="run", manifest=None, env=None): if op == "run": args = ["run", "--contract-path", str(self.contract), "--repo-root", str(self.repo), "--artifact-root", str(self.artifacts), "--plan", str(plan.name), "--output", output] elif op == "validate": args = ["validate", "--contract-path", str(self.contract), "--repo-root", str(self.repo), "--artifact-root", str(self.artifacts), "--manifest", str(manifest or output)] elif op == "validate-archived-v2": args = ["validate-archived-v2", "--contract-path", str(self.contract), "--repo-root", str(self.repo), "--artifact-root", str(self.artifacts), "--manifest", str(manifest or output)] else: raise AssertionError(op) proc = subprocess.Popen([sys.executable, str(SCRIPT), *args], cwd=self.repo, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, env=env, start_new_session=True) self.outer_processes.add(proc) try: stdout, stderr = proc.communicate(timeout=8) except subprocess.TimeoutExpired: self._kill_pid_group(proc.pid) stdout, stderr = proc.communicate(timeout=1) stderr += "\nOUTER_TIMEOUT\n" return subprocess.CompletedProcess(proc.args, proc.returncode, stdout, stderr) def make_plan(self, argv, timeout: float | None = 1.5, max_output_bytes=65536, expected_exit_code=0, extra=None, commands=None): plan = self.artifacts / "plan.json" if commands is None: commands = [{"id": "probe", "argv": argv, "timeout_seconds": timeout, "max_output_bytes": max_output_bytes, "expected_exit_code": expected_exit_code, "claims": ["probe"]}] data = {"hash_paths": ["input.txt"], "commands": commands} if extra: data.update(extra) plan.write_text(json.dumps(data)) plan.chmod(0o600) return plan def pid_file(self, name="child.pid"): path = Path(self.tmp.name) / name self.known_pid_files.add(path) return path def assert_no_live_pid(self, pid_file): self.assertTrue(pid_file.is_file(), "child must actually have executed") pid = int(pid_file.read_text().strip()) proc = Path("/proc") / str(pid) if not proc.exists(): return state = (proc / "stat").read_text().split()[2] self.assertIn(state, {"Z"}, f"descendant still live: pid={pid} state={state}") def test_leader_exits_zero_descendant_retains_stdout_is_killed(self): pid = self.pid_file() code = ("import os,signal,time; p=os.fork(); " f"(open({str(pid)!r},'w').write(str(os.getpid())) if p==0 else os._exit(0)); " "signal.signal(signal.SIGTERM,signal.SIG_IGN); time.sleep(30)") result = self.invoke(self.make_plan([sys.executable, "-c", code])) self.assertEqual(result.returncode, 1, result.stderr) record = json.loads((self.artifacts / "manifest.json").read_text())["commands"][0] self.assertEqual(record["execution_status"], "timeout") self.assertTrue(record["termination"]["kill_sent"]) self.assertTrue(record["termination"]["group_cleanup_completed"]) self.assert_no_live_pid(pid) def test_leader_dies_on_term_descendant_ignores_term_and_requires_kill(self): pid = self.pid_file() code = ("import os,signal,time; p=os.fork(); " f"(signal.signal(signal.SIGTERM,signal.SIG_IGN), open({str(pid)!r},'w').write(str(os.getpid())), time.sleep(30)) if p==0 else (signal.signal(signal.SIGTERM,signal.SIG_DFL), time.sleep(30))") result = self.invoke(self.make_plan([sys.executable, "-c", code])) self.assertEqual(result.returncode, 1, result.stderr) record = json.loads((self.artifacts / "manifest.json").read_text())["commands"][0] self.assertEqual(record["execution_status"], "timeout") self.assertTrue(record["termination"]["kill_sent"]) self.assertTrue(record["termination"]["group_cleanup_completed"]) self.assert_no_live_pid(pid) def test_normal_leader_exit_with_live_descendant_is_not_success(self): pid = self.pid_file() code = ("import os,time\np=os.fork()\nif p==0:\n" f" open({str(pid)!r},'w').write(str(os.getpid()))\n" " os.close(1)\n os.close(2)\n time.sleep(30)\nelse:\n os._exit(0)") result = self.invoke(self.make_plan([sys.executable, "-c", code])) self.assertEqual(result.returncode, 1, result.stderr) self.assert_no_live_pid(pid) record = json.loads((self.artifacts / "manifest.json").read_text())["commands"][0] self.assertEqual(record["execution_status"], "descendant_cleanup") self.assertTrue(record["termination"]["group_cleanup_completed"]) self.assertEqual(self.invoke(None, op="validate").returncode, 0) def test_stdout_eof_while_leader_runs_is_bounded(self): code = "import os,time; os.close(1); os.close(2); time.sleep(30)" started = time.monotonic() result = self.invoke(self.make_plan([sys.executable, "-c", code])) self.assertLess(time.monotonic() - started, 6) self.assertEqual(result.returncode, 1, result.stderr) record = json.loads((self.artifacts / "manifest.json").read_text())["commands"][0] self.assertEqual(record["execution_status"], "timeout") def test_timeout_expected_signal_exit_cannot_count_as_success(self): code = "import time; time.sleep(30)" plan = self.make_plan([sys.executable, "-c", code], expected_exit_code=-15) result = self.invoke(plan) self.assertEqual(result.returncode, 1, result.stderr) self.assertEqual(json.loads((self.artifacts / "manifest.json").read_text())["result"], "failure") def test_infrastructure_failure_stops_second_command_and_failure_validates(self): marker = self.repo / "second-ran.txt" commands = [ {"id": "first", "argv": [sys.executable, "-c", "import time; time.sleep(30)"], "timeout_seconds": 0.2, "claims": ["first"]}, {"id": "second", "argv": [sys.executable, "-c", f"from pathlib import Path; Path({str(marker)!r}).write_text('ran')"], "timeout_seconds": 1, "claims": ["second"]}, ] plan = self.make_plan([], commands=commands) result = self.invoke(plan) self.assertEqual(result.returncode, 1, result.stderr) data = json.loads((self.artifacts / "manifest.json").read_text()) self.assertEqual(len(data["commands"]), 1) self.assertFalse(marker.exists()) validated = self.invoke(plan, op="validate", manifest="manifest.json") self.assertEqual(validated.returncode, 0, validated.stderr) self.assertEqual(data["result"], "failure") def test_output_limit_retains_safe_bounded_failure(self): plan = self.make_plan([sys.executable, "-c", "print('safe-' * 10000)"], max_output_bytes=17) result = self.invoke(plan) self.assertEqual(result.returncode, 1, result.stderr) record = json.loads((self.artifacts / "manifest.json").read_text())["commands"][0] self.assertEqual(record["execution_status"], "output_limit") self.assertLessEqual(len(record["output"].encode()), 17) self.assertTrue(record["output"].startswith("safe-")) def test_invalid_utf8_output_is_bounded(self): plan = self.make_plan([sys.executable, "-c", "import sys; sys.stdout.buffer.write(b'\\xff' * 1000)"], timeout=2, max_output_bytes=7) result = self.invoke(plan) self.assertEqual(result.returncode, 1, result.stderr) record = json.loads((self.artifacts / "manifest.json").read_text())["commands"][0] self.assertEqual(record["execution_status"], "output_limit") self.assertLessEqual(len(record["output"].encode()), 7) def test_secret_on_timeout_is_rejected_without_manifest_or_leak(self): secret = "SYNTHETIC_TIMEOUT_SECRET_7f3" env = os.environ.copy(); env["SYNTHETIC_SECRET"] = secret plan = self.make_plan([sys.executable, "-c", "import os,time; print(os.environ['SYNTHETIC_SECRET'],flush=True); time.sleep(30)"], extra={"secret_env_names": ["SYNTHETIC_SECRET"]}) result = self.invoke(plan, env=env) self.assertEqual(result.returncode, 2) self.assertFalse((self.artifacts / "manifest.json").exists()) self.assertNotIn(secret, result.stdout + result.stderr) def test_secret_on_overflow_is_rejected_without_manifest_or_leak(self): secret = "SYNTHETIC_OVERFLOW_SECRET_9c2" env = os.environ.copy(); env["SYNTHETIC_SECRET"] = secret plan = self.make_plan([sys.executable, "-c", "import os; print(os.environ['SYNTHETIC_SECRET'] + 'x'*1000)"], max_output_bytes=16, extra={"secret_env_names": ["SYNTHETIC_SECRET"]}) result = self.invoke(plan, env=env) self.assertEqual(result.returncode, 2) self.assertFalse((self.artifacts / "manifest.json").exists()) self.assertNotIn(secret, result.stdout + result.stderr) def test_timeout_bounds_are_rejected_before_later_command_execution(self): invalids = [None, 0, -1, False, "0.2", float("nan"), float("inf"), 3601, 10**500] for index, invalid in enumerate(invalids): with self.subTest(timeout=invalid): marker = self.repo / f"later-ran-{index}.txt" first = {"id": "bad", "argv": ["true"], "claims": ["bad"]} if invalid is not None: first["timeout_seconds"] = invalid commands = [ first, {"id": "later", "argv": [sys.executable, "-c", f"from pathlib import Path; Path({str(marker)!r}).write_text('ran')"], "timeout_seconds": 1, "claims": ["later"]}, ] result = self.invoke(self.make_plan([], commands=commands), output=f"invalid-{index}.json") self.assertEqual(result.returncode, 2, result.stderr) self.assertFalse(marker.exists()) def test_archived_v2_shape_validates_read_only_but_fresh_v3_rejects(self): plan = self.make_plan(["true"], timeout=1) result = self.invoke(plan) self.assertEqual(result.returncode, 0, result.stderr) manifest = self.artifacts / "manifest.json" obj = json.loads(manifest.read_text()) archived_plan = json.loads(plan.read_text()) archived_plan["commands"][0].pop("timeout_seconds") plan.write_text(json.dumps(archived_plan)); plan.chmod(0o600) obj["schema"] = "evidence-manifest/v2" obj["plan"]["sha256"] = hashlib.sha256(plan.read_bytes()).hexdigest() for key in ("timeout_seconds", "execution_status", "termination"): obj["commands"][0].pop(key, None) obj["claims"]["probe"] = obj["claims"]["probe"][:6] manifest.write_text(json.dumps(obj, sort_keys=True, separators=(",", ":")) + "\n") before = manifest.read_bytes() fresh = self.invoke(plan, op="validate", manifest="manifest.json") self.assertEqual(fresh.returncode, 2) archived = self.invoke(plan, op="validate-archived-v2", manifest="manifest.json") self.assertEqual(archived.returncode, 0, archived.stderr) self.assertEqual(manifest.read_bytes(), before) def test_tampered_completed_status_or_termination_fails_validation(self): plan = self.make_plan(["true"], timeout=1) self.assertEqual(self.invoke(plan).returncode, 0) manifest = self.artifacts / "manifest.json" original = json.loads(manifest.read_text()) for mutation in (lambda obj: obj["commands"][0].__setitem__("execution_status", "timeout"), lambda obj: obj["commands"][0]["termination"].__setitem__("direct_child_reaped", False), lambda obj: obj["commands"][0]["termination"].__setitem__("group_cleanup_completed", False), lambda obj: obj["commands"][0]["termination"].__setitem__("term_sent", True), lambda obj: obj["commands"][0]["termination"].__setitem__("kill_sent", True)): manifest.write_text(json.dumps(original, sort_keys=True, separators=(",", ":")) + "\n") obj = json.loads(manifest.read_text()); mutation(obj) manifest.write_text(json.dumps(obj, sort_keys=True, separators=(",", ":")) + "\n") self.assertEqual(self.invoke(plan, op="validate", manifest="manifest.json").returncode, 2) def test_secret_prefix_after_safe_text_on_overflow_is_rejected(self): secret = "SYNTHETIC_BOUNDARY_SECRET_9c2" env = os.environ.copy(); env["SYNTHETIC_SECRET"] = secret plan = self.make_plan([sys.executable, "-c", "import os; print('safe-banner:' + os.environ['SYNTHETIC_SECRET'] + 'x'*1000)"], max_output_bytes=20, extra={"secret_env_names": ["SYNTHETIC_SECRET"]}) result = self.invoke(plan, env=env) self.assertEqual(result.returncode, 2, result.stderr) self.assertFalse((self.artifacts / "manifest.json").exists()) self.assertNotIn(secret[:8], result.stdout + result.stderr) def test_megabyte_invalid_utf8_processing_remains_bounded(self): plan = self.make_plan([sys.executable, "-c", "import sys; sys.stdout.buffer.write(b'\\xff' * (2**21))"], timeout=2, max_output_bytes=2**20) result = self.invoke(plan) self.assertEqual(result.returncode, 1, result.stderr) record = json.loads((self.artifacts / "manifest.json").read_text())["commands"][0] self.assertEqual(record["execution_status"], "output_limit") self.assertLessEqual(len(record["output"].encode()), 2**20) def test_silent_command_timeout_is_bounded_and_persists_failure(self): plan = self.make_plan([sys.executable, "-c", "import time; time.sleep(30)"]) started = time.monotonic() result = self.invoke(plan) elapsed = time.monotonic() - started self.assertEqual(result.returncode, 1, result.stderr) self.assertLess(elapsed, 6.0) record = json.loads((self.artifacts / "manifest.json").read_text())["commands"][0] self.assertEqual(record["execution_status"], "timeout") self.assertTrue(record["termination"]["direct_child_reaped"]) self.assertLessEqual(len(record["output"].encode()), record["max_output_bytes"]) def test_partial_output_survives_timeout_and_is_utf8_bounded(self): plan = self.make_plan([sys.executable, "-c", "import sys,time; sys.stdout.write('prefix\\n'); sys.stdout.flush(); time.sleep(30)"], timeout=1.5, max_output_bytes=64) result = self.invoke(plan) self.assertEqual(result.returncode, 1, result.stderr) record = json.loads((self.artifacts / "manifest.json").read_text())["commands"][0] self.assertEqual(record["output"], "prefix\n") self.assertLessEqual(len(record["output"].encode("utf-8")), 64) if __name__ == "__main__": unittest.main()