"""Bounded disposable-process observation; never used by production retirement."""
from dataclasses import dataclass
import os
import select
import signal
import subprocess
import sys
import time


@dataclass
class Result:
    returncode: int
    stdout: bytes
    stderr: bytes
    discarded: int


class FixtureTimeout(TimeoutError):
    def __init__(self, pid, output, errors, discarded, reaped):
        self.pid, self.stdout, self.stderr = pid, output, errors
        self.discarded, self.reaped = discarded, reaped
        super().__init__(f'fixture deadline: pid={pid} reaped={reaped} discarded={discarded} '
                         f'stdout={output!r} stderr={errors!r}')


def run_bounded(args, *, seconds=5, observe=None, **kwargs):
    """One deadline includes sampling, simultaneous pipe draining and exit.

    Retain WNOWAIT leader identity until descendant-group cleanup, then reap
    nonblockingly. An additional finite cleanup bound belongs ONLY to fixtures;
    failure reports an unreaped PID for the outer disposable unit to contain.
    """
    end = time.monotonic() + seconds
    process = subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                               start_new_session=True, **kwargs)
    streams = {process.stdout.fileno(): bytearray(), process.stderr.fileno(): bytearray()}
    output, errors = streams.values()
    pending = set(streams)
    discarded = 0
    timed_out = False
    reaped = False
    limit = 64 * 1024  # retain bounded diagnostics while always draining pipes

    def kill_group():
        try:
            os.killpg(process.pid, signal.SIGKILL)
        except ProcessLookupError:
            pass

    try:
        for fd in pending:
            os.set_blocking(fd, False)
        exited = False
        while pending or not exited:
            if time.monotonic() >= end:
                timed_out = True
                break
            if observe is not None:
                observe(process.pid)
            # Never reap the leader before killing any remaining descendants:
            # its PID/group identity cannot be reused during this observation.
            status = os.waitid(os.P_PID, process.pid, os.WEXITED | os.WNOHANG | os.WNOWAIT)
            if status is not None and not exited:
                exited = True
                kill_group()
            ready, _, _ = select.select(list(pending), [], [], min(0.01, max(0, end - time.monotonic())))
            for fd in ready:
                chunk = os.read(fd, 65536)
                if not chunk:
                    pending.remove(fd)
                    continue
                keep = min(len(chunk), limit - len(streams[fd]))
                streams[fd].extend(chunk[:keep])
                discarded += len(chunk) - keep
    finally:
        failure = sys.exc_info()[1]
        # Also run on observer/assertion failures. All signals precede reaping.
        kill_group()
        cleanup_end = time.monotonic() + 2
        while time.monotonic() < cleanup_end:
            if process.poll() is not None:
                reaped = True
                break
            time.sleep(0.005)
        process.stdout.close()
        process.stderr.close()
        if failure is not None:
            failure.add_note(f'fixture cleanup: pid={process.pid} reaped={reaped} discarded={discarded} '
                             f'stdout={bytes(output)!r} stderr={bytes(errors)!r}')
    if timed_out or not reaped:
        raise FixtureTimeout(process.pid, bytes(output), bytes(errors), discarded, reaped)
    return Result(process.returncode, bytes(output), bytes(errors), discarded)