How to Implement Session Memory Compaction for Long-Running Claude Conversations

Use background threading to generate summaries proactively, then swap them instantly when the token limit approaches to prevent context loss and user-visible latency.

The anthropics/claude-cookbooks repository demonstrates production-ready patterns for handling Claude's finite context window in long-running conversations. When conversations exceed the model's token limit, naive truncation causes silent loss of earlier context, breaking stateful assistants. The solution lies in session memory compaction—intelligently summarizing conversation history before the limit is hit.

Why Session Memory Compaction Matters

Claude's context window is finite (approximately 100k tokens for most models). Without compaction, exceeding this limit forces the API to drop older messages, erasing critical context like user preferences, completed tasks, or error history. The notebook misc/session_memory_compaction.ipynb provides two architectural approaches to solve this: traditional on-demand compaction and proactive instant compaction.

Traditional vs. Instant Compaction Strategies

The repository implements two distinct patterns for managing session memory:

• TraditionalCompactingChatSession – Waits until the token limit is breached, then pauses the conversation to generate a summary. This creates user-visible latency but requires simpler code.
• InstantCompactingChatSession – Uses a background thread to maintain a pre-generated summary as the conversation grows. When the limit approaches, the swap happens instantly with zero perceived delay.

For production-grade assistants that must remain responsive, the instant pattern is preferred because it eliminates blocking operations during active turns.

Implementing the InstantCompactingChatSession

The InstantCompactingChatSession class in misc/session_memory_compaction.ipynb provides the complete implementation for proactive memory management.

Initialization and Token Thresholds

The session configures three critical thresholds to manage when compaction occurs:

class InstantCompactingChatSession:
    def __init__(
        self,
        system_message: str = "You are a helpful assistant",
        context_limit: int = 12_000,
        min_tokens_to_init: int = 7_500,
        min_tokens_between_updates: int = 2_000,
    ):
        self.context_limit = context_limit
        self.min_tokens_to_init = min_tokens_to_init
        self.min_tokens_between_updates = min_tokens_between_updates
        self.messages: list[dict] = []
        self.session_memory: str | None = None
        self.last_summarized_index = 0
        self.tokens_at_last_update = 0
        self._lock = threading.Lock()
        self._update_thread: threading.Thread | None = None

The context_limit (12,000 tokens in the example) triggers the compaction, while min_tokens_to_init (7,500) determines when to create the first background summary, and min_tokens_between_updates (2,000) controls how frequently to refresh that summary.

The Chat Loop with Background Updates

The public chat() method handles user input while managing background compaction threads:

def chat(self, user_message: str) -> tuple[str, anthropic.types.Usage, str | None]:
    # Check if we need to compact before adding the new message

    if self.current_context_window_tokens + estimate_tokens(user_message) >= self.context_limit:
        self.compact()  # Instant swap because session_memory is ready

    self.messages.append({"role": "user", "content": user_message})
    
    response = client.messages.create(
        model=MODEL,
        max_tokens=3_500,
        system=self.system_message,
        messages=add_cache_control(self.messages),
    )
    
    assistant = response.content[0].text
    self.messages.append({"role": "assistant", "content": assistant})
    
    # Update token accounting

    cache_read = getattr(response.usage, "cache_read_input_tokens", 0) or 0
    total_input = response.usage.input_tokens + cache_read
    self.current_context_window_tokens = total_input + response.usage.output_tokens
    
    # Trigger background update if thresholds met

    background_status = None
    if self._should_init_memory() or self._should_update_memory():
        self._trigger_background_update()
        background_status = "initializing" if self.session_memory is None else "updating"
    
    return assistant, response.usage, background_status

The method checks _should_init_memory() (returns True when tokens exceed 7,500 and no memory exists) and _should_update_memory() (returns True when 2,000+ new tokens have accumulated since the last update) to determine when to spawn background threads.

The Compaction Method

When thresholds are breached, the compact() method performs an atomic swap of the conversation history:

def compact(self) -> None:
    """Swap the ready-made session memory into the conversation."""
    if not self.session_memory:
        # Fallback: generate on-the-fly if background never initialized

        self.session_memory = self._create_session_memory(self.messages)

    # Replace entire transcript with memory turn

    self.messages = [
        {
            "role": "user",
            "content": f"""This session is being continued from a previous conversation. 
Here is the session memory: {self.session_memory}. 
Continue from where we left off."""
        }
    ]
    self.current_context_window_tokens = estimate_tokens(self.session_memory)

This replacement strategy preserves essential context while resetting the token count to the summary size, effectively compressing thousands of tokens into a dense paragraph.

Creating and Updating Session Memory

The class uses two distinct methods to manage the summary lifecycle. Initial creation processes the full transcript:

def _create_session_memory(self, msgs: list[dict]) -> str:
    compaction_msg = [{"role": "user", "content": SESSION_MEMORY_PROMPT}]
    resp = client.messages.create(
        model=MODEL,
        max_tokens=5_000,
        system=self.system_message,
        messages=add_cache_control(msgs) + compaction_msg,
    )
    summary, _ = remove_thinking_blocks(resp.content[0].text)
    return summary

Incremental updates append new messages to the existing summary rather than reprocessing the entire history:

def _update_session_memory(self, new_msgs: list[dict]) -> str:
    compaction_msg = [{"role": "user", "content": SESSION_MEMORY_PROMPT}]
    resp = client.messages.create(
        model=MODEL,
        max_tokens=5_000,
        system=self.system_message,
        messages=add_cache_control(
            [{"role": "assistant", "content": self.session_memory}] + new_msgs
        ) + compaction_msg,
    )
    summary, _ = remove_thinking_blocks(resp.content[0].text)
    return summary

The Session Memory Prompt

At lines 63-77 of misc/session_memory_compaction.ipynb, the SESSION_MEMORY_PROMPT instructs Claude to preserve specific elements during compression: user intent, completed work, errors encountered, active work in progress, and key identifiers like IDs or names. This structured prompt ensures the summary maintains actionable context rather than generic conversational fluff.

Performance Optimization with Prompt Caching

The implementation leverages Claude's prompt caching feature to reduce latency and cost during compaction. The add_cache_control() function marks the last user message with cache_control: {"type": "ephemeral"}:

def add_cache_control(messages: list[dict]) -> list[MessageParam]:
    last_user = max(
        (i for i, m in enumerate(messages) if m["role"] == "user"), 
        default=-1
    )
    out = []
    for i, msg in enumerate(messages):
        txt = msg["content"] if isinstance(msg["content"], str) else msg["content"][0]["text"]
        block: TextBlockParam = {"type": "text", "text": txt}
        if i == last_user:
            block["cache_control"] = {"type": "ephemeral"}
        out.append({"role": msg["role"], "content": [block]})
    return out

By caching the conversation history, subsequent compaction requests hit the cache rather than reprocessing the full context from scratch, dramatically reducing both token costs and response time for the _create_session_memory() and _update_session_memory() calls.

Summary

• Session memory compaction prevents context loss in long-running Claude conversations by summarizing history before the token limit is reached.
• The InstantCompactingChatSession class uses background threads to maintain a ready-to-swap summary, eliminating user-visible latency.
• Threshold-based logic (min_tokens_to_init, min_tokens_between_updates) optimizes when to generate or refresh summaries without wasting API calls.
• Prompt caching via ephemeral cache control on the last user turn reduces costs for compaction operations that re-read conversation history.
• For production deployments, persist session_memory to disk after each update to survive process restarts and maintain bounded memory usage.

Frequently Asked Questions

How does session memory compaction differ from standard context window truncation?

Standard truncation silently drops older messages when the context limit is exceeded, causing permanent loss of earlier context. Session memory compaction proactively generates a semantic summary containing user intent, completed tasks, errors, and active work, then swaps the full transcript for this condensed version. This preserves essential state while freeing up token budget for continued conversation.

What token thresholds should I use for production applications?

The reference implementation uses a context_limit of 12,000 tokens (well below Claude's 100k maximum to allow headroom), initializes background memory at 7,500 tokens (min_tokens_to_init), and updates every 2,000 tokens thereafter (min_tokens_between_updates). Adjust these based on your expected turn length and latency requirements, ensuring the context_limit leaves sufficient room for the model's response and the compaction prompt itself.

Can I recover the original conversation messages after compaction?

No, the compact() method intentionally replaces the message list with a single user turn containing the session memory summary. If you need to retain full transcripts for audit or debugging purposes, you must persist the messages array to external storage (such as a database or file) before calling compact(), or maintain a parallel logging mechanism outside the session state.

Does the background threading approach introduce race conditions?

The implementation uses self._lock = threading.Lock() to protect the session_memory, last_summarized_index, and tokens_at_last_update variables during updates. The _trigger_background_update() method checks self._update_thread.is_alive() to prevent spawning duplicate threads if a previous update is still running. This ensures thread-safe operation when the main chat loop requests updates while background processing is active.

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