From 5702a5a1e24e583a7d8cee26ebf2bc85b979ba22 Mon Sep 17 00:00:00 2001 From: Joey Yakimowich-Payne Date: Fri, 13 Mar 2026 13:37:33 -0600 Subject: [PATCH 01/10] fix: wire bytes_saved through benchmark, restore _check_token_cap, apply block cleanup to outbound Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-opencode) Co-authored-by: Sisyphus --- src/mnemosyne/benchmark.py | 18 ++++++++++---- src/mnemosyne/gateway.py | 49 +++++++++++++++++++++++++++++++------- 2 files changed, 53 insertions(+), 14 deletions(-) diff --git a/src/mnemosyne/benchmark.py b/src/mnemosyne/benchmark.py index 789474a..ce50978 100644 --- a/src/mnemosyne/benchmark.py +++ b/src/mnemosyne/benchmark.py @@ -287,6 +287,8 @@ class TokenMetrics: # Payload sizes incoming_bytes_per_turn: list[int] = field(default_factory=list) outgoing_bytes_per_turn: list[int] = field(default_factory=list) + # Eviction savings (bytes removed from context per turn) + bytes_saved_per_turn: list[int] = field(default_factory=list) def record_turn( self, @@ -296,14 +298,15 @@ class TokenMetrics: cache_create: int, incoming_bytes: int, outgoing_bytes: int, + bytes_saved: int = 0, ) -> None: self.turns += 1 self.input_tokens_per_turn.append(input_tokens) self.effective_tokens_per_turn.append(effective_tokens) self.cache_read_per_turn.append(cache_read) - self.cache_create_per_turn.append(cache_create) self.incoming_bytes_per_turn.append(incoming_bytes) self.outgoing_bytes_per_turn.append(outgoing_bytes) + self.bytes_saved_per_turn.append(bytes_saved) @property def total_input_tokens(self) -> int: @@ -322,14 +325,19 @@ class TokenMetrics: total_cache = self.total_cache_read + sum(self.cache_create_per_turn) return self.total_cache_read / total_cache if total_cache > 0 else 0.0 + @property + def total_bytes_saved(self) -> int: + return sum(self.bytes_saved_per_turn) + @property def context_reduction_ratio(self) -> float: - """How much smaller outgoing payloads are vs incoming. - 1.0 = no reduction, 0.5 = halved, 0.2 = 80% reduction. + """Fraction of context bytes removed by eviction. + 0.0 = no reduction, 0.5 = half evicted, 1.0 = fully evicted. """ total_in = sum(self.incoming_bytes_per_turn) - total_out = sum(self.outgoing_bytes_per_turn) - return total_out / total_in if total_in > 0 else 1.0 + if total_in <= 0: + return 0.0 + return min(sum(self.bytes_saved_per_turn) / total_in, 1.0) def to_dict(self) -> dict[str, Any]: return { diff --git a/src/mnemosyne/gateway.py b/src/mnemosyne/gateway.py index e30253e..92e31aa 100644 --- a/src/mnemosyne/gateway.py +++ b/src/mnemosyne/gateway.py @@ -1428,9 +1428,10 @@ def create_app( # ── Pre/post processing ────────────────────────────────────────── - def _preprocess(payload: dict, session: Session) -> dict: + def _preprocess(payload: dict, session: Session) -> tuple[dict, int]: """Apply gateway transformations before pipeline and forwarding. + Returns (modified_payload, bytes_saved_by_eviction). Operates on the raw Anthropic-format payload (before normalization) because system prompt injection needs access to the system field directly. @@ -1459,12 +1460,15 @@ def create_app( ps = session.page_store ms = session.message_store + _bytes_saved = 0 + # 1. Ingest into MessageStore (asserts append-only, compacts) ingest = ms.ingest( incoming_messages, age_threshold=4, min_evict_size=min_evict_size, ) + _bytes_saved += ingest.bytes_saved if ingest.new_count > 0 or ingest.compacted_count > 0: parts = [] if ingest.new_count: @@ -1763,6 +1767,18 @@ def create_app( # 4. Build ephemeral outbound view — never mutate the physical store payload["messages"] = copy.deepcopy(ms.messages) + # Apply cleanup/block-state rewrites to the outbound view so model-authored + # drop/summarize/collapse operations actually affect future forwarded context. + cleanup_apply = session.block_store.apply_to_messages(payload["messages"]) + if any(cleanup_apply.values()): + print( + f" {_DIM}[{session.id}] outbound cleanup apply: " + f"drop={cleanup_apply['dropped']} " + f"sum={cleanup_apply['summarized']} " + f"anchor={cleanup_apply['anchored']}{_RESET}", + file=sys.stderr, + ) + # 4a. Phantom tool injection — DISABLED when proxying for opencode. # opencode validates tool calls against its own registry and rejects # unknown tools like memory_query. Phantom tools require SSE stream @@ -1809,13 +1825,7 @@ def create_app( with open(session._page_checkpoint, "w") as f: _json.dump(session.page_store.checkpoint(), f) - return payload - - def _check_token_cap(usage: dict, session: Session) -> None: - """Track usage and enforce token cap.""" - session.track_usage(usage) - if token_cap <= 0: - return + return payload, _bytes_saved effective = session.token_state["last_effective"] pct = effective / token_cap * 100 @@ -1881,11 +1891,28 @@ def create_app( exc_info=True, ) + def _check_token_cap(usage: dict, session: Session) -> None: + """Track usage and enforce token cap.""" + session.track_usage(usage) + if token_cap <= 0: + return + effective = session.token_state["last_effective"] + pct = effective / token_cap * 100 + sid = session.id + if effective > token_cap: + session.token_state["blocked"] = True + print( + f"{_RED} [{sid}] TOKEN CAP EXCEEDED: {effective:,} / {token_cap:,} " + f"({pct:.0f}%) — next request will be blocked{_RESET}", + file=sys.stderr, + ) + def _display_turn_status( usage: dict, session: Session, incoming_bytes: int = 0, outgoing_bytes: int = 0, + bytes_saved: int = 0, ) -> None: """Post-response status line with cache hit rate.""" sid = session.id @@ -1905,6 +1932,7 @@ def create_app( cache_create=cache_create, incoming_bytes=incoming_bytes, outgoing_bytes=outgoing_bytes, + bytes_saved=bytes_saved, ) cap_str = "" @@ -1971,8 +1999,9 @@ def create_app( # Cleanup now runs inside _preprocess (before manifest injection) # Pre-process: system status, block labeling + preprocess_bytes_saved = 0 if endpoint == "messages": - payload = _preprocess(payload, session) + payload, preprocess_bytes_saved = _preprocess(payload, session) # Optional provider-level model override for cost-controlled runs. if provider == "anthropic" and anthropic_model_override: @@ -2079,6 +2108,7 @@ def create_app( session, incoming_bytes=incoming_bytes, outgoing_bytes=outgoing_bytes, + bytes_saved=preprocess_bytes_saved if endpoint == "messages" else 0, ) _update_fidelity_pressure(usage, session) emit_event( @@ -2140,6 +2170,7 @@ def create_app( session, incoming_bytes=incoming_bytes, outgoing_bytes=outgoing_bytes, + bytes_saved=preprocess_bytes_saved, ) _update_fidelity_pressure(usage, session) From 235e88d41661a30adccaa51da35891f5ad92c938 Mon Sep 17 00:00:00 2001 From: Joey Yakimowich-Payne Date: Fri, 13 Mar 2026 13:38:26 -0600 Subject: [PATCH 02/10] fix: route mnemosyne provider instead of anthropic in opencode plugin Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-opencode) Co-authored-by: Sisyphus --- opencode-plugin/src/index.ts | 19 ++++++++++--------- 1 file changed, 10 insertions(+), 9 deletions(-) diff --git a/opencode-plugin/src/index.ts b/opencode-plugin/src/index.ts index d762e89..4a81ea8 100644 --- a/opencode-plugin/src/index.ts +++ b/opencode-plugin/src/index.ts @@ -266,21 +266,22 @@ const MnemosynePlugin: Plugin = async (ctx: PluginInput): Promise => { } } - // Inject baseURL for the Anthropic provider - // This routes all Anthropic API calls through the Mnemosyne gateway + // Inject baseURL for the Mnemosyne provider only. + // Keep provider ID `anthropic` untouched so Claude Code OAuth and + // oh-my-opencode variants continue to work against direct Anthropic. inputConfig.provider = inputConfig.provider ?? {}; - inputConfig.provider.anthropic = inputConfig.provider.anthropic ?? {}; - inputConfig.provider.anthropic.options = - inputConfig.provider.anthropic.options ?? {}; + inputConfig.provider.mnemosyne = inputConfig.provider.mnemosyne ?? {}; + inputConfig.provider.mnemosyne.options = + inputConfig.provider.mnemosyne.options ?? {}; // Only set if not already overridden by user - if (!inputConfig.provider.anthropic.options.baseURL) { + if (!inputConfig.provider.mnemosyne.options.baseURL) { // Anthropic SDK appends /messages to baseURL, so we need /v1 suffix - inputConfig.provider.anthropic.options.baseURL = `${gateway.url}/v1`; - log.info(`Routing Anthropic through gateway at ${gateway.url}/v1`); + inputConfig.provider.mnemosyne.options.baseURL = `${gateway.url}/v1`; + log.info(`Routing Mnemosyne through gateway at ${gateway.url}/v1`); } else { log.debug( - `baseURL already set to ${inputConfig.provider.anthropic.options.baseURL}, not overriding` + `baseURL already set to ${inputConfig.provider.mnemosyne.options.baseURL}, not overriding` ); } }, From 6719d3f3f0ff8c7cc58a0298a4017954abf3ae4c Mon Sep 17 00:00:00 2001 From: Joey Yakimowich-Payne Date: Fri, 13 Mar 2026 15:40:47 -0600 Subject: [PATCH 03/10] fix: render collapsed turn summaries in outbound context Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-opencode) Co-authored-by: Sisyphus --- src/mnemosyne/blocks.py | 140 ++++++++++++++++++++++++++++------------ 1 file changed, 99 insertions(+), 41 deletions(-) diff --git a/src/mnemosyne/blocks.py b/src/mnemosyne/blocks.py index 05ec772..747b15e 100644 --- a/src/mnemosyne/blocks.py +++ b/src/mnemosyne/blocks.py @@ -20,15 +20,18 @@ from pathlib import Path @dataclass class BlockEntry: """A tracked conversation block.""" - block_id: str # Short hex ID (first 8 chars of content hash) - content_hash: str # Full SHA-256 of content - size: int # Byte size of original content - turn: int # Turn when first seen - role: str # "user" or "assistant" - preview: str # First 80 chars for logging + + block_id: str # Short hex ID (first 8 chars of content hash) + content_hash: str # Full SHA-256 of content + size: int # Byte size of original content + turn: int # Turn when first seen + role: str # "user" or "assistant" + preview: str # First 80 chars for logging status: str = "resident" # resident | anchored | summarized | dropped original_content: str | None = None # Full content for fault restoration summary: str | None = None # Model-authored summary (if summarized) + collapse_start_turn: int | None = None + collapse_end_turn: int | None = None class BlockStore: @@ -108,8 +111,7 @@ class BlockStore: size_k = entry.size / 1024 block["text"] = f"[tensor:{entry.block_id} ({size_k:.1f}KB)]\n{text}" - def _get_or_create(self, content: str, role: str, - turn: int) -> BlockEntry | None: + def _get_or_create(self, content: str, role: str, turn: int) -> BlockEntry | None: """Get existing entry by content hash, or create a new one.""" content_hash = hashlib.sha256(content.encode()).hexdigest() short_id = content_hash[:8] @@ -181,8 +183,7 @@ class BlockStore: entry.status = "anchored" return True - def collapse_range(self, start_turn: int, end_turn: int, - summary: str) -> list[str]: + def collapse_range(self, start_turn: int, end_turn: int, summary: str) -> list[str]: """Replace all blocks in a turn range with a summary marker. Marks all resident/anchored blocks in [start_turn, end_turn] as @@ -194,8 +195,7 @@ class BlockStore: """ collapsed_ids = [] for entry in self._by_id.values(): - if (start_turn <= entry.turn <= end_turn - and entry.status in ("resident", "anchored")): + if start_turn <= entry.turn <= end_turn and entry.status in ("resident", "anchored"): entry.status = "dropped" collapsed_ids.append(entry.block_id) @@ -203,9 +203,7 @@ class BlockStore: return [] # Create a synthetic summary block for the range - synthetic_content = ( - f"[Turns {start_turn}-{end_turn} collapsed: {summary}]" - ) + synthetic_content = f"[Turns {start_turn}-{end_turn} collapsed: {summary}]" content_hash = hashlib.sha256(synthetic_content.encode()).hexdigest() short_id = content_hash[:8] @@ -223,6 +221,8 @@ class BlockStore: preview=synthetic_content[:80], status="summarized", summary=summary, + collapse_start_turn=start_turn, + collapse_end_turn=end_turn, ) self._by_id[short_id] = entry self._by_hash[content_hash] = short_id @@ -237,23 +237,67 @@ class BlockStore: Modifies messages in-place. Returns stats dict. """ stats = {"dropped": 0, "summarized": 0, "anchored": 0} + emitted_collapses: set[str] = set() + filtered_messages: list[dict] = [] for msg in messages: content = msg.get("content", "") if isinstance(content, str): - msg["content"] = self._apply_to_text(content, msg, stats) + new_text = self._apply_to_text(content, stats, emitted_collapses) + if not new_text.strip(): + continue + msg["content"] = new_text + filtered_messages.append(msg) elif isinstance(content, list): + new_blocks = [] for block in content: - if not isinstance(block, dict) or block.get("type") != "text": + if not isinstance(block, dict): + new_blocks.append(block) continue - text = block.get("text", "") - block["text"] = self._apply_to_text(text, msg, stats) + if block.get("type") != "text": + new_blocks.append(block) + continue + + text = block.get("text", "") + new_text = self._apply_to_text(text, stats, emitted_collapses) + if not new_text.strip(): + continue + block["text"] = new_text + new_blocks.append(block) + + if not new_blocks: + continue + msg["content"] = new_blocks + filtered_messages.append(msg) + + else: + filtered_messages.append(msg) + + messages[:] = filtered_messages return stats - def _apply_to_text(self, text: str, msg: dict, stats: dict) -> str: + def _find_collapse_summary(self, turn: int) -> BlockEntry | None: + """Return the newest synthetic collapse summary covering this turn.""" + for entry in reversed(list(self._by_id.values())): + if ( + entry.status == "summarized" + and entry.summary + and entry.collapse_start_turn is not None + and entry.collapse_end_turn is not None + and entry.collapse_start_turn <= turn <= entry.collapse_end_turn + ): + return entry + return None + + def _apply_to_text( + self, + text: str, + stats: dict, + emitted_collapses: set[str], + ) -> str: """Apply block status to a single text content.""" m = self._BLOCK_LABEL_RE.match(text) if not m: @@ -265,18 +309,28 @@ class BlockStore: return text if entry.status == "dropped": + collapse_summary = self._find_collapse_summary(entry.turn) + if collapse_summary is not None: + if collapse_summary.block_id not in emitted_collapses: + emitted_collapses.add(collapse_summary.block_id) + stats["summarized"] += 1 + start_turn = collapse_summary.collapse_start_turn + end_turn = collapse_summary.collapse_end_turn + return ( + f"[tensor:{collapse_summary.block_id} — summarized turns " + f"{start_turn}-{end_turn}]\n" + f"{collapse_summary.summary}" + ) + stats["dropped"] += 1 + return "" + stats["dropped"] += 1 turn_info = f"message {entry.turn} in session log" - return ( - f"[...archived {entry.size:,} chars, {turn_info}...]" - ) + return f"[...archived {entry.size:,} chars, {turn_info}...]" if entry.status == "summarized" and entry.summary: stats["summarized"] += 1 - return ( - f"[tensor:{block_id} — summarized, was {entry.size:,} chars]\n" - f"{entry.summary}" - ) + return f"[tensor:{block_id} — summarized, was {entry.size:,} chars]\n{entry.summary}" # resident or anchored — no change if entry.status == "anchored": @@ -290,14 +344,12 @@ class BlockStore: @property def total_bytes(self) -> int: - return sum(e.size for e in self._by_id.values() - if e.status == "resident") + return sum(e.size for e in self._by_id.values() if e.status == "resident") def large_blocks(self, min_size: int = 2000) -> list[BlockEntry]: """Return resident blocks larger than min_size, sorted by size.""" return sorted( - [e for e in self._by_id.values() - if e.status == "resident" and e.size >= min_size], + [e for e in self._by_id.values() if e.status == "resident" and e.size >= min_size], key=lambda e: e.size, reverse=True, ) @@ -325,16 +377,20 @@ class BlockStore: """ entries = [] for entry in self._by_id.values(): - entries.append({ - "block_id": entry.block_id, - "content_hash": entry.content_hash, - "size": entry.size, - "turn": entry.turn, - "role": entry.role, - "preview": entry.preview, - "status": entry.status, - "summary": entry.summary, - }) + entries.append( + { + "block_id": entry.block_id, + "content_hash": entry.content_hash, + "size": entry.size, + "turn": entry.turn, + "role": entry.role, + "preview": entry.preview, + "status": entry.status, + "summary": entry.summary, + "collapse_start_turn": entry.collapse_start_turn, + "collapse_end_turn": entry.collapse_end_turn, + } + ) tmp = path.with_suffix(".tmp") tmp.write_text(json.dumps(entries, indent=2)) @@ -369,6 +425,8 @@ class BlockStore: preview=rec["preview"], status=rec.get("status", "resident"), summary=rec.get("summary"), + collapse_start_turn=rec.get("collapse_start_turn"), + collapse_end_turn=rec.get("collapse_end_turn"), original_content=None, ) store._by_id[entry.block_id] = entry From 2bf6baaa331ea4e16a7cf9585080994f11b16443 Mon Sep 17 00:00:00 2001 From: Joey Yakimowich-Payne Date: Fri, 13 Mar 2026 15:40:48 -0600 Subject: [PATCH 04/10] fix: rebuild session history for undo and segment only new messages Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-opencode) Co-authored-by: Sisyphus --- src/mnemosyne/gateway.py | 35 +++++++---- src/mnemosyne/message_store.py | 103 ++++++++++++++++++++++++++------- 2 files changed, 106 insertions(+), 32 deletions(-) diff --git a/src/mnemosyne/gateway.py b/src/mnemosyne/gateway.py index 92e31aa..af7fed0 100644 --- a/src/mnemosyne/gateway.py +++ b/src/mnemosyne/gateway.py @@ -1469,10 +1469,19 @@ def create_app( min_evict_size=min_evict_size, ) _bytes_saved += ingest.bytes_saved - if ingest.new_count > 0 or ingest.compacted_count > 0: + if ingest.physical_tail_deleted > 0: + session._segmented_objects = [ + obj + for obj in session._segmented_objects + if obj.turn_end < ingest.deleted_physical_start + ] + + if ingest.new_count > 0 or ingest.compacted_count > 0 or ingest.physical_tail_deleted > 0: parts = [] if ingest.new_count: parts.append(f"+{ingest.new_count} msgs") + if ingest.physical_tail_deleted: + parts.append(f"undo removed {ingest.physical_tail_deleted} msgs") if ingest.compacted_count: parts.append(f"{ingest.compacted_count} evicted") if ingest.bytes_saved: @@ -1486,17 +1495,21 @@ def create_app( file=sys.stderr, ) - # 1b. Segment new messages into semantic objects and store in ObjectStore - # Phase 4d: admission control gates each object before storage + # 1b. Segment only newly ingested physical messages into semantic objects + # and store them in ObjectStore. Re-segmenting the full history here + # would balloon object counts across turns. try: - with Timer(session.benchmark.latency["segmentation"]): - segmented = session.segmenter.segment_incremental( - ms.messages, - session._segmented_objects, - start_turn=0, - ) - new_count = len(segmented) - len(session._segmented_objects) - if new_count > 0: + if ingest.new_count > 0: + new_physical_messages = ms.messages[ + ingest.new_physical_start : ingest.new_physical_start + ingest.new_count + ] + with Timer(session.benchmark.latency["segmentation"]): + segmented = session.segmenter.segment_incremental( + new_physical_messages, + session._segmented_objects, + start_turn=ingest.new_physical_start, + ) + new_count = len(segmented) - len(session._segmented_objects) new_objects = segmented[len(session._segmented_objects) :] admitted_count = 0 rejected_count = 0 diff --git a/src/mnemosyne/message_store.py b/src/mnemosyne/message_store.py index 43bc5a3..eea0cb7 100644 --- a/src/mnemosyne/message_store.py +++ b/src/mnemosyne/message_store.py @@ -79,9 +79,13 @@ def _fingerprint(msg: dict) -> str: @dataclass class IngestResult: """Result of ingesting a new turn's messages.""" + new_count: int = 0 + new_physical_start: int = 0 mutations_detected: int = 0 deletions_detected: int = 0 + physical_tail_deleted: int = 0 + deleted_physical_start: int = 0 compacted_count: int = 0 bytes_saved: int = 0 @@ -89,8 +93,7 @@ class IngestResult: class MessageStore: """Pichay's compacted conversation history for a session.""" - def __init__(self, session_id: str, page_store: PageStore, - log_path: Path | None = None): + def __init__(self, session_id: str, page_store: PageStore, log_path: Path | None = None): self.session_id = session_id self.page_store = page_store self.log_path = log_path @@ -124,10 +127,16 @@ class MessageStore: return json.dumps(content, default=str)[:limit] return str(content)[:limit] - def _log_violation(self, kind: str, index: int, msg: dict | None, - expected_fp: str, actual_fp: str, - old_msg: dict | None = None, - deleted_msgs: list[dict] | None = None) -> None: + def _log_violation( + self, + kind: str, + index: int, + msg: dict | None, + expected_fp: str, + actual_fp: str, + old_msg: dict | None = None, + deleted_msgs: list[dict] | None = None, + ) -> None: """Log append-only violations to file for later analysis.""" if self.log_path is None: return @@ -188,20 +197,26 @@ class MessageStore: self._turn += 1 result = IngestResult() client_known = len(self._client_fps) + first_mutation_index: int | None = None # ── Detect mutations in known client messages ──────────── check_limit = min(client_known, len(incoming)) for i in range(check_limit): fp = _fingerprint(incoming[i]) if fp != self._client_fps[i]: + if first_mutation_index is None: + first_mutation_index = i result.mutations_detected += 1 self.total_mutations += 1 # Look up physical message via mapping for comparison phys_idx = self._client_to_physical[i] old_msg = self._messages[phys_idx] if phys_idx < len(self._messages) else None self._log_violation( - "mutation", i, incoming[i], - self._client_fps[i], fp, + "mutation", + i, + incoming[i], + self._client_fps[i], + fp, old_msg=old_msg, ) print( @@ -210,8 +225,28 @@ class MessageStore: f"got {fp[:32]}{_RESET}", file=sys.stderr, ) - # Update CLIENT fingerprint only — physical store unchanged - self._client_fps[i] = fp + + # Tail mutation: rebuild physical/client history from first changed index onward. + if first_mutation_index is not None: + if first_mutation_index < len(self._client_to_physical): + physical_start = self._client_to_physical[first_mutation_index] + else: + physical_start = len(self._messages) + physical_start = max(0, min(physical_start, len(self._messages))) + removed = len(self._messages) - physical_start + self._messages = self._messages[:physical_start] + self._fingerprints = self._fingerprints[:physical_start] + self._client_fps = self._client_fps[:first_mutation_index] + self._client_to_physical = self._client_to_physical[:first_mutation_index] + client_known = len(self._client_fps) + result.physical_tail_deleted = max(result.physical_tail_deleted, removed) + result.deleted_physical_start = physical_start + print( + f" {_DIM}[{self.session_id}] CLIENT TAIL MUTATION APPLIED: " + f"rebuilt history from client index {first_mutation_index} " + f"(removed {removed} physical msgs){_RESET}", + file=sys.stderr, + ) # ── Detect client deletions (compaction) ───────────────── if len(incoming) < client_known: @@ -219,6 +254,7 @@ class MessageStore: result.deletions_detected = deleted self.total_deletions += deleted self.total_client_deletions_absorbed += deleted + undo_applied = False # Log what the client is deleting (from physical store via mapping) deleted_physical = [] for ci in range(len(incoming), client_known): @@ -226,24 +262,49 @@ class MessageStore: if pi < len(self._messages): deleted_physical.append(self._messages[pi]) self._log_violation( - "deletion", client_known, None, - f"expected_{client_known}", f"got_{len(incoming)}", + "deletion", + client_known, + None, + f"expected_{client_known}", + f"got_{len(incoming)}", deleted_msgs=deleted_physical, ) - print( - f" {_DIM}[{self.session_id}] CLIENT DELETION ABSORBED: " - f"{deleted} messages dropped by client, " - f"physical store unchanged ({len(self._messages)} msgs){_RESET}", - file=sys.stderr, - ) + deleted_indices = self._client_to_physical[len(incoming) : client_known] + if deleted_indices: + valid_deleted = sorted( + {pi for pi in deleted_indices if 0 <= pi < len(self._messages)} + ) + if valid_deleted: + tail_start = valid_deleted[0] + removed = len(self._messages) - tail_start + self._messages = self._messages[:tail_start] + self._fingerprints = self._fingerprints[:tail_start] + result.physical_tail_deleted = removed + result.deleted_physical_start = tail_start + undo_applied = True + print( + f" {_DIM}[{self.session_id}] CLIENT UNDO APPLIED: " + f"removed {removed} tail messages from physical store{_RESET}", + file=sys.stderr, + ) + + if not undo_applied: + print( + f" {_DIM}[{self.session_id}] CLIENT DELETION ABSORBED: " + f"{deleted} messages dropped by client, " + f"physical store unchanged ({len(self._messages)} msgs){_RESET}", + file=sys.stderr, + ) + # Truncate CLIENT tracking only — physical store stays intact - self._client_fps = self._client_fps[:len(incoming)] - self._client_to_physical = self._client_to_physical[:len(incoming)] + self._client_fps = self._client_fps[: len(incoming)] + self._client_to_physical = self._client_to_physical[: len(incoming)] # ── Extract and append new messages ────────────────────── new_start = min(client_known, len(incoming)) new_messages = incoming[new_start:] result.new_count = len(new_messages) + result.new_physical_start = len(self._messages) if new_messages: # Deep copy new messages so we own them @@ -256,7 +317,7 @@ class MessageStore: _strip_cache_control(msg) # Track in physical store and client mapping - phys_start = len(self._messages) + phys_start = result.new_physical_start for j, msg in enumerate(new_messages): fp = _fingerprint(msg) self._fingerprints.append(fp) From 92fba55f700632e09d71db0d0b9d65c21b3b481e Mon Sep 17 00:00:00 2001 From: Joey Yakimowich-Payne Date: Fri, 13 Mar 2026 21:07:52 -0600 Subject: [PATCH 05/10] fix: accurate context reduction stats and SSE cleanup tag filter Measure incoming_bytes before _preprocess() so bytes_saved reflects true reduction. Add SSECleanupFilter that intercepts memory_cleanup/yuyay-response tags in streaming responses, strips them from output, and executes ops (drops, collapses, releases) in real-time. Handles partial tags split across SSE chunks with a safety valve to flush stale buffers for prose. --- src/mnemosyne/benchmark.py | 5 +- src/mnemosyne/gateway.py | 282 ++++++++++++++++++++++++++++++++++++- 2 files changed, 282 insertions(+), 5 deletions(-) diff --git a/src/mnemosyne/benchmark.py b/src/mnemosyne/benchmark.py index ce50978..d7d5301 100644 --- a/src/mnemosyne/benchmark.py +++ b/src/mnemosyne/benchmark.py @@ -477,7 +477,10 @@ class BenchmarkCollector: "max_ms": round(max_time, 2), } - context_reduction = total_outgoing / total_incoming if total_incoming > 0 else 1.0 + total_bytes_saved = sum(s.tokens.total_bytes_saved for s in sessions) + context_reduction = ( + min(total_bytes_saved / total_incoming, 1.0) if total_incoming > 0 else 0.0 + ) return { "sessions": n, diff --git a/src/mnemosyne/gateway.py b/src/mnemosyne/gateway.py index af7fed0..89fa156 100644 --- a/src/mnemosyne/gateway.py +++ b/src/mnemosyne/gateway.py @@ -57,8 +57,274 @@ from mnemosyne.object_store import ObjectStoreBackend from mnemosyne.pager import PageStore, compact_messages from mnemosyne.message_store import MessageStore from mnemosyne.providers import adapters +from mnemosyne.tags import ( + parse_cleanup_tags, + parse_yuyay_response, + strip_cleanup_tags, + strip_yuyay_tags, +) from mnemosyne.telemetry import Telemetry +import re as _re + +# --------------------------------------------------------------------------- +# SSE Cleanup Filter — strips / tags from +# streaming responses and executes the contained ops in real-time. +# --------------------------------------------------------------------------- + +_TAG_OPEN_RE = _re.compile(r"<(memory_cleanup|yuyay-response)") +_TAG_CLOSE_RE = _re.compile(r"") + +# Matches a trailing '<' optionally followed by a prefix of a known tag name +# or ' bool: + """Return True if *buf* ends with a partial tag opener we care about. + + Only checks the last 25 characters — a complete tag name is at most + ```` (19 chars). This prevents false positives when + the model writes prose containing '<' earlier in the buffer. + """ + # Only look at the tail of the buffer for partial tags + window = buf[-25:] if len(buf) > 25 else buf + idx = window.rfind("<") + if idx == -1: + return False + tail = window[idx + 1 :] # everything after the last '<' + if not tail: + return True # bare '<' at the very end — could be start of any tag + # If the tail contains '>' then the tag is already closed — not partial + if ">" in tail: + return False + return any(p.startswith(tail) for p in _KNOWN_TAG_PREFIXES) + + +class SSECleanupFilter: + """Intercept SSE text deltas and strip/execute cleanup tags. + + Anthropic SSE streams emit ``content_block_delta`` events with + ``{"delta": {"type": "text_delta", "text": "..."}}``. This filter + accumulates those text fragments, detects complete + ```` / ```` blocks, executes the + operations against the session's BlockStore / PageStore, and rewrites + the SSE ``data:`` lines with the tags removed. + + Non-text events (ping, message_start, content_block_start, etc.) + pass through unchanged. + """ + + # Max text deltas to buffer while waiting for a tag to complete. + # Real tags complete within 2-3 deltas. If we exceed this, it's prose. + _MAX_BUFFERED_DELTAS = 6 + + def __init__(self, block_store: "BlockStore | None", page_store: "PageStore | None"): + self._bs = block_store + self._ps = page_store + self._buf = "" # accumulated text not yet flushed + self._inside_tag = False # currently buffering a tag body + self._buffered_deltas = 0 # how many deltas we've buffered without flushing + self._stats: list[str] = [] # executed ops log + + # ------------------------------------------------------------------ + # Public API + # ------------------------------------------------------------------ + + def filter_chunk(self, raw_chunk: bytes) -> bytes: + """Process a raw SSE chunk (may contain multiple lines). + + Returns the (possibly rewritten) chunk to forward downstream. + """ + lines = raw_chunk.split(b"\n") + out_lines: list[bytes] = [] + + for line in lines: + if not line.startswith(b"data: "): + out_lines.append(line) + continue + + json_bytes = line[6:] # strip "data: " prefix + if json_bytes.strip() in (b"", b"[DONE]"): + out_lines.append(line) + continue + + try: + event = json.loads(json_bytes) + except (json.JSONDecodeError, UnicodeDecodeError): + out_lines.append(line) + continue + + if event.get("type") != "content_block_delta": + out_lines.append(line) + continue + + delta = event.get("delta", {}) + if delta.get("type") != "text_delta": + out_lines.append(line) + continue + + text = delta.get("text", "") + if not text: + out_lines.append(line) + continue + + # Feed text into buffer and get cleaned output + cleaned = self._feed(text) + + if cleaned is None: + # Entire chunk is being buffered (inside a tag) — suppress + continue + elif cleaned == text: + # No change — pass through original bytes + out_lines.append(line) + elif not cleaned: + # Text was entirely a cleanup tag — suppress this event + continue + else: + # Rewrite the delta with cleaned text + delta["text"] = cleaned + out_lines.append(b"data: " + json.dumps(event, ensure_ascii=False).encode("utf-8")) + + return b"\n".join(out_lines) + + @property + def stats(self) -> str: + return "; ".join(self._stats) if self._stats else "" + + def flush(self) -> str: + """Flush any remaining buffered text (call at stream end).""" + if self._buf: + result = self._buf + self._buf = "" + self._inside_tag = False + return result + return "" + + # ------------------------------------------------------------------ + # Internal + # ------------------------------------------------------------------ + + def _feed(self, text: str) -> str | None: + """Feed a text delta fragment. Returns cleaned text or None to suppress. + + Returns: + str — cleaned text to emit (may be empty string to skip event) + None — suppress entirely (buffering inside a tag) + """ + self._buf += text + self._buffered_deltas += 1 + + has_open = bool(_TAG_OPEN_RE.search(self._buf)) + has_close = bool(_TAG_CLOSE_RE.search(self._buf)) + partial = _has_partial_tag(self._buf) + + # Safety valve: if we've buffered too many deltas without resolving + # a tag, this is just prose containing '<' — flush everything. + if self._buffered_deltas > self._MAX_BUFFERED_DELTAS and not has_close: + result = self._buf + self._buf = "" + self._inside_tag = False + self._buffered_deltas = 0 + return result + + # Fast path: no tag markers and no partial tag at the end + if not self._inside_tag and not has_open and not partial: + result = self._buf + self._buf = "" + self._buffered_deltas = 0 + return result + + # Partial tag opener at the end (e.g. " 0: + emit = self._buf[:idx] + self._buf = self._buf[idx:] + return emit + return None + + # We see a complete tag opening + if not self._inside_tag and has_open: + self._inside_tag = True + m = _TAG_OPEN_RE.search(self._buf) + if m and m.start() > 0: + emit = self._buf[: m.start()] + self._buf = self._buf[m.start() :] + return emit + + # Check if we have a complete tag + if self._inside_tag and has_close: + self._execute_ops(self._buf) + cleaned = strip_yuyay_tags(strip_cleanup_tags(self._buf)) + self._buf = "" + self._inside_tag = False + self._buffered_deltas = 0 + + if _TAG_OPEN_RE.search(cleaned): + self._buf = cleaned + self._inside_tag = True + return None + if _has_partial_tag(cleaned): + idx = cleaned.rfind("<") + if idx > 0: + emit = cleaned[:idx] + self._buf = cleaned[idx:] + return emit + self._buf = cleaned + return None + + return cleaned + + # Still inside an incomplete tag — keep buffering + if self._inside_tag: + return None + + # Shouldn't reach here, but safety + result = self._buf + self._buf = "" + self._buffered_deltas = 0 + return result + + def _execute_ops(self, text: str) -> None: + """Parse and execute cleanup/yuyay ops from buffered text.""" + ops_list = [] + + cleanup_ops = parse_cleanup_tags(text) + if not cleanup_ops.empty: + ops_list.append(cleanup_ops) + + yuyay_ops = parse_yuyay_response(text) + if not yuyay_ops.empty: + ops_list.append(yuyay_ops) + + for ops in ops_list: + if self._bs is not None: + for block_id in ops.drops: + if self._bs.drop(block_id): + self._stats.append(f"dropped {block_id}") + for block_id, summary in ops.summaries: + if self._bs.summarize(block_id, summary): + self._stats.append(f"summarized {block_id}") + for block_id in ops.anchors: + if self._bs.anchor(block_id): + self._stats.append(f"anchored {block_id}") + for collapse in ops.collapses: + collapsed = self._bs.collapse_range( + collapse.start_turn, collapse.end_turn, collapse.summary + ) + if collapsed: + self._stats.append( + f"collapsed turns {collapse.start_turn}-{collapse.end_turn} " + f"({len(collapsed)} blocks)" + ) + if self._ps is not None and ops.releases: + for path in ops.releases: + self._ps.mark_released(path) + self._stats.append(f"released {len(ops.releases)} path(s)") + # ANSI for stderr status lines _DIM = "\033[2m" @@ -2011,6 +2277,9 @@ def create_app( # Cleanup now runs inside _preprocess (before manifest injection) + # Measure raw incoming size BEFORE any preprocessing + incoming_bytes = len(json.dumps(payload, default=str).encode("utf-8")) + # Pre-process: system status, block labeling preprocess_bytes_saved = 0 if endpoint == "messages": @@ -2027,7 +2296,6 @@ def create_app( request_id = str(uuid.uuid4()) started = time.perf_counter() - incoming_bytes = len(json.dumps(payload, default=str).encode("utf-8")) session_id = session.id req = adapter.normalize_request(payload) @@ -2087,6 +2355,10 @@ def create_app( bytes_out = len(body) yield body else: + cleanup_filter = SSECleanupFilter( + block_store=session.block_store, + page_store=session.page_store, + ) for chunk in resp.iter_bytes(): bytes_out += len(chunk) chunk_count += 1 @@ -2099,7 +2371,9 @@ def create_app( provider=provider, usage_accumulator=usage, ) - yield chunk + filtered = cleanup_filter.filter_chunk(chunk) + if filtered: + yield filtered except Exception as e: stream_error = True emit_event( @@ -2121,7 +2395,7 @@ def create_app( session, incoming_bytes=incoming_bytes, outgoing_bytes=outgoing_bytes, - bytes_saved=preprocess_bytes_saved if endpoint == "messages" else 0, + bytes_saved=max(0, incoming_bytes - outgoing_bytes), ) _update_fidelity_pressure(usage, session) emit_event( @@ -2183,7 +2457,7 @@ def create_app( session, incoming_bytes=incoming_bytes, outgoing_bytes=outgoing_bytes, - bytes_saved=preprocess_bytes_saved, + bytes_saved=max(0, incoming_bytes - outgoing_bytes), ) _update_fidelity_pressure(usage, session) From 65e4e38a98ffcab0b8c6c75f2294e11f81d59365 Mon Sep 17 00:00:00 2001 From: Joey Yakimowich-Payne Date: Fri, 13 Mar 2026 21:13:38 -0600 Subject: [PATCH 06/10] fix: scale FM window_size to match real API pressure for fidelity degradation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The FidelityManager's internal pressure calculation uses its own tracked object tokens divided by window_size, which is always tiny compared to the real context. Temporarily scale window_size so the FM's pressure matches the actual API input_tokens/window ratio, triggering L0→L1→L2 degradations when context exceeds 50%. --- src/mnemosyne/gateway.py | 22 +++++++++++++++------- 1 file changed, 15 insertions(+), 7 deletions(-) diff --git a/src/mnemosyne/gateway.py b/src/mnemosyne/gateway.py index 89fa156..e8057de 100644 --- a/src/mnemosyne/gateway.py +++ b/src/mnemosyne/gateway.py @@ -2122,8 +2122,8 @@ def create_app( """Update FidelityManager window understanding from actual API usage. After receiving the response, we know the real input token count. - Update the fidelity manager's window_size understanding and schedule - degradation if pressure is above NORMAL for the next turn. + Scale the FM's window_size so its internal pressure calculation + reflects the real API token usage, then trigger degradation. """ input_tokens = usage.get("input_tokens", 0) if input_tokens <= 0: @@ -2132,14 +2132,22 @@ def create_app( fm = session.fidelity_manager turn = session.token_state.get("turn", 0) - # The FidelityManager tracks its own token budget via registered objects. - # Here we use the real API token count to check if we need proactive degradation. - # If real usage exceeds the fidelity window threshold, trigger degradation now - # so the NEXT turn benefits from reduced content. pressure_ratio = input_tokens / fm.window_size if fm.window_size > 0 else 1.0 if pressure_ratio >= fm.threshold_caution: - transitions = fm.degrade(turn) + # The FM's internal pressure uses total_tokens()/window_size, but + # total_tokens() only counts registered objects (a fraction of the + # real context). Temporarily scale window_size down so the FM's + # pressure matches the real API pressure, then restore it. + obj_tokens = fm.total_tokens() + if obj_tokens > 0: + # Set window_size so obj_tokens/window_size == pressure_ratio + saved_ws = fm.window_size + fm.window_size = max(1, int(obj_tokens / pressure_ratio)) + transitions = fm.degrade(turn) + fm.window_size = saved_ws + else: + transitions = fm.degrade(turn) if transitions: zone = fm.current_pressure() for _obj_id, old_level, new_level in transitions: From ad2c296ba3b83eab79c0cb90188893cc95b9b319 Mon Sep 17 00:00:00 2001 From: Joey Yakimowich-Payne Date: Fri, 13 Mar 2026 21:23:26 -0600 Subject: [PATCH 07/10] fix: parse XML-format cleanup tags and strip from SSE stream The model emits cleanup ops as XML elements (block:x, , turns N-M "summary") but the parser only handled prose format (drop: block:x). Add XML regex matchers alongside the existing prose parser so both formats are recognized, executed, and stripped from the streaming output. --- src/mnemosyne/tags.py | 71 ++++++++++++++++++++++++++++++------------- 1 file changed, 50 insertions(+), 21 deletions(-) diff --git a/src/mnemosyne/tags.py b/src/mnemosyne/tags.py index a200911..f38af67 100644 --- a/src/mnemosyne/tags.py +++ b/src/mnemosyne/tags.py @@ -22,6 +22,7 @@ from dataclasses import dataclass, field @dataclass class CollapseOp: """A turn-range collapse: replace multiple turns with a summary.""" + start_turn: int end_turn: int summary: str @@ -30,6 +31,7 @@ class CollapseOp: @dataclass class CleanupOps: """Parsed cleanup operations from a tag.""" + drops: list[str] = field(default_factory=list) summaries: list[tuple[str, str]] = field(default_factory=list) anchors: list[str] = field(default_factory=list) @@ -38,8 +40,7 @@ class CleanupOps: @property def empty(self) -> bool: - return not (self.drops or self.summaries or self.anchors - or self.releases or self.collapses) + return not (self.drops or self.summaries or self.anchors or self.releases or self.collapses) def __str__(self) -> str: parts = [] @@ -65,67 +66,95 @@ _TAG_PATTERN = re.compile( # Match tensor/block ID references: tensor:xxxxxxxx or block:xxxxxxxx (8-12 hex chars) _BLOCK_ID = re.compile(r"(?:tensor|block):([a-f0-9]{8,12})(?![a-f0-9])") -# Match summarize with quoted summary text +# Match summarize with quoted summary text (prose format) _SUMMARIZE_PATTERN = re.compile( r'summarize:\s*(?:tensor|block):([a-f0-9]{8,12})(?![a-f0-9])\s+"([^"]*)"' ) -# Match release with comma-separated paths +# Match release with comma-separated paths (prose format) _RELEASE_PATTERN = re.compile(r"release:\s*(.+)") # Match collapse with turn range and quoted summary # Format: collapse: turns 3-8 "Summary of what happened in those turns" -_COLLAPSE_PATTERN = re.compile( - r'collapse:\s*turns\s+(\d+)\s*-\s*(\d+)\s+"([^"]*)"' -) +# Also: turns 3-8 "Summary" +_COLLAPSE_PATTERN = re.compile(r'collapse[\s:>]+turns\s+(\d+)\s*-\s*(\d+)\s+"([^"]*)"') + +# --- XML element variants (model sometimes uses XML instead of prose) --- +# block:4ab5ee7a or tensor:4ab5ee7a +_XML_DROP = re.compile(r"(?:tensor|block):([a-f0-9]{8,12})") +# or +_XML_RELEASE = re.compile(r'block:abc123 +_XML_ANCHOR = re.compile(r"(?:tensor|block):([a-f0-9]{8,12})") +# summary text +_XML_SUMMARIZE = re.compile(r']*>([^<]*)') +# turns 3-8 "Summary" +_XML_COLLAPSE = re.compile(r'turns\s+(\d+)\s*-\s*(\d+)\s+"([^"]*)"') def parse_cleanup_tags(text: str) -> CleanupOps: - """Extract cleanup operations from text containing tags. + """Extract memory operations from ```` blocks. - Returns a CleanupOps with all parsed operations. Multiple tags in - the same text are merged into a single CleanupOps. + Supports both prose format (``drop: block:abc``) and XML element + format (``block:abc``). """ ops = CleanupOps() for match in _TAG_PATTERN.finditer(text): body = match.group(1) + + # --- XML element variants (scan full body first) --- + for m in _XML_DROP.finditer(body): + ops.drops.append(m.group(1)) + for m in _XML_RELEASE.finditer(body): + ops.releases.append(m.group(1)) + for m in _XML_ANCHOR.finditer(body): + ops.anchors.append(m.group(1)) + for m in _XML_SUMMARIZE.finditer(body): + ops.summaries.append((m.group(1), m.group(2))) + for m in _XML_COLLAPSE.finditer(body): + ops.collapses.append( + CollapseOp( + start_turn=int(m.group(1)), + end_turn=int(m.group(2)), + summary=m.group(3), + ) + ) + + # --- Prose line-based format --- for line in body.splitlines(): line = line.strip() - if not line: + if not line or line.startswith("<"): continue - # Summarize (must check before drop — both start with block ID) m = _SUMMARIZE_PATTERN.match(line) if m: ops.summaries.append((m.group(1), m.group(2))) continue - # Drop if line.startswith("drop:"): m = _BLOCK_ID.search(line) if m: ops.drops.append(m.group(1)) continue - # Anchor if line.startswith("anchor:"): m = _BLOCK_ID.search(line) if m: ops.anchors.append(m.group(1)) continue - # Collapse (turn range) m = _COLLAPSE_PATTERN.match(line) if m: - ops.collapses.append(CollapseOp( - start_turn=int(m.group(1)), - end_turn=int(m.group(2)), - summary=m.group(3), - )) + ops.collapses.append( + CollapseOp( + start_turn=int(m.group(1)), + end_turn=int(m.group(2)), + summary=m.group(3), + ) + ) continue - # Release m = _RELEASE_PATTERN.match(line) if m: paths = [p.strip() for p in m.group(1).split(",") if p.strip()] From e0af1edadf13bca525a52aa4d6a8dd69bce34fde Mon Sep 17 00:00:00 2001 From: Joey Yakimowich-Payne Date: Fri, 13 Mar 2026 21:27:08 -0600 Subject: [PATCH 08/10] fix: safety valve only flushes partial openers, not real tags MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Long cleanup tags (e.g. collapse summaries) can span 30+ SSE deltas. The safety valve was flushing after 6 deltas regardless, dumping incomplete tags into the output. Now only flushes when buffering a partial opener ( self._MAX_BUFFERED_DELTAS and not has_close: + # Safety valve: if we've buffered many deltas waiting for a *partial* + # opener to resolve (e.g. ""), + # flush everything. But NEVER flush when _inside_tag is True — that + # means we matched a real opening tag and are waiting for the close. + # Real tags (e.g. long collapse summaries) can span 30+ deltas. + if ( + self._buffered_deltas > self._MAX_BUFFERED_DELTAS + and not self._inside_tag + and not has_close + ): result = self._buf self._buf = "" self._inside_tag = False From 2c42f9b52a69af8b3eda31096a85c9321f87d97e Mon Sep 17 00:00:00 2001 From: Joey Yakimowich-Payne Date: Fri, 13 Mar 2026 21:35:19 -0600 Subject: [PATCH 09/10] fix: assign conversation turn numbers to blocks and add /api/blocks debug endpoint Blocks were all getting turn=1 because label_messages used a single global counter. Now derives turn from message position (each user msg increments the turn). Also updates turn on already-labeled blocks. Adds /api/blocks endpoint to inspect BlockStore state per session. This enables collapse_range(1,72) to correctly target early turns. --- src/mnemosyne/blocks.py | 30 +++++++++++++++++++++++++++--- src/mnemosyne/gateway.py | 34 ++++++++++++++++++++++++++++++++++ 2 files changed, 61 insertions(+), 3 deletions(-) diff --git a/src/mnemosyne/blocks.py b/src/mnemosyne/blocks.py index 747b15e..5150581 100644 --- a/src/mnemosyne/blocks.py +++ b/src/mnemosyne/blocks.py @@ -67,11 +67,22 @@ class BlockStore: its content hash. Labels are stable across turns as long as the content doesn't change. + Turn numbers are derived from message position: each user/assistant + pair is one conversation turn (user msg at index i → turn i//2 + 1). + This ensures ``collapse_range(1, 72)`` targets the right messages. + Only labels user and assistant text messages. Tool_use and tool_result blocks are managed by the PageStore, not here. """ - for msg in messages: + # Compute per-message turn based on position in conversation + turn_counter = 0 + for i, msg in enumerate(messages): role = msg.get("role", "") + # Increment turn on each user message (user+assistant = 1 turn) + if role == "user": + turn_counter += 1 + msg_turn = turn_counter if turn_counter > 0 else 1 + if role not in ("user", "assistant"): continue @@ -81,12 +92,14 @@ class BlockStore: if isinstance(content, str): # Skip if already labeled by us (validated against known IDs) if self._has_our_label(content): + # Update turn on existing block if it changed + self._update_turn(content, msg_turn) continue # Skip very short messages (not worth labeling) if len(content) < 200: continue - entry = self._get_or_create(content, role, current_turn) + entry = self._get_or_create(content, role, msg_turn) if entry and entry.status == "resident": size_k = entry.size / 1024 msg["content"] = f"[tensor:{entry.block_id} ({size_k:.1f}KB)]\n{content}" @@ -101,16 +114,27 @@ class BlockStore: text = block.get("text", "") # Skip if already labeled by us (validated against known IDs) if self._has_our_label(text): + self._update_turn(text, msg_turn) continue # Skip short blocks if len(text) < 200: continue - entry = self._get_or_create(text, role, current_turn) + entry = self._get_or_create(text, role, msg_turn) if entry and entry.status == "resident": size_k = entry.size / 1024 block["text"] = f"[tensor:{entry.block_id} ({size_k:.1f}KB)]\n{text}" + def _update_turn(self, labeled_content: str, turn: int) -> None: + """Update the turn number on an already-labeled block.""" + import re + + m = re.match(r"\[tensor:([a-f0-9]{8,12})", labeled_content) + if m: + entry = self._by_id.get(m.group(1)) + if entry: + entry.turn = turn + def _get_or_create(self, content: str, role: str, turn: int) -> BlockEntry | None: """Get existing entry by content hash, or create a new one.""" content_hash = hashlib.sha256(content.encode()).hexdigest() diff --git a/src/mnemosyne/gateway.py b/src/mnemosyne/gateway.py index 44591e0..998d727 100644 --- a/src/mnemosyne/gateway.py +++ b/src/mnemosyne/gateway.py @@ -1636,6 +1636,40 @@ def create_app( "session_tokens": total_tokens, } + @app.get("/api/blocks") + async def api_blocks(session_id: str | None = None) -> dict[str, Any]: + """Debug endpoint: expose BlockStore state per session.""" + all_sessions = sessions.all() + out: dict[str, Any] = {} + for sid, sess in all_sessions.items(): + if session_id and sid != session_id: + continue + bs = sess.block_store + blocks = [] + for bid, entry in bs._by_id.items(): + blocks.append( + { + "id": bid, + "status": entry.status, + "turn": entry.turn, + "role": entry.role, + "size": entry.size, + "preview": entry.preview[:80] if entry.preview else "", + "summary": entry.summary, + "collapse_start": getattr(entry, "collapse_start_turn", None), + "collapse_end": getattr(entry, "collapse_end_turn", None), + } + ) + out[sid] = { + "total_blocks": len(blocks), + "by_status": {}, + "blocks": blocks, + } + for b in blocks: + s = b["status"] + out[sid]["by_status"][s] = out[sid]["by_status"].get(s, 0) + 1 + return out + @app.get("/api/compaction-context") async def api_compaction_context( session_id: str, From f5c2c91057fa82f9af0648564cd6fcacc628e7e8 Mon Sep 17 00:00:00 2001 From: Joey Yakimowich-Payne Date: Fri, 13 Mar 2026 21:39:48 -0600 Subject: [PATCH 10/10] fix: remove orphaned SSE event headers when suppressing text deltas When the cleanup filter suppresses a text delta (buffering inside a tag), the preceding 'event: content_block_delta' header was left in the output, producing malformed SSE that caused opencode to retry rapidly and freeze. Now removes the event header alongside the data line. --- src/mnemosyne/gateway.py | 18 +++++++++++++----- 1 file changed, 13 insertions(+), 5 deletions(-) diff --git a/src/mnemosyne/gateway.py b/src/mnemosyne/gateway.py index 998d727..e1a9b4d 100644 --- a/src/mnemosyne/gateway.py +++ b/src/mnemosyne/gateway.py @@ -136,12 +136,17 @@ class SSECleanupFilter: """Process a raw SSE chunk (may contain multiple lines). Returns the (possibly rewritten) chunk to forward downstream. + When a text delta is suppressed (buffering inside a tag), the + preceding ``event:`` header line is also removed to avoid + producing malformed SSE (event header with no data). """ lines = raw_chunk.split(b"\n") out_lines: list[bytes] = [] for line in lines: if not line.startswith(b"data: "): + # Buffer event: lines — only emit them if the next data: line + # is kept. Non-event lines (empty lines, comments) pass through. out_lines.append(line) continue @@ -173,15 +178,18 @@ class SSECleanupFilter: # Feed text into buffer and get cleaned output cleaned = self._feed(text) - if cleaned is None: - # Entire chunk is being buffered (inside a tag) — suppress + if cleaned is None or not cleaned: + # Suppressed — also remove the preceding "event:" line + # to avoid malformed SSE (event header with no data) + if out_lines and out_lines[-1].startswith(b"event:"): + out_lines.pop() + # Also remove trailing empty line if present + if out_lines and out_lines[-1] == b"": + out_lines.pop() continue elif cleaned == text: # No change — pass through original bytes out_lines.append(line) - elif not cleaned: - # Text was entirely a cleanup tag — suppress this event - continue else: # Rewrite the delta with cleaned text delta["text"] = cleaned