# The Grid - Context Management System

## Technical Design Document

**Version:** 1.0
**Author:** Grid Executor (TICKET-010)
**Date:** 2026-01-24

---

## Executive Summary

The Grid Context Management System provides token budgeting, compression triggers, and context optimization strategies for multi-agent orchestration. This prevents context overflow, reduces costs, and maintains critical information across agent lifecycles.

### Key Features

1. **Token Budget Allocation** - Structured context distribution across agent roles
2. **Compression Triggers** - Automatic compression at 80%/90%/95% capacity
3. **Compression Strategies** - Sliding window, summarization, and prioritization
4. **MC Monitoring** - Pre-spawn context checks and automatic compression
5. **History Management** - Intelligent history pruning while preserving critical context

---

## Architecture Overview

```
                    ┌─────────────────────────────────────────┐
                    │           Master Control (MC)           │
                    │                                         │
                    │  ┌─────────────────────────────────────┐│
                    │  │       Context Monitor Gate          ││
                    │  │  (runs before EVERY Task() spawn)   ││
                    │  └─────────────────┬───────────────────┘│
                    └────────────────────┼────────────────────┘
                                         │
                    ┌────────────────────┼────────────────────┐
                    │                    ▼                    │
                    │        Context Budget Tracker           │
                    │                                         │
                    │  ┌─────────────────────────────────────┐│
                    │  │  total_budget: 200000 tokens        ││
                    │  │  current_usage: 145000 tokens       ││
                    │  │  usage_percent: 72.5%               ││
                    │  │  trigger_level: NORMAL              ││
                    │  └─────────────────────────────────────┘│
                    └─────────────────────────────────────────┘
                                         │
         ┌───────────────┬───────────────┼───────────────┬───────────────┐
         ▼               ▼               ▼               ▼               ▼
    ┌─────────┐    ┌─────────┐    ┌─────────┐    ┌─────────┐    ┌─────────┐
    │ System  │    │ Warmth  │    │  Plan   │    │ History │    │ Reserve │
    │  25K    │    │  10K    │    │  20K    │    │  75K    │    │  20K    │
    │  12.5%  │    │   5%    │    │  10%    │    │  37.5%  │    │  10%    │
    └─────────┘    └─────────┘    └─────────┘    └─────────┘    └─────────┘
```

---

## Token Budget Allocations

### Claude Opus Context Window: 200K Tokens

The 200K token budget is allocated across distinct purpose categories:

```yaml
token_budget:
  total: 200000
  allocations:
    system_instructions: 25000   # 12.5% - Agent role, protocols, rules
    warmth: 10000                # 5%    - Learnings from prior Programs
    plan: 20000                  # 10%   - Current execution plan
    current_input: 50000         # 25%   - User request + active context
    history: 75000               # 37.5% - Conversation history
    reserve: 20000               # 10%   - Safety buffer for responses
```

### Allocation Details

| Category | Tokens | Percentage | Contents |
|----------|--------|------------|----------|
| **System Instructions** | 25,000 | 12.5% | Agent markdown files, protocols, rules |
| **Warmth** | 10,000 | 5% | lessons_learned, codebase_patterns, gotchas |
| **Plan** | 20,000 | 10% | Current PLAN.md being executed |
| **Current Input** | 50,000 | 25% | User's current request, active files |
| **History** | 75,000 | 37.5% | Prior conversation turns, summaries |
| **Reserve** | 20,000 | 10% | Output generation, safety margin |

### Token Estimation

```python
def estimate_tokens(text: str) -> int:
    """
    Estimate token count from text.

    Claude's tokenizer averages ~4 characters per token for mixed
    English/code content.

    Args:
        text: Input text to estimate

    Returns:
        Estimated token count
    """
    return len(text) // 4


def estimate_category_tokens(content: dict) -> dict:
    """
    Estimate tokens for each budget category.

    Args:
        content: Dict with category keys and text values

    Returns:
        Dict with category keys and token estimates
    """
    return {
        category: estimate_tokens(text)
        for category, text in content.items()
    }
```

---

## Compression Triggers

### Threshold Levels

```
Context Usage:  0%         80%         90%        95%        100%
                │          │           │           │           │
                │  NORMAL  │  WARNING  │  COMPRESS │ EMERGENCY │ OVERFLOW
                │          │           │           │           │
                │ Continue │ Log warn  │ Auto-     │ Aggressive│ Block
                │ normally │ continue  │ compress  │ compress  │ spawn
                ▼          ▼           ▼           ▼           ▼
```

### Trigger Thresholds

| Level | Usage | Action | Automatic |
|-------|-------|--------|-----------|
| **NORMAL** | < 80% | Continue normally | N/A |
| **WARNING** | 80-89% | Log warning, continue | Yes |
| **COMPRESS** | 90-94% | Auto-compress history | Yes |
| **EMERGENCY** | 95-99% | Aggressive compression | Yes |
| **OVERFLOW** | >= 100% | Block spawn, require compression | Yes |

### Trigger Implementation

```python
def check_context_triggers(current_usage: int, total_budget: int = 200000) -> TriggerResult:
    """
    Check context usage against compression triggers.

    Args:
        current_usage: Current token count
        total_budget: Total available tokens (default 200K)

    Returns:
        TriggerResult with level and action
    """
    usage_percent = (current_usage / total_budget) * 100

    if usage_percent >= 100:
        return TriggerResult(
            level="OVERFLOW",
            action="block_spawn",
            message=f"Context at {usage_percent:.1f}% - cannot spawn without compression",
            requires_compression=True,
            compression_mode="aggressive"
        )

    if usage_percent >= 95:
        return TriggerResult(
            level="EMERGENCY",
            action="auto_compress",
            message=f"Context at {usage_percent:.1f}% - emergency compression active",
            requires_compression=True,
            compression_mode="aggressive"
        )

    if usage_percent >= 90:
        return TriggerResult(
            level="COMPRESS",
            action="auto_compress",
            message=f"Context at {usage_percent:.1f}% - auto-compressing history",
            requires_compression=True,
            compression_mode="standard"
        )

    if usage_percent >= 80:
        return TriggerResult(
            level="WARNING",
            action="log_and_continue",
            message=f"Context at {usage_percent:.1f}% - approaching limit",
            requires_compression=False,
            compression_mode=None
        )

    return TriggerResult(
        level="NORMAL",
        action="continue",
        message=None,
        requires_compression=False,
        compression_mode=None
    )
```

---

## Compression Strategies

### Strategy 1: Sliding Window

**Use case:** History compression when context grows incrementally.

**Method:** Keep the most recent N messages, archive older ones.

```python
def sliding_window_compress(history: list, keep_recent: int = 20) -> CompressResult:
    """
    Keep most recent messages, archive older ones.

    Args:
        history: List of message objects
        keep_recent: Number of recent messages to keep

    Returns:
        CompressResult with kept and archived messages
    """
    if len(history) <= keep_recent:
        return CompressResult(
            kept=history,
            archived=[],
            tokens_saved=0
        )

    archived = history[:-keep_recent]
    kept = history[-keep_recent:]

    # Estimate tokens saved
    archived_tokens = sum(estimate_tokens(str(m)) for m in archived)

    return CompressResult(
        kept=kept,
        archived=archived,
        tokens_saved=archived_tokens
    )
```

**Configuration:**
```yaml
sliding_window:
  default_keep: 20           # Keep last 20 messages
  emergency_keep: 10         # Keep only 10 in emergency
  preserve_checkpoints: true # Always keep checkpoint messages
  preserve_decisions: true   # Always keep decision messages
```

### Strategy 2: Summarization

**Use case:** When history contains valuable context that shouldn't be lost.

**Method:** LLM-summarize older content into condensed form.

```python
def summarize_compress(history: list, summary_window: int = 30) -> CompressResult:
    """
    Summarize older history into condensed form.

    Args:
        history: List of message objects
        summary_window: Number of messages to summarize at once

    Returns:
        CompressResult with summary + recent messages
    """
    if len(history) <= summary_window:
        return CompressResult(
            kept=history,
            summary=None,
            tokens_saved=0
        )

    to_summarize = history[:-summary_window // 2]
    to_keep = history[-summary_window // 2:]

    # Generate summary (this would use an LLM call)
    summary = generate_history_summary(to_summarize)

    original_tokens = sum(estimate_tokens(str(m)) for m in to_summarize)
    summary_tokens = estimate_tokens(summary)

    return CompressResult(
        kept=[{"role": "summary", "content": summary}] + to_keep,
        summary=summary,
        tokens_saved=original_tokens - summary_tokens
    )


def generate_history_summary(messages: list) -> str:
    """
    Generate a condensed summary of conversation history.

    Summary format:
    - Key decisions made
    - Files modified
    - Critical context
    - Unresolved items
    """
    # This would invoke a summarization task
    summary_prompt = """
    Summarize this conversation history into a condensed form.

    Include:
    1. Key decisions made (with rationale)
    2. Files created or modified
    3. Critical technical context
    4. Unresolved issues or pending items
    5. User preferences expressed

    Keep it concise but preserve critical information.
    Target length: 500-1000 tokens.
    """
    # ... LLM invocation
    return summary
```

**Summary Format:**
```markdown
## History Summary (Messages 1-50)

### Decisions Made
- Chose JWT auth over sessions (security concern)
- Using PostgreSQL with Prisma ORM
- Mobile-first responsive design

### Files Modified
- src/lib/auth.ts (created, JWT implementation)
- prisma/schema.prisma (User model added)
- src/api/login/route.ts (created)

### Critical Context
- User wants explicit error messages
- Database timestamps are UTC
- API routes use req.json() not req.body

### Pending
- CORS configuration needed
- Email verification not yet implemented
```

### Strategy 3: Prioritization

**Use case:** Emergency compression when aggressive token reduction needed.

**Method:** Score messages by relevance, keep highest priority.

```python
def prioritized_compress(history: list, target_tokens: int) -> CompressResult:
    """
    Keep messages by priority score until target reached.

    Priority scoring:
    - Checkpoints: 100
    - User decisions: 90
    - Errors/failures: 80
    - Code changes: 70
    - Progress updates: 50
    - Routine messages: 30

    Args:
        history: List of message objects
        target_tokens: Target token count to achieve

    Returns:
        CompressResult with prioritized messages
    """
    # Score each message
    scored = []
    for msg in history:
        score = calculate_priority_score(msg)
        tokens = estimate_tokens(str(msg))
        scored.append((score, tokens, msg))

    # Sort by priority (highest first)
    scored.sort(key=lambda x: x[0], reverse=True)

    # Keep messages until target reached
    kept = []
    current_tokens = 0
    discarded = []

    for score, tokens, msg in scored:
        if current_tokens + tokens <= target_tokens:
            kept.append(msg)
            current_tokens += tokens
        else:
            discarded.append(msg)

    # Re-sort kept messages by original order
    # (implementation detail omitted)

    original_tokens = sum(estimate_tokens(str(m)) for m in history)

    return CompressResult(
        kept=kept,
        discarded=discarded,
        tokens_saved=original_tokens - current_tokens
    )


def calculate_priority_score(message: dict) -> int:
    """
    Calculate priority score for a message.

    Higher score = more important to keep.
    """
    content = str(message.get('content', ''))

    # Checkpoints are critical
    if 'CHECKPOINT' in content:
        return 100

    # User decisions affect future work
    if message.get('role') == 'user' and any(kw in content.lower() for kw in ['decide', 'choose', 'prefer', 'want']):
        return 90

    # Errors need to be remembered
    if any(kw in content for kw in ['ERROR', 'FAILED', 'FAILURE', 'exception']):
        return 80

    # Code changes are valuable
    if any(kw in content for kw in ['created', 'modified', 'commit', 'git add']):
        return 70

    # File operations
    if any(kw in content for kw in ['wrote', 'edited', 'deleted']):
        return 60

    # Progress updates
    if any(kw in content for kw in ['complete', 'done', 'finished', 'progress']):
        return 50

    # Default
    return 30
```

---

## MC Context Monitoring

### Pre-Spawn Context Gate

Before spawning any agent, MC checks context budget:

```python
def context_gate(spawn_config: dict) -> GateResult:
    """
    Context check gate - runs before EVERY spawn.

    Args:
        spawn_config: Configuration for the spawn

    Returns:
        GateResult with allowed status and any required actions
    """
    # Estimate current context usage
    current_usage = estimate_current_context()

    # Check triggers
    trigger = check_context_triggers(current_usage)

    # Handle based on trigger level
    if trigger.level == "OVERFLOW":
        return GateResult(
            allowed=False,
            message=trigger.message,
            required_action="compress_before_spawn",
            trigger_level=trigger.level
        )

    if trigger.requires_compression:
        # Auto-compress before proceeding
        compression_result = auto_compress(
            mode=trigger.compression_mode,
            target_reduction=0.20  # Target 20% reduction
        )

        return GateResult(
            allowed=True,
            message=f"{trigger.message}. Compressed {compression_result.tokens_saved} tokens.",
            compression_applied=True,
            trigger_level=trigger.level
        )

    if trigger.level == "WARNING":
        # Log but continue
        log_context_warning(trigger.message)

    return GateResult(
        allowed=True,
        message=None,
        trigger_level=trigger.level
    )
```

### Context Budget Display

MC displays context status in progress updates:

```
Context Budget Status
─────────────────────
Usage:   [████████████████░░░░] 82.5%
Level:   WARNING
History: 62,400 / 75,000 tokens
Plan:    18,200 / 20,000 tokens
Warmth:  8,500 / 10,000 tokens

Note: Approaching compression threshold (90%)
```

### Automatic Compression Flow

```
┌─────────────────────────────────────────────────────────────┐
│                    Context Gate Check                        │
└─────────────────────────┬───────────────────────────────────┘
                          │
                          ▼
               ┌──────────────────┐
               │  Usage >= 90%?   │
               └────────┬─────────┘
                       │
           ┌───────────┴───────────┐
           │                       │
           ▼ Yes                   ▼ No
┌──────────────────┐    ┌──────────────────┐
│ Select Strategy  │    │    Continue      │
│ based on mode    │    │    normally      │
└────────┬─────────┘    └──────────────────┘
         │
         ▼
┌──────────────────────────────────────────┐
│  90-94%: Sliding Window (keep 20)        │
│  95-99%: Summarize + Sliding (keep 10)   │
│  100%+:  Prioritize + Block if needed    │
└────────────────────┬─────────────────────┘
                     │
                     ▼
          ┌──────────────────┐
          │ Apply Compression │
          │ Log tokens saved  │
          └────────┬─────────┘
                   │
                   ▼
          ┌──────────────────┐
          │ Proceed with     │
          │ spawn            │
          └──────────────────┘
```

---

## Plan Compression

Plans can grow large. Compress when needed:

### Plan Compression Strategy

```python
def compress_plan(plan_content: str, target_tokens: int) -> str:
    """
    Compress plan while preserving key information.

    Preserves:
    - Objective and must_haves
    - Thread names and verification criteria
    - Critical dependencies

    Removes:
    - Verbose context
    - Detailed rationale
    - Example code (unless critical)
    """
    # Parse plan sections
    sections = parse_plan_sections(plan_content)

    # Always keep these
    essential = [
        sections.get('frontmatter', ''),
        sections.get('objective', ''),
        sections.get('must_haves', ''),
    ]

    # Compress threads to names + verification only
    threads = sections.get('threads', [])
    compressed_threads = []
    for thread in threads:
        compressed_threads.append({
            'id': thread['id'],
            'name': thread['name'],
            'verification': thread.get('verification', ''),
            # Skip: context, implementation_notes, examples
        })

    # Rebuild compressed plan
    return rebuild_plan(essential, compressed_threads)
```

### Plan Compression Rules

| Section | Keep | Compress | Remove |
|---------|------|----------|--------|
| Frontmatter | Always | - | - |
| Objective | Always | - | - |
| Must-haves | Always | - | - |
| Context | If brief | If long | If redundant |
| Threads | Names, verification | Implementation notes | Examples |
| Dependencies | Always | - | - |

---

## Warmth Compression

Warmth accumulates across Programs. Compress when budget exceeded:

### Warmth Compression Strategy

```python
def compress_warmth(warmth_content: dict, target_tokens: int = 10000) -> dict:
    """
    Compress warmth while preserving most valuable learnings.

    Priority order:
    1. gotchas (prevent bugs)
    2. codebase_patterns (essential for work)
    3. user_preferences (quality)
    4. fragile_areas (prevent breakage)
    5. almost_did (nice-to-have)
    """
    # Estimate current warmth tokens
    current_tokens = estimate_tokens(yaml_dump(warmth_content))

    if current_tokens <= target_tokens:
        return warmth_content

    # Compress by category priority
    priority_order = [
        'gotchas',
        'codebase_patterns',
        'user_preferences',
        'fragile_areas',
        'almost_did'
    ]

    compressed = {}
    running_tokens = 0

    for category in priority_order:
        if category not in warmth_content:
            continue

        items = warmth_content[category]
        category_tokens = estimate_tokens(yaml_dump({category: items}))

        if running_tokens + category_tokens <= target_tokens:
            compressed[category] = items
            running_tokens += category_tokens
        else:
            # Partial include - keep most valuable items
            remaining_budget = target_tokens - running_tokens
            truncated = truncate_category(items, remaining_budget)
            if truncated:
                compressed[category] = truncated
            break

    return compressed
```

---

## Token Counting and Logging

### Context Usage Logging

Log context usage for monitoring:

```yaml
# .grid/context_log.yaml
sessions:
  - session_id: "sess-20260124-100000"
    events:
      - timestamp: "2026-01-24T10:00:00Z"
        type: "spawn"
        agent: "planner"
        context_before: 45000
        context_after: 72000
        trigger_level: "NORMAL"

      - timestamp: "2026-01-24T10:15:00Z"
        type: "spawn"
        agent: "executor"
        context_before: 165000
        context_after: 148000
        trigger_level: "COMPRESS"
        compression_applied: true
        tokens_saved: 23000
        strategy: "sliding_window"
```

### Monitoring Dashboard

```
CONTEXT MONITORING
══════════════════

Current Session: sess-20260124-100000
Duration: 2h 15m

Usage History:
  10:00 ████░░░░░░░░░░░░░░░░ 22.5% (spawn: planner)
  10:15 ████████░░░░░░░░░░░░ 36.0% (spawn: executor)
  10:45 ████████████░░░░░░░░ 58.5% (spawn: recognizer)
  11:00 ████████████████░░░░ 82.5% (WARNING)
  11:15 ████████████████████ 92.5% (COMPRESS - saved 23K)
  11:20 ████████████░░░░░░░░ 69.0% (after compression)

Compression Events:
  1. 11:15 - Sliding window (23,000 tokens saved)

Recommendations:
  - Consider breaking large blocks into smaller ones
  - Current warmth size is optimal
```

---

## Integration Points

### With Budget System

Context management integrates with cost budgeting:

```python
def combined_gate_check(spawn_config: dict) -> GateResult:
    """
    Combined budget + context gate.
    """
    # Check cost budget first
    cost_result = budget_gate(spawn_config)
    if not cost_result.allowed:
        return cost_result

    # Then check context budget
    context_result = context_gate(spawn_config)
    if not context_result.allowed:
        return context_result

    # Both passed
    return GateResult(
        allowed=True,
        cost_estimate=cost_result.estimated_cost,
        context_usage=context_result.current_usage,
        warnings=[
            w for w in [cost_result.message, context_result.message] if w
        ]
    )
```

### With Warmth Transfer

Warmth is compressed before transfer to fresh agents:

```python
def prepare_warmth_for_transfer(warmth: dict) -> dict:
    """
    Prepare warmth for transfer to new agent.

    Compress if needed to stay within budget.
    """
    current_tokens = estimate_tokens(yaml_dump(warmth))
    budget = TOKEN_BUDGET['warmth']  # 10,000 tokens

    if current_tokens > budget:
        return compress_warmth(warmth, budget)

    return warmth
```

### With Checkpoint Handling

Context is managed specially at checkpoints:

```python
def handle_checkpoint_context(checkpoint: dict) -> None:
    """
    Manage context at checkpoint boundaries.

    Checkpoints are natural compression points.
    """
    # Archive completed work context
    archive_history_to_summary()

    # Compress warmth for continuation
    compressed_warmth = compress_warmth(
        checkpoint.get('warmth', {}),
        target_tokens=8000  # Leave room for user response
    )

    # Store checkpoint context
    checkpoint['compressed_warmth'] = compressed_warmth
```

---

## Best Practices

### For Executors

1. **Be context-aware** - Check current usage before verbose operations
2. **Write concise summaries** - Compress learnings for SUMMARY.md
3. **Avoid redundancy** - Don't repeat context already in warmth
4. **Use structured output** - Easier to compress than prose

### For MC

1. **Monitor continuously** - Check context before every spawn
2. **Compress proactively** - Don't wait for overflow
3. **Preserve critical context** - Checkpoints, decisions, errors
4. **Log compression events** - For debugging and optimization

### For Plans

1. **Keep plans focused** - One clear objective per plan
2. **Use references** - Link to docs instead of inlining
3. **Separate context** - Don't embed large code samples
4. **Version summaries** - Compress completed blocks

---

## Configuration

### Context Management Config

```yaml
# .grid/context_config.yaml
context_management:
  enabled: true
  total_budget: 200000

  triggers:
    warning: 0.80
    compress: 0.90
    emergency: 0.95
    overflow: 1.00

  strategies:
    default: "sliding_window"
    emergency: "prioritize"

  sliding_window:
    default_keep: 20
    emergency_keep: 10

  summarization:
    enabled: true
    min_messages: 30
    target_length: 1000  # tokens

  logging:
    enabled: true
    verbose: false
    log_file: ".grid/context_log.yaml"
```

---

## Appendix: Token Estimation Accuracy

| Content Type | Actual Chars/Token | Estimate Used | Error Range |
|--------------|-------------------|---------------|-------------|
| English prose | 4.5 | 4 | +12.5% |
| Code | 3.5 | 4 | -14% |
| Mixed | 4.0 | 4 | +/- 5% |
| YAML/JSON | 3.8 | 4 | -5% |
| Markdown | 4.2 | 4 | +5% |

**Conservative approach:** Using 4 chars/token provides slight over-estimation, which is safer for budget management.

---

*End of Line.*