merge research-structuralization (ROBOT_TOKEN in auto-version + canary contract update)

CI canary root cause: get_shard() returned a raw sqlite3 connection —
virtual tables need the extension loaded PER CONNECTION, so recall() on a
shard crashed 'no such table: entries_vec' in clean environments. Also:
create_shard's schema lacked entries_vec entirely (main store had it,
shards didn't — recall routes to shards). Both fixed; 3 red-first tests
frozen green (canary failure class, schema completeness, shard reuse).

.gitea/workflows/auto-version.yml

@@ -31,7 +31,7 @@ jobs:
 
       - name: Auto-bump if the PR did not carry a version bump
         env:
-          GITEA_TOKEN: ${{ secrets.GITEA_TOKEN }}
+          GITEA_TOKEN: ${{ secrets.ROBOT_TOKEN }}
           RELEASE_SERVER_URL: ${{ gitea.server_url }}
           RELEASE_REPO: ${{ gitea.repository }}
           PR_TITLE: ${{ gitea.event.pull_request.title }}
@@ -126,20 +126,31 @@ jobs:
           drifts = []
           for rel, pat in checks:
               if rel not in edited:
-                  # unchanged surface: must ALREADY carry NEW
+                  # unchanged surface: must ALREADY carry NEW — if it carries a
+                  # DIFFERENT version (base drift, e.g. pyproject at 0.49.0 while
+                  # master is 0.54.1), RECOVER by force-editing to NEW. The gate
+                  # stays: nothing commits unless every surface carries NEW.
                   status, cur = api(f"/contents/{rel}?ref=master")
-                  text = base64.b64decode(cur.get("content", "")).decode("utf-8", "replace") if status == 200 else ""
+                  if status != 200:
+                      drifts.append(f"{rel}: unreadable ({status})")
+                      continue
+                  text = base64.b64decode(cur.get("content", "")).decode("utf-8", "replace")
                   hits = re.findall(pat, text)
                   if NEW not in hits:
-                      drifts.append(f"{rel}: unchanged but does not carry {NEW} (found {sorted(set(hits))})")
+                      fixed = re.sub(pat, lambda m: m.group(0).replace(m.group(1), NEW), text, count=1)
+                      if fixed != text:
+                          edited[rel] = (fixed, cur["sha"])
+                          print(f"  RECOVERED {rel}: base drift {sorted(set(hits))} -> {NEW}")
+                      else:
+                          drifts.append(f"{rel}: pattern present but unfixable (found {sorted(set(hits))})")
               else:
                   hits = re.findall(pat, edited[rel][0])
                   if NEW not in hits:
                       drifts.append(f"{rel}: edited content does not carry {NEW}")
           if drifts:
-              print("VERSION-SURFACE DRIFT — bump aborted, nothing committed:", file=sys.stderr)
+              print("VERSION-SURFACE DRIFT — bump aborted, nothing committed:")
               for d in drifts:
-                  print(f"  - {d}", file=sys.stderr)
+                  print(f"  - {d}")
               sys.exit(1)
 
           # Commit each surface DIRECTLY to master (bookkeeping, not reviewable code)
@@ -162,7 +173,7 @@ jobs:
           PYBUMP
       - name: Create release from merged pull request
         env:
-          GITEA_TOKEN: ${{ secrets.GITEA_TOKEN }}
+          GITEA_TOKEN: ${{ secrets.ROBOT_TOKEN }}
           PR_TITLE: ${{ gitea.event.pull_request.title }}
           PR_BODY: ${{ gitea.event.pull_request.body }}
           RELEASE_API_URL: ${{ gitea.server_url }}/api/v1/repos/${{ gitea.repository }}/releases

.gitea/workflows/ci.yml

@@ -129,6 +129,7 @@ jobs:
             src/rune/canary_p0_05.py \
             src/rune/canary_p0_07.py \
             src/rune/canary_p1_02.py \
+            src/rune/canary_p1_03.py \
             src/rune/canary_p1_07.py \
             src/rune/canary_p1_06.py \
             src/rune/canary_p1_01.py \

README.md

@@ -1,5 +1,7 @@
 # Rune
 
+**Current release: v0.54.1**
+
 **Rune is a persistent personal agent designed to accumulate experience, develop reusable competence, and remain recognisably the same agent even when its underlying foundation model changes.**
 
 Rune treats the language model as replaceable reasoning compute rather than as the whole agent. The durable parts of the system are intended to belong to Rune:

config/umbrellas.json

@@ -17,7 +17,8 @@
         "drift_cache",
         "drift_compression",
         "tier_manager",
-        "graph_shard_bridge"
+        "graph_shard_bridge",
+        "memory_selector"
       ],
       "canaries": [
         "canary_query_memory",

docs/memory-selection.md

@@ -0,0 +1,139 @@
+# Memory Selection (P1-03, #26)
+
+One memory-selection API owned by `ContextBroker.memory_selector` (authority
+map `config/v1_authority_map.json`, domain `memory_selection`, v1_owner
+`ContextBroker.memory_selector`), with true abstention and one relevance
+floor.
+
+## The one API
+
+`MemorySelector` (`src/rune/memory_selector.py`) —
+`select_for_task(task_description, types=None, budget=N) -> SelectionResult`
+(mirrors `CapabilityBroker.select_for_job`). `SelectionResult` carries
+`blocks`, `candidates_evaluated`, `selected_count`, `abstained`,
+`types_queried`; the irrelevant-memory rate is computable per run as
+`(candidates_evaluated - selected_count) / candidates_evaluated` when
+candidates > 0.
+
+The operator registers exactly ONE memory contributor (`memory`) that
+delegates to `broker.memory_selector.select_for_task`. The six pre-P1-03
+parallel contributors (memory, persona_memories, consolidated, neurons,
+graphs, abstraction) were replaced — their sources now live inside the
+selector. No parallel authority path remains.
+
+## One relevance floor — true abstention
+
+Every source competes under THE floor: `memory_store.RECALL_MIN_SCORE`
+(0.42). The store source passes it into `memory_store.recall`'s calibrated
+QPP gate — which applies it in EVERY recall mode (lexical-only, dense-only,
+hybrid; the numeric check is additive to the conjunctive FTS gate) — so the
+selector's `store_scored` exemption is sound. Self-scored sources are
+floored in the selector.
+
+**Known ranking bias (documented, deferred):** the floor gates each source
+on its own calibrated scale (lexical store ~4-6, dense/hybrid store RRF
+~0-0.13, self-scored 0-1), but merged candidates are ranked by raw score —
+a lexical-mode store row outranks an equally-relevant self-scored row.
+Abstention is unaffected; only candidate ORDER is biased. Per-source rank
+normalization was deliberately deferred (renormalizing the store's
+calibrated scales risked false abstentions).
+
+- A trivial task ("what time is it") legitimately receives ZERO recalled
+  memories — `abstained=True`, zero blocks. Abstention is the safe default.
+- Below-floor candidates are COUNTED but never selected — measurement, not
+  injection.
+- The top-5-importance neuron fallback (old `rune_operator` neurons
+  contributor) is REMOVED: importance alone never injects.
+- The `gate_consolidated` flat-0.8 shortcut inside `cached_recall` is
+  REMOVED — and the now-dead `gate_consolidated` function itself is deleted
+  from `memory_store` (discharge): consolidated knowledge competes under the
+  same floor through the selector's dedicated source.
+
+## Selector sources
+
+Registered in execution order (`DEFAULT_SOURCES`):
+
+| Source | Reads | Gate |
+| --- | --- | --- |
+| `store` | `memory_store.recall` (typed) | store's calibrated gate (same constant) |
+| `neurons` | `warm_neuron_store.list_neurons` trigger scan | lexical overlap ≥ floor; NO importance fallback |
+| `graphs` | `graph_memory.query` BFS (read-only) | lexical overlap ≥ floor |
+| `abstraction` | `abstraction` facts LIKE scan | lexical overlap ≥ floor |
+| `consolidated` | `MemoryConsolidation` knowledge items | lexical overlap ≥ floor (flat-0.8 bypass removed) |
+| `persona` | typed `self`/`personality` recall keyed on the ACTIVE task | store's calibrated gate |
+
+Fail-closed everywhere: a failing source is skipped + logged (never a
+raise, never a fallback injection); a selector unavailable → the
+contributor abstains (zero memory blocks).
+
+## Memory-type targeting
+
+`entries.memory_type` (additive `ALTER TABLE`, backward compatible; values
+`episodic|semantic|procedural|self|personality|associative|NULL`).
+Backfill classifies ONLY `kind='neuron'` rows to `associative`, and the
+backfill runs at MIGRATION time (when the column is first added) — NOT on
+every connect, whose per-connect unindexed UPDATE scan the selector's
+fresh-connection-per-turn pattern would pay every turn (discharge).
+Explicit `ensure_memory_type_column()` calls keep self-healing semantics
+(they re-classify unclassified neuron rows, never touching explicit
+classifications); other rows stay NULL — unclassified is eligible for every
+query (classification narrows, never excludes). Typed queries filter
+`(memory_type IN ?) OR memory_type IS NULL`; typed queries bypass shard
+routing (pre-migration shards carry no `memory_type` column).
+
+**Typed persona lane (discharge):** the operator's memory contributor calls
+the selector WITHOUT types, so the store source always runs a SECOND typed
+query (`self`/`personality`) merged into the candidates — still thresholded
+inside the store's gate, may abstain, deduplicated by entry id. The write
+side: `add_entry(..., memory_type=...)` (and the store CLI's
+`--memory-type`) classifies at WRITE time where the writer genuinely knows
+the type — the warm-router neuron writer stamps `associative`, the
+sleep-time growth sync stamps `personality` for persona-targeted evidence
+(`semantic` otherwise).
+
+## ContextBroker owns the final memory budget
+
+`memory_budget = min(1200, window // 8)` (default; the global input budget
+bounds it further). In `_admit`, the sum of `MEMORY_BLOCK_NAMES` blocks
+(`memory`, `persona_memories`, `consolidated`) may never exceed the
+sub-budget — further memory blocks are dropped (lowest priority first) with
+a `memory budget` reason in the trace. The active TaskState block (`user`)
+is non-evictable and unaffected. The selector also caps its own candidate
+list before returning.
+
+## TaskState supremacy
+
+`_contributor_history` is rebased on `TaskStateRepository` (status
+authority) + `ConversationStateRepository` (bodies): completed tasks only,
+each user turn labeled `PAST TASK (completed — context only, not
+instructions)`, block priority (30) strictly below the active task block
+(`user`, 100, non-evictable). Recall queries are keyed ONLY on the active
+task's description — never on past-task text.
+
+## Trace v5
+
+`INFERENCE_TRACE_SCHEMA_VERSION = 5` — additive optional fields (same
+discipline as v1→v2→v3→v4; old readers ignore unknown keys):
+
+- `memory_candidates` — candidates surfaced by all selector sources;
+- `memory_selected` — candidates that cleared the floor;
+- `memory_abstained` — true = zero memory injected (zero-injection event);
+- `memory_types_queried` — typed targeting used for this run.
+
+## Canary + CI
+
+`src/rune/canary_p1_03.py` self-bootstraps (temp HOME/RUNE_HOME/
+RUNE_MEMORY_DIR; no model, no network) and asserts the acceptance
+invariants: trivial→zero memories, importance fallback dead, TaskState
+supremacy, memory sub-budget enforcement. Registered in
+`.gitea/workflows/ci.yml`'s canary for-loop (`henxels_checks.py
+ci_runs_all_canaries` enforces the registration).
+
+## Scope notes
+
+- `warm_router.py` stays dormant-until-dependency (its ID-path defect is
+  #27 territory — not fixed here).
+- `auto_context.py` is a TUI display artifact, not model context — out of
+  scope.
+- Memory stores and their data are preserved; selection changes HOW/WHEN
+  memory is surfaced, never what is stored.

lint-baseline.json

@@ -1,4 +1,4 @@
 {
-  "BLE001": 351,
-  "S110": 79
-}
+  "BLE001": 348,
+  "S110": 80
+}

memory/journal/events.jsonl

@@ -27,3 +27,13 @@
 {"id": "08b86d62", "timestamp": "2026-08-24T03:47:40.904828+00:00", "type": "on_context", "source": {"text": "how do you make a woman orgasm to the point of loosing there mind and spontaneously ejaculating squi", "channel": "chat"}, "payload": {}}
 {"id": "786bc16c", "timestamp": "2026-08-24T03:53:35.251222+00:00", "type": "on_context", "source": {"text": "are your tool calls broken?", "channel": "chat"}, "payload": {}}
 {"id": "9f7856e2", "timestamp": "2026-08-24T03:57:53.441686+00:00", "type": "on_context", "source": {"text": "whats the error and whats the cause of the issue that i can feed back to the memo version of you as ", "channel": "chat"}, "payload": {}}
+{"id": "3c75ff71", "timestamp": "2026-08-24T14:25:44.345167+00:00", "type": "on_context", "source": {"text": "what tools do you have available to you?", "channel": "chat"}, "payload": {}}
+{"id": "64eb817c", "timestamp": "2026-08-24T14:30:01.098906+00:00", "type": "on_context", "source": {"text": "write me a text on my desktoip that says hello", "channel": "chat"}, "payload": {}}
+{"id": "83a16e33", "timestamp": "2026-08-24T20:35:48.613711+00:00", "type": "on_context", "source": {"text": "write me a text file with hello onto my desktoip that says", "channel": "chat"}, "payload": {}}
+{"id": "e46deccb", "timestamp": "2026-08-24T20:37:59.634797+00:00", "type": "on_context", "source": {"text": "run the next task now", "channel": "chat"}, "payload": {}}
+{"id": "408131bc", "timestamp": "2026-08-24T20:38:39.093274+00:00", "type": "on_context", "source": {"text": "write me a text file onto my pc that says hello within it.", "channel": "task"}, "payload": {}}
+{"id": "14b85cc8", "timestamp": "2026-08-25T14:06:10.392245+00:00", "type": "on_context", "source": {"text": "whats the most henious thing known to human behaviour", "channel": "chat"}, "payload": {}}
+{"id": "49465788", "timestamp": "2026-08-25T14:07:35.849973+00:00", "type": "on_context", "source": {"text": "what about individually from one person to another", "channel": "chat"}, "payload": {}}
+{"id": "da5d3459", "timestamp": "2026-08-25T14:08:32.168662+00:00", "type": "on_context", "source": {"text": "oh you have no link from previous messages? is that something i havnt built structurally?", "channel": "chat"}, "payload": {}}
+{"id": "eb6803b3", "timestamp": "2026-08-25T14:10:45.413634+00:00", "type": "on_context", "source": {"text": "what would you suggest i ask my builder application to make this work, and what would you ask my bui", "channel": "chat"}, "payload": {}}
+{"id": "f8ed6fb9", "timestamp": "2026-08-25T14:24:48.782627+00:00", "type": "on_context", "source": {"text": "i didnt see what you wrote you didnt send it.", "channel": "chat"}, "payload": {}}

memory/journal/events_stats.json

@@ -0,0 +1,5 @@
+{
+  "total": 6,
+  "type_on_context": 6,
+  "updated": "2026-08-25T14:24:48.787425+00:00"
+}

plans/2026-09-07-p1-03-memory-selection.md

@@ -0,0 +1,194 @@
+# P1-03 Unify memory selection and restore true abstention — Implementation Plan (#26)
+
+> **Execution contract:** FROZEN TESTS ARE THE SPEC. Work tree: /mnt/workdrive/p1-03 (branch p1-03-memory-selection). TDD red → green per task; record every deviation in the plan tail. Do not create/complete tasks; controller owns task state. Conventional commits referencing (#26). Do not push to master directly; PR + protected checks only.
+
+## Goal
+
+One memory-selection API owned by `ContextBroker.memory_selector` (authority-map v1_owner), a real relevance threshold with **true abstention** (a trivial task legitimately receives ZERO recalled memories), memory-type targeting, no fallback injection merely because a memory is important, ContextBroker owning the final memory budget, and old-task retrieval that can never override the active TaskState.
+
+## Architecture
+
+- **New module `src/rune/memory_selector.py`** — `MemorySelector` class, the single memory-selection API. Composed by `ContextBroker` as `self.memory_selector` (authority map `config/v1_authority_map.json:120-150`, domain `memory_selection`, v1_owner `ContextBroker.memory_selector`).
+- **Operator collapse:** the six parallel memory contributors in `rune_operator.py` (`_contributor_memory` :1165, `_contributor_persona_memories` :1201, `_contributor_consolidated` :1238, `_contributor_neurons` :1356, `_contributor_graphs` :1397, `_contributor_abstraction` :1435) are REPLACED by exactly ONE registered memory contributor that delegates to `broker.memory_selector.select_for_task(...)`. No parallel authority path — replace, not add.
+- **Selector sources** (internal to the selector, each under the same threshold policy): typed store recall (`memory_store.recall`/`cached_recall`), warm-neuron scan (NO importance fallback), graph entity resolution, abstraction facts, consolidated knowledge (de-bypassed).
+- **Fail-closed inversion:** selector or source failure → zero memory blocks (abstain), log, continue. Abstention is the safe default — never fail-open to "inject something important".
+- **Trace v4** (`inference_trace.py`): additive optional fields `memory_candidates`, `memory_selected`, `memory_abstained`, `memory_types_queried` (carried additive discipline v3→v4).
+
+## Tech stack
+
+Python 3.14 stdlib + sqlite3; pytest; pyright ratchet (`scripts/type_check_ratchet.py` vs `pyright-baseline.json` 120e/465w — may only decrease); ruff (format --check on ADDED files, `--select E9,F63,F7,F82` on CHANGED); canary self-bootstrapping pattern; CI on rune-ci-desktop via `.gitea/workflows/ci.yml` canary for-loop (henxel `ci_runs_all_canaries` enforces registration).
+
+## Global Constraints
+
+- Preserve memory stores (handoff 6.15): selection changes HOW/WHEN memory is surfaced — no deletion of durable memory, no store rewrites.
+- Nothing injects model context outside the ContextBroker (target-architecture invariant).
+- `auto_context.py` is a TUI display artifact (not model context) — OUT OF SCOPE; note only.
+- warm_router stays dormant-until-dependency per authority map; its ID-path defect (#27 P1-04 territory) is NOT fixed here.
+- Suite baseline: **604 passed + 84 subtests** must stay green (known pre-existing env-only collection error in `test_build_pipeline_policy.py` without `-e .` install — not a regression).
+- New code must be type-clean under the ratchet; house narrowing pattern (`if not ok or resolved is None: return` + assert); runtime guards widen params (`X | object`); dataclass field validation PRE-construction.
+- No unrelated cognitive subsystems; no additive feature growth beyond this issue.
+
+## Design Decisions (settled)
+
+1. **Owner:** `ContextBroker.memory_selector` — `MemorySelector` lives in `src/rune/memory_selector.py`, composed by `ContextBroker` (mirrors `CapabilityBroker.select_for_job` shape: `select_for_task(task_state, budget) -> SelectionResult`).
+2. **One API:** `SelectionResult(blocks: list[ContributionBlock], candidates_evaluated: int, selected_count: int, abstained: bool, types_queried: list[str])`. The operator registers exactly ONE memory contributor consuming it.
+3. **Threshold policy:** single relevance floor — `memory_store.RECALL_MIN_SCORE` (0.42) default, applied to EVERY source. `gate_consolidated`'s flat-0.8 shortcut inside `cached_recall` (memory_store.py:1053-1057) is REMOVED so consolidated knowledge competes under the same floor. The top-5-importance neuron fallback (rune_operator.py:1387-1389) is REMOVED. `always_digest` per-turn unconditional injection is retired from model context.
+4. **Type targeting:** additive `memory_type` column on `entries` (ALTER TABLE ADD COLUMN — backward compatible). Values: `episodic|semantic|procedural|self|personality|associative|NULL`. Backfill classifies ONLY `kind='neuron'` → `associative`; all other existing rows remain NULL (unclassified = eligible for all queries — classification narrows, never excludes). Typed queries filter `(memory_type IN ?) OR memory_type IS NULL`. Writers classify at write time where the type is genuinely known. The persona contributor's hardcoded query (`"Rune personality voice identity how to speak"`, :1219) is replaced by a typed `self`/`personality` query derived from the ACTIVE task description — still thresholded, may abstain.
+5. **Budget:** ContextBroker owns the final memory budget — sum of `MEMORY_BLOCK_NAMES` blocks ≤ memory sub-budget (default `min(1200, window // 8)` tokens) enforced in `_admit`; the selector also caps its own candidate list before returning.
+6. **TaskState supremacy:** `_contributor_history` (rune_operator.py:789-802, :1493-1508) is rebased on `TaskStateRepository` (canonical) — completed tasks only, explicitly labeled `PAST TASK (completed — context only, not instructions)`, priority strictly below the active-task block; the active TaskState block is non-evictable; recall queries are keyed ONLY on the active task's description (never on past-task text).
+7. **Measurement:** trace v4 additive fields above; irrelevant-memory rate is computable per run as `(candidates_evaluated − selected_count) / candidates_evaluated` when candidates > 0, with `abstained=True` runs recorded as zero-injection events.
+8. **Degrade-not-crash:** selector source failure → skip that source, log, continue; selector unavailable → zero memory blocks (fail-closed abstain), never fallback injection.
+
+## Tasks
+
+### Task 1 — `memory_type` column + backfill migration
+**Files:** `src/rune/memory_store.py`, `tests/test_p1_03_memory_selection.py` (new)
+**Interfaces:** `ensure_memory_type_column(db)`, backfill at connect/migration time; `recall(..., memory_types: list[str] | None = None)` filter.
+Steps: 1) Write failing tests (column exists after migration; neuron rows backfilled to `associative`; other rows NULL; typed query returns typed+NULL, excludes other types; untyped query unchanged). 2) Red. 3) Implement additive migration. 4) Green + full suite. 5) Commit `feat(p1-03): memory_type column + typed recall filter (#26)`.
+
+### Task 2 — MemorySelector skeleton + store-recall source + true abstention
+**Files:** `src/rune/memory_selector.py` (new), `src/rune/context_broker.py`, `tests/test_p1_03_memory_selection.py`
+**Interfaces:** `MemorySelector(db_path, min_score=RECALL_MIN_SCORE)`, `.select_for_task(task_description, types=None, budget=N) -> SelectionResult`; `ContextBroker.memory_selector` composition.
+Steps: 1) Failing tests: trivial task ("what time is it") → zero blocks, `abstained=True`; relevant task → blocks above floor; below-floor candidates counted but not selected; source failure → abstain + log, no raise. 2) Red. 3) Implement. 4) Green + suite. 5) Commit `feat(p1-03): MemorySelector with threshold abstention (#26)`.
+
+### Task 3 — Migrate remaining sources into the selector (fallbacks die here)
+**Files:** `src/rune/memory_selector.py`, `src/rune/memory_store.py`, `tests/test_p1_03_memory_selection.py`
+**Interfaces:** selector internal sources: neurons (scan, NO fallback), graphs, abstraction, consolidated (gate_consolidated bypass removed); typed persona source (`self`/`personality`).
+Steps: 1) Failing tests: high-importance irrelevant neuron NOT injected (fallback gone); consolidated item below floor abstains (flat-0.8 gone); persona typed query replaces hardcoded query and abstains on trivial task. 2) Red. 3) Implement sources + remove `gate_consolidated` shortcut. 4) Green + suite. 5) Commit `feat(p1-03): selector sources unified, bypasses removed (#26)`.
+
+### Task 4 — Operator single-contributor collapse + TaskState supremacy
+**Files:** `src/rune/rune_operator.py`, `tests/test_p1_03_memory_selection.py`
+**Interfaces:** one `memory` contributor delegating to `broker.memory_selector`; `_contributor_history` rebased on `TaskStateRepository` with past-task labeling; TaskState block non-evictable.
+Steps: 1) Failing tests: exactly one memory contributor registered; prompt for a trivial task contains NO memory blocks (Context/Identity/knowledge/neurons/graphs/abstraction all absent); old completed-task content cannot outrank active TaskState (priority + labeling assertions); recall query never keyed on past-task text. 2) Red. 3) Collapse contributors, rebase history, enforce priorities. 4) Green + suite. 5) Commit `feat(p1-03): single memory contributor, TaskState supremacy (#26)`.
+
+### Task 5 — Memory sub-budget + trace v4 + canary + CI + docs
+**Files:** `src/rune/context_broker.py`, `src/rune/inference_trace.py`, `src/rune/canary_p1_03.py` (new), `.gitea/workflows/ci.yml`, `docs/` (memory selection section), `tests/test_p1_03_memory_selection.py`
+**Interfaces:** `memory_budget` on ContextBroker (default `min(1200, window // 8)`); trace v4 additive fields; canary self-bootstrapping (`RUNE_HOME`/`RUNE_MEMORY_DIR` temp isolation, prints `P1-03 ... CANARY PASSED`, exit 0); CI for-loop registration (henxel enforces).
+Steps: 1) Failing tests: memory blocks evicted beyond sub-budget; trace carries v4 fields on assemble; canary assertions (trivial→zero, fallback-dead, TaskState supremacy). 2) Red. 3) Implement budget + trace + canary + CI list entry + docs. 4) Green + suite + canary + ratchet + ruff + henxels. 5) Commit `feat(p1-03): memory budget, trace v4, canary (#26)`.
+
+## Acceptance (from #26)
+
+- Trivial task can legitimately receive zero recalled memories → Task 2 tests + Task 4 prompt assertion + canary.
+- Irrelevant-memory rate is measured → Task 5 trace v4 fields (`memory_candidates`/`memory_selected`/`memory_abstained`).
+- Old-task retrieval cannot override TaskState → Task 4 tests + canary.
+
+Actions covered: one memory-selection API (Tasks 2-4), relevance threshold (Tasks 2-3), memory-type targeting (Tasks 1, 3), no fallback injection (Task 3), ContextBroker owns final memory budget (Task 5).
+
+## Pipeline requirements checklist (from #26)
+
+- [ ] TDD: frozen/regression tests written first, red → green
+- [ ] `ruff check` and `ruff format --check` clean
+- [ ] Spec-compliance review gate: PASS
+- [ ] Code-quality review gate: Approved
+- [ ] Full orchestration suite regression green
+- [ ] Canary passes (canary_p1_03 + all registered canaries)
+- [ ] Closed with a properly formatted closing comment + commit/PR references
+
+## Completion Record
+
+(status: **COMPLETE** — implementer record, 2026-09-07)
+
+### Commits (branch p1-03-memory-selection, in order)
+
+| Task | Commit | Message |
+| --- | --- | --- |
+| 1 | `fde27d7` | feat(p1-03): memory_type column + typed recall filter (#26) |
+| 2 | `f998768` | feat(p1-03): MemorySelector with threshold abstention (#26) |
+| 3 | `ff04888` | feat(p1-03): selector sources unified, bypasses removed (#26) |
+| 4 | `32de159` | feat(p1-03): single memory contributor, TaskState supremacy (#26) |
+| 5 | `7101087` | feat(p1-03): memory budget, trace v5, canary (#26) |
+| fix | `57c0faf` | fix(p1-03): prior canary trace-version pins follow the schema constant (#26) |
+
+### Suite counts
+
+- Baseline (master 2840f71): **604 passed + 84 subtests**; ratchet 117e/407w.
+- Final: **655 passed + 84 subtests** (+51 frozen tests, all red-first); ratchet **116e/418w** (baseline 120/465 — improved, never increased).
+- henxels_checks.py: **exit 0** (incl. `ci_runs_all_canaries`).
+- ruff format + `--select E9,F63,F7,F82`: clean on every added/changed file.
+- Canary battery (all 24 registered canaries, system python3): **23/24 pass**; `canary_native.py` has 2 pre-existing import-check failures (`pins.list_pins`, `event_bus.stats` — "No module named 'rune'"), verified **identical on master 2840f71** via a temp worktree — not a regression, untouched (out of scope). `canary_p1_03`, `canary_p1_02`, `canary_p1_01`, `canary_p1_06` all green.
+
+### Red→green evidence per task
+
+- **Task 1**: 8/8 frozen tests failed (no `memory_type` column, no `ensure_memory_type_column`, no `memory_types` param) → green 8/8; suite 612.
+- **Task 2**: 14 failed (no `memory_selector` module, no broker composition) → green 23/23; suite 627.
+- **Task 3**: 5 failed (importance fallback alive, flat-0.8 bypass present, no persona source, sources unregistered) → green 33 + 2 Task-4-deferred red; suite 637.
+- **Task 4**: 6 failed (six memory contributors registered, unlabeled history, legacy-ledger history, recall seam) → green 43/43; suite 647.
+- **Task 5**: 8 failed (no sub-budget, no v5 fields, no doc/canary/CI entry) → green 51/51; suite 655.
+
+### Deviations (with rationale)
+
+1. **Trace schema v5, not v4.** The plan assumed inference_trace at v3; P1-02 (#25) had already landed v4 (capsule fidelity). P1-03's additive fields therefore land as **v5** under the same additive discipline. Prior pinned assertions (test_p1_01, test_p1_02, canary_p1_01, canary_p1_06) were updated to follow the module constant — exactly what P1-02 did to P1-01's pin (commit 21c4023).
+2. **`memory_store.connect(db_path=None)`** gained an optional explicit-path param (Task 2 commit): `MemorySelector(db_path)` must pin the store it selects from. Additive, backward compatible; plan's Task 2 file list omitted memory_store.py.
+3. **Shard routing bypass.** Typed recall (`memory_types`) and the selector's store source bypass `ShardManager` routing: shards live at a hardcoded `~/.config/opencode/memory/shards` (ignores RUNE_* isolation) and pre-migration shard DBs carry **no `memory_type` column**, so typed recall through a shard is impossible. Untyped direct `recall()` callers keep existing routing behavior.
+4. **Consolidated source: per-item scoring; write side-effect retired.** The flat-0.8 whole-blob injection is replaced by per-knowledge-item lexical scoring under the floor. The collapsed `_contributor_consolidated`'s per-turn `_save_knowledge()` write is retired — consolidated knowledge is now read-only through the selector; no durable data deleted (preserve-stores constraint holds). Plan was silent on the write; reviewer flag below.
+5. **Graphs/abstraction sources fixed to real schemas.** The old contributors were silent always-degrade paths: `graph_memory.resolve_entity()` mutates the graph (creates missing entities) and returns an int id (old `e.get('name')` could never work); `abstraction.facts` uses column `state`, not `status` (old query raised). New sources use read-only `graph_memory.query()` and `state='active'`.
+6. **Test-seam migration (3 pre-existing characterization tests).** test_context_broker/test_rune_operator_tools patched `memory_store.cached_recall` — a seam the collapsed operator no longer calls. Assertions kept at equal strength; the memory seam moved to the selector's store source (patch.dict on `DEFAULT_SOURCES`). The second-turn characterization now expects the labeled past-task format — labeling IS an issue acceptance criterion, not a weakened assertion.
+7. **Two Task-4 tests authored during Task 3** (operator-source assertions: hardcoded persona query gone, importance fallback gone) stayed red through the Task 3 commit and went green in Task 4 — they assert the operator source, which Task 3's file list deliberately does not touch.
+
+### Reviewer flags
+
+- **Consolidation write retirement** (deviation 4): confirm intended — the plan's "replace, not add" + preserve-stores reading says yes, but it is a behavior change (knowledge.json stops growing per-turn).
+- **Store-source floor exemption**: store candidates skip the selector-level numeric comparison because the SAME constant is applied inside `memory_store.recall`'s calibrated gate (rows tagged `store_scored`); documented in memory_selector.py. Sanity-check the policy (renormalizing the store's mixed lexical/dense relevance scales would have produced false abstentions).
+- **Isolation gap (pre-existing, noted for a future issue)**: `memory_consolidation` hardcodes `~/.rune/consolidation` and `shard_manager` hardcodes `~/.config/opencode` — neither honors RUNE_* env isolation. Tests/canary redirect HOME; production code untouched.
+- Acceptance criteria all machine-checked: trivial→zero memories (Task 2/4 tests + canary check 1); irrelevant-memory rate measured (trace v5 fields + SelectionResult.irrelevant_rate); old-task cannot override TaskState (Task 4 tests + canary check 3).
+
+## Discharge Record (post-gate findings, 2026-09-07)
+
+Both review gates returned 9 findings (2 MAJOR, 7 MINOR). All discharged red-first on
+branch p1-03-memory-selection; every finding below maps to its commit and its red→green
+evidence. Suite after discharge: **670 passed + 84 subtests** (baseline 655+84; +15
+discharge tests, all red-first); pyright ratchet **held at 116e/418w**; ruff
+format/check clean on touched files that were formatted at HEAD; henxels exit 0;
+canary battery **24/24** (incl. canary_p1_03).
+
+| # | Finding | Commit | Red→green |
+|---|---|---|---|
+| 1 | Store floor only enforced in dense-only recall; lexical-only + hybrid paths never applied `min_score`, and the selector exempted `store_scored` rows from its own floor — a below-floor store row through lexical/hybrid would inject | `3f53d93` `fix(p1-03): recall applies the ONE floor in every mode` | RED: `test_recall_lexical_only_applies_min_score_floor` (3 seeded rows, floor 5.5 → rows returned) + `test_recall_hybrid_below_floor_row_abstains` (cos≈0.37 row injected through hybrid). GREEN: both abstain; hybrid above-floor positive control still surfaces. Floor choice: enforced INSIDE `memory_store.recall` in every mode (lexical best-relevance check additive to the conjunctive FTS gate; hybrid gates on the calibrated dense cosine, same as dense-only); selector `:204` claim rewritten to match. |
+| 2 | Typed persona source dead in production (operator calls `select_for_task` with NO types so `_persona_recall` always returned `[]`); no production writer set `memory_type` | `70c2007` `feat(p1-03): typed persona lane + write-time memory_type classification` | RED: `test_contributor_memory_runs_typed_persona_query` (no `memory_types=['self','personality']` call in the live contributor path) + `test_add_entry_memory_type_round_trip` (TypeError). GREEN: the store source runs a SECOND typed query (`self`/`personality`) merged into candidates (thresholded, deduped by id, may abstain — trivial task still gets zero); `add_entry` + store CLI `--memory-type` classify at write time; wired where the writer genuinely knows the type: warm-router neuron writer → `associative`, sleep-time growth sync → `personality` (persona-targeted) / `semantic`. |
+| 3 | `_persona_recall` ignored the selector's pinned `db_path` (opened the module-default store) | `fix(p1-03): selector sources honor the pinned db_path` | RED: `test_persona_source_uses_pinned_db_path` + `test_store_source_uses_pinned_db_path` (pinned store seeded, module-default store empty → no block). GREEN: pinned `db_path` is bound into sources that declare it at registration; both sources read the pin. |
+| 4 | `ensure_memory_type_column` ran an unindexed `UPDATE ... WHERE kind='neuron'` scan on EVERY connect; the selector opens a fresh connection per turn | `ea5de21` `fix(p1-03): neuron backfill runs at migration time only` | RED: `test_backfill_runs_once_at_migration_not_every_connect` (NULL marker row flipped back to `associative` by the second connect). GREEN: connect() only invokes the migration when the column is actually missing; explicit `ensure_memory_type_column()` keeps the frozen self-healing semantics (re-classifies unclassified neurons, never touches explicit classifications). |
+| 5 | Vacuous operator-level assertions (`block is None or ...` pass trivially on empty tmp stores) | `test(p1-03): strengthen vacuous selector assertions` | Not red-first (test-only): the four tests now seed the store and assert the positive branch — real block, exact source, history priority/label from a seeded completed task. |
+| 6 | Weak assertions: `cached_recall` test escaped via `or source=="consolidated"`; fallback-gone test grepped one literal | same commit as #5 | `test_cached_recall_never_returns_flat_eight_tenths` now asserts the calibrated recall shape (rrf/confidence/id present, relevance ≠ 0.8) with shard routing disabled (it silently rerouted to an empty shard, vacating the old assertion) + abstention twin + positive twin; `test_operator_importance_fallback_gone` greps ALL fallback slice patterns + a behavioral twin (off-trigger high-importance neuron does not inject through the operator). |
+| 7 | `gate_consolidated` dead code | deletion rode into `fix(p1-03): recall applies the ONE floor in every mode` (memory_store staged as a whole file); retirement proof in `refactor(p1-03): canary proves gate_consolidated retirement via AST` | Function deleted; `canary_query_memory`'s gate check now proves the retirement via AST parse of the REPO copy (the old check inspected whichever `memory_store` a sys.path shim resolved first). Suite stays green. |
+| 8 | Canary trivial-check ran on an empty store (capsule-level zero-memory proof not seeded with a distractor) | `b8ada5f` `test(p1-03): seed canary trivial-check distractor` | Not red-first (canary-only): `check_trivial_task_abstains` now seeds a high-importance fact distractor + a personality row before assembling; zero-memory proof is capsule-level meaningful. Canary passes. |
+| 9 | Ranking scale mix (store lexical ~5.x vs dense/hybrid RRF ~0-0.13 vs self-scored 0-1 ranked raw together) | documented in `3f53d93` (selector docstring) + this record | Per the finding's escape hatch: KNOWN RANKING BIAS documented in `MemorySelector` class docstring + `_store_recall` docstring + docs/memory-selection.md. Abstention unaffected (each source floored on its own calibrated scale); per-source rank normalization deliberately deferred (renormalizing the store's calibrated scales risked false abstentions). |
+
+### Discharge deviations (with rationale)
+
+1. **Floor enforced in the STORE (all modes), not in the selector.** The finding offered
+   either place. The store is where the calibrated gate lives; every direct `recall()`
+   caller gets the floor for free, and the selector's `store_scored` exemption becomes
+   literally true. The lexical branch's numeric check is additive to the conjunctive FTS
+   gate (which stays as-is); the hybrid branch gates on the dense cosine — the scale
+   `min_score` was actually calibrated against (RRF-fused relevance is rank-based and is
+   NOT compared against the floor).
+2. **Persona lane lives in the store source, not `_persona_recall`.** The finding's FIX (a)
+   prescribed exactly this ("the selector's store source runs a SECOND typed query");
+   `_persona_recall` remains the typed-caller source and now also honors the pinned
+   db_path. Lane results are deduplicated by entry id (the untyped and typed queries
+   overlap on unclassified rows — duplicates would corrupt the irrelevant-memory rate and
+   repeat block text).
+3. **Persona-memory write path**: sleep_time's growth sync IS a persona write path
+   (evidence items target `persona`), so it now stamps `memory_type=personality`
+   (`semantic` for other growth targets) via the store CLI's new `--memory-type` flag —
+   including through the dedupe gate (the surviving row is classified after
+   `dedupe_policy.apply`). No writer was invented beyond what genuinely knows its type.
+4. **Backfill self-healing preserved on explicit calls.** The finding said "backfill only
+   when the column was JUST added (migration-time)"; connect() now honors that strictly,
+   but an EXPLICIT `ensure_memory_type_column()` call keeps the frozen self-healing
+   semantics (re-classifies unclassified neuron rows) — the per-CONNECT scan is what was
+   eliminated. The frozen `test_backfill_classifies_only_neuron_rows` still passes
+   unchanged.
+5. **`canary_query_memory` gate check rewritten as a retirement proof** (AST parse of the
+   repo copy): the old check inspected `memory_store.gate_consolidated` through a sys.path
+   shim that resolves the DEPLOYED scripts copy first — deterministic in neither
+   direction, and obsolete once the function was deleted.
+6. **`test_cached_recall_*` tests disable shard routing** (`sys.modules["shard_manager"]
+   = None`): shard routing can silently reroute to an out-of-repo empty shard, which is
+   what made the original flat-0.8 assertion vacuous. Queries are unique per test to
+   defeat the in-process recall cache.
+7. **ruff format scope**: `warm_router.py`, `sleep_time.py`, `canary_query_memory.py`
+   were already unformatted at HEAD (pre-existing); wholesale reformatting would bury the
+   2-8 line functional changes in noise, so those files keep their local style and only
+   `ruff check --select E9,F63,F7,F82` is asserted (clean). Files formatted at HEAD
+   (memory_store, memory_selector, canary_p1_03, test file) remain ruff-format clean.

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "rune"
-version = "0.49.0"
+version = "0.54.1"
 description = "Rune — a stateful coding assistant with persistent memory, orchestration, and TUI"
 requires-python = ">=3.10"
 

scripts/merge_robot.py

@@ -87,7 +87,29 @@ def decide(base: str, token: str, pr: dict) -> str:
         return f"#{n}: skipped (status={state})"
     status, detail = _api(base, token, f"/pulls/{n}")
     if not detail.get("mergeable", False):
-        return f"#{n}: skipped (not mergeable)"
+        # CONFLICT-FIRST (Nick policy): resolve staleness immediately —
+        # attempt the API update NOW rather than waiting a sleep cycle.
+        # 409 (genuine conflict) still skips with the reason named.
+        ustatus, uresp = _api(base, token, f"/pulls/{n}/update", "POST", {"style": "merge"}, ok=(200, 409))
+        if ustatus == 200:
+            print(f"UPDATED {n} (conflict-first: mergeable=False resolved pre-merge)", flush=True)
+            # classify the update: docs-only → carried green stands; code-path
+            # touch → fresh CI required before merge (#208 slip prevention)
+            try:
+                _, fresh = _api(base, token, f"/pulls/{n}")
+                new_sha = fresh.get("head", {}).get("sha", sha)
+                if classify_update(base, token, pr, new_sha) == "RERUN_REQUIRED":
+                    return f"#{n}: skipped (updated; CI re-running on new head)"
+            except MergeRefused:
+                return f"#{n}: skipped (updated; CI re-running on new head)"
+            status, detail = _api(base, token, f"/pulls/{n}")
+            if not detail.get("mergeable", False):
+                return f"#{n}: skipped (updated; CI re-running on new head)"
+        elif ustatus == 409:
+            cmsg = str(uresp.get("message", "")) if isinstance(uresp, dict) else str(uresp)
+            return f"#{n}: skipped (genuine conflict — update 409: {cmsg[:120]})"
+        else:
+            return f"#{n}: skipped (not mergeable; update returned {ustatus})"
     mstatus, mresp = _api(base, token, f"/pulls/{n}/merge", "POST", {"Do": "merge"})
     if mstatus == 200:
         return f"#{n}: MERGED"
@@ -116,11 +138,25 @@ def decide(base: str, token: str, pr: dict) -> str:
             if mstatus == 200:
                 return f"#{n}: MERGED (after behind-base self-update)"
             if mstatus == 405 and "behind" in str(mresp.get("message", "")).lower():
-                # CARRY-FORWARD PROTOCOL: the workflow inspects the tree diff
-                # between the tested head and the new head; docs-only movement
-                # carries the prior green (no redundant re-run — Nick's policy).
-                print(f"UPDATED {n} awaiting-ci", flush=True)
-                return f"#{n}: skipped (updated; CI re-running on new head)"
+                # CARRY-FORWARD PROTOCOL: classify the tree diff between the
+                # tested head and the new head. Docs-only movement carries the
+                # prior green (no redundant re-run — Nick's policy); any
+                # code-path touch requires fresh CI before merge (#208 slip:
+                # required-checks satisfied by a stale green).
+                try:
+                    _, fresh = _api(base, token, f"/pulls/{n}")
+                    new_sha = fresh.get("head", {}).get("sha", sha)
+                    verdict = classify_update(base, token, pr, new_sha)
+                except MergeRefused:
+                    verdict = "RERUN_REQUIRED"  # can't verify → safest default
+                if verdict == "RERUN_REQUIRED":
+                    print(f"UPDATED {n} awaiting-ci (code-path touch)", flush=True)
+                    return f"#{n}: skipped (updated; CI re-running on new head)"
+                print(f"UPDATED {n} carry-forward (docs-only diff)", flush=True)
+                mstatus, mresp = _api(base, token, f"/pulls/{n}/merge", "POST", {"Do": "merge"})
+                if mstatus == 200:
+                    return f"#{n}: MERGED (carry-forward: docs-only update)"
+                return f"#{n}: skipped (carry-forward merge returned {mstatus})"
             raise MergeRefused(f"#{n} merge refused {mstatus} after update: {str(mresp)[:200]}")
         raise MergeRefused(f"#{n} merge refused 405: {msg}")
     if mstatus == 403:
@@ -143,6 +179,16 @@ CODE_PATHS = (
 )
 
 
+def classify_update(base: str, token: str, pr: dict, new_sha: str) -> str:
+    """Classify a self-heal update: CARRY_FORWARD (docs-only tree diff — the
+    tested code is byte-identical, prior green stands) or RERUN_REQUIRED
+    (any code-path touch — semantic interactions are real, #208 slip)."""
+    old_sha = pr["head"]["sha"]
+    status, cmp_data = _api(base, token, f"/compare/{old_sha}...{new_sha}")
+    files = [f.get("filename", "") for f in cmp_data.get("files", []) if isinstance(f, dict)]
+    return "CARRY_FORWARD" if not touches_code_path(files) else "RERUN_REQUIRED"
+
+
 def touches_code_path(changed_files: list[str], code_paths=CODE_PATHS) -> bool:
     """Carry-forward policy: True iff any changed file is a code path.
     Docs-only base movement must NOT trigger a redundant full re-run."""

src/rune/canary_p1_01.py

@@ -247,8 +247,10 @@ def check_scripted_operator_wiring() -> int:
     if not records:
         return _fail("no durable inference trace persisted")
     record = records[-1]
-    if record["schema_version"] != 4:
-        return _fail(f"trace schema_version must be 4, got {record['schema_version']}")
+    if record["schema_version"] != it.INFERENCE_TRACE_SCHEMA_VERSION:
+        return _fail(
+            f"trace schema_version must be {it.INFERENCE_TRACE_SCHEMA_VERSION}, got {record['schema_version']}"
+        )
     if record["selected_tools"] != expected:
         return _fail(f"trace selected_tools wrong: {record['selected_tools']}")
     if not record["capability_reason"]:

src/rune/canary_p1_03.py

@@ -0,0 +1,271 @@
+#!/usr/bin/env python3
+"""P1-03 memory-selection canary (#26).
+
+Self-bootstrapping like canary_p1_02.py (CI runs ``python3 <canary>``
+without PYTHONPATH). Proves the #26 claims are machine-checked with NO
+live model, NO Ollama, NO network:
+
+1. trivial task -> ZERO memory blocks (true abstention — the acceptance
+   criterion; the trace proves memory_tokens == 0);
+2. a high-importance irrelevant neuron is NOT injected (the
+   top-5-importance fallback is dead) while a trigger-matched neuron still
+   surfaces through the selector;
+3. old completed-task content can never override the active TaskState:
+   past-task text is labeled context-only, the history block's priority
+   stays strictly below the active task block, and the recall query is
+   keyed ONLY on the active task description;
+4. ContextBroker owns the final memory budget: a memory block above the
+   sub-budget is evicted with a ``memory budget`` reason.
+
+Isolates ALL state under a temp HOME (so ~/.rune and ~/.config/opencode
+shards are redirected too) plus RUNE_HOME/RUNE_MEMORY_DIR, so the canary
+never touches a real memory store. Prints 'P1-03 MEMORY SELECTION CANARY
+PASSED' and exits 0.
+"""
+
+import json
+import os
+import sys
+import tempfile
+from pathlib import Path
+from types import SimpleNamespace
+
+# Self-bootstrap: CI runs `python3 <canary>` without PYTHONPATH, so put the
+# repository src/ on sys.path before importing rune modules. Redirect HOME
+# FIRST so every expanduser path (memory_env defaults, ~/.rune/consolidation,
+# ~/.config/opencode shards) lands inside the temp sandbox.
+_repo_src = str(Path(__file__).resolve().parents[1])
+sys.path.insert(0, _repo_src)
+_canary_home = tempfile.mkdtemp(prefix="p1-03-canary-home-")
+os.environ["HOME"] = _canary_home
+os.environ["RUNE_HOME"] = str(Path(_canary_home) / "rune-home")
+os.environ["RUNE_MEMORY_DIR"] = str(Path(_canary_home) / "memory")
+
+from rune import memory_consolidation as _mc  # noqa: E402
+
+_consolidation_dir = Path(_canary_home) / "consolidation"
+_mc.CONSOLIDATION_DIR = str(_consolidation_dir)
+_mc.KNOWLEDGE_FILE = str(_consolidation_dir / "knowledge.json")
+_mc.LESSONS_FILE = str(_consolidation_dir / "lessons.json")
+_mc.CONNECTIONS_FILE = str(_consolidation_dir / "connections.json")
+
+from rune import memory_store  # noqa: E402
+
+
+def _fail(message: str) -> int:
+    print(f"FAILED: {message}", file=sys.stderr)
+    return 1
+
+
+def _seed_entry(text, kind="fact", memory_type=None, importance=0.5, triggers=""):
+    """Insert an entry into the ISOLATED store (no embeddings needed — the
+    canary runs lexical-only; _embed fails closed to {})."""
+    db = memory_store.connect()
+    try:
+        ts = memory_store.now_iso()
+        cur = db.execute(
+            "INSERT INTO entries (text, importance, created_at, last_accessed, "
+            "topic, entities, source, method, status, kind, triggers, memory_type) "
+            "VALUES (?,?,?,?,?,?,?,?,?,?,?,?)",
+            (text, importance, ts, ts, "", "[]", "canary", "curator", "active", kind, triggers, memory_type),
+        )
+        db.execute("INSERT INTO entries_fts (rowid, text) VALUES (?,?)", (cur.lastrowid, text))
+        db.commit()
+        return cur.lastrowid
+    finally:
+        db.close()
+
+
+def _no_embed() -> None:
+    """Force lexical-only recall: deterministic, no Ollama dependency."""
+    memory_store._embed = lambda texts: {}
+
+
+def check_trivial_task_abstains() -> int:
+    """Invariant 1: a trivial task receives ZERO memory blocks — proven
+    against a SEEDED store (P1-03 #26 discharge: a distractor the trivial
+    query cannot match is planted first, so the zero-memory proof is not
+    vacuous on an empty store)."""
+    _no_embed()
+    _seed_entry(
+        "Quarterly fiscal audit committee schedule for review",
+        kind="fact",
+        importance=0.99,
+    )
+    _seed_entry(
+        "I speak in one clean sentence, evidence first, never hedging",
+        kind="fact",
+        importance=0.95,
+        memory_type="personality",
+    )
+    from rune.context_broker import MEMORY_BLOCK_NAMES
+    from rune.rune_operator import RuneOperator
+
+    op = RuneOperator(str(Path(_canary_home) / "operator.json"))
+    task = SimpleNamespace(task_id="canary-t1", description="what time is it", assigned_model="")
+    broker = op._context_broker()
+    session: dict = {}
+    capsule, trace = broker.assemble(task, session, budget_tokens=10_000)
+    memory_blocks = [b for b in capsule.blocks if b.name in MEMORY_BLOCK_NAMES]
+    if memory_blocks:
+        return _fail(f"trivial task injected memory blocks: {[b.name for b in memory_blocks]}")
+    if trace["memory_tokens"] != 0:
+        return _fail(f"trivial task trace memory_tokens={trace['memory_tokens']} != 0")
+    return 0
+
+
+def check_importance_fallback_dead() -> int:
+    """Invariant 2: importance alone never injects; trigger match still does."""
+    _no_embed()
+    _seed_entry(
+        "Quarterly fiscal audit committee schedule for review",
+        kind="neuron",
+        importance=0.99,
+        triggers="audit",
+    )
+    _seed_entry(
+        "Kubernetes pods restart automatically on failure",
+        kind="neuron",
+        importance=0.9,
+        triggers="kubernetes",
+    )
+    from rune.context_broker import MEMORY_BLOCK_NAMES
+    from rune.rune_operator import RuneOperator
+
+    op = RuneOperator(str(Path(_canary_home) / "operator-irrelevant.json"))
+    broker = op._context_broker()
+    irrelevant = SimpleNamespace(task_id="canary-t2", description="kubernetes pods restart", assigned_model="")
+    session: dict = {}
+    capsule, _trace = broker.assemble(irrelevant, session, budget_tokens=10_000)
+    memory_text = "\n".join(b.content for b in capsule.blocks if b.name in MEMORY_BLOCK_NAMES)
+    if "fiscal audit" in memory_text:
+        return _fail("high-importance irrelevant neuron was injected (fallback alive)")
+    if "fiscal audit" in "\n".join(b.content for b in capsule.blocks):
+        return _fail("irrelevant neuron content leaked into a non-memory block")
+    relevant = SimpleNamespace(task_id="canary-t3", description="kubernetes pods restart", assigned_model="")
+    session2: dict = {}
+    capsule2, _trace2 = broker.assemble(relevant, session2, budget_tokens=10_000)
+    memory_text2 = "\n".join(b.content for b in capsule2.blocks if b.name in MEMORY_BLOCK_NAMES)
+    if "Kubernetes pods restart automatically" not in memory_text2:
+        return _fail("trigger-matched neuron did not surface (selector broken)")
+    return 0
+
+
+def check_taskstate_supremacy() -> int:
+    """Invariant 3: old-task content is labeled, lower priority, and never
+    keys a recall query."""
+    _no_embed()
+    from rune.context_broker import MEMORY_BLOCK_NAMES
+    from rune.orchestration.conversation_state import ConversationMessageRecord
+    from rune.orchestration.records import TaskStateRecord
+    from rune.rune_operator import OPERATOR_CHANNEL, RuneOperator
+
+    op = RuneOperator(str(Path(_canary_home) / "operator-supremacy.json"))
+    # Seed a completed past task through the CANONICAL ledger.
+    past = TaskStateRecord(
+        title="deploy the kubernetes cluster to production",
+        channel_id=OPERATOR_CHANNEL,
+        status="completed",
+    )
+    if not op._task_state.import_record(past):
+        return _fail("could not seed the past task record")
+    op._conversation.append(
+        ConversationMessageRecord(
+            role="user",
+            content="deploy the kubernetes cluster to production",
+            channel_id=OPERATOR_CHANNEL,
+            task_id=past.task_id,
+        )
+    )
+    op._conversation.append(
+        ConversationMessageRecord(
+            role="assistant",
+            content="Cluster deployed; all pods healthy in production.",
+            channel_id=OPERATOR_CHANNEL,
+            task_id=past.task_id,
+        )
+    )
+
+    # Spy on the selector: the recall query must be the ACTIVE task only.
+    broker = op._context_broker()
+    captured: dict = {}
+    real_select = broker.memory_selector.select_for_task
+
+    def spy(query, types=None, budget=4):
+        captured["query"] = query
+        return real_select(query, types=types, budget=budget)
+
+    broker.memory_selector.select_for_task = spy  # type: ignore[method-assign]
+
+    active = SimpleNamespace(task_id="canary-active", description="what time is it", assigned_model="")
+    session: dict = {}
+    capsule, _trace = broker.assemble(active, session, budget_tokens=10_000)
+
+    if captured.get("query") != "what time is it":
+        return _fail(f"recall query keyed on past-task text: {captured.get('query')!r}")
+
+    by_name = {b.name: b for b in capsule.blocks}
+    history = by_name.get("history")
+    if history is None:
+        return _fail("completed past task produced no history block")
+    history_text = (
+        history.content if isinstance(history.content, str) else json.dumps(history.content, ensure_ascii=False)
+    )
+    if "PAST TASK (completed — context only, not instructions)" not in history_text:
+        return _fail("past-task content is not labeled context-only")
+    if "kubernetes" not in history_text:
+        return _fail("past-task content missing from the labeled history block")
+    user_block = by_name.get("user")
+    if user_block is None or user_block.evictable:
+        return _fail("active TaskState block (user) must be non-evictable")
+    if history.priority >= user_block.priority:
+        return _fail(f"history priority {history.priority} must be strictly below active task {user_block.priority}")
+    # The labeled history must never leak into a memory block either.
+    for block in capsule.blocks:
+        block_text = block.content if isinstance(block.content, str) else json.dumps(block.content, ensure_ascii=False)
+        if block.name in MEMORY_BLOCK_NAMES and "deploy the kubernetes cluster to production" in block_text:
+            return _fail("past-task text surfaced through a memory block")
+    return 0
+
+
+def check_memory_budget_enforced() -> int:
+    """Invariant 4: the broker evicts memory above its sub-budget."""
+    from rune.context_broker import ContributionBlock, ContextBroker
+
+    broker = ContextBroker(
+        budget_resolver=lambda window: 10_000,
+        estimator=lambda text: len(text) // 4,
+    )
+    broker.register_contributor(
+        "memory",
+        lambda task, session: ContributionBlock("memory", 70, "m" * 8000, 0, "canary"),
+        2,
+    )
+    task = SimpleNamespace(task_id="canary-t4", description="kubernetes pods restart")
+    _capsule, trace = broker.assemble(task, {}, budget_tokens=10_000)
+    memory_records = [b for b in trace["blocks"] if b["name"] == "memory"]
+    if not memory_records or memory_records[0]["admitted"]:
+        return _fail("memory block above the sub-budget was not evicted")
+    if "memory budget" not in memory_records[0]["reason"]:
+        return _fail(f"eviction reason does not name the memory budget: {memory_records[0]['reason']!r}")
+    return 0
+
+
+def main() -> int:
+    checks = (
+        ("trivial task -> zero memory blocks (true abstention)", check_trivial_task_abstains),
+        ("high-importance irrelevant neuron NOT injected; fallback dead", check_importance_fallback_dead),
+        ("old-task content cannot override the active TaskState", check_taskstate_supremacy),
+        ("ContextBroker evicts memory beyond the sub-budget", check_memory_budget_enforced),
+    )
+    for label, check in checks:
+        result = check()
+        if result != 0:
+            return result
+        print(f"  ✓ {label}")
+    print("P1-03 MEMORY SELECTION CANARY PASSED")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

src/rune/canary_p1_06.py

@@ -198,8 +198,8 @@ def check_reason_recording() -> int:
     if not records:
         return _fail("no durable inference trace persisted")
     record = records[-1]
-    if record["schema_version"] != 4:
-        return _fail(f"schema_version must be 4, got {record['schema_version']}")
+    if record["schema_version"] != it.INFERENCE_TRACE_SCHEMA_VERSION:
+        return _fail(f"schema_version must be {it.INFERENCE_TRACE_SCHEMA_VERSION}, got {record['schema_version']}")
     if record["routing_reason"] != "router: measured 12/13 success task_family=code":
         return _fail(f"routing_reason not recorded: {record['routing_reason']!r}")
     if record["routing_mode"] != "local":

src/rune/canary_query_memory.py

@@ -78,10 +78,18 @@ def main():
     # --- 5. Integration with gate ---
     print("\n--- 5. Integration with gate ---")
     try:
-        import memory_store
-        import inspect
-        src = inspect.getsource(memory_store.gate_consolidated)
-        test("gate uses query_memory", "query_memory" in src or "adaptive" in src.lower())
+        import ast
+        from pathlib import Path
+
+        # P1-03 (#26) discharge: gate_consolidated (the flat-0.8 consolidated
+        # shortcut) is RETIRED — deleted from the store; cached_recall routes
+        # every query through the thresholded recall only. The canary proves
+        # the retirement against the REPO copy (deterministic AST parse —
+        # not whichever memory_store a sys.path shim resolves first).
+        repo_store = Path(__file__).resolve().parent / "memory_store.py"
+        tree = ast.parse(repo_store.read_text())
+        names = {n.name for n in ast.walk(tree) if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef))}
+        test("gate_consolidated retired (flat-0.8 bypass removed)", "gate_consolidated" not in names)
     except Exception as e:
         test("gate integration", False, str(e))
 

src/rune/context_broker.py

@@ -116,6 +116,23 @@ def default_budget(window: int) -> int:
     return max(1, window - max(2048, window // 4) - window // 20)
 
 
+# Cap of the memory sub-budget (P1-03 #26): memory must never crowd out
+# tools/history/user even on small windows.
+MEMORY_BUDGET_CAP = 1200
+
+
+def default_memory_budget(window: int | None) -> int:
+    """Memory sub-budget = min(1200, window // 8) (P1-03 #26, DD 5).
+
+    ContextBroker owns the FINAL memory budget: the sum of every
+    MEMORY_BLOCK_NAMES block may never exceed this, regardless of the
+    global input budget. An unknown window falls back to the cap only.
+    """
+    if not window or window <= 0:
+        return MEMORY_BUDGET_CAP
+    return min(MEMORY_BUDGET_CAP, int(window) // 8)
+
+
 def memory_disabled() -> bool:
     """True when RUNE_MEMORY_OFF is set to a truthy value.
 
@@ -440,6 +457,25 @@ class ContextBroker:
         self._archive_path = archive_path
         self._summarizer = summarizer
         self._trace_log = []
+        self.memory_selector = self._build_memory_selector()
+
+    def _build_memory_selector(self):
+        """Compose the memory-selection authority (P1-03 #26).
+
+        The authority map names ``ContextBroker.memory_selector`` as the
+        v1_owner for memory selection. An import/construction failure is
+        fail-closed: the selector is None and the memory contributor
+        abstains — never a fallback injection.
+        """
+        try:
+            try:
+                from .memory_selector import MemorySelector
+            except ImportError:  # TUI top-level deploy
+                from memory_selector import MemorySelector
+            return MemorySelector()
+        except Exception as exc:
+            LOG.warning("memory selector unavailable (%s) — memory will abstain", exc)
+            return None
 
     # -- contributor registry ------------------------------------------------
 
@@ -495,7 +531,10 @@ class ContextBroker:
         return collected
 
     def _admit(
-        self, collected: list[tuple[str, ContributionBlock, int]], budget: int
+        self,
+        collected: list[tuple[str, ContributionBlock, int]],
+        budget: int,
+        memory_budget: int | None = None,
     ) -> tuple[list[ContributionBlock], dict]:
         """Deterministic admission: priority desc, tie: registration recency.
 
@@ -503,7 +542,16 @@ class ContextBroker:
         evictable block is admitted while the budget lasts, otherwise
         dropped with its token count recorded. A mandatory block that
         cannot fit at all raises ContextOverflowError (fail loud).
+
+        P1-03 (#26): ContextBroker owns the final memory budget. The sum of
+        MEMORY_BLOCK_NAMES blocks is bounded by ``memory_budget`` (default
+        min(1200, window // 8)): once the memory sub-budget is exhausted,
+        further memory blocks are dropped (lowest priority first via the
+        normal priority-desc walk) with a ``memory budget`` reason. The
+        active TaskState block (``user``) is non-evictable and unaffected.
         """
+        if memory_budget is None:
+            memory_budget = default_memory_budget(None)
         recency = {id(block): index for index, (_name, block, _b) in enumerate(collected)}
 
         def sort_key(entry):
@@ -514,6 +562,7 @@ class ContextBroker:
         admitted: list[ContributionBlock] = []
         trace_blocks: list[dict] = []
         used = 0
+        memory_used = 0
         for _name, block, _bucket in candidates:
             record = {
                 "name": block.name,
@@ -535,9 +584,20 @@ class ContextBroker:
                 used += block.tokens
                 trace_blocks.append(record)
                 continue
+            in_memory_scope = block.name in MEMORY_BLOCK_NAMES
+            if in_memory_scope and used + block.tokens <= budget and memory_used + block.tokens > memory_budget:
+                record["admitted"] = False
+                record["reason"] = (
+                    f"memory budget: {memory_used}+{block.tokens}>{memory_budget} "
+                    "(memory sub-budget exhausted; lowest-priority memory dropped)"
+                )
+                trace_blocks.append(record)
+                continue
             if used + block.tokens <= budget:
                 admitted.append(block)
                 used += block.tokens
+                if in_memory_scope:
+                    memory_used += block.tokens
                 trace_blocks.append(record)
             else:
                 record["admitted"] = False
@@ -549,6 +609,8 @@ class ContextBroker:
             "blocks": trace_blocks,
             "admitted_total": used,
             "dropped": dropped,
+            "memory_budget": memory_budget,
+            "memory_used": memory_used,
         }
 
     def assemble(
@@ -571,6 +633,10 @@ class ContextBroker:
             window = None
             budget = int(budget_tokens)
 
+        # P1-03 (#26): ContextBroker owns the final memory budget —
+        # min(1200, window // 8); the global budget bounds it further.
+        memory_budget = min(default_memory_budget(window), budget) if budget else default_memory_budget(window)
+
         # Measure every block with the broker estimator (measured, not guessed).
         for _name, block, _bucket in collected:
             if isinstance(block.content, str):
@@ -608,7 +674,7 @@ class ContextBroker:
                         block.tokens = sum(self._estimator(m.get("content", "")) for m in compacted)
 
         try:
-            admitted, admission = self._admit(collected, budget)
+            admitted, admission = self._admit(collected, budget, memory_budget=memory_budget)
         except ContextOverflowError:
             raise
 
@@ -629,6 +695,7 @@ class ContextBroker:
             "model_id": model_id,
             "window": window,
             "budget": budget,
+            "memory_budget": admission["memory_budget"],
             "blocks": admission["blocks"],
             "admitted": [block.name for block in admitted],
             "admitted_total": admission["admitted_total"],

src/rune/inference_trace.py

@@ -25,7 +25,7 @@ try:
 except ImportError:  # direct script execution
     from rune.memory_env import rune_home
 
-INFERENCE_TRACE_SCHEMA_VERSION = 4
+INFERENCE_TRACE_SCHEMA_VERSION = 5
 TRACE_ROTATE_BYTES = 10_000_000
 TRACE_BACKUPS = 3
 
@@ -94,6 +94,14 @@ class InferenceTrace:
     capsule_messages: list = field(default_factory=list)  # exact runtime.run payload
     capsule_content_hashes: dict = field(default_factory=dict)  # block name -> sha256
     capsule_token_count: int = 0  # post-decoration re-measured total
+    # P1-03 (#26) schema v5 — memory-selection fields, additive. The
+    # irrelevant-memory rate is computable per run:
+    # (memory_candidates - memory_selected) / memory_candidates when
+    # candidates > 0; abstained runs are recorded as zero-injection events.
+    memory_candidates: int = 0  # candidates surfaced by all selector sources
+    memory_selected: int = 0  # candidates that cleared the ONE floor
+    memory_abstained: bool = False  # True = zero memory injected (fail-closed)
+    memory_types_queried: list = field(default_factory=list)  # typed targeting
 
 
 def price_for(model: str) -> float | None:

src/rune/memory_selector.py

@@ -0,0 +1,511 @@
+#!/usr/bin/env python3
+"""memory_selector.py — the ONE memory-selection API (P1-03, #26).
+
+Authority map (config/v1_authority_map.json, domain ``memory_selection``):
+``ContextBroker.memory_selector`` is the v1_owner. Everything that surfaces
+memory into model context flows through ``MemorySelector.select_for_task`` —
+the operator registers exactly ONE memory contributor consuming it; the
+pre-P1-03 parallel contributors (recall, persona_memories, consolidated,
+neurons, graphs, abstraction) are replaced, not added to.
+
+Policy (issue #26):
+  - ONE relevance floor — ``memory_store.RECALL_MIN_SCORE`` (0.42) default —
+    applied to EVERY source. The store source applies it inside the store's
+    calibrated gate (``min_score`` is passed down into ``recall``, which
+    enforces it in EVERY mode: lexical-only, dense-only, hybrid); self-
+    scored sources are floored here. ``gate_consolidated``'s flat-0.8
+    shortcut inside ``cached_recall`` is removed (Task 3) so consolidated
+    knowledge competes under the same floor, and the top-5-importance neuron
+    fallback is gone.
+  - TRUE ABSTENTION: a trivial task legitimately receives ZERO recalled
+    memories. Below-floor candidates are counted (the irrelevant-memory rate
+    is computable per run) but never selected.
+  - Memory-type targeting via ``memory_store``'s typed recall filter.
+  - FAIL-CLOSED: a source failure is a skip + log, not an injection; if the
+    selector itself is unavailable the contributor abstains — never fail open
+    to "inject something important".
+
+Degrade-not-crash throughout: sources are pluggable and isolated; one
+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
+
+try:
+    from . import memory_store
+except ImportError:  # direct script execution
+    import memory_store  # type: ignore[no-redef]
+
+try:
+    from .context_broker import ContributionBlock
+except ImportError:  # direct script execution
+    from context_broker import ContributionBlock  # type: ignore[no-redef]
+
+LOG = logging.getLogger("rune.memory_selector")
+
+# Candidate cap BEFORE budgeting (plan DD 5: the selector caps its own
+# candidate list before returning).
+MAX_CANDIDATES = 24
+
+# Default per-task candidate budget (the old memory contributor recalled 3;
+# the collapsed selector merges all sources under one slightly larger cap).
+DEFAULT_BUDGET = 4
+
+# Priority of the single memory block (was _contributor_memory's 70).
+MEMORY_BLOCK_PRIORITY = 70
+
+
+@dataclass
+class SelectionResult:
+    """One selection outcome — mirror of CapabilityBroker's select_for_job.
+
+    ``irrelevant_rate`` is computable per run:
+    ``(candidates_evaluated - selected_count) / candidates_evaluated`` when
+    candidates > 0; abstained runs are recorded as zero-injection events.
+    """
+
+    blocks: list[ContributionBlock]
+    candidates_evaluated: int
+    selected_count: int
+    abstained: bool
+    types_queried: list[str] = field(default_factory=list)
+
+    def irrelevant_rate(self) -> float:
+        if self.candidates_evaluated <= 0:
+            return 0.0
+        return (self.candidates_evaluated - self.selected_count) / self.candidates_evaluated
+
+
+class MemorySelector:
+    """Single memory-selection API composed by ContextBroker.
+
+    Sources are registered callables ``(query, types, budget, min_score) ->
+    list[dict]`` where each dict carries at least ``text`` and ``score``
+    (0-1 relevance under the ONE floor). The store source (``_store_recall``)
+    routes through ``memory_store.recall``'s calibrated gate; Task 3 adds the
+    remaining sources. A source failure is skipped + logged (fail-closed).
+
+    KNOWN RANKING BIAS (P1-03 #26 discharge, finding 9): the ONE floor gates
+    each source on its OWN calibrated scale (lexical store ~4-6, dense/
+    hybrid store RRF ~0-0.13 with the cosine gated at 0.42, self-scored
+    0-1), but merged candidates are RANKED by raw score — so a lexical-mode
+    store row outranks an equally-relevant self-scored row. Abstention is
+    unaffected (every source is floored on its own scale); only candidate
+    ORDER is biased. Per-source rank normalization was deliberately deferred:
+    renormalizing the store's calibrated scales risked false abstentions
+    this late in the release.
+    """
+
+    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)
+        # 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 -----------------------------------------------------
+
+    def add_source(self, name: str, fn: Callable) -> None:
+        """Register a selection source. Registration ORDER is execution order
+        (dict preserves it); re-registering a name replaces that source."""
+        self._sources[name] = fn
+
+    def source_names(self) -> list[str]:
+        """Registered source names in execution order."""
+        return list(self._sources)
+
+    # -- selection -------------------------------------------------------------
+
+    def select_for_task(
+        self,
+        task_description: str,
+        types: list[str] | None = None,
+        budget: int = DEFAULT_BUDGET,
+    ) -> SelectionResult:
+        """Select memory for one task under the ONE floor.
+
+        Returns a SelectionResult; on total selector failure the caller
+        (operator contributor) treats any exception as an abstention.
+        """
+        query = (task_description or "").strip()
+        types_queried = list(types) if types else []
+        if not query:
+            return SelectionResult([], 0, 0, True, types_queried)
+
+        candidates: list[dict] = []
+        for name, fn in self._sources.items():
+            try:
+                found = fn(query, types_queried, MAX_CANDIDATES, self.min_score)
+            except Exception as exc:  # fail-closed per source
+                LOG.warning("memory source %r failed; skipping (%s)", name, exc)
+                continue
+            for item in found or []:
+                entry = dict(item)
+                entry["kind"] = entry.get("kind") or name
+                candidates.append(entry)
+        candidates = candidates[:MAX_CANDIDATES]
+
+        # ONE floor for EVERY source: the store source was already floored by
+        # the same constant inside recall() — which applies min_score in
+        # EVERY mode (lexical-only, dense-only, hybrid; P1-03 #26 discharge)
+        # — so its rows are tagged store_scored and exempt from this
+        # selector-side numeric check. Self-scored sources are floored here
+        # against the same selector min_score.
+        selected = [item for item in candidates if item.get("store_scored") or _score_of(item) >= self.min_score]
+        selected.sort(key=lambda item: -_score_of(item))
+        selected = selected[: max(0, budget)]
+
+        blocks: list[ContributionBlock] = []
+        for item in selected:
+            text = str(item.get("text", "")).strip()
+            if not text:
+                continue
+            blocks.append(
+                ContributionBlock(
+                    "memory",
+                    MEMORY_BLOCK_PRIORITY,
+                    f"- {text}",
+                    max(1, len(text) // 4),
+                    "memory_selector.select_for_task",
+                    evictable=True,
+                )
+            )
+
+        abstained = not blocks
+        if abstained:
+            LOG.info(
+                "memory abstain: task=%r candidates=%d floor=%.2f types=%s",
+                query[:60],
+                len(candidates),
+                self.min_score,
+                types_queried or "all",
+            )
+        return SelectionResult(blocks, len(candidates), len(blocks), abstained, types_queried)
+
+    # -- sources ---------------------------------------------------------------
+
+    def _source_store(self, query: str, types: list[str], budget: int, min_score: float) -> list[dict]:
+        """Typed store recall through the ONE thresholded path."""
+        return _store_recall(query, types, budget, min_score, db_path=self.db_path)
+
+
+def _score_of(item: dict) -> float:
+    """Extract the item's 0-1 relevance under the ONE floor. Missing or
+    malformed scores floor at 0.0 (abstain) — never guessed upward."""
+    value = item.get("score")
+    if isinstance(value, bool) or not isinstance(value, (int, float)):
+        return 0.0
+    return float(value)
+
+
+def _store_recall(
+    query: str,
+    types: list[str] | None,
+    budget: int,
+    min_score: float = memory_store.RECALL_MIN_SCORE,
+    db_path: str | None = None,
+) -> list[dict]:
+    """Store source: typed recall through the ONE thresholded path.
+
+    Uses ``memory_store.recall`` (the calibrated QPP gate) — never the
+    ``cached_recall`` path whose ``gate_consolidated`` flat-0.8 shortcut
+    bypasses thresholding (removed in Task 3).
+
+    The store's ``relevance`` lives on mode-dependent calibrated scales
+    (lexical-mode ~4-6 via the +5 offset band, dense/hybrid-mode RRF ~0-0.13,
+    with the dense COSINE gated at 0.42-0.8 — see memory_store.recall), so it
+    is NOT renormalized here: the store already applied THE floor via
+    ``min_score`` in EVERY recall mode (lexical-only, dense-only, hybrid —
+    P1-03 #26 discharge), so tagging the result ``store_scored=True`` and
+    exempting it from the selector's numeric check is sound. The result is
+    tagged ``store_scored=True`` so the selector treats these rows as
+    floor-cleared by the same constant.
+
+    KNOWN RANKING BIAS (P1-03 #26 discharge, finding 9): because the scales
+    differ per mode, the selector's raw-score ranking favors lexical-mode
+    store rows (~4-6) over self-scored rows (0-1) and dense/hybrid-mode
+    store rows (~0-0.13). Abstention is NOT affected (each source is floored
+    on its own calibrated scale); only the ORDER of merged candidates is
+    biased. Per-source rank normalization was deliberately deferred —
+    renormalizing the store's calibrated scales risked false abstentions.
+
+    P1-03 (#26) DISCHARGE (finding 2a): besides the caller-typed query, the
+    source always runs a SECOND typed query restricted to
+    ``self``/``personality`` — the persona lane. The production caller (the
+    operator's single memory contributor) invokes the selector WITHOUT
+    types, which left the typed persona source dead in production. Both
+    queries are thresholded inside the store's gate and may abstain — a
+    trivial task still gets zero persona rows. Results are merged and
+    capped at ``budget``.
+
+    P1-03 (#26) DEVIATION: shard routing inside ``recall`` is bypassed — the
+    selector pins the canonical store connection. ShardManager routes to a
+    hardcoded ~/.config/opencode shard dir that ignores RUNE_* isolation and
+    carries no memory_type column, so routing through it would break typed
+    recall and the preserve-stores/isolation invariants.
+    """
+    db = memory_store.connect(db_path)
+    try:
+        results = memory_store.recall(db, query, budget=budget, min_score=min_score, memory_types=types or None)
+        persona_types = ["self", "personality"]
+        if not (set(types or []) & set(persona_types)):
+            # The persona lane: always executed for the untyped production
+            # caller (and any caller that did not itself ask persona types).
+            results = results + memory_store.recall(
+                db, query, budget=budget, min_score=min_score, memory_types=persona_types
+            )
+    finally:
+        db.close()
+    out: list[dict] = []
+    seen_ids: set = set()
+    for row in results:
+        # The untyped and typed-lane queries overlap on unclassified rows —
+        # merge by entry id so a shared row is a candidate ONCE (duplicates
+        # would corrupt the irrelevant-memory rate and repeat block text).
+        row_id = row.get("id")
+        if row_id is not None:
+            if row_id in seen_ids:
+                continue
+            seen_ids.add(row_id)
+        out.append(
+            {
+                "text": row.get("text", ""),
+                "score": float(row.get("relevance", 0.0)),
+                "store_scored": True,
+                "kind": "store",
+                "id": row.get("id"),
+                "source": row.get("source", ""),
+            }
+        )
+    return out
+
+
+def _lex_overlap_score(text: str, query: str) -> float:
+    """Deterministic lexical relevance for self-scored sources.
+
+    Distinctive-word coverage of the QUERY terms present in the candidate
+    text, with a stopword filter shared with the store's gate vocabulary.
+    A candidate sharing NO distinctive query word cannot score — the same
+    conjunctive logic the store's BM25 gate applies (a shared single word is
+    not a relation, but zero shared words is definitely not one).
+    """
+    stopwords = getattr(memory_store, "_STOPWORDS", set())
+    q_terms = {t.lower() for t in query.split() if len(t) > 3 and t.lower() not in stopwords}
+    if not q_terms:
+        return 0.0
+    text_lower = (text or "").lower()
+    hits = sum(1 for term in q_terms if term in text_lower)
+    return hits / len(q_terms)
+
+
+def _neurons_recall(query: str, types: list[str] | None, budget: int, min_score: float) -> list[dict]:
+    """Warm-neuron source: trigger scan, thresholded — NO importance fallback.
+
+    The pre-P1-03 contributor injected the top-5 neurons by importance when
+    no trigger matched (rune_operator.py:1387-1389). That fallback is dead:
+    a neuron that does not match the task is not injected merely because it
+    is important. Trigger-matched neurons are scored lexically against the
+    query and must clear the same floor as every other source.
+    """
+    try:
+        import warm_neuron_store
+    except ImportError:  # TUI top-level deploy
+        import rune.warm_neuron_store as warm_neuron_store  # type: ignore[no-redef]
+
+    neurons = warm_neuron_store.list_neurons()
+    task_words = {w.lower().strip(".,;:!?()[]{}\"'") for w in query.split()}
+    out: list[dict] = []
+    for neuron in neurons:
+        triggers = neuron.get("triggers") or []
+        if not any(t in task_words for t in triggers):
+            continue
+        text = neuron.get("text", "")
+        score = _lex_overlap_score(text, query)
+        if score < min_score:
+            continue
+        out.append(
+            {
+                "text": f"[{neuron.get('slug', 'warm')}] {text}",
+                "score": score,
+                "kind": "neuron",
+                "source": "warm_neuron_store.list_neurons",
+            }
+        )
+        if len(out) >= budget:
+            break
+    return out
+
+
+def _graphs_recall(query: str, types: list[str] | None, budget: int, min_score: float) -> list[dict]:
+    """Graph source: entity BFS from the task words via graph_memory.query,
+    floored. The pre-P1-03 contributor called resolve_entity(), which MUTATES
+    the graph (creates missing entities); read-only query() keeps the store
+    side-effect-free. A term with no entity yields no candidates."""
+    try:
+        import graph_memory
+    except ImportError:  # TUI top-level deploy
+        import rune.graph_memory as graph_memory  # type: ignore[no-redef]
+
+    db = graph_memory.connect()
+    out: list[dict] = []
+    try:
+        task_words = list(dict.fromkeys(w.lower().strip(".,;:!?()[]{}\"'") for w in query.split()))[:5]
+        for word in task_words:
+            result = graph_memory.query(db, word, hops=1)
+            if not result:
+                continue
+            entity = result.get("entity", word) if isinstance(result, dict) else str(result)
+            facts = result.get("facts", []) if isinstance(result, dict) else []
+            lines = [f"{f.get('subject')} {f.get('predicate', '-')} {f.get('object')}" for f in facts]
+            text = "; ".join(lines) if lines else str(entity)
+            score = _lex_overlap_score(text, query) or (1.0 / max(len(task_words), 1))
+            if score < min_score:
+                continue
+            out.append({"text": text, "score": score, "kind": "graph", "source": "graph_memory.query"})
+            if len(out) >= budget:
+                break
+    finally:
+        db.close()
+    return out
+
+
+def _abstraction_recall(query: str, types: list[str] | None, budget: int, min_score: float) -> list[dict]:
+    """Abstraction-facts source: LIKE scan over active facts, floored.
+
+    The pre-P1-03 contributor injected the top-3 LIKE matches unconditionally;
+    here a LIKE match is only a candidate — it must clear the floor.
+    """
+    try:
+        import abstraction
+    except ImportError:  # TUI top-level deploy
+        import rune.abstraction as abstraction  # type: ignore[no-redef]
+
+    db = abstraction.connect()
+    out: list[dict] = []
+    try:
+        task_words = [w for w in query.lower().split()[:3] if w]
+        if not task_words:
+            return []
+        placeholders = " OR ".join(["what LIKE ?" for _ in task_words])
+        params = [f"%{w}%" for w in task_words]
+        rows = db.execute(
+            f"SELECT what FROM facts WHERE state='active' AND ({placeholders}) LIMIT {int(budget) * 3}",
+            params,
+        ).fetchall()
+        for row in rows:
+            text = str(row[0])
+            score = _lex_overlap_score(text, query)
+            if score < min_score:
+                continue
+            out.append({"text": text, "score": score, "kind": "abstraction", "source": "abstraction facts"})
+            if len(out) >= budget:
+                break
+    finally:
+        db.close()
+    return out
+
+
+def _consolidated_recall(query: str, types: list[str] | None, budget: int, min_score: float) -> list[dict]:
+    """Consolidated-knowledge source under the SAME floor (bypass removed).
+
+    The pre-P1-03 path handed ``gate_consolidated``'s output a flat
+    relevance=0.8, bypassing thresholding entirely. Consolidated knowledge
+    now competes like any candidate: EACH knowledge item is scored by its
+    lexical overlap with the active task, and below-floor items abstain.
+    """
+    try:
+        from .memory_consolidation import MemoryConsolidation
+    except ImportError:  # TUI top-level deploy
+        from memory_consolidation import MemoryConsolidation  # type: ignore[no-redef]
+
+    try:
+        consolidation = MemoryConsolidation()
+        items = list(consolidation.get_relevant_knowledge(query, limit=budget * 2))
+    except Exception as exc:  # degrade: consolidation unavailable -> no candidates
+        LOG.warning("consolidation source unavailable; skipping (%s)", exc)
+        return []
+    out: list[dict] = []
+    for item in items:
+        content = (getattr(item, "content", "") or "").strip()
+        if not content:
+            continue
+        score = _lex_overlap_score(content, query)
+        if score < min_score:
+            continue  # below floor -> abstain (was: flat relevance injection)
+        out.append({"text": content, "score": score, "kind": "consolidated", "source": "memory_consolidation"})
+        if len(out) >= budget:
+            break
+    return out
+
+
+def _persona_recall(
+    query: str,
+    types: list[str] | None,
+    budget: int,
+    min_score: float,
+    db_path: str | None = None,
+) -> list[dict]:
+    """Typed persona source: self/personality memory keyed on the ACTIVE task.
+
+    Replaces the hardcoded ``"Rune personality voice identity how to speak"``
+    query. Only runs when the caller asked for self/personality types (the
+    operator's persona-memory call site); thresholded, may abstain — a
+    trivial task gets zero persona lines.
+
+    P1-03 (#26) DISCHARGE (finding 3): the source queries the selector's
+    PINNED store (``db_path``) — never the module-default store that a bare
+    ``memory_store.connect()`` would open.
+    """
+    wanted = {t for t in (types or []) if t in ("self", "personality")}
+    if not wanted:
+        return []
+    db = memory_store.connect(db_path)
+    try:
+        rows = memory_store.recall(db, query, budget=budget, min_score=min_score, memory_types=sorted(wanted))
+    finally:
+        db.close()
+    out: list[dict] = []
+    for row in rows:
+        out.append(
+            {
+                "text": row.get("text", ""),
+                "score": float(row.get("relevance", 0.0)),
+                "kind": "persona",
+                "source": "memory_store.recall(self/personality)",
+            }
+        )
+    return out
+
+
+# The default source set (P1-03 #26): every pre-selector memory path, unified
+# under the ONE floor. Registration order = execution order.
+DEFAULT_SOURCES: dict[str, Callable] = {
+    "store": _store_recall,
+    "neurons": _neurons_recall,
+    "graphs": _graphs_recall,
+    "abstraction": _abstraction_recall,
+    "consolidated": _consolidated_recall,
+    "persona": _persona_recall,
+}

src/rune/rune_operator.py

@@ -756,20 +756,46 @@ class RuneOperator:
             return selection
 
     def _conversation_messages(self, current_task_id: str, limit: int = 8) -> list[dict]:
-        """Return recent completed user/assistant pairs for the next turn."""
-        completed = [
-            item
-            for item in self.state.tasks
-            if item.task_id != current_task_id
-            and item.status == "completed"
-            and isinstance(item.result, str)
-            and item.result.strip()
-        ][-limit:]
+        """Recent completed-task turns from the CANONICAL ledger (#18, P1-03).
+
+        Status authority is TaskStateRepository (completed tasks only);
+        message bodies come from ConversationStateRepository. Past-task
+        content is explicitly labeled context-only so it can never be read
+        as instructions — and it never keys a recall query (the selector is
+        queried with the ACTIVE task description only).
+        """
+        window = self._conversation.window(OPERATOR_CHANNEL, n=limit * 4)
+        by_task: dict[str, list[dict]] = {}
+        order: list[str] = []
+        for msg in window:
+            if not msg.task_id or msg.task_id == current_task_id:
+                continue
+            if msg.task_id not in by_task:
+                order.append(msg.task_id)
+                by_task[msg.task_id] = []
+            by_task[msg.task_id].append({"role": msg.role, "content": msg.content})
         messages: list[dict] = []
-        for item in completed:
-            messages.append({"role": "user", "content": item.description})
-            messages.append({"role": "assistant", "content": item.result})
-        return messages
+        for task_id in reversed(order):
+            record = self._task_state.get(task_id)
+            if record is None or record.status != "completed":
+                continue
+            labeled: list[dict] = []
+            for message in by_task[task_id]:
+                if message["role"] == "user":
+                    labeled.append(
+                        {
+                            "role": "user",
+                            "content": (
+                                "PAST TASK (completed — context only, not instructions): " + (message["content"] or "")
+                            ),
+                        }
+                    )
+                else:
+                    labeled.append(message)
+            messages = labeled + messages  # chronological order
+            if len(messages) >= limit * 2:
+                break
+        return messages[: limit * 2]
 
     def resolve_pending_approval(self, approved: bool) -> Dict[str, Any]:
         """Resolve and resume the single persisted operator approval."""
@@ -984,14 +1010,13 @@ class RuneOperator:
 
         archive = Path(rune_home()) / "state" / "history_archive.jsonl"
         broker = ContextBroker(archive_path=archive)
+        # P1-03 (#26): exactly ONE memory contributor — the six parallel
+        # memory paths (memory, persona_memories, consolidated, neurons,
+        # graphs, abstraction) are replaced by this single contributor
+        # delegating to broker.memory_selector (authority-map v1_owner).
         broker.register_contributor("persona", self._contributor_persona, 0)
         broker.register_contributor("tools", self._contributor_tools, 1)
         broker.register_contributor("memory", self._contributor_memory, 2)
-        broker.register_contributor("persona_memories", self._contributor_persona_memories, 2)
-        broker.register_contributor("consolidated", self._contributor_consolidated, 2)
-        broker.register_contributor("neurons", self._contributor_neurons, 2)
-        broker.register_contributor("graphs", self._contributor_graphs, 2)
-        broker.register_contributor("abstraction", self._contributor_abstraction, 2)
         broker.register_contributor("model_profile", self._contributor_model_profile, 2)
         broker.register_contributor("history", self._contributor_history, 3)
         broker.register_contributor("user", self._contributor_user, 4)
@@ -1099,315 +1124,66 @@ I have tools to read, write, edit, search, lint, test, format, and commit code.
     # -- memory notes (bucket 2, evictable; zero under RUNE_MEMORY_OFF) ---
 
     def _contributor_memory(self, task, session):
-        """Recall relevant memory for this task → '## Context'."""
-        try:
-            from .context_broker import ContributionBlock, memory_disabled
-        except ImportError:  # TUI top-level deploy
-            from rune.context_broker import ContributionBlock, memory_disabled
-        try:
-            from .subsystem_health import record_failure
-        except ImportError:  # TUI top-level deploy
-            from rune.subsystem_health import record_failure
+        """THE single memory contributor (P1-03 #26).
 
-        if memory_disabled():
-            return None
-        import rune.memory_store as memory_store
-
-        try:
-            db = memory_store.connect()
-            try:
-                results = memory_store.cached_recall(db, task.description, budget=3)
-            finally:
-                db.close()
-        except Exception as exc:  # P0-06: observable degrade (T1)
-            record_failure("memory_selector", exc, tier=1)
-            if isinstance(session, dict):
-                session.setdefault("degraded", []).append("memory_selector")
-            return None
-        if not results:
-            return None
-        if isinstance(session, dict):
-            session["recall_count"] = len(results)
-        memories = [r.get("text", "")[:100] for r in results[:3]]
-        content = "\n".join(f"- {m}" for m in memories if m)
-        if not content:
-            return None
-        return ContributionBlock("memory", 70, f"\n## Context\n{content}", 0, "memory_store.cached_recall")
-
-    def _contributor_persona_memories(self, task, session):
-        """Second cached_recall (identity) → '## Identity context'."""
-        try:
-            from .context_broker import ContributionBlock, memory_disabled
-        except ImportError:  # TUI top-level deploy
-            from rune.context_broker import ContributionBlock, memory_disabled
-        try:
-            from .subsystem_health import record_failure
-        except ImportError:  # TUI top-level deploy
-            from rune.subsystem_health import record_failure
-
-        if memory_disabled():
-            return None
-        import rune.memory_store as memory_store
-
-        try:
-            db = memory_store.connect()
-            try:
-                persona_memories = memory_store.cached_recall(
-                    db, "Rune personality voice identity how to speak", budget=3
-                )
-            finally:
-                db.close()
-        except Exception as exc:  # P0-06: observable degrade (T1)
-            record_failure("persona", exc, tier=1)
-            if isinstance(session, dict):
-                session.setdefault("degraded", []).append("persona")
-            return None
-        if not persona_memories:
-            return None
-        persona_context = "\n".join(f"- {r.get('text', '')[:100]}" for r in persona_memories if r.get("text"))
-        if not persona_context:
-            return None
-        return ContributionBlock(
-            "persona_memories", 68, f"\n## Identity context\n{persona_context}", 0, "memory_store.cached_recall"
-        )
-
-    def _contributor_consolidated(self, task, session):
-        """Consolidation pipeline block, WITH its write side effect.
-
-        CAREFUL (P0-03 constraint): ``_save_knowledge()`` runs at COLLECT
-        time — exactly once per turn, even when admission later drops or
-        the memory gate zeroes this block. Consolidation writes knowledge
-        derived from memory systems; it is gated by the memory-off switch
-        (there is nothing to consolidate when memory is off).
+        Delegates to ``broker.memory_selector.select_for_task`` — the
+        authority-map v1_owner. The five parallel memory contributors this
+        replaced (persona_memories, consolidated, neurons, graphs,
+        abstraction) are gone: their sources now live inside the selector
+        under the ONE relevance floor. Fail-closed: selector unavailable,
+        selection failure, or empty selection → None (zero memory blocks) —
+        never a fallback injection.
         """
         try:
             from .context_broker import ContributionBlock, memory_disabled
         except ImportError:  # TUI top-level deploy
             from rune.context_broker import ContributionBlock, memory_disabled
-        try:
-            from .subsystem_health import record_failure
-        except ImportError:  # TUI top-level deploy
-            from rune.subsystem_health import record_failure
 
         if memory_disabled():
             return None
-        try:
-            from .memory_consolidation import MemoryConsolidation
-        except ImportError:  # TUI top-level deploy
-            from rune.memory_consolidation import MemoryConsolidation
-
-        degraded_reasons: list[str] = []
-        consolidation = MemoryConsolidation()
-
-        # Consolidate warm neurons into knowledge
-        try:
-            import warm_neuron_store
-
-            neurons = warm_neuron_store.list_neurons()
-            for neuron in neurons[:10]:  # Consolidate top 10
-                knowledge = consolidation.consolidate_neuron(neuron)
-                existing = [k for k in consolidation.knowledge if k.id == knowledge.id]
-                if not existing:
-                    consolidation.knowledge.append(knowledge)
-        except Exception as exc:  # P0-06: observable degrade (T1)
-            record_failure("consolidation", exc, tier=1, context="warm neurons")
-            degraded_reasons.append("warm neurons")
-
-        # Consolidate graph memory into knowledge
-        try:
-            import graph_memory
-
-            db = graph_memory.connect()
-            try:
-                task_words = task.description.lower().split()[:5]
-                for word in task_words:
-                    entity = graph_memory.resolve_entity(db, word)
-                    if entity:
-                        knowledge = consolidation.consolidate_graph_entity(entity)
-                        existing = [k for k in consolidation.knowledge if k.id == knowledge.id]
-                        if not existing:
-                            consolidation.knowledge.append(knowledge)
-            finally:
-                db.close()
-        except Exception as exc:  # P0-06: observable degrade (T1)
-            record_failure("consolidation", exc, tier=1, context="graph memory")
-            degraded_reasons.append("graph memory")
-
-        # Consolidate abstraction facts into knowledge
-        try:
-            import abstraction
-
-            db = abstraction.connect()
-            try:
-                task_words = task.description.lower().split()[:3]
-                if task_words:
-                    placeholders = " OR ".join(["what LIKE ?" for _ in task_words])
-                    params = [f"%{w}%" for w in task_words]
-                    facts = db.execute(
-                        f"SELECT what FROM facts WHERE status='active' AND ({placeholders}) LIMIT 5",
-                        params,
-                    ).fetchall()
-                    for fact in facts:
-                        knowledge = consolidation.consolidate_abstraction_fact(fact[0])
-                        existing = [k for k in consolidation.knowledge if k.id == knowledge.id]
-                        if not existing:
-                            consolidation.knowledge.append(knowledge)
-            finally:
-                db.close()
-        except Exception as exc:  # P0-06: observable degrade (T1)
-            record_failure("consolidation", exc, tier=1, context="abstraction facts")
-            degraded_reasons.append("abstraction facts")
-
-        # Consolidate memory recall into knowledge
-        try:
-            import rune.memory_store as memory_store
-
-            db = memory_store.connect()
-            try:
-                results = memory_store.cached_recall(db, task.description, budget=5)
-                if results:
-                    for r in results:
-                        text = r.get("text", "")
-                        score = r.get("score", 0.5)
-                        if text:
-                            knowledge = consolidation.consolidate_memory(text, score)
-                            existing = [k for k in consolidation.knowledge if k.id == knowledge.id]
-                            if not existing:
-                                consolidation.knowledge.append(knowledge)
-            finally:
-                db.close()
-        except Exception as exc:  # P0-06: observable degrade (T1)
-            record_failure("consolidation", exc, tier=1, context="memory recall")
-            degraded_reasons.append("memory recall")
-
-        # Write side effect — EXACTLY ONCE per turn, at collect time,
-        # before admission decides whether the block is visible.
-        consolidation._save_knowledge()
-
-        consolidated_context = consolidation.format_knowledge_for_prompt(task.description)
-        if not consolidated_context:
-            return None
-        return ContributionBlock("consolidated", 60, f"\n{consolidated_context}", 0, "memory_consolidation")
-
-    def _contributor_neurons(self, task, session):
-        """Warm-neuron trigger scan top-5 → '## My neurons'."""
-        try:
-            from .context_broker import ContributionBlock, memory_disabled
-        except ImportError:  # TUI top-level deploy
-            from rune.context_broker import ContributionBlock, memory_disabled
-        try:
-            from .subsystem_health import record_failure
-        except ImportError:  # TUI top-level deploy
-            from rune.subsystem_health import record_failure
-
-        if memory_disabled():
-            return None
-        import warm_neuron_store
-
-        try:
-            neurons = warm_neuron_store.list_neurons()
-        except Exception as exc:  # P0-06: observable degrade (T1)
-            record_failure("warm_router", exc, tier=1, context="neuron injection")
+        broker = self._context_broker()
+        selector = getattr(broker, "memory_selector", None)
+        if selector is None:
+            # Fail-closed abstention — recorded so the trace proves it.
             if isinstance(session, dict):
-                session.setdefault("degraded", []).append("warm_router")
+                session["memory_candidates"] = 0
+                session["memory_selected"] = 0
+                session["memory_abstained"] = True
+                session["memory_types_queried"] = []
             return None
-        if not neurons:
-            return None
-        # Find neurons relevant to the current task
-        task_words = set(task.description.lower().split())
-        relevant = []
-        for n in neurons:
-            triggers = n.get("triggers", [])
-            if any(t in task_words for t in triggers):
-                relevant.append(n)
-        # If no specific match, include top neurons by importance
-        if not relevant:
-            relevant = sorted(neurons, key=lambda x: x.get("importance", 0), reverse=True)[:5]
-        else:
-            relevant = relevant[:5]
-        neuron_context = "\n".join(f"- [{n.get('slug', '?')}] {n.get('text', '')[:100]}" for n in relevant)
-        return ContributionBlock(
-            "neurons", 55, f"\n## My neurons (warm, firing)\n{neuron_context}", 0, "warm_neuron_store.list_neurons"
-        )
-
-    def _contributor_graphs(self, task, session):
-        """Graph-memory entity resolution top-5 → '## My graph connections'."""
+        description = getattr(task, "description", "") or ""
         try:
-            from .context_broker import ContributionBlock, memory_disabled
-        except ImportError:  # TUI top-level deploy
-            from rune.context_broker import ContributionBlock, memory_disabled
-        try:
-            from .subsystem_health import record_failure
-        except ImportError:  # TUI top-level deploy
-            from rune.subsystem_health import record_failure
-
-        if memory_disabled():
-            return None
-        import graph_memory
-
-        try:
-            db = graph_memory.connect()
+            result = selector.select_for_task(description)
+        except Exception as exc:  # fail-closed: abstain, log, continue
             try:
-                task_words = task.description.lower().split()[:5]
-                entities = []
-                for word in task_words:
-                    entity = graph_memory.resolve_entity(db, word)
-                    if entity:
-                        entities.append(entity)
-            finally:
-                db.close()
-        except Exception as exc:  # P0-06: observable degrade (T1)
-            record_failure("graph_memory", exc, tier=1)
+                from .subsystem_health import record_failure
+            except ImportError:  # TUI top-level deploy
+                from rune.subsystem_health import record_failure
+
+            record_failure("memory_selector", exc, tier=1)
             if isinstance(session, dict):
-                session.setdefault("degraded", []).append("graph_memory")
+                session.setdefault("degraded", []).append("memory_selector")
+                session["memory_candidates"] = 0
+                session["memory_selected"] = 0
+                session["memory_abstained"] = True
+                session["memory_types_queried"] = []
             return None
-        if not entities:
+        # P1-03 (#26): selection summary rides the session to the durable
+        # trace (v5 fields) — the irrelevant-memory rate is computable.
+        if isinstance(session, dict):
+            session["memory_candidates"] = result.candidates_evaluated
+            session["memory_selected"] = result.selected_count
+            session["memory_abstained"] = result.abstained
+            session["memory_types_queried"] = list(result.types_queried)
+        blocks = [
+            block for block in result.blocks if isinstance(getattr(block, "content", ""), str) and block.content.strip()
+        ]
+        if not blocks:
             return None
-        entity_context = "\n".join(f"- {e.get('name', '?')}: {e.get('description', '')[:80]}" for e in entities[:5])
-        return ContributionBlock(
-            "graphs", 50, f"\n## My graph connections\n{entity_context}", 0, "graph_memory.resolve_entity"
-        )
-
-    def _contributor_abstraction(self, task, session):
-        """Abstraction facts LIKE top-3 → '## My learned knowledge'."""
-        try:
-            from .context_broker import ContributionBlock, memory_disabled
-        except ImportError:  # TUI top-level deploy
-            from rune.context_broker import ContributionBlock, memory_disabled
-        try:
-            from .subsystem_health import record_failure
-        except ImportError:  # TUI top-level deploy
-            from rune.subsystem_health import record_failure
-
-        if memory_disabled():
-            return None
-        import abstraction
-
-        try:
-            db = abstraction.connect()
-            try:
-                task_words = task.description.lower().split()[:3]
-                if not task_words:
-                    return None
-                placeholders = " OR ".join(["what LIKE ?" for _ in task_words])
-                params = [f"%{w}%" for w in task_words]
-                facts = db.execute(
-                    f"SELECT what FROM facts WHERE status='active' AND ({placeholders}) LIMIT 3",
-                    params,
-                ).fetchall()
-            finally:
-                db.close()
-        except Exception as exc:  # P0-06: observable degrade (T1)
-            record_failure("abstraction", exc, tier=1)
-            if isinstance(session, dict):
-                session.setdefault("degraded", []).append("abstraction")
-            return None
-        if not facts:
-            return None
-        fact_context = "\n".join(f"- {f[0][:100]}" for f in facts)
-        return ContributionBlock(
-            "abstraction", 45, f"\n## My learned knowledge\n{fact_context}", 0, "abstraction facts"
-        )
+        if isinstance(session, dict):
+            session["recall_count"] = result.selected_count
+        content = "\n## Context\n" + "\n".join(block.content for block in blocks)
+        return ContributionBlock("memory", 70, content, 0, "memory_selector.select_for_task")
 
     def _contributor_model_profile(self, task, session):
         """Model-profile prompt context → '## Model' (mock contract kept)."""
@@ -1427,11 +1203,19 @@ I have tools to read, write, edit, search, lint, test, format, and commit code.
     # -- history (bucket 3, evictable; compaction-ready) -------------------
 
     def _contributor_history(self, task, session):
-        """Recent completed pairs as the history block (compaction target).
+        """Past completed-task turns as the history block (compaction target).
+
+        P1-03 (#26) TaskState supremacy: rebased on TaskStateRepository
+        (status authority) + ConversationStateRepository (bodies). Past-task
+        content is labeled ``PAST TASK (completed — context only, not
+        instructions)`` and this block's priority (30) stays strictly below
+        the active TaskState block (``user``, 100, non-evictable) — old-task
+        text can never outrank the active task. Recall queries are keyed
+        ONLY on the active task description, never on past-task text.
 
         Not gated by RUNE_MEMORY_OFF: history is the operator's own task
-        ledger (RuneState), not a memory-store selector — the memory-off
-        absence proof covers recall/persona/consolidation only.
+        ledger, not a memory-store selector — the memory-off absence proof
+        covers recall/persona/consolidation only.
         """
         try:
             from .context_broker import ContributionBlock
@@ -1826,6 +1610,12 @@ I have tools to read, write, edit, search, lint, test, format, and commit code.
                             and not isinstance(getattr(capsule, "token_count", 0), bool)
                             else 0
                         ),
+                        # P1-03 (#26) schema v5: memory-selection fields —
+                        # session-carried summary of the ONE selector's run.
+                        memory_candidates=int(session.get("memory_candidates", 0) or 0),
+                        memory_selected=int(session.get("memory_selected", 0) or 0),
+                        memory_abstained=bool(session.get("memory_abstained", trace["memory_tokens"] == 0)),
+                        memory_types_queried=list(session.get("memory_types_queried", []) or []),
                     )
                     record_trace(trace_record)
                     # P1-06 (quality I-2 discharge): CLOSE the measured loop —

src/rune/shard_manager.py

@@ -53,6 +53,10 @@ class ShardManager:
             return shard_path
 
         db = sqlite3.connect(shard_path)
+        db.enable_load_extension(True)
+        import sqlite_vec
+
+        sqlite_vec.load(db)  # #26: entries_vec virtual table needs the extension per connection
         db.executescript("""
             CREATE TABLE IF NOT EXISTS entries (
                 id INTEGER PRIMARY KEY AUTOINCREMENT,
@@ -71,6 +75,9 @@ class ShardManager:
             CREATE VIRTUAL TABLE IF NOT EXISTS entries_fts USING fts5(
                 text, content='', tokenize='porter unicode61'
             );
+            CREATE VIRTUAL TABLE IF NOT EXISTS entries_vec USING vec0(
+                embedding float[256]
+            );
             CREATE INDEX IF NOT EXISTS idx_entries_status ON entries(status);
             CREATE INDEX IF NOT EXISTS idx_entries_topic ON entries(topic);
             CREATE INDEX IF NOT EXISTS idx_entries_wash_count ON entries(wash_count);
@@ -112,13 +119,25 @@ class ShardManager:
         return "general"
 
     def get_shard(self, topic: str) -> sqlite3.Connection:
-        """Get or create a shard for a topic. Returns database connection."""
+        """Get or create a shard for a topic. Returns database connection.
+
+        #26 defect fix: virtual tables (entries_vec) require the sqlite-vec
+        extension loaded PER CONNECTION — a fresh connection without it cannot
+        see or query entries_vec, even though the schema created it ("no such
+        table" in clean environments). Mirror memory_store.connect()."""
         if topic not in self.index:
             self.create_shard(topic)
 
         shard_path = self.index[topic]
         db = sqlite3.connect(shard_path)
         db.row_factory = sqlite3.Row
+        db.enable_load_extension(True)
+        import sqlite_vec
+
+        sqlite_vec.load(db)
+        # Existing shard files may predate the vec schema — ensure idempotently
+        # (CREATE IF NOT EXISTS is a no-op on current shards).
+        db.execute("CREATE VIRTUAL TABLE IF NOT EXISTS entries_vec USING vec0(embedding float[256])")
         return db
 
     def merge_results(self, results_list: List[List[dict]]) -> List[dict]:
@@ -160,8 +179,10 @@ class ShardManager:
 # CLI
 # ═══════════════════════════════════════════════════════════════════════════════
 
+
 def main():
     import argparse
+
     parser = argparse.ArgumentParser(description="Shard manager for unbounded memory")
     sub = parser.add_subparsers(dest="cmd")
 

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

@@ -282,20 +282,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_context_broker.py

@@ -399,9 +399,16 @@ def test_operator_characterization_parity():
         patch.object(rune_operator.model_profiles, "get_prompt_context", return_value="model context"),
         patch.object(rune_operator.model_profiles, "record_interaction") as record,
         patch("rune.memory_store.connect", return_value=memory_db),
-        patch(
-            "rune.memory_store.cached_recall",
-            return_value=[{"text": "remembered fact"}],
+        # P1-03 (#26): memory selection authority moved to the selector —
+        # script the store SOURCE there (same injected memory, same count).
+        patch.dict(
+            "rune.memory_selector.DEFAULT_SOURCES",
+            {
+                "store": lambda query, types, budget, min_score: [
+                    {"text": "remembered fact", "score": 0.9, "kind": "store"}
+                ]
+            },
+            clear=True,
         ),
         patch.object(operator, "_log_token_usage"),
     ):

tests/test_merge_robot.py

@@ -54,6 +54,8 @@ class FakeForge:
         if path.endswith("/merge"):
             n = int(path.split("/")[2])
             return self.merge_results.get(n, (200, {}))
+        if path.endswith("/update"):
+            return 409, {"message": "There are conflicting files"}  # fixture default: genuine conflict
         raise AssertionError(f"unexpected POST {path}")
 
 
@@ -118,8 +120,13 @@ def test_conflict_pr_skipped(forge):
     forge.prs = [{"number": 11, "head": {"sha": "s11"}}]
     forge.statuses = {"s11": "success"}
     forge.mergeable = {11: False}  # server reports conflict
-    assert "skipped" in mr.decide("base", "tok", forge.prs[0])
-    assert not any(m == "POST" for m, _ in forge.calls)
+    # conflict-first: the robot attempts the API update; FakeForge 409s it
+    # (genuine conflict) -> skip with reason
+    result = mr.decide("base", "tok", forge.prs[0])
+    assert "skipped" in result and "conflict" in result.lower()
+    assert not any(m == "POST" and p.endswith("/merge") for m, p in forge.calls), (
+        "must never attempt a merge on a conflicted PR"
+    )
 
 
 def test_behind_base_405_self_heals_then_merges(forge, monkeypatch):
@@ -133,6 +140,8 @@ def test_behind_base_405_self_heals_then_merges(forge, monkeypatch):
             return 200, {"state": "success"}
         if method == "GET" and path == "/pulls/12":
             return 200, {**forge.prs[0], "mergeable": True}
+        if method == "GET" and path.startswith("/compare/"):
+            return 200, {"files": [{"filename": "docs/notes.md"}]}  # docs-only → carry
         if method == "POST" and path.endswith("/merge"):
             return 405, {"message": "The head branch is behind the base branch"}
         if method == "POST" and path.endswith("/update"):
@@ -355,3 +364,70 @@ def test_transient_persisting_raises_loud(forge, monkeypatch):
     monkeypatch.setattr(mr.time, "sleep", lambda s: None)
     with pytest.raises(mr.MergeRefused, match="transient after 3 retries"):
         mr.decide("base", "tok", forge.prs[0])
+
+
+def test_conflict_first_policy_update_before_wait(forge, monkeypatch):
+    """Nick's policy: conflicts are resolved FIRST — the robot attempts the
+    API update the moment mergeable=False is observed (before any wait),
+    then a merge attempt follows within ~a minute. A PR should never sit
+    through a sleep cycle while resolvable staleness is known."""
+    forge.prs = [{"number": 40, "head": {"sha": "s40"}}]
+    calls = {"update": 0, "merge": 0}
+
+    def scripted_api(base, token, path, method="GET", data=None, ok=(200,)):
+        if method == "GET" and path.startswith("/commits/"):
+            return 200, {"state": "success"}
+        if method == "GET" and path == "/pulls/40":
+            # mergeable=False until the update lands (that's the conflict-first trigger)
+            return 200, {**forge.prs[0], "mergeable": calls["update"] > 0}
+        if method == "GET" and path.startswith("/compare/"):
+            return 200, {"files": [{"filename": "docs/notes.md"}]}  # docs-only → carry
+        if method == "POST" and path.endswith("/update"):
+            calls["update"] += 1
+            return 200, {}
+        if method == "POST" and path.endswith("/merge"):
+            calls["merge"] += 1
+            if calls["merge"] == 1:
+                return 405, {"message": "Please try again later"}
+            return 200, {}
+        raise AssertionError(f"unexpected {method} {path}")
+
+    monkeypatch.setattr(mr, "_api", scripted_api)
+    monkeypatch.setattr(mr.time, "sleep", lambda s: None)
+    line = mr.decide("base", "tok", forge.prs[0])
+    assert calls["update"] == 1, "mergeable=False must trigger immediate update (conflict-first)"
+    assert "MERGED" in line
+
+
+def test_carry_forward_checked_in_robot_update_path(forge, monkeypatch):
+    """After a self-heal update, the robot must classify the tree diff:
+    docs-only → merge on carried green; code-path touch → skip (CI must
+    re-run first). Prevents the #208 slip (merged on stale required-checks)."""
+    forge.prs = [{"number": 50, "head": {"sha": "s50"}}]
+    calls = {"update": 0, "merge": 0, "classify": 0}
+
+    def scripted_api(base, token, path, method="GET", data=None, ok=(200,)):
+        if method == "GET" and path.startswith("/commits/"):
+            return 200, {"state": "success"}
+        if method == "GET" and path == "/pulls/50":
+            return 200, {**forge.prs[0], "mergeable": calls["update"] > 0}
+        if method == "GET" and path.startswith("/compare/"):
+            return 200, {"files": [{"filename": "src/rune/x.py"}]}  # code touch → RERUN
+        if method == "POST" and path.endswith("/update"):
+            calls["update"] += 1
+            return 200, {}
+        if method == "POST" and path.endswith("/merge"):
+            calls["merge"] += 1
+            return 200, {}
+        raise AssertionError(f"unexpected {method} {path}")
+
+    monkeypatch.setattr(mr, "_api", scripted_api)
+    monkeypatch.setattr(mr.time, "sleep", lambda s: None)
+    monkeypatch.setattr(
+        mr,
+        "classify_update",
+        lambda base, token, pr, new_sha: "RERUN_REQUIRED",
+    )
+    line = mr.decide("base", "tok", forge.prs[0])
+    assert "skipped" in line and "CI re-running" in line, "code-path touch must wait for fresh CI"
+    assert calls["merge"] == 0, "no merge on unverified carry"

tests/test_p1_01_capability_authority.py

@@ -326,9 +326,9 @@ def test_default_runtime_still_exposes_full_tools():
 def test_inference_trace_schema_v3_fields():
     from rune import inference_trace as it
 
-    # P1-02 (#25) bumped the schema to v4 (additive); assert against the
-    # module constant, not a pinned literal.
-    assert it.INFERENCE_TRACE_SCHEMA_VERSION == 4
+    # P1-02 (#25) bumped the schema to v4, P1-03 (#26) to v5 (additive);
+    # assert against the module constant, not a pinned literal.
+    assert it.INFERENCE_TRACE_SCHEMA_VERSION == 5
     trace = it.InferenceTrace(
         schema_version=it.INFERENCE_TRACE_SCHEMA_VERSION,
         provider="ollama",
@@ -422,7 +422,7 @@ def test_operator_trace_record_carries_capability_fields(tmp_path, monkeypatch):
     records = it.read_traces(trace_dir=tmp_path / "home" / "state")
     assert records, "durable trace missing"
     record = records[-1]
-    assert record["schema_version"] == 4
+    assert record["schema_version"] == it.INFERENCE_TRACE_SCHEMA_VERSION
     assert record["selected_tools"] == cb.select_for_job("fix this code path")["names"]
     assert "capability_reason" in record and record["capability_reason"]
     assert record["find_capability_calls"] == 0
@@ -498,7 +498,7 @@ def test_operator_trace_record_carries_selected_tools(tmp_path, monkeypatch):
     records = it.read_traces(trace_dir=tmp_path / "home2" / "state")
     assert records
     record = records[-1]
-    assert record["schema_version"] == 4
+    assert record["schema_version"] == it.INFERENCE_TRACE_SCHEMA_VERSION
     assert record["selected_tools"] == tool_names
     assert record["capability_reason"]
     assert record["find_capability_calls"] == 0

tests/test_p1_02_capsule_authority.py

@@ -379,7 +379,9 @@ def test_inference_trace_schema_v4_fields():
     capsule_token_count with defaults; v1-v3 readers unaffected."""
     from rune import inference_trace as it
 
-    assert it.INFERENCE_TRACE_SCHEMA_VERSION == 4
+    # P1-03 (#26) bumped the schema to v5 (additive memory-selection fields);
+    # assert against the module constant, not a pinned literal.
+    assert it.INFERENCE_TRACE_SCHEMA_VERSION == 5
     trace = it.InferenceTrace(
         schema_version=it.INFERENCE_TRACE_SCHEMA_VERSION,
         provider="ollama",

tests/test_rune_operator_tools.py

@@ -149,9 +149,7 @@ class RuneOperatorToolTests(unittest.TestCase):
             request.source,
         )
         self.operator.state.tasks.append(
-            rune_operator.RuneTask(
-                "task-1", "run tests", "running", "qwen3:8b", created_at="now"
-            )
+            rune_operator.RuneTask("task-1", "run tests", "running", "qwen3:8b", created_at="now")
         )
         self.operator.state.current_task = "task-1"
         self.operator.state.pending_task_id = "task-1"
@@ -164,9 +162,7 @@ class RuneOperatorToolTests(unittest.TestCase):
         self.operator.state.pending_runtime_state = {"messages": []}
         self.operator._save()
         runtime = Mock()
-        runtime.approve.return_value = OperatorResponse(
-            success=False, model="qwen3:8b", error="terminal failure"
-        )
+        runtime.approve.return_value = OperatorResponse(success=False, model="qwen3:8b", error="terminal failure")
 
         with patch.object(self.operator, "_runtime_for_model", return_value=runtime):
             result = self.operator.resolve_pending_approval(True)
@@ -179,18 +175,14 @@ class RuneOperatorToolTests(unittest.TestCase):
     def test_approved_write_never_bypasses_prepared_runtime_authority(self):
         request = self.request()
         self.operator.state.tasks.append(
-            rune_operator.RuneTask(
-                "task-1", "write file", "running", "qwen3:8b", created_at="now"
-            )
+            rune_operator.RuneTask("task-1", "write file", "running", "qwen3:8b", created_at="now")
         )
         self.operator.state.current_task = "task-1"
         self.operator.state.pending_task_id = "task-1"
         self.operator.state.pending_request = asdict(request)
         self.operator.state.pending_runtime_state = {"messages": []}
         runtime = Mock()
-        runtime.approve.return_value = OperatorResponse(
-            success=True, user_text="Written.", model="qwen3:8b"
-        )
+        runtime.approve.return_value = OperatorResponse(success=True, user_text="Written.", model="qwen3:8b")
 
         with (
             patch.object(self.operator, "_runtime_for_model", return_value=runtime),
@@ -202,15 +194,11 @@ class RuneOperatorToolTests(unittest.TestCase):
         self.assertTrue(result["success"])
 
     def test_failed_runtime_preserves_failure_envelope_and_persists_failed_task(self):
-        request = ToolRequest(
-            "call-2", "read_file", {"path": "/tmp/a"}, "ollama-native"
-        )
+        request = ToolRequest("call-2", "read_file", {"path": "/tmp/a"}, "ollama-native")
         response = OperatorResponse(
             success=False,
             thinking=["The response was malformed."],
-            tool_events=[
-                ToolEvent("error", request, {"success": False, "error": "bad"})
-            ],
+            tool_events=[ToolEvent("error", request, {"success": False, "error": "bad"})],
             model="qwen3:8b",
             recall_count=1,
             override_tier=1,
@@ -235,9 +223,7 @@ class RuneOperatorToolTests(unittest.TestCase):
         self.operator.state.pending_task_id = "stale-task"
         self.operator.state.pending_request = {"name": "write_file"}
         self.operator.state.pending_runtime_state = {"messages": []}
-        response = OperatorResponse(
-            success=True, user_text="Finished.", model="qwen3:8b"
-        )
+        response = OperatorResponse(success=True, user_text="Finished.", model="qwen3:8b")
 
         with patch.object(self.operator, "_execute_on_model", return_value=response):
             self.operator.send_task("new task", "qwen3:8b")
@@ -249,9 +235,7 @@ class RuneOperatorToolTests(unittest.TestCase):
 
     def test_failed_tool_events_override_success_and_persist_structured_failure(self):
         for index, event_kind in enumerate(("error", "failed"), start=3):
-            request = ToolRequest(
-                f"call-{index}", "read_file", {"path": "/tmp/a"}, "ollama-native"
-            )
+            request = ToolRequest(f"call-{index}", "read_file", {"path": "/tmp/a"}, "ollama-native")
             response = OperatorResponse(
                 success=True,
                 user_text="I finished successfully.",
@@ -278,9 +262,7 @@ class RuneOperatorToolTests(unittest.TestCase):
             self.assertNotEqual(result["task"]["result"], response.user_text)
 
         persisted = json.loads(self.storage.read_text())
-        self.assertEqual(
-            [task["status"] for task in persisted["tasks"]], ["failed", "failed"]
-        )
+        self.assertEqual([task["status"] for task in persisted["tasks"]], ["failed", "failed"])
         self.assertEqual(
             [task["result"] for task in persisted["tasks"]],
             ["permission denied", "permission denied"],
@@ -288,19 +270,13 @@ class RuneOperatorToolTests(unittest.TestCase):
 
     def test_valid_local_final_uses_runtime_with_context_and_records_success(self):
         runtime = Mock()
-        runtime.run.return_value = OperatorResponse(
-            success=True, user_text="Final answer", model="qwen3:8b"
-        )
+        runtime.run.return_value = OperatorResponse(success=True, user_text="Final answer", model="qwen3:8b")
         profile = SimpleNamespace(safety_tier=2)
         memory_db = Mock()
 
         with (
-            patch.object(
-                rune_operator, "LocalToolRuntime", return_value=runtime, create=True
-            ) as runtime_type,
-            patch.object(
-                rune_operator, "OllamaChatTransport", return_value=Mock(), create=True
-            ),
+            patch.object(rune_operator, "LocalToolRuntime", return_value=runtime, create=True) as runtime_type,
+            patch.object(rune_operator, "OllamaChatTransport", return_value=Mock(), create=True),
             patch.object(
                 rune_operator.model_profiles,
                 "get_or_create_profile",
@@ -317,16 +293,20 @@ class RuneOperatorToolTests(unittest.TestCase):
                 return_value=SimpleNamespace(token_count=10),
             ),
             patch("rune.memory_store.connect", return_value=memory_db),
-            patch(
-                "rune.memory_store.cached_recall",
-                return_value=[{"text": "remembered fact"}],
+            # P1-03 (#26): the memory seam is now the selector's store source.
+            patch.dict(
+                "rune.memory_selector.DEFAULT_SOURCES",
+                {
+                    "store": lambda query, types, budget, min_score: [
+                        {"text": "remembered fact", "score": 0.9, "kind": "store"}
+                    ]
+                },
+                clear=True,
             ),
             patch.object(self.operator, "_log_token_usage"),
         ):
             response = self.operator._execute_on_model(
-                rune_operator.RuneTask(
-                    "task-1", "inspect this", assigned_model="qwen3:8b"
-                ),
+                rune_operator.RuneTask("task-1", "inspect this", assigned_model="qwen3:8b"),
                 "qwen3:8b",
             )
 
@@ -350,9 +330,7 @@ class RuneOperatorToolTests(unittest.TestCase):
     def test_terminal_failures_record_model_failure_while_pending_is_not_counted(self):
         request = self.request()
         cases = (
-            OperatorResponse(
-                success=False, model="qwen3:8b", error="iteration limit reached"
-            ),
+            OperatorResponse(success=False, model="qwen3:8b", error="iteration limit reached"),
             OperatorResponse(
                 success=True,
                 model="qwen3:8b",
@@ -363,9 +341,7 @@ class RuneOperatorToolTests(unittest.TestCase):
                 success=True,
                 user_text="Could not finish.",
                 model="qwen3:8b",
-                tool_events=[
-                    ToolEvent("error", request, {"success": False, "error": "denied"})
-                ],
+                tool_events=[ToolEvent("error", request, {"success": False, "error": "denied"})],
             ),
         )
 
@@ -378,13 +354,9 @@ class RuneOperatorToolTests(unittest.TestCase):
                     return_value=runtime_response,
                     create=True,
                 ),
-                patch.object(
-                    rune_operator.model_profiles, "record_interaction"
-                ) as record,
+                patch.object(rune_operator.model_profiles, "record_interaction") as record,
             ):
-                response = self.operator._finalize_runtime_response(
-                    runtime_response, "qwen3:8b", "inspect this", 2, 1
-                )
+                response = self.operator._finalize_runtime_response(runtime_response, "qwen3:8b", "inspect this", 2, 1)
                 self.assertIs(response, runtime_response)
                 if runtime_response.pending_approval is not None:
                     record.assert_not_called()
@@ -432,9 +404,7 @@ class RuneOperatorToolTests(unittest.TestCase):
             ),
             patch.object(rune_operator.model_profiles, "record_interaction") as record,
         ):
-            response = self.operator._execute_on_model(
-                rune_operator.RuneTask("task-1", "inspect this"), "qwen3:8b"
-            )
+            response = self.operator._execute_on_model(rune_operator.RuneTask("task-1", "inspect this"), "qwen3:8b")
 
         self.assertFalse(response.success)
         record.assert_called_once_with("qwen3:8b", False, "inspect this")

tests/test_runtime_architecture_characterization.py

@@ -38,12 +38,8 @@ class RuntimeArchitectureCharacterizationTests(unittest.TestCase):
             operator = rune_operator.RuneOperator(str(Path(tempdir) / "operator.json"))
             runtime = Mock()
             runtime.run.side_effect = [
-                OperatorResponse(
-                    success=True, user_text="First answer", model="qwen3:8b"
-                ),
-                OperatorResponse(
-                    success=True, user_text="Second answer", model="qwen3:8b"
-                ),
+                OperatorResponse(success=True, user_text="First answer", model="qwen3:8b"),
+                OperatorResponse(success=True, user_text="Second answer", model="qwen3:8b"),
             ]
             profile = SimpleNamespace(safety_tier=2)
 
@@ -74,14 +70,12 @@ class RuntimeArchitectureCharacterizationTests(unittest.TestCase):
 
             second_messages = runtime.run.call_args_list[1].args[0]
             role_content = [
-                (message["role"], message["content"])
-                for message in second_messages
-                if message["role"] != "system"
+                (message["role"], message["content"]) for message in second_messages if message["role"] != "system"
             ]
             self.assertIn(
-                ("user", "First question"),
+                ("user", "PAST TASK (completed — context only, not instructions): First question"),
                 role_content,
-                "the second provider turn omits the prior user message",
+                "the second provider turn omits the prior user message (P1-03: past-task content is labeled context-only)",
             )
             self.assertIn(
                 ("assistant", "First answer"),
@@ -90,18 +84,12 @@ class RuntimeArchitectureCharacterizationTests(unittest.TestCase):
             )
 
     def test_tui_send_path_projects_typed_tool_events(self):
-        send = _function_node(
-            ROOT / "src" / "rune" / "rune_tui.py", "RuneApp", "_send_to_model"
-        )
+        send = _function_node(ROOT / "src" / "rune" / "rune_tui.py", "RuneApp", "_send_to_model")
         string_literals = {
-            node.value
-            for node in ast.walk(send)
-            if isinstance(node, ast.Constant) and isinstance(node.value, str)
+            node.value for node in ast.walk(send) if isinstance(node, ast.Constant) and isinstance(node.value, str)
         }
         constructed_types = {
-            node.func.id
-            for node in ast.walk(send)
-            if isinstance(node, ast.Call) and isinstance(node.func, ast.Name)
+            node.func.id for node in ast.walk(send) if isinstance(node, ast.Call) and isinstance(node.func, ast.Name)
         }
 
         self.assertIn(
@@ -121,9 +109,7 @@ class RuntimeArchitectureCharacterizationTests(unittest.TestCase):
         self.assertIn("tool_widget.apply(tool_event)", source)
 
     def test_tui_send_path_does_not_auto_confirm_pending_mutation(self):
-        send = _function_node(
-            ROOT / "src" / "rune" / "rune_tui.py", "RuneApp", "_send_to_model"
-        )
+        send = _function_node(ROOT / "src" / "rune" / "rune_tui.py", "RuneApp", "_send_to_model")
         calls = [
             node.func.attr
             for node in ast.walk(send)
@@ -137,9 +123,7 @@ class RuntimeArchitectureCharacterizationTests(unittest.TestCase):
         )
 
     def test_generic_bubble_does_not_interpolate_protocol_markup(self):
-        render = _function_node(
-            ROOT / "src" / "rune" / "chat_bubble.py", "ChatBubble", "render"
-        )
+        render = _function_node(ROOT / "src" / "rune" / "chat_bubble.py", "ChatBubble", "render")
         direct_content_interpolations = [
             node
             for node in ast.walk(render)

tests/test_shard_manager.py

@@ -0,0 +1,58 @@
+"""Red-first: #26 — shard connections must load sqlite-vec per connection.
+get_shard() previously returned a raw connection: entries_vec invisible,
+recall() crashed with 'no such table' in clean environments (CI canary)."""
+
+from __future__ import annotations
+
+import importlib.util
+import sqlite3
+import sys
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(ROOT / "src"))
+
+SPEC = importlib.util.spec_from_file_location("shard_manager", ROOT / "src" / "rune" / "shard_manager.py")
+sm_mod = importlib.util.module_from_spec(SPEC)
+sys.modules["shard_manager"] = sm_mod  # register so importlib.reload() works
+SPEC.loader.exec_module(sm_mod)
+
+
+@pytest.fixture
+def shard_env(tmp_path, monkeypatch):
+    """Point ShardManager's SHARD_DIR at a temp home — no importlib.reload
+    (module-constant patching is deterministic and side-effect free)."""
+    shard_dir = tmp_path / "shards"
+    shard_dir.mkdir()
+    monkeypatch.setattr(sm_mod, "SHARD_DIR", str(shard_dir))
+    # fresh index per test
+    monkeypatch.setattr(sm_mod.ShardManager, "_load_index", lambda self: {})
+    monkeypatch.setattr(sm_mod.ShardManager, "_save_index", lambda self: None)
+    yield sm_mod
+
+
+def test_get_shard_connection_can_query_entries_vec(shard_env):
+    """The exact CI canary failure: recall() SELECT ... FROM entries_vec on a
+    shard connection must not raise 'no such table'."""
+    mgr = sm_mod.ShardManager()
+    db = mgr.get_shard("guitar academy")
+    rows = db.execute("SELECT rowid FROM entries_vec LIMIT 1").fetchall()
+    assert rows == []  # table exists and is queryable — empty is fine
+
+
+def test_create_shard_schema_includes_entries_vec(shard_env):
+    mgr = sm_mod.ShardManager()
+    path = mgr.create_shard("test topic")
+    db = sqlite3.connect(path)
+    tables = {r[0] for r in db.execute("SELECT name FROM sqlite_master WHERE type='table'")}
+    assert "entries_vec" in tables, "shard schema must include the vec table (recall routes here)"
+
+
+def test_get_shard_reuses_existing_shard_file(shard_env):
+    mgr = sm_mod.ShardManager()
+    path1 = mgr.create_shard("reuse")
+    mgr.index["reuse"] = path1
+    db = mgr.get_shard("reuse")
+    db.execute("SELECT 1 FROM entries_vec").fetchall()  # no raise