| 1 | """Real subprocess tests use only synthetic Python children, never model CLIs."""
|
| 2 | from dataclasses import replace
|
| 3 | import json
|
| 4 | import os
|
| 5 | from pathlib import Path
|
| 6 | import signal
|
| 7 | import subprocess
|
| 8 | import sys
|
| 9 | import tempfile
|
| 10 | import time
|
| 11 | from types import SimpleNamespace
|
| 12 | import unittest
|
| 13 | from unittest import mock
|
| 14 |
|
| 15 | sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
|
| 16 | import agent_workflow as aw
|
| 17 | import agent_metrics_schema
|
| 18 | import workflow_ab
|
| 19 | from oakbench.agent_execution import run_captured
|
| 20 | from oakbench.environment import base_env
|
| 21 |
|
| 22 |
|
| 23 | class CaptureLimitsTest(unittest.TestCase):
|
| 24 | def setUp(self):
|
| 25 | self.temp = tempfile.TemporaryDirectory()
|
| 26 | self.root = Path(self.temp.name)
|
| 27 | self.addCleanup(self.temp.cleanup)
|
| 28 | self.stdout = self.root / "stdout"
|
| 29 | self.stderr = self.root / "stderr"
|
| 30 |
|
| 31 | def run_script(self, script, timeout=3, limit=4096):
|
| 32 | return run_captured([sys.executable, "-c", script], self.root, base_env(), None,
|
| 33 | self.stdout, self.stderr, timeout, limit)
|
| 34 |
|
| 35 | def test_exact_raw_byte_counts_and_no_stdin_prompt(self):
|
| 36 | result = self.run_script("import sys; assert sys.stdin.read() == ''; sys.stdout.buffer.write(bytes([255,13,10])); sys.stderr.buffer.write(b'err')")
|
| 37 | self.assertEqual((result.stdout_observed_bytes, result.stderr_observed_bytes), (3, 3))
|
| 38 | self.assertEqual(self.stdout.read_bytes(), b"\xff\r\n")
|
| 39 | self.assertEqual(result.returncode, 0)
|
| 40 | self.assertTrue(result.capture_complete)
|
| 41 |
|
| 42 | def test_combined_output_limit_bounds_disk_and_stops_child(self):
|
| 43 | result = self.run_script("import os\nwhile True:\n os.write(1,b'x'*65536)\n os.write(2,b'y'*65536)", limit=1024)
|
| 44 | self.assertTrue(result.output_limited)
|
| 45 | self.assertEqual(result.returncode, 125)
|
| 46 | self.assertLessEqual(self.stdout.stat().st_size + self.stderr.stat().st_size, 1024)
|
| 47 | self.assertGreater(result.stdout_observed_bytes + result.stderr_observed_bytes, 1024)
|
| 48 |
|
| 49 | def test_large_stdin_is_fully_delivered_without_blocking_deadline(self):
|
| 50 | payload = "p" * 200000
|
| 51 | result = run_captured([sys.executable, "-c", "import sys; print(len(sys.stdin.read()))"],
|
| 52 | self.root, base_env(), payload, self.stdout, self.stderr, 3, 4096)
|
| 53 | self.assertEqual(result.returncode, 0)
|
| 54 | self.assertEqual(self.stdout.read_text().strip(), str(len(payload)))
|
| 55 |
|
| 56 | def test_invalid_limits_and_launcher_errors(self):
|
| 57 | for timeout in (0, -1, float("nan"), float("inf")):
|
| 58 | with self.assertRaises(ValueError):
|
| 59 | self.run_script("", timeout=timeout)
|
| 60 | for limit in (0, -1, True):
|
| 61 | with self.assertRaises(ValueError):
|
| 62 | self.run_script("", limit=limit)
|
| 63 | with self.assertRaises(FileNotFoundError):
|
| 64 | run_captured(["/missing/synthetic-agent"], self.root, base_env(), None,
|
| 65 | self.stdout, self.stderr, 1, 10)
|
| 66 | # Failure closes both files; they remain readable/removable.
|
| 67 | self.stdout.unlink(); self.stderr.unlink()
|
| 68 |
|
| 69 | @unittest.skipUnless(os.name == "posix", "POSIX process groups required")
|
| 70 | def test_timeout_kills_grandchild_ignoring_term(self):
|
| 71 | marker = self.root / "late-write"
|
| 72 | child = f"import signal,time,pathlib; signal.signal(signal.SIGTERM,signal.SIG_IGN); time.sleep(2); pathlib.Path({str(marker)!r}).write_text('escaped'); time.sleep(30)"
|
| 73 | parent = f"import subprocess,sys,time; subprocess.Popen([sys.executable,'-c',{child!r}]); time.sleep(30)"
|
| 74 | result = self.run_script(parent, timeout=0.15)
|
| 75 | self.assertTrue(result.timed_out)
|
| 76 | self.assertEqual(result.returncode, 124)
|
| 77 | time.sleep(0.2)
|
| 78 | self.assertFalse(marker.exists())
|
| 79 | self.assertTrue(result.capture_complete)
|
| 80 |
|
| 81 | @unittest.skipUnless(os.name == "posix", "POSIX process groups required")
|
| 82 | def test_exit_zero_parent_still_cleans_background_tools(self):
|
| 83 | marker = self.root / "late-write"
|
| 84 | child = f"import time,pathlib; time.sleep(.6); pathlib.Path({str(marker)!r}).write_text('escaped')"
|
| 85 | parent = f"import subprocess,sys; subprocess.Popen([sys.executable,'-c',{child!r}]); print('done')"
|
| 86 | result = self.run_script(parent)
|
| 87 | self.assertEqual(result.returncode, 0)
|
| 88 | time.sleep(0.7)
|
| 89 | self.assertFalse(marker.exists())
|
| 90 | self.assertTrue(result.capture_complete)
|
| 91 |
|
| 92 | @unittest.skipUnless(os.name == "posix", "POSIX process groups required")
|
| 93 | def test_keyboard_interrupt_cleans_owned_group(self):
|
| 94 | marker = self.root / "interrupt-write"
|
| 95 | real_popen = subprocess.Popen
|
| 96 | def interrupted_popen(*args, **kwargs):
|
| 97 | process = real_popen(*args, **kwargs)
|
| 98 | real_wait = process.wait
|
| 99 | called = False
|
| 100 | def interrupt_once(*args, **kwargs):
|
| 101 | nonlocal called
|
| 102 | if not called:
|
| 103 | called = True
|
| 104 | raise KeyboardInterrupt
|
| 105 | return real_wait(*args, **kwargs)
|
| 106 | process.wait = interrupt_once
|
| 107 | return process
|
| 108 | with mock.patch("oakbench.agent_execution.subprocess.Popen", side_effect=interrupted_popen):
|
| 109 | with self.assertRaises(KeyboardInterrupt):
|
| 110 | self.run_script(f"import time,pathlib; time.sleep(.4); pathlib.Path({str(marker)!r}).write_text('escaped')")
|
| 111 | time.sleep(.5)
|
| 112 | self.assertFalse(marker.exists())
|
| 113 |
|
| 114 | @unittest.skipUnless(os.name == "posix", "POSIX detached sessions required")
|
| 115 | def test_escaped_session_does_not_hang_capture_and_is_reported_incomplete(self):
|
| 116 | pidfile = self.root / "escaped.pid"
|
| 117 | child = f"import os,pathlib,time; pathlib.Path({str(pidfile)!r}).write_text(str(os.getpid())); time.sleep(30)"
|
| 118 | parent = (f"import subprocess,sys,time,pathlib; subprocess.Popen([sys.executable,'-c',{child!r}],start_new_session=True); "
|
| 119 | f"p=pathlib.Path({str(pidfile)!r})\nwhile not p.exists(): time.sleep(.01)")
|
| 120 | started = time.monotonic()
|
| 121 | try:
|
| 122 | result = self.run_script(parent)
|
| 123 | self.assertLess(time.monotonic() - started, 4)
|
| 124 | self.assertFalse(result.capture_complete)
|
| 125 | self.assertEqual(result.returncode, 125)
|
| 126 | finally:
|
| 127 | if pidfile.exists():
|
| 128 | try:
|
| 129 | os.kill(int(pidfile.read_text()), signal.SIGKILL)
|
| 130 | except ProcessLookupError:
|
| 131 | pass
|
| 132 |
|
| 133 |
|
| 134 | class TrialMetadataTest(unittest.TestCase):
|
| 135 | def test_raised_capture_limit_preserves_complete_usage_beyond_default(self):
|
| 136 | with tempfile.TemporaryDirectory() as tmp, mock.patch.object(aw, "DEFAULT_OUTPUT_LIMIT_BYTES", 64):
|
| 137 | root = Path(tmp)
|
| 138 | stdout, stderr, prompt = [root / name for name in ("stdout", "stderr", "prompt")]
|
| 139 | stdout.write_text("\n" * 80 + '{"type":"turn.completed","usage":{"input_tokens":10,"output_tokens":2}}\n')
|
| 140 | stderr.write_text(""); prompt.write_text("task")
|
| 141 | execution = aw.AgentExecution(0, False, 1, stdout, stderr, prompt, capture_metadata={"output_limit_bytes": 1024})
|
| 142 | oracle = SimpleNamespace(elapsed_ms=1, vcs_ms=1, test_ms=1, success=True, checks={}, failure_reason=None,
|
| 143 | changed_files=[], test_returncode=0, test_output_bytes=0, dirty_files_at_end=0, commit_delta=1,
|
| 144 | bytes_written=0, artifact_paths=[])
|
| 145 | row = aw.make_row("synthetic", "2026-09-10T00:00:00Z", "test",
|
| 146 | aw.AgentConfig("synthetic", "codex_exec_json", "test", Path(sys.executable), "test", "test"),
|
| 147 | workflow_ab.Subject("git", "git", "Git", Path("/git")), workflow_ab.WORKFLOWS["bugfix_test_loop"],
|
| 148 | "agent-default", 0, root, root, 0, 0, execution, oracle, "task", {}, {}, {}, "minimal", "full-docs")
|
| 149 | self.assertFalse(row["output_truncated"])
|
| 150 | self.assertEqual(row["token_metrics"]["total_tokens_reported"], 12)
|
| 151 | self.assertEqual(row["workflow_integrity_evidence_class"], "tests_and_subject_status")
|
| 152 |
|
| 153 | def test_fresh_home_sanitizes_git_but_preserves_credentials(self):
|
| 154 | with tempfile.TemporaryDirectory() as tmp, mock.patch.dict(os.environ, {
|
| 155 | "GIT_DIR": "/not-the-fixture", "GIT_WORK_TREE": "/wrong-tree",
|
| 156 | "GIT_CONFIG_COUNT": "1", "OPENAI_API_KEY": "synthetic-token", "CODEX_HOME": "/synthetic-auth-home",
|
| 157 | "OAK_ADMIN_API_KEY": "must-not-leak", "CLAUDE_CODE_ARBITRARY": "must-not-leak",
|
| 158 | "UNRELATED_SERVICE_KEY": "must-not-leak",
|
| 159 | }):
|
| 160 | first, first_meta = aw.trial_agent_env(Path(tmp))
|
| 161 | second, _ = aw.trial_agent_env(Path(tmp))
|
| 162 | self.assertNotEqual(first["HOME"], second["HOME"])
|
| 163 | self.assertNotIn("GIT_DIR", first)
|
| 164 | self.assertNotIn("GIT_WORK_TREE", first)
|
| 165 | self.assertNotIn("GIT_CONFIG_COUNT", first)
|
| 166 | self.assertEqual(first["OPENAI_API_KEY"], "synthetic-token")
|
| 167 | self.assertEqual(first["CODEX_HOME"], "/synthetic-auth-home")
|
| 168 | self.assertNotIn("OAK_ADMIN_API_KEY", first)
|
| 169 | self.assertNotIn("CLAUDE_CODE_ARBITRARY", first)
|
| 170 | self.assertNotIn("UNRELATED_SERVICE_KEY", first)
|
| 171 | self.assertIn("CODEX_HOME", first_meta["preserved_config_override_names"])
|
| 172 | self.assertFalse(first_meta["os_sandbox_enforced"])
|
| 173 |
|
| 174 | def test_missing_oracle_preserves_observed_agent_failures(self):
|
| 175 | with tempfile.TemporaryDirectory() as tmp:
|
| 176 | root = Path(tmp)
|
| 177 | stdout, stderr, prompt = [root / name for name in ("stdout", "stderr", "prompt")]
|
| 178 | for path in (stdout, stderr, prompt): path.write_text("")
|
| 179 | oracle = SimpleNamespace(elapsed_ms=1, vcs_ms=1, test_ms=1, success=False,
|
| 180 | checks={"required_checks_passed": False, "final_vcs_clean": True}, failure_reason="required_checks_passed",
|
| 181 | changed_files=[], test_returncode=77, test_output_bytes=0, dirty_files_at_end=0, commit_delta=1,
|
| 182 | bytes_written=0, artifact_paths=[], evidence={"oracle_version": aw.ORACLE_VERSION,
|
| 183 | "workflow_integrity_source": "trusted_oracle_instrument_unavailable", "workflow_integrity_passed": None,
|
| 184 | "workflow_integrity_unmeasured_reason": "missing", "workflow_integrity_failure_reason": None})
|
| 185 | for code, timeout, expected in ((0, False, 77), (3, False, 1), (124, True, 1)):
|
| 186 | execution = aw.AgentExecution(code, timeout, 1, stdout, stderr, prompt)
|
| 187 | row = aw.make_row("synthetic", "2026-09-10T00:00:00Z", "test",
|
| 188 | aw.AgentConfig("synthetic", "codex_exec_json", "test", Path(sys.executable), "test", "test"),
|
| 189 | workflow_ab.Subject("git", "git", "Git", Path("/git")), workflow_ab.WORKFLOWS["wide_config_refactor"],
|
| 190 | "agent-default", 0, root, root, 0, 0, execution, oracle, "task", {}, {}, {}, "minimal", "full-docs")
|
| 191 | self.assertEqual(row["returncode"], expected)
|
| 192 | from oakbench.rows import validate_rows
|
| 193 | from oakbench.results import ResultsStore
|
| 194 | self.assertEqual(validate_rows([row], "agent"), [])
|
| 195 | ResultsStore(root / "results", "agent").write("producer", [row])
|
| 196 | self.assertEqual(row["skip_reason"], "missing" if expected == 77 else None)
|
| 197 | self.assertIsNone(row["test_metrics"]["passed_commands"])
|
| 198 | self.assertIsNone(row["test_metrics"]["failed_commands"])
|
| 199 | self.assertEqual(agent_metrics_schema.validate_row(row), [])
|
| 200 | self.assertEqual(row["env_isolation_version"], aw.ENV_ISOLATION_VERSION)
|
| 201 | if timeout: self.assertEqual(row["outcome"], "timeout")
|
| 202 |
|
| 203 | def test_partial_usage_is_null_and_output_limit_never_passes(self):
|
| 204 | with tempfile.TemporaryDirectory() as tmp:
|
| 205 | root = Path(tmp)
|
| 206 | stdout, stderr, prompt = [root / name for name in ("stdout", "stderr", "prompt")]
|
| 207 | stdout.write_bytes(b'{"type":"turn.completed","usage":{"input_tokens":10,"output_tokens":2}}\n\xff')
|
| 208 | stderr.write_bytes(b""); prompt.write_text("task")
|
| 209 | execution = aw.AgentExecution(0, False, 1, stdout, stderr, prompt)
|
| 210 | oracle = SimpleNamespace(elapsed_ms=1, vcs_ms=1, test_ms=1, success=True,
|
| 211 | checks={}, failure_reason=None, changed_files=[], test_returncode=0,
|
| 212 | test_output_bytes=0, dirty_files_at_end=0, commit_delta=1,
|
| 213 | bytes_written=0, artifact_paths=[])
|
| 214 | for partial in (replace(execution, output_limited=True), replace(execution, capture_complete=False),
|
| 215 | replace(execution, returncode=3)):
|
| 216 | row = aw.make_row("synthetic", "2026-09-10T00:00:00Z", "test",
|
| 217 | aw.AgentConfig("synthetic", "codex_exec_json", "test", Path(sys.executable), "test", "test"),
|
| 218 | workflow_ab.Subject("git", "git", "Git", Path("/git")),
|
| 219 | workflow_ab.WORKFLOWS["bugfix_test_loop"], "agent-default", 0, root, root,
|
| 220 | 0, 0, partial, oracle, "task", {}, {}, {}, "minimal", "full-docs")
|
| 221 | self.assertFalse(row["success"])
|
| 222 | self.assertEqual(row["outcome"], "fail" if partial.returncode == 3 else "error")
|
| 223 | self.assertEqual(row["output_truncated"], partial.returncode != 3)
|
| 224 | self.assertEqual(row["stdout_bytes"], stdout.stat().st_size)
|
| 225 | self.assertIsNone(row["token_metrics"]["total_tokens_reported"])
|
| 226 | self.assertIsNone(row["token_metrics"]["cost_weighted_total"])
|
| 227 | self.assertFalse(row["source"]["agent_execution"]["usage_complete"])
|
| 228 | self.assertFalse(row["source"]["agent_execution"]["transcript_metrics_complete"])
|
| 229 | self.assertIsNone(row["turn_metrics"]["assistant_turns_total"])
|
| 230 | self.assertEqual(row["turn_metrics"]["turn_source"], "unmeasured")
|
| 231 | self.assertEqual(agent_metrics_schema.validate_row(row), [])
|
| 232 |
|
| 233 |
|
| 234 | if __name__ == "__main__":
|
| 235 | unittest.main()
|