diff --git a/lifecycle/forgetting.py b/lifecycle/forgetting.py index 907a53b..9572c9e 100644 --- a/lifecycle/forgetting.py +++ b/lifecycle/forgetting.py @@ -135,6 +135,78 @@ def _carried_into_core(row: Any) -> frozenset[str]: return _CARRIED_INTO_CORE - {"memory_kind"} +def _episode_requested(row: Any) -> tuple[float, float | None]: + """Return the weight and time a restore asks `save()` to write, in one place. + + Two callers must agree: `_restore_episode` writes the pair, and + `restore_entries` asks about it in `dry_run` — the same question, before the + write the preview does not run. A second copy of the rule would let the + preview and the pass disagree about which values were requested, which is the + same shape of defect `_split_core_content` exists to prevent on the core route. + + `None` is checked, not truthiness. `row["importance"] or 0.5` — the earlier + form — treats a recorded `0.0` as absent and returns the episode with a + different weight than it had. **This was live, not latent.** An earlier + version of this comment said the base held no such row; that was measured on + one instance and is false of the set of them. Measured 2026-10-11 from copies: + the `-hermes` base holds one archived episode with `importance=0.0` and reason + `forget_primitive_fuzzy_l3`, so the old form would have restored that episode + at 0.5. There is no single base to measure — the archive is per instance. + + The bin now carries the moment the episode happened. It is absent on rows + archived before the column existed (3159 of them on one live base, + measured 2026-11-03), and `archived_at` is not a substitute: it is when + the row was removed, not when it happened. So a missing value keeps the + old behaviour and the episode is stamped with its return. + + The key check is for a base whose bin predates the column and where + `ensure()` has not run in this process yet — `restore_entries` reads the + table directly and must not crash on a schema it can still serve. + `in row` would test VALUES, not column names on sqlite3.Row, so the key + check has to go through `.keys()` (SIM118 does not apply to this type). + """ + weight = 0.5 if row["importance"] is None else float(row["importance"]) + created_at = None + if "original_created_at" in row.keys(): # noqa: SIM118 + recorded = row["original_created_at"] + created_at = None if recorded is None else float(recorded) + return weight, created_at + + +def _episode_not_applied(actual: Any, created_at: float | None, weight: float) -> frozenset[str]: + """Name the requested values the row in `episodes` does not hold — one comparison. + + `_restore_episode` asks this of its post-write read-back, and `restore_entries` + asks it of the dedup survivor in `dry_run`: the same rule in the one place both + modes read, which is what keeps the two answers identical about the same row. + A `None` `created_at` was never requested, so it cannot be "not applied". + """ + not_applied: set[str] = set() + if actual is not None: + if created_at is not None and actual["created_at"] != created_at: + not_applied.add("created_at") + if actual["emotional_weight"] != weight: + not_applied.add("emotional_weight") + return frozenset(not_applied) + + +async def _episode_dedup_survivor(conn: Any, layer: str, user_id: str, summary: str) -> Any: + """Return the row `EpisodicMemory.save` would dedup against, or `None`. + + Precisely the dedup query `save()` runs before it writes + (`core/episodic.py:91-94`) — same columns in the WHERE, same ordering — asked + in `dry_run` so the preview and the pass cannot answer differently about + whether a restore will write or dedup. It is a read, and the preview writes + nothing else. + """ + return await ( + await conn.execute( + "SELECT created_at, emotional_weight FROM episodes WHERE layer=? AND user_id=? AND summary=? ORDER BY episode_id ASC LIMIT 1", + (layer, user_id, summary), + ) + ).fetchone() + + async def _not_carried(conn: Any, table: str, carried: frozenset[str]) -> set[str]: """Return the columns of `table` an archive row cannot fill — **read from the schema**. @@ -478,10 +550,23 @@ async def restore_entries(self, ids: list[int], user_id: str, memory_type: str | } ) continue + not_applied: frozenset[str] = frozenset() + if destination == "episodes" and dry_run: + # Asked in the preview because the pass answers it, and the two + # must not answer differently about the same row: `_restore_episode` + # reads the surviving row back after `save()` and reports which + # requested values it does not hold. A preview that skipped the + # question reported a deduped row's `created_at` as returned while + # the pass reported it as lost. Closed the way the core key check + # above closes it: the question the write will ask is asked in both + # modes, and the write itself still does not run in a preview. + weight, created_at = _episode_requested(row) + survivor = await _episode_dedup_survivor(conn, self.layer, user_id, str(row["content"])) + not_applied = _episode_not_applied(survivor, created_at, weight) if not dry_run: try: if destination == "episodes": - await self._restore_episode(row, user_id) + _, not_applied = await self._restore_episode(row, user_id) else: await self._restore_core(conn, row, original_id, now) await conn.execute("DELETE FROM archived_memories WHERE id = ?", (row["id"],)) @@ -516,7 +601,7 @@ async def restore_entries(self, ids: list[int], user_id: str, memory_type: str | # The limit, per row, where the answer is read. A caller who # gets a summary back and no tags should learn why from this # answer rather than from a missing tag weeks later. - "not_recorded": sorted(await _not_carried(conn, destination, carried)), + "not_recorded": sorted(await _not_carried(conn, destination, carried - not_applied)), } ) # Only the pass commits. An unconditional `commit()` here was reached on the @@ -557,7 +642,7 @@ async def restore_entries(self, ids: list[int], user_id: str, memory_type: str | ) return outcome - async def _restore_episode(self, row: Any, user_id: str) -> int: + async def _restore_episode(self, row: Any, user_id: str) -> tuple[int, frozenset[str]]: """Put an archived episode back into `episodes`, where it came from. `EpisodicMemory.save()` dedups on `(layer, user_id, summary)` and returns @@ -569,35 +654,30 @@ async def _restore_episode(self, row: Any, user_id: str) -> int: The episode is given back with the time it happened when the bin recorded it (`original_created_at`). A row archived before that column existed has no time to give back, so it is stamped with the moment of its return. + + Returns the id and the carried fields the **surviving row** did not get. + The read-back is deliberate: when the dedup fires, `save()` writes nothing + and the row that was already there keeps its own `created_at` and + `emotional_weight`. Reporting the bin's capability instead of the row's + state would tell a caller those values came back when they did not. + `restore_entries` asks the same question in `dry_run` — through the same + dedup query and the same comparison — so the preview and this pass answer + identically about the same row. """ from core.episodic import EpisodicMemory mem = EpisodicMemory(cm=self._cm, layer=self.layer) - # `None` is checked, not truthiness. `row["importance"] or 0.5` — the earlier - # form — treats a recorded `0.0` as absent and returns the episode with a - # different weight than it had. **This was live, not latent.** An earlier - # version of this comment said the base held no such row; that was measured on - # one instance and is false of the set of them. Measured 2026-10-11 from copies: - # the `-hermes` base holds one archived episode with `importance=0.0` and reason - # `forget_primitive_fuzzy_l3`, so the old form would have restored that episode - # at 0.5. There is no single base to measure — the archive is per instance. - weight = 0.5 if row["importance"] is None else float(row["importance"]) - # The bin now carries the moment the episode happened. It is absent on rows - # archived before the column existed (3159 of them on one live base, - # measured 2026-11-03), and `archived_at` is not a substitute: it is when - # the row was removed, not when it happened. So a missing value keeps the - # old behaviour and the episode is stamped with its return. - # - # The key check is for a base whose bin predates the column and where - # `ensure()` has not run in this process yet — `restore_entries` reads the - # table directly and must not crash on a schema it can still serve. - created_at = None - # `in row` would test VALUES, not column names on sqlite3.Row, so the key - # check has to go through `.keys()` (SIM118 does not apply to this type). - if "original_created_at" in row.keys(): # noqa: SIM118 - recorded = row["original_created_at"] - created_at = None if recorded is None else float(recorded) - return await mem.save(user_id, str(row["content"]), emotional_weight=weight, created_at=created_at) + weight, created_at = _episode_requested(row) + episode_id = await mem.save(user_id, str(row["content"]), emotional_weight=weight, created_at=created_at) + # Read the row back through the dispatcher that wrote it: the shared + # connection `restore_entries` holds may already be inside a transaction + # (its per-row `DELETE`), whose snapshot predates this write. + actual = await mem._dispatch.select_one( + DB_NAME, + "SELECT created_at, emotional_weight FROM episodes WHERE episode_id = ?", + (episode_id,), + ) + return episode_id, _episode_not_applied(actual, created_at, weight) async def _restore_core(self, conn: Any, row: Any, original_id: int, now: float) -> None: """Put an archived L4 row back into `core_memory`. diff --git a/scripts/purge_foreign_rows.py b/scripts/purge_foreign_rows.py index fd00489..976432f 100755 --- a/scripts/purge_foreign_rows.py +++ b/scripts/purge_foreign_rows.py @@ -32,11 +32,16 @@ episode, so a text that legitimately belongs to the persona is never taken out because a foreign row happened to produce the same summary. -NOTHING IS ERASED - -Every removed row is copied to `archived_memories` with its reason before the -delete, using the same class `features.compression` uses. Deleting without -archiving would be the same class of mistake as the bug that put the rows there. +WHAT IS ERASED, AND WHAT IS NOT + +Episodes and `core_memory` facts are copied to `archived_memories` with reason +`foreign_agent_contamination` before the delete, using the same class +`features.compression` uses. The graph is not: `epi_nodes`, their `epi_edges` +and `epi_tags` are deleted outright. A node's `content` is derived text — it +already lives in an episode or a fact — and this repository has no restore route +for a node, so there is nothing a restore could return. The cost is stated rather +than hidden: **after this purge the graph cannot be rebuilt from the archive**, +and the edges removed with a node are gone with it. **These rows are not restorable through anything an agent can call.** The archive reason is `foreign_agent_contamination`, and `PURGE_REASONS` makes that a refusal diff --git a/scripts/purge_speaker_layer_leaks.py b/scripts/purge_speaker_layer_leaks.py index 1cb26ef..4238bea 100755 --- a/scripts/purge_speaker_layer_leaks.py +++ b/scripts/purge_speaker_layer_leaks.py @@ -34,19 +34,30 @@ the move would make junk my own PERMANENT canon instead of the owner's — the same defect, relocated. They are archived here, so nothing is lost. -NOTHING IS ERASED - -Every removed row is copied to `archived_memories` with reason `speaker_layer_leak` -before the delete. That reason is deliberately NOT in `PURGE_REASONS`: removal was -deliberate, but restorable, and a restorable removal is the safer default. - -EVERY REMOVAL LEAVES A TRAIL - -One `audit_log` row per removed record — `action='speaker_layer_leak_purge'`, the -shape `purge_l3_legacy` writes. Per-record and not a summary: `features/blame.py` -answers «what happened to this row» with `WHERE target_id=?`, and -`lifecycle/importance_scheduler.py` counts `target_id`, so a single summary row -would answer neither question. +WHAT IS ERASED, AND WHAT IS NOT + +`core_memory` facts and episodes are copied to `archived_memories` with reason +`speaker_layer_leak` before the delete. That reason is deliberately NOT in +`PURGE_REASONS`: removal was deliberate, but restorable, and a restorable removal +is the safer default. + +The graph is not archived. `epi_nodes`, their `epi_edges` and `epi_tags` are +deleted outright, and the audit row for a node says `deleted`, never `archive`: +a node's `content` is derived text — it already lives in an episode or a fact — +and this repository has no restore route for a node, so there is nothing a +restore could return. The cost is stated plainly: **after this purge +the graph cannot be rebuilt from the archive**, and the edges removed with a node +are gone with it. + +WHAT LEAVES A TRAIL, AND WHAT DOES NOT + +Facts, episodes and nodes each get one `audit_log` row — +`action='speaker_layer_leak_purge'`, the shape `purge_l3_legacy` writes. +Per-record and not a summary: `features/blame.py` answers «what happened to this +row» with `WHERE target_id=?`, and `lifecycle/importance_scheduler.py` counts +`target_id`, so a single summary row would answer neither question. `epi_edges` +and `epi_tags` leave no audit row of their own: they are deleted with their node, +and the node's row is the only trail they get. USAGE @@ -178,11 +189,12 @@ async def _run(conn: Any) -> tuple[int, int, int]: """ async def _audit(target_id: str, details: dict[str, Any]) -> None: - """One row per removed record, the shape `purge_l3_legacy` writes. + """One row per removed fact, episode or node — edges and tags get none. - Per-record and not a summary line: `features/blame.py` answers «what happened - to this row» with `WHERE target_id=?`, and `lifecycle/importance_scheduler.py` - counts `target_id`. A single summary row would answer neither. + The shape is the one `purge_l3_legacy` writes. Per-record and not a summary + line: `features/blame.py` answers «what happened to this row» with + `WHERE target_id=?`, and `lifecycle/importance_scheduler.py` counts + `target_id`. A single summary row would answer neither. """ await conn.execute( "INSERT INTO audit_log (user_id, action, layer, target_id, details, timestamp) VALUES (?,?,?,?,?,?)", @@ -245,7 +257,11 @@ async def _audit(target_id: str, details: dict[str, Any]) -> None: await conn.execute(f"DELETE FROM epi_tags WHERE node_id IN ({ph})", tuple(p["node_ids"])) await conn.execute(f"DELETE FROM epi_nodes WHERE node_id IN ({ph})", tuple(p["node_ids"])) for nid in p["node_ids"]: - await _audit(f"epi_node:{int(nid)}", {"klass": ARCHIVE_REASON, "archive": f"archived_memories:{ARCHIVE_REASON}"}) + # A node's `content` is derived text and there is no restore route + # for a node, so nothing went to `archived_memories`. `deleted` states + # the fact the other rows state with `archive`; an `archive` key here + # would claim a copy that does not exist and that no restore could return. + await _audit(f"epi_node:{int(nid)}", {"klass": ARCHIVE_REASON, "deleted": True, "table": "epi_nodes"}) return len(ids), len(ep_ids), len(p["node_ids"]) diff --git a/tests/test_lifecycle/test_restore_entries.py b/tests/test_lifecycle/test_restore_entries.py index f545b01..bbb7f99 100644 --- a/tests/test_lifecycle/test_restore_entries.py +++ b/tests/test_lifecycle/test_restore_entries.py @@ -718,3 +718,126 @@ async def test_the_listing_never_offers_what_the_restore_refuses(cm: Any) -> Non for original_id in hidden: outcome = await fs.restore_entries([original_id], user_id=U1) assert outcome.refused, f"the listing hid id {original_id}, but the restore accepts it" + + +async def test_a_deduped_restore_reports_what_the_surviving_row_kept(cm: Any) -> None: + """The report describes the row in the table, not the bin's capability. + + `EpisodicMemory.save()` dedups on `(layer, user_id, summary)` and returns the + existing id without writing. The bin held a time and a weight, but the row + already in `episodes` kept its own, so `not_recorded` must name both: a caller + reading the answer would otherwise believe the archived values came back. + """ + conn = await cm.get("memory.db") + await conn.execute( + "INSERT INTO episodes (layer, user_id, summary, emotional_weight, tags, created_at)" + " VALUES ('user', ?, 'same words', 0.9, '[]', 1700099999.0)", + (U1,), + ) + await conn.commit() + await _archive( + cm, + original_id=61, + content="same words", + memory_type="episode", + importance=0.2, + original_created_at=1700000000.0, + ) + + outcome = await _fs(cm).restore_entries([61], user_id=U1) + + assert outcome.restored == 1 + not_recorded = outcome.restored_items[0]["not_recorded"] + assert "created_at" in not_recorded, "the deduped row kept its own time, not the archived one" + assert "emotional_weight" in not_recorded, "the deduped row kept its own weight, not the archived one" + + +async def test_a_clean_restore_does_not_report_the_time_or_weight_as_lost(cm: Any) -> None: + """Control for the dedup report: with no existing row both values do land.""" + t0 = 1700000000.0 + await _archive( + cm, + original_id=62, + content="a fresh moment", + memory_type="episode", + importance=0.2, + original_created_at=t0, + ) + + outcome = await _fs(cm).restore_entries([62], user_id=U1) + + assert outcome.restored == 1 + not_recorded = outcome.restored_items[0]["not_recorded"] + assert "created_at" not in not_recorded + assert "emotional_weight" not in not_recorded + row = (await _rows(cm, "SELECT created_at, emotional_weight FROM episodes WHERE summary='a fresh moment'"))[0] + assert row["created_at"] == pytest.approx(t0) + assert row["emotional_weight"] == pytest.approx(0.2) + + +async def test_dry_run_and_pass_report_the_same_not_recorded_for_a_deduped_restore(cm: Any) -> None: + """The preview is the pass: one question, answered identically in both modes. + + `EpisodicMemory.save()` dedups on `(layer, user_id, summary)`. When the text is + already in `episodes`, the pass reads the surviving row back and reports its + `created_at` and `emotional_weight` as not returned. The preview used to skip + that question — `not_applied` stayed empty in `dry_run` — and reported both + values as returned for the same row the pass reported them as lost. Measured + on scratch: preview `not_recorded = [episode_id, layer, memory_kind, tags]`, + pass adds `[created_at, emotional_weight]`. The preview must run the same + dedup query `save()` runs (`core/episodic.py:91-94`) and say what the pass + will say; the core key check at `lifecycle/forgetting.py:460-466` closes the + same class of drift the same way. + """ + conn = await cm.get("memory.db") + await conn.execute( + "INSERT INTO episodes (layer, user_id, summary, emotional_weight, tags, created_at)" + " VALUES ('user', ?, 'same words', 0.9, '[]', 1700099999.0)", + (U1,), + ) + await conn.commit() + await _archive( + cm, + original_id=63, + content="same words", + memory_type="episode", + importance=0.2, + original_created_at=1700000000.0, + ) + + preview = await _fs(cm).restore_entries([63], user_id=U1, dry_run=True) + + assert preview.restored == 0, "a preview restores nothing" + assert len(await _rows(cm, "SELECT episode_id FROM episodes WHERE summary='same words'")) == 1, "the preview wrote a row" + assert await _archived_count(cm, 63) == 1, "the preview consumed the archive row" + + actual = await _fs(cm).restore_entries([63], user_id=U1) + + assert actual.restored == 1 + assert preview.restored_items[0]["not_recorded"] == actual.restored_items[0]["not_recorded"], ( + "the preview and the pass answered differently about the same row" + ) + assert "created_at" in preview.restored_items[0]["not_recorded"], "the preview reported a deduped time as returned" + assert "emotional_weight" in preview.restored_items[0]["not_recorded"], "the preview reported a deduped weight as returned" + + +async def test_dry_run_and_pass_agree_on_a_clean_restore(cm: Any) -> None: + """Control for the drift test: with no surviving row both modes say the values + land. Without this the repair could collapse into "the preview always reports + the values as lost", which would pass the dedup test and still be a lie.""" + await _archive( + cm, + original_id=64, + content="a fresh moment", + memory_type="episode", + importance=0.2, + original_created_at=1700000000.0, + ) + + preview = await _fs(cm).restore_entries([64], user_id=U1, dry_run=True) + actual = await _fs(cm).restore_entries([64], user_id=U1) + + assert actual.restored == 1 + assert preview.restored_items[0]["not_recorded"] == actual.restored_items[0]["not_recorded"] + assert "created_at" not in preview.restored_items[0]["not_recorded"] + assert "emotional_weight" not in preview.restored_items[0]["not_recorded"] diff --git a/tests/test_scripts/test_purge_speaker_layer_leaks.py b/tests/test_scripts/test_purge_speaker_layer_leaks.py index 784ba2d..e1dbec3 100644 --- a/tests/test_scripts/test_purge_speaker_layer_leaks.py +++ b/tests/test_scripts/test_purge_speaker_layer_leaks.py @@ -66,6 +66,17 @@ async def _episode(layer: str, summary: str, raw_id: int) -> int: return int(cur.lastrowid or 0) +async def _node(layer: str, content: str) -> int: + conn = await connection_manager.get(DB_NAME) + cur = await conn.execute( + "INSERT INTO epi_nodes (layer, user_id, content, node_type, tags, confidence, created_at)" + " VALUES (?, 'default', ?, 'concept', '[]', 0.5, 1.0)", + (layer, content), + ) + await conn.commit() + return int(cur.lastrowid or 0) + + async def test_selects_only_the_owner_row_whose_source_is_the_agent(db) -> None: """Слой говорящего — по источнику: agent-L0 в user-слое = утечка, user-L0 — нет.""" from scripts.purge_speaker_layer_leaks import plan @@ -256,3 +267,36 @@ async def test_apply_removes_the_episodes_the_plan_reports(db, tmp_path) -> None assert left == 0, "эпизод из плана обязан быть удалён, а не только показан" archived = await (await conn.execute("SELECT COUNT(*) FROM archived_memories WHERE archive_reason=?", ("speaker_layer_leak",))).fetchone() assert int((archived or (0,))[0]) == 2, "заархивированы и запись, и эпизод" + + +async def test_a_deleted_node_is_not_recorded_as_archived(db, tmp_path) -> None: + """A node's `content` is derived text and there is no restore route for a node; + the audit line must say deleted. + + The graph is derived structure: `apply()` deletes `epi_nodes`/`epi_edges`/`epi_tags` + without calling `archive()`. The audit row used to carry + `archive: archived_memories:speaker_layer_leak`, which told a reader that a copy + existed when none did. + """ + from pathlib import Path + + agent_l0 = await _l0("agent", "реплика агента") + await _fact("user", "rule:утекло", "моя реплика", agent_l0) + nid = await _node("user", "моя реплика") + + repo = Path(__file__).resolve().parents[2] + proc = _run_cli(repo, _cli_env(tmp_path), "--apply") + assert proc.returncode == 0, proc.stderr[-800:] + + conn = await connection_manager.get(DB_NAME) + rows = await ( + await conn.execute( + "SELECT details FROM audit_log WHERE action='speaker_layer_leak_purge' AND target_id=?", + (f"epi_node:{nid}",), + ) + ).fetchall() + assert len(rows) == 1, "a removed node must leave one audit row" + details = json.loads(str(rows[0][0])) + assert details.get("deleted") is True, f"the audit line does not say the node was deleted: {details}" + assert "archive" not in details, f"the audit line still claims an archived copy: {details}" + assert "archived_memories" not in json.dumps(details), f"the audit line still names the bin: {details}"