purge: erase a claim's text from every index, by handle

README.md

@@ -169,7 +169,7 @@ advisory flag set at recall when a code-anchored claim's anchors all fail to res
 | var | meaning |
 |---|---|
 | `KYPP_MEMORY_DB` | store path (default `~/.kypp/memory.db`) |
-| `KYPP_URL` | a hosted kypp Worker: `recall`, `show`, `briefing`, `remember`, `correct`, `reject` and `verify` use it through `POST /v1/operator` instead of the local store (unset → local). Needs `KYPP_OPERATOR_TOKEN`; see `worker/README.md` |
+| `KYPP_URL` | a hosted kypp Worker: `recall`, `show`, `briefing`, `remember`, `correct`, `reject`, `purge` and `verify` use it through `POST /v1/operator` instead of the local store (unset → local). Needs `KYPP_OPERATOR_TOKEN`; see `worker/README.md` |
 | `KYPP_PROJECT` | project binding; unset → **derived from `KYPP_REPO_ROOT`** (its path as a `/`→`-` key, matching what `sweep` files under) so serve + CLI + capture share one bucket per repo; `sweep`/`capture` instead derive each session's project from its pillbox log path (`global/` vs `projects/<key>/`) |
 | `KYPP_REPO_ROOT` | repo the ripgrep code-resolver grounds against, and the source of the derived `KYPP_PROJECT` key (default `.`) |
 | `KYPP_USER` | authoring human, stamped on every claim as provenance (default: OS login) |

docs/features/hosted-worker.md

@@ -31,7 +31,8 @@ TypeSafe API's shape, for GitHub Actions OIDC tokens (`worker/src/oidc.ts`) and
   nothing, never browse results. Recall order ends in `c.id`, so ties don't depend on the engine.
   `worker/bench/replay` replays one query set against every store to check this.
 - `POST /v1/operator` (`worker/src/operator.ts`, `kypp/remote.py`) is the only hosted writer of human
-  authority, `verify` commands and verified results, behind `KYPP_OPERATOR_TOKEN`. Keep
+  authority, `verify` commands and verified results, behind `KYPP_OPERATOR_TOKEN`. Its `purge` op
+  reaches any claim by handle (`anyClaim`, every scope and lane) and logs only the reason. Keep
   `D1Store.operatorClaim`, `markVerified` and `operatorReject` off every MCP path, and log each in
   `defrag_log`. `markVerified` mirrors `MemoryStore.mark_verified`: change one, change both.
 - Hosted operator `correct` is an explicit replacement in the existing `memory_claims` store,

docs/features/recall.md

@@ -13,7 +13,7 @@ tests:
 ---
 # Recall, briefing and handles
 
-`kypp recall`, `briefing`, `show`, `remember`, `correct`, `reject` and `usage` print one compact
+`kypp recall`, `briefing`, `show`, `remember`, `correct`, `reject`, `purge` and `usage` print one compact
 handle line per claim, and `show` expands a handle to the full claim. With `KYPP_URL` set, all but
 `usage` go to the hosted Worker (`kypp/remote.py`) and never open the local store.
 

docs/features/store.md

@@ -55,4 +55,11 @@ the file path, so search can go from a file to entries and from an entry to file
   file bodies; they go stale and burn context.
 - Vector recall is an exact scan on purpose (pyturso and libsql). libsql can build a DiskANN
   index; that trades recall for speed and is the wrong quality choice at this corpus size.
-- Memory is append-only: claims and observations are never deleted.
+- Memory is append-only: claims and observations are never deleted. The one erase is `purge`
+  (`MemoryStore.purge`, operator `purge` on the Worker): it blanks the claim's text, embedding,
+  anchors and verify, drops its FTS and pointer rows and its usage queries, and leaves a rejected
+  `[purged]` tombstone. A recall query is copied onto every claim it returned, so purge clears
+  every copy; hosted purge also blanks the claim's earlier `defrag_log` reasons. Verdicts skip a
+  claim with no `verify`, so a run that started before the purge can't revive it. A new table or
+  column that copies claim text must be added to both purges.
+  Purge does not touch the observations a claim was distilled from.

kypp/backends.py

@@ -37,6 +37,9 @@ def _claim_side_writes(self, cid, subject, content, code_refs, now):
         ))
         return rows
 
+    def _purge_side_writes(self, claim_id):
+        return [*super()._purge_side_writes(claim_id), ("DELETE FROM claims_fts WHERE claim_id=?", (claim_id,))]
+
     def _fts_ids(self, query: str, where: list[str], params: list, limit: int) -> list[str]:
         terms = query_tokens(query)
         if not terms:

kypp/cli.py

@@ -11,6 +11,7 @@
     "remember": "store a claim: SUBJECT CONTENT (subject = identity key)",
     "correct": "human correction: assert the right answer for a subject, supersede the rest",
     "reject": "demote a claim by handle out of recall/briefing (outcome-driven; not deletion)",
+    "purge": "erase a claim by handle from every index (leaked key, badly wrong lesson)",
     "verify": "re-check verifier-carrying config/fact claims → mark verified/stale",
     "briefing": "session-start digest: strongest accepted memory, pitfalls first",
     "usage": "usage provenance: which claims a run saw / which runs saw a claim",
@@ -45,6 +46,8 @@ def main():
         from .shell import correct_main as run
     elif cmd == "reject":
         from .shell import reject_main as run
+    elif cmd == "purge":
+        from .shell import purge_main as run
     elif cmd == "verify":
         from .verify import main as run
     elif cmd == "briefing":

kypp/shell.py

@@ -216,6 +216,24 @@ def reject_main():
     print(f"{cid[:8]} [{status} {ctype}] {subject}" + (f" — {args.reason}" if args.reason else ""))
 
 
+def purge_main():
+    ap = argparse.ArgumentParser(
+        description="erase a claim BY HANDLE: its subject, content, code anchors and the recall queries "
+                    "that found it, from every index. The row stays as a rejected '[purged]' tombstone. "
+                    "For a leaked key or a badly wrong lesson; to retire an ordinary bad claim, use `reject`.")
+    ap.add_argument("handle", nargs="+", help="claim id or 8-char prefix; several to purge copies")
+    ap.add_argument("--reason", required=True, help="why, for the audit log. Never paste the purged text here")
+    args = ap.parse_args()
+    for handle in args.handle:
+        if remote.remote_url():
+            cid = remote.call("purge", handle=handle, reason=args.reason)["claim"]["id"]
+        else:
+            store = store_from_env()
+            cid = _resolve_handle(store, handle).id
+            store.purge(cid, args.reason)
+        print(f"{cid[:8]} purged — {args.reason}")
+
+
 def project_main():
     ap = argparse.ArgumentParser(
         description="the hosted store's projects (needs KYPP_URL): list them, or add one so claims under "

kypp/store.py

@@ -42,6 +42,8 @@
 )
 from .vocab import AUTHORITIES, OLLAMA_DEFAULT_HOST, SCOPES, STATUSES, SURFACES, TYPES  # single-sourced vocabulary (turso-free leaf)
 
+PURGED = "[purged]"  # what a purged claim's subject and content read as
+
 SCHEMA = """
 CREATE TABLE IF NOT EXISTS observations (
   id TEXT PRIMARY KEY, actor TEXT NOT NULL, project TEXT, scope TEXT NOT NULL,
@@ -401,18 +403,19 @@ def mark_verified(self, claim_id: str, passed: bool) -> None:
         it, and since recency breaks ties between equal human claims, a sweep landing between
         `kypp correct`'s insert and its consolidate must not make the old answer look newest. The
         verifier's only write."""
-        # Verify raises agent → verified, never lowers human.
+        # Verify raises agent → verified, never lowers human. A claim whose verify was cleared (purged)
+        # takes no verdict, so a run that loaded it before the purge can't revive it.
         if passed:
             self._write([("UPDATE memory_claims SET "
                           "status=CASE WHEN authority='human' AND status='superseded' THEN status ELSE 'accepted' END, "
                           "authority=CASE WHEN authority='human' THEN 'human' ELSE 'verified' END, "
                           "updated_at=CASE WHEN authority='human' THEN updated_at ELSE ? END "
-                          "WHERE id=?", (_now(), claim_id))])
+                          "WHERE id=? AND verify IS NOT NULL", (_now(), claim_id))])
         else:
             self._write([("UPDATE memory_claims SET "
                           "status=CASE WHEN authority='human' AND status='superseded' THEN status ELSE 'rejected' END, "
                           "updated_at=CASE WHEN authority='human' THEN updated_at ELSE ? END "
-                          "WHERE id=?", (_now(), claim_id))])
+                          "WHERE id=? AND verify IS NOT NULL", (_now(), claim_id))])
 
     def reject(self, claim_id: str, reason: str | None = None) -> None:
         """Outcome-driven demote: set a claim to `rejected` so recall/briefing drop it, optionally
@@ -435,6 +438,27 @@ def reject(self, claim_id: str, reason: str | None = None) -> None:
                       "metadata=?, updated_at=? WHERE id=?",
                       (json.dumps(meta), _now(), claim_id))])
 
+    def purge(self, claim_id: str, reason: str) -> None:
+        """Erase one claim's text everywhere recall reads it (row, embedding, pointers, usage queries,
+        FTS on the adapters that have it), keeping a rejected tombstone so the id still resolves. The
+        one erase in an append-only store: for a leaked key or a badly wrong lesson. The observations
+        a claim was distilled from are not touched."""
+        _require(bool(reason and reason.strip()), "purge needs a reason")
+        now = _now()
+        self._write([
+            ("UPDATE memory_claims SET subject=?, content=?, status='rejected', verify=NULL, embedding=NULL, "
+             "code_refs='[]', source_ids='[]', metadata=?, updated_at=? WHERE id=?",
+             (PURGED, PURGED, json.dumps({"purged_at": now, "purged_reason": reason}), now, claim_id)),
+            ("DELETE FROM code_pointers WHERE claim_id=?", (claim_id,)),
+            # A recall query is copied onto every claim it returned; clear each copy, not just this claim's.
+            ("UPDATE claim_usages SET query=NULL WHERE query IN "
+             "(SELECT query FROM claim_usages WHERE claim_id=? AND query IS NOT NULL)", (claim_id,)),
+            *self._purge_side_writes(claim_id),
+        ])
+
+    def _purge_side_writes(self, claim_id: str) -> list[tuple[str, tuple]]:
+        return []
+
     # --- usage = the run -> claims-consumed link (memory credit-assignment foundation) ---
     def record_usage(self, consumer: str, claims, *, surface: str, project: str | None = None,
                      scope: str | None = None, query: str | None = None) -> int:

tests/test_project_cli.py

@@ -1,4 +1,4 @@
-"""`kypp project` is hosted-only and prints what the operator route answered."""
+"""`kypp project` and `kypp purge` over KYPP_URL print what the operator route answered."""
 import contextlib
 import io
 import os
@@ -9,15 +9,15 @@
 from kypp import remote, shell
 
 
-def _run(argv, answers):
+def _run(argv, answers, main=shell.project_main, prog="kypp project"):
     calls = []
     def fake(op, **args):
         calls.append((op, args))
         return answers[op]
     out = io.StringIO()
-    with mock.patch.object(sys, "argv", ["kypp project", *argv]), mock.patch.object(remote, "call", fake), \
+    with mock.patch.object(sys, "argv", [prog, *argv]), mock.patch.object(remote, "call", fake), \
             mock.patch.dict(os.environ, {"KYPP_URL": "https://k.example"}), contextlib.redirect_stdout(out):
-        shell.project_main()
+        main()
     return calls, out.getvalue()
 
 
@@ -39,5 +39,19 @@ def test_needs_a_hosted_store(self):
         self.assertIn("KYPP_URL", str(e.exception))
 
 
+class PurgeCliTest(unittest.TestCase):
+    def test_purges_each_handle_with_the_reason(self):
+        calls, out = _run(["abcd1234", "ef567890", "--reason", "leaked key"],
+                          {"purge": {"claim": {"id": "abcd1234" + "0" * 24}}}, shell.purge_main, "kypp purge")
+        self.assertEqual(calls, [("purge", {"handle": "abcd1234", "reason": "leaked key"}),
+                                 ("purge", {"handle": "ef567890", "reason": "leaked key"})])
+        self.assertEqual(out.splitlines(), ["abcd1234 purged — leaked key"] * 2)
+
+    def test_needs_a_reason(self):
+        with mock.patch.object(sys, "argv", ["kypp purge", "abcd1234"]), contextlib.redirect_stderr(io.StringIO()):
+            with self.assertRaises(SystemExit):
+                shell.purge_main()
+
+
 if __name__ == "__main__":
     unittest.main()

tests/test_store_contract.py

@@ -78,6 +78,34 @@ def test_put_get_and_keyword_search(self):
         hits = store.search("codesign", project="pillbox")
         self.assertEqual([c.id for c in hits], [cid])
 
+    def test_purge_erases_text_from_every_index_and_keeps_a_tombstone(self):
+        store = self.open(self.path, embed=_embed if self.semantic else None)
+        key = "sk-live-leakedvalue0123"
+        cid = store.put("fact", "deploy token", f"use {key} for ftsmarker deploys", scope="project",
+                        project="p", accept=True, code_refs=[{"symbol": "TOKEN", "path": "deploy/token.py"}])
+        other = store.put("fact", "deploys", "deploys go out on green", scope="project", project="p", accept=True)
+        # One recall returned both claims, so its query (and the key) is copied onto both usage rows.
+        store.record_usage("run", [store.get(cid), store.get(other)], surface="recall", project="p", query=f"which {key}")
+        self.assertIn(cid, [c.id for c in store.search("leakedvalue0123", project="p")])
+        store.purge(cid, "leaked key")
+        store.mark_verified(cid, True)  # a verify run that loaded the claim before the purge
+        self.assertEqual(store.get(cid).status, "rejected", "a late verdict can't revive a tombstone")
+        self.assertEqual(store.get(other).content, "deploys go out on green")
+        got = store.get(cid)
+        self.assertEqual((got.status, got.subject, got.content), ("rejected", "[purged]", "[purged]"))
+        self.assertEqual(got.code_refs, [])
+        self.assertNotIn(cid, [c.id for c in store.search("leakedvalue0123", project="p", include_candidates=True)])
+        self.assertNotIn(cid, [c.id for c in store.search("ftsmarker", project="p", include_candidates=True)])
+        self.assertEqual(store.entries_for_path("deploy/token.py"), [])
+        cur = store.db.cursor()
+        tables = ["memory_claims", "claim_usages", "code_pointers"] + (["claims_fts"] if self.fts else [])
+        for table in tables:
+            rows = cur.execute(f"SELECT * FROM {table}").fetchall()
+            self.assertNotIn(key, repr(rows), table)
+        self.assertIsNone(cur.execute("SELECT verify FROM memory_claims WHERE id=?", (cid,)).fetchone()[0])
+        with self.assertRaises(ValueError):
+            store.purge(cid, " ")
+
     def test_stale_claims_rank_after_live_ones_unless_named(self):
         store = self.open(self.path)
         store.resolver = _GoneResolver()

worker/src/operator.ts

@@ -75,6 +75,14 @@ async function run(store: D1Store, a: Args, opts: OperatorOptions): Promise<unkn
       await store.operatorReject(c.id, str(a.reason) || null);
       return { claim: await store.get(c.id, OPERATOR_USER) };
     }
+    case "purge": {
+      const reason = str(a.reason).trim();
+      if (!reason) throw new Error("purge needs a reason (it is logged; don't paste the purged text into it)");
+      const c = await store.anyClaim(str(a.handle));
+      if (!c) throw new Error(`no claim ${JSON.stringify(a.handle)}`);
+      await store.operatorPurge(c.id, reason);
+      return { claim: await store.anyClaim(c.id) };
+    }
     case "verifiable": {
       const where = await project();
       if (!where) throw new Error(`unknown project ${JSON.stringify(a.project)}`);

worker/src/store.ts

@@ -177,6 +177,8 @@ export interface ClaimInput {
   sharedFrom?: string;
 }
 
+export const PURGED = "[purged]";
+
 export class D1Store {
   private db: D1Database;
   private screened: boolean;
@@ -301,6 +303,26 @@ export class D1Store {
     ]);
   }
 
+  // Purge is the one erase: memory is otherwise never deleted, but a leaked key or a badly wrong
+  // lesson must leave every copy kypp reads (row text, FTS, pointers, recall queries). The row stays
+  // as a rejected tombstone so its id still resolves, and the log keeps the reason, never the text.
+  async operatorPurge(id: string, reason: string): Promise<void> {
+    const ts = now();
+    await this.db.batch([
+      this.db.prepare(
+        "UPDATE memory_claims SET subject = ?, content = ?, status = 'rejected', verify = NULL, code_refs = '[]',"
+        + " source_ids = '[]', metadata = json_object('purged_at', ?), updated_at = ? WHERE id = ?",
+      ).bind(PURGED, PURGED, ts, ts, id),
+      this.db.prepare("DELETE FROM claims_fts WHERE claim_id = ?").bind(id),
+      this.db.prepare("DELETE FROM code_pointers WHERE claim_id = ?").bind(id),
+      // A recall query is copied onto every claim it returned; clear each copy, not just this claim's.
+      this.db.prepare("UPDATE claim_usages SET query = NULL WHERE query IN (SELECT query FROM claim_usages WHERE claim_id = ? AND query IS NOT NULL)").bind(id),
+      // Earlier free-text reasons (a reject's) may quote the leaked text; the actions and times stay.
+      this.db.prepare("UPDATE defrag_log SET reason = NULL WHERE claim_id = ?").bind(id),
+      this.logOperator(id, "purge", null, reason),
+    ]);
+  }
+
   // One defrag_log row for an operator write, so every human/verified change can be traced.
   // `when` (a query on the claim id) skips the row when the write it logs would not apply.
   private logOperator(id: string, op: string, project: string | null, reason: string | null, when?: string): D1PreparedStatement {
@@ -408,6 +430,15 @@ export class D1Store {
     return results.length ? hydrate(results[0]) : null;
   }
 
+  // The operator's reach: any claim by handle, whatever its scope or lane (purge must reach a key
+  // leaked into a room, another user's scope or an unsorted claim).
+  async anyClaim(handle: string): Promise<Claim | null> {
+    if (!/^[0-9a-f]{8,32}$/.test(handle)) throw new Error(`bad claim handle ${JSON.stringify(handle)} (expected 8-32 hex chars)`);
+    const { results } = await this.db.prepare("SELECT * FROM memory_claims WHERE id LIKE ? LIMIT 2").bind(`${handle}%`).all();
+    if (results.length > 1) throw new Error(`ambiguous claim handle ${handle} (use more chars)`);
+    return results.length ? hydrate(results[0]) : null;
+  }
+
   async recordUsage(consumer: string | null, claims: Claim[], surface: "recall" | "briefing" | "expand",
     project: string | null, query: string | null = null): Promise<void> {
     if (!consumer || !claims.length) return;