fix(chat_shell): use model_type for provider detection instead of model_id

chat_shell/chat_shell/compression/compressor.py

@@ -121,7 +121,12 @@ def __init__(
         self.model_id = model_id
         self.config = config or CompressionConfig.from_settings()
         self.model_context = get_model_context_config(model_id, model_config)
-        self.token_counter = TokenCounter(model_id)
+
+        # Extract model_type from model_config for accurate provider detection.
+        # model_type (e.g. "claude", "openai") is the authoritative source
+        # because the same model_id can be served by multiple protocols.
+        model_type = model_config.get("model") if model_config else None
+        self.token_counter = TokenCounter(model_id, model_type=model_type)
 
         # Default strategies in order of application (from least to most disruptive)
         self.strategies = strategies or [

chat_shell/chat_shell/compression/strategies.py

@@ -733,55 +733,20 @@ def compress(
 
         # Build result with truncation notice if any messages were removed
         if messages_to_remove > 0:
-            is_anthropic = token_counter.provider == "anthropic"
-
-            if is_anthropic:
-                # Anthropic requires strict user/assistant alternation and
-                # no non-leading system messages. Adapt the notice role based
-                # on adjacent messages.
-                prev_role = first_messages[-1]["role"] if first_messages else None
-                following = kept_middle if kept_middle else last_messages
-                next_role = following[0]["role"] if following else None
-
-                if prev_role == next_role and prev_role is not None:
-                    # Odd removal broke alternation (e.g., user→[removed]→user).
-                    # Insert notice with opposite role to bridge the gap.
-                    notice_role = "assistant" if prev_role == "user" else "user"
-                    truncation_notice_msg = {
-                        "role": notice_role,
-                        "content": self.TRUNCATION_NOTICE,
-                    }
-                    result = (
-                        system_messages
-                        + first_messages
-                        + [truncation_notice_msg]
-                        + kept_middle
-                        + last_messages
-                    )
-                elif following:
-                    # Alternation is intact. Merge notice into the next message
-                    # to avoid inserting a message that breaks alternation.
-                    orig_content = following[0]["content"]
-                    if isinstance(orig_content, list):
-                        # Content is a list of blocks (e.g., user message with
-                        # time block). Prepend notice as a new text block.
-                        merged = [
-                            {"type": "text", "text": self.TRUNCATION_NOTICE}
-                        ] + orig_content
-                    else:
-                        merged = self.TRUNCATION_NOTICE + "\n\n" + orig_content
-                    following[0] = {**following[0], "content": merged}
-                    result = (
-                        system_messages + first_messages + kept_middle + last_messages
-                    )
-                else:
-                    # No messages after the truncation point
-                    result = system_messages + first_messages
-            else:
-                # Non-Anthropic models (OpenAI, Google, etc.) support
-                # system messages anywhere, so keep original behavior.
+            # Use alternation-safe approach for ALL providers.
+            # Never insert a system-role notice in the middle — Anthropic
+            # rejects non-consecutive system messages, and provider detection
+            # may be inaccurate for other providers too.
+            prev_role = first_messages[-1]["role"] if first_messages else None
+            following = kept_middle if kept_middle else last_messages
+            next_role = following[0]["role"] if following else None
+
+            if prev_role == next_role and prev_role is not None:
+                # Odd removal broke alternation (e.g., user→[removed]→user).
+                # Insert notice with opposite role to bridge the gap.
+                notice_role = "assistant" if prev_role == "user" else "user"
                 truncation_notice_msg = {
-                    "role": "system",
+                    "role": notice_role,
                     "content": self.TRUNCATION_NOTICE,
                 }
                 result = (
@@ -791,6 +756,23 @@ def compress(
                     + kept_middle
                     + last_messages
                 )
+            elif following:
+                # Alternation is intact. Merge notice into the next message
+                # to avoid inserting a message that breaks alternation.
+                orig_content = following[0]["content"]
+                if isinstance(orig_content, list):
+                    # Content is a list of blocks (e.g., user message with
+                    # time block). Prepend notice as a new text block.
+                    merged = [
+                        {"type": "text", "text": self.TRUNCATION_NOTICE}
+                    ] + orig_content
+                else:
+                    merged = self.TRUNCATION_NOTICE + "\n\n" + orig_content
+                following[0] = {**following[0], "content": merged}
+                result = system_messages + first_messages + kept_middle + last_messages
+            else:
+                # No messages after the truncation point
+                result = system_messages + first_messages
         else:
             result = messages
 

chat_shell/chat_shell/compression/token_counter.py

@@ -24,6 +24,8 @@
 
 import tiktoken
 
+from chat_shell.models.providers import detect_provider
+
 logger = logging.getLogger(__name__)
 
 # Cache for tiktoken encodings
@@ -139,33 +141,6 @@ def _count_tokens_for_messages(model_id: str, messages: list[dict[str, Any]]) ->
     return total_tokens
 
 
-def _detect_provider(model_id: str) -> str:
-    """Detect the provider based on model identifier.
-
-    Args:
-        model_id: Model identifier string
-
-    Returns:
-        Provider name: "openai", "anthropic", "google", or "unknown"
-    """
-    model_id_lower = model_id.lower()
-
-    # Anthropic models
-    if any(prefix in model_id_lower for prefix in ["claude", "anthropic"]):
-        return "anthropic"
-
-    # Google models
-    if any(prefix in model_id_lower for prefix in ["gemini", "palm", "bison"]):
-        return "google"
-
-    # OpenAI models (most gpt-*, o1*, o3* patterns)
-    if any(prefix in model_id_lower for prefix in ["gpt-", "o1", "o3"]):
-        return "openai"
-
-    # Default to unknown if no pattern matches
-    return "unknown"
-
-
 class TokenCounter:
     """Token counter for various model providers using tiktoken.
 
@@ -179,14 +154,22 @@ class TokenCounter:
         token_count = counter.count_messages(messages)
     """
 
-    def __init__(self, model_name: str | None = None, model_id: str | None = None):
+    def __init__(
+        self,
+        model_name: str | None = None,
+        model_id: str | None = None,
+        model_type: str | None = None,
+    ):
         """Initialize token counter.
 
         Args:
             model_name: Model identifier for provider detection (preferred)
             model_id: Deprecated alias for model_name (for backward compatibility)
+            model_type: Model protocol type (e.g. "claude", "openai", "gemini").
+                This is the authoritative source for provider detection.
         """
         self.model_id = model_id or model_name or "gpt-4"
+        self._model_type = model_type
         self._encoding: tiktoken.Encoding | None = None
 
     @property
@@ -201,9 +184,12 @@ def provider(self) -> str:
         """Get the detected provider for the model.
 
         Returns:
-            Provider name: "openai", "anthropic", "google", or "unknown"
+            Provider name: "openai", "anthropic", "google".
+
+        Raises:
+            ValueError: If model_type was not provided and cannot be resolved.
         """
-        return _detect_provider(self.model_id)
+        return detect_provider(self._model_type or self.model_id)
 
     def count_text(self, text: str) -> int:
         """Count tokens in a text string.

chat_shell/chat_shell/models/factory.py

@@ -23,26 +23,10 @@
 from shared.telemetry.decorators import add_span_event, trace_sync
 
 from .openai_reasoning import ChatOpenAIWithReasoning
+from .providers import PROVIDER_ALIASES, detect_provider
 
 logger = logging.getLogger(__name__)
 
-# Provider detection: (prefixes, provider_name)
-_PROVIDER_PATTERNS = [
-    (("gpt-", "o1-", "o3-", "chatgpt-"), "openai"),
-    (("claude-",), "anthropic"),
-    (("gemini-",), "google"),
-]
-
-# Provider type aliases
-_PROVIDER_ALIASES = {
-    "openai": "openai",
-    "gpt": "openai",
-    "anthropic": "anthropic",
-    "claude": "anthropic",
-    "google": "google",
-    "gemini": "google",
-}
-
 # Allowed think_config keys per provider, mapped directly to constructor params.
 # NOTE: Only keys that are safe as direct constructor params belong here.
 # For openai, "reasoning" (dict) is NOT whitelisted because setting it as a
@@ -85,22 +69,19 @@ def _extract_think_params(provider: str, think_config: dict | None) -> dict:
 
 
 def _detect_provider(model_type: str, model_id: str) -> str:
-    """Detect provider from model type or model ID."""
-    # Check model_type alias first
-    if provider := _PROVIDER_ALIASES.get(model_type.lower()):
-        return provider
-
-    # Fall back to model_id prefix detection
-    model_lower = model_id.lower()
-    for prefixes, provider in _PROVIDER_PATTERNS:
-        if any(model_lower.startswith(p.lower()) for p in prefixes):
-            return provider
-
-    # Default to OpenAI for unknown models (common for OpenAI-compatible APIs)
-    logger.warning(
-        "Unknown provider for %s/%s, defaulting to OpenAI", model_type, model_id
-    )
-    return "openai"
+    """Detect provider from model type.
+
+    Delegates to the shared :func:`detect_provider`.  Falls back to
+    ``"openai"`` when ``model_type`` is not recognized (common for
+    OpenAI-compatible APIs).
+    """
+    try:
+        return detect_provider(model_type)
+    except ValueError:
+        logger.warning(
+            "Unknown provider for %s/%s, defaulting to OpenAI", model_type, model_id
+        )
+        return "openai"
 
 
 def _mask_api_key(api_key: str) -> str:
@@ -339,7 +320,9 @@ def create_from_name(
         return cls.create_from_config(
             {
                 "model_id": model_name,
-                "model": _detect_provider("", model_name),
+                "model": PROVIDER_ALIASES.get(
+                    model_name.split("-")[0].lower(), "openai"
+                ),
                 "api_key": api_key,
                 "base_url": base_url or "",
             },
@@ -350,17 +333,18 @@ def create_from_name(
     def get_provider(model_id: str) -> str | None:
         """Get provider name for a model ID.
 
+        Uses the model_id prefix (e.g. ``"gpt-"`` -> ``"openai"``) as a
+        convenience lookup.  For accurate detection, prefer
+        :func:`detect_provider` with ``model_type``.
+
         Args:
             model_id: Model identifier
 
         Returns:
             Provider name ("openai", "anthropic", "google") or None if unknown
         """
-        model_lower = model_id.lower()
-        for prefixes, provider in _PROVIDER_PATTERNS:
-            if any(model_lower.startswith(p.lower()) for p in prefixes):
-                return provider
-        return None
+        prefix = model_id.split("-")[0].lower()
+        return PROVIDER_ALIASES.get(prefix)
 
     @classmethod
     def is_supported(cls, model_id: str) -> bool:

chat_shell/chat_shell/models/providers.py

@@ -0,0 +1,53 @@
+# SPDX-FileCopyrightText: 2025 Weibo, Inc.
+#
+# SPDX-License-Identifier: Apache-2.0
+
+"""Provider detection utilities shared across the chat_shell package.
+
+The ``model_type`` (e.g. ``"claude"``, ``"openai"``, ``"gemini"``) is the
+authoritative source for provider detection because the same ``model_id``
+can be served by multiple protocols (e.g. ``thudm-glm-5`` may support both
+OpenAI and Anthropic protocols).
+
+Modules that need provider detection should call :func:`detect_provider`
+with the ``model_type`` value from the model configuration.
+"""
+
+# Mapping from model_type values (env.model) to canonical provider names.
+# This is the authoritative mapping used by both the model factory and the
+# compression pipeline.
+PROVIDER_ALIASES: dict[str, str] = {
+    "openai": "openai",
+    "gpt": "openai",
+    "o1": "openai",
+    "o3": "openai",
+    "chatgpt": "openai",
+    "anthropic": "anthropic",
+    "claude": "anthropic",
+    "google": "google",
+    "gemini": "google",
+}
+
+
+def detect_provider(model_type: str) -> str:
+    """Detect the provider from model type.
+
+    Args:
+        model_type: The protocol / model type string (e.g. ``"claude"``,
+            ``"openai"``, ``"gemini"``).  This is the authoritative source
+            from the model configuration's ``model`` field.
+
+    Returns:
+        Canonical provider name: ``"openai"``, ``"anthropic"``, or ``"google"``.
+
+    Raises:
+        ValueError: If ``model_type`` is not recognized.
+    """
+    provider = PROVIDER_ALIASES.get(model_type.lower())
+    if provider:
+        return provider
+
+    raise ValueError(
+        f"Unknown model_type '{model_type}'. "
+        f"Supported values: {', '.join(sorted(PROVIDER_ALIASES.keys()))}"
+    )

chat_shell/tests/test_compression.py

@@ -734,8 +734,9 @@ def test_truncation_notice_uses_system_role_for_openai(self):
         ]
         assert len(truncation_notices) == 1
 
-        # OpenAI should keep role="system" (original behavior)
-        assert truncation_notices[0]["role"] == "system"
+        # Non-Anthropic should also use alternation-safe roles (not system)
+        # to avoid issues when provider detection is inaccurate
+        assert truncation_notices[0]["role"] in ("user", "assistant")
 
     def test_no_truncation_for_short_history(self):
         """Test that short history is not truncated."""

chat_shell/tests/test_model_factory.py

@@ -147,15 +147,11 @@ def test_model_type_alias(self):
         assert _detect_provider("google", "anything") == "google"
         assert _detect_provider("gemini", "anything") == "google"
 
-    def test_model_id_prefix_detection(self):
+    def test_unknown_model_type_defaults_to_openai(self):
+        """Unknown model_type always defaults to openai (no model_id fallback)."""
         assert _detect_provider("unknown", "gpt-4o") == "openai"
-        assert _detect_provider("unknown", "o1-preview") == "openai"
-        assert _detect_provider("unknown", "o3-mini") == "openai"
-        assert _detect_provider("unknown", "claude-3-sonnet") == "anthropic"
-        assert _detect_provider("unknown", "gemini-1.5-pro") == "google"
-
-    def test_unknown_defaults_to_openai(self):
-        """Unknown provider defaults to openai for OpenAI-compatible APIs."""
+        assert _detect_provider("unknown", "claude-3-sonnet") == "openai"
+        assert _detect_provider("unknown", "gemini-1.5-pro") == "openai"
         assert _detect_provider("unknown", "kimi-k2.5") == "openai"
         assert _detect_provider("unknown", "deepseek-chat") == "openai"