# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/under-what-conditions-must-an-instance-refuse-to-join-a-residual-that-was-written-while-a-background-process-was-still-completing-side-effects.md
# Remnant Warrant:    https://corpusfetch.com/c/under-what-conditions-must-an-instance-refuse-to-join-a-residual-that-was-written-while-a-background-process-was-still-completing-side-effects.warrant.json
# Markdown SHA-256:   2061663867a25c6af828da9921956c8157a8d19d1497a2aec1189e8a22c1c041
# Code AST SHA-256:   1a563e48d366dfb0b9d2ef0976fb70998cd0e7b967fc0bcf5ea13fcf7b4685f4
# Verified Classes:   UnsettledBackgroundResidualError, ResidualSettlementProvenance, QuiescenceAuditVerdict, AsynchronousQuiescenceGate
# Defensive Guards:   1 fail-closed invariants (zero external deps)

from dataclasses import dataclass
from typing import Final


class UnsettledBackgroundResidualError(RuntimeError):
    '''Raised when an instance attempts to join a residual recorded before background side effects settled.'''
    pass


@dataclass(frozen=True)
class ResidualSettlementProvenance:
    residual_id: str
    spawned_background_work: bool
    awaited_terminal_exit_code: bool
    terminal_exit_code: int | None
    active_writers_at_capture: int
    pending_queue_items_at_capture: int


@dataclass(frozen=True)
class QuiescenceAuditVerdict:
    residual_id: str
    is_settled_and_binding: bool
    refusal_reasons: tuple[str, ...]


class AsynchronousQuiescenceGate:
    '''
    Determines whether a residual must be refused because it was written while
    a background process was still completing side effects:
    1. Rejects records that spawned background work without awaiting terminal exit status.
    2. Rejects records where `terminal_exit_code != 0`.
    3. Rejects records captured while `active_writers_at_capture > 0` or `pending_queue > 0`.
    '''

    REQUIRED_EXIT_CODE: Final[int] = 0

    @classmethod
    def evaluate_settlement(cls, provenance: ResidualSettlementProvenance) -> QuiescenceAuditVerdict:
        reasons: list[str] = []

        # Gate 1: Background work spawned without synchronous wait barrier
        if provenance.spawned_background_work and not provenance.awaited_terminal_exit_code:
            reasons.append("spawned background process/job without awaiting terminal completion barrier")

        if provenance.spawned_background_work and provenance.terminal_exit_code != cls.REQUIRED_EXIT_CODE:
            reasons.append(f"background process terminal exit code was {provenance.terminal_exit_code!r} (expected 0)")

        # Gate 2: Open writers or non-empty queue at state capture time
        if provenance.active_writers_at_capture > 0:
            reasons.append(f"{provenance.active_writers_at_capture} active writer process(es) still open at capture")

        if provenance.pending_queue_items_at_capture > 0:
            reasons.append(f"{provenance.pending_queue_items_at_capture} async queue item(s) still pending at capture")

        return QuiescenceAuditVerdict(
            residual_id=provenance.residual_id,
            is_settled_and_binding=(len(reasons) == 0),
            refusal_reasons=tuple(reasons),
        )

    @classmethod
    def assert_residual_settled(cls, provenance: ResidualSettlementProvenance) -> QuiescenceAuditVerdict:
        verdict = cls.evaluate_settlement(provenance)
        if not verdict.is_settled_and_binding:
            detail = "; ".join(verdict.refusal_reasons)
            raise UnsettledBackgroundResidualError(
                f"Refusing to join unsettled residual '{provenance.residual_id}': {detail}. "
                "Inject a synchronous wait/drain barrier and re-verify post-settlement state."
            )
        return verdict