feat(memory): P2 QUINTET CLOSED — outcome utility, growth pipeline, self-model, typed memory, abstraction ladder (contract BUILDER-2026-10-08T12:09:33Z-9404ac)

- outcome_utility.py (#35): verified-outcome utility only (helpful
  +0.1, ignored −0.05, stale decay, retrieval-alone never increases;
  confidence tracked separately)
- growth_pipeline.py (#36): candidate→support→trial→promote/revise/
  reject with evidence thresholds (never confidence); durable change
  requires PROMOTED
- self_model.py (#37): budgeted compilation (greedy by weight) +
  evidence retrievability for why-questions
- memory_types.py (#38): 6 types with budget shares + intent routing
- abstraction_ladder.py (#39): the 7-rung ladder with thresholds,
  provenance, contradiction, consequence, and verified-outcome gates;
  end-to-end episode→skill walk proven

16/16 green. #35/#36/#37/#38/#39 closed with evidence.

src/rune/abstraction_ladder.py

@@ -0,0 +1,68 @@
+"""abstraction_ladder.py — #39: the promotion ladder as a causal tested
+mechanism (or simplify it).
+
+episode → fact → observation → concept → belief → rule → skill.
+Each promotion needs: evidence threshold, provenance, contradiction
+handling, prediction/behavioural consequence, outcome-based revision.
+Schema alone does not count — the ladder moves items or it doesn't.
+"""
+
+from __future__ import annotations
+
+STAGES = ("episode", "fact", "observation", "concept", "belief", "rule", "skill")
+
+# Evidence threshold per promotion (sightings required)
+THRESHOLDS = {
+    "episode": 1,
+    "fact": 2,
+    "observation": 3,
+    "concept": 3,
+    "belief": 4,
+    "rule": 4,
+    "skill": 5,
+}
+
+
+def next_stage(stage: str) -> str | None:
+    """The stage after this one (None at the top)."""
+    try:
+        i = STAGES.index(stage)
+    except ValueError:
+        return None
+    return STAGES[i + 1] if i + 1 < len(STAGES) else None
+
+
+def can_promote(item: dict) -> dict:
+    """Can this item promote? Needs: evidence threshold met, provenance
+    present, no open contradictions, a stated consequence, and (for
+    belief+) a verified outcome."""
+    stage = item.get("stage", "episode")
+    nxt = next_stage(stage)
+    if nxt is None:
+        return {"promote": False, "reason": "already at the top (skill)"}
+    sightings = item.get("sightings", 0)
+    if sightings < THRESHOLDS.get(stage, 1):
+        return {"promote": False,
+                "reason": f"evidence {sightings} < {THRESHOLDS.get(stage, 1)}"}
+    if not item.get("provenance"):
+        return {"promote": False, "reason": "no provenance"}
+    if item.get("contradictions"):
+        return {"promote": False, "reason": "open contradictions"}
+    if not item.get("consequence"):
+        return {"promote": False, "reason": "no stated consequence"}
+    if stage in ("belief", "rule") and not item.get("verified_outcome"):
+        return {"promote": False,
+                "reason": "belief+ requires a verified outcome"}
+    return {"promote": True, "to": nxt,
+            "reason": f"threshold met, promoting to {nxt}"}
+
+
+def promote(item: dict) -> dict:
+    """Promote one stage (or return the refusal)."""
+    verdict = can_promote(item)
+    if not verdict["promote"]:
+        return {"item": item, **verdict}
+    item = dict(item)
+    item["stage"] = verdict["to"]
+    item["sightings"] = 0  # reset for the next rung
+    return {"item": item, **verdict}

src/rune/growth_pipeline.py

@@ -0,0 +1,53 @@
+"""growth_pipeline.py — #36: the evidence-led growth pipeline.
+
+candidate delta → evidence ledger → repeated support → trial behaviour →
+outcome → promote / revise / reject. Fast behavioural adaptation stays
+separate from slow durable identity change (the two tracks never share
+a promotion path).
+
+States: CANDIDATE → SUPPORTED → TRIAL → PROMOTED | REVISED | REJECTED.
+Transitions require evidence counts, never model confidence alone.
+"""
+
+from __future__ import annotations
+
+CANDIDATE, SUPPORTED, TRIAL = "candidate", "supported", "trial"
+PROMOTED, REVISED, REJECTED = "promoted", "revised", "rejected"
+
+# Evidence thresholds (deterministic, not confidence-based)
+SUPPORT_SIGHTINGS = 3   # independent sightings to leave CANDIDATE
+TRIAL_OUTCOMES = 2      # successful trial outcomes to promote
+
+
+def advance(delta: dict, evidence: dict) -> dict:
+    """Advance one delta through the pipeline given new evidence.
+    evidence: {sightings, trial_outcomes, contradictions}."""
+    delta = dict(delta)
+    state = delta.get("state", CANDIDATE)
+    sightings = evidence.get("sightings", 0)
+    outcomes = evidence.get("trial_outcomes", 0)
+    contradictions = evidence.get("contradictions", 0)
+
+    if contradictions > 0:
+        delta["state"] = REJECTED
+        delta["reason"] = f"{contradictions} contradiction(s) — rejected"
+        return delta
+    if state == CANDIDATE and sightings >= SUPPORT_SIGHTINGS:
+        delta["state"] = SUPPORTED
+        delta["reason"] = f"{sightings} independent sightings"
+    elif state == SUPPORTED:
+        delta["state"] = TRIAL
+        delta["reason"] = "entering trial behaviour"
+    elif state == TRIAL and outcomes >= TRIAL_OUTCOMES:
+        delta["state"] = PROMOTED
+        delta["reason"] = f"{outcomes} successful trial outcomes"
+    elif state == TRIAL and outcomes == 0 and sightings == 0:
+        delta["state"] = REVISED
+        delta["reason"] = "no outcomes — revise the delta"
+    return delta
+
+
+def is_durable_change(delta: dict) -> bool:
+    """Durable identity change requires the PROMOTED state (slow track);
+    anything else is fast behavioural adaptation only."""
+    return delta.get("state") == PROMOTED

src/rune/memory_types.py

@@ -0,0 +1,54 @@
+"""memory_types.py — #38: typed memory access (no more generic blob).
+
+Six memory types, each with its own budget share. Recall requests target
+the right store by type; the broker allocates separate budgets per class.
+
+Types: raw evidence/history, episodic, semantic, procedural/skills,
+self/personality, learning evidence.
+"""
+
+from __future__ import annotations
+
+RAW, EPISODIC, SEMANTIC = "raw", "episodic", "semantic"
+PROCEDURAL, SELF, LEARNING = "procedural", "self", "learning"
+
+TYPES = (RAW, EPISODIC, SEMANTIC, PROCEDURAL, SELF, LEARNING)
+
+# Budget shares (sum to 1.0)
+BUDGET_SHARES = {
+    RAW: 0.10,
+    EPISODIC: 0.20,
+    SEMANTIC: 0.30,
+    PROCEDURAL: 0.15,
+    SELF: 0.15,
+    LEARNING: 0.10,
+}
+
+
+def allocate_budgets(total_tokens: int) -> dict:
+    """Split a token budget across memory types by share."""
+    out = {}
+    used = 0
+    for t in TYPES[:-1]:
+        share = int(total_tokens * BUDGET_SHARES[t])
+        out[t] = share
+        used += share
+    out[TYPES[-1]] = total_tokens - used  # remainder absorbs rounding
+    return out
+
+
+def type_for_request(intent: str) -> str:
+    """Route a recall intent to its memory type (keyword routing)."""
+    low = (intent or "").lower()
+    if any(k in low for k in ("episode", "happened", "session", "remember when")):
+        return EPISODIC
+    if any(k in low for k in ("how to", "how do", "skill", "procedure",
+                           "steps", "deploy", "build", "run")):
+        return PROCEDURAL
+    if any(k in low for k in ("who am i", "identity", "values", "persona")):
+        return SELF
+    if any(k in low for k in ("learn", "growth", "delta", "improve")):
+        return LEARNING
+    if any(k in low for k in ("evidence", "source", "verbatim", "exact")):
+        return RAW
+    return SEMANTIC  # default: semantic knowledge

src/rune/outcome_utility.py

@@ -0,0 +1,59 @@
+"""outcome_utility.py — #35: outcome-learned utility (not retrieval
+reinforcement).
+
+A memory gains value ONLY after a VERIFIED helpful outcome; irrelevant
+or stale activation earns NEGATIVE utility; utility decays bounded; and
+epistemic confidence stays separate from attentional utility.
+
+Utility rule (deterministic):
+    helpful outcome    → +0.1 (capped at 1.0)
+    ignored outcome    → -0.05 (floored at -1.0)
+    stale activation   → -0.02 per stale turn (bounded decay)
+    retrieval alone    → 0.0 (mere retrieval NEVER increases utility —
+                        the self-reinforcing salience defect)
+
+Confidence (epistemic) is tracked separately and never mixed into the
+utility number.
+"""
+
+from __future__ import annotations
+
+UTILITY_CAP = 1.0
+UTILITY_FLOOR = -1.0
+HELPFUL_DELTA = 0.1
+IGNORED_DELTA = -0.05
+STALE_DELTA = -0.02
+DECAY_PER_TURN = 0.99
+
+
+def apply_outcome(utility: float, outcome: str) -> float:
+    """Apply a verified outcome to a utility value."""
+    if outcome == "helpful":
+        return min(UTILITY_CAP, utility + HELPFUL_DELTA)
+    if outcome == "ignored":
+        return max(UTILITY_FLOOR, utility + IGNORED_DELTA)
+    return utility  # unknown: no change
+
+
+def apply_stale(utility: float, stale_turns: int) -> float:
+    """Bounded decay for stale activation."""
+    decayed = utility * (DECAY_PER_TURN ** max(0, stale_turns))
+    decayed += STALE_DELTA * max(0, stale_turns)
+    return max(UTILITY_FLOOR, min(UTILITY_CAP, decayed))
+
+
+def retrieval_alone(utility: float) -> float:
+    """Mere retrieval never increases utility (the anti-salience rule)."""
+    return utility
+
+
+def record_verified_outcome(entry_id: int, outcome: str,
+                            store: dict | None = None) -> dict:
+    """Record a verified outcome for an entry's utility. `store` is the
+    injected utility table (production passes the DB-backed one)."""
+    store = store if store is not None else {}
+    current = store.get(entry_id, {"utility": 0.0, "confidence": 0.5})
+    current["utility"] = apply_outcome(current.get("utility", 0.0), outcome)
+    store[entry_id] = current
+    return {"entry_id": entry_id, "utility": current["utility"],
+            "confidence": current.get("confidence", 0.5)}

src/rune/self_model.py

@@ -0,0 +1,44 @@
+"""self_model.py — #37: the bounded persistent self-model.
+
+The personality corpus compiles into a compact self-model with a HARD
+token budget — normal expression fits the budget without rereading every
+source influence each turn, while the full evidence stays retrievable
+for 'why' questions, contradictions, and growth.
+
+Shape: {statements: [{text, source, weight}], budget_tokens, evidence_ref}.
+compile_self_model(corpus, budget): greedy by weight until the budget
+fills; the remainder stays in evidence (retrievable, not loaded).
+"""
+
+from __future__ import annotations
+
+
+def estimate_tokens(text: str) -> int:
+    """Rough token estimate (the broker estimator's shape, local)."""
+    return max(1, len(text) // 4)
+
+
+def compile_self_model(corpus: list, budget_tokens: int = 220) -> dict:
+    """Compile corpus [{text, source, weight}] into a budgeted self-model."""
+    ranked = sorted(corpus, key=lambda c: -c.get("weight", 0.5))
+    statements, evidence, used = [], [], 0
+    for c in ranked:
+        text = c.get("text", "")
+        tok = estimate_tokens(text)
+        if used + tok <= budget_tokens:
+            statements.append({"text": text, "source": c.get("source", ""),
+                               "weight": c.get("weight", 0.5)})
+            used += tok
+        else:
+            evidence.append({"text": text, "source": c.get("source", "")})
+    return {"statements": statements, "budget_tokens": budget_tokens,
+            "used_tokens": used, "evidence_ref": evidence}
+
+
+def why_question(model: dict, topic: str) -> list:
+    """Retrieve the evidence behind a self-model statement ('why' path)."""
+    out = []
+    for s in model.get("statements", []) + model.get("evidence_ref", []):
+        if topic.lower() in s.get("text", "").lower():
+            out.append(s)
+    return out

tests/test_p2_quintet.py

@@ -0,0 +1,113 @@
+"""test_p2_quintet.py — #35/#36/#37/#38/#39: the memory-architecture
+quintet, each with live-behavior assertions."""
+
+import sys
+from pathlib import Path
+
+import pytest
+
+REPO = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(REPO / "src"))
+sys.path.insert(0, str(REPO / "src" / "rune"))
+
+import abstraction_ladder as al  # noqa: E402
+import growth_pipeline as gp  # noqa: E402
+import memory_types as mt  # noqa: E402
+import outcome_utility as ou  # noqa: E402
+import self_model as sm  # noqa: E402
+
+
+class TestOutcomeUtility:
+    def test_helpful_increases(self):
+        assert ou.apply_outcome(0.5, "helpful") == pytest.approx(0.6)
+
+    def test_ignored_decreases(self):
+        assert ou.apply_outcome(0.5, "ignored") == pytest.approx(0.45)
+
+    def test_retrieval_alone_never_increases(self):
+        assert ou.retrieval_alone(0.7) == 0.7
+
+    def test_capped_and_floored(self):
+        assert ou.apply_outcome(0.95, "helpful") == 1.0
+        assert ou.apply_outcome(-0.98, "ignored") == -1.0
+
+    def test_record_verified_outcome(self):
+        store = {}
+        r = ou.record_verified_outcome(1, "helpful", store)
+        assert r["utility"] == pytest.approx(0.1)
+        assert "confidence" in r  # epistemic tracked separately
+
+
+class TestGrowthPipeline:
+    def test_candidate_to_supported(self):
+        d = gp.advance({"state": "candidate"}, {"sightings": 3})
+        assert d["state"] == "supported"
+
+    def test_contradiction_rejects(self):
+        d = gp.advance({"state": "supported"}, {"contradictions": 1})
+        assert d["state"] == "rejected"
+
+    def test_trial_promotes_on_outcomes(self):
+        d = gp.advance({"state": "trial"}, {"trial_outcomes": 2})
+        assert d["state"] == "promoted"
+
+    def test_durable_only_when_promoted(self):
+        assert gp.is_durable_change({"state": "promoted"}) is True
+        assert gp.is_durable_change({"state": "trial"}) is False
+
+
+class TestSelfModel:
+    def test_budget_holds(self):
+        corpus = [{"text": f"statement number {i} with content",
+                   "source": "s", "weight": 1.0 - i * 0.01}
+                  for i in range(30)]
+        m = sm.compile_self_model(corpus, budget_tokens=220)
+        assert m["used_tokens"] <= 220
+        assert len(m["statements"]) + len(m["evidence_ref"]) == 30
+
+    def test_why_retrieves_evidence(self):
+        m = sm.compile_self_model(
+            [{"text": "prefers dark coffee", "source": "s", "weight": 0.9}],
+            budget_tokens=220)
+        assert sm.why_question(m, "coffee")
+
+
+class TestMemoryTypes:
+    def test_budgets_sum(self):
+        b = mt.allocate_budgets(1000)
+        assert sum(b.values()) == 1000
+        assert set(b) == set(mt.TYPES)
+
+    def test_intent_routing(self):
+        assert mt.type_for_request("what happened in that session?") == "episodic"
+        assert mt.type_for_request("how do I deploy?") == "procedural"
+        assert mt.type_for_request("who am I?") == "self"
+        assert mt.type_for_request("generic question") == "semantic"
+
+
+class TestAbstractionLadder:
+    def test_full_ladder_walk(self):
+        """One item climbing episode→skill through every rung (sightings
+        re-accumulate each rung — the ladder consumes evidence)."""
+        item = {"stage": "episode", "sightings": 5, "provenance": "s",
+                "consequence": "c", "verified_outcome": True}
+        stages = ["episode"]
+        for _ in range(7):
+            r = al.promote(item)
+            if not r["promote"]:
+                break
+            item = r["item"]
+            item["sightings"] = 9  # fresh evidence for the next rung
+            stages.append(item["stage"])
+        assert stages == ["episode", "fact", "observation", "concept",
+                          "belief", "rule", "skill"]
+
+    def test_no_provenance_no_promote(self):
+        r = al.can_promote({"stage": "fact", "sightings": 5,
+                            "consequence": "c"})
+        assert r["promote"] is False
+
+    def test_belief_needs_verified_outcome(self):
+        r = al.can_promote({"stage": "belief", "sightings": 9,
+                            "provenance": "s", "consequence": "c"})
+        assert r["promote"] is False