diff --git a/README.md b/README.md index d8bb3b2..b970698 100644 --- a/README.md +++ b/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//`) | | `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) | diff --git a/docs/features/hosted-worker.md b/docs/features/hosted-worker.md index ebe3fc8..4c72250 100644 --- a/docs/features/hosted-worker.md +++ b/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, diff --git a/docs/features/recall.md b/docs/features/recall.md index 7d59d55..41e122e 100644 --- a/docs/features/recall.md +++ b/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. diff --git a/docs/features/store.md b/docs/features/store.md index 2ced3c4..36ad8b7 100644 --- a/docs/features/store.md +++ b/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. diff --git a/kypp/backends.py b/kypp/backends.py index a019993..6eda034 100644 --- a/kypp/backends.py +++ b/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: diff --git a/kypp/cli.py b/kypp/cli.py index 6cd4275..4497cb9 100644 --- a/kypp/cli.py +++ b/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": diff --git a/kypp/shell.py b/kypp/shell.py index 3443bf2..333f9da 100644 --- a/kypp/shell.py +++ b/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 " diff --git a/kypp/store.py b/kypp/store.py index 2999422..d3f2b4d 100644 --- a/kypp/store.py +++ b/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: diff --git a/tests/test_project_cli.py b/tests/test_project_cli.py index d4695d7..45043cd 100644 --- a/tests/test_project_cli.py +++ b/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() diff --git a/tests/test_store_contract.py b/tests/test_store_contract.py index 7b95384..514b051 100644 --- a/tests/test_store_contract.py +++ b/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() diff --git a/worker/src/operator.ts b/worker/src/operator.ts index 2e7cefa..f09f8ea 100644 --- a/worker/src/operator.ts +++ b/worker/src/operator.ts @@ -75,6 +75,14 @@ async function run(store: D1Store, a: Args, opts: OperatorOptions): Promise { + 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 { + 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 { if (!consumer || !claims.length) return; diff --git a/worker/test/operator.test.ts b/worker/test/operator.test.ts index 835ccfe..bdeda38 100644 --- a/worker/test/operator.test.ts +++ b/worker/test/operator.test.ts @@ -232,3 +232,38 @@ test("register adds a project once and files the claims already waiting under it assert.equal((await call({ op: "register", project: "vu1n/grokbot" })).status, 400); assert.equal((await call({ op: "register", project: "" })).status, 400); }); + +test("purge erases a claim's text from every place recall reads, keeps a tombstone, and is logged", async () => { + const { db, call, row, logged } = await setup(); + const store = new D1Store(db); + const KEY = "sk-live-leakedvalue0123"; + // A key leaked into another user's scope: the operator reaches it anyway. + const id = await store.claim({ type: "fact", subject: "deploy token", content: `use ${KEY} for deploys`, scope: "user", project: null, + confidence: 0.7, sourceIds: ["session:a"], codeRefs: [{ path: "deploy/token.ts", symbol: "TOKEN" }], accept: true, agent: "t", user: "someone" }); + await db.prepare("UPDATE memory_claims SET verify = 'true' WHERE id = ?").bind(id).run(); + const before = await db.prepare("SELECT (SELECT COUNT(*) FROM claims_fts WHERE claim_id = ?) AS fts, (SELECT COUNT(*) FROM code_pointers WHERE claim_id = ?) AS ptr") + .bind(id, id).first() as Record; + assert.ok(before.fts > 0 && before.ptr > 0, "the claim is indexed before the purge"); + const other = await store.claim({ type: "fact", subject: "deploys", content: "deploys go out on green", scope: "user", project: null, + confidence: 0.7, sourceIds: [], codeRefs: [], accept: true, agent: "t", user: "someone" }); + // One recall returned both claims, so its query (and the key) is copied onto both usage rows. + await store.recordUsage("s", [(await store.anyClaim(id))!, (await store.anyClaim(other))!], "recall", null, `which ${KEY}`); + // An earlier reject whose reason quotes the leaked text. + await store.operatorReject(id, `wrong: ${KEY}`); + assert.equal((await call({ op: "purge", handle: id.slice(0, 8) })).status, 400, "a reason is required"); + const res = await call({ op: "purge", handle: id.slice(0, 8), reason: "leaked key" }); + assert.equal(res.status, 200, JSON.stringify(res.body)); + const r = await row(id); + assert.equal(r.status, "rejected"); + assert.equal(r.verify, null, "a purged claim can't be revived by verify"); + for (const col of ["subject", "content", "metadata", "code_refs", "source_ids"]) assert.ok(!String(r[col]).includes(KEY) && !String(r[col]).includes("deploy/token"), col); + const leftovers = await db.prepare( + "SELECT (SELECT COUNT(*) FROM claims_fts WHERE claim_id = ?) AS fts, (SELECT COUNT(*) FROM code_pointers WHERE claim_id = ?) AS ptr," + + " (SELECT COUNT(*) FROM claim_usages WHERE query IS NOT NULL) AS q").bind(id, id).first() as Record; + assert.deepEqual({ ...leftovers }, { fts: 0, ptr: 0, q: 0 }); + assert.equal((await store.recall("leakedvalue0123", null, "someone")).length, 0); + assert.equal((await store.recall("deploy/token.ts", null, "someone")).length, 0); + assert.deepEqual(await logged(id), ["operator:reject", "operator:purge"]); + const reasons = (await db.prepare("SELECT reason FROM defrag_log WHERE claim_id = ? ORDER BY created_at").bind(id).all()).results.map((r) => r.reason); + assert.deepEqual(reasons, [null, "leaked key"]); +});