agent_memory_hygiene v1.0.0

name: agent_memory_hygiene description: "Audit, prune, decay-score, and de-duplicate an agent's long-term memory store so it stays accurate and small. Triggers when a memory store has grown stale, duplicated, or contradictory, or when setting up a recurring memory cleanup pass." metadata: { "includeInPrompt": true }

Agent Memory Hygiene

Purpose

Long-term memory stores grow by accretion. Old facts stay after they stop being true, the same fact gets written three times in three wordings, and a corrected fact sits next to the version it replaced. Recall then returns stale or conflicting entries, and the store gets slower to search as it grows. This skill is the maintenance pass: inventory what is in the store, flag duplicates and likely contradictions, score staleness, archive what is safe to remove, and prove recall still works afterwards.

Scope lines against close cousins:

• agentic-memory covers how to store and recall memories in the first place; this skill maintains a store that already exists.
• context-handoffs covers moving context between sessions; this skill curates one store in place.
• agent-observability covers traces and token accounting for agent runs; this skill works on memory content itself.

Workflow

1. Back up the store first

Copy the memory file to a dated backup before any step that writes. Pruning in step 6 writes new files and leaves the original untouched, but the backup is what makes swapping the pruned file into live use reversible.

2. Get the store into the helper's format

The helper at the end of this skill reads a plain JSON list or a JSONL file (one JSON object per line). Each record uses these fields:

• id (required): unique string.
• text (required): the memory content.
• created_at (required): ISO date or datetime, for example 2026-01-15.
• updated_at (optional): ISO date or datetime; defaults to created_at.
• last_recalled_at (optional): ISO date or datetime of the last successful recall.
• recall_count (optional): integer, defaults to 0.
• pinned (optional): boolean, defaults to false. Pinned records are never archived.
• key (optional): string. Records sharing a key are claims about the same fact.

If the live store uses a different format or lives in a database, export it into this shape first and keep the export script with the backup. The helper makes no claim about any specific memory framework or vector store API; it works on the exported file only.

3. Save the helper and run the inventory report

Save the script in the final section of this skill as audit.py, then run:

python3 audit.py report memories.jsonl --now 2026-10-12

--now pins the reference date so age and staleness numbers are reproducible. Omit it to use the current date. The report is JSON and includes:

• total_records, total_text_chars: store size.
• oldest_age_days, median_age_days, never_recalled, pinned: the age and use profile.
• exact_duplicate_groups: groups of ids whose normalized text is identical (lowercased, punctuation stripped, whitespace collapsed).
• near_duplicate_pairs: pairs with token-set Jaccard similarity at or above 0.8 that are not exact duplicates.
• contradiction_flags: pairs flagged as likely contradictions (see step 5 for how weak or strong that signal is).
• staleness: one entry per record with its score, age in days, recall count, and pinned state.

4. Score staleness and decay

The score is computed exactly as follows, so it can be explained and argued with:

• Days idle = days between --now and the record's last touch, where last touch is last_recalled_at if present, else updated_at, else created_at.
• Base score = min(days idle / 180, 1.0). A record untouched for 180 days or more has a base of 1.0.
• Recall factor: multiply by 1.0 if never recalled, 0.75 if recalled 1 to 4 times, 0.5 if recalled 5 or more times.

High staleness alone is not a deletion order. It means the record has not earned its place recently. Records that describe durable facts can be old, untouched, and still correct; that is what the review steps and the pinned flag are for.

5. Review duplicates and contradiction flags by hand

• Exact duplicates: safe to reduce to one copy. Step 6 does this automatically.
• Near duplicates: read both texts. Merge them into one record with the clearer wording, or keep both if the wording difference carries real meaning. The helper never archives near duplicates on its own.
• Contradiction flags: the helper flags a pair when token similarity is at least 0.5 and exactly one side contains a negation word (not, no, never, false), or when two records share a key but have different normalized text. Treat these as candidates, not proof. The check is wording-based: it misses contradictions stated in unrelated words, and it can flag pairs that merely discuss the same topic. Decide each flagged pair from source and date evidence outside the script, keep the surviving text, and archive the loser with a note in the review summary. Never let the script resolve a contradiction.

6. Write recall checks before pruning anything

Build a small checks.json: a JSON list of {"query": "...", "expect_id": "..."} entries for facts that must still be findable after pruning, plus any known critical memories. Save the checks with the backup so the same checks can be rerun after every future pass.

7. Prune into kept and archive files

python3 audit.py prune memories.jsonl --out kept.jsonl --archive archived.jsonl --now 2026-10-12

Defaults are --stale-threshold 0.8 and --min-age-days 90; pass different values if the store turns over faster or slower. Rules, in order:

1. Pinned records are always kept.
2. Exact duplicates: the first record in each group is kept, the rest are archived with reason exact duplicate.
3. Records with staleness at or above the threshold and age at or above the minimum age are archived with reason stale score <value>.
4. Everything else is kept. Near duplicates and contradiction flags do not cause archiving.

Archiving is the deletion mechanism. Archived records are copied to the archive file with an added _archive_reason field, so a bad prune can be undone by merging the archive back. Do not hard-delete the archive in the same pass; retire it only after the pruned store has run clean for a while.

8. Verify recall still works

python3 audit.py verify kept.jsonl checks.json

Verify ranks records by query token overlap and passes a check when the expected id is in the top 3 hits. Exit code is 0 when every check passes and 1 when any check fails. A failed check means the prune removed or buried something that matters: restore it from the archive, adjust the thresholds, and rerun from step 7. Only swap kept.jsonl into live use after verify passes. Also rerun report on the kept file and record the before and after record counts in the review summary.

9. Worked example

Fixture memories.jsonl (written and run during drafting of this skill):

{"id": "m1", "text": "alpha beta gamma delta epsilon zeta eta theta iota kappa", "created_at": "2026-10-01", "recall_count": 6, "last_recalled_at": "2026-10-10"} {"id": "m2", "text": "alpha beta gamma delta epsilon zeta eta theta iota kappa", "created_at": "2026-10-02", "recall_count": 0} {"id": "m3", "text": "alpha beta gamma delta epsilon zeta eta theta iota lambda", "created_at": "2026-09-20", "recall_count": 2, "last_recalled_at": "2026-10-05"} {"id": "m4", "text": "Old vendor pricing was forty dollars per seat", "created_at": "2024-01-01", "recall_count": 0} {"id": "m5", "text": "The office wifi is not open to guests", "created_at": "2026-08-01", "recall_count": 3, "last_recalled_at": "2026-10-01"} {"id": "m6", "text": "The office wifi is open to guests", "created_at": "2026-09-01", "recall_count": 1, "last_recalled_at": "2026-10-02"} {"id": "m7", "text": "Pinned safety note keep this record", "created_at": "2024-06-01", "recall_count": 0, "pinned": true}

Running report with --now 2026-10-12 gave: 7 records, one exact duplicate group (m1, m2), three near duplicate pairs (including m5/m6 at Jaccard 0.875), one contradiction flag (m5/m6, reason negation differs), and staleness 1.0 for both m4 (1015 days idle) and m7 (863 days idle). Running prune with default thresholds kept 5 records and archived 2: m2 as an exact duplicate and m4 as stale. Pinned m7 was kept despite its score of 1.0. Running verify on the kept file with checks expecting m5 for "office wifi guests" and m1 for "alpha beta gamma" passed 2 of 2 with exit code 0. A third check expecting the archived m4 failed and verify exited 1, which is the intended behavior: the check suite catches a prune that removed something the checks say must survive.

10. Helper script

Save the following as audit.py. It uses only the Python standard library.

python
#!/usr/bin/env python3
"""Audit, prune, and verify a plain JSON or JSONL agent memory file.

Record schema (one JSON object per record):
  id               str, required, unique
  text             str, required
  created_at       ISO date or datetime, required, e.g. 2026-01-15
  updated_at       ISO date or datetime, optional, defaults to created_at
  last_recalled_at ISO date or datetime, optional
  recall_count     int, optional, defaults to 0
  pinned           bool, optional, defaults to False
  key              str, optional; records sharing a key claim the same fact

Usage:
  python3 audit.py report memories.jsonl [--now 2026-10-12]
  python3 audit.py prune memories.jsonl --out kept.jsonl --archive archived.jsonl
                         [--stale-threshold 0.8] [--min-age-days 90] [--now 2026-10-12]
  python3 audit.py verify kept.jsonl checks.json
checks.json is a JSON list of {"query": "...", "expect_id": "..."} objects.
Exit code: report 0; prune 0; verify 0 if every check passes, else 1.
"""
import argparse, json, re, sys
from datetime import datetime, timezone
from pathlib import Path

NEGATIONS = {"not", "no", "never", "false"}

def parse_date(value):
    text = str(value).strip().replace("Z", "+00:00")
    dt = datetime.fromisoformat(text)
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    return dt

def tokens(text):
    return re.findall(r"[a-z0-9]+", str(text).lower())

def norm_text(text):
    return " ".join(tokens(text))

def token_set(text):
    return set(tokens(text))

def jaccard(a, b):
    if not a and not b:
        return 1.0
    if not a or not b:
        return 0.0
    return len(a & b) / len(a | b)

def load_records(path):
    raw = Path(path).read_text(encoding="utf-8").strip()
    if not raw:
        return []
    if raw.startswith("["):
        data = json.loads(raw)
        if not isinstance(data, list):
            raise ValueError("top-level JSON must be a list of records")
        return data
    return [json.loads(line) for line in raw.splitlines() if line.strip()]

def write_jsonl(path, records):
    with open(path, "w", encoding="utf-8") as fh:
        for rec in records:
            fh.write(json.dumps(rec, ensure_ascii=False) + "\n")

def last_touch(rec):
    for field in ("last_recalled_at", "updated_at", "created_at"):
        if rec.get(field):
            return parse_date(rec[field])
    raise ValueError(f"record {rec.get('id')!r} has no date field")

def staleness(rec, now):
    age_days = max((now - last_touch(rec)).days, 0)
    score = min(age_days / 180.0, 1.0)
    count = int(rec.get("recall_count", 0))
    if count >= 5:
        score *= 0.5
    elif count >= 1:
        score *= 0.75
    return round(score, 3), age_days

def analyse(records, now):
    sets = [token_set(r.get("text", "")) for r in records]
    exact = {}
    for rec in records:
        exact.setdefault(norm_text(rec.get("text", "")), []).append(rec["id"])
    exact_groups = [ids for ids in exact.values() if len(ids) > 1]
    near_pairs, contradictions = [], []
    for i in range(len(records)):
        for j in range(i + 1, len(records)):
            sim = jaccard(sets[i], sets[j])
            a, b = records[i], records[j]
            if sim >= 0.8 and norm_text(a.get("text", "")) != norm_text(b.get("text", "")):
                near_pairs.append({"ids": [a["id"], b["id"]], "jaccard": round(sim, 3)})
            neg_diff = bool(sets[i] & NEGATIONS) != bool(sets[j] & NEGATIONS)
            same_key = a.get("key") and a.get("key") == b.get("key")
            if (sim >= 0.5 and neg_diff) or (same_key and norm_text(a.get("text", "")) != norm_text(b.get("text", ""))):
                contradictions.append({"ids": [a["id"], b["id"]], "jaccard": round(sim, 3),
                                         "reason": "negation differs" if neg_diff else "same key, different text"})
    scores = []
    for rec in records:
        score, age = staleness(rec, now)
        scores.append({"id": rec["id"], "staleness": score, "age_days": age,
                       "recall_count": int(rec.get("recall_count", 0)),
                       "pinned": bool(rec.get("pinned", False))})
    return exact_groups, near_pairs, contradictions, scores

def cmd_report(args):
    records = load_records(args.file)
    now = parse_date(args.now) if args.now else datetime.now(timezone.utc)
    exact, near, contra, scores = analyse(records, now)
    ages = sorted(s["age_days"] for s in scores)
    report = {
        "total_records": len(records),
        "total_text_chars": sum(len(str(r.get("text", ""))) for r in records),
        "oldest_age_days": ages[-1] if ages else 0,
        "median_age_days": ages[len(ages) // 2] if ages else 0,
        "never_recalled": sum(1 for r in records if int(r.get("recall_count", 0)) == 0),
        "pinned": sum(1 for r in records if r.get("pinned", False)),
        "exact_duplicate_groups": exact,
        "near_duplicate_pairs": near,
        "contradiction_flags": contra,
        "staleness": scores,
    }
    print(json.dumps(report, indent=2))
    return 0

def cmd_prune(args):
    records = load_records(args.file)
    now = parse_date(args.now) if args.now else datetime.now(timezone.utc)
    exact, _near, _contra, scores = analyse(records, now)
    score_by_id = {s["id"]: s for s in scores}
    dup_ids = set()
    for group in exact:
        dup_ids.update(group[1:])
    keep, archive = [], []
    for rec in records:
        s = score_by_id[rec["id"]]
        if rec.get("pinned", False):
            keep.append(rec)
        elif rec["id"] in dup_ids:
            archive.append({**rec, "_archive_reason": "exact duplicate"})
        elif s["staleness"] >= args.stale_threshold and s["age_days"] >= args.min_age_days:
            archive.append({**rec, "_archive_reason": f"stale score {s['staleness']}"})
        else:
            keep.append(rec)
    write_jsonl(args.out, keep)
    write_jsonl(args.archive, archive)
    print(json.dumps({"kept": len(keep), "archived": len(archive),
                      "kept_ids": [r["id"] for r in keep],
                      "archived_ids": [r["id"] for r in archive]}, indent=2))
    return 0

def recall(records, query, top=3):
    q = token_set(query)
    scored = [(len(q & token_set(r.get("text", ""))), r["id"]) for r in records]
    scored.sort(key=lambda pair: (-pair[0], pair[1]))
    return [rid for score, rid in scored if score > 0][:top]

def cmd_verify(args):
    records = load_records(args.file)
    checks = json.loads(Path(args.checks).read_text(encoding="utf-8"))
    failed = 0
    for check in checks:
        hits = recall(records, check["query"])
        ok = check["expect_id"] in hits
        if not ok:
            failed += 1
        print(json.dumps({"query": check["query"], "expect_id": check["expect_id"],
                          "top_hits": hits, "pass": ok}))
    print(f"{len(checks) - failed}/{len(checks)} checks passed")
    return 1 if failed else 0

def main():
    parser = argparse.ArgumentParser(description="Agent memory hygiene audit")
    sub = parser.add_subparsers(dest="cmd", required=True)
    p = sub.add_parser("report"); p.add_argument("file"); p.add_argument("--now"); p.set_defaults(fn=cmd_report)
    p = sub.add_parser("prune"); p.add_argument("file")
    p.add_argument("--out", required=True); p.add_argument("--archive", required=True)
    p.add_argument("--stale-threshold", type=float, default=0.8)
    p.add_argument("--min-age-days", type=int, default=90)
    p.add_argument("--now"); p.set_defaults(fn=cmd_prune)
    p = sub.add_parser("verify"); p.add_argument("file"); p.add_argument("checks"); p.set_defaults(fn=cmd_verify)
    args = parser.parse_args()
    sys.exit(args.fn(args))

if __name__ == "__main__":
    main()

Output Contract

A completed hygiene pass produces all of the following:

• The report JSON from the pre-prune run, saved next to the backup.
• kept.jsonl: the pruned store, not yet swapped into live use until verify passes.
• archived.jsonl: every removed record with its _archive_reason field.
• The verify output for kept.jsonl against checks.json, showing every check passing.
• A short review summary in plain text: record counts before and after, the thresholds used, how each near duplicate pair and contradiction flag was resolved and by what evidence, and the date of the pass.

Operating Rules

1. Back up before writing anything. Never run a prune against the only copy of a store.
2. Archive, never hard-delete, inside a pass. Removal means the record moves to the archive file with a reason.
3. Pinned records are exempt from archiving. If a pinned record looks wrong, that is a review decision, not a pruning outcome.
4. Exact duplicates may be archived automatically. Near duplicates and contradiction flags always need a human-readable decision recorded in the review summary.
5. Contradiction flags are wording heuristics, not findings of fact. State their limits when reporting them: unrelated wording is missed, same-topic pairs can be flagged in error.
6. Recall checks are written before pruning, never after. Checks written after the fact only prove the pruned store recalls what survived.
7. Do not swap the pruned file into live use until verify exits 0. On failure, restore from the archive and adjust thresholds instead of editing the kept file by hand.
8. Record the thresholds and the --now date with each pass, so the next pass can reproduce the numbers.
9. If the store cannot be exported to the JSON or JSONL shape in step 2, stop and say so. Do not guess at a framework-specific API to reach the records.
10. Keep the pass boring and repeatable: same steps, same checks file, growing only when new critical memories appear.