diff --git a/src/mnemosyne/fidelity.py b/src/mnemosyne/fidelity.py index d68fced..e48266a 100644 --- a/src/mnemosyne/fidelity.py +++ b/src/mnemosyne/fidelity.py @@ -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: diff --git a/src/mnemosyne/gateway.py b/src/mnemosyne/gateway.py index c58bb27..b4bab1e 100644 --- a/src/mnemosyne/gateway.py +++ b/src/mnemosyne/gateway.py @@ -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)