From 5702a5a1e24e583a7d8cee26ebf2bc85b979ba22 Mon Sep 17 00:00:00 2001 From: Joey Yakimowich-Payne Date: Fri, 13 Mar 2026 13:37:33 -0600 Subject: [PATCH] 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)