fix(memory): count tokens at decision boundaries through one shared helper (#13694)

autobot-backend/chat_history/context_overflow.py

@@ -19,6 +19,8 @@
 from autobot_shared.logging_manager import get_logger
 from autobot_shared.redis_client import get_async_redis_client
 from autobot_shared.redis_utils import decode_redis_value
+from autobot_shared.ssot_constants import CategoryDefaults
+from autobot_shared.token_count import estimate_fast
 
 logger = get_logger(__name__)
 
@@ -176,7 +178,7 @@ def _sanitize_tool_messages(msgs: List[Dict]) -> List[Dict]:
                 cleaned.append(m)
             # else: orphan — drop silently
             continue
-        if role == "assistant" and m.get("tool_calls"):
+        if role == CategoryDefaults.ROLE_ASSISTANT and m.get("tool_calls"):
             in_batch = True
         else:
             in_batch = False
@@ -254,7 +256,7 @@ async def summarize_messages(
 
             # Generate summary via LLM
             response = await gateway.chat_completion(
-                messages=[{"role": "user", "content": prompt}],
+                messages=[{"role": CategoryDefaults.ROLE_USER, "content": prompt}],
                 model=model_name,
                 temperature=0.3,  # Low temp for consistent summaries
                 max_tokens=500,  # Cap summary length
@@ -458,12 +460,13 @@ async def _create_summary(
         # Reset token tracker (conversation now starts from summary)
         await self.tracker.reset_session(session_id)
 
-        # Re-add tokens for remaining messages
+        # Re-add tokens for remaining messages.
+        # #13694: was an inline `len(text) // 4` — a fourth estimator beside the
+        # three that already existed. These are the numbers the 80/90% trigger
+        # runs on until the next provider response supplies an authoritative
+        # count, so they use the shared fast path.
         for msg in messages[split_point:]:
-            # Estimate tokens (rough approximation)
-            text = msg.get("text", "")
-            estimated_tokens = len(text) // 4
-            await self.tracker.add_message_tokens(session_id, prompt_tokens=estimated_tokens)
+            await self.tracker.add_message_tokens(session_id, prompt_tokens=estimate_fast(msg.get("text", "")))
 
         return summary
 

autobot-backend/memory/essential_story.py

@@ -20,6 +20,7 @@
 import yaml
 
 from autobot_shared.logging_manager import get_logger
+from autobot_shared.token_count import estimate_fast
 
 logger = get_logger(__name__)
 
@@ -193,8 +194,19 @@ async def _get_token_budget(self, model_name: str) -> int:
             return _DEFAULT_BUDGET
 
     def _estimate_tokens(self, text: str) -> int:
-        """Approximate token count: word count * 1.3."""
-        return int(len(text.split()) * 1.3)
+        """Approximate token count via the shared estimator (#13694).
+
+        Was ``len(text.split()) * 1.3``, which returns **1** for any
+        non-space-delimited script — so a CJK essential story was budgeted as a
+        single token and every fact fitted. AutoBot is i18n across 11 locales,
+        so that is not hypothetical.
+
+        The replacement is not accurate either; see
+        ``autobot_shared.token_count`` for the measured error in both
+        directions. What changes is that this site now agrees with every other
+        token estimate in the system instead of disagreeing with all of them.
+        """
+        return estimate_fast(text)
 
     async def _fetch_top_facts(self, max_tokens: int) -> List[Dict[str, Any]]:
         """Query KB, sort by quality_score desc, return top facts within budget."""

autobot-backend/services/memory/compression.py

@@ -15,27 +15,30 @@
 import yaml
 
 from autobot_shared.logging_manager import get_logger
+from autobot_shared.ssot_constants import CategoryDefaults
+from autobot_shared.token_count import estimate_fast
 
 logger = get_logger(__name__)
 
 # Default threshold: only compress for models with context_window_tokens <= 8192.
 _DEFAULT_COMPRESSION_THRESHOLD = 8192
 
-# Token estimation: words * 1.3 approximates BPE token count without an
-# external tokenizer dependency (GPT-style subword tokenisation averages
-# ~1.3 tokens per whitespace-delimited word for English prose).
-_WORDS_TO_TOKENS = 1.3
-
 _CONFIG_PATH = Path(__file__).parent.parent.parent / "config" / "context_windows.yaml"
 
 
 def _estimate_tokens(text: str) -> int:
-    """Estimate token count using word-count heuristic.
+    """Estimate token count (#13694).
+
+    Was ``len(text.split()) * 1.3``, whose docstring conceded it held "for
+    English prose". Measured against the character estimate it under-counted
+    JSON by ~19x and CJK by ~26x — ``text.split()`` returns one element for a
+    non-space-delimited script — and this number gates whether compression runs
+    at all. Under-counting means the threshold is never reached on exactly the
+    payloads that need compressing.
 
-    Approximation: len(text.split()) * 1.3.  Avoids any external tokenizer
-    dependency while staying within ~10 % of real BPE counts for English prose.
+    Delegates to the shared fast path so this is not a fourth estimator.
     """
-    return int(len(text.split()) * _WORDS_TO_TOKENS)
+    return estimate_fast(text)
 
 
 def _message_tokens(msg: Dict[str, Any]) -> int:
@@ -162,7 +165,7 @@ def _trim_with_summary(self, messages: List[Dict[str, Any]], target_tokens: int)
             return messages
 
         summary = {
-            "role": "assistant",
+            "role": CategoryDefaults.ROLE_ASSISTANT,
             "content": (
                 f"[Summary: {dropped_count} earlier message(s) were omitted to fit"
                 f" the {target_tokens}-token history budget.]"

autobot-backend/tests/memory/test_essential_story.py

@@ -18,6 +18,8 @@
 
 import pytest
 
+from autobot_shared.token_count import estimate_fast
+
 # #11796: snapshot sys.modules before the stub bootstrap.  The stubs below
 # leaked (e.g. a bare "knowledge._composed" stub broke the real knowledge
 # package's lazy `from knowledge import KnowledgeBase` for every test module
@@ -188,9 +190,32 @@ def test_empty_string(self):
         assert gen._estimate_tokens("") == 0
 
     def test_ten_words(self):
+        """#13694: agreement with the canonical estimator, not words * 1.3.
+
+        The old assertion was ``int(10 * 1.3)``. That formula was replaced
+        because ``text.split()`` sees one element for a non-space-delimited
+        script, so it under-counted CJK by ~26x and JSON by ~19x — and this
+        number decides how much of the story fits in the budget.
+
+        Asserted against the shared helper rather than a literal, so it pins
+        agreement rather than a second copy of the arithmetic.
+        """
         gen = EssentialStoryGenerator()
         text = "one two three four five six seven eight nine ten"
-        assert gen._estimate_tokens(text) == int(10 * 1.3)
+
+        assert gen._estimate_tokens(text) == estimate_fast(text)
+
+    def test_a_non_space_delimited_script_is_not_under_counted(self):
+        """The defect itself, not its arithmetic.
+
+        A value-only test goes green again on any ratio tweak; this one fails
+        for the whole class of input the previous formula could not see.
+        """
+        gen = EssentialStoryGenerator()
+        cjk = "\u4e2d\u6587\u5b57\u7b26\u4e32" * 40  # 200 chars, one "word" to split()
+
+        assert len(cjk.split()) == 1, "fixture no longer exercises the defect"
+        assert gen._estimate_tokens(cjk) > 10
 
 
 class TestFetchTopFacts:
@@ -218,7 +243,8 @@ async def test_sorts_by_quality_desc(self):
     @pytest.mark.asyncio
     async def test_respects_token_budget(self):
         """Only facts that fit within max_tokens are returned."""
-        # Each fact ~10 words * 1.3 = 13 tokens
+        # Budget maths follows the canonical character estimate (#13694),
+        # not the removed words * 1.3 ratio.
         facts = _make_facts(20, quality_step=0.05)
         mock_kb = AsyncMock()
         mock_kb.get_all_facts = AsyncMock(return_value=facts)

autobot-backend/tests/services/test_context_compression.py

@@ -6,6 +6,7 @@
 
 import pytest
 
+from autobot_shared.token_count import estimate_fast
 from services.memory.compression import (
     _DEFAULT_COMPRESSION_THRESHOLD,
     ContextCompressionService,
@@ -41,9 +42,33 @@ def test_single_word(self) -> None:
         assert _estimate_tokens("hello") == 1
 
     def test_multiple_words(self) -> None:
+        """#13694: the estimate is character-based now, not words * 1.3.
+
+        The old assertion was ``10 * 1.3 = 13``. That formula was replaced
+        because it under-counted JSON by ~19x and CJK by ~26x, and this number
+        gates whether compression runs at all — so under-counting disabled
+        compression on exactly the payloads that needed it.
+
+        Asserted against the shared helper rather than a literal, so this test
+        pins *agreement with the canonical estimator* rather than a second copy
+        of its arithmetic that would drift the moment the ratio is tuned.
+        """
         text = " ".join(["word"] * 10)
-        # 10 * 1.3 = 13
-        assert _estimate_tokens(text) == 13
+
+        assert _estimate_tokens(text) == estimate_fast(text)
+
+    def test_a_non_space_delimited_script_is_not_under_counted(self) -> None:
+        """The defect the change exists to fix, not just its arithmetic.
+
+        ``text.split()`` returns a single element for CJK, so the old estimator
+        scored a long Chinese string as 1 word -> 1 token. A value-only test
+        would go green again on any ratio tweak; this one fails for the whole
+        class of input the previous formula could not see.
+        """
+        cjk = "\u4e2d\u6587\u5b57\u7b26\u4e32" * 40  # 200 chars, one "word" to split()
+
+        assert len(cjk.split()) == 1, "fixture no longer exercises the defect"
+        assert _estimate_tokens(cjk) > 10, "a non-space-delimited script is being counted as one token"
 
 
 # ---------------------------------------------------------------------------
@@ -109,7 +134,7 @@ async def test_drops_old_messages_and_prepends_summary(self) -> None:
         # Create 5 messages; each ~6 tokens ("word word word word word .")
         messages = [_msg("user", " ".join(["word"] * 5 + ["."])) for _ in range(5)]
         # Allow budget for only 2 messages (~6 tokens each -> 12 tokens needed)
-        # Each message is 6 words * 1.3 = 7 tokens
+        # Budget maths follows the canonical character estimate (#13694).
         result = await svc.compress_history(messages, 15)
         # First element must be an assistant summary (system mid-history is invalid)
         assert result[0]["role"] == "assistant"

autobot-backend/tests/test_token_count_boundaries.py

@@ -0,0 +1,129 @@
+# Copyright 2025-2026 mrveiss
+# SPDX-License-Identifier: Apache-2.0
+# AutoBot - AI-Powered Automation Platform
+# Author: mrveiss
+"""Token counting at context-fit decision boundaries (#13694).
+
+Every context-fitting decision was made against `len(text.split()) * 1.3`,
+whose own docstring conceded it held "for English prose" — which is exactly
+where large prompts do not come from.
+
+The three failure classes below are the ones the issue named, and the numbers
+are why the fix matters rather than being tidiness: the estimate gates the 90%
+auto-summarise trigger and whether compression runs at all, so under-counting
+means the threshold is reached only after the real window is blown.
+"""
+
+import pytest
+
+from autobot_shared.token_count import estimate_fast, exact_from_usage, resolve_tokens
+
+
+# The heuristic this replaced, kept here as the thing under comparison.
+def _words_times_1_3(text: str) -> int:
+    return int(len(text.split()) * 1.3)
+
+
+CODE = (
+    "async def get_chat_history_manager(request: Request) -> ChatHistoryManager:\n"
+    "    return await ResourceFactory.get_chat_history_manager(request)"
+)
+JSON_PAYLOAD = '{"rows":[{"id":1,"name":"a"},{"id":2,"name":"b"}],"meta":{"total":2,"ok":true}}'
+CJK = "部署已於上午九點完成並且沒有任何錯誤發生系統運行正常" * 4
+PROSE = "The deployment ran at nine in the morning and completed without any errors at all."
+
+
+class TestTheThreeFailureClasses:
+    """Each asserts the new estimate no longer collapses the way the old one did."""
+
+    def test_cjk_is_no_longer_counted_as_a_single_word(self):
+        """`text.split()` returns ONE element for a non-space-delimited script.
+
+        AutoBot is i18n across 11 locales, so this is not hypothetical.
+        """
+        old = _words_times_1_3(CJK)
+        new = estimate_fast(CJK)
+
+        assert old <= 2, "precondition: the old heuristic collapses CJK"
+        assert new > 10 * old, f"expected a large correction, got {old} -> {new}"
+
+    def test_json_is_no_longer_counted_as_a_single_word(self):
+        old = _words_times_1_3(JSON_PAYLOAD)
+        new = estimate_fast(JSON_PAYLOAD)
+
+        assert new > 10 * old, f"punctuation-dense payload still under-counted: {old} -> {new}"
+
+    def test_code_is_no_longer_under_counted(self):
+        """Identifiers are one 'word' and several tokens."""
+        old = _words_times_1_3(CODE)
+        new = estimate_fast(CODE)
+
+        assert new > 2 * old, f"code still under-counted: {old} -> {new}"
+
+    def test_english_prose_is_substantially_unchanged(self):
+        """The old heuristic was genuinely fine here — the fix must not
+        overcorrect on the case it handled well."""
+        old = _words_times_1_3(PROSE)
+        new = estimate_fast(PROSE)
+
+        assert 0.5 * old <= new <= 2 * old, f"prose drifted: {old} -> {new}"
+
+
+class TestProviderCountsWin:
+    """AC: where the provider returns authoritative usage, prefer it."""
+
+    def test_total_tokens_is_used_when_present(self):
+        assert exact_from_usage({"total_tokens": 1234}) == 1234
+
+    def test_prompt_and_completion_are_summed_when_total_is_absent(self):
+        assert exact_from_usage({"prompt_tokens": 100, "completion_tokens": 50}) == 150
+
+    def test_absent_usage_is_none_not_zero(self):
+        """None means 'no authoritative count', which is different from 'zero
+        tokens' — conflating them would silently pin a session at 0% fill."""
+        assert exact_from_usage(None) is None
+        assert exact_from_usage({}) is None
+        assert exact_from_usage({"prompt_tokens": 0, "completion_tokens": 0}) is None
+
+    def test_resolve_prefers_the_provider_over_the_estimate(self):
+        text = "x" * 4000  # would estimate ~1000
+
+        assert resolve_tokens(text, {"total_tokens": 7}) == 7
+
+    def test_resolve_falls_back_knowingly_when_no_usage(self):
+        text = "x" * 4000
+
+        assert resolve_tokens(text, None) == estimate_fast(text)
+
+
+class TestNoFourthEstimator:
+    """AC: one shared helper; no second estimation concept introduced."""
+
+    def test_the_fast_path_delegates_to_the_canonical_estimator(self):
+        from autobot_shared.doc_chunking import estimate_tokens
+
+        for text in (PROSE, CODE, JSON_PAYLOAD, CJK):
+            assert estimate_fast(text) == estimate_tokens(text)
+
+    def test_the_compression_gate_uses_the_shared_helper(self):
+        """`compression._estimate_tokens` gated whether compression runs at all."""
+        from services.memory.compression import _estimate_tokens
+
+        for text in (CODE, JSON_PAYLOAD, CJK):
+            assert _estimate_tokens(text) == estimate_fast(text)
+
+    def test_empty_text_is_zero_not_an_error(self):
+        assert estimate_fast("") == 0
+
+
+class TestTheGateActuallyMoves:
+    """The point of the fix: the corrected count changes the decision."""
+
+    @pytest.mark.parametrize("payload", [JSON_PAYLOAD, CJK])
+    def test_a_payload_that_never_tripped_the_threshold_now_can(self, payload):
+        # Scaled to a realistic conversation's worth of one payload type.
+        big = payload * 600
+        threshold = 8192  # services/memory/compression._DEFAULT_COMPRESSION_THRESHOLD
+
+        assert _words_times_1_3(big) < threshold, "precondition: old estimate stayed under"
+        assert estimate_fast(big) > threshold, "corrected estimate must now trip it"