Merge branch 'headroomlabs-ai:main' into main

Author: nangsontay

.github/workflows/ci.yml

@@ -43,7 +43,9 @@ jobs:
     timeout-minutes: 5
     outputs:
       code: ${{ steps.filter.outputs.code }}
-      e2e: ${{ steps.filter.outputs.e2e }}
+      native: ${{ steps.filter.outputs.native }}
+      dashboard: ${{ steps.filter.outputs.dashboard }}
+      packaging: ${{ steps.filter.outputs.packaging }}
       workflows: ${{ steps.filter.outputs.workflows }}
     steps:
       - uses: actions/checkout@v7
@@ -61,14 +63,41 @@ jobs:
               - 'tests/**'
               - 'scripts/**'
               - '.github/workflows/**'
-            e2e:
-              - 'headroom/**'
+            # native = anything that can change the compiled wrapper, the native
+            # install flow, or the docker image (drives the scarce macOS/Windows
+            # runners + docker E2E). A pure-Python logic change hits none of these.
+            native:
+              - 'headroom/cli/**'
+              - 'headroom/install/**'
+              - 'headroom/providers/**'
+              - 'headroom/rtk/**'
               - 'crates/**'
+              - '**/*.rs'
+              - 'Cargo.toml'
+              - 'Cargo.lock'
+              - 'rust-toolchain.toml'
               - 'docker/**'
               - 'Dockerfile'
               - 'e2e/**'
               - 'scripts/install*'
               - 'pyproject.toml'
+              - '.github/workflows/**'
+            dashboard:
+              - 'headroom/dashboard/**'
+              - '.github/workflows/**'
+            # packaging = anything that changes how the wheel is built (so the
+            # cross-platform wheel build only reruns when the build actually changes).
+            packaging:
+              - 'pyproject.toml'
+              - 'Cargo.toml'
+              - 'Cargo.lock'
+              - 'uv.lock'
+              - 'rust-toolchain.toml'
+              - 'crates/**'
+              - '**/*.rs'
+              - 'scripts/**'
+              - 'MANIFEST.in'
+              - '.github/workflows/**'
             workflows:
               - '.github/workflows/**'
 
@@ -122,7 +151,7 @@ jobs:
 
   build-wheel-windows:
     needs: changes
-    if: needs.changes.outputs.code == 'true'
+    if: needs.changes.outputs.packaging == 'true'
     runs-on: windows-latest
     timeout-minutes: 45
     steps:
@@ -354,7 +383,7 @@ jobs:
 
   test-dashboard-ui:
     needs: [changes, build-wheel]
-    if: needs.changes.outputs.code == 'true'
+    if: needs.changes.outputs.dashboard == 'true'
     runs-on: ubuntu-latest
     timeout-minutes: 20
     steps:
@@ -474,7 +503,7 @@ jobs:
 
   docker-native-e2e:
     needs: changes
-    if: needs.changes.outputs.e2e == 'true'
+    if: needs.changes.outputs.native == 'true'
     runs-on: ubuntu-latest
     timeout-minutes: 45
     steps:
@@ -518,7 +547,7 @@ jobs:
 
   windows-native-wrapper:
     needs: changes
-    if: needs.changes.outputs.e2e == 'true'
+    if: needs.changes.outputs.native == 'true'
     runs-on: windows-latest
     timeout-minutes: 20
     steps:
@@ -535,7 +564,7 @@ jobs:
 
   macos-native-wrapper:
     needs: changes
-    if: needs.changes.outputs.e2e == 'true'
+    if: needs.changes.outputs.native == 'true'
     runs-on: macos-latest
     timeout-minutes: 20
     steps:

.github/workflows/init-e2e.yml

@@ -3,8 +3,12 @@ name: Init E2E
 on:
   pull_request:
     branches: [main]
+    # Scoped to what the `headroom init` flow actually exercises (mirrors
+    # init-native-e2e), not all of headroom/** — a pure-Python logic change
+    # elsewhere shouldn't spin up a docker init E2E.
     paths:
-      - 'headroom/**'
+      - 'headroom/cli/**'
+      - 'headroom/install/**'
       - 'crates/**'
       - 'docker/**'
       - 'Dockerfile'

.github/workflows/wrap-e2e.yml

@@ -3,8 +3,13 @@ name: Wrap E2E
 on:
   pull_request:
     branches: [main]
+    # Scoped to what the `headroom wrap` flow actually exercises (mirrors
+    # wrap-native-e2e), not all of headroom/** — a pure-Python logic change
+    # elsewhere shouldn't spin up a docker wrap E2E.
     paths:
-      - 'headroom/**'
+      - 'headroom/cli/**'
+      - 'headroom/providers/**'
+      - 'headroom/rtk/**'
       - 'crates/**'
       - 'docker/**'
       - 'Dockerfile'

CHANGELOG.md

@@ -158,9 +158,15 @@ def apply_proxy_env_defaults(self, env: MutableMapping[str, str]) -> MutableMapp
         # append-only forwarding), not by refusing to touch user turns.
         compress_user_messages=True,
         compress_system_messages=False,  # system prompt is the hottest cache
-        protect_recent=2,  # keep the active code working set verbatim
+        # Compress the newest observation delta. In cache mode the delta IS the
+        # newest turn(s), so a positional guard here suppresses the only thing
+        # cache mode can compress (large recent grep/test/build tool output).
+        # The code working set stays byte-exact via protect_reads below — a TYPE
+        # guard on file reads — which is the real fidelity protection, so 0 loses
+        # no correctness while unblocking compression of recent non-read output.
+        protect_recent=0,
         protect_analysis_context=True,
-        min_tokens_to_compress=25,  # low → compression is visible
+        min_tokens_to_compress=10,  # low → even modest deltas are eligible
         max_items_after_crush=15,
         smart_crusher_with_compaction=True,
         force_kompress=False,  # don't override diff/log lossless with lossy ML

headroom/agent_savings.py

@@ -669,13 +669,20 @@ def _convert_messages_for_litellm(self, messages: list[dict[str, Any]]) -> list[
                             tr_content = "\n".join(
                                 b.get("text", "") for b in tr_content if b.get("type") == "text"
                             )
-                        converted.append(
-                            {
-                                "role": "tool",
-                                "tool_call_id": tr["tool_use_id"],
-                                "content": str(tr_content),
-                            }
-                        )
+                        tool_msg: dict[str, Any] = {
+                            "role": "tool",
+                            "tool_call_id": tr["tool_use_id"],
+                            "content": str(tr_content),
+                        }
+                        # Claude Code's moving cache breakpoint usually lands on the
+                        # tail tool_result, not just the system prompt. Carry
+                        # cache_control through so LiteLLM's Bedrock Converse
+                        # transformation can inject a cachePoint here too (#1390
+                        # covers the system-prompt/text-block case; this is the
+                        # tool_result case, out of scope there).
+                        if "cache_control" in tr:
+                            tool_msg["cache_control"] = tr["cache_control"]
+                        converted.append(tool_msg)
                     continue
 
                 # tool_use blocks → OpenAI assistant message with tool_calls
@@ -966,15 +973,39 @@ async def stream_message(
                 },
             )
 
+            # Request usage in the final streaming chunk so cache metrics
+            # (cache_read_input_tokens / cache_creation_input_tokens) come back at
+            # all. Without this, LiteLLM/Bedrock never emits a usage chunk over SSE
+            # and the caller's cache stats always read 0, even when caching is
+            # working server-side.
+            kwargs["stream_options"] = {"include_usage": True}
+
             # Stream content — blocks emitted dynamically based on response
             response = await acompletion(**kwargs)
             output_tokens = 0
             current_block_index = -1
             active_block_type: str | None = None  # "text" or "tool_use"
             tool_block_map: dict[int, int] = {}  # litellm tc.index → SSE block index
             stop_reason = "end_turn"
+            # Populated from the final usage chunk (stream_options.include_usage=True
+            # above). The message_start emitted before this loop always carries
+            # input_tokens=0 and no cache fields because LiteLLM/Bedrock only reports
+            # usage on the trailing chunk. Carry the final cache stats on the terminal
+            # message_delta instead of emitting a second protocol-invalid
+            # message_start after content has already streamed.
+            final_input_tokens = 0
+            final_cache_read_tokens = 0
+            final_cache_write_tokens = 0
 
             async for chunk in response:
+                if hasattr(chunk, "usage") and chunk.usage:
+                    cu = chunk.usage
+                    final_input_tokens = int(getattr(cu, "prompt_tokens", 0) or 0)
+                    final_cache_read_tokens = int(getattr(cu, "cache_read_input_tokens", 0) or 0)
+                    final_cache_write_tokens = int(
+                        getattr(cu, "cache_creation_input_tokens", 0) or 0
+                    )
+
                 if not hasattr(chunk, "choices") or not chunk.choices:
                     continue
 
@@ -1080,13 +1111,21 @@ async def stream_message(
                     data={"type": "content_block_stop", "index": current_block_index},
                 )
 
+            delta_usage: dict[str, Any] = {"output_tokens": output_tokens}
+            if final_input_tokens or final_cache_read_tokens or final_cache_write_tokens:
+                delta_usage["input_tokens"] = final_input_tokens
+                if final_cache_read_tokens:
+                    delta_usage["cache_read_input_tokens"] = final_cache_read_tokens
+                if final_cache_write_tokens:
+                    delta_usage["cache_creation_input_tokens"] = final_cache_write_tokens
+
             # Emit message_delta with correct stop reason
             yield StreamEvent(
                 event_type="message_delta",
                 data={
                     "type": "message_delta",
                     "delta": {"stop_reason": stop_reason, "stop_sequence": None},
-                    "usage": {"output_tokens": output_tokens},
+                    "usage": delta_usage,
                 },
             )
 

headroom/backends/litellm.py

@@ -40,12 +40,29 @@ def _is_tool_result_message(msg: dict) -> bool:
     return False
 
 
+def _extract_text_from_blocks(blocks: list) -> str | None:
+    """Extract joined text from a list-of-blocks content (e.g. Anthropic list-of-text-blocks).
+
+    Modern Claude Code sends ``tool_result`` content as a list of typed
+    blocks (``[{"type": "text", "text": "..."}]``) instead of a plain
+    string.  This helper extracts text from ``type == "text"`` blocks and
+    joins them.
+    """
+    texts = [b.get("text", "") for b in blocks if isinstance(b, dict) and b.get("type") == "text"]
+    return "\n".join(t for t in texts if t != "") or None
+
+
 def _extract_tool_result_content(msg: dict) -> str | None:
     """Extract text content from a tool result message (both formats)."""
     # OpenAI format
     if msg.get("role") == "tool":
         content = msg.get("content")
-        return content if isinstance(content, str) else None
+        if isinstance(content, str):
+            return content
+        # OpenAI content can also be a list of content parts
+        if isinstance(content, list):
+            return _extract_text_from_blocks(content)
+        return None
     # Anthropic format
     content = msg.get("content")
     if isinstance(content, list):
@@ -54,6 +71,8 @@ def _extract_tool_result_content(msg: dict) -> str | None:
                 inner = block.get("content")
                 if isinstance(inner, str):
                     return inner
+                if isinstance(inner, list):
+                    return _extract_text_from_blocks(inner)
     return None
 
 
@@ -69,7 +88,16 @@ def _swap_tool_result_content(msg: dict, new_content: str) -> dict:
     if isinstance(content, list):
         for block in content:
             if isinstance(block, dict) and block.get("type") == "tool_result":
-                block["content"] = new_content
+                inner = block.get("content")
+                if isinstance(inner, list):
+                    # Collapse list-of-blocks to a single text block.
+                    # The compressed content is a single string; preserving
+                    # multiple text blocks would produce a different joined
+                    # output on re-extraction (first text block replaced,
+                    # remaining text blocks still joined).
+                    block["content"] = [{"type": "text", "text": new_content}]
+                else:
+                    block["content"] = new_content
                 break
     return new_msg
 

headroom/cache/compression_cache.py

@@ -30,6 +30,25 @@
 
 logger = logging.getLogger(__name__)
 
+# Residual-CCR status signals (provider-generic).
+#
+# ``handle_response`` may return a response that still contains
+# ``headroom_retrieve`` tool calls. Callers need to know *why* so they can
+# decide whether that is a safe passthrough or a genuine failure:
+#
+# - RESIDUAL_CCR_RESOLVED:       no CCR tool calls remain — fully handled.
+# - RESIDUAL_CCR_SKIPPED_MIXED:  CCR was intentionally skipped because the model
+#                                emitted headroom_retrieve alongside a non-CCR
+#                                client tool (#839). The client must resolve both
+#                                tool calls; the proxy must pass the turn through
+#                                unchanged (200), not fail closed.
+# - RESIDUAL_CCR_ERROR:          CCR tool calls remain with no accompanying client
+#                                tool — i.e. a real conversion/handling failure the
+#                                proxy could not resolve. Callers should fail closed.
+RESIDUAL_CCR_RESOLVED = "resolved"
+RESIDUAL_CCR_SKIPPED_MIXED = "skipped_mixed_tools"
+RESIDUAL_CCR_ERROR = "error"
+
 
 @dataclass
 class CCRToolResult:
@@ -110,6 +129,33 @@ def has_ccr_tool_calls(
         """
         return has_ccr_tool_calls(response, provider)
 
+    def residual_ccr_status(
+        self,
+        response: dict[str, Any],
+        provider: str = "anthropic",
+    ) -> str:
+        """Classify why (if at all) CCR tool calls remain in a handled response.
+
+        This is a stateless, provider-generic signal derived from the same
+        parsing ``handle_response`` uses, so it stays correct under concurrency
+        and works identically for every provider/harness.
+
+        Returns one of:
+        - ``RESIDUAL_CCR_RESOLVED``: no headroom_retrieve tool calls remain.
+        - ``RESIDUAL_CCR_SKIPPED_MIXED``: headroom_retrieve remains *alongside*
+          a non-CCR client tool call. This is an intentional skip (#839) — the
+          proxy cannot synthesize the client tool_result, so the turn must be
+          handed back to the client unchanged rather than failed closed.
+        - ``RESIDUAL_CCR_ERROR``: headroom_retrieve remains with no accompanying
+          client tool call — a genuine handling/conversion failure.
+        """
+        ccr_calls, other_calls = self._parse_ccr_tool_calls(response, provider)
+        if not ccr_calls:
+            return RESIDUAL_CCR_RESOLVED
+        if other_calls:
+            return RESIDUAL_CCR_SKIPPED_MIXED
+        return RESIDUAL_CCR_ERROR
+
     def _extract_tool_calls(
         self,
         response: dict[str, Any],

headroom/ccr/response_handler.py

@@ -199,6 +199,14 @@ class CCRToolInjector:
             # `<<ccr:HASH,KIND,SIZE>>`. HASH is 12-24 hex chars, terminated by a
             # space, comma, or the closing `>>`.
             re.compile(r"<<ccr:([a-f0-9]{12,24})\b"),
+            # read_lifecycle STALE/SUPERSEDED markers:
+            # `[Read content stale/superseded: ... Retrieve original: hash=xxx]`.
+            # These carry a retrievable CCR hash but never contain the word
+            # "compressed", so the patterns above miss them -- and the retrieve
+            # tool is then not injected, leaving the model a marker it cannot
+            # redeem (silent data loss, #1006). Match the load-bearing
+            # "Retrieve original: hash=" phrase directly.
+            re.compile(r"Retrieve original: hash=([a-f0-9]{12,24})"),
         ]
     )
 

headroom/ccr/tool_injection.py

@@ -458,8 +458,16 @@ async def index_batch(self, memories: list[Memory]) -> int:
                     new_memories.append((memory, embedding, hnsw_id))
                     self._next_hnsw_id += 1
 
-            # Resize if needed
-            required_capacity = len(self._memory_to_hnsw) + len(new_memories)
+            # Resize if needed. hnswlib never frees a slot on mark_deleted
+            # (remove/evict), so its capacity is bounded by the high-water mark
+            # of assigned ids (_next_hnsw_id, already incremented for the new
+            # memories above), NOT the live entry count. After deletions or
+            # evictions the live count is well below _next_hnsw_id, so keying the
+            # resize off `len(self._memory_to_hnsw)` under-provisions and the
+            # add_items below raises "number of elements exceeds the specified
+            # limit". This mirrors the single-item index() guard, which resizes
+            # off _next_hnsw_id.
+            required_capacity = self._next_hnsw_id
             if required_capacity > self._max_elements:
                 new_max = max(self._max_elements * 2, required_capacity + 1000)
                 self._resize_index(new_max)