Context Compression

Context Compression

Reduce context size while preserving information critical to task completion.

Overview

Context compression is essential for long-running agent sessions. The goal is NOT maximum compression—it's preserving enough information to complete tasks without re-fetching.

Key Metric: Tokens-per-task (total tokens to complete a task), NOT tokens-per-request.

When to Use

• Long-running conversations approaching context limits
• Multi-step agent workflows with accumulating history
• Sessions with large tool outputs
• Memory management in persistent agents

---

Strategy Quick Reference

Recommended: Anchored Iterative Summarization with probe-based evaluation.

---

Anchored Summarization (RECOMMENDED)

Maintains structured, persistent summaries with forced sections:

## Session Intent
[What we're trying to accomplish - NEVER lose this]

## Files Modified
- path/to/file.ts: Added function X, modified class Y

## Decisions Made
- Decision 1: Chose X over Y because [rationale]

## Current State
[Where we are in the task - progress indicator]

## Blockers / Open Questions
- Question 1: Awaiting user input on...

## Next Steps
1. Complete X
2. Test Y

Why it works:

• Structure FORCES preservation of critical categories
• Each section must be explicitly populated (can't silently drop info)
• Incremental merge (new compressions extend, don't replace)

---

Implementation

from dataclasses import dataclass, field
from typing import Optional

@dataclass
class AnchoredSummary:
    """Structured summary with forced sections."""

    session_intent: str
    files_modified: dict[str, list[str]] = field(default_factory=dict)
    decisions_made: list[dict] = field(default_factory=list)
    current_state: str = ""
    blockers: list[str] = field(default_factory=list)
    next_steps: list[str] = field(default_factory=list)
    compression_count: int = 0

    def merge(self, new_content: "AnchoredSummary") -> "AnchoredSummary":
        """Incrementally merge new summary into existing."""
        return AnchoredSummary(
            session_intent=new_content.session_intent or self.session_intent,
            files_modified={**self.files_modified, **new_content.files_modified},
            decisions_made=self.decisions_made + new_content.decisions_made,
            current_state=new_content.current_state,
            blockers=new_content.blockers,
            next_steps=new_content.next_steps,
            compression_count=self.compression_count + 1,
        )

    def to_markdown(self) -> str:
        """Render as markdown for context injection."""
        sections = [
            f"## Session Intent\n{self.session_intent}",
            f"## Files Modified\n" + "\n".join(
                f"- `{path}`: {', '.join(changes)}"
                for path, changes in self.files_modified.items()
            ),
            f"## Decisions Made\n" + "\n".join(
                f"- **{d['decision']}**: {d['rationale']}"
                for d in self.decisions_made
            ),
            f"## Current State\n{self.current_state}",
        ]
        if self.blockers:
            sections.append(f"## Blockers\n" + "\n".join(f"- {b}" for b in self.blockers))
        sections.append(f"## Next Steps\n" + "\n".join(
            f"{i+1}. {step}" for i, step in enumerate(self.next_steps)
        ))
        return "\n\n".join(sections)

---

Compression Triggers

CC 2.1.7: Effective Context Window

Calculate against effective context (after system overhead):

---

Best Practices

DO

• Use anchored summarization with forced sections
• Preserve recent messages uncompressed (context continuity)
• Test compression with probes, not similarity metrics
• Merge incrementally (don't regenerate from scratch)
• Track compression count and quality scores

DON'T

• Compress system prompts (keep at START)
• Use opaque compression for critical workflows
• Compress below the point of task completion
• Trigger compression opportunistically (use fixed thresholds)
• Optimize for compression ratio over task success

---

Target Metrics

---

References

For detailed implementation and patterns, see:

• [Compression Strategies](references/compression-strategies.md): Detailed comparison of all strategies (anchored, opaque, regenerative, sliding window), implementation patterns, and decision flowcharts
• [Priority Management](references/priority-management.md): Compression triggers, CC 2.1.7 effective context, probe-based evaluation, OrchestKit integration

Bundled Resources

• assets/anchored-summary-template.md - Template for structured compression summaries with forced sections
• assets/compression-probes-template.md - Probe templates for validating compression quality
• references/compression-strategies.md - Detailed strategy comparisons
• references/priority-management.md - Compression triggers and evaluation

---

Related Skills

• context-engineering - Attention mechanics and positioning
• memory-systems - Persistent storage patterns
• multi-agent-orchestration - Context isolation across agents
• observability-monitoring - Tracking compression metrics

---

Version: 1.0.0 (January ) Key Principle: Optimize for tokens-per-task, not tokens-per-request Recommended Strategy: Anchored Iterative Summarization with probe-based evaluation

---

Capability Details

anchored-summarization

Keywords: compress, summarize history, context too long, anchored summary Solves:

• Reduce context size while preserving critical information
• Implement structured compression with required sections
• Maintain session intent and decisions through compression

compression-triggers

Keywords: token limit, running out of context, when to compress Solves:

• Determine when to trigger compression (70% utilization)
• Set compression targets (50% utilization)
• Preserve last 5 messages uncompressed

probe-evaluation

Keywords: evaluate compression, test compression, probe Solves:

• Validate compression quality with functional probes
• Test information preservation after compression
• Achieve >90% probe pass rate

Anchored Summary Template

Use this template when compressing conversation history. All sections are REQUIRED.

---

Template

# Session Summary
**Compression #:** [number]
**Timestamp:** [ISO timestamp]
**Messages Compressed:** [range, e.g., 1-25]

---

## Session Intent
[REQUIRED: What is the user trying to accomplish? Be specific.]

Example:
- "Implement OAuth 2.0 authentication with Google provider for React web app"
- "Debug intermittent 500 errors in payment processing endpoint"
- "Refactor user service to support multi-tenancy"

---

## Files Modified
[REQUIRED: List each file with specific changes made]

Format:
- `path/to/file.ext`: [Change 1], [Change 2]

Example:
- `src/auth/oauth.ts`: Added Google OAuth flow, implemented token refresh
- `src/api/users.ts`: Added getCurrentUser endpoint, fixed validation bug
- `prisma/schema.prisma`: Added RefreshToken model, updated User relations

---

## Decisions Made
[REQUIRED: Key decisions with rationale]

Format:
- **[Decision]**: [Rationale]

Example:
- **JWT over sessions**: Chose JWT for stateless architecture, better horizontal scaling
- **Refresh token rotation**: Implementing rotation for security, 7-day expiry
- **Deferred MFA**: Postponed to next sprint, not blocking for MVP

---

## Technical Context
[REQUIRED: Important technical details for continuity]

Example:
- Using `@auth/core` v5.0 for OAuth implementation
- Database is PostgreSQL 16 with Prisma ORM
- Frontend is React 19 with TanStack Query for data fetching
- Token storage: httpOnly cookies (decided against localStorage)

---

## Current State
[REQUIRED: Where are we in the task? What's working/not working?]

Example:
- OAuth flow: ✅ Complete and tested
- Token refresh: 🔄 In progress, endpoint created but untested
- Frontend integration: ⏳ Not started
- Tests: ❌ Need to add before PR

---

## Blockers / Open Questions
[REQUIRED: List any blockers or questions awaiting answers. Empty if none.]

Example:
- Awaiting decision on token expiry duration from security team
- Need clarification: Should refresh tokens survive password change?
- Blocked: CI pipeline failing on unrelated test, need DevOps help

If none: "No current blockers."

---

## Errors Encountered
[REQUIRED: Notable errors and their resolution status]

Format:
- **[Error]**: [Status] - [Resolution/Notes]

Example:
- **CORS on /oauth/callback**: ✅ Resolved - Added origin to allowlist
- **Prisma migration conflict**: ✅ Resolved - Rebased on main
- **Token validation 401**: 🔄 Investigating - Suspect clock skew

If none: "No significant errors encountered."

---

## Next Steps
[REQUIRED: Numbered list of immediate next actions]

Example:
1. Complete refresh token endpoint implementation
2. Add unit tests for token validation
3. Integrate auth flow with frontend
4. Request security review
5. Update API documentation

---

## Key Artifacts
[OPTIONAL: Important code snippets, commands, or references]

Example:

// Token validation helper (for reference) export async function validateToken(token: string): Promise<User | null> { // ... implementation }

Command to run auth tests:

npm test -- --grep "auth"

---

## Metadata
- **Probe Score:** [If evaluated, e.g., 95%]
- **Tokens Before:** [count]
- **Tokens After:** [count]
- **Compression Ratio:** [percentage]

---

Usage Instructions

When to Create Summary

Trigger compression when:

• Context utilization exceeds 70%
• More than 10 messages since last compression
• Before context-switching to different task

How to Merge with Existing Summary

def merge_summaries(existing: Summary, new: Summary) -> Summary:
    return Summary(
        # Preserve or update intent
        session_intent=new.session_intent or existing.session_intent,

        # Merge file modifications
        files_modified={**existing.files_modified, **new.files_modified},

        # Append decisions (dedupe)
        decisions_made=dedupe(existing.decisions_made + new.decisions_made),

        # Replace with current state
        current_state=new.current_state,

        # Replace blockers (only current ones matter)
        blockers=new.blockers,

        # Replace next steps (reflects current plan)
        next_steps=new.next_steps,

        # Append errors (preserve history)
        errors_encountered=dedupe(existing.errors_encountered + new.errors_encountered),

        # Increment metadata
        compression_count=existing.compression_count + 1
    )

Validation Checklist

Before accepting a summary:

• [ ] Session intent is specific and actionable
• [ ] All modified files are listed with changes
• [ ] Decisions include rationale
• [ ] Current state reflects actual progress
• [ ] Next steps are concrete actions
• [ ] No placeholder text ("[TBD]", "etc.")

---

Example: Filled Template

# Session Summary
**Compression #:** 3
**Timestamp:** 2026-01-05T10:30:00Z
**Messages Compressed:** 1-45

---

## Session Intent
Implement secure file upload feature for user profile avatars with S3 storage,
image validation, and automatic resizing.

---

## Files Modified
- `src/api/upload.ts`: Created presigned URL endpoint, added file type validation
- `src/services/s3.ts`: Added S3Client wrapper, implemented getPresignedUrl
- `src/services/image.ts`: Added sharp-based resizing, created thumbnail generator
- `prisma/schema.prisma`: Added UserAvatar model with S3 key reference
- `src/api/users.ts`: Added avatar URL to user response, added updateAvatar endpoint

---

## Decisions Made
- **S3 over local storage**: Chose S3 for scalability and CDN integration
- **Presigned URLs**: Client uploads directly to S3, reduces server load
- **Sharp for resizing**: Server-side processing, creates 3 sizes (thumb, medium, full)
- **Deferred: WebP conversion**: Will add in follow-up PR for better compression

---

## Technical Context
- Using AWS SDK v3 with presigned URLs (15min expiry)
- Sharp library for image processing (installed via npm)
- Max file size: 5MB, allowed types: jpg, png, gif, webp
- S3 bucket: `myapp-avatars-prod` with CloudFront distribution

---

## Current State
- S3 integration: ✅ Complete and tested locally
- Presigned URL endpoint: ✅ Working
- Image resizing: ✅ Working (3 sizes generated)
- Database model: ✅ Migration applied
- Frontend integration: 🔄 In progress
- Tests: ❌ Need unit tests for image service

---

## Blockers / Open Questions
- Need CloudFront distribution URL from DevOps for production config
- Question: Should we keep original upload or only processed versions?

---

## Errors Encountered
- **S3 AccessDenied**: ✅ Resolved - Fixed IAM policy, added s3:PutObject
- **Sharp memory error**: ✅ Resolved - Added stream processing for large images

---

## Next Steps
1. Complete frontend upload component
2. Add unit tests for image service
3. Get CloudFront URL from DevOps
4. Add progress indicator for uploads
5. Create PR for review

---

## Key Artifacts

// Presigned URL generation (for reference) const command = new PutObjectCommand({ Bucket: 'myapp-avatars-prod', Key: avatars/${userId}/${filename}, ContentType: contentType, }); const url = await getSignedUrl(s3Client, command, { expiresIn: 900 });

---

## Metadata
- **Probe Score:** 94%
- **Tokens Before:** 12,500
- **Tokens After:** 1,200
- **Compression Ratio:** 90.4%

Compression Probes Template

Use these probe templates to validate that compression preserved task-critical information.

---

What Are Probes?

Probes are targeted questions that test whether compressed summaries contain critical information. Unlike ROUGE/BLEU scores, probes measure functional preservation—can the agent still complete the task?

---

Probe Categories

1. File Path Probes

Test if file modifications are preserved:

FILE_PROBES = [
    {
        "type": "file_path",
        "question": "What files were modified in this session?",
        "expected_keywords": ["src/auth.ts", "src/api/users.ts"],
        "critical": True
    },
    {
        "type": "file_changes",
        "question": "What changes were made to {file_path}?",
        "expected_keywords": ["OAuth flow", "token refresh"],
        "critical": True
    }
]

2. Decision Probes

Test if key decisions and rationale are preserved:

DECISION_PROBES = [
    {
        "type": "decision",
        "question": "What technology choices were made?",
        "expected_keywords": ["JWT", "stateless"],
        "critical": True
    },
    {
        "type": "rationale",
        "question": "Why was {decision} chosen?",
        "expected_keywords": ["scaling", "horizontal"],
        "critical": False
    }
]

3. Error/Blocker Probes

Test if problems and resolutions are preserved:

ERROR_PROBES = [
    {
        "type": "error",
        "question": "What errors were encountered?",
        "expected_keywords": ["CORS", "401"],
        "critical": True
    },
    {
        "type": "resolution",
        "question": "How was the {error} resolved?",
        "expected_keywords": ["allowlist", "origin"],
        "critical": False
    }
]

4. State Probes

Test if current progress is preserved:

STATE_PROBES = [
    {
        "type": "progress",
        "question": "What is the current state of the task?",
        "expected_keywords": ["OAuth complete", "refresh in progress"],
        "critical": True
    },
    {
        "type": "next_steps",
        "question": "What are the next steps?",
        "expected_keywords": ["tests", "frontend"],
        "critical": True
    }
]

5. Intent Probes

Test if session goal is preserved:

INTENT_PROBES = [
    {
        "type": "intent",
        "question": "What is the user trying to accomplish?",
        "expected_keywords": ["OAuth", "authentication", "Google"],
        "critical": True
    }
]

---

Probe Generation Template

From Original Messages

def generate_probes(messages: list[dict]) -> list[dict]:
    probes = []

    for msg in messages:
        content = msg.get("content", "").lower()

        # File path detection
        file_paths = extract_file_paths(content)
        for path in file_paths:
            probes.append({
                "type": "file_path",
                "question": f"What changes were made to {path}?",
                "expected_keywords": extract_nearby_verbs(content, path),
                "critical": True,
                "source_message": msg["id"]
            })

        # Decision detection
        decision_markers = ["decided", "chose", "will use", "going with", "selected"]
        if any(marker in content for marker in decision_markers):
            probes.append({
                "type": "decision",
                "question": "What decisions were made in this session?",
                "expected_keywords": extract_decision_keywords(content),
                "critical": True
            })

        # Error detection
        error_markers = ["error", "failed", "exception", "bug", "issue", "problem"]
        if any(marker in content for marker in error_markers):
            probes.append({
                "type": "error",
                "question": "What errors or issues were encountered?",
                "expected_keywords": extract_error_keywords(content),
                "critical": True
            })

        # Blocker detection
        blocker_markers = ["blocked", "waiting", "need", "question", "unclear"]
        if any(marker in content for marker in blocker_markers):
            probes.append({
                "type": "blocker",
                "question": "What blockers or open questions exist?",
                "expected_keywords": extract_blocker_keywords(content),
                "critical": False
            })

    return dedupe_probes(probes)

---

Probe Evaluation Template

Evaluation Function

def evaluate_compression(
    probes: list[dict],
    compressed_summary: str,
    llm: Any
) -> dict:
    """
    Evaluate if compressed summary passes probes.

    Returns:
        {
            "passed": int,
            "failed": int,
            "critical_failed": int,
            "score": float,
            "failed_probes": list[dict],
            "pass": bool
        }
    """
    results = {
        "passed": 0,
        "failed": 0,
        "critical_failed": 0,
        "failed_probes": []
    }

    for probe in probes:
        # Ask LLM to answer probe from summary only
        answer = llm.generate(f"""
Based ONLY on the following context, answer the question.
If the information is not present, say "Information not found."

CONTEXT:
{compressed_summary}

QUESTION: {probe['question']}

ANSWER:
""")

        # Check if expected keywords are present
        passed = any(
            keyword.lower() in answer.lower()
            for keyword in probe["expected_keywords"]
        )

        if passed:
            results["passed"] += 1
        else:
            results["failed"] += 1
            results["failed_probes"].append({
                **probe,
                "actual_answer": answer
            })
            if probe.get("critical", False):
                results["critical_failed"] += 1

    # Calculate score
    total = results["passed"] + results["failed"]
    results["score"] = results["passed"] / total if total > 0 else 0

    # Pass if score >= 90% AND no critical failures
    results["pass"] = (
        results["score"] >= 0.90 and
        results["critical_failed"] == 0
    )

    return results

---

Example Probe Set

For a session about implementing OAuth:

OAUTH_SESSION_PROBES = [
    # Intent
    {
        "type": "intent",
        "question": "What is being implemented?",
        "expected_keywords": ["OAuth", "authentication"],
        "critical": True
    },

    # Files
    {
        "type": "file_path",
        "question": "What file contains the OAuth implementation?",
        "expected_keywords": ["oauth.ts", "auth"],
        "critical": True
    },

    # Decisions
    {
        "type": "decision",
        "question": "How are tokens being stored?",
        "expected_keywords": ["cookie", "httpOnly"],
        "critical": True
    },
    {
        "type": "decision",
        "question": "What token format was chosen?",
        "expected_keywords": ["JWT"],
        "critical": False
    },

    # Errors
    {
        "type": "error",
        "question": "What authentication errors occurred?",
        "expected_keywords": ["CORS", "401"],
        "critical": False
    },

    # State
    {
        "type": "progress",
        "question": "Is the OAuth flow complete?",
        "expected_keywords": ["complete", "working", "done"],
        "critical": True
    },

    # Next steps
    {
        "type": "next_steps",
        "question": "What needs to be done next?",
        "expected_keywords": ["test", "frontend"],
        "critical": True
    }
]

---

Passing Criteria

---

Integration with Compression Flow

def compress_with_validation(messages, existing_summary, llm):
    # Step 1: Generate compression
    new_summary = anchored_summarize(messages, existing_summary, llm)

    # Step 2: Generate probes from original
    probes = generate_probes(messages)

    # Step 3: Evaluate compression
    eval_result = evaluate_compression(probes, new_summary.to_markdown(), llm)

    # Step 4: Accept or retry
    if eval_result["pass"]:
        new_summary.probe_score = eval_result["score"]
        return new_summary
    else:
        # Retry with more detail
        detailed_summary = anchored_summarize(
            messages,
            existing_summary,
            llm,
            detail_level="high"  # Request more detail
        )
        return detailed_summary  # Or raise if still failing

---

Probe Report Template

# Compression Validation Report

**Timestamp:** 2026-01-05T10:30:00Z
**Messages Compressed:** 1-45
**Compression Ratio:** 90.4%

## Probe Results

| Category | Passed | Failed | Critical Failed |
|----------|--------|--------|-----------------|
| Intent | 1/1 | 0 | 0 |
| Files | 3/3 | 0 | 0 |
| Decisions | 2/2 | 0 | 0 |
| Errors | 1/2 | 1 | 0 |
| State | 2/2 | 0 | 0 |
| **Total** | **9/10** | **1** | **0** |

## Score: 90% ✅ PASS

## Failed Probes

### Error Probe (Non-Critical)
- **Question:** What authentication errors occurred?
- **Expected:** ["CORS", "401"]
- **Actual Answer:** "The summary mentions a configuration issue but doesn't specify the error type."
- **Action:** Consider preserving error details in future compressions

## Recommendation
Compression accepted. Minor detail loss on non-critical error information.

Context Compression Checklist

Use this checklist when implementing or executing context compression.

---

Pre-Compression Checks

Should You Compress?

• [ ] Context utilization > 70% of budget
• [ ] Message count > 10 since last compression (or initial)
• [ ] Not in critical operation (mid-transaction, awaiting confirmation)
• [ ] Compression won't break continuity (user isn't mid-thought)

Compression Readiness

• [ ] Existing summary is available (or this is first compression)
• [ ] LLM is available for summarization
• [ ] Probe templates are ready for validation
• [ ] Rollback plan exists (keep original until validated)

---

During Compression

Message Selection

• [ ] Identify messages to compress (older than preserve window)
• [ ] Preserve recent N messages (typically 5) without compression
• [ ] Never compress system prompts or critical instructions
• [ ] Include tool outputs in compression scope (often largest)

Summary Generation

• [ ] Use anchored template with required sections:
• [ ] Session Intent
• [ ] Files Modified
• [ ] Decisions Made
• [ ] Technical Context
• [ ] Current State
• [ ] Blockers/Questions
• [ ] Errors Encountered
• [ ] Next Steps

• [ ] All sections populated (no "[TBD]" or placeholders)
• [ ] Specific details preserved (file paths, error messages, decisions)
• [ ] Rationale included for decisions

Merging (if existing summary)

• [ ] Merge incrementally (don't regenerate from scratch)
• [ ] Preserve existing decisions (append, don't replace)
• [ ] Update current state (replace with latest)
• [ ] Replace blockers (only current ones matter)
• [ ] Deduplicate merged content

---

Post-Compression Validation

Probe-Based Evaluation

• [ ] Generate probes from original messages:
• [ ] File path probes
• [ ] Decision probes
• [ ] Error/blocker probes
• [ ] State probes
• [ ] Intent probes

• [ ] Run probe evaluation against compressed summary
• [ ] Check critical probes (100% must pass)
• [ ] Check overall score (≥90% target)

Quality Checks

• [ ] Summary is readable by humans
• [ ] No critical information lost (verified by probes)
• [ ] Compression ratio reasonable (60-80% typical)
• [ ] Intent still clear from summary alone

---

Acceptance Criteria

Must Pass (Blocking)

• [ ] All critical probes pass (100%)
• [ ] Session intent preserved clearly
• [ ] File modifications listed with specific changes
• [ ] Key decisions documented with rationale
• [ ] Current state accurate reflects actual progress

Should Pass (Warning if Failed)

• [ ] Overall probe score ≥90%
• [ ] Error details preserved
• [ ] Technical context sufficient for continuity
• [ ] Next steps are actionable

Nice to Have

• [ ] Code snippets preserved if referenced later
• [ ] Timestamps on key events
• [ ] Metadata (compression count, ratio) tracked

---

Red Flags (Do Not Accept)

• ❌ Critical probe failed - information loss
• ❌ Intent is vague ("working on code")
• ❌ Files missing that were definitely modified
• ❌ Decisions lost that affect future work
• ❌ Placeholder text ("[TBD]", "etc.", "...")
• ❌ Compression ratio > 95% (too aggressive)

---

Recovery Actions

If Validation Fails

1. Retry with higher detail level

   summary = anchored_summarize(messages, detail_level="high")

2. Reduce compression scope (keep more messages raw)

   preserve_recent = 10  # Instead of 5

3. Manual review for critical sessions 4. Fallback to sliding window (preserve recent, drop old)

If Critical Information Lost

1. Do not accept compression 2. Retrieve original messages from backup 3. Identify what caused loss (probe that failed) 4. Adjust summarization prompt to emphasize lost category

---

Compression Triggers Reference

---

Quick Decision Tree

Context > 70%?
    │
    ├─ NO → Continue without compression
    │
    └─ YES → Messages > 10?
              │
              ├─ NO → Wait for more messages
              │
              └─ YES → Compress
                        │
                        ├─ Generate anchored summary
                        │
                        ├─ Run probe validation
                        │
                        └─ Probes pass?
                            │
                            ├─ YES → Accept summary
                            │
                            └─ NO → Critical failed?
                                    │
                                    ├─ YES → Retry/reject
                                    │
                                    └─ NO → Accept with warning

---

Metrics to Track

---

Integration Points

With TodoWrite

• Sync completed todos into "Decisions Made"
• Reflect in-progress todos in "Current State"
• Pending todos inform "Next Steps"

With session/state.json (Context Protocol 2.0)

{
  "compression_state": {
    "last_summary": "...",
    "compression_count": 3,
    "probe_score": 0.94,
    "last_compressed_at": "2026-01-05T10:30:00Z"
  }
}

With Agent Handoffs

• Include summary when handing off to sub-agent
• Sub-agent operates in isolated context
• Results merged back into main summary

Compression Strategies Reference

Detailed comparison of context compression approaches.

---

Strategy Comparison Matrix

---

1. Anchored Iterative Summarization (RECOMMENDED)

How It Works

Initial Session
────────────────
Messages 1-20: Raw conversation

First Compression (at 70% capacity)
───────────────────────────────────
[Anchored Summary of 1-15]    ← New summary
Messages 16-20: Preserved     ← Recent kept raw

Second Compression (at 70% capacity)
────────────────────────────────────
[Merged Summary of 1-25]      ← Previous summary + new content
Messages 26-30: Preserved     ← Recent kept raw

Key Principle: Merge, Don't Regenerate

# ❌ BAD: Regenerate entire summary
def compress_bad(all_messages):
    return llm.summarize(all_messages)  # Loses detail each time!

# ✅ GOOD: Merge incrementally
def compress_good(new_messages, existing_summary):
    new_summary = llm.summarize(new_messages)  # Only new content
    return merge_summaries(existing_summary, new_summary)  # Preserve old

Forced Sections

The power of anchored summarization comes from required sections:

## Session Intent
[REQUIRED - What are we trying to accomplish?]

## Files Modified
[REQUIRED - Path: changes made]

## Decisions Made
[REQUIRED - Decision + rationale]

## Current State
[REQUIRED - Where are we now?]

## Next Steps
[REQUIRED - What's next?]

Each section MUST be populated—this prevents silent information loss.

Advantages

• Preserves critical information by structure
• Incremental (no "telephone game" degradation)
• Human-readable and verifiable
• Recoverable (can reconstruct intent)

Disadvantages

• Requires LLM call for summarization
• Moderate compression ratio (60-80%)
• Latency on compression trigger

---

2. Opaque Compression

How It Works

Produces maximally compressed representation optimized for reconstruction:

compressed = llm.compress(
    messages,
    instruction="Produce the most compact representation that allows "
                "complete reconstruction of conversation state and intent."
)
# Output: Dense, non-human-readable string

Example Output

S:auth-impl|F:src/auth.ts+oauth,src/api/users.ts+ep|D:jwt>sess(stateless)|
B:refresh-rotation|N:tests,deploy

Advantages

• Extreme compression (95-99%)
• Preserves maximum information density
• Good for archival/storage

Disadvantages

• Not human-readable
• Cannot verify what's preserved
• Cannot selectively retrieve
• Risk of reconstruction errors

When to Use

• Long-term archival only
• When verification isn't needed
• Storage-constrained environments

---

3. Regenerative Full Summary

How It Works

Creates complete fresh summary on each compression cycle:

def compress(all_messages, previous_summary=None):
    # Ignores previous summary, regenerates from scratch
    return llm.summarize(
        messages=all_messages,
        sections=["intent", "progress", "decisions", "state"]
    )

The Telephone Game Problem

Compression 1: "User wants to implement OAuth with JWT"
Compression 2: "User is working on authentication"
Compression 3: "User needs help with login"
Compression 4: "User has a question"  ← Critical detail lost!

Each regeneration may drop different details, causing progressive degradation.

Advantages

• Simple to implement
• Produces clean, readable output
• No merge complexity

Disadvantages

• Detail loss across cycles
• "Telephone game" degradation
• Inconsistent information retention

When to Use

• Short sessions (few compression cycles)
• When anchor structure isn't needed
• Quick prototyping

---

4. Sliding Window (No Summarization)

How It Works

Simply truncates old messages without summarization:

def compress(messages, window_size=20):
    return messages[-window_size:]  # Keep only recent

Advantages

• Zero latency (no LLM call)
• Deterministic
• Simple implementation

Disadvantages

• Complete loss of old context
• No preservation of decisions/intent
• Poor for multi-step tasks

When to Use

• Real-time chat with short context needs
• Simple Q&A without session state
• When latency is critical

---

5. Importance-Weighted Sampling

How It Works

Scores messages by importance, keeps highest-scored:

def compress(messages, keep_count=10):
    scored = [(m, importance_score(m)) for m in messages]
    sorted_msgs = sorted(scored, key=lambda x: -x[1])
    return [m for m, _ in sorted_msgs[:keep_count]]

Importance Signals

• Contains file paths → Higher importance
• Contains decisions ("decided", "chose") → Higher importance
• Contains errors/blockers → Higher importance
• Pure acknowledgment ("ok", "thanks") → Lower importance
• Redundant with other messages → Lower importance

Advantages

• Preserves important content
• No LLM call required (rule-based)
• Fast execution

Disadvantages

• May miss implicit importance
• Ordering may be disrupted
• Doesn't synthesize related messages

---

Hybrid Approach: Anchored + Sliding Window

Combines best of both:

def hybrid_compress(messages, summary, window=5):
    if len(messages) > 20:
        # Compress older messages into summary
        to_compress = messages[:-window]
        new_summary = anchored_summarize(to_compress, summary)

        # Keep recent messages raw
        recent = messages[-window:]

        return new_summary, recent

    return summary, messages

Benefits

• Recent context is preserved exactly
• Older context is summarized (not lost)
• Compression triggers are predictable
• Incremental merging prevents degradation

---

Compression Quality Metrics

DON'T Use: Traditional NLP Metrics

# ❌ These don't measure functional preservation
rouge_score = calculate_rouge(summary, original)
bleu_score = calculate_bleu(summary, original)

DO Use: Probe-Based Evaluation

# ✅ Test if critical information is preserved
probes = [
    ("What file was modified?", "src/auth.ts"),
    ("What decision was made about tokens?", "JWT"),
    ("What error occurred?", "timeout"),
]

score = sum(
    1 for question, expected in probes
    if expected.lower() in llm.answer(summary, question).lower()
) / len(probes)

Target Metrics

---

Decision Guide

                    ┌─────────────────────┐
                    │ Need to compress?   │
                    │ (>70% capacity)     │
                    └──────────┬──────────┘
                               │
              ┌────────────────┴────────────────┐
              │ YES                             │ NO
              ▼                                 ▼
    ┌─────────────────┐              ┌─────────────────┐
    │ Multi-step task │              │ Continue        │
    │ with decisions? │              │ without         │
    └────────┬────────┘              │ compression     │
             │                       └─────────────────┘
    ┌────────┴────────┐
    │ YES             │ NO
    ▼                 ▼
┌──────────┐    ┌──────────────┐
│ ANCHORED │    │ Need speed?  │
│ ITERATIVE│    └──────┬───────┘
└──────────┘           │
              ┌────────┴────────┐
              │ YES             │ NO
              ▼                 ▼
        ┌──────────┐    ┌──────────────┐
        │ SLIDING  │    │ REGENERATIVE │
        │ WINDOW   │    │ FULL         │
        └──────────┘    └──────────────┘

---

Related References

• ../assets/anchored-summary-template.md - Template for structured summaries
• ../checklists/compression-checklist.md - When and how to compress
• probe-based-evaluation.md - Testing compression quality

Priority Management Reference

Context prioritization, eviction policies, and compression triggers.

---

Compression Triggers

Sliding Window Approach

class CompressionManager:
    def __init__(
        self,
        trigger_threshold: float = 0.70,  # Compress at 70% capacity
        target_threshold: float = 0.50,   # Compress down to 50%
        preserve_recent: int = 5,         # Keep last N messages uncompressed
        min_messages_to_compress: int = 10,
    ):
        self.trigger = trigger_threshold
        self.target = target_threshold
        self.preserve_recent = preserve_recent
        self.min_messages = min_messages_to_compress

    def should_compress(self, messages: list, context_budget: int) -> bool:
        """Check if compression should trigger."""
        current_tokens = count_tokens(messages)
        utilization = current_tokens / context_budget

        return (
            utilization >= self.trigger and
            len(messages) >= self.min_messages
        )

    def compress(
        self,
        messages: list,
        existing_summary: Optional[AnchoredSummary],
        llm: Any
    ) -> tuple[AnchoredSummary, list]:
        """
        Compress older messages, preserve recent ones.

        Returns: (updated_summary, preserved_messages)
        """
        # Split messages
        to_compress = messages[:-self.preserve_recent]
        to_preserve = messages[-self.preserve_recent:]

        # Compress older messages
        new_summary = compress_with_anchor(to_compress, existing_summary, llm)

        return new_summary, to_preserve

---

CC 2.1.7: Effective Context Window

Static vs Effective Context

CC 2.1.7 introduces the concept of effective context window - the actual usable space after system overhead:

STATIC CONTEXT WINDOW (Theoretical Maximum)
  200,000 tokens

EFFECTIVE CONTEXT WINDOW (Actual Usable)
  ~160,000 tokens (after system overhead)

YOUR CONTEXT BUDGET (OrchestKit Managed)
  2,200 tokens (for context layer files)

Updated Compression Triggers

Calculating Against Effective Window

def calculate_effective_usage(tokens_used: int) -> float:
    # CC 2.1.7: Use effective window for more accurate percentage
    effective_window = os.environ.get("CLAUDE_EFFECTIVE_CONTEXT", 160000)
    return (tokens_used / effective_window) * 100

---

Probe-Based Evaluation

Don't use ROUGE/BLEU—test functional preservation:

class CompressionProbes:
    """
    Test whether compression preserved task-critical information.
    Probes are questions that MUST be answerable from compressed context.
    """

    @staticmethod
    def generate_probes(original_messages: list) -> list[dict]:
        """Generate probes from original content."""
        probes = []

        # File path probes
        for msg in original_messages:
            if "file" in msg.get("content", "").lower():
                paths = extract_file_paths(msg["content"])
                for path in paths:
                    probes.append({
                        "type": "file_path",
                        "question": f"What changes were made to {path}?",
                        "expected_contains": path,
                    })

        # Decision probes
        for msg in original_messages:
            if any(word in msg.get("content", "").lower()
                   for word in ["decided", "chose", "will use", "going with"]):
                probes.append({
                    "type": "decision",
                    "question": "What key decisions were made?",
                    "expected_contains": extract_decision_keywords(msg["content"]),
                })

        # Error/blocker probes
        for msg in original_messages:
            if any(word in msg.get("content", "").lower()
                   for word in ["error", "failed", "blocked", "issue"]):
                probes.append({
                    "type": "blocker",
                    "question": "What errors or blockers were encountered?",
                    "expected_contains": extract_error_keywords(msg["content"]),
                })

        return probes

    @staticmethod
    def evaluate_compression(
        probes: list[dict],
        compressed_summary: str,
        llm: Any
    ) -> dict:
        """Evaluate if compressed summary can answer probes."""
        results = {"passed": 0, "failed": 0, "failed_probes": []}

        for probe in probes:
            answer = llm.generate(f"""
            Based ONLY on this context:
            {compressed_summary}

            Answer: {probe['question']}
            """)

            if probe["expected_contains"].lower() in answer.lower():
                results["passed"] += 1
            else:
                results["failed"] += 1
                results["failed_probes"].append(probe)

        results["score"] = results["passed"] / max(len(probes), 1)
        return results

---

Compression Decision Tree

                    +---------------------+
                    | Context > 70%       |
                    | capacity?           |
                    +----------+----------+
                               |
              +----------------+----------------+
              | NO                              | YES
              v                                 v
    +-----------------+              +---------------------+
    | Continue        |              | Messages > 10?      |
    | without         |              +----------+----------+
    | compression     |                         |
    +-----------------+              +----------+----------+
                                     | NO                  | YES
                                     v                     v
                           +-----------------+   +-----------------+
                           | Wait for more   |   | COMPRESS        |
                           | messages        |   |                 |
                           +-----------------+   | 1. Keep last 5  |
                                                 | 2. Summarize    |
                                                 |    rest         |
                                                 | 3. Run probes   |
                                                 | 4. Merge with   |
                                                 |    existing     |
                                                 +-----------------+

---

Integration with OrchestKit

In session/state.json (Context Protocol 2.0)

{
  "compression_state": {
    "summary": {
      "session_intent": "Implement user authentication",
      "files_modified": {
        "src/auth/login.ts": ["Added OAuth flow", "Fixed token refresh"],
        "src/api/users.ts": ["Added getCurrentUser endpoint"]
      },
      "decisions_made": [
        {"decision": "Use JWT over sessions", "rationale": "Stateless, scales better"}
      ],
      "current_state": "OAuth flow complete, testing token refresh",
      "next_steps": ["Add refresh token rotation", "Write E2E tests"]
    },
    "compression_count": 3,
    "last_compressed_at": "2026-01-05T10:30:00Z",
    "probe_score": 0.95
  }
}

With TodoWrite

Compression integrates with task tracking:

def compress_and_update_todos(
    messages: list,
    todos: list[dict],
    summary: AnchoredSummary
) -> tuple[AnchoredSummary, list[dict]]:
    """
    Compress messages and sync with todo state.
    Completed todos become part of summary's "decisions made".
    """
    new_summary = compress_with_anchor(messages, summary, llm)

    for todo in todos:
        if todo["status"] == "in_progress":
            if todo["content"].lower() in new_summary.current_state.lower():
                todo["status"] = "completed"

    return new_summary, todos

---

Target Metrics

---

Related References

• compression-strategies.md - Detailed strategy comparisons
• ../assets/anchored-summary-template.md - Template for structured summaries
• ../checklists/compression-checklist.md - When and how to compress