feat(p1-03): typed persona lane + write-time memory_type classification (#26)

Gate finding 2: the typed persona source was dead in production — the
operator's single memory contributor calls select_for_task WITHOUT
types, so _persona_recall always returned []; and no production writer
ever set memory_type. The store source now always runs a SECOND typed
query (self/personality) merged into the candidates — still thresholded
inside the store's gate, deduplicated by entry id, may abstain (a
trivial task still gets zero persona rows). add_entry gains an additive
memory_type param (default None = unclassified, eligible everywhere)
plus a --memory-type store CLI flag wired through the dedupe gate;
wired where the writer genuinely knows the type: warm-router neuron
writer -> associative, sleep-time growth sync -> personality
(persona-targeted) / semantic.

Gate finding 3: _persona_recall ignored the selector's pinned db_path
(bare memory_store.connect() opened the module-default store). The pin
is now bound into sources that declare db_path at registration time;
both store-backed sources honor it.

Red-first evidence: no typed call in the live contributor path;
add_entry TypeError; persona block missing from the pinned-only store.

src/rune/memory_selector.py

@@ -31,6 +31,8 @@ exploding source never takes down assembly.
 
 from __future__ import annotations
 
+import functools
+import inspect
 import logging
 from dataclasses import dataclass, field
 from typing import Callable
@@ -103,7 +105,25 @@ class MemorySelector:
     def __init__(self, db_path: str | None = None, min_score: float | None = None):
         self.db_path = db_path if db_path is not None else memory_store.DB_PATH
         self.min_score = float(min_score) if min_score is not None else float(memory_store.RECALL_MIN_SCORE)
-        self._sources: dict[str, Callable] = dict(DEFAULT_SOURCES)
+        # P1-03 (#26) DISCHARGE (finding 3): the selector's PINNED db_path is
+        # bound into store-backed sources at registration time — a bare
+        # ``memory_store.connect()`` inside a source used to open the
+        # module-default store, ignoring the pin. Only sources that actually
+        # declare a ``db_path`` parameter are bound (replacement sources
+        # registered via add_source keep the documented 4-arg contract).
+        self._sources: dict[str, Callable] = {}
+        for name, fn in DEFAULT_SOURCES.items():
+            try:
+                params = inspect.signature(fn).parameters
+                has_db_path = "db_path" in params or any(
+                    p.kind is inspect.Parameter.VAR_KEYWORD for p in params.values()
+                )
+            except (TypeError, ValueError):
+                has_db_path = False
+            if has_db_path:
+                self._sources[name] = functools.partial(fn, db_path=self.db_path)
+            else:
+                self._sources[name] = fn
 
     # -- source registry -----------------------------------------------------
 

src/rune/memory_store.py

@@ -505,6 +505,7 @@ def add_entry(
     temperature=None,
     always_on=0,
     priority=5,
+    memory_type=None,
 ):
     """Add an atomic memory entry with its embedding + FTS + metadata.
 
@@ -514,10 +515,18 @@ def add_entry(
     is — stale facts rank lower but are never deleted (lukewarm, not cold).
     `always_on` (0/1) + `priority` (0-10) mark the always-in-context core:
     constitution, safety, core identity, operating essentials. These are
-    compiled by memory_compiler.py into the session's core context."""
+    compiled by memory_compiler.py into the session's core context.
+    P1-03 (#26) DISCHARGE (finding 2b): optional `memory_type` classifies at
+    WRITE time (episodic|semantic|procedural|self|personality|associative —
+    see MEMORY_TYPES); None (default) leaves the row unclassified, which is
+    eligible for EVERY typed query (classification narrows, never excludes)."""
     text = (text or "").strip()
     if not text or len(text) < 8:
         return False
+    if memory_type is not None:
+        memory_type = str(memory_type)
+        if memory_type not in MEMORY_TYPES:
+            raise ValueError(f"invalid memory_type {memory_type!r}; expected one of {sorted(MEMORY_TYPES)}")
     vec = _embed([text]).get(text)
     ts = now_iso()
     conf = 0.7 if confidence is None else float(confidence)
@@ -525,8 +534,8 @@ def add_entry(
     cur = db.execute(
         "INSERT INTO entries (text, importance, created_at, last_accessed, "
         "topic, entities, source, method, status, valid_from, valid_until, "
-        "confidence, temperature, always_on, priority, tier) "
-        "VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
+        "confidence, temperature, always_on, priority, tier, memory_type) "
+        "VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
         (
             text,
             float(importance),
@@ -544,6 +553,7 @@ def add_entry(
             int(always_on),
             int(priority),
             "warm",
+            memory_type,
         ),
     )
     eid = cur.lastrowid

src/rune/sleep_time.py

@@ -743,7 +743,13 @@ def main():
                              "--topic", ev.get("target", "general"),
                              "--importance", str(min(0.95, 0.4 + 0.1 * int(ev.get("strength", 1) or 1))),
                              "--source", ev.get("source", "curator"),
-                             "--method", ev.get("source", "curator")],
+                             "--method", ev.get("source", "curator"),
+                             # P1-03 (#26) discharge (finding 2b): growth
+                             # evidence targets persona/identity growth —
+                             # the writer genuinely knows the type, so
+                             # classify at write time.
+                             "--memory-type",
+                             "personality" if ev.get("target") == "persona" else "semantic"],
                             capture_output=True, text=True, timeout=30)
                         if '"added": True' in (r.stdout or ""):
                             added += 1

src/rune/warm_router.py

@@ -277,20 +277,20 @@ def write_neuron(slug, triggers, body, importance=0.5):
                                   "AND slug=?", (slug,)).fetchone()
             if existing:
                 db.execute("UPDATE entries SET text=?, triggers=?, importance=?, "
-                           "last_accessed=?, summary=? WHERE id=?",
+                           "last_accessed=?, summary=?, memory_type=? WHERE id=?",
                            ((body or "").strip(), ", ".join(triggers),
-                            float(importance), ts, _aaak(body), existing[0]))
+                            float(importance), ts, _aaak(body), "associative", existing[0]))
                 eid = existing[0]
             else:
                 cur = db.execute(
                     "INSERT INTO entries (text, importance, created_at, "
                     "last_accessed, topic, source, method, status, valid_from, "
                     "confidence, temperature, always_on, priority, triggers, "
-                    "kind, slug, summary) "
-                    "VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
+                    "kind, slug, summary, memory_type) "
+                    "VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
                     ((body or "").strip(), float(importance), ts, ts, "warm",
                      "warm_router", "warm_router", "active", ts, 0.9, 1.0, 0, 5,
-                     ", ".join(triggers), "neuron", slug, _aaak(body)))
+                     ", ".join(triggers), "neuron", slug, _aaak(body), "associative"))
                 eid = cur.lastrowid
             db.commit()
         finally:

tests/test_p1_03_memory_selection.py

@@ -1160,3 +1160,133 @@ def test_recall_dense_only_above_floor_paraphrase_selected(monkeypatch):
         assert any("overview of the money" in r["text"] for r in results)
     finally:
         db.close()
+
+# --- Finding 2: typed persona lane alive in production; write-time types ---
+
+
+def test_contributor_memory_runs_typed_persona_query(tmp_path, monkeypatch):
+    """Discharge #2a: the typed persona query (memory_types=['self',
+    'personality']) executes in the LIVE contributor path — the operator's
+    untyped select_for_task call used to leave the persona source dead."""
+    _no_embed(monkeypatch)
+    op = _operator(tmp_path)
+    broker = op._context_broker()
+    seen: list = []
+    original = ms.recall
+
+    def spy(db, query, **kwargs):
+        if kwargs.get("memory_types"):
+            seen.append(list(kwargs["memory_types"]))
+        return original(db, query, **kwargs)
+
+    monkeypatch.setattr(ms, "recall", spy)
+    task = mock.Mock(description="kubernetes pods restart")
+    task.task_id = "t1"
+    op._contributor_memory(task, {})
+    assert ["self", "personality"] in seen, (
+        "the typed persona query must execute in the live contributor path"
+    )
+
+
+def test_persona_lane_merges_without_duplicate_candidates(monkeypatch):
+    """Discharge #2a: the store source's second typed query merges into the
+    candidates — rows that BOTH queries return (unclassified rows match
+    untyped AND typed) must appear ONCE, not double-count (duplicates would
+    corrupt the irrelevant-memory rate and can produce repeated block text)."""
+    _no_embed(monkeypatch)
+    db = ms.connect()
+    try:
+        _add(db, "Kubernetes pods restart automatically on failure")
+        _add(db, "Kubernetes pods restart policy for the cluster")
+    finally:
+        db.close()
+    from rune.memory_selector import MemorySelector
+
+    sel = MemorySelector(db_path=ms.DB_PATH)
+    sel.add_source("neurons", lambda query, types, budget, min_score: [])
+    sel.add_source("graphs", lambda query, types, budget, min_score: [])
+    sel.add_source("abstraction", lambda query, types, budget, min_score: [])
+    sel.add_source("consolidated", lambda query, types, budget, min_score: [])
+    sel.add_source("persona", lambda query, types, budget, min_score: [])
+    result = sel.select_for_task("kubernetes pods restart")
+    texts = [b.content for b in result.blocks]
+    assert result.candidates_evaluated == 2, "two shared rows must be merged once each"
+    assert len(texts) == len(set(texts)), "merged candidates must not duplicate block text"
+
+
+def test_persona_lane_still_abstains_on_trivial_task(monkeypatch):
+    """Discharge #2a: the lane is thresholded like every source — a trivial
+    task through the UNTYPED production caller still gets zero memory."""
+    _no_embed(monkeypatch)
+    db = ms.connect()
+    try:
+        _add(db, "I speak in one clean sentence, evidence first", memory_type="personality", importance=0.95)
+    finally:
+        db.close()
+    from rune.memory_selector import MemorySelector
+
+    sel = MemorySelector(db_path=ms.DB_PATH)
+    result = sel.select_for_task("what time is it")
+    assert result.blocks == []
+    assert result.abstained is True
+
+
+def test_add_entry_memory_type_round_trip(monkeypatch):
+    """Discharge #2b: add_entry classifies at write time (additive optional
+    param, default None = unclassified/eligible everywhere)."""
+    _no_embed(monkeypatch)
+    db = ms.connect()
+    try:
+        assert ms.add_entry(db, "I speak in one clean sentence, evidence first", memory_type="personality")
+        assert ms.add_entry(db, "Kubernetes pods restart automatically on failure")
+        rows = {r["text"]: r["memory_type"] for r in db.execute("SELECT text, memory_type FROM entries")}
+        assert rows["I speak in one clean sentence, evidence first"] == "personality"
+        assert rows["Kubernetes pods restart automatically on failure"] is None, (
+            "default must stay NULL (unclassified = eligible for every query)"
+        )
+    finally:
+        db.close()
+
+
+# --- Finding 3: sources honor the selector's pinned db_path ----------------
+
+
+def test_persona_source_uses_pinned_db_path(tmp_path, monkeypatch):
+    """Discharge #3: the persona source must query the selector's PINNED
+    store — not memory_store.connect()'s module default."""
+    _no_embed(monkeypatch)
+    pinned = str(tmp_path / "pinned.db")
+    pdb = ms.connect(pinned)
+    try:
+        _add(pdb, "I speak in one clean sentence, evidence first", memory_type="personality")
+    finally:
+        pdb.close()
+    from rune.memory_selector import MemorySelector
+
+    sel = MemorySelector(db_path=pinned)
+    sel.add_source("store", lambda query, types, budget, min_score: [])  # isolate the persona source
+    result = sel.select_for_task("how do I speak in one clean sentence", types=["self", "personality"])
+    assert any("one clean sentence" in b.content for b in result.blocks), (
+        "persona source must read the PINNED db_path, not the module default"
+    )
+
+
+def test_store_source_uses_pinned_db_path(tmp_path, monkeypatch):
+    """Discharge #3 twin: the store source must honor the pinned db_path too
+    (same unbound-default hole)."""
+    _no_embed(monkeypatch)
+    pinned = str(tmp_path / "pinned-store.db")
+    pdb = ms.connect(pinned)
+    try:
+        _add(pdb, "Kubernetes pods restart automatically on failure")
+        _add(pdb, "Kubernetes pods restart policy for the cluster")
+    finally:
+        pdb.close()
+    from rune.memory_selector import MemorySelector
+
+    sel = MemorySelector(db_path=pinned)
+    sel.add_source("persona", lambda query, types, budget, min_score: [])  # isolate the store source
+    result = sel.select_for_task("kubernetes pods restart")
+    assert any("Kubernetes" in b.content for b in result.blocks), (
+        "store source must read the PINNED db_path, not the module default"
+    )