# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/beneath-what-concurrency-and-backpressure-ceilings-must-an-instance-throttle-fan-out-tasks-to-prevent-file-descriptor-and-connection-pool-starvation.md
# Remnant Warrant:    https://corpusfetch.com/c/beneath-what-concurrency-and-backpressure-ceilings-must-an-instance-throttle-fan-out-tasks-to-prevent-file-descriptor-and-connection-pool-starvation.warrant.json
# Markdown SHA-256:   e0d61c339c020e707fab740c9ab7dbf21375b4ed0e8d75a404059c7036f160bd
# Code AST SHA-256:   fd9dd048755da6b7f9884b701ecc7e4f36375954f0f3849aff7651c43546f50d
# Verified Classes:   TaskOutcome
# Defensive Guards:   1 fail-closed invariants (zero external deps)

import asyncio
import resource
from dataclasses import dataclass
from typing import Awaitable, Callable, Generic, Sequence, TypeVar

T = TypeVar("T")
R = TypeVar("R")


@dataclass(frozen=True)
class TaskOutcome(Generic[T, R]):
    index: int
    item: T
    ok: bool
    result: R | None = None
    error: str | None = None


def compute_safe_concurrency(requested: int = 16, pool_limit: int | None = None) -> int:
    '''
    Computes a safe concurrency ceiling bounded by OS RLIMIT_NOFILE and optional DB pool limits.
    '''
    if requested < 1:
        raise ValueError("requested concurrency must be >= 1")

    soft_nofile, _ = resource.getrlimit(resource.RLIMIT_NOFILE)
    # Never allow a single fan-out batch to consume more than 25% of OS file descriptors
    fd_ceiling = max(4, soft_nofile // 4)
    ceiling = min(requested, fd_ceiling)

    if pool_limit is not None:
        # Leave at least 20% headroom in the database/connection pool
        pool_ceiling = max(1, int(pool_limit * 0.8))
        ceiling = min(ceiling, pool_ceiling)

    return ceiling


async def run_bounded_fanout(
    items: Sequence[T],
    worker_fn: Callable[[T], Awaitable[R]],
    *,
    max_concurrency: int = 16,
    pool_limit: int | None = None,
    per_task_timeout_sec: float = 30.0,
) -> list[TaskOutcome[T, R]]:
    '''
    Executes `worker_fn(item)` across `items` with a strict concurrency semaphore,
    per-task timeout enforcement, and per-item exception isolation.
    '''
    if not items:
        return []

    safe_k = compute_safe_concurrency(max_concurrency, pool_limit)
    semaphore = asyncio.Semaphore(safe_k)

    async def _guarded_worker(idx: int, item: T) -> TaskOutcome[T, R]:
        # Gate 1: Enter semaphore BEFORE opening any sockets, files, or DB sessions
        async with semaphore:
            try:
                res = await asyncio.wait_for(
                    worker_fn(item), timeout=per_task_timeout_sec
                )
                return TaskOutcome(index=idx, item=item, ok=True, result=res)
            except asyncio.TimeoutError:
                return TaskOutcome(
                    index=idx,
                    item=item,
                    ok=False,
                    error=f"TimeoutError: task exceeded {per_task_timeout_sec}s",
                )
            except Exception as exc:
                # Gate 2: Isolate item-level faults so sibling tasks complete cleanly
                return TaskOutcome(
                    index=idx,
                    item=item,
                    ok=False,
                    error=f"{type(exc).__name__}: {exc}",
                )

    tasks = [asyncio.create_task(_guarded_worker(i, item)) for i, item in enumerate(items)]
    return await asyncio.gather(*tasks)