Fix/ccr tighten proactive expansion eligibility

.github/workflows/opencode-plugin.yml

@@ -11,12 +11,14 @@ on:
     paths:
       - "plugins/opencode/**"
       - "headroom/providers/opencode/_dist/**"
+      - "headroom/providers/opencode/hook-shim/**"
       - ".github/workflows/opencode-plugin.yml"
   push:
     branches: [main]
     paths:
       - "plugins/opencode/**"
       - "headroom/providers/opencode/_dist/**"
+      - "headroom/providers/opencode/hook-shim/**"
       - ".github/workflows/opencode-plugin.yml"
 
 permissions:
@@ -51,3 +53,6 @@ jobs:
           cmp dist-standalone/entry.opencode.js \
             ../../headroom/providers/opencode/_dist/entry.opencode.js \
             || { echo "::error::headroom/providers/opencode/_dist/entry.opencode.js is stale - run 'npm run build:standalone' in plugins/opencode and commit the result"; exit 1; }
+          cmp dist-standalone/hook-shim/handler.js \
+            ../../headroom/providers/opencode/hook-shim/handler.js \
+            || { echo "::error::headroom/providers/opencode/hook-shim/handler.js is stale - run 'npm run build:standalone' in plugins/opencode and commit the result"; exit 1; }

headroom/backends/anyllm.py

@@ -25,6 +25,44 @@
     AnyLLM = None  # type: ignore
 
 
+def _convert_anthropic_tool(tool: dict[str, Any]) -> dict[str, Any]:
+    """Convert an Anthropic tool definition to the OpenAI function shape.
+
+    any-llm speaks OpenAI, so an Anthropic ``{name, description, input_schema}``
+    tool must become ``{type: function, function: {name, description,
+    parameters}}`` before it is forwarded, or the provider ignores/rejects the
+    tools array and the model never calls a tool. Mirrors the LiteLLM backend's
+    converter so both OpenAI-compatible backends send the same shape.
+    """
+    func: dict[str, Any] = {"name": tool.get("name", "")}
+    if "description" in tool:
+        func["description"] = tool["description"]
+    if "input_schema" in tool:
+        func["parameters"] = tool["input_schema"]
+    return {"type": "function", "function": func}
+
+
+def _convert_tool_choice(choice: Any) -> Any:
+    """Convert an Anthropic ``tool_choice`` to the OpenAI shape (mirrors LiteLLM).
+
+    Anthropic: ``{"type": "auto"}``, ``{"type": "any"}``, ``{"type": "tool",
+    "name": ...}``. OpenAI: ``"auto"``, ``"required"``, ``{"type": "function",
+    "function": {"name": ...}}``. Passing the raw Anthropic dict through makes
+    the provider reject or ignore it.
+    """
+    if isinstance(choice, str):
+        return choice
+    if isinstance(choice, dict):
+        choice_type = choice.get("type", "auto")
+        if choice_type == "auto":
+            return "auto"
+        if choice_type == "any":
+            return "required"
+        if choice_type == "tool":
+            return {"type": "function", "function": {"name": choice.get("name", "")}}
+    return "auto"
+
+
 class AnyLLMBackend(Backend):
     """Backend using any-llm for multi-provider support."""
 
@@ -251,9 +289,9 @@ async def send_message(
             if "stop_sequences" in body:
                 kwargs["stop"] = body["stop_sequences"]
             if "tools" in body:
-                kwargs["tools"] = body["tools"]
+                kwargs["tools"] = [_convert_anthropic_tool(t) for t in body["tools"]]
             if "tool_choice" in body:
-                kwargs["tool_choice"] = body["tool_choice"]
+                kwargs["tool_choice"] = _convert_tool_choice(body["tool_choice"])
 
             logger.debug(f"any-llm request: provider={self.provider}, model={original_model}")
 
@@ -301,9 +339,9 @@ async def stream_message(
             if "stop_sequences" in body:
                 kwargs["stop"] = body["stop_sequences"]
             if "tools" in body:
-                kwargs["tools"] = body["tools"]
+                kwargs["tools"] = [_convert_anthropic_tool(t) for t in body["tools"]]
             if "tool_choice" in body:
-                kwargs["tool_choice"] = body["tool_choice"]
+                kwargs["tool_choice"] = _convert_tool_choice(body["tool_choice"])
 
             msg_id = f"msg_{uuid.uuid4().hex[:24]}"
 
@@ -324,42 +362,131 @@ async def stream_message(
                 },
             )
 
-            yield StreamEvent(
-                event_type="content_block_start",
-                data={
-                    "type": "content_block_start",
-                    "index": 0,
-                    "content_block": {"type": "text", "text": ""},
-                },
-            )
-
             stream_response = await self.llm.acompletion(**kwargs)
             output_tokens = 0
+            # Stream text immediately in a single text block, but BUFFER tool
+            # calls and emit them as complete blocks at the end. OpenAI streams
+            # parallel tool calls interleaved by index (index 0 and 1 introduced
+            # together, then a fragment for 0, then for 1), while Anthropic
+            # requires each content block to be fully emitted — start, deltas,
+            # stop — before the next opens. Reassembling per index and flushing
+            # complete blocks keeps every delta inside its own block's start/stop
+            # for any interleaving. (The previous version pre-opened one text
+            # block and dropped tool calls entirely; a naive open-on-new-index
+            # instead mis-sequenced parallel calls, emitting a fragment for an
+            # already-stopped block.)
+            current_block_index = -1
+            text_block_open = False
+            # provider tool index -> {"id", "name", "arguments"}, first-seen order
+            tool_calls: dict[int, dict[str, Any]] = {}
+            tool_order: list[int] = []
+            stop_reason = "end_turn"
 
             async for chunk in cast(AsyncIterator[Any], stream_response):
-                if hasattr(chunk, "choices") and chunk.choices:
-                    delta = chunk.choices[0].delta
-                    if hasattr(delta, "content") and delta.content:
+                if not (hasattr(chunk, "choices") and chunk.choices):
+                    continue
+                choice = chunk.choices[0]
+                delta = choice.delta
+
+                # Map OpenAI finish_reason to the Anthropic stop_reason so a tool
+                # call or a length truncation is not reported as end_turn.
+                finish_reason = getattr(choice, "finish_reason", None)
+                if finish_reason == "tool_calls":
+                    stop_reason = "tool_use"
+                elif finish_reason == "length":
+                    stop_reason = "max_tokens"
+                elif finish_reason == "stop":
+                    stop_reason = "end_turn"
+
+                if getattr(delta, "tool_calls", None):
+                    for tc in delta.tool_calls:
+                        idx = tc.index if getattr(tc, "index", None) is not None else 0
+                        buf = tool_calls.get(idx)
+                        if buf is None:
+                            buf = {"id": None, "name": "", "arguments": ""}
+                            tool_calls[idx] = buf
+                            tool_order.append(idx)
+                        if getattr(tc, "id", None):
+                            buf["id"] = tc.id
+                        func = getattr(tc, "function", None)
+                        if func is not None:
+                            if getattr(func, "name", None):
+                                buf["name"] = func.name
+                            if getattr(func, "arguments", None):
+                                buf["arguments"] += func.arguments
+
+                elif getattr(delta, "content", None):
+                    if not text_block_open:
+                        current_block_index += 1
+                        text_block_open = True
                         yield StreamEvent(
-                            event_type="content_block_delta",
+                            event_type="content_block_start",
                             data={
-                                "type": "content_block_delta",
-                                "index": 0,
-                                "delta": {"type": "text_delta", "text": delta.content},
+                                "type": "content_block_start",
+                                "index": current_block_index,
+                                "content_block": {"type": "text", "text": ""},
                             },
                         )
-                        output_tokens += 1
+                    yield StreamEvent(
+                        event_type="content_block_delta",
+                        data={
+                            "type": "content_block_delta",
+                            "index": current_block_index,
+                            "delta": {"type": "text_delta", "text": delta.content},
+                        },
+                    )
+                    output_tokens += 1
+
+            # Close the text block before any tool blocks (Anthropic orders
+            # content blocks sequentially, text then tool_use).
+            if text_block_open:
+                yield StreamEvent(
+                    event_type="content_block_stop",
+                    data={"type": "content_block_stop", "index": current_block_index},
+                )
 
-            yield StreamEvent(
-                event_type="content_block_stop",
-                data={"type": "content_block_stop", "index": 0},
-            )
+            # Flush each buffered tool call as a complete, self-contained block:
+            # start, one input_json_delta with the reassembled arguments, stop.
+            for idx in tool_order:
+                buf = tool_calls[idx]
+                current_block_index += 1
+                tool_id = buf["id"] or f"toolu_{uuid.uuid4().hex[:24]}"
+                yield StreamEvent(
+                    event_type="content_block_start",
+                    data={
+                        "type": "content_block_start",
+                        "index": current_block_index,
+                        "content_block": {
+                            "type": "tool_use",
+                            "id": tool_id,
+                            "name": buf["name"],
+                            "input": {},
+                        },
+                    },
+                )
+                if buf["arguments"]:
+                    yield StreamEvent(
+                        event_type="content_block_delta",
+                        data={
+                            "type": "content_block_delta",
+                            "index": current_block_index,
+                            "delta": {
+                                "type": "input_json_delta",
+                                "partial_json": buf["arguments"],
+                            },
+                        },
+                    )
+                    output_tokens += 1
+                yield StreamEvent(
+                    event_type="content_block_stop",
+                    data={"type": "content_block_stop", "index": current_block_index},
+                )
 
             yield StreamEvent(
                 event_type="message_delta",
                 data={
                     "type": "message_delta",
-                    "delta": {"stop_reason": "end_turn", "stop_sequence": None},
+                    "delta": {"stop_reason": stop_reason, "stop_sequence": None},
                     "usage": {"output_tokens": output_tokens},
                 },
             )

headroom/cache/compression_cache.py

@@ -135,8 +135,10 @@ def __init__(self, max_entries: int = 10000) -> None:
         # `compute_frozen_count` (bounded above by the `min` clamp at
         # `proxy/handlers/anthropic.py`) and `update_from_result`'s
         # "unchanged content" tracking.
-        self._stable_hashes: set[str] = set()
-        self._first_seen: dict[str, float] = {}
+        # Ordered mappings preserve set/dict-style membership while allowing
+        # deterministic oldest-first eviction.
+        self._stable_hashes: OrderedDict[str, None] = OrderedDict()
+        self._first_seen: OrderedDict[str, float] = OrderedDict()
         self._hits: int = 0
         self._misses: int = 0
         self._total_tokens_saved: int = 0
@@ -172,6 +174,34 @@ def store_compressed(self, hash: str, compressed: str, tokens_saved: int) -> Non
                 _, evicted = self._cache.popitem(last=False)
                 self._total_tokens_saved -= evicted.tokens_saved
 
+    def _mark_stable_locked(self, content_hash: str) -> None:
+        """Record a stable hash while bounding retained bookkeeping."""
+        self._stable_hashes[content_hash] = None
+        self._stable_hashes.move_to_end(content_hash)
+
+        while len(self._stable_hashes) > self.max_entries:
+            self._stable_hashes.popitem(last=False)
+
+    def _record_first_seen_locked(self, content_hash: str, seen_at: float) -> None:
+        """Record a first-seen timestamp while bounding retained bookkeeping."""
+        self._first_seen[content_hash] = seen_at
+        self._first_seen.move_to_end(content_hash)
+
+        while len(self._first_seen) > self.max_entries:
+            self._first_seen.popitem(last=False)
+
+    def _prune_expired_first_seen_locked(
+        self,
+        now: float,
+        ttl_seconds: float,
+    ) -> None:
+        """Remove first-seen entries whose cache timing window has expired."""
+        while self._first_seen:
+            _, oldest_seen_at = next(iter(self._first_seen.items()))
+            if now - oldest_seen_at < ttl_seconds:
+                break
+            self._first_seen.popitem(last=False)
+
     def mark_stable(self, content_hash: str) -> None:
         """Mark a content hash as stable (unchanged, not compressed).
 
@@ -180,7 +210,7 @@ def mark_stable(self, content_hash: str) -> None:
         even though no compressed version exists in the cache.
         """
         with self._lock:
-            self._stable_hashes.add(content_hash)
+            self._mark_stable_locked(content_hash)
 
     def mark_stable_from_messages(self, messages: list[dict], up_to: int) -> None:
         """Mark all tool_result hashes in messages[:up_to] as stable."""
@@ -189,7 +219,7 @@ def mark_stable_from_messages(self, messages: list[dict], up_to: int) -> None:
                 if _is_tool_result_message(msg):
                     content = _extract_tool_result_content(msg)
                     if content is not None:
-                        self._stable_hashes.add(self.content_hash(content))
+                        self._mark_stable_locked(self.content_hash(content))
 
     def should_defer_compression(
         self,
@@ -216,13 +246,18 @@ def should_defer_compression(
         """
         with self._lock:
             now = time.time()
+            self._prune_expired_first_seen_locked(now, ttl_seconds)
+
             first_seen = self._first_seen.get(content_hash)
             if first_seen is None:
-                self._first_seen[content_hash] = now
+                self._record_first_seen_locked(content_hash, now)
                 return False  # First time — compress now (no cache entry to preserve)
+
             age = now - first_seen
             if age >= ttl_seconds - batch_window:
+                self._record_first_seen_locked(content_hash, now)
                 return False  # Near TTL boundary — compress now (batch window)
+
             return True  # Seen recently within TTL — defer to preserve cache
 
     def get_stats(self) -> dict:
@@ -335,7 +370,7 @@ def update_from_result(self, originals: list[dict], compressed: list[dict]) -> N
                     continue
                 if orig_content == comp_content:
                     # Content unchanged — mark as stable for frozen count walk
-                    self._stable_hashes.add(self.content_hash(orig_content))
+                    self._mark_stable_locked(self.content_hash(orig_content))
                     continue
                 h = self.content_hash(orig_content)
                 tokens_saved = len(orig_content) // 4 - len(comp_content) // 4

headroom/cache/compression_store.py

@@ -52,6 +52,10 @@
 CCR_TTL_SECONDS_ENV = "HEADROOM_CCR_TTL_SECONDS"
 
 _RETRIEVAL_LOG_PREVIEW_CHARS = 4096
+# Previews carry verbatim tool-result content (post-redaction), which makes
+# proxy.log too sensitive for users to share in bug reports. Set to
+# 0/false/no/off to log byte counts only.
+PAYLOAD_PREVIEW_ENV = "HEADROOM_LOG_PAYLOAD_PREVIEW"
 _SECRET_KEY_VALUE_RE = re.compile(
     r"(?i)\b([A-Z0-9_-]*(?:API[_-]?KEY|TOKEN|SECRET|PASSWORD|CREDENTIAL|AUTH)[A-Z0-9_-]*)"
     r"(\s*[:=]\s*)([\"']?)([^\"'\s,}]+)"
@@ -108,7 +112,21 @@ def _redact_retrieval_log_payload(payload: str) -> str:
     return _API_KEY_VALUE_RE.sub("sk-[REDACTED]", redacted)
 
 
+def _payload_preview_enabled() -> bool:
+    raw = os.environ.get(PAYLOAD_PREVIEW_ENV)
+    if raw is None:
+        return True
+    return raw.strip().lower() not in ("0", "false", "no", "off")
+
+
 def _payload_for_retrieval_log(payload: str) -> dict[str, Any]:
+    if not _payload_preview_enabled():
+        return {
+            "payload_chars": len(payload),
+            "payload_preview_chars": 0,
+            "payload_truncated": len(payload) > 0,
+            "payload_preview": "",
+        }
     redacted = _redact_retrieval_log_payload(payload)
     preview = redacted[:_RETRIEVAL_LOG_PREVIEW_CHARS]
     truncated = len(redacted) > len(preview)