Backpressure & Flow Control

Contents

Overview

flowchart TD
  CHECK["estimateTokens()"]
  DECIDE{"tokens > context_window - reserve?"}
  NORMAL["Continue normally"]
  SPLIT["splitMessagesByTokenShare()"]
  SUMMARIZE["summarizeInStages()"]
  MERGE["Merge summaries + recent messages"]
  VALIDATE{"compacted tokens < limit?"}
  TRUNC["tool-result-truncation.ts"]
  PRUNE["history-image-prune.ts"]
  GUARD["tool-result-context-guard.ts"]
  DONE["Compacted history ready"]
  RETRY["Retry with reduced budget"]

  CHECK --> DECIDE
  DECIDE -->|No| NORMAL
  DECIDE -->|Yes| TRUNC
  TRUNC --> PRUNE
  PRUNE --> GUARD
  GUARD --> SPLIT
  SPLIT --> SUMMARIZE
  SUMMARIZE --> MERGE
  MERGE --> VALIDATE
  VALIDATE -->|Yes| DONE
  VALIDATE -->|No| RETRY
  RETRY --> SPLIT

The overview flowchart shows the full compaction pipeline. Token estimation runs before every LLM call. When the estimate exceeds the safe threshold (context window minus a reserve buffer), the system first tries lightweight mitigations (truncating tool results, pruning old images). If those are insufficient, full compaction kicks in: split the history, summarize each chunk via LLM, merge, and validate. If the merged result still exceeds the limit, the system retries with a more aggressive split ratio.

Decision Points

How It Works

Backpressure management operates at three layers: pre-compaction mitigations, adaptive compaction, and session truncation as a last resort. Each layer activates only when the previous layer fails to bring token counts below the safe threshold.

Layer 1 — Pre-Compaction Mitigations

Before triggering full compaction, the system tries two lightweight measures that preserve full conversation detail:

Tool result truncation (tool-result-truncation.ts) scans all tool results in the history and truncates any that exceed a configurable byte limit. It preserves the first and last N characters with an [...truncated...] marker in between, ensuring the LLM still sees the beginning and end of large outputs.

Image pruning (history-image-prune.ts) removes image attachments from messages older than a configurable number of turns. Images are expensive in tokens (a single screenshot can cost 1000+ tokens) and become less relevant as the conversation progresses.

Layer 2 — Adaptive Compaction

When pre-compaction mitigations are insufficient, the full compaction algorithm runs. It is implemented primarily in compact.ts and orchestrated by compaction-runtime-context.ts.

// compact.ts — core compaction algorithm
const BASE_CHUNK_RATIO = 0.4;
const MIN_CHUNK_RATIO = 0.15;
const SAFETY_MARGIN = 1.2;

async function compactHistory(
  messages: Message[],
  contextWindow: number,
  reserve: number,
  attempt: number = 0,
): Promise<Message[]> {
  const targetTokens = contextWindow - reserve;
  const ratio = Math.max(
    MIN_CHUNK_RATIO,
    BASE_CHUNK_RATIO - attempt * 0.1,
  );

  // Split: keep recent messages verbatim, compact older ones
  const { toCompact, toKeep } = splitMessagesByTokenShare(
    messages,
    ratio,
    targetTokens,
  );

  // Summarize older messages in stages
  const summaries = await summarizeInStages(toCompact, targetTokens * ratio);

  // Merge: summary + recent verbatim messages
  const compacted = [...summaries, ...toKeep];

  // Validate: did we actually reduce enough?
  const estimate = estimateTokens(compacted) * SAFETY_MARGIN;
  if (estimate > targetTokens && attempt < 3) {
    return compactHistory(messages, contextWindow, reserve, attempt + 1);
  }

  return compacted;
}

Summarization Strategy

The summarizeInStages() function breaks the to-compact portion into smaller chunks (each fitting within the LLM's context window) and summarizes them sequentially. Each stage's output becomes input for the next, progressively distilling the conversation. The function respects message boundaries — it never splits a single message across summary stages.

Layer 3 — Session Truncation

If compaction fails (e.g., the summary itself is too long, or the timeout fires), session-truncation.ts performs hard truncation: it drops the oldest messages entirely until the history fits. This is the last resort — it loses information but guarantees the session can continue.

Behavior Diagram

stateDiagram-v2
  [*] --> Normal
  Normal --> CheckingTokens : before LLM call
  CheckingTokens --> Normal : under limit
  CheckingTokens --> PreCompaction : over limit
  PreCompaction --> TruncatingToolResults : truncate large results
  TruncatingToolResults --> PruningImages : prune old images
  PruningImages --> Recheck : re-estimate tokens
  Recheck --> Normal : under limit
  Recheck --> Compacting : still over limit
  Compacting --> Splitting : splitMessagesByTokenShare()
  Splitting --> Summarizing : summarizeInStages()
  Summarizing --> Merging : combine summaries + recent
  Merging --> Validating : estimateTokens() * SAFETY_MARGIN
  Validating --> Normal : under limit
  Validating --> RetryCompaction : still over, attempt < 3
  RetryCompaction --> Splitting : reduced ratio
  Validating --> HardTruncation : attempt >= 3
  HardTruncation --> Normal : oldest messages dropped
  Summarizing --> TimedOut : compaction-safety-timeout
  TimedOut --> HardTruncation : fall back to truncation

The state diagram shows the complete lifecycle of a single compaction cycle. Most sessions never leave the Normal → CheckingTokens → Normal loop. Long-running sessions may cycle through PreCompaction several times before needing full compaction. The TimedOut transition guards against summarization hangs, ensuring the session never blocks indefinitely.

Annotated Code Walkthrough

Walk through the key functions in execution order during a compaction cycle.

// compaction-runtime-context.ts
interface CompactionRuntimeContext {
  messages: Message[];
  contextWindow: number;
  reserveTokens: number;
  estimatedTokens: number;
  toolResultSizes: Map<string, number>;
  imageCount: number;
}

function buildCompactionContext(
  session: Session,
  model: ModelConfig,
): CompactionRuntimeContext {
  const messages = session.getHistory();
  const estimated = estimateTokens(messages);
  return {
    messages,
    contextWindow: model.contextWindow,
    reserveTokens: model.maxOutputTokens + 512, // +512 for system prompt
    estimatedTokens: estimated,
    toolResultSizes: scanToolResultSizes(messages),
    imageCount: countImages(messages),
  };
}

// compact.ts — splitting logic
function splitMessagesByTokenShare(
  messages: Message[],
  ratio: number,
  targetTokens: number,
): { toCompact: Message[]; toKeep: Message[] } {
  const compactBudget = Math.floor(targetTokens * ratio);
  let accumulated = 0;
  let splitIndex = 0;

  for (let i = 0; i < messages.length; i++) {
    const msgTokens = estimateTokens([messages[i]]);
    if (accumulated + msgTokens > compactBudget) {
      splitIndex = i;
      break;
    }
    accumulated += msgTokens;
    splitIndex = i + 1;
  }

  // Always keep system message in both halves
  const systemMsg = messages.find(m => m.role === "system");
  return {
    toCompact: messages.slice(0, splitIndex),
    toKeep: systemMsg
      ? [systemMsg, ...messages.slice(splitIndex).filter(m => m.role !== "system")]
      : messages.slice(splitIndex),
  };
}

// compaction-safety-timeout.ts
const COMPACTION_TIMEOUT_MS = 30_000;

async function withCompactionTimeout<T>(
  fn: () => Promise<T>,
  fallback: () => T,
): Promise<T> {
  const controller = new AbortController();
  const timer = setTimeout(
    () => controller.abort(),
    COMPACTION_TIMEOUT_MS,
  );

  try {
    const result = await fn();
    clearTimeout(timer);
    return result;
  } catch (err) {
    clearTimeout(timer);
    if (controller.signal.aborted) {
      console.warn("Compaction timed out, falling back to truncation");
      return fallback();
    }
    throw err;
  }
}

Edge Cases & Error Handling

Configuration

Related Topics

See also:

• Streaming overview — the parent L2 page covering the full streaming pipeline
• Chunk Assembly — how raw SSE events are normalized into unified payloads
• Error Recovery — how errors during streaming are categorized and recovered from

Directory Map

src/agents/pi-embedded-runner/
├── compaction/
│   ├── compact.ts                        ← Core compaction algorithm with splitMessagesByTokenShare()
│   ├── compaction-runtime-context.ts     ← Runtime context builder and concurrency mutex
│   ├── compaction-safety-timeout.ts      ← Timeout guard for compaction pipeline
│   └── session-truncation.ts             ← Last-resort hard truncation
├── context-guard/
│   ├── tool-result-truncation.ts         ← Truncates oversized tool results
│   └── tool-result-context-guard.ts      ← Proactive guard on incoming tool results
├── history/
│   └── history-image-prune.ts            ← Prunes old images from conversation history
└── streaming/
    └── llm-idle-timeout.ts               ← Idle LLM connection timeout handler