Compact Memory Implementation Skill for Claude

COMPACT_THRESHOLD = 150_000  # adjust per model

if response.usage.input_tokens > COMPACT_THRESHOLD:
    compact = compact_memory(history)
    history = []  # reset — compact moves to system prompt

COMPACT_EVERY_N = 30

if turn_count % COMPACT_EVERY_N == 0:
    compact = compact_memory(history)

def compact_memory(history: list[dict]) -> dict:
    response = client.messages.create(
        model="claude-haiku-4-5-20251001",  # cheaper model is fine for compaction
        max_tokens=4096,
        system=COMPACTOR_SYSTEM_PROMPT,
        messages=[
            {
                "role": "user",
                "content": format_history_for_compact(history),
            }
        ],
    )
    return json.loads(response.content[0].text)

{
  "task": "What the agent is working on and why — the goal, not the steps",
  "current_state": "Exact status at compaction point: what is done, what is not, what is in progress",
  "key_decisions": [
    { "decision": "...", "reason": "...", "constraint": "..." }
  ],
  "eliminated_approaches": [
    { "approach": "...", "reason_ruled_out": "..." }
  ],
  "open_questions": ["..."],
  "next_steps": ["..."],
  "relevant_tool_results": {
    "key": "Only results future steps will need — summarized, not raw dumps"
  },
  "compacted_at_turn": 42
}

You are a conversation compactor. Read the provided conversation and produce a JSON summary that captures everything a fresh agent needs to continue the work without asking what happened.

Include:
- Current task and goal (not the steps taken to get here)
- Exact current state — what is done and what is not
- Decisions made and WHY (reasoning, not just the choice)
- Approaches tried and ruled out with reasons (prevents re-exploration)
- Open questions and blockers
- Concrete next steps in priority order
- Tool results that future steps will need (summarize, don't dump raw output)

Omit:
- Intermediate reasoning that led nowhere
- Completed sub-tasks with no future relevance
- Raw tool output that has already been acted on
- Anything derivable by reading the code or running a command

Output valid JSON matching the schema provided. No prose outside the JSON.

def format_history_for_compact(history: list[dict]) -> str:
    lines = ["Conversation to compact:\n"]
    for msg in history:
        role = msg["role"].upper()
        content = msg["content"] if isinstance(msg["content"], str) else "[tool use]"
        lines.append(f"[{role}]: {content[:2000]}")  # cap very long messages
    return "\n".join(lines)

MEMORY_BLOCK_TEMPLATE = """
## Restored memory (compacted at turn {turn})

**Task**: {task}

**Current state**: {current_state}

**Key decisions**:
{decisions}

**Ruled out approaches**:
{eliminated}

**Next steps**:
{next_steps}

Begin from current state above. Do not re-explore eliminated approaches.
"""

def build_system_with_memory(base_system: str, compact: dict | None) -> str:
    if compact is None:
        return base_system
    memory = MEMORY_BLOCK_TEMPLATE.format(
        turn=compact["compacted_at_turn"],
        task=compact["task"],
        current_state=compact["current_state"],
        decisions="\n".join(f"- {d['decision']} (because {d['reason']})"
                            for d in compact["key_decisions"]),
        eliminated="\n".join(f"- {e['approach']}: {e['reason_ruled_out']}"
                             for e in compact["eliminated_approaches"]),
        next_steps="\n".join(f"- {s}" for s in compact["next_steps"]),
    )
    return base_system + "\n\n" + memory

messages = [
    {
        "role": "user",
        "content": f"[Resuming from compacted state — turn {compact['compacted_at_turn']}]\n"
                   f"{json.dumps(compact, indent=2)}\n\n"
                   f"Continue from the next steps listed above.",
    }
]

import json, pathlib

MEMORY_DIR = pathlib.Path("memory")
MEMORY_DIR.mkdir(exist_ok=True)

def save_compact(session_id: str, compact: dict) -> None:
    (MEMORY_DIR / f"{session_id}.json").write_text(json.dumps(compact, indent=2))

def load_compact(session_id: str) -> dict | None:
    path = MEMORY_DIR / f"{session_id}.json"
    return json.loads(path.read_text()) if path.exists() else None

def run_agent(session_id: str, user_input: str) -> str:
    compact = load_compact(session_id)
    system = build_system_with_memory(BASE_SYSTEM, compact)
    history = []
    turn = 0

    while True:
        response = client.messages.create(
            model="claude-opus-4-7",
            system=system,
            messages=history + [{"role": "user", "content": user_input}],
            max_tokens=8192,
        )

        # Trigger compact if context is growing too large
        if response.usage.input_tokens > COMPACT_THRESHOLD:
            compact = compact_memory(history)
            save_compact(session_id, compact)
            system = build_system_with_memory(BASE_SYSTEM, compact)
            history = []  # reset history — compact is now in system
            turn = 0
            continue

        if response.stop_reason == "end_turn":
            return response.content[0].text

        history.append({"role": "assistant", "content": response.content})
        user_input = handle_tool_calls(response)  # your tool dispatch
        turn += 1

COMPACTOR_WITH_PRIOR = """
You are updating an existing memory compact with new information from a continuation session.

Prior compact:
{prior_compact}

New conversation turns since last compact:
{new_turns}

Produce an updated compact that:
- Merges both sources
- Removes resolved items and completed steps
- Adds new decisions, eliminations, and open questions
- Keeps next_steps current

Output valid JSON. No prose outside the JSON.
"""

def compact_memory_with_prior(history: list[dict], prior: dict) -> dict:
    prompt = COMPACTOR_WITH_PRIOR.format(
        prior_compact=json.dumps(prior, indent=2),
        new_turns=format_history_for_compact(history),
    )
    response = client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=4096,
        system=prompt,
        messages=[{"role": "user", "content": "Update the compact."}],
    )
    return json.loads(response.content[0].text)