How to Implement Custom CompactStrategy Subclasses in Forge

To implement a custom CompactStrategy subclass, extend the abstract base class CompactStrategy in src/forge/context/strategies.py and implement the compact() method that returns a tuple of (new message list, phase integer) while preserving the system prompt and original user input.

Forge uses compact strategies to enforce token budgets during LLM conversations. When context windows fill up, these strategies decide which historical messages to prune while keeping critical information intact. By creating custom CompactStrategy subclasses, you can tailor the compaction logic to your specific workflow needs—whether that means keeping recent assistant messages, truncating tool results, or applying aggressive multi-phase filtering.

Understanding the CompactStrategy Interface

The compaction abstraction lives in src/forge/context/strategies.py and centers on the CompactStrategy abstract base class (ABC). This interface defines a single required method that all implementations must provide:

class CompactStrategy(ABC):
    @abstractmethod
    def compact(
        self,
        messages: list[Message],
        budget_tokens: int,
        *,
        step_hint: str = "",
    ) -> tuple[list[Message], int]:
        ...

A ContextManager instance holds a concrete strategy and invokes compact() via maybe_compact() whenever token usage approaches the configured limit. The method receives the full conversation history and must return a new list (never mutating the original) plus an integer phase indicating compaction aggressiveness (0 means no changes were made).

Core Requirements for Custom Implementations

Your subclass must satisfy four specific contracts to remain compatible with Forge’s context management pipeline:

1. Method Signature – Implement exactly compact(messages, budget_tokens, *, step_hint="") → tuple[list[Message], int].

2. Preservation Guarantees – Never drop messages[0] (the system prompt) or messages[1] (the original user input). These indexes are protected by convention.

3. Token-Budget Logic – Use the supplied budget_tokens alongside the helper _estimate_tokens(messages) to decide whether compaction is necessary.

4. Return Value Contract – Return a copy of the message list (e.g., list(messages)) and a phase number. Return phase 0 if unchanged, otherwise 1 (or higher values for custom phase indicators).

The repository includes three reference implementations—NoCompact, SlidingWindowCompact, and TieredCompact—that demonstrate these rules in practice.

Building a Custom Compact Strategy

Minimal Implementation Example

Below is a complete custom strategy called RecentOnlyCompact. It retains the system prompt, the initial user message, and only the most recent N assistant messages, discarding everything else when token usage exceeds a threshold.

from forge.context.strategies import CompactStrategy, _estimate_tokens
from forge.core.messages import Message, MessageRole

class RecentOnlyCompact(CompactStrategy):
    """Keep only the last *keep_recent* assistant messages.

    This strategy is useful when the workflow is shallow and you only need
    the most recent assistant output to stay in context.
    """

    def __init__(self, keep_recent: int = 3, compact_threshold: float = 0.80):
        self.keep_recent = keep_recent
        self.compact_threshold = compact_threshold

    def compact(
        self,
        messages: list[Message],
        budget_tokens: int,
        *,
        step_hint: str = "",
    ) -> tuple[list[Message], int]:
        # Estimate current token usage

        current_tokens = _estimate_tokens(messages)

        # If below threshold, do nothing

        if current_tokens < int(budget_tokens * self.compact_threshold):
            return list(messages), 0

        # Identify indexes of the last keep_recent assistant messages

        recent_assistant_idxs = [
            i for i, m in enumerate(messages) if m.role == MessageRole.ASSISTANT
        ][-self.keep_recent :]

        # Build compacted list: keep first two messages, then recent assistants

        compacted = [
            m
            for i, m in enumerate(messages)
            if i < 2 or i in recent_assistant_idxs
        ]

        # Return new list and phase 1

        return compacted, 1

Extending Built-in Strategies

For workflows requiring the multi-phase approach of TieredCompact but with bespoke rules, subclass the existing strategy and override specific phase helpers. This example drops tool results entirely in Phase 1 rather than truncating them:

from forge.context.strategies import TieredCompact
from forge.core.messages import MessageType

class HybridTieredCompact(TieredCompact):
    """Same three-phase logic as TieredCompact but drops tool results
    completely in Phase 1 rather than truncating them.
    """

    def _phase1(self, messages, eligible_end):
        result = []
        for i, msg in enumerate(messages):
            if 2 <= i < eligible_end and msg.metadata.type == MessageType.TOOL_RESULT:
                # Skip the whole tool result instead of truncating

                continue
            result.append(msg)
        return result

Integrating Your Strategy with ContextManager

Once defined, pass your custom strategy to the ContextManager constructor alongside your token budget. The manager handles the orchestration:

from forge.context.manager import ContextManager

ctx = ContextManager(
    strategy=RecentOnlyCompact(keep_recent=4),
    budget_tokens=4000,
)

# During your workflow, the manager automatically evaluates compaction:

messages = ctx.maybe_compact(messages, step_index=current_step)

The ContextManager delegates to your strategy’s compact() method only when internal heuristics suggest the budget might be exceeded, respecting the compact_threshold you define within your implementation.

Key Source Files Reference

When implementing custom CompactStrategy subclasses, reference these Forge source files:

• src/forge/context/strategies.py – Defines the CompactStrategy ABC and built-in implementations (NoCompact, SlidingWindowCompact, TieredCompact).
• src/forge/core/messages.py – Provides Message, MessageMeta, MessageRole, and MessageType enums used for filtering logic.
• src/forge/context/manager.py – Houses ContextManager, which coordinates token budgeting and delegates to your compact strategy.
• tests/unit/test_strategies.py – Unit tests documenting expected behaviors; use this as a reference for assertions your custom strategy should satisfy.

Summary

• Extend CompactStrategy from src/forge/context/strategies.py to create custom compaction logic.
• Implement the compact() method with the exact signature (list[Message], int, *, str) -> tuple[list[Message], int].
• Always preserve messages[0] (system prompt) and messages[1] (original user input) in your output.
• Return a new list (never mutate input) and a phase integer (0 for no change, 1+ for custom phases).
• Leverage _estimate_tokens() to evaluate token counts against budget_tokens.
• Instantiate your strategy and pass it to ContextManager to activate it in your workflow.

Frequently Asked Questions

What does the phase integer returned by compact() represent?

The phase integer indicates how aggressive the compaction was. Return 0 when no messages are removed. Return 1 when your strategy performs its standard compaction. You may define additional phase values (e.g., 2, 3) to signal increasingly aggressive pruning, similar to how TieredCompact uses multiple phases.

Can a custom strategy modify the messages list in place?

No. The compact() method must return a new list containing the filtered messages. Mutating the input list violates the contract expected by ContextManager and can cause unexpected side effects in your conversation history. Use list(messages) or list comprehensions to generate the output.

How do I test a custom CompactStrategy implementation?

Write unit tests in a file similar to tests/unit/test_strategies.py. Verify that your implementation preserves messages[0] and messages[1], returns phase 0 when under budget, correctly estimates tokens using _estimate_tokens(), and returns a new list instance rather than the original reference.

Should I subclass TieredCompact or CompactStrategy for multi-phase logic?

Subclass TieredCompact if you want to reuse its three-phase framework (removing nudges, then tool results, then reasoning) while overriding specific phase behaviors. Subclass the base CompactStrategy directly if you need a completely custom algorithm that does not follow the tiered pattern, such as the RecentOnlyCompact example above.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}