Consolidation

Every memory system that only writes eventually stops working — not with an error, but with a slow decline in recall as useful records get outvoted by accumulated sediment. Consolidation is the pump that runs against that: it merges near-duplicates, promotes episodes into facts, resolves what can be resolved, and prunes what has earned pruning.

It is also the only skill in this repo that destroys information. A merge bug is silent, permanent, and discovered weeks later by someone asking why the agent forgot something. So the design here is built around three properties that matter more than the merging itself: lineage (every derived record names its sources), dry run by default (nothing writes until a human has read the plan), and idempotence (a pass that dies halfway can simply be re-run).

  select candidates ──► blocking, not all-pairs
        │
        ▼
  propose operations ──► merge / promote / supersede / decay / prune
        │
        ▼
  ┌── DRY RUN ──► print the plan ──► human reads it ──┐
  │                                                   │
  └── APPLY ──► one transaction per operation ────────┘
                     │
                     ▼
              lineage written, sources kept until first successful use

Use this when

• The store is growing without bound and recall is measurably declining.
• You have an episodic log and want the distillate — this is the pump between memory-episodic and memory-semantic.
• Near-duplicates are crowding the top-k: five phrasings of one fact occupying all five slots.

Do not use this when the store is small. Under a few thousand records, a nightly pass is a cron job that mostly proves itself unnecessary, and you have imported the riskiest component in the system to solve a problem you do not have. Measure first: run the probe suite, look at distinct@k, and only build this when duplicates are demonstrably displacing real answers.

Workflow

1. Ask what is allowed to be destroyed. Which record types can be merged? Which must never be touched? Is anything under a retention obligation? If the user has no answer, default to: merge only near-identical facts, promote but never delete episodes, and prune nothing at all in the first version.
2. Ask what triggers a pass. Schedule, record count, or manual. Default to manual plus a documented cron line — a consolidation pass firing unattended on day one is how silent data loss happens.
3. Scaffold the project into the directory the user names. Kebab-case folder, snake_case package, hatchling + pyproject.toml, src/ layout.
4. Write the six modules: operations.py (the operation types), candidates.py (blocking), merge.py, promote.py, prune.py, runner.py (planning, dry run, transactional apply). Then cli.py.
5. Verify against the contract in references/evaluation.md, plus the consolidation-specific tests below.

The decisions that matter

1. Candidate selection — never compare all pairs

Naive deduplication is O(n²) embedding comparisons. At 50,000 records that is 1.25 billion, which is not a performance problem so much as a pass that never finishes. Use blocking: only compare records that share a cheap key.

def blocks(records):
    """Group by cheap keys. Only compare within a block."""
    by_key: dict[tuple, list] = defaultdict(list)
    for record in records:
        by_key[("subject", record.metadata.get("subject"))].append(record)
        by_key[("day", int(record.created_at // 86400))].append(record)
        by_key[("shingle", minhash_band(record.text))].append(record)
    return [group for group in by_key.values() if 2 <= len(group) <= 200]

Three blocking keys with different failure modes, which is the point of using several: subject catches facts about the same entity, day catches episodes from one session, and a MinHash band catches lexical near-duplicates that share neither. A pair missed by all three survives to the next pass — blocking trades recall for tractability, and that trade is correct here because a missed merge costs one redundant record while a bad merge costs the truth.

Cap block size. A block of 5,000 records sharing subject="user" is not a block, it is the whole store wearing a hat.

2. What may be merged automatically, and what may never

The rule underneath: automatic operations must be reversible or provably information-preserving. If undoing an operation requires information the operation destroyed, a human approves it or it does not happen.

3. Lineage, on every derived record

derived_from: list[str]      # the source record ids
derivation: str              # "merge" | "promote" | "compact"
derived_at: float

Lineage is what makes consolidation auditable, what makes forget(scope) cascade correctly (see references/forgetting.md), and what lets you answer "where did this come from" when a consolidated fact turns out to be wrong.

Keep the sources until the derived record has been retrieved and used at least once. Consolidation bugs surface on first use, and that is the only moment when rollback is still cheap. Mark sources archived rather than deleting them, and let a later pass delete archived sources whose derived record has a non-zero access_count.

4. Dry run is the default

<command> consolidate            # plans, prints, writes nothing
<command> consolidate --apply    # writes

Not a flag on a destructive default — a non-destructive default with an opt-in. The plan output must be readable by a human who did not write the system:

MERGE  3 records -> 1
  keep:  f_a19c  "Arttu deploys arttuan.com on Vercel"
  merge: f_77de  "Arttu uses Vercel for arttuan.com"        sim 0.96
         f_2b01  "arttuan.com is deployed on Vercel"        sim 0.95
  reason: same (subject, predicate, object), cosine > 0.95

PROMOTE episode cluster -> fact
  from:  ep_4a2, ep_91f, ep_c03  (3 episodes, 2 verified successes)
  fact:  "The CI pipeline has no redis service"
  confidence: 0.6  (promoted, not user-stated)

PRUNE  0 records  (retention policy: none configured)

SKIPPED 4 operations - run with --explain to see why

SKIPPED in the summary matters as much as the operations. A pass that skips nothing is not applying its safety rules.

5. Idempotence and crash safety

A consolidation pass will die halfway — a killed process, a full disk, a model timeout. Design for it:

• One transaction per operation, not one per pass. A pass that dies after 40 of 100 merges leaves 40 complete merges and 60 untouched records, which is a valid state.
• Operations are keyed by their inputs. A merge of {f_a, f_b, f_c} has a deterministic ID; re-running the pass finds the merge already recorded and skips it.
• Never mutate in place. Write the derived record, then mark the sources archived. In that order: a crash between them leaves a duplicate (harmless and fixed next pass), while the reverse order loses data.

operation_id = "op_" + blake2b(
    (kind + "|" + "|".join(sorted(source_ids))).encode(), digest_size=8
).hexdigest()

Build it

<project>/
├── pyproject.toml
├── .env.example
├── README.md
└── src/<package>/
    ├── __init__.py
    ├── config.py        # thresholds, retention, what may be auto-merged
    ├── operations.py    # Operation types + deterministic ids
    ├── candidates.py    # blocking: subject / day / minhash
    ├── merge.py         # near-duplicate merging
    ├── promote.py       # episodes -> facts, with corroboration
    ├── prune.py         # retention, decay, archived-source cleanup
    ├── runner.py        # plan -> dry run -> transactional apply
    └── cli.py           # consolidate / plan / explain / undo / stats

operations.py

from __future__ import annotations

import hashlib
import time
from dataclasses import dataclass, field
from enum import Enum


class Kind(str, Enum):
    MERGE = "merge"
    PROMOTE = "promote"
    COMPACT = "compact"
    SUPERSEDE = "supersede"
    ARCHIVE = "archive"
    PRUNE = "prune"


AUTO_APPROVED = {Kind.MERGE, Kind.PROMOTE, Kind.COMPACT, Kind.ARCHIVE}
NEEDS_HUMAN = {Kind.PRUNE}          # and entity merges, which live in memory-graph


@dataclass
class Operation:
    kind: Kind
    sources: list[str]
    produces: object | None = None       # the derived record, if any
    reason: str = ""
    score: float = 0.0
    skipped: str | None = None           # set when a safety rule blocked it
    created_at: float = field(default_factory=time.time)
    id: str = ""

    def __post_init__(self):
        if not self.id:
            payload = f"{self.kind.value}|" + "|".join(sorted(self.sources))
            self.id = "op_" + hashlib.blake2b(payload.encode(), digest_size=8).hexdigest()

    @property
    def auto(self) -> bool:
        return self.kind in AUTO_APPROVED and self.skipped is None

    def render(self) -> str:
        head = f"{self.kind.value.upper():9} {len(self.sources)} records"
        if self.skipped:
            return f"{head}  SKIPPED: {self.skipped}"
        lines = [f"{head}  ({self.reason})"]
        if self.produces is not None:
            lines.append(f"  -> {getattr(self.produces, 'text', self.produces)[:100]}")
        for source in self.sources[:5]:
            lines.append(f"     from {source}")
        return "\n".join(lines)

merge.py

"""Near-duplicate merging. Refuses anything that smells like a contradiction."""

from __future__ import annotations

from .operations import Kind, Operation

MERGE_AT = 0.95
CONTRADICTION_BAND = (0.85, 0.95)     # similar but not identical: suspicious


def plan_merges(block: list, *, similarity, keep_rule=None) -> list[Operation]:
    """One operation per cluster of near-identical records."""
    operations: list[Operation] = []
    used: set[str] = set()
    keep_rule = keep_rule or (lambda records: max(
        records, key=lambda r: (r.metadata.get("confidence", 0.5), r.created_at)))

    for i, record in enumerate(block):
        if record.id in used:
            continue
        cluster = [record]
        for other in block[i + 1:]:
            if other.id in used:
                continue
            score = similarity(record.text, other.text)
            if score >= MERGE_AT:
                if _contradicts(record, other):
                    operations.append(Operation(
                        Kind.MERGE, [record.id, other.id],
                        reason=f"sim {score:.2f}",
                        skipped="same key, different value - this is a contradiction, "
                                "not a duplicate; route it to the resolver"))
                    continue
                cluster.append(other)
            elif CONTRADICTION_BAND[0] <= score < CONTRADICTION_BAND[1] and _contradicts(record, other):
                operations.append(Operation(
                    Kind.MERGE, [record.id, other.id], reason=f"sim {score:.2f}",
                    skipped="near-duplicate with a differing value - needs a human"))

        if len(cluster) < 2:
            continue
        keeper = keep_rule(cluster)
        merged = _combine(keeper, cluster)
        used.update(r.id for r in cluster)
        operations.append(Operation(
            Kind.MERGE, [r.id for r in cluster], produces=merged,
            reason=f"{len(cluster)} records, cosine >= {MERGE_AT}",
            score=min(similarity(keeper.text, r.text) for r in cluster if r.id != keeper.id),
        ))
    return operations


def _contradicts(a, b) -> bool:
    key_a = (a.metadata.get("subject"), a.metadata.get("predicate"))
    key_b = (b.metadata.get("subject"), b.metadata.get("predicate"))
    if not all(key_a) or key_a != key_b:
        return False
    return a.metadata.get("object") != b.metadata.get("object")


def _combine(keeper, cluster):
    """The merged record keeps the best text and the union of everything else."""
    merged = replace(keeper)
    merged.metadata = dict(keeper.metadata)
    merged.metadata["derived_from"] = sorted({r.id for r in cluster})
    merged.metadata["derivation"] = "merge"
    merged.metadata["access_count"] = sum(r.metadata.get("access_count", 0) for r in cluster)
    # Corroboration raises confidence, but never past a user statement's own level.
    best = max(r.metadata.get("confidence", 0.5) for r in cluster)
    merged.metadata["confidence"] = min(0.99, best + 0.01 * (len(cluster) - 1))
    merged.created_at = min(r.created_at for r in cluster)      # keep the earliest claim date
    return merged

created_at = min(...) is deliberate: the merged fact has been true since the earliest time anyone said it, not since the merge ran. Taking the max quietly makes every consolidated fact look fresh, which then defeats recency decay.

promote.py

"""Episodes -> facts. The pump. Requires corroboration, caps confidence."""

from __future__ import annotations

from .operations import Kind, Operation

PROMOTE_PROMPT = """These episodes describe things that happened. Extract any
durable fact that MORE THAN ONE of them independently supports.

Rules:
- The fact must be supported by at least two separate episodes.
- Do not extract what happened (that is already recorded). Extract what is true.
- Do not extract anything an episode merely assumed.
- Empty array if nothing is corroborated. That is the common answer.

Reply as JSON: [{{"text": "...", "subject": "...", "predicate": "...",
                  "object": "...", "supported_by": [<episode indexes>]}}]

Episodes:
{episodes}"""

MIN_SUPPORT = 2
PROMOTED_CONFIDENCE = 0.6        # below any user-stated fact, deliberately


def plan_promotions(cluster: list, client, *, make_fact) -> list[Operation]:
    if len(cluster) < MIN_SUPPORT:
        return []
    rendered = "\n\n".join(f"[{i}] {e.render()}" for i, e in enumerate(cluster))
    reply = client.complete([{"role": "user", "content": PROMOTE_PROMPT.format(episodes=rendered)}])
    raw = (reply.content or "")
    raw = raw[raw.find("[") : raw.rfind("]") + 1] if "[" in raw else "[]"
    try:
        candidates = __import__("json").loads(raw)
    except Exception:
        return []

    operations = []
    for candidate in candidates if isinstance(candidates, list) else []:
        support = [i for i in candidate.get("supported_by", []) if 0 <= i < len(cluster)]
        if len(set(support)) < MIN_SUPPORT:
            operations.append(Operation(
                Kind.PROMOTE, [cluster[i].id for i in support], reason="candidate fact",
                skipped=f"only {len(set(support))} supporting episode(s)"))
            continue
        fact = make_fact(candidate, confidence=PROMOTED_CONFIDENCE,
                         derived_from=[cluster[i].id for i in support])
        operations.append(Operation(
            Kind.PROMOTE, [cluster[i].id for i in support], produces=fact,
            reason=f"corroborated by {len(set(support))} episodes"))
    return operations

Promotion never deletes the episodes. The log is the evidence; the fact is the claim. Deleting the evidence when you derive a claim is how a system becomes unable to explain itself.

runner.py

"""Plan, print, apply. One transaction per operation. Resumable."""

from __future__ import annotations

from dataclasses import dataclass, field

from .operations import Kind, Operation


@dataclass
class Plan:
    operations: list[Operation] = field(default_factory=list)

    @property
    def applicable(self) -> list[Operation]:
        return [o for o in self.operations if o.auto]

    @property
    def skipped(self) -> list[Operation]:
        return [o for o in self.operations if o.skipped]

    def render(self, *, explain: bool = False) -> str:
        lines = [o.render() for o in self.applicable]
        if self.skipped:
            lines.append(f"\nSKIPPED {len(self.skipped)} operations"
                         + (":" if explain else " - run with --explain to see why"))
            if explain:
                lines += [f"  {o.render()}" for o in self.skipped]
        by_kind = {}
        for operation in self.applicable:
            by_kind[operation.kind.value] = by_kind.get(operation.kind.value, 0) + 1
        lines.append("\n" + ", ".join(f"{v} {k}" for k, v in sorted(by_kind.items())) or "nothing to do")
        return "\n".join(lines)


class Runner:
    def __init__(self, store, *, max_compression: float = 10.0):
        self.store = store
        self.max_compression = max_compression

    def apply(self, plan: Plan, *, dry_run: bool = True) -> dict:
        if dry_run:
            return {"planned": len(plan.applicable), "applied": 0, "dry_run": True}

        applied = skipped = 0
        for operation in plan.applicable:
            if self.store.operation_done(operation.id):
                skipped += 1                       # idempotent: already applied
                continue
            ratio = len(operation.sources) / 1.0
            if operation.kind is Kind.COMPACT and ratio > self.max_compression:
                skipped += 1                       # not summarising, discarding
                continue
            with self.store.transaction():
                if operation.produces is not None:
                    self.store.put(operation.produces)      # derived record FIRST
                if operation.kind in (Kind.MERGE, Kind.COMPACT):
                    for source in operation.sources:
                        self.store.archive(source, superseded_by=getattr(operation.produces, "id", None))
                self.store.record_operation(operation)
            applied += 1
        return {"planned": len(plan.applicable), "applied": applied, "skipped": skipped}

The derived record is written before the sources are archived, inside one transaction, and the operation is recorded last. A crash at any point leaves either nothing done or everything done for that operation, and re-running skips what is already recorded.

cli.py

• <command> plan [--explain] — the dry run. The default. Prints the plan and writes nothing.
• <command> consolidate --apply — runs it. Prints the same plan plus a summary.
• <command> undo <operation_id> — restore archived sources, remove the derived record. Possible only while sources are archived rather than pruned, which is why the archive stage exists.
• <command> stats — store size over time, duplicate estimate, promotion count, oldest record. The growth curve is the number that tells you whether this is working.

Failure modes

• A merge that was a contradiction. Two facts with the same key and different values, cosine 0.96 because the wording matches, merged into whichever one the keep-rule picked. The disagreement is now gone and the agent asserts one side with raised confidence. The _contradicts check exists for exactly this; never merge on text similarity without comparing the structured key.
• Compaction that is deletion in disguise. Forty records into one paragraph is not a summary. Cap the compression ratio and alert rather than accepting it.
• Consolidating across scopes. Two users' records merged into one. A data leak wearing a tidy-up costume. Block on scope first, always, before any other blocking key.
• Lost lineage. A consolidated fact turns out to be wrong and there is no way to find what produced it or to re-derive it correctly. Lineage is three fields and it is the difference between a bug and an unexplainable system.
• A pass that deletes its own evidence. Promoting episodes into facts and then pruning the episodes leaves claims with no support, and a later forget(scope) that cannot cascade.
• Non-idempotent operations. A re-run after a crash double-merges, or merges a record into a record that was itself merged. Deterministic operation IDs plus a record_operation table; check before applying.
• Unattended first run. A consolidation pass on a cron schedule, on day one, against a store nobody has inspected. Run it manually, with --explain, for at least a week before it gets a schedule.
• Confidence inflation. Merging five copies of a rumour into one fact at confidence 0.99. Corroboration from independent sources is evidence; five copies of the same source is not. Cap the lift, and prefer to count distinct derived_from sources rather than record count.

Required tests

All offline. See references/evaluation.md for the universal set; these are mandatory:

def test_merging_preserves_every_source_id(store):
    plan = plan_merges(three_near_identical(), similarity=exact_similarity)
    merged = plan[0].produces
    assert set(merged.metadata["derived_from"]) == {"f_1", "f_2", "f_3"}


def test_contradiction_is_skipped_not_merged():
    a = fact("The deploy target is staging", subject="deploy", predicate="target", object="staging")
    b = fact("The deploy target is prod", subject="deploy", predicate="target", object="prod")
    operations = plan_merges([a, b], similarity=lambda x, y: 0.96)
    assert all(o.skipped for o in operations)
    assert "contradiction" in operations[0].skipped


def test_dry_run_writes_nothing(store, plan):
    before = snapshot(store)
    Runner(store).apply(plan, dry_run=True)
    assert snapshot(store) == before


def test_apply_is_idempotent(store, plan):
    first = Runner(store).apply(plan, dry_run=False)
    second = Runner(store).apply(plan, dry_run=False)
    assert first["applied"] > 0
    assert second["applied"] == 0 and second["skipped"] == first["applied"]


def test_crash_between_operations_leaves_a_valid_state(store, plan):
    runner = Runner(store)
    with crash_after(operations=2):
        with contextlib.suppress(Crash):
            runner.apply(plan, dry_run=False)
    assert store.integrity_ok()
    runner.apply(plan, dry_run=False)          # resumes cleanly
    assert store.operation_done(plan.applicable[-1].id)


def test_sources_are_archived_not_deleted(store, plan):
    Runner(store).apply(plan, dry_run=False)
    for source in plan.applicable[0].sources:
        assert store.get(source) is not None
        assert store.get(source).metadata["status"] == "archived"


def test_undo_restores_the_sources(store, plan):
    Runner(store).apply(plan, dry_run=False)
    store.undo(plan.applicable[0].id)
    for source in plan.applicable[0].sources:
        assert store.get(source).metadata["status"] == "active"


def test_promotion_requires_two_episodes(promoter):
    promoter.client = FakeChat(['[{"text":"x","supported_by":[0]}]'])
    operations = plan_promotions(two_episodes(), promoter.client, make_fact=make_fact)
    assert operations[0].skipped and "1 supporting" in operations[0].skipped


def test_promoted_facts_have_capped_confidence(promoter):
    operations = plan_promotions(three_episodes(), promoter.client, make_fact=make_fact)
    assert operations[0].produces.metadata["confidence"] <= 0.6


def test_promotion_does_not_delete_episodes(store, plan):
    before = set(store.ids(kind="episode"))
    Runner(store).apply(promotion_plan(), dry_run=False)
    assert set(store.ids(kind="episode")) == before


def test_blocking_never_crosses_scopes(store):
    groups = blocks(store.all())
    for group in groups:
        assert len({r.metadata["scope"] for r in group}) == 1


def test_over_compression_is_refused(store):
    plan = Plan([Operation(Kind.COMPACT, [f"ep_{i}" for i in range(40)], produces=one_summary())])
    result = Runner(store, max_compression=10.0).apply(plan, dry_run=False)
    assert result["applied"] == 0


def test_merged_record_keeps_the_earliest_created_at():
    merged = plan_merges(records_dated(["2020-01-01", "2024-01-01"]),
                         similarity=lambda a, b: 0.99)[0].produces
    assert merged.created_at == iso("2020-01-01")

Verify

Follow references/evaluation.md. For this skill specifically:

1. Tier 0 — install, import, --help, <command> plan on an empty store (should print "nothing to do", not crash).
2. Tier 1 — pytest -q, offline. Every test above, especially the crash and idempotence tests — those are the ones that matter in production.
3. Tier 2 — the before/after measurement, which is the only real proof this skill works. Run the probe suite, run consolidation, run it again:
   • recall@5 must not drop. If it does, consolidation destroyed something the probes needed. Stop and find out what.
   • distinct@5 should rise. That is the duplicates clearing out.
   • Store size should fall, and the growth curve over repeated simulated sessions should flatten rather than stay linear. Report all three numbers.
4. Tier 3 — run plan --explain against the user's real store and read every operation. Every single one, the first time. This is the skill where reading the plan is not optional, and it is the last cheap moment to catch a merge rule that is wrong.

Then tell the user what actually ran: operations planned, applied, skipped and why, plus the three Tier 2 numbers. If recall@5 fell at all, say so first and recommend against scheduling the pass until it is understood.