diff --git a/amplifier_module_context_simple/__init__.py b/amplifier_module_context_simple/__init__.py index 7baa8d2..ccc54ac 100644 --- a/amplifier_module_context_simple/__init__.py +++ b/amplifier_module_context_simple/__init__.py @@ -170,13 +170,28 @@ def __init__( # potentially shifting) on every single get_messages_for_request() # call. See _apply_sticky_decisions() / _compact_ephemeral(). self._next_seq: int = 0 - self._removed_seqs: set[int] = set() + # seq -> why it was removed. A bare set answered "was this removed?" + # but never "why", and every diagnosis during the incident + # investigation had to reconstruct intent from logs that no longer + # existed. Membership tests, `len()`, and iteration are identical on a + # dict, so this is prefix-neutral by construction: no compaction + # decision reads the reason. + self._removed_seqs: dict[int, str] = {} self._truncated_seqs: set[int] = set() self._stubbed_seqs: set[int] = set() # Cumulative highest progressive strategy level (1-8) ever reached. # Reported in compaction stats / notice so the LLM sees the total # accumulated effect, not just the most recent escalation step. self._sticky_level: int = 0 + # Log-once latch for an arithmetically unreachable target. Reset with + # the rest of the sticky state so a genuinely new situation speaks up. + self._infeasible_target_reported: bool = False + # Runaway-compaction breaker. Counts CONSECUTIVE escalations that failed + # to meaningfully reduce the post-compaction token count. Frequency alone + # is the wrong signal -- see `_MAX_INEFFECTIVE_ESCALATIONS`. + self._ineffective_escalations: int = 0 + self._last_after_tokens: int | None = None + self._escalation_breaker_reported: bool = False async def add_message(self, message: dict[str, Any]) -> None: """Add a message to the context. @@ -320,6 +335,13 @@ async def get_messages_for_request( token_count = self._estimate_tokens(working_messages) # Check if compaction needed (using effective budget with notice reserve deducted) + if not self._should_compact(token_count, effective_budget): + # No compaction needed at all: the strongest possible evidence that + # pressure is relieved, so the runaway breaker re-arms here too. + # Its bookkeeping otherwise lives inside `_compact_ephemeral`, which + # by definition does not run on this path -- leaving a stale count + # that could trip the breaker on an unrelated later burst. + self._ineffective_escalations = 0 if self._should_compact(token_count, effective_budget): # Compact EPHEMERALLY - returns new list, working_messages unchanged compacted = await self._compact_ephemeral( @@ -484,10 +506,14 @@ async def set_messages(self, messages: list[dict[str, Any]]) -> None: restamped.append({**msg, "metadata": meta}) self.messages = restamped self._next_seq = len(restamped) - self._removed_seqs = set() + self._removed_seqs = {} self._truncated_seqs = set() self._stubbed_seqs = set() self._sticky_level = 0 + self._infeasible_target_reported = False + self._ineffective_escalations = 0 + self._last_after_tokens = None + self._escalation_breaker_reported = False self._last_compaction_stats = None logger.info(f"Restored {len(messages)} messages to context") @@ -495,10 +521,14 @@ async def clear(self) -> None: """Clear all messages.""" self.messages = [] self._next_seq = 0 - self._removed_seqs = set() + self._removed_seqs = {} self._truncated_seqs = set() self._stubbed_seqs = set() self._sticky_level = 0 + self._infeasible_target_reported = False + self._ineffective_escalations = 0 + self._last_after_tokens = None + self._escalation_breaker_reported = False self._last_compaction_stats = None logger.info("Context cleared") @@ -555,11 +585,16 @@ def _extract_seq(msg: dict[str, Any]) -> int | None: """ return (msg.get("metadata") or {}).get("_seq") - def _record_removed(self, msg: dict[str, Any]) -> None: - """Permanently record that a message has been removed by compaction.""" + def _record_removed(self, msg: dict[str, Any], reason: str = "unspecified") -> None: + """Permanently record that a message has been removed by compaction. + + *reason* is diagnostic only -- nothing in the compaction path reads it. + It exists because "663 messages were removed" is not actionable and + "663 messages were removed chasing an unreachable target at level 8" is. + """ seq = self._extract_seq(msg) if seq is not None: - self._removed_seqs.add(seq) + self._removed_seqs[seq] = reason # A message can only be in one terminal state; removal supersedes # any earlier truncate/stub decision for the same seq. self._truncated_seqs.discard(seq) @@ -720,6 +755,83 @@ async def _compact_ephemeral( needs_escalation = ( budget > 0 and (current_tokens / budget) >= self.compact_threshold ) + + # FEASIBILITY PRE-CHECK. System messages are never compacted, so once + # their share alone exceeds the target, the target is arithmetically + # unreachable -- no escalation level can ever get there. Escalating + # anyway does not fail; it deletes real conversation on every request to + # chase a number that cannot come down, pinned at maximum level, while + # `after_tokens` never moves. + # + # Measured on this module before the check existed, with a 12,153-token + # system prompt against a 40,000 budget (target 10,000) and NO images: + # level pinned at 8 from the fifth call, `after_tokens` stuck at ~14,776, + # and `_removed_seqs` ratcheting 6 -> 12 -> 18 -> 30 -> 42 -> 54 of 58 + # messages ever added, the returned view sawtoothing 4 -> 8 -> 4 as + # history regrew and was destroyed again. Silently: the existing + # over-budget warning below is gated on `> budget`, and this state sits + # at 37% of budget. + # + # Only skip while the view still fits the ACTUAL budget. Over budget, + # a partial reduction beats none and we escalate as before. + target_unreachable = system_tokens > target_tokens + if needs_escalation and target_unreachable and current_tokens <= budget: + if not self._infeasible_target_reported: + logger.warning( + f"Compaction target is unreachable and further escalation would only " + f"destroy conversation: the system prompt alone is {system_tokens:,} " + f"tokens against a target of {target_tokens:,}. System messages are " + f"never compacted, so no level can reach the target. The view is " + f"{current_tokens:,} tokens, still within the {budget:,} budget, so it " + f"is being returned as-is. Reduce the system prompt or raise the budget." + ) + self._infeasible_target_reported = True + needs_escalation = False + + # RUNAWAY BREAKER. + # + # Every guard above tests whether THIS call's target is reachable. None + # of them notices that the same answer has been reached over and over. + # The incident ran 235 compactions across 266 model calls, every one at + # maximum level, `after_tokens` never moving -- and the only signal was + # an INFO line that fired 235 times and nobody saw. + # + # FREQUENCY IS THE WRONG SIGNAL, and that was the first design here. + # A session genuinely over budget must keep compacting on every call; + # freezing it there converts "destroying conversation" into "guaranteed + # provider rejection", which is worse. The test that caught this is + # `test_going_over_budget_still_escalates` -- under a frequency breaker + # the view grew to 61,190 tokens against a 40,000 budget. + # + # The signal that actually separates "this workload needs compacting" + # from "compaction is not working" is EFFECTIVENESS: in the incident + # `after_tokens` never once dropped below 1,797,300 across all 235 + # passes. Zero improvement, 235 times. + # + # Freezing is PREFIX-SAFE by construction: the sticky decisions already + # made are re-applied unchanged, which is strictly more stable than + # re-deriving them. Same shape as the two guards above -- decide whether + # to escalate, never what the view contains. + if ( + needs_escalation + and self._ineffective_escalations >= self._MAX_INEFFECTIVE_ESCALATIONS + ): + if not self._escalation_breaker_reported: + logger.error( + f"Compaction has escalated {self._ineffective_escalations} " + f"times in a row without reducing the result (currently " + f"{current_tokens:,} tokens against a {budget:,} budget, " + f"target {target_tokens:,}, cumulative level " + f"{self._sticky_level}). Further escalation is deleting " + f"conversation without moving the number, so it is being " + f"stopped and the existing view returned. This is a " + f"configuration or estimation problem, not a workload " + f"problem: check the system prompt size, the token budget, " + f"and whether any single message is larger than the target." + ) + self._escalation_breaker_reported = True + needs_escalation = False + if not needs_escalation: # Sticky state alone already keeps us under the threshold that # triggered compaction in the first place -- nothing NEW needs @@ -823,6 +935,7 @@ async def _compact_ephemeral( target_tokens, protected_recent=level3_protection, system_tokens=system_tokens, + budget=budget, ) ) total_removed += removed @@ -891,6 +1004,7 @@ async def _compact_ephemeral( target_tokens, protected_recent=level5_protection, system_tokens=system_tokens, + budget=budget, ) ) total_removed += removed @@ -956,6 +1070,7 @@ async def _compact_ephemeral( target_tokens, protected_recent=level7_protection, system_tokens=system_tokens, + budget=budget, ) ) total_removed += removed @@ -986,15 +1101,26 @@ async def _compact_ephemeral( if first_user_idx is not None and first_user_idx != last_user_idx: first_msg = working_messages[first_user_idx] if not first_msg.get("_stubbed"): - content = first_msg.get("content", "") - if isinstance(content, str) and len(content) > 80: - working_messages[first_user_idx] = self._stub_user_message( - first_msg - ) + # Let `_stub_user_message` decide. An `isinstance(content, + # str)` guard here would re-impose the shape-based exemption + # that commit 56fecd6 removed FROM THE HELPER -- the helper + # handles block content, and a guard at the call site makes + # that branch unreachable from production. + before_tokens = self._estimate_message_tokens(first_msg) + stubbed_msg = self._stub_user_message(first_msg) + if stubbed_msg is not first_msg: + working_messages[first_user_idx] = stubbed_msg # Sticky: record before `first_msg` var is superseded. self._record_stubbed(first_msg) total_stubbed += 1 - savings = (len(content) - 70) // 4 + # UNITS: measured, not derived from `len(content)`. + # `len()` on block content is a BLOCK COUNT, so the old + # `(len(content) - 70) // 4` was arithmetic on the wrong + # quantity the moment the shape stopped being a string -- + # the same unit-mismatch class as ad7936a. + savings = before_tokens - self._estimate_message_tokens( + stubbed_msg + ) current_tokens -= savings logger.info( f"Level 8: Stubbed first user message (saved ~{savings} tokens)" @@ -1025,7 +1151,10 @@ async def _compact_ephemeral( if indices_to_remove: # Sticky: record before filtering the list out from under them. for i in indices_to_remove: - self._record_removed(working_messages[i]) + self._record_removed( + working_messages[i], + f"level-{self._sticky_level} sweep of already-stubbed messages", + ) working_messages = [ msg for i, msg in enumerate(working_messages) @@ -1110,9 +1239,10 @@ def _truncate_tool_wave( # total-vs-total after this first mutation replaces the # caller-supplied (already total) seed value. true_total = self._estimate_tokens(messages) + system_tokens - old_len = len(str(msg)) // 4 + # UNITS: content-aware on both sides, matching `true_total`. + old_len = self._estimate_message_tokens(msg) messages[i] = self._truncate_tool_result(msg) - new_len = len(str(messages[i])) // 4 + new_len = self._estimate_message_tokens(messages[i]) true_total += new_len - old_len truncated += 1 current_tokens = true_total @@ -1127,6 +1257,7 @@ def _remove_messages_with_protection( target_tokens: int, protected_recent: float, system_tokens: int, + budget: int = 0, ) -> tuple[list[dict[str, Any]], int, int, int]: """ Remove oldest messages with specified protection level. @@ -1201,7 +1332,13 @@ def _remove_messages_with_protection( # `messages` is constant here, the base total only needs computing # once, and the removed-token total only needs an O(1) delta per # newly-removed index. - token_lens = [len(str(msg)) // 4 for msg in messages] + # UNITS: must match `base_tokens` below, which is content-aware. The + # old `len(str(msg)) // 4` counted a base64 payload as prose, so + # removing one image-bearing message credited ~100k tokens against a + # baseline that had counted it at ~1.6k -- `current_tokens` went hard + # negative, the loop exited on its first candidate, and compaction + # silently UNDER-shot. Deltas and baseline must come from one estimator. + token_lens = [self._estimate_message_tokens(msg) for msg in messages] tool_call_id_to_indices: dict[str, list[int]] = {} for idx, m in enumerate(messages): tcid = m.get("tool_call_id") @@ -1217,6 +1354,40 @@ def _remove_messages_with_protection( ) # computed once; messages is constant here current_tokens = base_tokens + # STOP CONDITION FEASIBILITY. + # + # The loop below exits when `current_tokens <= target_tokens`. If + # removing EVERY eligible candidate still cannot reach the target, that + # condition never becomes true, so the loop runs to exhaustion and + # deletes everything it is allowed to -- permanently, because + # `_removed_seqs` is re-applied on every later rebuild -- for no gain. + # + # The target is unreachable whenever un-compactable content alone + # exceeds it. System messages are one way (guarded earlier, before the + # level ladder) but NOT the only one: the last user message and the last + # `protected_tool_results` tool results are equally immune. One + # `read_file` on a large file puts an enormous tool result inside that + # protected window and arms this. + # + # Measured on this module before the clamp, with a 16-token system + # prompt and no images -- so the earlier system-floor guard cannot fire: + # a 64-message conversation collapsed to 6 messages on the FIRST call at + # maximum level, 58 messages removed permanently, and by the second call + # the view was 229 tokens against a 40,000 budget. The condition that + # caused it is TRANSIENT -- the blob leaves the protected tool window + # within a few turns and becomes truncatable -- but the removals are not. + # + # So: aim at the requirement that is actually achievable. `target_usage` + # is a hysteresis preference; fitting the budget is the contract. If even + # the budget is out of reach this changes nothing and the loop behaves + # exactly as before. + if budget > 0: + achievable_floor = base_tokens - sum( + token_lens[i] for i in removal_candidates + ) + if achievable_floor > target_tokens: + target_tokens = max(target_tokens, budget) + for i in removal_candidates: if current_tokens <= target_tokens: break @@ -1275,10 +1446,16 @@ def _remove_messages_with_protection( if current_tokens <= target_tokens: break msg = messages[i] - content = msg.get("content", "") - if isinstance(content, str) and len(content) > 80: + # Same reasoning as the Level 8 site above: the helper owns the + # "is there anything worth stubbing here" decision for BOTH content + # shapes, and savings are measured rather than derived from + # `len(content)`, which is a block count for block content. + stubbed_msg = self._stub_user_message(msg) + if stubbed_msg is not msg: indices_to_stub.add(i) - savings = (len(content) - 70) // 4 # Stub is ~70 chars + savings = self._estimate_message_tokens( + msg + ) - self._estimate_message_tokens(stubbed_msg) current_tokens -= savings # Sticky: record these NEW decisions by stable seq id before building @@ -1288,7 +1465,10 @@ def _remove_messages_with_protection( # already absent from `messages` by the time this runs; stub # candidates explicitly exclude already-`_stubbed` messages above). for i in indices_to_remove: - self._record_removed(messages[i]) + self._record_removed( + messages[i], + f"level-{self._sticky_level} removal, target {target_tokens:,} tokens", + ) for i in indices_to_stub: self._record_stubbed(messages[i]) @@ -1442,7 +1622,12 @@ async def _finalize_compaction_with_stats( # operator knows which knob actually moves (shrink the system prompt, # or raise the budget -- compacting harder will not help). system_tokens = self._estimate_tokens(system_messages) - if budget > 0 and final_tokens > budget: + # Fires on the DESTRUCTIVE condition, not only when over budget. A view + # can sit far under budget while compaction runs at maximum level and + # deletes on every request, which is the state that previously ended here + # in total silence -- the gate was `> budget` while the damage begins at + # `> target_tokens`. + if budget > 0 and (final_tokens > budget or system_tokens > target_tokens): if system_tokens > target_tokens: cause = ( f"the system prompt ALONE is {system_tokens:,} tokens, which already " @@ -1478,9 +1663,37 @@ async def _finalize_compaction_with_stats( # not the per-call deltas passed in -- this reports the total # accumulated effect on the conversation, which is what the notice # and any observability consumer actually wants to know. + # EFFECTIVENESS BOOKKEEPING for the runaway breaker at the escalation + # gate. A pass that cannot move the result is the incident's signature: + # `after_tokens` never once dropped below 1,797,300 across all 235 + # passes. Compare against the PREVIOUS result rather than this pass's + # own before/after, because the pathology is that successive passes all + # land on the same number. + # A pass is INEFFECTIVE when it finishes still over the real budget. + # + # Two earlier definitions were wrong, and each was caught by + # `test_going_over_budget_still_escalates` rather than by reasoning: + # + # - "escalated N times in a row" punishes a session that is genuinely + # over budget and must compact on every call. + # - "did not reduce vs the previous pass" punishes compaction that is + # correctly HOLDING THE LINE while new turns arrive. A result that + # plateaus just under budget is compaction working, not failing. + # + # Landing over budget is the honest failure signal: the view being + # returned will not fit, so the pass did not accomplish the one thing it + # exists to do. In the incident every pass landed at ~1.84x the budget, + # 235 times running. + if budget > 0 and final_tokens > budget: + self._ineffective_escalations += 1 + else: + self._ineffective_escalations = 0 + self._last_after_tokens = final_tokens + stats = { "before_tokens": old_tokens, "after_tokens": final_tokens, + "ineffective_escalations": self._ineffective_escalations, "before_messages": old_count, "after_messages": len(final_messages), "messages_removed": len(self._removed_seqs), @@ -1522,27 +1735,82 @@ def _truncate_tool_result(self, msg: dict[str, Any]) -> dict[str, Any]: "_original_tokens": original_tokens, } + _MAX_INEFFECTIVE_ESCALATIONS = 10 + """Consecutive INEFFECTIVE escalations before the runaway breaker trips. + + The incident ran **235 compactions across 266 model calls**, all pinned at + maximum strategy, with `after_tokens` never once dropping below 1,797,300. + Zero improvement, 235 times. Nothing anywhere counted, and the only signal + was an INFO line that fired 235 times and nobody saw. + + Counting *frequency* was the obvious first design and it is wrong: a session + that is genuinely over budget must keep compacting every call, and freezing + it there converts "destroying conversation" into "guaranteed provider + rejection". Ineffectiveness is the signal that actually separates "this + workload needs compacting" from "compaction is not working". + """ + + def _stub_text(self, content: str) -> str: + """The stub body: a 50-char preview of what was there.""" + preview = content[:50].replace("\n", " ").strip() + if len(content) > 50: + preview += "..." + return f'[User message compacted - original: "{preview}"]' + def _stub_user_message(self, msg: dict[str, Any]) -> dict[str, Any]: """ Create a stub for a user message to preserve thread while reducing tokens. + Handles BOTH content shapes. A message whose content is a list of blocks + used to be returned unchanged, because the guard was + ``isinstance(content, str)`` -- so a multimodal message was structurally + exempt from the only mechanism that could shrink it, while a small + text-only message carrying the user's actual instructions could still be + stubbed. Protection ran by TYPE rather than by cost. + + Non-text blocks are preserved rather than stubbed: they are counted at a + flat cost by ``_estimate_content_tokens``, so removing one buys almost + nothing and loses the attachment. Only the text is compacted. + Returns a NEW dict - does not modify the original. """ content = msg.get("content", "") - if not isinstance(content, str) or len(content) <= 80: - return msg # Too short to stub - # Take first 50 chars, clean up for display - preview = content[:50].replace("\n", " ").strip() - if len(content) > 50: - preview += "..." + if isinstance(content, str): + if len(content) <= 80: + return msg # Too short to stub + return { + **msg, + "content": self._stub_text(content), + "_stubbed": True, + "_original_length": len(content), + } - return { - **msg, - "content": f'[User message compacted - original: "{preview}"]', - "_stubbed": True, - "_original_length": len(content), - } + if isinstance(content, list): + text_blocks = [ + block + for block in content + if isinstance(block, dict) and block.get("type") == "text" + ] + joined = "".join(str(block.get("text", "")) for block in text_blocks) + if len(joined) <= 80: + return msg # Nothing worth stubbing; attachments stay as they are + preserved = [ + block + for block in content + if not (isinstance(block, dict) and block.get("type") == "text") + ] + return { + **msg, + "content": [ + {"type": "text", "text": self._stub_text(joined)}, + *preserved, + ], + "_stubbed": True, + "_original_length": len(joined), + } + + return msg def _format_compaction_notice(self) -> str: """ @@ -1706,6 +1974,66 @@ def _calculate_budget(self, token_budget: int | None, provider: Any | None) -> i logger.info(f"Using fallback max_tokens budget: {self.max_tokens:,}") return self.max_tokens + # A non-text content block costs a flat approximation, never the length of + # its payload. A base64 image measured as len(payload)/4 reads as hundreds + # of thousands of tokens while actually costing ~1-2k, and because such a + # message is structurally protected from shrinking, the compactor's target + # becomes unreachable: it re-runs on every request, deleting real + # conversation to chase a number that cannot come down. Any fixed value in + # the low thousands is ~1000x closer to truth than the payload length. + _NON_TEXT_BLOCK_TOKENS = 1600 + _NON_TEXT_BLOCK_TYPES = frozenset( + { + "image", + "image_url", + "input_image", + "input_audio", + "audio", + "video", + "document", + "file", + } + ) + def _estimate_tokens(self, messages: list[dict[str, Any]]) -> int: - """Rough token estimation (chars / 4).""" - return sum(len(str(msg)) // 4 for msg in messages) + """Rough token estimation, content-aware. + + Text is counted at chars/4. Non-text blocks are counted at a flat + per-block cost rather than the size of their encoded payload -- see + ``_NON_TEXT_BLOCK_TOKENS``. + """ + return sum(self._estimate_message_tokens(msg) for msg in messages) + + def _estimate_message_tokens(self, msg: dict[str, Any]) -> int: + """Estimate one message: envelope overhead plus content.""" + if not isinstance(msg, dict): + return len(str(msg)) // 4 + envelope = {key: value for key, value in msg.items() if key != "content"} + overhead = len(str(envelope)) // 4 if envelope else 0 + return overhead + self._estimate_content_tokens(msg.get("content")) + + def _estimate_content_tokens(self, content: Any) -> int: + """Estimate a content value, descending into structured blocks.""" + if content is None: + return 0 + if isinstance(content, str): + return len(content) // 4 + if not isinstance(content, list): + return len(str(content)) // 4 + + total = 0 + for block in content: + if not isinstance(block, dict): + total += len(str(block)) // 4 + continue + if block.get("type") in self._NON_TEXT_BLOCK_TYPES: + total += self._NON_TEXT_BLOCK_TOKENS + continue + # A tool_result can carry blocks of its own, including images. + nested = block.get("content") + if isinstance(nested, (list, str)): + envelope = {k: v for k, v in block.items() if k != "content"} + total += len(str(envelope)) // 4 + self._estimate_content_tokens(nested) + continue + total += len(str(block)) // 4 + return total diff --git a/tests/test_compaction_unit_consistency.py b/tests/test_compaction_unit_consistency.py new file mode 100644 index 0000000..fc58a70 --- /dev/null +++ b/tests/test_compaction_unit_consistency.py @@ -0,0 +1,122 @@ +"""Compaction arithmetic must be done in ONE unit. + +The estimator fix (content-aware counting, so a base64 payload is not measured +as prose) landed in `_estimate_tokens` but NOT in the two hot paths that do +delta arithmetic on top of it. Those kept the old `len(str(msg)) // 4`. + +The result was a second, opposite defect introduced by the fix for the first: + + baseline (content-aware, whole list) : 1,621 + per-message delta (old formula) : 100,031 <- what the loop subtracted + running total after one removal : -98,410 <- hard negative + +`_remove_messages_with_protection` exits as soon as `current_tokens <= +target_tokens`, so a single image-bearing removal drove the total negative and +the loop stopped on its first candidate. Compaction silently UNDER-shot on +exactly the conversations the estimator fix was written for -- and +`final_tokens` is honestly re-measured at the end, so the reported stats looked +correct while the loop that produced them had been flying on garbage. + +These tests pin the invariant rather than the incident: whatever the estimator +does, the whole-list figure and the per-message figures must be the same +quantity, because the removal and truncation loops subtract one from the other. +""" + +from __future__ import annotations + +from typing import Any + +from amplifier_module_context_simple import SimpleContextManager + + +def _probe() -> SimpleContextManager: + return SimpleContextManager.__new__(SimpleContextManager) + + +def _image_message(payload_chars: int = 400_000) -> dict[str, Any]: + return { + "role": "user", + "content": [ + {"type": "text", "text": "here is the mockup"}, + { + "type": "image", + "source": {"type": "base64", "data": "A" * payload_chars}, + }, + ], + } + + +def _conversation() -> list[dict[str, Any]]: + return [ + {"role": "user", "content": "lets build it"}, + _image_message(), + {"role": "assistant", "content": "looking at the mockup"}, + {"role": "tool", "tool_call_id": "c1", "content": "R" * 5_000}, + _image_message(120_000), + {"role": "assistant", "content": "done"}, + ] + + +def test_per_message_estimates_sum_to_the_whole_list_estimate() -> None: + """The invariant both hot loops depend on. + + `_remove_messages_with_protection` seeds a running total from the whole-list + estimate and then subtracts per-message figures; `_truncate_tool_wave` does + the same with a before/after pair. If the two disagree by even a constant + factor the running total is meaningless, and on image-bearing messages they + disagreed by ~60x. + """ + probe = _probe() + messages = _conversation() + + whole = probe._estimate_tokens(messages) + parts = sum(probe._estimate_message_tokens(message) for message in messages) + + assert whole == parts, ( + f"whole-list estimate {whole:,} != sum of per-message estimates {parts:,}; " + "the removal and truncation loops subtract one from the other" + ) + + +def test_removing_any_message_leaves_the_running_total_sane() -> None: + """The concrete failure: one removal drove the total hard negative. + + A negative running total satisfies `current_tokens <= target_tokens` + immediately, so the loop stopped after its first candidate and compaction + under-shot -- silently, because the final figure is re-measured honestly. + """ + probe = _probe() + messages = _conversation() + total = probe._estimate_tokens(messages) + + for index, message in enumerate(messages): + remaining = total - probe._estimate_message_tokens(message) + assert remaining >= 0, ( + f"removing message {index} ({message.get('role')}) drove the running " + f"total to {remaining:,} against a baseline of {total:,}" + ) + # And it must equal what a fresh estimate of the shortened list says. + rest = messages[:index] + messages[index + 1 :] + assert remaining == probe._estimate_tokens(rest) + + +def test_truncating_a_tool_result_moves_the_total_by_its_own_delta() -> None: + """The second hot path: `_truncate_tool_wave`'s before/after pair.""" + # A real instance: `_truncate_tool_result` reads configured state + # (`truncate_chars`), unlike the pure estimator methods above. + probe = SimpleContextManager(max_tokens=40_000, truncate_chars=100) + message = {"role": "tool", "tool_call_id": "c1", "content": "R" * 20_000} + messages = [{"role": "user", "content": "go"}, message] + + before_total = probe._estimate_tokens(messages) + old_len = probe._estimate_message_tokens(message) + truncated = probe._truncate_tool_result(message) + new_len = probe._estimate_message_tokens(truncated) + + predicted = before_total + (new_len - old_len) + actual = probe._estimate_tokens([messages[0], truncated]) + + assert predicted == actual, ( + f"delta arithmetic predicted {predicted:,} but a fresh estimate says {actual:,}" + ) + assert new_len < old_len, "truncation must actually reduce the estimate" diff --git a/tests/test_infeasible_target_guard.py b/tests/test_infeasible_target_guard.py new file mode 100644 index 0000000..a3bfbb5 --- /dev/null +++ b/tests/test_infeasible_target_guard.py @@ -0,0 +1,140 @@ +"""Never escalate toward a target that arithmetic says is unreachable. + +System messages are never compacted. Once their share alone exceeds the +compaction target, no escalation level can reach it -- the predicate never +clears, so compaction re-decides on every request, pinned at maximum level, +deleting real conversation to chase a number that cannot come down. + +Reproduced on this module before the check existed, with a 12,153-token system +prompt against a 40,000 budget (target 10,000) and **no images anywhere**: + + call level after_tokens removed view + 5 8 14,475 6 4 + 13 8 14,776 18 8 + 29 8 14,776 54 4 <- 54 of 58 messages ever added + +`after_tokens` never moved. The view sawtoothed as history regrew and was +destroyed again. And it was silent: the existing over-budget warning is gated on +`final_tokens > budget`, while this state sits at 37% of budget. + +The guard only declines to escalate while the view still fits the ACTUAL budget. +Over budget a partial reduction beats none, so escalation proceeds as before. +""" + +from __future__ import annotations + +import logging + +import pytest +from amplifier_module_context_simple import SimpleContextManager + +MODULE_LOGGER = "amplifier_module_context_simple" + +# ~12,153 tokens: larger than the 10,000 target, smaller than the 40,000 budget. +BIG_SYSTEM = "S" * 48_524 +BUDGET = 40_000 +TARGET = 10_000 + + +def _manager(**overrides) -> SimpleContextManager: + config = { + "max_tokens": BUDGET, + "target_usage": 0.25, # target = 10,000 + "compact_threshold": 0.5, # compaction considered from 20,000 + "protected_recent": 0.30, + "compaction_notice_enabled": False, + } + config.update(overrides) + return SimpleContextManager(**config) + + +async def _grow(context: SimpleContextManager, turns: int, words: int = 600) -> None: + for i in range(turns): + await context.add_message({"role": "user", "content": f"turn {i} " * words}) + await context.add_message( + {"role": "assistant", "content": f"reply {i} " * words} + ) + await context.get_messages_for_request() + + +@pytest.mark.asyncio +async def test_an_unreachable_target_does_not_destroy_a_context_that_fits() -> None: + """The defect: deleting conversation while comfortably inside the budget.""" + context = _manager() + await context.add_message({"role": "system", "content": BIG_SYSTEM}) + await context.add_message( + {"role": "user", "content": "PROJECT PATH IS ~/Desktop/ora"} + ) + + system_tokens = context._estimate_tokens([context.messages[0]]) + assert system_tokens > TARGET, "fixture must make the target unreachable" + + await _grow(context, turns=4) + view = await context.get_messages_for_request() + + assert context._estimate_tokens(view) <= BUDGET, ( + "fixture must stay inside the budget" + ) + assert not context._removed_seqs, ( + f"removed {len(context._removed_seqs)} messages while the view still fit the " + f"budget, chasing a target that no level can reach" + ) + assert any("PROJECT PATH IS" in str(m.get("content", "")) for m in view) + + +@pytest.mark.asyncio +async def test_it_says_so_once_and_names_the_knob( + caplog: pytest.LogCaptureFixture, +) -> None: + """Silence is the failure mode this module cannot afford; so is 235 repeats.""" + context = _manager() + await context.add_message({"role": "system", "content": BIG_SYSTEM}) + + with caplog.at_level(logging.WARNING, logger=MODULE_LOGGER): + await _grow(context, turns=6) + + unreachable = [r for r in caplog.records if "target is unreachable" in r.message] + assert len(unreachable) == 1, ( + f"expected exactly one warning, got {len(unreachable)} -- a warning repeated " + f"per request trains operators to ignore it" + ) + message = unreachable[0].message + assert "system prompt alone" in message + assert f"{TARGET:,}" in message + assert "Reduce the system prompt or raise the budget" in message, ( + "the warning must name the knob that actually moves" + ) + + +@pytest.mark.asyncio +async def test_going_over_budget_still_escalates() -> None: + """Declining to chase the target must not become declining to compact. + + Over the real budget a partial reduction beats none, even when the target + stays out of reach. + """ + context = _manager() + await context.add_message({"role": "system", "content": BIG_SYSTEM}) + await _grow(context, turns=30) + + view = await context.get_messages_for_request() + + assert context._removed_seqs, "compaction must still act once genuinely over budget" + assert context._estimate_tokens(view) <= BUDGET, ( + "compaction ran but did not bring the view back inside the budget" + ) + + +@pytest.mark.asyncio +async def test_a_reachable_target_is_unaffected() -> None: + """The guard must be invisible whenever the target is actually achievable.""" + context = _manager(max_tokens=400_000, target_usage=0.5) # target 200,000 + await context.add_message({"role": "system", "content": BIG_SYSTEM}) + + system_tokens = context._estimate_tokens([context.messages[0]]) + assert system_tokens < 200_000, "fixture must make the target reachable" + + await _grow(context, turns=10) + view = await context.get_messages_for_request() + + assert context._estimate_tokens(view) <= 400_000 diff --git a/tests/test_multimodal_token_estimate.py b/tests/test_multimodal_token_estimate.py new file mode 100644 index 0000000..2c3f137 --- /dev/null +++ b/tests/test_multimodal_token_estimate.py @@ -0,0 +1,167 @@ +"""Token estimation must not measure a base64 payload as if it were prose. + +Regression cover for a session (`eec9ae98`) destroyed by this defect. Two +pasted screenshots were persisted as raw base64 inside the transcript. The +estimator counted them at ``len(str(msg)) // 4``, so 9.8M characters of image +data read as ~2.46M tokens against a 978,720 budget -- while the provider's own +usage accounting showed the model actually receiving 9,187-58,428 input tokens. + +The consequences compounded: because an image-bearing message is structurally +protected from shrinking, the compactor could never reach its target, so its +predicate never cleared and it re-ran on 88% of all model calls -- 235 times, +pinned at maximum strategy, deleting 663 of 686 messages and permanently +stubbing the user instructions that carried the project path. The session then +could not find the project it had been working on for hours. + +The arithmetic below is taken verbatim from that report so the numbers stay +falsifiable. +""" + +from __future__ import annotations + +from typing import Any + +import pytest +from amplifier_module_context_simple import SimpleContextManager + +# The two screenshots, in characters, exactly as persisted. +FIRST_IMAGE_CHARS = 7_182_876 +SECOND_IMAGE_CHARS = 2_647_967 + +# What the old estimator produced, and the observed compaction floor it matched. +OLD_ESTIMATE_FIRST_IMAGE = FIRST_IMAGE_CHARS // 4 # 1,795,719 +OBSERVED_STUCK_FLOOR = 1_797_300 + +# The session's real configuration. +BUDGET = 978_720 +TARGET = 489_360 + + +def _probe() -> SimpleContextManager: + """The estimator reads only class constants, so it needs no built state.""" + return SimpleContextManager.__new__(SimpleContextManager) + + +def _image_message(payload_chars: int) -> dict[str, Any]: + return { + "role": "user", + "content": [ + { + "type": "text", + "text": "here is the mockup, the project is at ~/Desktop/ora", + }, + { + "type": "image", + "source": { + "type": "base64", + "media_type": "image/png", + "data": "A" * payload_chars, + }, + }, + ], + "metadata": {"source": "tui-clipboard", "attachment_count": 1}, + } + + +def test_base64_image_is_not_measured_by_payload_length() -> None: + """The single defect: an image counted as prose.""" + estimate = _probe()._estimate_tokens([_image_message(FIRST_IMAGE_CHARS)]) + + assert estimate < 5_000, ( + f"A single screenshot estimated at {estimate:,} tokens. The old formula " + f"gave {OLD_ESTIMATE_FIRST_IMAGE:,}, which matched the observed compaction " + f"floor of {OBSERVED_STUCK_FLOOR:,} that never came down." + ) + + +def test_both_screenshots_fit_well_inside_the_budget_they_used_to_blow() -> None: + """Together the two images read as 2.5x the budget; they cost ~3k.""" + messages = [_image_message(FIRST_IMAGE_CHARS), _image_message(SECOND_IMAGE_CHARS)] + + estimate = _probe()._estimate_tokens(messages) + + assert estimate < TARGET, ( + f"Two screenshots estimated at {estimate:,} tokens against a compaction " + f"target of {TARGET:,}. When this exceeded the target, the target became " + f"arithmetically unreachable and compaction looped forever." + ) + + +def test_an_image_nested_in_a_tool_result_is_also_bounded() -> None: + """A screenshot returned by a tool is the same payload in a different place.""" + message = { + "role": "user", + "content": [ + { + "type": "tool_result", + "tool_use_id": "call-1", + "content": [ + {"type": "text", "text": "captured"}, + { + "type": "image", + "source": {"type": "base64", "data": "A" * FIRST_IMAGE_CHARS}, + }, + ], + } + ], + } + + assert _probe()._estimate_tokens([message]) < 5_000 + + +def test_text_estimation_is_unchanged() -> None: + """The fix must not quietly re-scale ordinary text.""" + probe = _probe() + body = "word " * 4_000 # 20,000 chars -> ~5,000 tokens + + plain = probe._estimate_tokens([{"role": "user", "content": body}]) + blocked = probe._estimate_tokens( + [{"role": "user", "content": [{"type": "text", "text": body}]}] + ) + + assert 4_900 <= plain <= 5_200, plain + # Block form carries a little structural overhead but must not diverge. + assert abs(blocked - plain) < 200, (plain, blocked) + + +@pytest.mark.asyncio +async def test_image_heavy_conversation_does_not_trigger_compaction() -> None: + """End to end: the session that shredded itself now compacts zero times. + + Real content here is a few hundred tokens. Before the fix, the first + compaction fired on the exact timestamp the first image arrived, and 235 + more followed. + """ + context = SimpleContextManager( + max_tokens=BUDGET, + target_usage=0.50, + protected_recent=0.30, + protected_tool_results=5, + compaction_notice_enabled=False, + ) + + await context.add_message( + {"role": "system", "content": "You are a helpful assistant."} + ) + await context.add_message( + {"role": "user", "content": "lets build the ora overseer"} + ) + await context.add_message(_image_message(FIRST_IMAGE_CHARS)) + await context.add_message( + {"role": "assistant", "content": "Looking at the mockup now."} + ) + await context.add_message(_image_message(SECOND_IMAGE_CHARS)) + for i in range(8): + await context.add_message({"role": "user", "content": f"continue {i}"}) + await context.add_message({"role": "assistant", "content": f"working {i}"}) + + view = await context.get_messages_for_request() + + assert context._last_compaction_stats is None, ( + "Compaction fired on a conversation whose real content is a few hundred " + f"tokens: {context._last_compaction_stats}" + ) + # Nothing was stubbed, so the message carrying the project path survives. + assert any( + "~/Desktop/ora" in str(message.get("content", "")) for message in view + ), "The user message carrying the project path did not survive." diff --git a/tests/test_removal_stop_condition.py b/tests/test_removal_stop_condition.py new file mode 100644 index 0000000..6ba3d38 --- /dev/null +++ b/tests/test_removal_stop_condition.py @@ -0,0 +1,167 @@ +"""Never delete everything eligible to chase a target that cannot be reached. + +`_remove_messages_with_protection` stops when `current_tokens <= target_tokens`. +When the target is unreachable that condition never becomes true, so the loop +runs to exhaustion and removes every eligible candidate -- permanently, because +`_removed_seqs` is re-applied on every later rebuild -- for no gain at all. + +The system-message floor is guarded before the level ladder, but it is not the +only un-compactable content: the last user message and the last +`protected_tool_results` tool results are equally immune. One `read_file` on a +large file puts an enormous tool result inside that protected window and arms +this, with a 16-token system prompt and no images anywhere. + +Measured on this module before the clamp: + + call hist view tokens removed + 1 64 6 35,273 58 + 2 66 5 229 61 <- 229 tokens against a 40,000 budget + +and after: + + 1 64 62 39,777 2 + 2 66 62 39,745 4 + +The condition that causes it is TRANSIENT -- the blob leaves the protected tool +window within a few turns and becomes truncatable -- but the removals are not. +""" + +from __future__ import annotations + +from typing import Any + +import pytest + +from amplifier_module_context_simple import SimpleContextManager + +BUDGET = 40_000 +BIG_TOOL_RESULT = "X" * 140_000 + + +def _manager(**overrides: Any) -> SimpleContextManager: + config: dict[str, Any] = { + "max_tokens": BUDGET, + "target_usage": 0.5, # target 20,000 + "compact_threshold": 0.5, + "protected_recent": 0.30, + "protected_tool_results": 5, + "compaction_notice_enabled": False, + } + config.update(overrides) + return SimpleContextManager(**config) + + +async def _ordinary_session_then_one_big_read(context: SimpleContextManager) -> None: + """A small system prompt, 30 turns of chat, one large tool result. + + Deliberately mundane: nothing here is an image, and the system prompt is far + below the target, so the system-floor guard cannot fire. + """ + await context.add_message({"role": "system", "content": "You are helpful."}) + await context.add_message( + {"role": "user", "content": "PROJECT PATH IS ~/Desktop/ora"} + ) + for i in range(30): + await context.add_message({"role": "user", "content": f"turn {i} " * 30}) + await context.add_message({"role": "assistant", "content": f"reply {i} " * 30}) + await context.add_message( + { + "role": "assistant", + "content": "reading", + "tool_calls": [{"id": "t1", "name": "read_file"}], + } + ) + await context.add_message( + {"role": "tool", "tool_call_id": "t1", "content": BIG_TOOL_RESULT} + ) + + +@pytest.mark.asyncio +async def test_one_large_tool_result_does_not_delete_the_conversation() -> None: + """The defect: a routine `read_file` destroyed 58 of 64 messages on call one.""" + context = _manager() + await _ordinary_session_then_one_big_read(context) + + view = await context.get_messages_for_request() + + assert len(context._removed_seqs) < 20, ( + f"removed {len(context._removed_seqs)} messages on the first call while " + f"chasing an unreachable target" + ) + assert len(view) > 40, f"view collapsed to {len(view)} messages" + + +@pytest.mark.asyncio +async def test_the_view_never_collapses_far_below_the_budget() -> None: + """A view of 229 tokens against a 40,000 budget is not compaction, it is loss. + + Aiming at an unreachable target made the loop remove everything it was + allowed to; the result undershot the budget by two orders of magnitude. + """ + context = _manager() + await _ordinary_session_then_one_big_read(context) + + for _ in range(3): + view = await context.get_messages_for_request() + tokens = context._estimate_tokens(view) + assert tokens > BUDGET // 10, ( + f"view fell to {tokens:,} tokens against a {BUDGET:,} budget -- " + f"far more was deleted than the budget ever required" + ) + await context.add_message({"role": "user", "content": "next " * 30}) + await context.add_message({"role": "assistant", "content": "ok " * 30}) + + +@pytest.mark.asyncio +async def test_removals_stay_bounded_as_the_session_continues() -> None: + """The transient condition must not keep ratcheting `_removed_seqs`.""" + context = _manager() + await _ordinary_session_then_one_big_read(context) + + for _ in range(10): + await context.get_messages_for_request() + await context.add_message({"role": "user", "content": "next " * 30}) + await context.add_message({"role": "assistant", "content": "ok " * 30}) + + assert len(context._removed_seqs) < 45, ( + f"{len(context._removed_seqs)} messages permanently removed; the " + f"condition that caused it lasts only a few turns" + ) + + +@pytest.mark.asyncio +async def test_a_reachable_target_is_still_pursued() -> None: + """The clamp must not become an excuse to stop compacting. + + When the target IS achievable the loop must still drive to it, or the fix + for over-removal becomes a cause of under-removal. + """ + context = _manager(max_tokens=200_000, target_usage=0.5) + await context.add_message({"role": "system", "content": "You are helpful."}) + for i in range(120): + await context.add_message({"role": "user", "content": f"turn {i} " * 300}) + await context.add_message({"role": "assistant", "content": f"reply {i} " * 300}) + + view = await context.get_messages_for_request() + + assert context._estimate_tokens(view) <= 200_000, "view must fit the budget" + assert context._removed_seqs, "a reachable target must still be pursued" + + +@pytest.mark.asyncio +async def test_the_clamp_is_inert_without_a_budget() -> None: + """`budget=0` (the default) must behave exactly as before. + + The parameter is threaded from the three call sites; a caller that does not + supply it must not silently change behaviour. + """ + context = _manager() + await _ordinary_session_then_one_big_read(context) + messages = [dict(m) for m in context.messages] + + kept, removed, _stubbed, _tokens = context._remove_messages_with_protection( + messages, target_tokens=20_000, protected_recent=0.30, system_tokens=16 + ) + + assert removed > 0, "without a budget the old exhaustive behaviour stands" + assert len(kept) < len(messages) diff --git a/tests/test_runaway_compaction_breaker.py b/tests/test_runaway_compaction_breaker.py new file mode 100644 index 0000000..729f10f --- /dev/null +++ b/tests/test_runaway_compaction_breaker.py @@ -0,0 +1,160 @@ +"""Stop compacting when compaction is demonstrably not working. + +The incident ran **235 compactions across 266 model calls** -- 88% of every +request preceded by a full compaction, all pinned at maximum strategy, every one +finishing at roughly 1.84x the budget. Nothing anywhere counted the repetition, +and the only signal was an INFO line that fired 235 times and nobody saw. + +Defining "not working" took three attempts, and each wrong definition was caught +by `test_going_over_budget_still_escalates` rather than by reasoning: + +1. **"escalated N times in a row"** punishes a session that is genuinely over + budget and *must* compact on every call. Under it the view grew to 61,190 + tokens against a 40,000 budget -- the breaker turned "destroying + conversation" into "guaranteed provider rejection", which is worse. +2. **"did not reduce versus the previous pass"** punishes compaction that is + correctly holding the line while new turns arrive. A result that plateaus + just *under* budget is compaction working, not failing. +3. **"finished still over the real budget"** -- the definition here. The view + being returned will not fit, so the pass did not accomplish the one thing it + exists to do. +""" + +from __future__ import annotations + +import logging +from typing import Any + +import pytest + +from amplifier_module_context_simple import SimpleContextManager + +MODULE_LOGGER = "amplifier_module_context_simple" +BUDGET = 40_000 +# Larger than the whole budget: no amount of conversation compaction can make a +# view containing it fit, so every pass lands over budget. +UNCOMPACTABLE_SYSTEM = "S" * 200_000 + + +def _manager(**overrides: Any) -> SimpleContextManager: + config: dict[str, Any] = { + "max_tokens": BUDGET, + "target_usage": 0.5, + "compact_threshold": 0.5, + "protected_recent": 0.30, + "compaction_notice_enabled": False, + } + config.update(overrides) + return SimpleContextManager(**config) + + +async def _grow(context: SimpleContextManager, turns: int, words: int = 300) -> None: + for i in range(turns): + await context.add_message({"role": "user", "content": f"turn {i} " * words}) + await context.add_message( + {"role": "assistant", "content": f"reply {i} " * words} + ) + await context.get_messages_for_request() + + +@pytest.mark.asyncio +async def test_the_breaker_trips_when_every_pass_lands_over_budget() -> None: + """The incident's shape: compaction that never once produces a usable view.""" + context = _manager() + await context.add_message({"role": "system", "content": UNCOMPACTABLE_SYSTEM}) + + await _grow(context, turns=20) + + assert context._escalation_breaker_reported is True, ( + "compaction finished over budget on every pass and nothing ever stopped it" + ) + + +@pytest.mark.asyncio +async def test_the_breaker_says_so_once(caplog: pytest.LogCaptureFixture) -> None: + """235 identical INFO lines is why nobody saw the incident. + + One ERROR that names the knobs is the whole point; repeating it per request + would recreate the noise it replaces. + """ + context = _manager() + await context.add_message({"role": "system", "content": UNCOMPACTABLE_SYSTEM}) + + with caplog.at_level(logging.ERROR, logger=MODULE_LOGGER): + await _grow(context, turns=20) + + tripped = [r for r in caplog.records if "times in a row" in r.message] + assert len(tripped) == 1, f"expected exactly one ERROR, got {len(tripped)}" + message = tripped[0].message + assert "system prompt size" in message + assert "token budget" in message + + +@pytest.mark.asyncio +async def test_a_session_that_compacts_successfully_never_trips() -> None: + """Holding the line under budget is compaction working, not failing. + + This is the case definition (2) got wrong: the result plateaus because new + turns keep arriving, not because the compactor is stuck. + """ + context = _manager(max_tokens=200_000) + await context.add_message({"role": "system", "content": "You are helpful."}) + + await _grow(context, turns=40) + + assert context._escalation_breaker_reported is False + assert context._ineffective_escalations == 0 + + +@pytest.mark.asyncio +async def test_one_successful_pass_re_arms_the_breaker() -> None: + """The pathology is a consecutive run, so recovery must reset the count.""" + context = _manager() + await context.add_message({"role": "system", "content": "You are helpful."}) + await _grow(context, turns=10) + + context._ineffective_escalations = 5 # pretend a rough patch + await context.add_message({"role": "user", "content": "small"}) + await context.get_messages_for_request() + + assert context._ineffective_escalations == 0, ( + "a pass that produced a view inside the budget must clear the count" + ) + + +@pytest.mark.asyncio +async def test_the_count_is_reported_for_observability() -> None: + """A breaker nobody can see the approach of is a breaker nobody trusts.""" + context = _manager() + await context.add_message({"role": "system", "content": UNCOMPACTABLE_SYSTEM}) + await _grow(context, turns=6) + + stats = context._last_compaction_stats or {} + assert "ineffective_escalations" in stats + assert stats["ineffective_escalations"] > 0 + + +@pytest.mark.asyncio +async def test_tripping_freezes_rather_than_re_deriving() -> None: + """Prefix safety: the breaker must return the decisions already made. + + Freezing re-applies the existing sticky set unchanged, which is strictly + more stable for prompt caching than re-deriving the whole decision. + """ + context = _manager() + await context.add_message({"role": "system", "content": UNCOMPACTABLE_SYSTEM}) + await _grow(context, turns=20) + assert context._escalation_breaker_reported is True + + removed_at_trip = set(context._removed_seqs) + level_at_trip = context._sticky_level + + await context.add_message({"role": "user", "content": "another turn " * 300}) + await context.get_messages_for_request() + + # Compare KEY SETS: `_removed_seqs` maps seq -> why it was removed, so a + # bare `== removed_at_trip` compares a dict against a set and is always + # False. The invariant under test is "no new removals", not "the reason + # strings are byte-identical". + assert set(context._removed_seqs) == removed_at_trip, "kept deleting after the trip" + assert context._sticky_level == level_at_trip, "kept escalating after the trip" diff --git a/tests/test_sticky_compaction_and_tail_notice.py b/tests/test_sticky_compaction_and_tail_notice.py index 8cdabf7..d8cb657 100644 --- a/tests/test_sticky_compaction_and_tail_notice.py +++ b/tests/test_sticky_compaction_and_tail_notice.py @@ -302,9 +302,20 @@ async def test_large_system_message_counts_toward_compaction_trigger(): ) +# A probe instance: the estimator reads only class-level constants, so it needs +# no constructed state. +_ESTIMATOR_PROBE = SimpleContextManager.__new__(SimpleContextManager) + + def _estimate(messages: list[dict]) -> int: - """Mirror of SimpleContextManager._estimate_tokens for test-side assertions.""" - return sum(len(str(m)) // 4 for m in messages) + """Delegate to the real estimator instead of mirroring it. + + This was a hand-copied `sum(len(str(m)) // 4 ...)` mirror, which silently + went stale the moment estimation became content-aware -- the assertions + then compared production against a formula production no longer used. + Measuring "exactly as the module measures it" means calling the module. + """ + return _ESTIMATOR_PROBE._estimate_tokens(messages) async def _build_large_system_scenario( diff --git a/tests/test_stub_call_sites.py b/tests/test_stub_call_sites.py new file mode 100644 index 0000000..6637740 --- /dev/null +++ b/tests/test_stub_call_sites.py @@ -0,0 +1,153 @@ +"""The stub call sites must not re-impose the shape exemption the helper dropped. + +Commit `56fecd6` made `_stub_user_message` shape-agnostic: a message whose +content is a list of blocks can be compacted, with non-text blocks preserved. +Both *callers* kept an `isinstance(content, str)` guard, so that branch of the +helper was unreachable from production -- the fix was live in the unit under +test and dead in the code path. + +The guards also computed savings as `(len(content) - 70) // 4`. On block content +`len()` is a BLOCK COUNT, not a character count, so lifting the guard without +fixing the arithmetic would have produced a savings figure in the wrong unit -- +the same class of defect as `ad7936a`, where per-message deltas and the baseline +disagreed. + +HONEST SCOPE: these tests pin the call sites' behaviour directly. Driving the +same difference end-to-end through `get_messages_for_request()` was attempted +four times and could not be reproduced -- `stub_candidates` excludes the first +and last user message at levels 1-7, and in every fixture built, removal reached +the target before the stub stage ran at all. So this is a latent-defect fix: +the inconsistency and the unit error are real and verified here; a user-visible +symptom was not demonstrated. +""" + +from __future__ import annotations + +from typing import Any + +import pytest + +from amplifier_module_context_simple import SimpleContextManager + +LONG_TEXT = ( + "the project lives at ~/Desktop/ora and the overseer app is inside it, " * 40 +) + + +def _manager() -> SimpleContextManager: + return SimpleContextManager( + max_tokens=40_000, + target_usage=0.5, + protected_recent=0.10, + compaction_notice_enabled=False, + ) + + +def _user(content: Any, *, subject: bool = False) -> dict[str, Any]: + msg: dict[str, Any] = {"role": "user", "content": content} + if subject: + # Removal shifts indices, so the subject is located by marker, never by + # position -- an earlier version of this file asserted on kept[2] and + # was reading whichever message happened to land there. + msg["_probe_subject"] = True + return msg + + +def _blocks(text: str) -> list[dict[str, Any]]: + return [{"type": "text", "text": text}] + + +def _find_subject(messages: list[dict[str, Any]]) -> dict[str, Any] | None: + for msg in messages: + if msg.get("_probe_subject"): + return msg + return None + + +def _conversation(subject: dict[str, Any]) -> list[dict[str, Any]]: + """`subject` sits mid-history so it is a stub candidate, not first or last.""" + return [ + _user("first user message"), + {"role": "assistant", "content": "ok"}, + subject, + {"role": "assistant", "content": "ok"}, + _user("last user message"), + ] + + +@pytest.mark.parametrize( + ("label", "make_content"), + [ + pytest.param("string", lambda: LONG_TEXT, id="string-content"), + pytest.param("blocks", lambda: _blocks(LONG_TEXT), id="block-content"), + ], +) +def test_both_content_shapes_are_stubbable_from_the_call_site( + label: str, make_content: Any +) -> None: + """The defect: only one of these two shapes could ever be stubbed.""" + context = _manager() + messages = _conversation(_user(make_content(), subject=True)) + + kept, _removed, stubbed, _tokens = context._remove_messages_with_protection( + messages, target_tokens=1, protected_recent=0.10, system_tokens=0 + ) + + assert stubbed >= 1, f"{label} content was never offered to the stubber" + found = _find_subject(kept) + assert found is not None, f"{label} subject was removed entirely, not stubbed" + assert found.get("_stubbed") is True, f"{label} subject not stubbed: {found}" + + +def test_a_stubbed_block_message_actually_gets_smaller() -> None: + """Reachability is not enough -- the compaction has to save something.""" + context = _manager() + subject = _user(_blocks(LONG_TEXT), subject=True) + before = context._estimate_message_tokens(subject) + + kept, _removed, _stubbed, _tokens = context._remove_messages_with_protection( + _conversation(subject), target_tokens=1, protected_recent=0.10, system_tokens=0 + ) + + found = _find_subject(kept) + assert found is not None + after = context._estimate_message_tokens(found) + assert after < before, f"stubbing block content saved nothing: {before} -> {after}" + + +def test_savings_are_measured_in_tokens_not_len() -> None: + """`len()` on block content is a block count, not a character count. + + The old `(len(content) - 70) // 4` on a one-block message yields a large + NEGATIVE number, which would have been subtracted from the running total -- + inflating it, and driving further compaction. This asserts the reported + post-compaction figure agrees with a fresh estimate of what was returned. + """ + context = _manager() + messages = _conversation(_user(_blocks(LONG_TEXT), subject=True)) + + kept, _removed, _stubbed, reported = context._remove_messages_with_protection( + messages, target_tokens=1, protected_recent=0.10, system_tokens=0 + ) + + assert reported >= 0, f"reported token count went negative: {reported}" + fresh = context._estimate_tokens(kept) + assert abs(reported - fresh) < max(200, fresh // 2), ( + f"reported {reported:,} disagrees with a fresh estimate {fresh:,} of the " + f"returned view -- the savings arithmetic is in the wrong unit" + ) + + +def test_a_short_block_message_is_left_alone() -> None: + """Deferring to the helper must not mean stubbing everything in sight.""" + context = _manager() + subject = _user(_blocks("short"), subject=True) + + kept, _removed, stubbed, _tokens = context._remove_messages_with_protection( + _conversation(subject), target_tokens=1, protected_recent=0.10, system_tokens=0 + ) + + found = _find_subject(kept) + assert found is not None + assert found.get("_stubbed") is not True + assert stubbed == 0, "a message with nothing worth compacting was stubbed anyway" diff --git a/tests/test_stub_protection_by_cost.py b/tests/test_stub_protection_by_cost.py new file mode 100644 index 0000000..119c970 --- /dev/null +++ b/tests/test_stub_protection_by_cost.py @@ -0,0 +1,108 @@ +"""Stubbing must protect by cost, not by content shape. + +Companion to `test_multimodal_token_estimate.py`, covering the second half of +report section 1. `_stub_user_message` guarded on `isinstance(content, str)`, so +a message whose content was a list of blocks was returned unchanged -- it could +not be stubbed at all. + +In the incident that meant the two largest messages in the context were +structurally exempt from the only mechanism that could shrink them, while small +text-only messages carrying the user's actual instructions were stubbed on all +235 passes. Protection ran by TYPE; it should run by COST. +""" + +from __future__ import annotations + +from typing import Any + +from amplifier_module_context_simple import SimpleContextManager + +LONG_TEXT = "the project lives at ~/Desktop/ora and the overseer app is inside it, " * 3 + + +def _probe() -> SimpleContextManager: + """Stubbing reads no constructed state.""" + return SimpleContextManager.__new__(SimpleContextManager) + + +def _image_block(payload: str = "A" * 2048) -> dict[str, Any]: + return {"type": "image", "source": {"type": "base64", "data": payload}} + + +def test_string_content_still_stubs() -> None: + """The path that already worked must keep working.""" + msg = {"role": "user", "content": LONG_TEXT} + + stubbed = _probe()._stub_user_message(msg) + + assert stubbed is not msg, "must return a new dict, never mutate" + assert stubbed["_stubbed"] is True + assert stubbed["_original_length"] == len(LONG_TEXT) + assert "User message compacted" in stubbed["content"] + + +def test_short_string_content_is_left_alone() -> None: + msg = {"role": "user", "content": "short"} + assert _probe()._stub_user_message(msg) is msg + + +def test_block_content_with_long_text_is_now_stubbable() -> None: + """The defect: this message used to be returned unchanged.""" + msg = { + "role": "user", + "content": [{"type": "text", "text": LONG_TEXT}], + } + + stubbed = _probe()._stub_user_message(msg) + + assert stubbed is not msg, ( + "block-shaped content was exempt from stubbing -- the largest messages " + "could not be shrunk while small ones carrying instructions could" + ) + assert stubbed["_stubbed"] is True + assert stubbed["_original_length"] == len(LONG_TEXT) + assert stubbed["content"][0]["type"] == "text" + assert "User message compacted" in stubbed["content"][0]["text"] + + +def test_attachments_survive_stubbing() -> None: + """Only the text is compacted; a non-text block is not worth dropping. + + Non-text blocks are counted at a flat cost by the estimator, so removing one + buys almost nothing and loses the attachment outright. + """ + image = _image_block() + msg = { + "role": "user", + "content": [ + {"type": "text", "text": LONG_TEXT}, + image, + {"type": "text", "text": " and one more note"}, + ], + } + + stubbed = _probe()._stub_user_message(msg) + + blocks = stubbed["content"] + assert len(blocks) == 2, blocks + assert blocks[0]["type"] == "text" + assert blocks[1] == image, "the attachment must survive verbatim" + # Both text runs are accounted for in the recorded original length. + assert stubbed["_original_length"] == len(LONG_TEXT) + len(" and one more note") + + +def test_block_content_with_little_text_is_left_alone() -> None: + """An image with a short caption has nothing worth compacting.""" + msg = { + "role": "user", + "content": [{"type": "text", "text": "look"}, _image_block()], + } + assert _probe()._stub_user_message(msg) is msg + + +def test_unexpected_content_shapes_pass_through() -> None: + """Never guess at a shape we do not recognise.""" + probe = _probe() + for content in (None, 42, {"type": "text"}): + msg = {"role": "user", "content": content} + assert probe._stub_user_message(msg) is msg, content