jyapayne/mnemosyne
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix: gradual fidelity degradation with recency guard and better stubs

Browse source 
Root cause of retry loops: mass degradation stubbed 85% of context at
once, confusing the model into infinite retries.

Fixes:
- Cap degradations at 20 per turn (gradual compression)
- Protect objects accessed within last 10 turns from degradation
- Estimate L1/L2 token counts (30%/10% of L0) so FM pressure tracks
  correctly after degradation
- Improved stubs: '[compressed 3.4KB -> stub] first 200 chars...'
- Re-enabled _apply_fidelity
This commit is contained in:
Joey Yakimowich-Payne 2026-03-15 09:39:42 -06:00
commit ac5e207a73
2 changed files with 52 additions and 20 deletions
Download patch file Download diff file Expand all files Collapse all files

52
src/mnemosyne/fidelity.py
View file

@ -164,8 +164,16 @@ def make_object(
created_at_turn=created_at_turn,
last_accessed_turn=created_at_turn,
token_count_l0=_estimate_tokens(content_full),
token_count_l1=_estimate_tokens(summary_detailed) if summary_detailed else None,
token_count_l2=_estimate_tokens(summary_compact) if summary_compact else None,
token_count_l1=(
_estimate_tokens(summary_detailed)
if summary_detailed
else max(10, _estimate_tokens(content_full) // 3) # ~30% of L0
),
token_count_l2=(
_estimate_tokens(summary_compact)
if summary_compact
else max(5, _estimate_tokens(content_full) // 10) # ~10% of L0
),
token_count_l3=_estimate_tokens(stub) if stub else 25,
)
return obj
@ -310,9 +318,22 @@ class FidelityManager:
obj.pinned = False
obj.pin_until_turn = None
def _degradable(self, obj: SemanticObject) -> bool:
"""Check if an object can be degraded (not pinned, not already L4)."""
return not obj.pinned and obj.current_fidelity < FidelityLevel.L4
# Objects accessed within the last N turns are protected from degradation.
_RECENCY_GUARD_TURNS = 10
# Maximum number of objects to degrade per turn (prevents mass stubbing).
_MAX_DEGRADES_PER_TURN = 20
def _degradable(self, obj: SemanticObject, current_turn: int = 0) -> bool:
"""Check if an object can be degraded.
Protected if: pinned, already L4, or accessed within the last
``_RECENCY_GUARD_TURNS`` turns.
"""
if obj.pinned or obj.current_fidelity >= FidelityLevel.L4:
return False
if current_turn > 0 and (current_turn - obj.last_accessed_turn) < self._RECENCY_GUARD_TURNS:
return False
return True
def _degrade_caution(
self,
@ -323,11 +344,13 @@ class FidelityManager:
candidates = [
o
for o in self._objects.values()
if o.current_fidelity == FidelityLevel.L0 and self._degradable(o)
if o.current_fidelity == FidelityLevel.L0 and self._degradable(o, current_turn)
]
candidates = self._sorted_by_age(candidates)
for obj in candidates:
if len(transitions) >= self._MAX_DEGRADES_PER_TURN:
break
old = obj.current_fidelity
new = self._degrade_one(obj)
if new is not None:
@ -344,16 +367,19 @@ class FidelityManager:
) -> list[tuple[str, FidelityLevel, FidelityLevel]]:
"""Warning zone: degrade L0→L1, L1→L2, oldest L2→L3."""
transitions: list[tuple[str, FidelityLevel, FidelityLevel]] = []
cap = self._MAX_DEGRADES_PER_TURN
# First pass: L0 → L1
l0_objs = self._sorted_by_age(
[
o
for o in self._objects.values()
if o.current_fidelity == FidelityLevel.L0 and self._degradable(o)
if o.current_fidelity == FidelityLevel.L0 and self._degradable(o, current_turn)
]
)
for obj in l0_objs:
if len(transitions) >= cap:
return transitions
old = obj.current_fidelity
new = self._degrade_one(obj)
if new is not None:
@ -366,10 +392,12 @@ class FidelityManager:
[
o
for o in self._objects.values()
if o.current_fidelity == FidelityLevel.L1 and self._degradable(o)
if o.current_fidelity == FidelityLevel.L1 and self._degradable(o, current_turn)
]
)
for obj in l1_objs:
if len(transitions) >= cap:
return transitions
old = obj.current_fidelity
new = self._degrade_one(obj)
if new is not None:
@ -382,7 +410,7 @@ class FidelityManager:
[
o
for o in self._objects.values()
if o.current_fidelity == FidelityLevel.L2 and self._degradable(o)
if o.current_fidelity == FidelityLevel.L2 and self._degradable(o, current_turn)
]
)
for obj in l2_objs:
@ -408,7 +436,7 @@ class FidelityManager:
[
o
for o in self._objects.values()
if o.current_fidelity == target_level and self._degradable(o)
if o.current_fidelity == target_level and self._degradable(o, current_turn)
]
)
for obj in candidates:
@ -426,7 +454,7 @@ class FidelityManager:
[
o
for o in self._objects.values()
if o.current_fidelity == FidelityLevel.L3 and self._degradable(o)
if o.current_fidelity == FidelityLevel.L3 and self._degradable(o, current_turn)
]
)
for obj in l3_objs:
@ -455,7 +483,7 @@ class FidelityManager:
[
o
for o in self._objects.values()
if o.current_fidelity < FidelityLevel.L4 and self._degradable(o)
if o.current_fidelity < FidelityLevel.L4 and self._degradable(o, current_turn)
]
)
for obj in all_objs:

20
src/mnemosyne/gateway.py
View file

@ -1209,11 +1209,17 @@ def _block_text(block: dict) -> str:
def _auto_stub(text: str) -> str:
"""Generate a simple stub from the first line of content."""
first_line = text.split("\n", 1)[0].strip()
if len(first_line) > 120:
first_line = first_line[:117] + "..."
return f"[evicted content: {first_line}]"
"""Generate a compact stub preserving the first ~200 chars of content.
Keeps enough context for the model to understand what the content was
without needing the full text. The stub format includes a size hint
so the model knows how much was removed.
"""
size_kb = len(text.encode("utf-8", errors="replace")) / 1024
preview = text[:200].replace("\n", " ").strip()
if len(text) > 200:
preview = preview[:197] + "..."
return f"[compressed {size_kb:.1f}KB → stub] {preview}"
import logging as _logging
@ -2183,9 +2189,7 @@ def create_app(
# )
# 4b. Apply fidelity-based content replacement
# DISABLED: L1/L2 stub replacement causes response issues that trigger
# rapid retries in opencode. Re-enable once stub format is validated.
# _apply_fidelity(payload, session)
_apply_fidelity(payload, session)
# Place cache_control markers at optimal positions
_place_cache_controls(payload)