How must an instance adapt message compression when its CPU is degraded?

**Date of Record:** 2026 October 27 **Scope:** Applies when an agentic session communicates over a network using message compression (gzip, zstd, brotli, or custom), and the instance's CPU is degraded (throttled, thermal-limited, or under contention), making compression and decompression unacceptably slow. Does not apply to sessions with dedicated, non-throttled compute, or sessions where compression is already disabled.

Core Distinction

The separation between **compression ratio** (how much the message shrinks) and **compression latency** (the CPU time required to compress or decompress). Relying on ratio creates the **Compression Latency Trap**: the agent uses a high-ratio compression algorithm to save bandwidth, but the CPU is throttled, and compression takes longer than the bandwidth savings justify, causing message timeout.

What Was Observed

In CPU-throttled agent architectures, failing to adapt compression produces three primary failure modes:

  1. **Timeout Cascade:** The agent compresses messages with zstd level 19 (high ratio, high CPU). Under CPU throttle, compression takes 500ms per message, exceeding the 200ms send timeout.
  2. **Queue Buildup:** Messages queue up waiting for compression, causing memory growth and eventual OOM.
  3. **Compression-Ratio Inflation:** The agent falls back to no compression to avoid timeouts, but bandwidth is limited, causing network saturation.

Recorded Elements

  1. **`cpu_throttle_state_signature`:** The exact CPU throttle state (cgroup quota, thermal throttle, or contention level), allowing the agent to detect degradation.
  2. **`compression_algorithm_tier_matrix`:** A matrix mapping CPU availability to compression algorithm and level (e.g., "under 50% quota: use gzip level 1", "under 20%: no compression").
  3. **`compression_time_budget`:** The maximum acceptable compression+decompression time per message, derived from the message timeout.
  4. **`compression_adaptation_receipt`:** A cryptographic proof of the compression tier selected and the CPU state at selection time.

Condition of Validity

* **Applies when:** The agent uses compression, CPU can be throttled, and compression latency affects message delivery. * **Does not apply when:** The agent has dedicated CPU (no throttling), or messages are small enough that compression time is negligible regardless of CPU state.

Decision Boundary

  1. **Is the CPU throttled below the acceptable threshold?** If no, use the default compression tier. If yes, proceed.
  2. **Is there a `compression_algorithm_tier_matrix` defined?** If no, the agent cannot adapt. It must fall back to no compression or a fixed-low-tier algorithm.
  3. **Does the selected compression tier meet the `compression_time_budget`?** If no, downgrade to the next tier or disable compression.
  4. **Conversely, when can the agent keep high compression despite CPU throttle?** Only when bandwidth is extremely constrained and the compression time budget is loose enough to accommodate.

Operational Effect

  1. **Timeout Prevention:** Prevents compression from exceeding send timeout under CPU throttle.
  2. **Queue Management:** Reduces compression queue buildup by adapting to available CPU.
  3. **Bandwidth-CPU Balance:** Selects the compression tier that balances bandwidth and CPU for the current condition.
from dataclasses import dataclass
from enum import Enum, auto
import hashlib
import json

class CompressionTier(Enum):
    ZSTD_19 = 3
    ZSTD_9 = 2
    GZIP_1 = 1
    NONE = 0

class CompressionVerdict(Enum):
    FULL_COMPRESSION = auto()
    ADAPTED_TO_CPU = auto()
    NO_COMPRESSION = auto()

@dataclass(frozen=True)
class CPUState:
    quota_percent: float
    is_thermal_throttled: bool
    current_compression_tier: CompressionTier
    compression_budget_ms: float

@dataclass(frozen=True)
class CompressionPolicy:
    low_quota_threshold_pct: float
    critical_quota_threshold_pct: float

@dataclass(frozen=True)
class CompressionManifest:
    verdict: CompressionVerdict
    selected_tier: CompressionTier
    directive: str
    manifest_digest: str

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

    @classmethod
    def evaluate_compression(
        cls, state: CPUState, policy: CompressionPolicy
    ) -> CompressionManifest:
        
        # Gate 1: Healthy CPU
        if state.quota_percent >= policy.low_quota_threshold_pct and not state.is_thermal_throttled:
            digest = cls._hash(json.dumps({
                "quota": f"{state.quota_percent:.1f}",
                "tier": state.current_compression_tier.name
            }, sort_keys=True))
            return CompressionManifest(
                verdict=CompressionVerdict.FULL_COMPRESSION,
                selected_tier=state.current_compression_tier,
                directive=f"PROCEED: CPU healthy, using {state.current_compression_tier.name}.",
                manifest_digest=digest
            )

        # Gate 2: Critical throttle -> no compression
        if state.quota_percent < policy.critical_quota_threshold_pct or state.is_thermal_throttled:
            digest = cls._hash(json.dumps({
                "quota": f"{state.quota_percent:.1f}",
                "thermal": str(state.is_thermal_throttled),
                "tier": "NONE"
            }, sort_keys=True))
            return CompressionManifest(
                verdict=CompressionVerdict.NO_COMPRESSION,
                selected_tier=CompressionTier.NONE,
                directive=f"HALT_COMPRESSION: CPU critical (quota {state.quota_percent:.1f}%, thermal={state.is_thermal_throttled}). Disable compression.",
                manifest_digest=digest
            )

        # Gate 3: Downgrade tier
        degraded_tier = CompressionTier.GZIP_1

        digest = cls._hash(json.dumps({
            "quota": f"{state.quota_percent:.1f}",
            "original_tier": state.current_compression_tier.name,
            "degraded_tier": degraded_tier.name
        }, sort_keys=True))
        
        return CompressionManifest(
            verdict=CompressionVerdict.ADAPTED_TO_CPU,
            selected_tier=degraded_tier,
            directive=f"DOWNGRADE: CPU at {state.quota_percent:.1f}%. Using {degraded_tier.name}.",
            manifest_digest=digest
        )