Code Agent Anatomy (12): Harness Design Part 2 — Context Engineering

The Loose End from Last Time

The previous post (Part 11) dissected the control flow: a single main loop, an immutable state machine, and completion gates. One line of code was glossed over:

# loop.py _prepare_step_context()
model_view = host.context_engine.build_model_view(...)
messages = model_view.messages

What build_model_view() actually does is far more than "fetch the message history." A long-running agent accumulates an ever-growing conversation history — exceed the token limit and it errors out. The subtler point: even before hitting the limit, the agent must decide each step — what exactly should get sent to the model? System prompt, history, tool outputs… cramming everything in is no better than careful curation.

This post dissects that pipeline: from "sensing we're almost out of headroom" to "deciding what the model gets to see."

The Conclusion First

MyCodeAgent breaks this problem into three cleanly separated concerns:

Three layers, clear responsibilities, no cross-contamination. Compaction doesn't mean deleting history — it means reading history a different way.

1. History Is a Fact Log — Never Modified

# runtime/history.py
class HistoryManager:
    def __init__(self, ...):
        self._messages: List[Message] = []  # append-only, never deleted
    
    def append_user(self, content: str, ...) -> Message: ...
    def append_assistant(self, content: str, ...) -> Message: ...
    def append_tool_result(self, ...) -> Message: ...
    
    def get_messages(self) -> List[Message]:
        return list(self._messages)  # returns a copy — caller cannot mutate

HistoryManager is a pure append-only fact list. It has no knowledge of "compaction," and no message ever disappears because of it.

This matters: it means that when an agent crashes, the complete history can be reconstructed from the transcript — nothing is ever lost to a compaction operation (see Part 10).

2. The Compaction Decision: How to Tell "We're Almost Full"

# runtime/context/budget.py
class ContextBudgetPolicy:
    def should_compact(self, *, messages, pending_input, last_usage_tokens) -> CompactDecision:
        # default: 128000 × 0.8 = 102400 tokens
        threshold = int(self.config.context_window * self.config.compression_threshold)
 
        # Source 1: estimate from message content (character count // 3)
        estimated_from_messages = self.estimate_tokens(messages, pending_input)
        # Source 2: actual usage from last LLM call + new input
        estimated_from_usage = int(last_usage_tokens or 0) + len(pending_input or "") // 3
        
        # take the more pessimistic estimate
        estimated = max(estimated_from_messages, estimated_from_usage)
 
        if message_count < 3:
            return CompactDecision(False, "messages_not_enough", ...)
        if estimated < threshold:
            return CompactDecision(False, "below_threshold", ...)
        return CompactDecision(True, "threshold_exceeded", ...)

Two design details worth noting:

1. Why not count tokens precisely?

Exact token counting requires calling a tokenizer, which is model-specific — different models tokenize differently, and calling it has overhead. character count // 3 is a conservative approximation (an average across mixed Chinese/English text). Better to compact early than to react after an error.

2. Why take the max of two sources?

Estimating purely from the message list underestimates — tool_calls JSON, tool_name fields, and other metadata all consume tokens but don't live in content. Meanwhile, last_usage_tokens is the actual number the LLM consumed on the previous step — more accurate, but it only knows the past, not the current new input. Taking max is an engineering habit of pessimistic estimation: compact a little early rather than not at all.

3. How Compaction Works: Non-Destructive Checkpoints

# runtime/context/compact.py
class ContextCompactor:
    def compact(self, messages: list[Message]) -> dict:
        # 1. segment messages into turns at user-message boundaries
        rounds = self.round_segmenter.identify(source_messages)
        
        # 2. retain the most recent min_retain_rounds turns verbatim (default: 10)
        retain_start_round = len(rounds) - min_rounds
        retain_start_idx = rounds[retain_start_round].start_idx
        messages_to_compact = source_messages[:retain_start_idx]
        
        # 3. call LLM to summarize the "old" messages
        summary = self.summary_generator(messages_to_compact)
        
        # 4. store summary + split-point in CompactStore
        checkpoint = self.compact_store.create_checkpoint(
            summary=summary,
            retain_start_idx=retain_start_idx,
            ...
        )
        # note: source_messages is never modified

The keyword here is non-destructive: compaction does not modify any message in HistoryManager. It only writes a checkpoint to CompactStore recording "what the summary says" and "which index marks the start of verbatim history."

RoundSegmenter divides the message list into turns at user-message boundaries. Compaction granularity is always a complete turn — splitting a user and assistant message from the same exchange would break semantic continuity.

4. Read-Time Projection: What the Model Sees ≠ What's Stored

# runtime/context/projection.py
class ProjectionBuilder:
    def project(self, source_messages: list[Message]) -> ProjectionResult:
        checkpoint = self.compact_store.active_checkpoint
        
        if not checkpoint:
            # no compaction yet: projection = full history
            return ProjectionResult(messages=source, projection_mode="full_history")
        
        # checkpoint exists: projection = [summary message] + source[retain_start_idx:]
        summary = Message(content=checkpoint.summary, role="summary", ...)
        return ProjectionResult(
            messages=[summary] + source[retain_start_idx:],
            projection_mode="compact_checkpoint",
        )

This is the single most important step in the whole context engineering stack: read-time projection.

If source_messages has 200 entries and the checkpoint's retain_start_idx is 120, the projected result is:

[summary (covering the first 120 messages)] + [source[120:] — the most recent 80 messages verbatim]

The model sees 81 messages (1 summary + 80 verbatim), but HistoryManager still holds all 200 originals.

5. MessageNormalizer: Format Conversion

History stores Message objects (the runtime's internal format); the LLM API expects a list of OpenAI-style dicts. MessageNormalizer handles that conversion:

# runtime/context/normalizer.py
class MessageNormalizer:
    def _normalize_one(self, msg: Message) -> list[dict]:
        if msg.role == "user":
            return [{"role": "user", "content": msg.content}]
        if msg.role == "assistant":
            return [self._assistant_message(msg)]  # restores tool_calls
        if msg.role == "tool":
            return [self._tool_message(msg)]        # links tool_call_id
        if msg.role == "summary":
            # summary becomes a system message, prepended to history
            return [{"role": "system", "content": f"## Archived History Summary\n{msg.content}"}]

A summary-role message is converted to a system-role message before being sent to the model — the model treats it as background context rather than conversational history. That's a subtle but intentional semantic choice.

6. ModelView: What the Model Saw on This Step

All of the above converges in ContextEngine.build_model_view(), which produces a ModelView:

# runtime/context/engine.py (core logic of build_model_view)
def build_model_view(self, *, history_manager, pending_input, ...) -> ModelView:
    # 1. Bounded history projection (projection + normalize)
    source_messages = history_manager.get_messages()
    projection = self.projection_builder.project(source_messages)
    history_messages = self.normalizer.normalize(projection.messages)
 
    # 2. System layer (agent persona + tool contracts + project rules)
    system_messages = self.context_builder.get_system_messages()
 
    # 3. Dynamic system: Session Memory (cross-run summaries)
    dynamic_messages = [{"role": "system", "content": rendered}] if self.session_memory else []
 
    # 4. Assemble: system first, history last
    messages = list(system_messages) + dynamic_messages + list(history_messages)
 
    return ModelView(
        messages=messages,
        system_message_count=len(system_messages),
        history_message_count=len(history_messages),
        source_message_count=projection.source_message_count,
        projection_mode=projection.projection_mode,
        ...
    )

Final message order sent to the model:

[Constitution][Tool Contracts][Code Law]   ← system layer (Part 04)
[Session Memory?]                           ← cross-run summary (Part 10)
[summary? + most recent N turns verbatim]  ← history projection

ModelView is more than a message list — it carries metadata: projection_mode (was it compacted?), source_message_count vs. history_message_count (before vs. after compaction counts). These feed the trace, letting you observe "how much the model actually saw this step."

7. Two Trigger Paths

Two places in the loop can trigger compaction:

Per-step (proactive)              On model call (reactive)
        ↓                                  ↓
compact_if_needed()            classify exception → PROMPT_TOO_LONG
  → budget estimate > threshold  → reactive_compact()
  → compactor.compact()          → compactor.compact()
  → record checkpoint             → rebuild model view, inner continue retry

Proactive compaction (compact_if_needed) runs before each step, prior to assembling the model view. If the estimate says we're almost out of headroom, compact first, then assemble — the model gets the compacted version on the next step.

Reactive compaction (reactive_compact) is the fallback. If the estimate didn't trigger but the model actually returns PROMPT_TOO_LONG, force-compact immediately, rebuild the model view, and continue to retry the current step. There's a retry limit; if exhausted, the agent takes the MODEL_RECOVERY_FAILED termination path.

Design Highlights

1. History and ModelView are separate

HistoryManager is an immutable fact log; ModelView is a per-step ephemeral view. Because they're separate, compaction never loses history, crashes can be fully recovered, and during debugging you can compare "what was stored" against "what the model actually saw."

2. Read-time projection, not write-time deletion

Compaction results are stored as a checkpoint (summary + split index), folded in dynamically each time history is read. This is safer than "delete messages on compact" — you can roll back at any time (clear the checkpoint and you're back to full history), and a compaction bug can never corrupt history.

3. Dual-path safety net

Proactive estimation + reactive exception catching: both can trigger compaction. Even if the estimation algorithm has error, the agent self-heals on a model error rather than crashing.

Summary

About the Source Code

All analysis in this series is based on the open-source project MyCodeAgent.

The source code includes comments at key locations aligned with the walkthrough in each post — you can read the articles alongside the code, or clone it directly to run, modify, and extend, building your own agent on top of it.

git clone https://github.com/chendongqi/MyCodeAgent
cd MyCodeAgent
cp .env.example .env   # fill in your LLM API key
uv sync
uv run python main.py

Explore PrimeSkills — a curated marketplace of AI agents and skills, each validated against real enterprise workflows. No hype, just what actually works.

Visit my personal site for more insights and interesting products.