Sync local changes: compaction telemetry, CBOR disk-full handling, routing and tests

- Add compaction telemetry domain types and metrics recorder; extend history compaction, config, and interfaces.

- Harden CBOR wire capture on ENOSPC/I/O (disable capture, throttle logs, avoid close chaining).

- Propagate BackendError/RateLimitExceededError from ZAI Coding Plan without generic wrapper.

- Update Anthropic connector, backend request preparation, routing compliance script, and related unit/integration tests.

Made-with: Cursor

Author: Mateusz

dev/scripts/check_routing_unification_compliance.py

@@ -21,6 +21,7 @@
 APPROVED_INTERNAL_DELEGATE_CALL_CHAINS = {
     "self.call_completion",
     "self._backend_completion_flow.call_completion",
+    "self._completion_flow.call_completion",
 }
 
 OUTBOUND_CALL_METHOD_SUFFIXES = (

src/connectors/anthropic.py

@@ -26,6 +26,7 @@
 from src.core.common.exceptions import (
     AuthenticationError,
     ConfigurationError,
+    InvalidRequestError,
     ServiceUnavailableError,
 )
 from src.core.config.app_config import AppConfig
@@ -543,6 +544,14 @@ async def chat_completions(  # type: ignore[override]
         through :class:`ConnectorChatCompletionsRequest`; legacy positional call shapes
         are not supported at this boundary.
         """
+        if not isinstance(request, ConnectorChatCompletionsRequest):  # type: ignore[unreachable]
+            raise InvalidRequestError(
+                message=(
+                    f"chat_completions requires ConnectorChatCompletionsRequest, "
+                    f"got {type(request).__name__}"
+                ),
+                details={"connector": "anthropic"},
+            )
         return await self._chat_completions_canonical(request)
 
     # -----------------------------------------------------------

src/connectors/zai_coding_plan.py

@@ -110,7 +110,7 @@ async def initialize(self, **kwargs: Any) -> None:
                 code="missing_api_key",
             )
 
-        # Log masked API key for verification (show first 4 and last 4 chars)
+        # Log masked API key for verification (show first 4 and last 4 chars)
         masked_key = self._mask_api_key(self.api_key)
         if logger.isEnabledFor(logging.INFO):
             logger.info(
@@ -573,6 +573,10 @@ async def chat_completions(  # type: ignore[override]
                 status_code=getattr(exc, "status_code", 502),
                 code=f"zai_error_{getattr(exc, 'status_code', 'unknown')}",
             ) from exc
+        except BackendError:
+            # RateLimitExceededError and other mapped domain errors must propagate
+            # to resilience / HTTP mapping; do not wrap as zai_error_unexpected.
+            raise
         except Exception as exc:
             if logger.isEnabledFor(logging.ERROR):
                 logger.error(
@@ -1030,8 +1034,13 @@ async def _prepare_payload(
             payload["tools"] = request_data.tools
             logger.debug("Including tools: %d tools", len(request_data.tools))
         if hasattr(request_data, "tool_choice") and request_data.tool_choice:
-            payload["tool_choice"] = request_data.tool_choice
-            logger.debug("Including tool_choice: %s", request_data.tool_choice)
+            tool_choice = request_data.tool_choice
+            if isinstance(tool_choice, dict) and tool_choice.get("type") == "function":
+                # ZAI coding-plan rejects function-enforced tool_choice objects;
+                # use OpenAI-compatible automatic tool selection instead.
+                tool_choice = "auto"
+            payload["tool_choice"] = tool_choice
+            logger.debug("Including tool_choice: %s", tool_choice)
 
         # ZAI coding-plan gateway is strict about accepted payload shape.
         # Only forward a minimal set of known-safe keys to avoid WAF/429 triggers.

src/core/config/env/from_env_part3.py

@@ -101,6 +101,50 @@ def _load_replacement_rules_from_env(
         return []
 
 
+def _apply_gemini_backend(
+    config_backends: dict[str, Any],
+    env: Mapping[str, str],
+    gemini_key: str,
+    resolution: ParameterResolution | None,
+) -> None:
+    if logger.isEnabledFor(logging.INFO):
+        _, gemini_key_source = get_env_value_with_windows_persistent_fallback(
+            "GEMINI_API_KEY", environ=env
+        )
+        logger.info(
+            "Gemini key diagnostics [from_env_part3]: env_type=%s source=%s",
+            type(env).__name__,
+            gemini_key_source,
+        )
+
+    config_backends["gemini"] = config_backends.get("gemini", {})
+    config_backends["gemini"]["api_key"] = gemini_key
+    config_backends["gemini"]["api_url"] = _get_env_value(
+        env,
+        "GEMINI_API_BASE_URL",
+        "https://generativelanguage.googleapis.com",
+        path="backends.gemini.api_url",
+        resolution=resolution,
+    )
+    gemini_timeout = _get_env_value(
+        env,
+        "GEMINI_TIMEOUT",
+        None,
+        path="backends.gemini.timeout",
+        resolution=resolution,
+        transform=lambda value: _to_int(value, 0),
+    )
+    if gemini_timeout:
+        config_backends["gemini"]["timeout"] = gemini_timeout
+    if resolution is not None:
+        resolution.record(
+            "backends.gemini.api_key",
+            config_backends["gemini"]["api_key"],
+            ParameterSource.ENVIRONMENT,
+            origin="GEMINI_API_KEY",
+        )
+
+
 def apply_config_part3(
     config: dict[str, Any],
     env: Mapping[str, str],
@@ -137,43 +181,11 @@ def apply_config_part3(
                 origin="OPENROUTER_API_KEY",
             )
 
-    gemini_key, gemini_key_source = get_env_value_with_windows_persistent_fallback(
+    gemini_key, _gemini_key_source = get_env_value_with_windows_persistent_fallback(
         "GEMINI_API_KEY", environ=env
     )
     if gemini_key and not _has_numbered_env_variants(env, "GEMINI_API_KEY"):
-        if logger.isEnabledFor(logging.INFO):
-            logger.info(
-                "Gemini key diagnostics [from_env_part3]: env_type=%s source=%s",
-                type(env).__name__,
-                gemini_key_source,
-            )
-
-        config_backends["gemini"] = config_backends.get("gemini", {})
-        config_backends["gemini"]["api_key"] = gemini_key
-        config_backends["gemini"]["api_url"] = _get_env_value(
-            env,
-            "GEMINI_API_BASE_URL",
-            "https://generativelanguage.googleapis.com",
-            path="backends.gemini.api_url",
-            resolution=resolution,
-        )
-        gemini_timeout = _get_env_value(
-            env,
-            "GEMINI_TIMEOUT",
-            None,
-            path="backends.gemini.timeout",
-            resolution=resolution,
-            transform=lambda value: _to_int(value, 0),
-        )
-        if gemini_timeout:
-            config_backends["gemini"]["timeout"] = gemini_timeout
-        if resolution is not None:
-            resolution.record(
-                "backends.gemini.api_key",
-                config_backends["gemini"]["api_key"],
-                ParameterSource.ENVIRONMENT,
-                origin="GEMINI_API_KEY",
-            )
+        _apply_gemini_backend(config_backends, env, gemini_key, resolution)
 
     if env.get("ANTHROPIC_API_KEY"):
         config_backends["anthropic"] = config_backends.get("anthropic", {})

src/core/domain/compaction_telemetry.py

@@ -0,0 +1,74 @@
+"""Telemetry domain models for history compaction.
+
+Mirrors dynamic compression telemetry shape while keeping compaction
+semantics honest — no fake method pipelines, no recovery handles unless
+intentionally added later.
+"""
+
+from __future__ import annotations
+
+from pydantic import Field
+
+from src.core.interfaces.model_bases import DomainModel
+
+
+class CompactionEventRecord(DomainModel):
+    """Diagnostics for one compaction evaluation (per stale resource or no-op pass)."""
+
+    correlation_id: str | None = None
+    tool_call_id: str | None = None
+    tool_name: str = ""
+    tool_category: str = ""
+    resource_identity_hash: str = ""
+    resource_identity_preview: str | None = None
+    resource_preview_redacted: bool = False
+    original_bytes: int = Field(default=0, ge=0)
+    compacted_bytes: int = Field(default=0, ge=0)
+    saved_bytes: int = Field(default=0, ge=0)
+    original_tokens_estimate: int = Field(default=0, ge=0)
+    saved_tokens_estimate: int = Field(default=0, ge=0)
+    applied: bool = False
+    decision_reason: str = "no_stale_results"
+    failed_open: bool = False
+    failure_reason: str | None = None
+    elapsed_total_ms: float = Field(default=0.0, ge=0)
+    original_sha256: str | None = None
+    compacted_sha256: str | None = None
+    warnings: list[str] = Field(default_factory=list)
+    message_index: int | None = None
+
+
+class CompactionAggregateMetrics(DomainModel):
+    """Running aggregate stats for operator diagnostics."""
+
+    processed_evaluations: int = Field(default=0, ge=0)
+    applied_evaluations: int = Field(default=0, ge=0)
+    fail_open_count: int = Field(default=0, ge=0)
+    total_original_bytes: int = Field(default=0, ge=0)
+    total_compacted_bytes: int = Field(default=0, ge=0)
+    total_saved_bytes: int = Field(default=0, ge=0)
+    total_saved_tokens_estimate: int = Field(default=0, ge=0)
+    by_category: dict[str, int] = Field(default_factory=dict)
+    by_decision_reason: dict[str, int] = Field(default_factory=dict)
+
+
+class CompactionAlertRecord(DomainModel):
+    """Rate-limited alert emitted for frequent compaction issues."""
+
+    alert_type: str
+    threshold: int = Field(ge=1)
+    observed_count: int = Field(ge=0)
+    window_seconds: int = Field(ge=1)
+    warning: str
+    category: str | None = None
+
+
+class EffectiveCompactionConfigDiagnostics(DomainModel):
+    """Redaction-safe effective-configuration diagnostics for operators."""
+
+    active_controls: list[str] = Field(default_factory=list)
+    inactive_controls: list[str] = Field(default_factory=list)
+    ignored_controls: list[str] = Field(default_factory=list)
+    reasons: dict[str, str] = Field(default_factory=dict)
+    fingerprint: str | None = None
+    warnings: list[str] = Field(default_factory=list)

src/core/domain/configuration/compaction_config.py

@@ -15,6 +15,33 @@
 from src.core.domain.compaction import ToolCategory
 
 
+@dataclass
+class CompactionAlertsConfig:
+    """Operator alert thresholds for compaction issues."""
+
+    enabled: bool = False
+    window_seconds: int = 300
+    cooldown_seconds: int = 600
+    fail_open_threshold: int = 3
+    overflow_risk_threshold: int = 3
+    no_op_above_threshold_threshold: int = 5
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any] | None) -> "CompactionAlertsConfig":
+        """Create alert config from dictionary payload."""
+        payload = data or {}
+        return cls(
+            enabled=bool(payload.get("enabled", False)),
+            window_seconds=int(payload.get("window_seconds", 300)),
+            cooldown_seconds=int(payload.get("cooldown_seconds", 600)),
+            fail_open_threshold=int(payload.get("fail_open_threshold", 3)),
+            overflow_risk_threshold=int(payload.get("overflow_risk_threshold", 3)),
+            no_op_above_threshold_threshold=int(
+                payload.get("no_op_above_threshold_threshold", 5)
+            ),
+        )
+
+
 @dataclass
 class CompactionConfig:
     """Configuration for context compaction feature.
@@ -67,6 +94,9 @@ class CompactionConfig:
         "because a newer result for this resource exists later in the conversation."
     )
 
+    # Telemetry / alerting
+    alerts: CompactionAlertsConfig = field(default_factory=CompactionAlertsConfig)
+
     def is_tool_category_allowed(self, category: ToolCategory) -> bool:
         """Check if a tool category is eligible for compaction.
 
@@ -97,6 +127,7 @@ def is_tool_category_allowed(self, category: ToolCategory) -> bool:
     @classmethod
     def from_dict(cls, data: dict[str, Any]) -> "CompactionConfig":
         """Create configuration from a dictionary."""
+        alerts = CompactionAlertsConfig.from_dict(data.get("alerts", {}))
         return cls(
             enabled=data.get("enabled", False),
             token_threshold=data.get("token_threshold", 100_000),
@@ -110,12 +141,13 @@ def from_dict(cls, data: dict[str, Any]) -> "CompactionConfig":
             preserve_last_n_results=data.get("preserve_last_n_results", 1),
             stub_template=data.get("stub_template", cls.stub_template),
             redact_resource_identifiers=data.get("redact_resource_identifiers", False),
+            alerts=alerts,
         )
 
     @classmethod
     def disabled(cls) -> "CompactionConfig":
         """Create a disabled configuration."""
-        return cls(enabled=False)
+        return cls(enabled=False, alerts=CompactionAlertsConfig(enabled=False))
 
     @classmethod
     def default(cls) -> "CompactionConfig":
@@ -138,6 +170,7 @@ def default(cls) -> "CompactionConfig":
             ],
             max_stubs_per_resource=1,
             preserve_last_n_results=1,
+            alerts=CompactionAlertsConfig(enabled=False),
         )
 
 

src/core/interfaces/history_compaction_interface.py

@@ -17,6 +17,12 @@
 from pydantic import BaseModel, Field
 
 from src.core.domain.chat import ChatMessage
+from src.core.domain.compaction_telemetry import (
+    CompactionAggregateMetrics,
+    CompactionAlertRecord,
+    CompactionEventRecord,
+    EffectiveCompactionConfigDiagnostics,
+)
 from src.core.domain.configuration.compaction_config import (
     CompactionConfig,
     CompactionPolicies,
@@ -104,6 +110,10 @@ class CompactionResult:
     original_message_count: int = 0
     stale_resources: set[str] = field(default_factory=set)
     error: str | None = None
+    event_records: list[CompactionEventRecord] = field(default_factory=list)
+    aggregate_metrics: CompactionAggregateMetrics | None = None
+    alerts: list[CompactionAlertRecord] = field(default_factory=list)
+    effective_config_diagnostics: EffectiveCompactionConfigDiagnostics | None = None
 
     @property
     def was_compacted(self) -> bool: