feat(p1-03): memory budget, trace v5, canary (#26)

.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,112 @@
+# 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; self-scored sources are floored in the selector.
+
+- 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: 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`; 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).
+
+## 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.

src/rune/canary_p1_03.py

@@ -0,0 +1,257 @@
+#!/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."""
+    _no_embed()
+    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/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.
 
@@ -514,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.
 
@@ -522,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):
@@ -533,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,
@@ -554,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
@@ -568,6 +609,8 @@ class ContextBroker:
             "blocks": trace_blocks,
             "admitted_total": used,
             "dropped": dropped,
+            "memory_budget": memory_budget,
+            "memory_used": memory_used,
         }
 
     def assemble(
@@ -590,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):
@@ -627,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
 
@@ -648,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/rune_operator.py

@@ -1140,6 +1140,12 @@ I have tools to read, write, edit, search, lint, test, format, and commit code.
         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["memory_candidates"] = 0
+                session["memory_selected"] = 0
+                session["memory_abstained"] = True
+                session["memory_types_queried"] = []
             return None
         description = getattr(task, "description", "") or ""
         try:
@@ -1153,7 +1159,18 @@ I have tools to read, write, edit, search, lint, test, format, and commit code.
             record_failure("memory_selector", exc, tier=1)
             if isinstance(session, dict):
                 session.setdefault("degraded", []).append("memory_selector")
+                session["memory_candidates"] = 0
+                session["memory_selected"] = 0
+                session["memory_abstained"] = True
+                session["memory_types_queried"] = []
             return None
+        # 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()
         ]
@@ -1589,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 —

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_p1_03_memory_selection.py

@@ -756,3 +756,221 @@ def test_selector_unavailable_means_zero_memory_blocks(tmp_path):
         task = mock.Mock(description="kubernetes pods restart automatically")
         task.task_id = "t1"
         assert op._contributor_memory(task, {}) is None
+
+
+# ---------------------------------------------------------------------------
+# Task 5 — memory sub-budget + trace fields + canary + docs
+# ---------------------------------------------------------------------------
+
+
+def test_memory_budget_default_is_min_cap_window_eighth():
+    from rune.context_broker import default_memory_budget
+
+    assert default_memory_budget(9600) == 1200, "min(1200, window // 8)"
+    assert default_memory_budget(3200) == 400, "small windows scale down"
+    assert default_memory_budget(None) == 1200, "no window -> cap only"
+
+
+def test_memory_blocks_evicted_beyond_sub_budget(tmp_path):
+    """ContextBroker owns the final memory budget: memory blocks beyond the
+    sub-budget are dropped even when the GLOBAL budget could fit them."""
+    from rune.context_broker import ContributionBlock, ContextBroker
+
+    broker = ContextBroker(
+        budget_resolver=lambda window: 10_000,
+        estimator=lambda text: len(text) // 4,
+    )
+    big_memory = "m" * 8000  # ~2000 tokens > the 1200-token sub-budget
+    broker.register_contributor(
+        "memory",
+        lambda task, session: ContributionBlock("memory", 70, big_memory, 0, "test"),
+        2,
+    )
+    broker.register_contributor(
+        "notes",
+        lambda task, session: ContributionBlock("notes", 65, "n" * 400, 0, "test"),
+        2,
+    )
+    task = mock.Mock(description="kubernetes pods restart")
+    task.task_id = "t1"
+    _capsule, trace = broker.assemble(task, {}, budget_tokens=10_000)
+    memory_records = [b for b in trace["blocks"] if b["name"] == "memory"]
+    assert memory_records and memory_records[0]["admitted"] is False, (
+        "a memory block above the sub-budget must be evicted"
+    )
+    assert "memory budget" in memory_records[0]["reason"]
+    assert trace["memory_tokens"] == 0
+    notes_records = [b for b in trace["blocks"] if b["name"] == "notes"]
+    assert notes_records and notes_records[0]["admitted"] is True, (
+        "non-memory blocks are unaffected by the memory sub-budget"
+    )
+
+
+def test_memory_blocks_within_sub_budget_admitted(tmp_path):
+    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" * 400, 0, "test"),
+        2,
+    )
+    task = mock.Mock(description="kubernetes pods restart")
+    task.task_id = "t1"
+    _capsule, trace = broker.assemble(task, {}, budget_tokens=10_000)
+    memory_records = [b for b in trace["blocks"] if b["name"] == "memory"]
+    assert memory_records[0]["admitted"] is True
+    assert trace["memory_tokens"] > 0
+    assert trace["memory_budget"] == 1200
+
+
+def test_lowest_priority_memory_block_evicted_first(tmp_path):
+    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, "a" * 4800, 0, "test"),
+        2,
+    )
+    broker.register_contributor(
+        "persona_memories",
+        lambda task, session: ContributionBlock("persona_memories", 68, "b" * 4800, 0, "test"),
+        2,
+    )
+    task = mock.Mock(description="kubernetes pods restart")
+    task.task_id = "t1"
+    _capsule, trace = broker.assemble(task, {}, budget_tokens=10_000)
+    by_name = {b["name"]: b for b in trace["blocks"]}
+    assert by_name["memory"]["admitted"] is True, "higher-priority memory fits the sub-budget"
+    assert by_name["persona_memories"]["admitted"] is False, "lowest-priority memory evicted first"
+
+
+def test_trace_v5_memory_fields_additive_defaults():
+    """P1-03 additive fields (schema v5 — v4 was taken by P1-02 capsule
+    fields): defaults keep v1-v4 readers and rows fully compatible."""
+    from rune import inference_trace as it
+
+    assert it.INFERENCE_TRACE_SCHEMA_VERSION == 5
+    trace = it.InferenceTrace(
+        schema_version=it.INFERENCE_TRACE_SCHEMA_VERSION,
+        provider="ollama",
+        model="m",
+        task_id="t",
+        channel="c",
+        resolved_context_window=0,
+        input_budget=0,
+        model_calls=0,
+        prompt_tokens_estimated=0,
+        prompt_tokens_served=0,
+        output_tokens=0,
+        latency_ms=0.0,
+        model_time_ms=0.0,
+        per_block_tokens={},
+        dropped_blocks=[],
+        tools_exposed=[],
+        memories_exposed=[],
+        memory_tokens=0,
+        degraded=[],
+        cache=None,
+        cost_usd=None,
+        cost_basis="local-unpriced",
+        success=True,
+        error="",
+    )
+    assert trace.memory_candidates == 0
+    assert trace.memory_selected == 0
+    assert trace.memory_abstained is False
+    assert trace.memory_types_queried == []
+
+
+def test_operator_trace_record_carries_memory_selection_fields(tmp_path, monkeypatch):
+    """A scripted operator run persists the v5 memory-selection fields."""
+    monkeypatch.setenv("RUNE_HOME", str(tmp_path / "home-m"))
+    (tmp_path / "home-m" / "state").mkdir(parents=True, exist_ok=True)
+    from rune import inference_trace as it
+
+    storage = tmp_path / "operator-m.json"
+    op = ro.RuneOperator(str(storage))
+    trace = {
+        "model_id": "m",
+        "window": 40960,
+        "budget": 32768,
+        "blocks": [
+            {"name": "persona", "tokens": 10, "admitted": True, "reason": ""},
+            {"name": "user", "tokens": 5, "admitted": True, "reason": ""},
+        ],
+        "admitted_total": 15,
+        "dropped": [],
+        "memory_tokens": 0,
+        "degraded": [],
+        "compaction": [],
+        "elapsed_ms": 1.0,
+    }
+    capsule = mock.Mock(
+        token_count=15,
+        active_objective="",
+        blocks=[],
+        to_messages=lambda: [{"role": "user", "content": "do it"}],
+    )
+    broker = mock.Mock(assemble=lambda task, session, model_id: (capsule, trace))
+    response = ro.OperatorResponse(success=True, user_text="done", model="m")
+
+    class _RT:
+        transport = mock.Mock(provider="ollama")
+        tool_names = frozenset()
+
+        def run(self, messages, **kwargs):
+            return response
+
+    with (
+        mock.patch.object(op, "_context_broker", return_value=broker),
+        mock.patch.object(ro.model_profiles, "get_or_create_profile", return_value=mock.Mock(safety_tier=0)),
+        mock.patch.object(ro.model_profiles, "record_interaction"),
+        mock.patch.object(ro, "LocalToolRuntime", return_value=_RT()),
+        mock.patch.object(ro, "OllamaChatTransport", return_value=mock.Mock(provider="ollama")),
+        mock.patch("rune.local_tool_runtime.context_length", lambda model_: 40960),
+        mock.patch.object(
+            ro.safety_override.SafetyOverride,
+            "get_params",
+            lambda self, text: {"tier": 0, "temperature": 0.7, "prefix": "", "suffix": ""},
+        ),
+        mock.patch.object(ro.safety_override.SafetyOverride, "is_refusal", lambda self, text: False),
+        mock.patch.object(op, "_log_token_usage"),
+        mock.patch("rune.event_bus._store_path", lambda: tmp_path / "journal-m-events.jsonl"),
+    ):
+        envelope = op.send_task("fix this code path")
+
+    assert envelope.get("success") is True
+    records = it.read_traces(trace_dir=tmp_path / "home-m" / "state")
+    assert records, "durable trace missing"
+    record = records[-1]
+    assert record["schema_version"] == 5
+    assert record["memory_candidates"] == 0
+    assert record["memory_selected"] == 0
+    assert record["memory_abstained"] is True, "no memory ran in this scripted broker -> abstained"
+    assert record["memory_types_queried"] == []
+
+
+def test_memory_selection_doc_exists():
+    doc = ROOT / "docs" / "memory-selection.md"
+    assert doc.is_file()
+    text = doc.read_text()
+    assert "MemorySelector" in text
+    assert "select_for_task" in text
+    assert "RECALL_MIN_SCORE" in text
+    assert "abstain" in text.lower()
+    assert "memory budget" in text.lower()
+
+
+def test_canary_p1_03_registered_in_ci():
+    canary = ROOT / "src" / "rune" / "canary_p1_03.py"
+    assert canary.is_file(), "canary must exist"
+    ci = ROOT / ".gitea" / "workflows" / "ci.yml"
+    assert "canary_p1_03.py" in ci.read_text(), "canary must be registered in the CI canary loop"