Merge remote-tracking branch 'gitea/master' into p1-03-memory-selection

Gate finding 8: the canary's trivial-task check ran on an EMPTY store —
the capsule-level zero-memory proof was vacuous. It now seeds a
high-importance fact distractor plus a personality row before
assembling; the trivial task must still produce zero memory blocks and
zero memory tokens.

Also lands the discharge record: docs updated (floor-in-every-mode
claim, known ranking bias, typed persona lane, write-time memory_type,
migration-time backfill) and the plan's Completion Record gains the
Discharge Record section mapping every gate finding to its commit and
red-first evidence.

.gitea/workflows/ci.yml

@@ -91,6 +91,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 \

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.

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.

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

@@ -81,6 +81,7 @@ def _build_model_router():
         record_failure("model_routing", exc, tier=1, context="build_model_router")
         return None
 
+
 try:
     try:
         from .local_tool_runtime import LocalToolRuntime, OllamaChatTransport
@@ -131,6 +132,7 @@ try:
         TaskStateRepository,
         migrate_operator_tasks,
     )
+
     try:
         from .orchestration.event_store import DurableEventStore
     except ImportError:  # TUI top-level deploy
@@ -208,11 +210,7 @@ def _memory_query_from_trace(trace: Dict[str, Any], task_id: str) -> MemoryQuery
     Admitted blocks become the {block_name: tokens} map; dropped blocks ride
     through with their recorded reasons. Observable selection state only.
     """
-    admitted = {
-        block["name"]: block["tokens"]
-        for block in trace.get("blocks", [])
-        if block.get("admitted")
-    }
+    admitted = {block["name"]: block["tokens"] for block in trace.get("blocks", []) if block.get("admitted")}
     dropped = [
         {"name": block.get("name", ""), "tokens": block.get("tokens", 0)}
         for block in trace.get("blocks", [])
@@ -261,9 +259,7 @@ def _observations_from_degraded(degraded: list[str], task_id: str) -> list:
     ]
 
 
-def _verification_and_outcome(
-    response: OperatorResponse, task_id: str, degraded: list[str]
-) -> tuple:
+def _verification_and_outcome(response: OperatorResponse, task_id: str, degraded: list[str]) -> tuple:
     """VerificationResult + linked Outcome for the final response (#30).
 
     Method contract (frozen): 'exit_code' when a run_command tool event
@@ -290,9 +286,7 @@ def _verification_and_outcome(
             evidence = _bounded(result.get("result") or "run_tests passed")
             break
 
-    has_tool_failure = any(
-        event.kind in {"error", "failed"} for event in response.tool_events
-    )
+    has_tool_failure = any(event.kind in {"error", "failed"} for event in response.tool_events)
     passed = bool(response.success) and not has_tool_failure
     if method == "none":
         # No objective evidence ran — never self-report (plan: the
@@ -320,6 +314,7 @@ def _verification_and_outcome(
     )
     return verification, outcome
 
+
 #: A message that looks like a confirmation/follow-up of the active task
 #: gets amended onto it instead of minting a new task (P0-02 #18).
 _FOLLOW_UP_RE = re.compile(
@@ -439,9 +434,7 @@ class RuneOperator:
         self._events_store = DurableEventStore(root / "events.jsonl")
         self._hot_store = AtomicStore(root / "hot")
         self._task_state = TaskStateRepository(self._events_store, self._hot_store)
-        self._conversation = ConversationStateRepository(
-            self._events_store, self._hot_store, tasks=self._task_state
-        )
+        self._conversation = ConversationStateRepository(self._events_store, self._hot_store, tasks=self._task_state)
         try:
             self._migration_report = migrate_operator_tasks(
                 Path(self.storage_path), self._task_state, self._conversation
@@ -449,9 +442,7 @@ class RuneOperator:
         except Exception as exc:  # P0-06 contract: degrade loudly, never silently
             import logging
 
-            logging.getLogger(__name__).warning(
-                "Legacy operator.json migration failed: %s", exc
-            )
+            logging.getLogger(__name__).warning("Legacy operator.json migration failed: %s", exc)
             self._migration_report = None
 
     def _load(self):
@@ -466,13 +457,7 @@ class RuneOperator:
                     current_task=data.get("current_task"),
                     active_model=data.get("active_model"),
                     tasks=[
-                        RuneTask(
-                            **{
-                                k: v
-                                for k, v in t.items()
-                                if k in RuneTask.__dataclass_fields__
-                            }
-                        )
+                        RuneTask(**{k: v for k, v in t.items() if k in RuneTask.__dataclass_fields__})
                         for t in data.get("tasks", [])
                     ],
                     pending_task_id=data.get("pending_task_id"),
@@ -482,9 +467,7 @@ class RuneOperator:
             except Exception as e:
                 import logging
 
-                logging.getLogger(__name__).warning(
-                    "Failed to load operator state: %s", e
-                )
+                logging.getLogger(__name__).warning("Failed to load operator state: %s", e)
 
     def _save(self):
         """Save Rune's state to storage."""
@@ -499,15 +482,9 @@ class RuneOperator:
             "role": self.state.role,
             "current_task": self.state.current_task,
             "active_model": self.state.active_model,
-            "tasks_pending": len(
-                [t for t in self.state.tasks if t.status == "pending"]
-            ),
-            "tasks_running": len(
-                [t for t in self.state.tasks if t.status == "running"]
-            ),
-            "tasks_completed": len(
-                [t for t in self.state.tasks if t.status == "completed"]
-            ),
+            "tasks_pending": len([t for t in self.state.tasks if t.status == "pending"]),
+            "tasks_running": len([t for t in self.state.tasks if t.status == "running"]),
+            "tasks_completed": len([t for t in self.state.tasks if t.status == "completed"]),
         }
 
     def send_task(self, description: str, model: str | None = None, channel: str = OPERATOR_CHANNEL) -> Dict[str, Any]:
@@ -519,9 +496,7 @@ class RuneOperator:
         a fresh task records its parent for lineage.
         """
         active = self._task_state.get_active(channel)
-        if active is not None and (
-            active.status == "running" or _is_follow_up(description)
-        ):
+        if active is not None and (active.status == "running" or _is_follow_up(description)):
             state_task = self._task_state.amend(active.task_id, description)
         else:
             latest = self._task_state.latest(channel)
@@ -577,9 +552,7 @@ class RuneOperator:
             if decision is not None:
                 model = decision.model
                 routing_reason = (
-                    decision.reason
-                    if decision.reason.startswith("router:")
-                    else f"router: {decision.reason}"
+                    decision.reason if decision.reason.startswith("router:") else f"router: {decision.reason}"
                 )
                 routing_mode = "cloud" if decision.provider not in ("ollama", "local") else "local"
             else:
@@ -602,11 +575,7 @@ class RuneOperator:
         )
 
         failed_event = next(
-            (
-                event
-                for event in result.tool_events
-                if event.kind in {"error", "failed"}
-            ),
+            (event for event in result.tool_events if event.kind in {"error", "failed"}),
             None,
         )
         if failed_event is not None:
@@ -645,9 +614,7 @@ class RuneOperator:
             task.result = result.user_text
             task.completed_at = datetime.now(timezone.utc).isoformat()
             self.state.current_task = None
-            self._task_state.transition(
-                task.task_id, "completed", outcome_ref="response:" + task.task_id
-            )
+            self._task_state.transition(task.task_id, "completed", outcome_ref="response:" + task.task_id)
             self._conversation.append(
                 ConversationMessageRecord(
                     role="assistant",
@@ -680,9 +647,7 @@ class RuneOperator:
             if "status" not in obj or obj.get("task_id") != task.task_id:
                 continue
             obj["status"] = task.status
-            verification = next(
-                (item for item in result.runtime_objects if "method" in item), None
-            )
+            verification = next((item for item in result.runtime_objects if "method" in item), None)
             obj["verification_id"] = (
                 verification["verification_id"]
                 if verification is not None and verification.get("passed") and task.status == "completed"
@@ -786,57 +751,67 @@ class RuneOperator:
             self._last_capability_selection = selection
             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:]
+    def _conversation_messages(self, current_task_id: str, limit: int = 8) -> list[dict]:
+        """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."""
         task = next(
-            (
-                item
-                for item in self.state.tasks
-                if item.task_id == self.state.pending_task_id
-            ),
+            (item for item in self.state.tasks if item.task_id == self.state.pending_task_id),
             None,
         )
         request = self.state.pending_request
         pending_state = self.state.pending_runtime_state
         if task is None or request is None or pending_state is None:
-            return self._response_envelope(
-                OperatorResponse(success=False, error="no pending approval"), task
-            )
+            return self._response_envelope(OperatorResponse(success=False, error="no pending approval"), task)
 
         model = task.assigned_model or self.state.active_model or ""
         runtime = self._runtime_for_model(model, task.description)
-        result = (
-            runtime.approve(pending_state)
-            if approved
-            else runtime.reject(pending_state)
-        )
+        result = runtime.approve(pending_state) if approved else runtime.reject(pending_state)
         result = self._finalize_runtime_response(
             result, model, task.description, result.override_tier, result.recall_count
         )
         failed_event = next(
-            (
-                event
-                for event in result.tool_events
-                if event.kind in {"error", "failed"}
-            ),
+            (event for event in result.tool_events if event.kind in {"error", "failed"}),
             None,
         )
         if failed_event is not None:
@@ -872,9 +847,7 @@ class RuneOperator:
         """
         try:
             runtime_objects: list = []
-            for observation in _observations_from_degraded(
-                getattr(response, "degraded", None) or [], task_id
-            ):
+            for observation in _observations_from_degraded(getattr(response, "degraded", None) or [], task_id):
                 runtime_objects.append(observation.to_dict())
             if response.pending_approval is None:
                 verification, outcome = _verification_and_outcome(
@@ -962,26 +935,18 @@ class RuneOperator:
         return [m.to_dict() for m in self._conversation.window(channel, n=n)]
 
     @staticmethod
-    def _response_envelope(
-        result: OperatorResponse, task: Optional[RuneTask]
-    ) -> Dict[str, Any]:
+    def _response_envelope(result: OperatorResponse, task: Optional[RuneTask]) -> Dict[str, Any]:
         envelope = asdict(result)
         envelope["task"] = task.to_dict() if task is not None else None
         envelope["response"] = result.user_text
         if result.pending_approval is not None:
             pending_event = next(
-                (
-                    event
-                    for event in reversed(result.tool_events)
-                    if event.kind == "pending_approval"
-                ),
+                (event for event in reversed(result.tool_events) if event.kind == "pending_approval"),
                 None,
             )
             envelope["needs_confirm"] = True
             envelope["tool_call"] = asdict(result.pending_approval)
-            envelope["tool_result"] = (
-                dict(pending_event.result or {}) if pending_event else {}
-            )
+            envelope["tool_result"] = dict(pending_event.result or {}) if pending_event else {}
         return envelope
 
     def _fallback_model(self, description: str) -> str:
@@ -994,22 +959,15 @@ class RuneOperator:
         desc_lower = description.lower()
 
         # Quick tasks — laptop model
-        if any(
-            word in desc_lower for word in ["hello", "hi", "thanks", "ok", "status"]
-        ):
+        if any(word in desc_lower for word in ["hello", "hi", "thanks", "ok", "status"]):
             return "qwen2.5:7b"
 
         # Coding tasks — local coding model
-        if any(
-            word in desc_lower for word in ["fix", "code", "debug", "test", "refactor"]
-        ):
+        if any(word in desc_lower for word in ["fix", "code", "debug", "test", "refactor"]):
             return "qwen3:14b"
 
         # Complex tasks — strong local model
-        if any(
-            word in desc_lower
-            for word in ["design", "architecture", "research", "analyze"]
-        ):
+        if any(word in desc_lower for word in ["design", "architecture", "research", "analyze"]):
             return "hf.co/DavidAU/Qwen3.6-27B-Fable-Fusion-711-Uncensored-Heretic-NM-DAU-NEO-MAX-MTP-GGUF:IQ2_M"
 
         # Default to laptop model
@@ -1048,14 +1006,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)
@@ -1163,315 +1120,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)."""
@@ -1491,11 +1199,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
@@ -1553,20 +1269,24 @@ I have tools to read, write, edit, search, lint, test, format, and commit code.
             if not hasattr(broker, "assemble"):
                 from types import SimpleNamespace as _NS
 
-                broker = _NS(assemble=lambda *a, **k: (
-                    _NS(token_count=0, blocks=[], active_objective="", to_messages=lambda: []),
-                    {},
-                ))
+                broker = _NS(
+                    assemble=lambda *a, **k: (
+                        _NS(token_count=0, blocks=[], active_objective="", to_messages=lambda: []),
+                        {},
+                    )
+                )
             assemble = getattr(broker, "assemble", None)
             if assemble is None:
                 # Degrade-not-crash: a stand-in broker without assemble gets an
                 # empty capsule + trace (scripted tests, Mock runtimes).
                 from types import SimpleNamespace as _NS
 
-                broker = _NS(assemble=lambda *a, **k: (
-                    _NS(token_count=0, blocks=[], active_objective="", to_messages=lambda: []),
-                    {},
-                ))
+                broker = _NS(
+                    assemble=lambda *a, **k: (
+                        _NS(token_count=0, blocks=[], active_objective="", to_messages=lambda: []),
+                        {},
+                    )
+                )
             capsule, trace = broker.assemble(task, session, model_id=model)  # type: ignore[attr-defined]
             self._last_trace = trace
             for subsystem in trace.get("degraded", []):
@@ -1637,9 +1357,7 @@ I have tools to read, write, edit, search, lint, test, format, and commit code.
                     "output": getattr(transport, "usage_output_tokens", 0),
                     "duration_ns": getattr(transport, "usage_total_duration_ns", 0),
                 }
-                prev = _prev_usage.setdefault(
-                    id(transport), {"calls": 0, "prompt": 0, "output": 0, "duration_ns": 0}
-                )
+                prev = _prev_usage.setdefault(id(transport), {"calls": 0, "prompt": 0, "output": 0, "duration_ns": 0})
                 for key, value in cur.items():
                     if isinstance(value, int) and not isinstance(value, bool):
                         run_usage[key] += max(0, value - prev[key])
@@ -1667,9 +1385,7 @@ I have tools to read, write, edit, search, lint, test, format, and commit code.
                         for trace_block in trace["blocks"]:
                             if trace_block.get("name") == "user" and trace_block.get("admitted"):
                                 trace_block["tokens"] = block.tokens
-                        capsule.token_count = sum(
-                            block.tokens for block in capsule.blocks if hasattr(block, "tokens")
-                        )
+                        capsule.token_count = sum(block.tokens for block in capsule.blocks if hasattr(block, "tokens"))
                 messages = capsule.to_messages()
                 # P1-02 (#25) acceptance: the log must capture the EXACT
                 # payload the runtime receives — capture BEFORE run, assert
@@ -1727,14 +1443,10 @@ I have tools to read, write, edit, search, lint, test, format, and commit code.
                         "active_objective": capsule.active_objective,
                         "blocks": [block.to_dict() for block in capsule.blocks],
                     }
-                    response.pending_state["exposed_tools"] = _sorted_tool_names(
-                        getattr(runtime, "tool_names", ())
-                    )
+                    response.pending_state["exposed_tools"] = _sorted_tool_names(getattr(runtime, "tool_names", ()))
 
                 terminal_refusal = (
-                    response.success
-                    and response.pending_approval is None
-                    and override.is_refusal(response.user_text)
+                    response.success and response.pending_approval is None and override.is_refusal(response.user_text)
                 )
                 if terminal_refusal and current_tier < max_tier:
                     current_tier += 1
@@ -1775,16 +1487,12 @@ I have tools to read, write, edit, search, lint, test, format, and commit code.
                     runtime_objects: list = []
                     runtime_objects.append(_memory_query_from_trace(trace, task_id).to_dict())
                     for compaction_event in trace.get("compaction", []):
-                        runtime_objects.append(
-                            _observation_from_compaction(compaction_event, task_id).to_dict()
-                        )
+                        runtime_objects.append(_observation_from_compaction(compaction_event, task_id).to_dict())
                     response.runtime_objects = runtime_objects
                     # Shared emission (verification/outcome/degraded) — same
                     # helper the approval-resume path uses (quality I-2).
                     self._attach_runtime_objects(response, task_id)
-                    response.runtime_objects = (
-                        runtime_objects + list(response.runtime_objects)
-                    )
+                    response.runtime_objects = runtime_objects + list(response.runtime_objects)
                 except Exception as exc:
                     try:
                         from .subsystem_health import record_failure
@@ -1890,9 +1598,7 @@ I have tools to read, write, edit, search, lint, test, format, and commit code.
                         # P1-02 (#25) schema v4: capsule fidelity fields —
                         # the logged messages EXACTLY equal the runtime.run
                         # payload (captured pre-run on the mutated capsule).
-                        capsule_messages=[
-                            dict(m) for m in _sent if isinstance(m, dict)
-                        ],
+                        capsule_messages=[dict(m) for m in _sent if isinstance(m, dict)],
                         capsule_content_hashes=capsule_hashes,
                         capsule_token_count=(
                             int(capsule.token_count)
@@ -1900,6 +1606,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 —
@@ -1911,9 +1623,7 @@ I have tools to read, write, edit, search, lint, test, format, and commit code.
                     except ImportError:  # TUI top-level deploy (#168)
                         from rune.model_routing import update_profiles_from_traces
 
-                    merged = update_profiles_from_traces(
-                        self._routing_profiles_path(), [asdict(trace_record)]
-                    )
+                    merged = update_profiles_from_traces(self._routing_profiles_path(), [asdict(trace_record)])
                     from .model_routing import save_profiles as _save_profiles
 
                     _save_profiles(self._routing_profiles_path(), merged)
@@ -1940,7 +1650,9 @@ I have tools to read, write, edit, search, lint, test, format, and commit code.
 
         # Unreachable in practice (the tier loop always returns or raises),
         # but the declared contract requires a value on all code paths.
-        return OperatorResponse(success=False, model=model, error="task execution ended without a result", degraded=sorted(set(degraded)))
+        return OperatorResponse(
+            success=False, model=model, error="task execution ended without a result", degraded=sorted(set(degraded))
+        )
 
     @staticmethod
     def _finalize_runtime_response(
@@ -1956,9 +1668,7 @@ I have tools to read, write, edit, search, lint, test, format, and commit code.
         response.model = response.model or model
         response.override_tier = override_tier
         response.recall_count = recall_count
-        has_tool_failure = any(
-            event.kind in {"error", "failed"} for event in response.tool_events
-        )
+        has_tool_failure = any(event.kind in {"error", "failed"} for event in response.tool_events)
         valid_final = (
             response.success
             and response.pending_approval is None
@@ -1980,10 +1690,7 @@ I have tools to read, write, edit, search, lint, test, format, and commit code.
 
         # Rotate if file exceeds 10MB
         try:
-            if (
-                os.path.exists(log_path)
-                and os.path.getsize(log_path) > 10 * 1024 * 1024
-            ):
+            if os.path.exists(log_path) and os.path.getsize(log_path) > 10 * 1024 * 1024:
                 rotated = log_path + ".old"
                 os.replace(log_path, rotated)
         except OSError:

src/rune/sleep_time.py

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

src/rune/warm_router.py

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

tests/test_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_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)