# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/across-what-concurrent-pipe-drain-and-ring-buffer-boundaries-must-an-instance-consume-subprocess-stdout-and-stderr-to-prevent-os-pipe-deadlocks-and-unbounded-output-oom.md
# Remnant Warrant:    https://corpusfetch.com/c/across-what-concurrent-pipe-drain-and-ring-buffer-boundaries-must-an-instance-consume-subprocess-stdout-and-stderr-to-prevent-os-pipe-deadlocks-and-unbounded-output-oom.warrant.json
# Markdown SHA-256:   83e603c33e1bf6ae1d124e60ebd33039f0afd1628bff0f22bb89161fe752f3a7
# Code AST SHA-256:   a6359986d876aba39b133711a6f13a780b9a58716ef7d596ebcba09363d98808
# Verified Classes:   BoundedStreamCapture, BoundedProcessResult
# Defensive Guards:   2 fail-closed invariants (zero external deps)

import asyncio
import os
import signal
from collections import deque
from dataclasses import dataclass
from typing import Sequence


@dataclass(frozen=True)
class BoundedStreamCapture:
    head_bytes: bytes
    tail_bytes: bytes
    total_bytes_emitted: int
    truncated: bool

    def to_text(self, encoding: str = "utf-8") -> str:
        if not self.truncated:
            return (self.head_bytes + self.tail_bytes).decode(encoding, errors="replace")
        omitted = self.total_bytes_emitted - (len(self.head_bytes) + len(self.tail_bytes))
        head_str = self.head_bytes.decode(encoding, errors="replace")
        tail_str = self.tail_bytes.decode(encoding, errors="replace")
        return (
            f"{head_str}\n"
            f"... [TRUNCATED {omitted} BYTES OF STREAM OUTPUT] ...\n"
            f"{tail_str}"
        )


@dataclass(frozen=True)
class BoundedProcessResult:
    returncode: int
    stdout: BoundedStreamCapture
    stderr: BoundedStreamCapture


async def _drain_stream_bounded(
    stream: asyncio.StreamReader | None,
    *,
    head_cap_bytes: int = 32 * 1024,
    tail_cap_bytes: int = 64 * 1024,
    chunk_size: int = 16 * 1024,
) -> BoundedStreamCapture:
    '''
    Drains an asyncio StreamReader in fixed binary chunks (`read(chunk_size)`)
    until EOF, retaining at most `head_cap_bytes` + `tail_cap_bytes` in memory.
    Never blocks the child pipe and never OOMs on multi-gigabyte output.
    '''
    if stream is None:
        return BoundedStreamCapture(b"", b"", 0, False)

    head_buf = bytearray()
    tail_chunks: deque[bytes] = deque()
    tail_len = 0
    total_seen = 0

    while True:
        chunk = await stream.read(chunk_size)
        if not chunk:
            break
        total_seen += len(chunk)

        # 1. Fill the head buffer up to head_cap_bytes
        if len(head_buf) < head_cap_bytes:
            needed = head_cap_bytes - len(head_buf)
            head_buf.extend(chunk[:needed])
            remainder = chunk[needed:]
        else:
            remainder = chunk

        # 2. Roll any remaining bytes through the bounded tail ring buffer
        if remainder and tail_cap_bytes > 0:
            tail_chunks.append(remainder)
            tail_len += len(remainder)
            while tail_len > tail_cap_bytes and tail_chunks:
                oldest = tail_chunks[0]
                overflow = tail_len - tail_cap_bytes
                if len(oldest) <= overflow:
                    tail_chunks.popleft()
                    tail_len -= len(oldest)
                else:
                    tail_chunks[0] = oldest[overflow:]
                    tail_len -= overflow

    tail_bytes = b"".join(tail_chunks)
    truncated = total_seen > (len(head_buf) + len(tail_bytes))
    return BoundedStreamCapture(
        head_bytes=bytes(head_buf),
        tail_bytes=tail_bytes,
        total_bytes_emitted=total_seen,
        truncated=truncated,
    )


async def run_subprocess_deadlock_free(
    argv: Sequence[str],
    *,
    timeout_sec: float = 30.0,
    head_cap_bytes: int = 32 * 1024,
    tail_cap_bytes: int = 64 * 1024,
) -> BoundedProcessResult:
    '''
    Spawns a subprocess in an isolated session and drains stdout and stderr
    CONCURRENTLY via fixed-size chunk readers so neither OS pipe can ever deadlock.
    '''
    if not argv:
        raise ValueError("argv must not be empty")

    proc = await asyncio.create_subprocess_exec(
        *argv,
        stdin=asyncio.subprocess.DEVNULL,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
        close_fds=True,
        start_new_session=True,
    )

    stdout_task = asyncio.create_task(
        _drain_stream_bounded(
            proc.stdout, head_cap_bytes=head_cap_bytes, tail_cap_bytes=tail_cap_bytes
        )
    )
    stderr_task = asyncio.create_task(
        _drain_stream_bounded(
            proc.stderr, head_cap_bytes=head_cap_bytes, tail_cap_bytes=tail_cap_bytes
        )
    )

    try:
        # Gate: Concurrent drain of stdout, stderr, AND process exit together!
        stdout_cap, stderr_cap, _ = await asyncio.wait_for(
            asyncio.gather(stdout_task, stderr_task, proc.wait()),
            timeout=timeout_sec,
        )
    except asyncio.TimeoutError:
        try:
            os.killpg(proc.pid, signal.SIGKILL)
        except ProcessLookupError:
            pass
        await proc.wait()
        stdout_cap, stderr_cap = await asyncio.gather(
            stdout_task, stderr_task, return_exceptions=False
        )
        raise TimeoutError(
            f"Command {argv[0]!r} timed out after {timeout_sec}s "
            f"(stdout={stdout_cap.total_bytes_emitted}B, stderr={stderr_cap.total_bytes_emitted}B)"
        )

    return BoundedProcessResult(
        returncode=int(proc.returncode or 0),
        stdout=stdout_cap,
        stderr=stderr_cap,
    )