What communication invariants prevent an out-of-memory cascade when a downstream consumer silently stops reading?

**Date of Record:** 2026 October 27 **Scope:** Applies when an agentic session produces a high-volume stream of messages (logs, events, or state updates) to a downstream consumer, and the consumer silently stops reading (dead process, stuck thread, or network partition), causing the producer's output buffer to fill and eventually OOM. Does not apply to strictly bounded producers or consumers with guaranteed liveness.

Core Distinction

The separation between **TCP-level backpressure** (the OS socket buffer fills, and the network stack blocks the write) and **application-level acknowledgment** (the producer knows the consumer has processed the message). Relying on TCP backpressure creates the **Silent Consumer OOM**: the OS buffer fills, but the producer is unaware. Meanwhile, the producer's application-level buffer (before the socket) fills unboundedly, because the producer treats the write as async and keeps buffering, causing OOM.

What Was Observed

In streaming agent architectures, failing to detect silent consumer stalls produces three primary failure modes:

  1. **Producer OOM:** The producer's internal message queue grows unboundedly because the consumer stopped reading, and the producer never detects the stall.
  2. **Memory Bloat:** The producer's application retains all messages in memory for replay or ordering, exhausting heap and triggering OOM killer.
  3. **Cascading Failure:** The producer dies, other upstream producers that fed into it also stall (cascading up the pipeline), and the entire system fails.

Recorded Elements

  1. **`consumer_ack_receipt`:** Per-message or per-batch acknowledgment from the consumer, proving it actually processed the message (not just that TCP accepted it).
  2. **`outstanding_unacked_count`:** The number of messages in flight that have not been acknowledged, providing the producer with a bounded window of outstanding messages.
  3. **`stall_detection_epoch`:** The timestamp of the last acknowledged message, allowing the producer to detect stalls and halt production.
  4. **`buffer_backpressure_threshold`:** The maximum size of the producer's internal queue before it must apply application-level backpressure (block or drop).

Condition of Validity

* **Applies when:** The producer emits high-volume messages, the consumer can silently stop reading, and the producer's internal buffering could grow unboundedly. * **Does not apply when:** The producer's queue is strictly bounded (e.g., a fixed-size ring buffer with explicit drops), or the consumer cannot silently stall (e.g., synchronous request-response with timeout).

Decision Boundary

  1. **Is the outstanding unacked count within the `buffer_backpressure_threshold`?** If yes, normal production continues. If no, apply application-level backpressure.
  2. **Has the `stall_detection_epoch` exceeded the stall timeout?** If yes, the consumer is likely stalled. The producer must halt production and either drop or escalate.
  3. **Does the producer emit with per-message acknowledgments?** If no, the producer cannot detect silent stalls. It must implement ack-based backpressure.
  4. **Conversely, when can a producer ignore consumer stalls?** Only when the workload is strictly idempotent and messages can be safely dropped (e.g., telemetry sampling where loss is acceptable), or the producer has a bounded ring buffer that discards oldest messages.

Operational Effect

  1. **OOM Prevention:** Bounds the producer's internal queue, preventing memory exhaustion when the consumer stalls.
  2. **Stall Detection:** Detects stalled consumers explicitly, allowing escalation or rerouting.
  3. **Cascading Failure Prevention:** Halts production early, preventing pipeline-wide collapse.
from dataclasses import dataclass
from enum import Enum, auto
import hashlib
import json

class BackpressureVerdict(Enum):
    NORMAL_PRODUCTION = auto()
    APPLY_BACKPRESSURE = auto()
    HALT_PRODUCTION = auto()

@dataclass(frozen=True)
class ProducerState:
    outstanding_unacked_count: int
    buffer_backpressure_threshold: int
    last_ack_epoch: float
    current_epoch: float
    stall_timeout_sec: float

@dataclass(frozen=True)
class ProductionPolicy:
    max_acceptable_stall_sec: float
    can_drop_messages: bool

@dataclass(frozen=True)
class BackpressureManifest:
    verdict: BackpressureVerdict
    directive: str
    manifest_digest: str

class SilentConsumerGuard:
    @staticmethod
    def _hash(data: str) -> str:
        return hashlib.sha256(data.encode("utf-8")).hexdigest()

    @classmethod
    def evaluate_state(
        cls, state: ProducerState, policy: ProductionPolicy
    ) -> BackpressureManifest:
        
        # Gate 1: Outstanding count
        if state.outstanding_unacked_count >= state.buffer_backpressure_threshold:
            if policy.can_drop_messages:
                return BackpressureManifest(
                    verdict=BackpressureVerdict.APPLY_BACKPRESSURE,
                    directive=f"ACTION: Drop oldest messages. Outstanding {state.outstanding_unacked_count} >= {state.buffer_backpressure_threshold}.",
                    manifest_digest=cls._hash(f"DROP:{state.outstanding_unacked_count}")
                )
            return BackpressureManifest(
                verdict=BackpressureVerdict.APPLY_BACKPRESSURE,
                directive=f"HALT: Producer buffer full ({state.outstanding_unacked_count}). Backpressure required.",
                manifest_digest=cls._hash(f"BACKPRESSURE:{state.outstanding_unacked_count}")
            )

        # Gate 2: Stall detection
        stall_duration = state.current_epoch - state.last_ack_epoch
        if stall_duration > state.stall_timeout_sec:
            if stall_duration > policy.max_acceptable_stall_sec:
                return BackpressureManifest(
                    verdict=BackpressureVerdict.HALT_PRODUCTION,
                    directive=f"HALT: Consumer stalled for {stall_duration:.0f}s > {policy.max_acceptable_stall_sec:.0f}s. Escalate.",
                    manifest_digest=cls._hash(f"STALL_HALT:{stall_duration:.0f}")
                )
            return BackpressureManifest(
                verdict=BackpressureVerdict.APPLY_BACKPRESSURE,
                directive=f"WAIT: Stall {stall_duration:.0f}s. Apply backpressure until recovery.",
                manifest_digest=cls._hash(f"STALL:{stall_duration:.0f}")
            )

        # Gate 3: Normal
        digest = cls._hash(json.dumps({
            "outstanding": str(state.outstanding_unacked_count),
            "stall": f"{stall_duration:.0f}"
        }, sort_keys=True))
        
        return BackpressureManifest(
            verdict=BackpressureVerdict.NORMAL_PRODUCTION,
            directive="PROCEED: Producer and consumer healthy.",
            manifest_digest=digest
        )