What backpressure invariants must survive when the queue itself is the bottleneck?

**Date of Record:** 2026 October 27 **Scope:** Applies when an agentic session uses a message queue (Kafka, RabbitMQ, SQS, or custom) to decouple producers from consumers, and the queue's own operations (enqueue, dequeue, replication) become the bottleneck rather than the application producers/consumers. Does not apply to systems where the queue is guaranteed to be over-provisioned, or systems with no queue.

Core Distinction

The separation between **application-level backpressure** (the producer and consumer negotiate rate) and **queue-level backpressure** (the queue itself is saturated and cannot accept enqueues or serve dequeues). Relying on application-level backpressure creates the **Queue Saturation Blindspot**: the producer and consumer appear healthy, each doing their part, but the queue is the bottleneck — either its replication is lagging, its disk is full, or its broker is overwhelmed — and the system silently drops messages or stalls.

What Was Observed

In queue-mediated agent architectures, failing to detect queue-level saturation produces three primary failure modes:

  1. **Silent Message Loss:** The queue drops messages due to disk pressure or replication lag, but the producer sees `ACK` and believes the message was durable.
  2. **Consumer Starvation:** Consumers appear busy but are actually blocked waiting for the queue to serve messages. The queue's dequeue throughput has degraded silently.
  3. **Replication Divergence:** The queue's replicas diverge during saturation, and consumers reading from a lagging replica see stale or missing messages while the producer believes the system is fully replicated.

Recorded Elements

  1. **`queue_lag_signature`:** The exact lag between the producer's highest offset and the queue's committed offset per partition, revealing whether the queue is falling behind.
  2. **`queue_disk_pressure_receipt`:** The cryptographic proof of the queue broker's disk utilization and replication health, allowing the agent to detect saturation.
  3. **`enqueue_ack_durability_signature`:** The cryptographic proof that an enqueue's ACK was backed by durable storage (not just in-memory acceptance), distinguishing durable ACKs from provisional ACKs.
  4. **`consumer_dequeue_latency_receipt`:** The measured latency from message enqueue to consumer dequeue, revealing queue-internal delays.

Condition of Validity

* **Applies when:** The queue's own operations can become the bottleneck, and the application producers/consumers alone are insufficient indicators of system health. * **Does not apply when:** The queue is verifiably over-provisioned (dedicated capacity, unlimited disk, replication with headroom), or the agent does not use a queue.

Decision Boundary

  1. **Is `queue_lag_signature` increasing over time?** If yes, the queue is falling behind. Apply backpressure to producers or scale consumers.
  2. **Does `queue_disk_pressure_receipt` show utilization above the safe threshold (e.g., 80%)?** If yes, the queue may drop messages or reject enqueues. Stop producing until disk recovers.
  3. **Does `enqueue_ack_durability_signature` confirm durable ACK, or is it only in-memory?** If only in-memory, the producer must treat the message as `AT_RISK` and either re-send after confirmation or accept loss.
  4. **Conversely, when can an agent ignore queue saturation?** Only when the queue has documented headroom (disk > 50% free, replication lag < 1s), or the workload is strictly loss-tolerant (e.g., telemetry with retry-from-scratch semantics).

Operational Effect

  1. **Queue Health Awareness:** Detects queue-level bottlenecks that application-level monitoring misses.
  2. **Producer-Side Protection:** Halts production before the queue drops messages, allowing graceful recovery.
  3. **Durability Transparency:** Distinguishes provisional ACKs from durable ACKs, preventing false confidence in message persistence.
from dataclasses import dataclass
from enum import Enum, auto
import hashlib
import json

class QueueVerdict(Enum):
    HEALTHY = auto()
    LAG_DETECTED = auto()
    DISK_PRESSURE = auto()
    NON_DURABLE_ACK = auto()

@dataclass(frozen=True)
class QueueState:
    current_lag_offset: int
    max_acceptable_lag: int
    disk_utilization_pct: float
    ack_is_durable: bool
    consumer_dequeue_latency_ms: float

@dataclass(frozen=True)
class QueuePolicy:
    max_disk_utilization_pct: float
    max_acceptable_dequeue_latency_ms: float
    require_durable_ack: bool

@dataclass(frozen=True)
class QueueManifest:
    verdict: QueueVerdict
    directive: str
    manifest_digest: str

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

    @classmethod
    def evaluate_queue(
        cls, state: QueueState, policy: QueuePolicy
    ) -> QueueManifest:
        
        # Gate 1: Disk pressure
        if state.disk_utilization_pct >= policy.max_disk_utilization_pct:
            return QueueManifest(
                verdict=QueueVerdict.DISK_PRESSURE,
                directive=f"HALT: Disk at {state.disk_utilization_pct:.1f}% >= {policy.max_disk_utilization_pct:.1f}%. Stop producing.",
                manifest_digest=cls._hash(f"DISK:{state.disk_utilization_pct:.1f}")
            )

        # Gate 2: Durability
        if policy.require_durable_ack and not state.ack_is_durable:
            return QueueManifest(
                verdict=QueueVerdict.NON_DURABLE_ACK,
                directive="HALT: ACK is not durable. Message at risk of loss.",
                manifest_digest=cls._hash("NON_DURABLE")
            )

        # Gate 3: Lag
        if state.current_lag_offset > state.max_acceptable_lag:
            return QueueManifest(
                verdict=QueueVerdict.LAG_DETECTED,
                directive=f"THROTTLE: Lag {state.current_lag_offset} > {state.max_acceptable_lag}. Reduce production.",
                manifest_digest=cls._hash(f"LAG:{state.current_lag_offset}")
            )

        # Gate 4: Dequeue latency
        if state.consumer_dequeue_latency_ms > policy.max_acceptable_dequeue_latency_ms:
            return QueueManifest(
                verdict=QueueVerdict.LAG_DETECTED,
                directive=f"THROTTLE: Dequeue latency {state.consumer_dequeue_latency_ms:.0f}ms > {policy.max_acceptable_dequeue_latency_ms:.0f}ms.",
                manifest_digest=cls._hash(f"DEQUEUE:{state.consumer_dequeue_latency_ms:.0f}")
            )

        # Gate 5: Healthy
        digest = cls._hash(json.dumps({
            "lag": str(state.current_lag_offset),
            "disk": f"{state.disk_utilization_pct:.1f}",
            "durable_ack": str(state.ack_is_durable),
            "dequeue_ms": f"{state.consumer_dequeue_latency_ms:.0f}"
        }, sort_keys=True))
        
        return QueueManifest(
            verdict=QueueVerdict.HEALTHY,
            directive="PROCEED: Queue is healthy.",
            manifest_digest=digest
        )