diff --git a/features/replay.py b/features/replay.py index 38ba94b..3ecc192 100644 --- a/features/replay.py +++ b/features/replay.py @@ -200,6 +200,11 @@ async def replay( written into the decision as `gate_override`, so the record shows a person chose it rather than implying the gate admitted it. + The override covers the gate, not the row's nature: a `service` row named by + id is refused before it is claimed, and counted in `refused_service`. Audit + prose is not speech, and the name cannot change that; refusing visibly (a + count, not a silent skip) tells the caller the id can never be replayed. + `max_rows` bounds a single run. A config change re-opens every row whose decisions carry the old config_hash — 564 refused rows on one live base — and without a bound the whole backlog would be distilled in one burst. Bounding it @@ -208,24 +213,41 @@ async def replay( from core import MemoryManager from graph.epistemic import EpistemicGraph from lifecycle.distiller import distill_and_route, score_text - from shared.l0 import STATUS_RECEIVED, classify_received + from shared.l0 import SERVICE_RAW_TYPE, STATUS_RECEIVED, classify_received dispatch = get_write_dispatch() cutoff = time.time() - since_days * 86400 chash = config_hash() + refused_service = 0 if ids: placeholders = ",".join("?" for _ in ids) rows = await dispatch.query( DB_NAME, - f"SELECT id, ts, layer, user_id, text, status, decisions FROM l0_journal WHERE id IN ({placeholders}) ORDER BY id", + f"SELECT id, ts, layer, user_id, text, status, decisions, raw_type FROM l0_journal WHERE id IN ({placeholders}) ORDER BY id", tuple(ids), ) + # `ids` overrides the GATE ("this one, despite the threshold"), not what + # the row IS. A service row is audit material, never speech: naming one + # must not turn the sweep summary or re-captured junk into a message. + # This branch has no raw_type predicate by design (a named import must + # stay reachable), so the refusal happens here, and it is COUNTED: a + # silent filter would be indistinguishable from "already verdicted", + # and the caller would keep retrying a row that can never be replayed. + refused_service = sum(1 for r in rows if str(r["raw_type"] or "") == SERVICE_RAW_TYPE) + if refused_service: + rows = [r for r in rows if str(r["raw_type"] or "") != SERVICE_RAW_TYPE] + logger.warning("replay: refused %d named service row(s) — audit material, not speech", refused_service) else: rows = await dispatch.query( DB_NAME, + # Service rows (sweep summary, graph-cleanup capture) are audit + # material, not speech. The distiller already refuses a machine + # report by text, but a structural fact must not depend on that + # content detector: without this filter every nightly pass claimed + # the sweep row and only then discarded it. "SELECT id, ts, layer, user_id, text, status, decisions FROM l0_journal" - " WHERE ts > ? AND (status IN ('received', 'gated_out') OR (status='processing' AND processed_at < ?)) ORDER BY id", - (cutoff, time.time() - 600.0), + " WHERE ts > ? AND raw_type != ? AND (status IN ('received', 'gated_out') OR (status='processing' AND processed_at < ?)) ORDER BY id", + (cutoff, SERVICE_RAW_TYPE, time.time() - 600.0), ) # An explicit id list is a human decision to wake that row; a window run is @@ -401,4 +423,11 @@ async def _claim(conn: Any, gate_name: str, chash_now: str) -> bool: (new_status, time.time(), json.dumps(decisions, ensure_ascii=False), row["id"]), ) processed += 1 - return {"processed": processed, "skipped": skipped, "conflicts": conflicts, "gated": gated, "limit": max_rows} + return { + "processed": processed, + "skipped": skipped, + "conflicts": conflicts, + "gated": gated, + "limit": max_rows, + "refused_service": refused_service, + } diff --git a/lifecycle/graph_enrich.py b/lifecycle/graph_enrich.py index 37ef761..65a6916 100644 --- a/lifecycle/graph_enrich.py +++ b/lifecycle/graph_enrich.py @@ -477,7 +477,7 @@ async def graph_enrich(layer: str = "user") -> dict[str, Any]: """Pre-clean JSON junk from the graph, then run miners. Returns stats.""" from graph.epistemic import EpistemicGraph from lifecycle.graph_miners import MINERS - from shared.l0 import capture + from shared.l0 import SERVICE_RAW_TYPE, capture cm = connection_manager # Housekeeping must not ride, nor wait out, somebody else's transaction: a @@ -504,7 +504,16 @@ async def graph_enrich(layer: str = "user") -> dict[str, Any]: with _timed_phase(layer, "junk_capture"): for row in junk: # capture() never raises; junk must reach L0 before its node is gone. - await capture(event="graph_cleanup", layer=layer, user_id=str(row["user_id"]), text=str(row["content"])) + # Explicit raw_type: the deleted node's prose can look like the + # owner's message (classify_raw only anchors the opening), and then + # replay distilled the junk straight back into memory. + await capture( + event="graph_cleanup", + layer=layer, + user_id=str(row["user_id"]), + text=str(row["content"]), + raw_type=SERVICE_RAW_TYPE, + ) ids.append(int(row["node_id"])) if ids: with _timed_phase(layer, "junk_delete"): diff --git a/lifecycle/l0_sweep.py b/lifecycle/l0_sweep.py index 9c4ec0b..6950771 100644 --- a/lifecycle/l0_sweep.py +++ b/lifecycle/l0_sweep.py @@ -115,6 +115,9 @@ async def _journal(summary: dict[str, Any], layer: str | None) -> None: import contextlib with contextlib.suppress(Exception): - from shared.l0 import capture + from shared.l0 import SERVICE_RAW_TYPE, capture - await capture("l0_sweep", layer or "user", "default", f"cleaner_summary: {summary}", decisions=[summary]) + # Explicit raw_type: classify_raw() reads `cleaner_summary: {...}` as a + # `user-message`, which made the sweep's own audit line a turn of the + # owner's speech for replay, L2 enrichment and the graph miners. + await capture("l0_sweep", layer or "user", "default", f"cleaner_summary: {summary}", raw_type=SERVICE_RAW_TYPE, decisions=[summary]) diff --git a/lifecycle/l0_tiers.py b/lifecycle/l0_tiers.py index 0c534b9..b1c2016 100644 --- a/lifecycle/l0_tiers.py +++ b/lifecycle/l0_tiers.py @@ -83,6 +83,9 @@ async def _close(conn: Any) -> int: # clause keeps them safe if that migration has not run or failed # halfway. The cost of being wrong here is one owner-requested # history silently closed, so the redundant guard is worth it. + # Service rows (sweep summary, graph cleanup) DO pass here on + # purpose: sweep_received skips them and replay never reads them, + # so this expiry is the only thing that closes a journal-only row. "SELECT id, decisions FROM l0_journal WHERE status='received' AND raw_type != 'import' AND ts <= ? ORDER BY id", (cutoff,), ) diff --git a/lifecycle/segment_consolidation.py b/lifecycle/segment_consolidation.py index cd03869..82a9726 100644 --- a/lifecycle/segment_consolidation.py +++ b/lifecycle/segment_consolidation.py @@ -98,13 +98,17 @@ async def segment_l0(since_hours: float = 24.0, layer: str = "user") -> dict[str consumes the segments as a grouping. """ from shared.connection import connection_manager + from shared.l0 import SERVICE_RAW_TYPE from shared.write_dispatch import get_write_dispatch cutoff = __import__("time").time() - since_hours * 3600 rows = await get_write_dispatch(connection_manager).query( DB_NAME, - "SELECT id, text FROM l0_journal WHERE ts >= ? AND layer=? ORDER BY id ASC LIMIT 500", - (cutoff, layer), + # Segmentation groups conversation into coherent batches. A maintenance + # row is a boundary the owner never spoke; letting one in would shape + # the grouping graph_enrich consumes. + "SELECT id, text FROM l0_journal WHERE ts >= ? AND layer=? AND raw_type != ? ORDER BY id ASC LIMIT 500", + (cutoff, layer, SERVICE_RAW_TYPE), ) records = [(int(r["id"]), str(r["text"])) for r in rows] segments = detect_boundaries(records) diff --git a/shared/l0.py b/shared/l0.py index 89c834f..a3905c9 100644 --- a/shared/l0.py +++ b/shared/l0.py @@ -34,6 +34,14 @@ STATUS_RECEIVED = "received" STATUS_PARKED = "parked" +#: Raw type of a maintenance job's own journal row (the sweep summary, the junk +#: re-captured by graph cleanup). These rows are audit material, not speech: +#: classifying their text made `cleaner_summary: {...}` a `user-message`, so the +#: replay distilled it and the miners bound it to the owner's sessions. A +#: dedicated type keeps them out of every `user-message` reader without teaching +#: each reader to recognise the text. +SERVICE_RAW_TYPE = "service" + #: The close reason for a row the capture path captured but never stamped: #: the body returned or raised before the write happened, so NOTHING was #: written and the row must be re-enterable. It is the one close reason that @@ -686,8 +694,11 @@ async def sweep_received(*, grace_s: float = 600.0) -> dict[str, int]: # with no age window — keeps them safe if that migration has not # run or failed halfway. The cost of being wrong is one # owner-requested history silently closed by tier_l0 the same night. - "SELECT id FROM l0_journal WHERE status=? AND raw_type != 'import'", - (STATUS_RECEIVED,), + # Service rows are skipped too: a journal-only audit line has no + # memory evidence to find, so judging it would only stamp it + # UNACCOUNTED_EXIT for a write that was never meant to happen. + "SELECT id FROM l0_journal WHERE status=? AND raw_type NOT IN ('import', ?)", + (STATUS_RECEIVED, SERVICE_RAW_TYPE), ) except Exception as exc: logger.warning("sweep_received: cannot read the journal: %s", exc) diff --git a/tests/test_features/test_replay.py b/tests/test_features/test_replay.py index df1b8e0..c960b2e 100644 --- a/tests/test_features/test_replay.py +++ b/tests/test_features/test_replay.py @@ -162,6 +162,47 @@ async def test_replay_by_ids_ignores_the_status_filter(cm: Any) -> None: assert (await _rows(cm))[0]["status"] in {"promoted_l4", "saved_l3"} +@pytest.mark.asyncio +async def test_replay_by_ids_refuses_a_service_row(cm: Any) -> None: + """A maintenance row is not speech, so naming it by id must not distil it. + + `ids` is a human override of the GATE — "this one, despite the threshold" — + not of what the row IS. A service row (sweep summary, re-captured junk from + graph cleanup) holds audit prose, and its text can pass for a user message: + the `[ariel recall]` marker mid-text defeats `classify_raw`'s prefix check, + so without this the named row was classified and distilled back into memory + — the exact resurrection the service type exists to prevent. The run must + refuse it AND say so, not skip it silently like an already-verdicted row; + the other ids named in the same call must still be replayed. + """ + from features.replay import replay + from shared.l0 import SERVICE_RAW_TYPE, capture + + junk = await capture( + "graph_cleanup", + "user", + "u1", + "старый мусор: [ariel recall] внутри текста", + raw_type=SERVICE_RAW_TYPE, + ) + wanted = await capture("new_message", "user", "u1", "я решила перейти на PostgreSQL для проекта") + assert None not in (junk, wanted) + await _age(cm, junk, wanted) + + res = await replay(since_days=1, ids=[junk, wanted]) + + by_id = {r["id"]: r for r in await _rows(cm)} + assert by_id[junk]["status"] == "received", "the service row must not be claimed" + assert by_id[wanted]["status"] in {"promoted_l4", "saved_l3"}, "a real row named alongside it still replays" + + conn = await cm.get("memory.db") + revived = await (await conn.execute("SELECT COUNT(*) FROM episodes WHERE summary LIKE '%ariel recall%'")).fetchone() + assert int(revived[0]) == 0, "deleted-node prose must not return as an episode" + + assert res["refused_service"] == 1, res + assert res["processed"] == 1, res + + @pytest.mark.asyncio async def test_replay_respects_the_importance_gate(cm: Any) -> None: """A replay must not re-admit what the live gate refused. diff --git a/tests/test_features/test_session_close.py b/tests/test_features/test_session_close.py index 09b06c5..59f9cc8 100644 --- a/tests/test_features/test_session_close.py +++ b/tests/test_features/test_session_close.py @@ -76,3 +76,23 @@ async def test_enrich_rebuilds_summary_from_l0(cm: Any) -> None: row = await (await conn.execute("SELECT summary FROM sessions WHERE session_id='sess_t7'")).fetchone() assert "postgres" in row[0] and "backup" in row[0] assert "старое резюме" not in row[0] + + +@pytest.mark.asyncio +async def test_enrich_ignores_sweep_summary(cm: Any) -> None: + """L2 enrichment binds user messages only; a sweep log line is not a topic.""" + from features.l2_enrich import enrich_sessions + from lifecycle.l0_sweep import sweep_expired + + conn = await cm.get("memory.db") + await conn.execute( + "INSERT INTO sessions (session_id, user_id, summary, started_at) VALUES (?, ?, ?, ?)", + ("sess_sweep", "default", "старое резюме", 1.0), + ) + await conn.commit() + await sweep_expired(min_remain=0) + + await enrich_sessions(days=1) + + row = await (await conn.execute("SELECT summary FROM sessions WHERE session_id='sess_sweep'")).fetchone() + assert "cleaner_summary" not in row[0] diff --git a/tests/test_lifecycle/test_graph_enrich.py b/tests/test_lifecycle/test_graph_enrich.py index 09fe190..7601843 100644 --- a/tests/test_lifecycle/test_graph_enrich.py +++ b/tests/test_lifecycle/test_graph_enrich.py @@ -409,3 +409,21 @@ def test_timed_phase_suppress_still_decides(): reached = True assert reached + + +@pytest.mark.asyncio +async def test_graph_cleanup_rows_are_service_not_user_speech(graph): + """Re-captured junk is audit material: replay must not resurrect it as speech. + + This node is junk by the `%[ariel recall]%` pattern, but the marker is mid-text, + so classify_raw() reads the prose as `user-message` and replay would distil the + deleted node straight back into memory. + """ + await graph.add_node("gu", "старый мусор: [ariel recall] внутри текста", "fact") + from lifecycle.graph_enrich import graph_enrich + + await graph_enrich(layer="user") + conn = await connection_manager.get("memory.db") + row = await (await conn.execute("SELECT raw_type FROM l0_journal WHERE event='graph_cleanup'")).fetchone() + assert row is not None + assert row["raw_type"] == "service" diff --git a/tests/test_lifecycle/test_graph_miners.py b/tests/test_lifecycle/test_graph_miners.py index e843a29..da83dfa 100644 --- a/tests/test_lifecycle/test_graph_miners.py +++ b/tests/test_lifecycle/test_graph_miners.py @@ -1333,3 +1333,21 @@ def test_no_maintenance_wrapped_function_commits_on_its_own(): if isinstance(sub, ast.Call) and isinstance(sub.func, ast.Attribute) and sub.func.attr == "commit": offenders.append(f"{path}:{sub.lineno} in {node.name}") assert not offenders, f"@_maintenance function commits on its own: {offenders}" + + +@pytest.mark.asyncio +async def test_miner_sessions_ignores_sweep_summary(db): + """A maintenance summary must not open a user session and bind nodes to it.""" + from lifecycle.graph_miners import miner_sessions + from lifecycle.l0_sweep import sweep_expired + + await sweep_expired(min_remain=0) + conn = await connection_manager.get(DB_NAME) + row = await (await conn.execute("SELECT ts FROM l0_journal WHERE event='l0_sweep' ORDER BY id DESC LIMIT 1")).fetchone() + ts = float(row["ts"]) + await _node("alpha quantum", ts + 10) + await _node("beta neutrino", ts + 30) + + result = await miner_sessions(db, "user") + + assert result["edges"] == 0 diff --git a/tests/test_lifecycle/test_ttl_sweep.py b/tests/test_lifecycle/test_ttl_sweep.py index 6e06f35..2d2a00f 100644 --- a/tests/test_lifecycle/test_ttl_sweep.py +++ b/tests/test_lifecycle/test_ttl_sweep.py @@ -219,3 +219,39 @@ async def test_sweep_scopes_to_the_named_layer(app): left = {r["key"] for r in await (await conn.execute("SELECT key FROM core_memory WHERE user_id='lay'")).fetchall()} assert "agent_exp" in left, "a user-layer sweep reached into the agent layer" assert "user_exp" not in left + + +@pytest.mark.asyncio +async def test_sweep_summary_is_service_not_user_message(app): + """The cleaner's own summary is audit material, not the owner's speech. + + classify_raw() reads `cleaner_summary: {...}` as `user-message`, so every + L0 reader that selects user messages treated the sweep's own log line as a + turn of the owner's. + """ + from lifecycle.l0_sweep import sweep_expired + + await sweep_expired(min_remain=0) + conn = await connection_manager.get("memory.db") + row = await (await conn.execute("SELECT raw_type FROM l0_journal WHERE event='l0_sweep' ORDER BY id DESC LIMIT 1")).fetchone() + assert row is not None + assert row["raw_type"] == "service" + + +@pytest.mark.asyncio +async def test_sweep_summary_never_becomes_an_episode_or_graph_node(app): + """Replay must not distil the sweep summary into L3/L4 or wire it into the graph.""" + from features.replay import replay + from lifecycle.l0_sweep import sweep_expired + + await sweep_expired(min_remain=0) + conn = await connection_manager.get("memory.db") + await conn.execute("UPDATE l0_journal SET ts = ts - 1200 WHERE event='l0_sweep'") + await conn.commit() + + await replay(since_days=1, respect_gate=False) + + episodes = await (await conn.execute("SELECT COUNT(*) FROM episodes WHERE summary LIKE 'cleaner_summary:%'")).fetchone() + nodes = await (await conn.execute("SELECT COUNT(*) FROM epi_nodes WHERE content LIKE 'cleaner_summary:%'")).fetchone() + assert int(episodes[0]) == 0, "the sweep summary became an L3 episode" + assert int(nodes[0]) == 0, "the sweep summary was wired into the graph"