fix(memory): a maintenance row is audit material, not the owner's speech

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,
+    }

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"):

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])

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,),
         )

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)

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)

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.

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]

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"

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

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"