feat(forgetting): thread narrative_weight arc pressure to fade threshold

Open-arc intensity now modulates the forgetting threshold so memories
resist fading during heavy narrative periods. effective_fade = FADE_THRESHOLD /
(1 + narrative_weight) — at full arc pressure the bar is halved. The
unfade path is symmetrically lowered so recovery is also easier.

Wire: _run_felt_time_tick now returns IntensityDrivers; run_folded caches
the last computed value and passes it to forgetting_run_pass; run_pass
threads narrative_weight to policy.next_state. LOST_THRESHOLD is unchanged —
arc pressure only delays the fade gate, not the final deletion gate.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: Hana

brain/bridge/supervisor.py

@@ -112,6 +112,7 @@ def run_folded(
     )
     last_heartbeat_at = time.monotonic() if heartbeat_interval_s is not None else None
     last_soul_review_at = time.monotonic() if soul_review_interval_s is not None else None
+    _last_intensity_drivers: IntensityDrivers | None = None
     last_finalize_at = time.monotonic() if finalize_interval_s is not None else None
     last_log_rotation_at = time.monotonic() if log_rotation_interval_s is not None else None
     last_initiate_review_at = time.monotonic() if initiate_review_interval_s is not None else None
@@ -191,7 +192,7 @@ def run_folded(
                 # accessor is identified.
                 # TODO(v0.0.15): replace 0 chat_turns + 0 reflex_firings with
                 # real counters from the event bus or engine state.
-                _run_felt_time_tick(
+                _last_intensity_drivers = _run_felt_time_tick(
                     persona_dir,
                     wall_clock_s_since_last=wall_s,
                     heartbeats_since_last=1,
@@ -215,7 +216,11 @@ def run_folded(
             except Exception:
                 logger.exception("supervisor soul-review tick raised")
             try:
-                forgetting_run_pass(persona_dir, event_bus=event_bus)
+                forgetting_run_pass(
+                    persona_dir,
+                    event_bus=event_bus,
+                    intensity_drivers=_last_intensity_drivers,
+                )
             except Exception:
                 logger.exception("supervisor forgetting pass raised")
             # Narrative-memory arc-update runs AFTER forgetting in the same
@@ -302,13 +307,16 @@ def _run_felt_time_tick(
     heartbeats_since_last: int,
     chat_turns_since_last: int,
     reflex_firings_since_last: int,
-) -> None:
+) -> IntensityDrivers:
     """Fold one supervisor heartbeat cycle into felt-time state.
 
     drivers values are derived from existing body + emotion accessors;
     cold-start cases (no body state, no emotion vector) collapse all
     drivers to 0.0 so lived-age advances at baseline.
 
+    Returns the computed IntensityDrivers so the caller can cache them for
+    the next forgetting pass (arc pressure modulates the fade threshold).
+
     Fault-isolated upstream: caller wraps in try/except so a raise here
     cannot cascade into bridge shutdown or take down the heartbeat loop.
     """
@@ -324,6 +332,7 @@ def _run_felt_time_tick(
             drivers=drivers,
         )
     )
+    return drivers
 
 
 def _derive_intensity_drivers(

brain/forgetting/__init__.py

@@ -12,6 +12,7 @@
 from pathlib import Path
 from typing import Any
 
+from brain.felt_time.lived_age import IntensityDrivers
 from brain.felt_time.state import load_or_recover as load_felt_time
 from brain.forgetting import graveyard, policy, salience, tombstone
 from brain.forgetting.policy import Transition
@@ -92,7 +93,12 @@ def _load_migration_grace(persona_dir: Path) -> tuple[datetime | None, float]:
     return mig, lived
 
 
-def run_pass(persona_dir: Path, *, event_bus: Any) -> dict[str, int]:
+def run_pass(
+    persona_dir: Path,
+    *,
+    event_bus: Any,
+    intensity_drivers: IntensityDrivers | None = None,
+) -> dict[str, int]:
     """Run one forgetting pass over all active+fading memories.
 
     Returns an aggregate summary dict with counts; also publishes a
@@ -164,7 +170,13 @@ def run_pass(persona_dir: Path, *, event_bus: Any) -> dict[str, int]:
                 next_low = prev_low + 1
             else:
                 next_low = 0
-            transition = policy.next_state(memory, salience=s, consecutive_low_passes=next_low)
+            nw = intensity_drivers.narrative_weight if intensity_drivers else 0.0
+            transition = policy.next_state(
+                memory,
+                salience=s,
+                consecutive_low_passes=next_low,
+                narrative_weight=nw,
+            )
             if transition == Transition.FADE:
                 summary_text = tombstone.summarise(memory.content)
                 store.fade(memory_id, summary=summary_text)

brain/forgetting/policy.py

@@ -38,20 +38,29 @@ def next_state(
     *,
     salience: float,
     consecutive_low_passes: int,
+    narrative_weight: float = 0.0,
 ) -> Transition:
     """Compute the transition for one memory in one forgetting pass.
 
     consecutive_low_passes: the count from the persisted forgetting_state.json
     tracking how many recent passes saw salience < LOST_THRESHOLD for this
     memory. The orchestrator increments/resets this counter; this function
     just reads it.
+
+    narrative_weight: open-arc pressure in [0, 1] from IntensityDrivers.
+    A heavy open arc lowers the effective fade threshold so memories resist
+    fading during intense narrative periods. LOST_THRESHOLD is unchanged —
+    arc pressure only delays the fade gate, not the final deletion gate.
     """
+    # Arc pressure lowers the effective threshold proportionally.
+    # narrative_weight=0 → baseline; narrative_weight=1 → threshold halved.
+    effective_fade = FADE_THRESHOLD / (1.0 + narrative_weight)
     if memory.state == "active":
-        if salience < FADE_THRESHOLD:
+        if salience < effective_fade:
             return Transition.FADE
         return Transition.NONE
     if memory.state == "fading":
-        if salience >= FADE_THRESHOLD:
+        if salience >= effective_fade:
             return Transition.UNFADE
         if salience < LOST_THRESHOLD and consecutive_low_passes >= LOST_PASS_COUNT:
             return Transition.LOSE

tests/bridge/test_supervisor_forgetting.py

@@ -97,3 +97,53 @@ def _soul_review_counter(*a, **k):
 
     # Soul-review kept running even though forgetting raised each time.
     assert soul_review_calls[0] >= 2
+
+
+def test_supervisor_passes_intensity_drivers_to_forgetting(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """run_folded caches IntensityDrivers from the felt-time tick and forwards them
+    to forgetting_run_pass so arc pressure can modulate the fade threshold."""
+    from brain.felt_time.lived_age import IntensityDrivers
+
+    persona_dir = tmp_path / "persona"
+    persona_dir.mkdir()
+
+    fake_drivers = IntensityDrivers(narrative_weight=0.7)
+    captured_drivers: list = []
+    stop_event = threading.Event()
+
+    def _fake_felt_time_tick(*a, **k):
+        return fake_drivers  # supervisor caches this return value
+
+    def _fake_forgetting_pass(persona_dir, *, event_bus, intensity_drivers=None, **_):
+        captured_drivers.append(intensity_drivers)
+        stop_event.set()
+        return {"faded": 0, "lost": 0, "total": 0, "exempt": 0, "unfaded": 0, "duration_ms": 0}
+
+    monkeypatch.setattr("brain.bridge.supervisor._run_felt_time_tick", _fake_felt_time_tick)
+    monkeypatch.setattr("brain.bridge.supervisor.forgetting_run_pass", _fake_forgetting_pass)
+    monkeypatch.setattr("brain.bridge.supervisor._run_soul_review_tick", lambda *a, **k: None)
+    monkeypatch.setattr("brain.bridge.supervisor._run_heartbeat_tick", lambda *a, **k: None)
+    monkeypatch.setattr("brain.bridge.supervisor._run_narrative_memory_pass", lambda *a, **k: None)
+
+    run_folded(
+        stop_event,
+        persona_dir=persona_dir,
+        provider=MagicMock(),
+        event_bus=MagicMock(),
+        tick_interval_s=0.05,
+        # Both intervals set to 0 so they fire on the first loop iteration.
+        # Heartbeat block runs first in the loop body, then soul-review — so
+        # drivers computed during heartbeat are available when forgetting fires.
+        heartbeat_interval_s=0.0,
+        soul_review_interval_s=0.0,
+        finalize_interval_s=None,
+        initiate_review_interval_s=None,
+        voice_reflection_interval_s=None,
+        log_rotation_interval_s=None,
+    )
+
+    assert len(captured_drivers) >= 1, "forgetting pass should have been called"
+    assert captured_drivers[0] is not None, "intensity_drivers should be forwarded, not None"
+    assert captured_drivers[0].narrative_weight == pytest.approx(0.7)

tests/forgetting/test_orchestrator.py

@@ -108,6 +108,48 @@ def test_run_pass_cold_start_no_memories(tmp_path):
     assert summary["lost"] == 0
 
 
+def test_run_pass_intensity_drivers_protects_borderline_memory(tmp_path):
+    """intensity_drivers.narrative_weight threads to policy and protects memories from fading."""
+    from datetime import timedelta
+    from unittest.mock import patch
+
+    from brain.felt_time.lived_age import IntensityDrivers
+    from brain.forgetting import policy
+
+    persist_felt_time(
+        FeltTimeState(lived_age_hours=100.0, last_tick_ts="2026-05-18T00:00:00+00:00"),
+        tmp_path,
+    )
+    store = MemoryStore(tmp_path / "memories.db")
+    m = Memory.create_new(content="arc memory", memory_type="episodic", domain="chat", emotions={})
+    object.__setattr__(m, "created_at", datetime.now(UTC) - timedelta(days=10))
+    store.create(m)
+    store.close()
+
+    # Pin salience to just below baseline FADE_THRESHOLD but above the arc-adjusted threshold.
+    # effective_fade at narrative_weight=1.0 → FADE_THRESHOLD / 2 = 0.125
+    borderline_salience = 0.20  # 0.125 < 0.20 < 0.25
+
+    event_bus = MagicMock()
+    with patch("brain.forgetting.salience.score", return_value=borderline_salience):
+        # Without arc pressure: salience 0.20 < FADE_THRESHOLD 0.25 → should fade
+        summary_no_arc = run_pass(tmp_path, event_bus=event_bus)
+        assert summary_no_arc["faded"] >= 1, "baseline: borderline memory should fade without arc"
+
+    # Reload memory to active state
+    store = MemoryStore(tmp_path / "memories.db")
+    store._conn.execute("UPDATE memories SET state = 'active' WHERE id = ?", (m.id,))
+    store._conn.commit()
+    store.close()
+
+    event_bus2 = MagicMock()
+    drivers = IntensityDrivers(narrative_weight=1.0)
+    with patch("brain.forgetting.salience.score", return_value=borderline_salience):
+        # With full arc pressure: effective_fade = 0.125, 0.20 >= 0.125 → should NOT fade
+        summary_arc = run_pass(tmp_path, event_bus=event_bus2, intensity_drivers=drivers)
+        assert summary_arc["faded"] == 0, "arc pressure: borderline memory should be protected"
+
+
 def test_run_pass_recovers_from_corrupt_forgetting_state(persona_with_low_salience_memory):
     persona_dir, _mem_id = persona_with_low_salience_memory
     # Pre-corrupt the state file.

tests/forgetting/test_policy.py

@@ -163,3 +163,42 @@ def test_clock_skew_clamped():
     assert policy.is_within_import_grace(
         mem, migrated_at_utc=mig, lived_age_hours_at_migration=100.0,
         current_lived_age_hours=50.0) is True
+
+
+# ---------------------------------------------------------------------------
+# narrative_weight — arc pressure lowers the effective fade threshold
+# ---------------------------------------------------------------------------
+
+
+def test_narrative_weight_protects_borderline_active_memory():
+    """Arc pressure lowers effective fade threshold → memory held that would otherwise fade."""
+    m = _make_memory(state="active")
+    # Just below the baseline threshold — fades without arc pressure
+    borderline = FADE_THRESHOLD - 0.01  # 0.24
+    assert next_state(m, salience=borderline, consecutive_low_passes=0) == Transition.FADE
+    # Strong arc pressure halves the threshold → 0.24 is now safely above it
+    assert next_state(
+        m, salience=borderline, consecutive_low_passes=0, narrative_weight=0.8
+    ) == Transition.NONE
+
+
+def test_narrative_weight_eases_unfade_for_fading_memory():
+    """Under arc pressure the reduced threshold makes it easier to recover a fading memory."""
+    m = _make_memory(state="fading")
+    # salience between effective threshold (arc) and baseline threshold
+    low_salience = 0.18
+    # Without arc pressure: 0.18 < 0.25 → stays fading
+    assert next_state(m, salience=low_salience, consecutive_low_passes=0) == Transition.NONE
+    # With arc pressure: effective_fade ≈ 0.139 → 0.18 >= 0.139 → unfades
+    assert next_state(
+        m, salience=low_salience, consecutive_low_passes=0, narrative_weight=0.8
+    ) == Transition.UNFADE
+
+
+def test_zero_narrative_weight_is_baseline():
+    """narrative_weight=0.0 behaves identically to omitting the argument."""
+    m = _make_memory(state="active")
+    assert (
+        next_state(m, salience=0.20, consecutive_low_passes=0, narrative_weight=0.0)
+        == next_state(m, salience=0.20, consecutive_low_passes=0)
+    )