"""Real subprocess tests use only synthetic Python children, never model CLIs."""

from dataclasses import replace

import json

import os

from pathlib import Path

import signal

import subprocess

import sys

import tempfile

import time

from types import SimpleNamespace

import unittest

from unittest import mock

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))

import agent_workflow as aw

import agent_metrics_schema

import workflow_ab

from oakbench.agent_execution import run_captured

from oakbench.environment import base_env

class CaptureLimitsTest(unittest.TestCase):

    def setUp(self):

        self.temp = tempfile.TemporaryDirectory()

        self.root = Path(self.temp.name)

        self.addCleanup(self.temp.cleanup)

        self.stdout = self.root / "stdout"

        self.stderr = self.root / "stderr"

    def run_script(self, script, timeout=3, limit=4096):

        return run_captured([sys.executable, "-c", script], self.root, base_env(), None,

                            self.stdout, self.stderr, timeout, limit)

    def test_exact_raw_byte_counts_and_no_stdin_prompt(self):

        result = self.run_script("import sys; assert sys.stdin.read() == ''; sys.stdout.buffer.write(bytes([255,13,10])); sys.stderr.buffer.write(b'err')")

        self.assertEqual((result.stdout_observed_bytes, result.stderr_observed_bytes), (3, 3))

        self.assertEqual(self.stdout.read_bytes(), b"\xff\r\n")

        self.assertEqual(result.returncode, 0)

        self.assertTrue(result.capture_complete)

    def test_combined_output_limit_bounds_disk_and_stops_child(self):

        result = self.run_script("import os\nwhile True:\n os.write(1,b'x'*65536)\n os.write(2,b'y'*65536)", limit=1024)

        self.assertTrue(result.output_limited)

        self.assertEqual(result.returncode, 125)

        self.assertLessEqual(self.stdout.stat().st_size + self.stderr.stat().st_size, 1024)

        self.assertGreater(result.stdout_observed_bytes + result.stderr_observed_bytes, 1024)

    def test_large_stdin_is_fully_delivered_without_blocking_deadline(self):

        payload = "p" * 200000

        result = run_captured([sys.executable, "-c", "import sys; print(len(sys.stdin.read()))"],

                    self.root, base_env(), payload, self.stdout, self.stderr, 3, 4096)

        self.assertEqual(result.returncode, 0)

        self.assertEqual(self.stdout.read_text().strip(), str(len(payload)))

    def test_invalid_limits_and_launcher_errors(self):

        for timeout in (0, -1, float("nan"), float("inf")):

            with self.assertRaises(ValueError):

                self.run_script("", timeout=timeout)

        for limit in (0, -1, True):

            with self.assertRaises(ValueError):

                self.run_script("", limit=limit)

        with self.assertRaises(FileNotFoundError):

            run_captured(["/missing/synthetic-agent"], self.root, base_env(), None,

                         self.stdout, self.stderr, 1, 10)

        # Failure closes both files; they remain readable/removable.

        self.stdout.unlink(); self.stderr.unlink()

    @unittest.skipUnless(os.name == "posix", "POSIX process groups required")

    def test_timeout_kills_grandchild_ignoring_term(self):

        marker = self.root / "late-write"

        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)"

        parent = f"import subprocess,sys,time; subprocess.Popen([sys.executable,'-c',{child!r}]); time.sleep(30)"

        result = self.run_script(parent, timeout=0.15)

        self.assertTrue(result.timed_out)

        self.assertEqual(result.returncode, 124)

        time.sleep(0.2)

        self.assertFalse(marker.exists())

        self.assertTrue(result.capture_complete)

    @unittest.skipUnless(os.name == "posix", "POSIX process groups required")

    def test_exit_zero_parent_still_cleans_background_tools(self):

        marker = self.root / "late-write"

        child = f"import time,pathlib; time.sleep(.6); pathlib.Path({str(marker)!r}).write_text('escaped')"

        parent = f"import subprocess,sys; subprocess.Popen([sys.executable,'-c',{child!r}]); print('done')"

        result = self.run_script(parent)

        self.assertEqual(result.returncode, 0)

        time.sleep(0.7)

        self.assertFalse(marker.exists())

        self.assertTrue(result.capture_complete)

    @unittest.skipUnless(os.name == "posix", "POSIX process groups required")

    def test_keyboard_interrupt_cleans_owned_group(self):

        marker = self.root / "interrupt-write"

        real_popen = subprocess.Popen

        def interrupted_popen(*args, **kwargs):

            process = real_popen(*args, **kwargs)

            real_wait = process.wait

            called = False

            def interrupt_once(*args, **kwargs):

                nonlocal called

                if not called:

                    called = True

                    raise KeyboardInterrupt

                return real_wait(*args, **kwargs)

            process.wait = interrupt_once

            return process

        with mock.patch("oakbench.agent_execution.subprocess.Popen", side_effect=interrupted_popen):

            with self.assertRaises(KeyboardInterrupt):

                self.run_script(f"import time,pathlib; time.sleep(.4); pathlib.Path({str(marker)!r}).write_text('escaped')")

        time.sleep(.5)

        self.assertFalse(marker.exists())

    @unittest.skipUnless(os.name == "posix", "POSIX detached sessions required")

    def test_escaped_session_does_not_hang_capture_and_is_reported_incomplete(self):

        pidfile = self.root / "escaped.pid"

        child = f"import os,pathlib,time; pathlib.Path({str(pidfile)!r}).write_text(str(os.getpid())); time.sleep(30)"

        parent = (f"import subprocess,sys,time,pathlib; subprocess.Popen([sys.executable,'-c',{child!r}],start_new_session=True); "

                  f"p=pathlib.Path({str(pidfile)!r})\nwhile not p.exists(): time.sleep(.01)")

        started = time.monotonic()

        try:

            result = self.run_script(parent)

            self.assertLess(time.monotonic() - started, 4)

            self.assertFalse(result.capture_complete)

            self.assertEqual(result.returncode, 125)

        finally:

            if pidfile.exists():

                try:

                    os.kill(int(pidfile.read_text()), signal.SIGKILL)

                except ProcessLookupError:

                    pass

class TrialMetadataTest(unittest.TestCase):

    def test_raised_capture_limit_preserves_complete_usage_beyond_default(self):

        with tempfile.TemporaryDirectory() as tmp, mock.patch.object(aw, "DEFAULT_OUTPUT_LIMIT_BYTES", 64):

            root = Path(tmp)

            stdout, stderr, prompt = [root / name for name in ("stdout", "stderr", "prompt")]

            stdout.write_text("\n" * 80 + '{"type":"turn.completed","usage":{"input_tokens":10,"output_tokens":2}}\n')

            stderr.write_text(""); prompt.write_text("task")

            execution = aw.AgentExecution(0, False, 1, stdout, stderr, prompt, capture_metadata={"output_limit_bytes": 1024})

            oracle = SimpleNamespace(elapsed_ms=1, vcs_ms=1, test_ms=1, success=True, checks={}, failure_reason=None,

                     changed_files=[], test_returncode=0, test_output_bytes=0, dirty_files_at_end=0, commit_delta=1,

                     bytes_written=0, artifact_paths=[])

            row = aw.make_row("synthetic", "2026-09-10T00:00:00Z", "test",

                aw.AgentConfig("synthetic", "codex_exec_json", "test", Path(sys.executable), "test", "test"),

                workflow_ab.Subject("git", "git", "Git", Path("/git")), workflow_ab.WORKFLOWS["bugfix_test_loop"],

                "agent-default", 0, root, root, 0, 0, execution, oracle, "task", {}, {}, {}, "minimal", "full-docs")

            self.assertFalse(row["output_truncated"])

            self.assertEqual(row["token_metrics"]["total_tokens_reported"], 12)

    def test_fresh_home_sanitizes_git_but_preserves_credentials(self):

        with tempfile.TemporaryDirectory() as tmp, mock.patch.dict(os.environ, {

            "GIT_DIR": "/not-the-fixture", "GIT_WORK_TREE": "/wrong-tree",

            "GIT_CONFIG_COUNT": "1", "OPENAI_API_KEY": "synthetic-token", "CODEX_HOME": "/synthetic-auth-home",

            "OAK_ADMIN_API_KEY": "must-not-leak", "CLAUDE_CODE_ARBITRARY": "must-not-leak",

            "UNRELATED_SERVICE_KEY": "must-not-leak",

        }):

            first, first_meta = aw.trial_agent_env(Path(tmp))

            second, _ = aw.trial_agent_env(Path(tmp))

            self.assertNotEqual(first["HOME"], second["HOME"])

            self.assertNotIn("GIT_DIR", first)

            self.assertNotIn("GIT_WORK_TREE", first)

            self.assertNotIn("GIT_CONFIG_COUNT", first)

            self.assertEqual(first["OPENAI_API_KEY"], "synthetic-token")

            self.assertEqual(first["CODEX_HOME"], "/synthetic-auth-home")

            self.assertNotIn("OAK_ADMIN_API_KEY", first)

            self.assertNotIn("CLAUDE_CODE_ARBITRARY", first)

            self.assertNotIn("UNRELATED_SERVICE_KEY", first)

            self.assertIn("CODEX_HOME", first_meta["preserved_config_override_names"])

            self.assertFalse(first_meta["os_sandbox_enforced"])

    def test_missing_oracle_preserves_observed_agent_failures(self):

        with tempfile.TemporaryDirectory() as tmp:

            root = Path(tmp)

            stdout, stderr, prompt = [root / name for name in ("stdout", "stderr", "prompt")]

            for path in (stdout, stderr, prompt): path.write_text("")

            oracle = SimpleNamespace(elapsed_ms=1, vcs_ms=1, test_ms=1, success=False,

                checks={"required_checks_passed": False, "final_vcs_clean": True}, failure_reason="required_checks_passed",

                changed_files=[], test_returncode=77, test_output_bytes=0, dirty_files_at_end=0, commit_delta=1,

                bytes_written=0, artifact_paths=[], evidence={"oracle_version": aw.ORACLE_VERSION,

                    "workflow_integrity_source": "trusted_oracle_instrument_unavailable", "workflow_integrity_passed": None,

                    "workflow_integrity_unmeasured_reason": "missing", "workflow_integrity_failure_reason": None})

            for code, timeout, expected in ((0, False, 77), (3, False, 1), (124, True, 1)):

                execution = aw.AgentExecution(code, timeout, 1, stdout, stderr, prompt)

                row = aw.make_row("synthetic", "2026-09-10T00:00:00Z", "test",

                    aw.AgentConfig("synthetic", "codex_exec_json", "test", Path(sys.executable), "test", "test"),

                    workflow_ab.Subject("git", "git", "Git", Path("/git")), workflow_ab.WORKFLOWS["wide_config_refactor"],

                    "agent-default", 0, root, root, 0, 0, execution, oracle, "task", {}, {}, {}, "minimal", "full-docs")

                self.assertEqual(row["returncode"], expected)

                from oakbench.rows import validate_rows

                from oakbench.results import ResultsStore

                self.assertEqual(validate_rows([row], "agent"), [])

                ResultsStore(root / "results", "agent").write("producer", [row])

                self.assertEqual(row["skip_reason"], "missing" if expected == 77 else None)

                self.assertEqual(agent_metrics_schema.validate_row(row), [])

                self.assertEqual(row["env_isolation_version"], aw.ENV_ISOLATION_VERSION)

                if timeout: self.assertEqual(row["outcome"], "timeout")

    def test_partial_usage_is_null_and_output_limit_never_passes(self):

        with tempfile.TemporaryDirectory() as tmp:

            root = Path(tmp)

            stdout, stderr, prompt = [root / name for name in ("stdout", "stderr", "prompt")]

            stdout.write_bytes(b'{"type":"turn.completed","usage":{"input_tokens":10,"output_tokens":2}}\n\xff')

            stderr.write_bytes(b""); prompt.write_text("task")

            execution = aw.AgentExecution(0, False, 1, stdout, stderr, prompt)

            oracle = SimpleNamespace(elapsed_ms=1, vcs_ms=1, test_ms=1, success=True,

                     checks={}, failure_reason=None, changed_files=[], test_returncode=0,

                     test_output_bytes=0, dirty_files_at_end=0, commit_delta=1,

                     bytes_written=0, artifact_paths=[])

            for partial in (replace(execution, output_limited=True), replace(execution, capture_complete=False),

                            replace(execution, returncode=3)):

                row = aw.make_row("synthetic", "2026-09-10T00:00:00Z", "test",

                    aw.AgentConfig("synthetic", "codex_exec_json", "test", Path(sys.executable), "test", "test"),

                    workflow_ab.Subject("git", "git", "Git", Path("/git")),

                    workflow_ab.WORKFLOWS["bugfix_test_loop"], "agent-default", 0, root, root,

                    0, 0, partial, oracle, "task", {}, {}, {}, "minimal", "full-docs")

                self.assertFalse(row["success"])

                self.assertEqual(row["outcome"], "fail" if partial.returncode == 3 else "error")

                self.assertEqual(row["output_truncated"], partial.returncode != 3)

                self.assertEqual(row["stdout_bytes"], stdout.stat().st_size)

                self.assertIsNone(row["token_metrics"]["total_tokens_reported"])

                self.assertIsNone(row["token_metrics"]["cost_weighted_total"])

                self.assertFalse(row["source"]["agent_execution"]["usage_complete"])

                self.assertFalse(row["source"]["agent_execution"]["transcript_metrics_complete"])

                self.assertIsNone(row["turn_metrics"]["assistant_turns_total"])

                self.assertEqual(row["turn_metrics"]["turn_source"], "unmeasured")

                self.assertEqual(agent_metrics_schema.validate_row(row), [])

if __name__ == "__main__":

    unittest.main()