Folded Memory Implementation Skill for Claude

L1 Working memory    [ turn 38..50 ] — raw turns, full detail, short window
L2 Episodic memory   [ turn 10..37 ] — compressed episodes, medium detail
L3 Semantic memory   [ turn 1..9  ] — abstract facts and decisions, sparse

from dataclasses import dataclass, field
from typing import Any

@dataclass
class Episode:
    episode_id: int
    turn_range: tuple[int, int]       # (start_turn, end_turn)
    summary: str
    decisions: list[dict]             # [{"decision": ..., "reason": ...}]
    eliminated: list[dict]            # [{"approach": ..., "why": ...}]
    open_questions: list[str]
    tool_results: dict[str, Any]      # summarized results worth keeping

@dataclass
class SemanticMemory:
    facts: list[str]                  # stable domain facts
    decisions: list[dict]             # durable decisions (never re-litigate)
    eliminated_approaches: list[dict] # things proven not to work
    patterns: list[str]               # recurring patterns observed

@dataclass
class FoldedMemory:
    semantic: SemanticMemory
    episodes: list[Episode]
    current_episode_id: int = 0
    total_turns_seen: int = 0

EPISODE_COMPACTOR_PROMPT = """
You are an episode compactor. Read the provided conversation turns and produce a structured Episode summary.

An Episode captures:
- What was attempted and what happened (not the full dialogue — the outcome)
- Decisions made and WHY (the reasoning behind them, not just the choice)
- Approaches tried and ruled out, with reasons (prevents re-exploration)
- Open questions that were not resolved
- Tool results that future turns will need (summarize, don't dump raw output)

Do NOT include:
- Intermediate back-and-forth that led to a conclusion (keep the conclusion, drop the path)
- Tool outputs that have already been acted on and have no future relevance
- Anything a fresh agent could derive by reading the code or running a command

Output valid JSON:
{
  "summary": "2-3 sentence narrative of what happened in this episode",
  "decisions": [{"decision": "...", "reason": "...", "constraint": "..."}],
  "eliminated": [{"approach": "...", "why": "..."}],
  "open_questions": ["..."],
  "tool_results": {"key": "summarized result"}
}
"""

def compact_l1_to_episode(turns: list[dict], episode_id: int, turn_range: tuple) -> Episode:
    response = client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=2048,
        system=EPISODE_COMPACTOR_PROMPT,
        messages=[{"role": "user", "content": format_turns(turns)}],
    )
    data = json.loads(response.content[0].text)
    return Episode(
        episode_id=episode_id,
        turn_range=turn_range,
        **data,
    )

SEMANTIC_EXTRACTOR_PROMPT = """
You are a semantic memory extractor. Read the provided episode summaries and extract only
knowledge that is durable — facts, decisions, and patterns that will still matter many
sessions from now.

Extract:
- Stable domain facts discovered ("the API always returns 200 even on failure — check data.success")
- Decisions that should never be re-litigated ("chose optimistic locking because DB doesn't support SELECT FOR UPDATE")
- Approaches definitively ruled out ("tried polling every 5s — causes rate limiting, don't retry")
- Recurring patterns that should inform future behavior

Do NOT extract:
- Task-specific state that will be resolved (current blockers, in-progress work)
- Results tied to a specific turn or tool call
- Anything that changes frequently

Merge with the existing semantic memory provided — update facts that were contradicted,
remove decisions that are now resolved, add new ones.

Output valid JSON:
{
  "facts": ["..."],
  "decisions": [{"decision": "...", "reason": "...", "constraint": "..."}],
  "eliminated_approaches": [{"approach": "...", "why": "..."}],
  "patterns": ["..."]
}
"""

def distill_episodes_to_semantic(
    episodes: list[Episode],
    existing_semantic: SemanticMemory,
) -> SemanticMemory:
    payload = {
        "existing_semantic": asdict(existing_semantic),
        "episodes_to_distill": [asdict(e) for e in episodes],
    }
    response = client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=2048,
        system=SEMANTIC_EXTRACTOR_PROMPT,
        messages=[{"role": "user", "content": json.dumps(payload, indent=2)}],
    )
    data = json.loads(response.content[0].text)
    return SemanticMemory(**data)

L1_MAX_TURNS = 20        # working memory window
L2_MAX_EPISODES = 8      # episodic memory window

def maybe_fold(memory: FoldedMemory, l1_turns: list[dict]) -> tuple[FoldedMemory, list[dict]]:
    """Check thresholds and fold layers if needed. Returns updated memory and remaining L1."""

    # L1 → L2: flush oldest turns when L1 is full
    if len(l1_turns) >= L1_MAX_TURNS:
        flush_count = L1_MAX_TURNS // 2             # flush half, keep recent half
        turns_to_flush = l1_turns[:flush_count]
        turn_range = (
            memory.total_turns_seen - len(l1_turns),
            memory.total_turns_seen - len(l1_turns) + flush_count - 1,
        )
        episode = compact_l1_to_episode(
            turns_to_flush,
            episode_id=memory.current_episode_id,
            turn_range=turn_range,
        )
        memory.episodes.append(episode)
        memory.current_episode_id += 1
        l1_turns = l1_turns[flush_count:]           # keep the recent half

    # L2 → L3: distill oldest episodes when L2 is full
    if len(memory.episodes) >= L2_MAX_EPISODES:
        flush_count = L2_MAX_EPISODES // 2
        episodes_to_distill = memory.episodes[:flush_count]
        memory.semantic = distill_episodes_to_semantic(episodes_to_distill, memory.semantic)
        memory.episodes = memory.episodes[flush_count:]

    return memory, l1_turns

def build_system_with_folded_memory(base_system: str, memory: FoldedMemory) -> str:
    blocks = [base_system]

    # L3 — always present
    if memory.semantic.facts or memory.semantic.decisions:
        blocks.append(_format_semantic(memory.semantic))

    # L2 — episode digest (most recent episodes first)
    if memory.episodes:
        blocks.append(_format_episodes(memory.episodes))

    return "\n\n".join(blocks)

def _format_semantic(s: SemanticMemory) -> str:
    lines = ["## Semantic memory (durable knowledge)"]
    if s.facts:
        lines += ["**Facts**:"] + [f"- {f}" for f in s.facts]
    if s.decisions:
        lines += ["**Decisions**:"] + [f"- {d['decision']} (because {d['reason']})" for d in s.decisions]
    if s.eliminated_approaches:
        lines += ["**Ruled out**:"] + [f"- {e['approach']}: {e['why']}" for e in s.eliminated_approaches]
    if s.patterns:
        lines += ["**Patterns**:"] + [f"- {p}" for p in s.patterns]
    return "\n".join(lines)

def _format_episodes(episodes: list[Episode]) -> str:
    lines = ["## Episode memory (recent history, oldest → newest)"]
    for ep in episodes:
        lines.append(f"\n### Episode {ep.episode_id} (turns {ep.turn_range[0]}–{ep.turn_range[1]})")
        lines.append(ep.summary)
        if ep.decisions:
            lines += ["Decisions:"] + [f"- {d['decision']}" for d in ep.decisions]
        if ep.open_questions:
            lines += ["Open:"] + [f"- {q}" for q in ep.open_questions]
    return "\n".join(lines)

def run_agent(session_id: str, user_input: str) -> str:
    memory = load_folded_memory(session_id)   # returns empty FoldedMemory if new session
    l1_turns = []

    while True:
        system = build_system_with_folded_memory(BASE_SYSTEM, memory)

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

        # Fold if needed
        l1_turns.append({"role": "user", "content": user_input})
        l1_turns.append({"role": "assistant", "content": response.content[0].text})
        memory, l1_turns = maybe_fold(memory, l1_turns)

        save_folded_memory(session_id, memory, l1_turns)

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

        user_input = handle_tool_calls(response)

import json, pathlib
from dataclasses import asdict

MEMORY_DIR = pathlib.Path("memory")

def save_folded_memory(session_id: str, memory: FoldedMemory, l1_turns: list[dict]) -> None:
    MEMORY_DIR.mkdir(exist_ok=True)
    (MEMORY_DIR / f"{session_id}_folded.json").write_text(
        json.dumps({"memory": asdict(memory), "l1_turns": l1_turns}, indent=2)
    )

def load_folded_memory(session_id: str) -> tuple[FoldedMemory, list[dict]]:
    path = MEMORY_DIR / f"{session_id}_folded.json"
    if not path.exists():
        return FoldedMemory(semantic=SemanticMemory([], [], [], [])), []
    data = json.loads(path.read_text())
    memory = FoldedMemory(
        semantic=SemanticMemory(**data["memory"]["semantic"]),
        episodes=[Episode(**e) for e in data["memory"]["episodes"]],
        current_episode_id=data["memory"]["current_episode_id"],
        total_turns_seen=data["memory"]["total_turns_seen"],
    )
    return memory, data["l1_turns"]