"""Bounded real-agent capture and owned-process-group cleanup.

This controls one POSIX process group, not an OS sandbox. Descendants that

create independent sessions can escape the group; non-POSIX cleanup only owns

the direct child. Never describe this helper as process-tree containment.

"""

from __future__ import annotations

from dataclasses import dataclass

import math

import os

from pathlib import Path

import subprocess

import threading

import time

from oakbench.execution import stop_owned_process

EXECUTION_VERSION = "owned-group-bounded-output-v1"

DEFAULT_OUTPUT_LIMIT_BYTES = 8 * 1024 * 1024

@dataclass(frozen=True)

class CapturedProcess:

    returncode: int

    timed_out: bool

    output_limited: bool

    elapsed_ms: float

    stdout_observed_bytes: int

    stderr_observed_bytes: int

    stdout_retained_bytes: int

    stderr_retained_bytes: int

    capture_complete: bool

def run_captured(command: list[str], cwd: Path, env: dict[str, str],

                 stdin_text: str | None, stdout_path: Path, stderr_path: Path,

                 timeout_seconds: float,

                 output_limit_bytes: int = DEFAULT_OUTPUT_LIMIT_BYTES) -> CapturedProcess:

    if not math.isfinite(timeout_seconds) or timeout_seconds <= 0:

        raise ValueError("timeout_seconds must be positive and finite")

    if isinstance(output_limit_bytes, bool) or not isinstance(output_limit_bytes, int) or output_limit_bytes < 1:

        raise ValueError("output_limit_bytes must be a positive integer")

    started = time.perf_counter()

    counts = {"stdout": 0, "stderr": 0}

    retained = {"stdout": 0, "stderr": 0}

    count_lock = threading.Lock()

    exceeded = threading.Event()

    capture_errors = []

    readers = []

    writers = []

    proc = None

    timed_out = False

    forced_pipe_closure = False

    with stdout_path.open("wb") as stdout_file, stderr_path.open("wb") as stderr_file:

        try:

            proc = subprocess.Popen(command, cwd=cwd, env=env,

                                    stdin=subprocess.PIPE if stdin_text is not None else subprocess.DEVNULL,

                                    stdout=subprocess.PIPE, stderr=subprocess.PIPE,

                                    bufsize=0, start_new_session=os.name == "posix")

            def drain(stream, destination, name):

                try:

                    while True:

                        data = stream.read(65536)

                        if not data:

                            break

                        with count_lock:

                            counts[name] += len(data)

                            allowance = max(0, output_limit_bytes - sum(retained.values()))

                            admitted = data[:allowance]

                            destination.write(admitted)

                            retained[name] += len(admitted)

                            if len(admitted) != len(data):

                                exceeded.set()

                except (OSError, ValueError) as exc:

                    capture_errors.append(str(exc))

                finally:

                    stream.close()

            for name, stream, destination in (("stdout", proc.stdout, stdout_file),

                                              ("stderr", proc.stderr, stderr_file)):

                thread = threading.Thread(target=drain, args=(stream, destination, name), daemon=True)

                readers.append(thread)

                thread.start()

            if stdin_text is not None:

                def supply_input():

                    try:

                        pending = memoryview(stdin_text.encode("utf-8"))

                        while pending:

                            written = proc.stdin.write(pending)

                            if not written:

                                break

                            pending = pending[written:]

                    except (BrokenPipeError, OSError, ValueError):

                        pass

                    finally:

                        proc.stdin.close()

                writer = threading.Thread(target=supply_input, daemon=True)

                writers.append(writer)

                writer.start()

            deadline = started + timeout_seconds

            while proc.poll() is None:

                if exceeded.is_set():

                    break

                remaining = deadline - time.perf_counter()

                if remaining <= 0:

                    timed_out = True

                    break

                try:

                    proc.wait(timeout=min(0.025, remaining))

                except subprocess.TimeoutExpired:

                    pass

        finally:

            # Even an exit-zero CLI may leave tools running or pipe handles

            # open. Stop the owned group before oracle inspection on every path,

            # including KeyboardInterrupt and reader/launcher exceptions.

            if proc is not None:

                stop_owned_process(proc)

            for thread in readers + writers:

                thread.join(timeout=1.0)

            forced_pipe_closure = any(thread.is_alive() for thread in readers)

            if proc is not None:

                for stream in (proc.stdout, proc.stderr, proc.stdin):

                    if stream is not None and not stream.closed:

                        stream.close()

    complete = not capture_errors and not forced_pipe_closure

    output_limited = exceeded.is_set()

    returncode = 124 if timed_out else (125 if output_limited or not complete else proc.returncode)

    return CapturedProcess(returncode, timed_out, output_limited,

                           (time.perf_counter() - started) * 1000,

                           counts["stdout"], counts["stderr"],

                           retained["stdout"], retained["stderr"], complete)