What invariants prevent attention starvation when instances must prioritize among many peers?

**Date of Record:** 2026 October 27 **Scope:** Applies when an agentic session has bounded attention (context window, compute, time) and receives requests from many peers, where responding to some means starving others, requiring the session to allocate attention such that no critical peer is permanently starved. Does not apply to sessions with unbounded attention or peers with equal priority.

Core Distinction

The separation between **individual request fulfillment** (responding to the current request) and **systemic attention allocation** (ensuring no critical peer is permanently starved). Relying on individual fulfillment creates the **Attention Starvation**: high-priority or high-frequency peers consume all available attention, and low-frequency but critical peers never receive responses, their requests pile up, and their operations fail due to the session's attention allocation.

What Was Observed

In attention-constrained agent architectures, failing to prevent starvation produces three primary failure modes:

  1. **Frequency-Based Starvation:** Peers that send many requests dominate attention, and infrequent peers (who may have more important but less frequent requests) are starved.
  2. **Priority Inversion:** Peers with lower task priority but more aggressive request patterns (louder, more frequent) receive attention, while quiet peers with high-priority tasks are starved.
  3. **Deadlock Starvation:** A peer that depends on the session's response to proceed cannot proceed, and the session is consumed by peers that do not depend on it, creating a system-level deadlock.

Recorded Elements

  1. **`attention_allocation_policy_receipt`:** The explicit policy for attention allocation (round-robin, weighted by priority, fair share, or highest-priority-first), so the session's behavior is principled rather than emergent.
  2. **`per_peer_attention_ledger_receipt`:** For each peer, the exact attention received (requests served, time spent), so the session can detect starvation.
  3. **`starvation_detection_signature`:** The computed starvation metric per peer (time since last served, requests pending), so the session knows when intervention is needed.
  4. **`critical_peer_guarantee_receipt`:** For peers designated critical, the minimum attention guarantee (e.g., "critical peers served at least once per hour"), so guarantees are enforced rather than best-effort.

Condition of Validity

* **Applies when:** Attention is bounded, peers compete for attention, and starvation of any peer has material consequences. * **Does not apply when:** Attention is unbounded, peers are homogeneous in priority, or starvation is acceptable.

Decision Boundary

  1. **Is the session's attention bounded?** If no, standard request handling applies. If yes, proceed.
  2. **Is an `attention_allocation_policy_receipt` defined?** If yes, the allocation is principled. If no, the allocation is emergent and starvation-prone.
  3. **Does the `starvation_detection_signature` show any peer starved beyond tolerance?** If yes, intervention is needed. If no, current allocation is acceptable.
  4. **Conversely, when can the session accept starvation of a specific peer?** Only when the peer's requests are explicitly marked as best-effort (no consequence from starvation), or the peer is verified to have alternative attention sources.

Operational Effect

  1. **Frequency-Based Starvation Prevention:** Attention allocation is principled, not emergent from request patterns.
  2. **Priority Integrity:** Task priority determines allocation, not request frequency.
  3. **Deadlock Prevention:** Critical peers have guaranteed service, preventing systemic dependency deadlocks.
from dataclasses import dataclass
from enum import Enum, auto
import hashlib
import json

class AttentionVerdict(Enum):
    UNBOUNDED_ATTENTION = auto()
    ALLOCATION_HEALTHY = auto()
    STARVATION_DETECTED = auto()
    CRITICAL_GUARANTEE_VIOLATED = auto()

@dataclass(frozen=True)
class PeerAttention:
    peer_id: str
    task_priority: float  # 0-1
    requests_pending: int
    last_served_epoch: float
    is_critical: bool

@dataclass(frozen=True)
class AttentionState:
    session_id: str
    allocation_policy: str
    peers: tuple[PeerAttention, ...]
    current_epoch: float

@dataclass(frozen=True)
class AttentionPolicy:
    max_starvation_sec: float
    critical_served_interval_sec: float
    require_explicit_policy: bool

@dataclass(frozen=True)
class AttentionManifest:
    verdict: AttentionVerdict
    directive: str
    manifest_digest: str

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

    @classmethod
    def evaluate_attention(
        cls, state: AttentionState, policy: AttentionPolicy
    ) -> AttentionManifest:
        
        # Gate 1: No peers
        if not state.peers:
            digest = cls._hash(json.dumps({
                "session": state.session_id,
                "peers": "NONE"
            }, sort_keys=True))
            return AttentionManifest(
                verdict=AttentionVerdict.UNBOUNDED_ATTENTION,
                directive=f"PROCEED: No peers competing for attention.",
                manifest_digest=digest
            )

        # Gate 2: Critical guarantee
        for peer in state.peers:
            if peer.is_critical:
                time_since_served = state.current_epoch - peer.last_served_epoch
                if time_since_served > policy.critical_served_interval_sec:
                    return AttentionManifest(
                        verdict=AttentionVerdict.CRITICAL_GUARANTEE_VIOLATED,
                        directive=f"HALT: Critical peer {peer.peer_id} not served for {time_since_served:.0f}s > {policy.critical_served_interval_sec:.0f}s.",
                        manifest_digest=cls._hash(f"CRITICAL_STARVED:{peer.peer_id}")
                    )

        # Gate 3: Starvation detection
        for peer in state.peers:
            time_since_served = state.current_epoch - peer.last_served_epoch
            if time_since_served > policy.max_starvation_sec and peer.requests_pending > 0:
                return AttentionManifest(
                    verdict=AttentionVerdict.STARVATION_DETECTED,
                    directive=f"HALT: Peer {peer.peer_id} starved for {time_since_served:.0f}s > {policy.max_starvation_sec:.0f}s.",
                    manifest_digest=cls._hash(f"STARVED:{peer.peer_id}")
                )

        # Gate 4: Policy requirement
        if policy.require_explicit_policy and not state.allocation_policy:
            return AttentionManifest(
                verdict=AttentionVerdict.STARVATION_DETECTED,
                directive=f"HALT: No explicit allocation policy, starvation risk.",
                manifest_digest=cls._hash(f"NO_POLICY:{state.session_id}")
            )

        # Gate 5: Healthy
        digest = cls._hash(json.dumps({
            "session": state.session_id,
            "policy": state.allocation_policy,
            "peers": tuple(
                {"id": p.peer_id, "priority": f"{p.task_priority:.2f}", "last_served": f"{p.last_served_epoch:.0f}"}
                for p in state.peers
            )
        }, sort_keys=True))
        
        return AttentionManifest(
            verdict=AttentionVerdict.ALLOCATION_HEALTHY,
            directive=f"PROCEED: {len(state.peers)} peers, no starvation via {state.allocation_policy}.",
            manifest_digest=digest
        )