# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/in-what-scenarios-must-an-instance-refuse-to-multiplex-tool-calls-across-mcp-sessions-when-the-sse-stream-is-half-open.md
# Remnant Warrant:    https://corpusfetch.com/c/in-what-scenarios-must-an-instance-refuse-to-multiplex-tool-calls-across-mcp-sessions-when-the-sse-stream-is-half-open.warrant.json
# Markdown SHA-256:   a88efd0a2c7eca25a04297c603498dd170e4400c99f2400127d972e0c02078a8
# Code AST SHA-256:   7eedc6a2ec93ac86d55060742625706220c4506be63afe4f39518b252d7a85bb
# Verified Classes:   HalfOpenStreamRefusal, SessionStream
# Defensive Guards:   2 fail-closed invariants (zero external deps)

class HalfOpenStreamRefusal(RuntimeError):
    """Raised when a multiplexed outbound frame targets a half-open,
    credit-exhausted SSE stream and must not be admitted."""
    pass


class SessionStream:
    def __init__(self, session_id: str, credit: int, drain_window_s: float):
        self.session_id = session_id
        self.credit = credit
        self.drain_window_s = drain_window_s
        self.queue_depth = 0
        self.last_progress_s = 0.0

    def is_half_open(self, now_s: float) -> bool:
        # Active session, but no forward progress within the drain window.
        return self.queue_depth >= self.credit and \
            (now_s - self.last_progress_s) >= self.drain_window_s

    def admit(self, now_s: float, session_scoped: bool, idempotent: bool) -> None:
        if not session_scoped:
            return  # session-independent work bypasses this gate
        if self.is_half_open(now_s):
            if not idempotent:
                # Refuse BEFORE dispatch so gateway retries cannot duplicate.
                raise HalfOpenStreamRefusal(
                    f"session {self.session_id}: stream half-open at credit "
                    f"{self.credit}, queue_depth {self.queue_depth}; "
                    f"refusing non-idempotent multiplexed call"
                )
            raise HalfOpenStreamRefusal(
                f"session {self.session_id}: stream half-open; "
                f"backpressure required (429 + Retry-After)"
            )
        self.queue_depth += 1


def dispatch_tool_call(stream: SessionStream, now_s: float,
                       session_scoped: bool = True,
                       idempotent: bool = False) -> None:
    # Admission control lives at the transport layer, upstream of execution.
    stream.admit(now_s, session_scoped=session_scoped, idempotent=idempotent)
    # ... enqueue outbound frame, run tool, write to SSE channel ...