feat(provider): add pipeline reasoning controls (#2373)

* feat(provider): add pipeline reasoning controls

* fix(provider): preserve local agent model compatibility

* refactor(web): use shadcn reasoning slider

* fix(runtime): stabilize reasoning chat delivery

* fix(provider): route reasoning controls by model family

* fix(provider): handle hosted Kimi reasoning protocols

* fix(provider): map qwen reasoning levels to budgets

* fix(provider): preserve think tags in streamed reasoning

* fix(provider): preserve reasoning tool metadata

* style(provider): satisfy ruff checks after merge

* fix(persistence): preserve reasoning migration compatibility

src/langbot/pkg/api/http/service/model.py

@@ -10,6 +10,7 @@ from ....core import app
 from ....entity.persistence import model as persistence_model
 from ....entity.persistence import pipeline as persistence_pipeline
 from ....provider.modelmgr import requester as model_requester
+from ....provider.modelmgr import reasoning as model_reasoning
 from ....workspace.errors import WorkspaceNotFoundError
 from .secrets import mask_secret_value, redact_secrets, restore_secret_placeholders
 from .tenant import TenantContext, require_workspace_uuid, scope_statement
@@ -55,6 +56,53 @@ def _redact_model_secrets(model_data: dict) -> dict:
     return redacted
 
 
+def _normalize_llm_reasoning(model_data: dict) -> None:
+    model_data['reasoning_config'] = model_reasoning.validate_reasoning_config(
+        model_data.get('reasoning_config'),
+        model_data.get('abilities'),
+        model_data.get('extra_args'),
+    )
+
+
+def _validate_llm_reasoning_capability(
+    model_entity: persistence_model.LLMModel,
+    runtime_provider: model_requester.RuntimeProvider,
+) -> None:
+    config = model_reasoning.normalize_reasoning_config(model_entity.reasoning_config)
+    if config['level'] == 'provider_default':
+        return
+
+    runtime_model = model_requester.RuntimeLLMModel(
+        execution_context=runtime_provider.execution_context,
+        model_entity=model_entity,
+        provider=runtime_provider,
+    )
+    capabilities = runtime_provider.requester.get_reasoning_capabilities(runtime_model)
+    model_reasoning.validate_reasoning_capabilities(config, capabilities, model_entity.name)
+
+
+def _reasoning_capabilities(ap: app.Application, model: persistence_model.LLMModel) -> dict:
+    model_mgr = getattr(ap, 'model_mgr', None)
+    runtime_models = getattr(model_mgr, 'llm_model_dict', {}) if model_mgr is not None else {}
+    for runtime_model in runtime_models.values():
+        if (
+            runtime_model.model_entity.uuid == model.uuid
+            and runtime_model.model_entity.workspace_uuid == model.workspace_uuid
+        ):
+            return runtime_model.provider.requester.get_reasoning_capabilities(runtime_model)
+    return model_reasoning.default_reasoning_capabilities(
+        supported='reasoning' in (model.abilities or []),
+        source='manual' if 'reasoning' in (model.abilities or []) else 'unknown',
+    )
+
+
+def _serialize_llm_model(ap: app.Application, model: persistence_model.LLMModel) -> dict:
+    model_dict = ap.persistence_mgr.serialize_model(persistence_model.LLMModel, model)
+    model_dict['reasoning_config'] = model_reasoning.normalize_reasoning_config(model_dict.get('reasoning_config'))
+    model_dict['reasoning_capabilities'] = _reasoning_capabilities(ap, model)
+    return model_dict
+
+
 async def _validate_provider_supports(
     ap: app.Application,
     context: TenantContext,
@@ -165,7 +213,7 @@ class LLMModelsService:
 
         models_list = []
         for model in models:
-            model_dict = self.ap.persistence_mgr.serialize_model(persistence_model.LLMModel, model)
+            model_dict = _serialize_llm_model(self.ap, model)
             provider = providers.get(model.provider_uuid)
             if provider:
                 provider_dict = self.ap.persistence_mgr.serialize_model(persistence_model.ModelProvider, provider)
@@ -196,7 +244,7 @@ class LLMModelsService:
             )
         )
         models = result.all()
-        serialized = [self.ap.persistence_mgr.serialize_model(persistence_model.LLMModel, m) for m in models]
+        serialized = [_serialize_llm_model(self.ap, model) for model in models]
         return serialized if include_secret else [_redact_model_secrets(model) for model in serialized]
 
     async def create_llm_model(
@@ -233,13 +281,17 @@ class LLMModelsService:
         await _require_workspace_provider(self.ap, context, model_data['provider_uuid'])
         await _assert_cloud_managed_provider_mutable(self.ap, context, model_data['provider_uuid'])
         await _validate_provider_supports(self.ap, context, model_data['provider_uuid'], 'llm')
+        _normalize_llm_reasoning(model_data)
+
+        runtime_provider = await _require_runtime_provider(self.ap, context, model_data['provider_uuid'])
+        model_entity = persistence_model.LLMModel(**model_data)
+        _validate_llm_reasoning_capability(model_entity, runtime_provider)
 
         await self.ap.persistence_mgr.execute_async(sqlalchemy.insert(persistence_model.LLMModel).values(**model_data))
 
-        runtime_provider = await _require_runtime_provider(self.ap, context, model_data['provider_uuid'])
         runtime_llm_model = await self.ap.model_mgr.load_llm_model_with_provider(
             context,
-            persistence_model.LLMModel(**model_data),
+            model_entity,
             runtime_provider,
         )
         await self.ap.model_mgr.cache_llm_model(context, runtime_llm_model)
@@ -287,7 +339,7 @@ class LLMModelsService:
         if model is None:
             return None
 
-        model_dict = self.ap.persistence_mgr.serialize_model(persistence_model.LLMModel, model)
+        model_dict = _serialize_llm_model(self.ap, model)
 
         # Get provider
         provider_result = await self.ap.persistence_mgr.execute_async(
@@ -349,6 +401,18 @@ class LLMModelsService:
         await _assert_cloud_managed_provider_mutable(self.ap, context, provider_uuid)
         await _validate_provider_supports(self.ap, context, provider_uuid, 'llm')
 
+        merged_model_data = {
+            key: value
+            for key, value in {**existing_model, **model_data, 'provider_uuid': provider_uuid}.items()
+            if key not in {'provider', 'created_at', 'updated_at', 'reasoning_capabilities'}
+        }
+        _normalize_llm_reasoning(merged_model_data)
+        model_data['reasoning_config'] = merged_model_data['reasoning_config']
+
+        runtime_provider = await _require_runtime_provider(self.ap, context, provider_uuid)
+        model_entity = persistence_model.LLMModel(**_runtime_model_data(model_uuid, merged_model_data))
+        _validate_llm_reasoning_capability(model_entity, runtime_provider)
+
         result = await self.ap.persistence_mgr.execute_async(
             scope_statement(
                 sqlalchemy.update(persistence_model.LLMModel)
@@ -362,19 +426,9 @@ class LLMModelsService:
             raise WorkspaceNotFoundError('Model not found')
 
         await self.ap.model_mgr.remove_llm_model(context, model_uuid)
-        runtime_provider = await _require_runtime_provider(self.ap, context, provider_uuid)
         runtime_llm_model = await self.ap.model_mgr.load_llm_model_with_provider(
             context,
-            persistence_model.LLMModel(
-                **_runtime_model_data(
-                    model_uuid,
-                    {
-                        key: value
-                        for key, value in {**existing_model, **model_data, 'provider_uuid': provider_uuid}.items()
-                        if key not in {'provider', 'created_at', 'updated_at'}
-                    },
-                )
-            ),
+            model_entity,
             runtime_provider,
         )
         await self.ap.model_mgr.cache_llm_model(context, runtime_llm_model)
@@ -407,6 +461,7 @@ class LLMModelsService:
                 raise WorkspaceNotFoundError('Model not found')
             runtime_llm_model = await self.ap.model_mgr.get_model_by_uuid(context, model_uuid)
         else:
+            _normalize_llm_reasoning(model_data)
             runtime_llm_model = await self.ap.model_mgr.init_temporary_runtime_llm_model(context, model_data)
 
         extra_args = model_data.get('extra_args', {})

src/langbot/pkg/entity/persistence/model.py

@@ -48,6 +48,12 @@ class LLMModel(Base):
     provider_uuid = sqlalchemy.Column(sqlalchemy.String(255), nullable=False)
     abilities = sqlalchemy.Column(sqlalchemy.JSON, nullable=False, default=[])
     context_length = sqlalchemy.Column(sqlalchemy.Integer, nullable=True)
+    reasoning_config = sqlalchemy.Column(
+        sqlalchemy.JSON,
+        nullable=False,
+        default=lambda: {'level': 'provider_default'},
+        server_default=sqlalchemy.text('\'{"level":"provider_default"}\''),
+    )
     extra_args = sqlalchemy.Column(sqlalchemy.JSON, nullable=False, default={})
     prefered_ranking = sqlalchemy.Column(sqlalchemy.Integer, nullable=False, default=0)
     created_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=False, server_default=sqlalchemy.func.now())

src/langbot/pkg/persistence/alembic/versions/0018_add_llm_reasoning_config.py

@@ -0,0 +1,57 @@
+"""add llm reasoning config
+
+Revision ID: 0018_llm_reasoning_config
+Revises: 0017_oss_workspace_identity
+Create Date: 2026-07-27
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = '0018_llm_reasoning_config'
+down_revision = '0017_oss_workspace_identity'
+branch_labels = None
+depends_on = None
+
+
+_LLM_MODELS = sa.table(
+    'llm_models',
+    sa.column('reasoning_config', sa.JSON()),
+)
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    inspector = sa.inspect(conn)
+    if 'llm_models' not in inspector.get_table_names():
+        return
+
+    columns = {column['name'] for column in inspector.get_columns('llm_models')}
+    if 'reasoning_config' in columns:
+        return
+
+    op.add_column(
+        'llm_models',
+        sa.Column(
+            'reasoning_config',
+            sa.JSON(),
+            nullable=True,
+            server_default=sa.text('\'{"level":"provider_default"}\''),
+        ),
+    )
+    conn.execute(_LLM_MODELS.update().values(reasoning_config={'level': 'provider_default'}))
+    with op.batch_alter_table('llm_models') as batch_op:
+        batch_op.alter_column('reasoning_config', existing_type=sa.JSON(), nullable=False)
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+    inspector = sa.inspect(conn)
+    if 'llm_models' not in inspector.get_table_names():
+        return
+    columns = {column['name'] for column in inspector.get_columns('llm_models')}
+    if 'reasoning_config' in columns:
+        with op.batch_alter_table('llm_models') as batch_op:
+            batch_op.drop_column('reasoning_config')

src/langbot/pkg/persistence/alembic/versions/0021_merge_reasoning_config.py

@@ -0,0 +1,21 @@
+"""merge reasoning config with the main migration branch
+
+Revision ID: 0021_merge_reasoning_config
+Revises: 0020_membership_source, 0018_llm_reasoning_config
+Create Date: 2026-08-09
+"""
+
+from __future__ import annotations
+
+revision = '0021_merge_reasoning_config'
+down_revision = ('0020_membership_source', '0018_llm_reasoning_config')
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    pass
+
+
+def downgrade() -> None:
+    pass

src/langbot/pkg/pipeline/controller.py

@@ -132,7 +132,7 @@ class Controller:
 
                             break
 
-                    if not selected_query:  # 没找到 说明：没有请求 或者 所有query对应的session都已达到并发上限
+                    if not selected_query:  # No query is runnable under the current session limits.
                         await self.ap.query_pool.condition.wait()
                         continue
 

src/langbot/pkg/platform/sources/websocket_adapter.py

@@ -5,6 +5,7 @@ import contextvars
 import logging
 import time
 import typing
+from dataclasses import dataclass
 from datetime import datetime
 
 import pydantic
@@ -25,6 +26,15 @@ _current_pipeline_uuid: contextvars.ContextVar[str | None] = contextvars.Context
 )
 
 
+@dataclass(frozen=True)
+class WebSocketReplyContext:
+    """Trusted routing context retained when the originating socket reconnects."""
+
+    scope: WebSocketScope
+    pipeline_uuid: str
+    session_id: str | None
+
+
 class WebSocketMessage(pydantic.BaseModel):
     """WebSocket消息格式"""
 
@@ -265,6 +275,11 @@ class WebSocketAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
         embed_target = self._parse_embed_target(sender_id)
         if embed_target is not None:
             return embed_target
+        reply_context = getattr(message_source, '_websocket_reply_context', None)
+        if isinstance(reply_context, WebSocketReplyContext):
+            if reply_context.scope != self._scope():
+                raise ValueError('WebSocket reply context does not match this adapter scope')
+            return reply_context.pipeline_uuid, reply_context.session_id
         raise ValueError('WebSocket reply target is not bound to this adapter scope')
 
     async def send_message(
@@ -685,6 +700,16 @@ class WebSocketAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
 
         # 异步触发事件处理
         # Use owner_bot's listeners if available, otherwise fall back to proxy bot
+        object.__setattr__(
+            event,
+            '_websocket_reply_context',
+            WebSocketReplyContext(
+                scope=connection.scope,
+                pipeline_uuid=pipeline_uuid,
+                session_id=connection.session_id,
+            ),
+        )
+
         listeners = (
             owner_bot.adapter.listeners
             if (owner_bot and hasattr(owner_bot.adapter, 'listeners') and owner_bot.adapter.listeners)

src/langbot/pkg/provider/modelmgr/modelmgr.py

@@ -649,6 +649,7 @@ class ModelManager:
             provider_uuid=runtime_provider.provider_entity.uuid,
             abilities=model_info.get('abilities', []),
             context_length=model_info.get('context_length'),
+            reasoning_config=model_info.get('reasoning_config', {'level': 'provider_default'}),
             extra_args=model_info.get('extra_args', {}),
         )
         return self._build_llm_model(execution_context, model_entity, runtime_provider)
@@ -717,7 +718,10 @@ class ModelManager:
         provider_entity = self._coerce_provider(provider_info, context)
         requester_manifest = self.get_available_requester_manifest_by_name(provider_entity.requester)
         litellm_provider = self._get_litellm_provider_from_manifest(requester_manifest)
-        config = {'base_url': provider_entity.base_url}
+        config = {
+            'base_url': provider_entity.base_url,
+            'requester_name': provider_entity.requester,
+        }
 
         if litellm_provider:
             from .requesters import litellmchat

src/langbot/pkg/provider/modelmgr/reasoning.py

@@ -0,0 +1,125 @@
+from __future__ import annotations
+
+import typing
+
+
+ReasoningLevel = typing.Literal[
+    'provider_default',
+    'disabled',
+    'enabled',
+    'minimal',
+    'low',
+    'medium',
+    'high',
+    'xhigh',
+    'max',
+]
+
+REASONING_LEVELS: tuple[str, ...] = (
+    'provider_default',
+    'disabled',
+    'enabled',
+    'minimal',
+    'low',
+    'medium',
+    'high',
+    'xhigh',
+    'max',
+)
+DEFAULT_REASONING_CONFIG: dict[str, str] = {'level': 'provider_default'}
+
+_CONFLICTING_TOP_LEVEL_ARGS = {
+    'reasoning_effort',
+    'thinking',
+    'enable_thinking',
+    'thinking_budget',
+    'reasoning',
+}
+_CONFLICTING_EXTRA_BODY_ARGS = {
+    'reasoning_effort',
+    'thinking',
+    'enable_thinking',
+    'thinking_budget',
+    'reasoning',
+}
+
+
+def normalize_reasoning_config(value: typing.Any) -> dict[str, str]:
+    """Return the canonical model reasoning configuration."""
+    if value is None:
+        return dict(DEFAULT_REASONING_CONFIG)
+    if not isinstance(value, dict):
+        raise ValueError('reasoning_config must be an object')
+
+    unknown_fields = set(value) - {'level'}
+    if unknown_fields:
+        raise ValueError(f'Unsupported reasoning_config fields: {", ".join(sorted(unknown_fields))}')
+
+    level = value.get('level', 'provider_default')
+    if level not in REASONING_LEVELS:
+        raise ValueError(f'Unsupported reasoning level: {level}')
+    return {'level': typing.cast(str, level)}
+
+
+def validate_reasoning_config(
+    value: typing.Any,
+    abilities: typing.Iterable[str] | None,
+    extra_args: typing.Any,
+) -> dict[str, str]:
+    """Validate a model-facing reasoning config and conflicting raw arguments."""
+    config = normalize_reasoning_config(value)
+    if config['level'] == 'provider_default':
+        return config
+
+    if 'reasoning' not in set(abilities or []):
+        raise ValueError('The reasoning ability must be enabled before selecting a reasoning level')
+
+    conflicts = find_reasoning_arg_conflicts(extra_args)
+    if conflicts:
+        raise ValueError('reasoning_config conflicts with advanced parameters: ' + ', '.join(conflicts))
+    return config
+
+
+def find_reasoning_arg_conflicts(extra_args: typing.Any) -> list[str]:
+    if not isinstance(extra_args, dict):
+        return []
+
+    conflicts = [key for key in sorted(_CONFLICTING_TOP_LEVEL_ARGS) if key in extra_args]
+    extra_body = extra_args.get('extra_body')
+    if isinstance(extra_body, dict):
+        conflicts.extend(f'extra_body.{key}' for key in sorted(_CONFLICTING_EXTRA_BODY_ARGS) if key in extra_body)
+    return conflicts
+
+
+def validate_reasoning_capabilities(
+    config: typing.Any,
+    capabilities: typing.Mapping[str, typing.Any],
+    model_name: str,
+) -> None:
+    """Ensure an explicit reasoning level can be honored by the requester."""
+    level = normalize_reasoning_config(config)['level']
+    if level == 'provider_default':
+        return
+
+    available_levels = capabilities.get('levels')
+    if not isinstance(available_levels, list):
+        available_levels = []
+    legacy_levels = capabilities.get('legacy_levels')
+    if not isinstance(legacy_levels, list):
+        legacy_levels = []
+    if capabilities.get('supported') is not True or (level not in available_levels and level not in legacy_levels):
+        available_text = ', '.join(str(item) for item in available_levels) or 'provider_default'
+        raise ValueError(
+            f'Reasoning level "{level}" is not supported by model {model_name}. Available levels: {available_text}'
+        )
+
+
+def default_reasoning_capabilities(
+    supported: bool = False,
+    source: str = 'unknown',
+) -> dict[str, typing.Any]:
+    return {
+        'supported': supported,
+        'levels': ['provider_default'],
+        'source': source,
+    }

src/langbot/pkg/provider/modelmgr/requester.py

@@ -10,6 +10,7 @@ from ...entity.persistence import model as persistence_model
 from ...workspace.errors import WorkspaceInvariantError
 import langbot_plugin.api.entities.builtin.resource.tool as resource_tool
 from . import token
+from . import reasoning
 import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query
 import langbot_plugin.api.entities.builtin.provider.message as provider_message
 
@@ -377,11 +378,15 @@ class RuntimeLLMModel:
     provider: RuntimeProvider
     """提供商实例"""
 
+    reasoning_config_override: dict[str, str] | None
+    """Request-scoped reasoning policy supplied by the active pipeline."""
+
     def __init__(
         self,
         execution_context: ExecutionContext,
         model_entity: persistence_model.LLMModel,
         provider: RuntimeProvider,
+        reasoning_config_override: dict[str, str] | None = None,
     ):
         _ensure_same_execution_scope(provider.execution_context, execution_context, resource='LLM model')
         if model_entity.workspace_uuid != execution_context.workspace_uuid:
@@ -391,6 +396,7 @@ class RuntimeLLMModel:
         self.execution_context = execution_context
         self.model_entity = model_entity
         self.provider = provider
+        self.reasoning_config_override = reasoning_config_override
 
 
 class RuntimeEmbeddingModel:
@@ -482,6 +488,13 @@ class ProviderAPIRequester(metaclass=abc.ABCMeta):
         """
         raise NotImplementedError('This provider does not support model scanning')
 
+    def get_reasoning_capabilities(self, model: RuntimeLLMModel) -> dict[str, typing.Any]:
+        """Return normalized reasoning controls supported by a model."""
+        return reasoning.default_reasoning_capabilities(
+            supported='reasoning' in (model.model_entity.abilities or []),
+            source='manual' if 'reasoning' in (model.model_entity.abilities or []) else 'unknown',
+        )
+
     @abc.abstractmethod
     async def invoke_llm(
         self,

src/langbot/pkg/provider/modelmgr/requesters/litellmchat.py

@@ -7,7 +7,7 @@ import typing
 import litellm
 from litellm import acompletion, aembedding, arerank
 
-from .. import errors, requester
+from .. import errors, reasoning, requester
 from ....utils import httpclient
 import langbot_plugin.api.entities.builtin.resource.tool as resource_tool
 import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query
@@ -164,6 +164,39 @@ class LiteLLMRequester(requester.ProviderAPIRequester):
 
     _EMBEDDING_MODEL_HINTS = ('embedding', 'embed', 'bge-', 'e5-', 'm3e', 'gte-', 'text-embedding')
     _RERANK_MODEL_HINTS = ('rerank', 're-rank', 're_rank')
+    _QWEN_DEDICATED_THINKING_MODELS = frozenset(
+        {
+            'qwen3.7-max-preview',
+            'qwen3.7-max-2026-05-17',
+        }
+    )
+    _QWEN_REASONING_BUDGETS = {
+        'low': 1024,
+        'medium': 4096,
+        'high': 8192,
+    }
+    _INFERRED_EFFORT_PROVIDERS = frozenset(
+        {
+            'anthropic',
+            'gemini',
+            'groq',
+            'mistral',
+            'openai',
+            'openrouter',
+            'together_ai',
+            'xai',
+        }
+    )
+    _REQUESTER_REASONING_FAMILIES = {
+        'openai-chat-completions': 'openai',
+        'anthropic-messages': 'anthropic',
+        'deepseek-chat-completions': 'deepseek',
+        'moonshot-chat-completions': 'kimi',
+        'moonshot-cn-chat-completions': 'kimi',
+        'bailian-chat-completions': 'qwen',
+        'doubao-chat-completions': 'doubao',
+        'mimo-chat-completions': 'mimo',
+    }
 
     default_config: dict[str, typing.Any] = {
         'base_url': '',
@@ -172,6 +205,7 @@ class LiteLLMRequester(requester.ProviderAPIRequester):
         'drop_params': False,
         'num_retries': 0,
         'api_version': '',
+        'requester_name': '',
     }
 
     async def initialize(self):
@@ -201,7 +235,10 @@ class LiteLLMRequester(requester.ProviderAPIRequester):
             return False
 
         provider = self._get_custom_llm_provider()
-        candidates: list[tuple[str, str | None]] = [(model_name, provider)]
+        candidates: list[tuple[str, str | None]] = [
+            (candidate, None) for candidate in self._metadata_model_candidates(model_name)
+        ]
+        candidates.append((model_name, provider))
         litellm_model_name = self._build_litellm_model_name(model_name)
         if litellm_model_name != model_name:
             candidates.append((litellm_model_name, None))
@@ -268,6 +305,14 @@ class LiteLLMRequester(requester.ProviderAPIRequester):
                 deduped_candidates.append(candidate)
         return deduped_candidates
 
+    @staticmethod
+    def _metadata_model_candidates(model_name: str) -> list[str]:
+        """Return known equivalent model IDs used only for LiteLLM metadata lookup."""
+        normalized_model_name = (model_name or '').lower()
+        if normalized_model_name.startswith('mimo-v2.5'):
+            return [f'openrouter/xiaomi/{normalized_model_name}']
+        return []
+
     def _known_context_length_fallback(self, model_name: str) -> int | None:
         normalized_model_name = (model_name or '').lower()
         if normalized_model_name.startswith('deepseek-v4-'):
@@ -287,7 +332,8 @@ class LiteLLMRequester(requester.ProviderAPIRequester):
         if not callable(helper):
             return self._known_context_length_fallback(model_name)
 
-        candidates = [model_name]
+        candidates = self._metadata_model_candidates(model_name)
+        candidates.append(model_name)
         litellm_model_name = self._build_litellm_model_name(model_name)
         if litellm_model_name != model_name:
             candidates.append(litellm_model_name)
@@ -314,6 +360,297 @@ class LiteLLMRequester(requester.ProviderAPIRequester):
     def _supports_vision(self, model_name: str) -> bool:
         return self._safe_litellm_bool_helper('supports_vision', model_name)
 
+    def _supports_reasoning(self, model_name: str) -> bool:
+        return self._safe_litellm_bool_helper('supports_reasoning', model_name)
+
+    def _requester_name(self, model: requester.RuntimeLLMModel | None = None) -> str:
+        if model is not None:
+            provider_entity = getattr(getattr(model, 'provider', None), 'provider_entity', None)
+            name = getattr(provider_entity, 'requester', None)
+            if isinstance(name, str) and name:
+                return name.lower()
+        return str(self.requester_cfg.get('requester_name') or '').lower()
+
+    @staticmethod
+    def _infer_reasoning_family_from_model_name(model_name: str) -> str:
+        normalized_name = (model_name or '').lower()
+        basename = normalized_name.rsplit('/', 1)[-1]
+        if basename.startswith(('gpt-', 'chatgpt-', 'o1', 'o3', 'o4')):
+            return 'openai'
+        if basename.startswith('claude-'):
+            return 'anthropic'
+        if basename.startswith('deepseek-'):
+            return 'deepseek'
+        if basename.startswith(('kimi-', 'moonshot-')):
+            return 'kimi'
+        if basename.startswith(('qwen-', 'qwen3', 'qwq')):
+            return 'qwen'
+        if basename.startswith(('doubao-', 'seed-')):
+            return 'doubao'
+        if basename.startswith('mimo-'):
+            return 'mimo'
+        return ''
+
+    def _reasoning_family(
+        self,
+        model_name: str,
+        model: requester.RuntimeLLMModel | None = None,
+    ) -> str:
+        requester_name = self._requester_name(model)
+        if requester_name in {'new-api-chat-completions', 'volcark-chat-completions'}:
+            inferred_family = self._infer_reasoning_family_from_model_name(model_name)
+            if inferred_family:
+                return inferred_family
+            return 'volcengine' if requester_name == 'volcark-chat-completions' else ''
+
+        # Bailian's compatible endpoint also hosts Kimi models. Keep those
+        # models on Kimi's ``thinking`` protocol instead of Qwen's
+        # ``enable_thinking`` protocol.
+        if requester_name == 'bailian-chat-completions':
+            inferred_family = self._infer_reasoning_family_from_model_name(model_name)
+            if inferred_family == 'kimi':
+                return inferred_family
+
+        requester_family = self._REQUESTER_REASONING_FAMILIES.get(requester_name)
+        if requester_family:
+            return requester_family
+
+        inferred_family = self._infer_reasoning_family_from_model_name(model_name)
+        provider = (self._get_custom_llm_provider() or '').lower()
+        if provider == 'openai':
+            return inferred_family or ('openai' if requester_name in {'', 'openai'} else '')
+        if provider:
+            return provider
+        return inferred_family
+
+    @staticmethod
+    def _is_anthropic_adaptive_model(model_name: str) -> bool:
+        basename = model_name.lower().rsplit('/', 1)[-1]
+        if 'mythos-preview' in basename:
+            return True
+
+        parts = basename.split('-')
+        if len(parts) < 3 or parts[0] != 'claude':
+            return False
+        model_families = {'opus', 'sonnet', 'fable', 'mythos'}
+        if parts[1] in model_families:
+            if parts[2] == '5':
+                return True
+            return len(parts) >= 4 and parts[2] == '4' and parts[3] in {'6', '7', '8'}
+        return parts[1] == '5' and parts[2] in model_families
+
+    @staticmethod
+    def _is_anthropic_always_thinking_model(model_name: str) -> bool:
+        normalized_name = model_name.lower()
+        return any(marker in normalized_name for marker in ('fable-5', 'mythos-5', 'mythos-preview'))
+
+    @staticmethod
+    def _is_dedicated_qwen_thinking_model(model_name: str) -> bool:
+        normalized_name = model_name.lower().rsplit('/', 1)[-1]
+        return (
+            normalized_name in LiteLLMRequester._QWEN_DEDICATED_THINKING_MODELS
+            or normalized_name.startswith('qwq')
+            or '-thinking' in normalized_name
+        )
+
+    @staticmethod
+    def _supports_qwen_thinking_budget(model_name: str) -> bool:
+        """Return whether the documented Qwen3 family supports thinking_budget."""
+        normalized_name = model_name.lower().rsplit('/', 1)[-1]
+        return normalized_name.startswith('qwen3')
+
+    def _known_reasoning_levels(self, model_name: str, family: str) -> list[str] | None:
+        normalized_name = model_name.lower().rsplit('/', 1)[-1]
+
+        if family == 'deepseek' and normalized_name.startswith('deepseek-'):
+            if normalized_name.startswith('deepseek-v4-'):
+                return ['provider_default', 'disabled', 'low', 'high', 'xhigh', 'max']
+            if 'reasoner' in normalized_name or '-r1' in normalized_name:
+                return ['provider_default']
+            return ['provider_default', 'disabled', 'enabled']
+
+        if family == 'kimi':
+            if normalized_name.startswith('kimi-k3'):
+                return ['provider_default', 'low', 'high', 'max']
+            if normalized_name.startswith('kimi-k2.7-code'):
+                return ['provider_default']
+            if normalized_name.startswith(('kimi-k2.5', 'kimi-k2.6')):
+                return ['provider_default', 'disabled', 'enabled']
+            if 'thinking' in normalized_name:
+                return ['provider_default']
+
+        if family == 'qwen' and normalized_name.startswith(('qwen-', 'qwen3', 'qwq')):
+            if self._is_dedicated_qwen_thinking_model(normalized_name):
+                if self._supports_qwen_thinking_budget(normalized_name):
+                    return ['provider_default', 'low', 'medium', 'high']
+                return ['provider_default']
+            if self._supports_qwen_thinking_budget(normalized_name):
+                return ['provider_default', 'disabled', 'low', 'medium', 'high']
+            return ['provider_default', 'disabled', 'enabled']
+
+        if family == 'doubao' and normalized_name.startswith(('doubao-', 'seed-')):
+            return ['provider_default', 'disabled', 'low', 'medium', 'high']
+
+        if family == 'mimo' and normalized_name.startswith(('mimo-v2.5',)):
+            return ['provider_default', 'disabled', 'enabled']
+
+        if family == 'anthropic' and normalized_name.startswith('claude-'):
+            levels = ['provider_default']
+            adaptive = self._is_anthropic_adaptive_model(normalized_name)
+            if adaptive and not self._is_anthropic_always_thinking_model(normalized_name):
+                levels.append('disabled')
+            levels.extend(['low', 'medium', 'high'])
+            if adaptive:
+                levels.extend(['xhigh', 'max'])
+            return levels
+
+        if family == 'openai' and normalized_name.startswith(('gpt-5', 'o1', 'o3', 'o4')):
+            return ['provider_default', 'low', 'medium', 'high']
+
+        return None
+
+    def _openai_reasoning_levels(self, model_name: str) -> list[str]:
+        model_info = self._safe_model_info(model_name)
+        levels = ['provider_default']
+        if model_info.get('supports_none_reasoning_effort') is True:
+            levels.append('disabled')
+        if model_info.get('supports_minimal_reasoning_effort') is True:
+            levels.append('minimal')
+        for level in ('low', 'medium', 'high'):
+            if model_info.get(f'supports_{level}_reasoning_effort') is not False:
+                levels.append(level)
+        for level in ('xhigh', 'max'):
+            if model_info.get(f'supports_{level}_reasoning_effort') is True:
+                levels.append(level)
+        return levels
+
+    def _safe_model_info(self, model_name: str) -> dict[str, typing.Any]:
+        helper = getattr(litellm, 'get_model_info', None)
+        if not callable(helper):
+            return {}
+
+        candidates = [
+            *self._metadata_model_candidates(model_name),
+            model_name,
+            self._build_litellm_model_name(model_name),
+        ]
+        for candidate in candidates:
+            try:
+                info = helper(candidate)
+            except Exception:
+                continue
+            if isinstance(info, dict):
+                return info
+            model_dump = getattr(info, 'model_dump', None)
+            if callable(model_dump):
+                try:
+                    dumped = model_dump()
+                    if isinstance(dumped, dict):
+                        return dumped
+                except Exception:
+                    continue
+        return {}
+
+    def get_reasoning_capabilities(self, model: requester.RuntimeLLMModel) -> dict[str, typing.Any]:
+        model_name = model.model_entity.name
+        abilities = model.model_entity.abilities or []
+        detected = self._supports_reasoning(model_name)
+        declared = 'reasoning' in abilities
+        family = self._reasoning_family(model_name, model)
+        known_levels = self._known_reasoning_levels(model_name, family)
+        supported = detected or declared or known_levels is not None
+        if not supported:
+            return reasoning.default_reasoning_capabilities()
+
+        normalized_name = model_name.lower()
+        if family == 'openai':
+            levels = self._openai_reasoning_levels(model_name)
+        elif known_levels is not None:
+            levels = known_levels
+        elif family == 'anthropic':
+            levels = ['provider_default', 'low', 'medium', 'high']
+        elif family in {'deepseek', 'qwen', 'mimo', 'volcengine'}:
+            levels = ['provider_default', 'disabled', 'enabled']
+        elif family == 'doubao':
+            levels = ['provider_default', 'disabled', 'low', 'medium', 'high']
+        elif family == 'ollama':
+            levels = ['provider_default']
+            levels.append('disabled')
+            if normalized_name.startswith('gpt-oss') or '/gpt-oss' in normalized_name:
+                levels.extend(['low', 'medium', 'high'])
+            else:
+                levels.append('enabled')
+        elif family in self._INFERRED_EFFORT_PROVIDERS:
+            levels = ['provider_default', 'low', 'medium', 'high']
+        else:
+            levels = ['provider_default']
+
+        capabilities = {
+            'supported': True,
+            'levels': list(dict.fromkeys(levels)),
+            'source': 'litellm' if detected else ('provider' if known_levels is not None else 'manual'),
+        }
+        if family == 'qwen' and 'disabled' in capabilities['levels'] and 'enabled' not in capabilities['levels']:
+            capabilities['legacy_levels'] = ['enabled']
+        return capabilities
+
+    def _build_reasoning_args(self, model: requester.RuntimeLLMModel) -> dict[str, typing.Any]:
+        level = self._reasoning_level(model)
+        if level == 'provider_default':
+            return {}
+
+        config = {'level': level}
+        capabilities = self.get_reasoning_capabilities(model)
+        try:
+            reasoning.validate_reasoning_capabilities(config, capabilities, model.model_entity.name)
+        except ValueError as exc:
+            raise errors.RequesterError(str(exc)) from exc
+
+        family = self._reasoning_family(model.model_entity.name, model)
+        if level == 'disabled':
+            if family in {'deepseek', 'kimi', 'mimo', 'doubao'}:
+                return {'extra_body': {'thinking': {'type': 'disabled'}}}
+            if family == 'qwen':
+                return {'extra_body': {'enable_thinking': False}}
+            if family == 'volcengine':
+                return {'extra_body': {'thinking': {'type': 'disabled'}}}
+            if family == 'anthropic':
+                return {'thinking': {'type': 'disabled'}}
+            return {'reasoning_effort': 'none'}
+        if level == 'enabled':
+            if family in {'deepseek', 'kimi', 'mimo', 'volcengine'}:
+                return {'extra_body': {'thinking': {'type': 'enabled'}}}
+            if family == 'qwen':
+                return {'extra_body': {'enable_thinking': True}}
+            return {'reasoning_effort': 'low'}
+        if family == 'qwen' and level in self._QWEN_REASONING_BUDGETS:
+            return {
+                'extra_body': {
+                    'enable_thinking': True,
+                    'thinking_budget': self._QWEN_REASONING_BUDGETS[level],
+                }
+            }
+        if family == 'deepseek':
+            return {
+                'extra_body': {
+                    'thinking': {'type': 'enabled'},
+                    'reasoning_effort': level,
+                }
+            }
+        return {'reasoning_effort': level}
+
+    @staticmethod
+    def _reasoning_config_value(model: requester.RuntimeLLMModel) -> typing.Any:
+        raw_config = getattr(model, 'reasoning_config_override', None)
+        if raw_config is None:
+            raw_config = getattr(model.model_entity, 'reasoning_config', None)
+        if not isinstance(raw_config, dict):
+            return None
+        return raw_config
+
+    def _reasoning_level(self, model: requester.RuntimeLLMModel) -> str:
+        return reasoning.normalize_reasoning_config(self._reasoning_config_value(model))['level']
+
     def _infer_model_type(self, model_id: str) -> str:
         normalized_id = (model_id or '').lower()
         if any(kw in normalized_id for kw in self._RERANK_MODEL_HINTS):
@@ -344,6 +681,13 @@ class LiteLLMRequester(requester.ProviderAPIRequester):
             )
             if supports_provider_reported_vision or self._supports_vision(model_id):
                 abilities.append('vision')
+            supports_provider_reported_reasoning = bool(
+                model_payload and model_payload.get('supports_reasoning') is True
+            )
+            family = self._reasoning_family(model_id)
+            supports_known_reasoning = self._known_reasoning_levels(model_id, family) is not None
+            if supports_provider_reported_reasoning or supports_known_reasoning or self._supports_reasoning(model_id):
+                abilities.append('reasoning')
             scanned_model['abilities'] = abilities
 
             context_length = self._context_length_from_scan_payload(model_payload)
@@ -354,13 +698,51 @@ class LiteLLMRequester(requester.ProviderAPIRequester):
 
         return scanned_model
 
-    def _convert_messages(self, messages: typing.List[provider_message.Message]) -> list[dict]:
+    def _convert_messages(
+        self,
+        messages: typing.List[provider_message.Message],
+        reasoning_family: str = '',
+        include_reasoning_context: bool = True,
+    ) -> list[dict]:
         """Convert LangBot messages to LiteLLM/OpenAI format."""
         req_messages = []
         for m in messages:
             msg_dict = m.dict(exclude_none=True)
             content = msg_dict.get('content')
 
+            if msg_dict.get('role') == 'assistant' and reasoning_family:
+                provider_fields = msg_dict.get('provider_specific_fields')
+                if isinstance(provider_fields, dict):
+                    cleaned_provider_fields = dict(provider_fields)
+                    reasoning_content = cleaned_provider_fields.pop('reasoning_content', None)
+                    thinking_blocks = cleaned_provider_fields.pop('thinking_blocks', None)
+
+                    # ``content`` is also used for the user-facing rendering.
+                    # Do not replay that rendered <think> wrapper alongside the
+                    # structured provider reasoning on the next request.
+                    if reasoning_content or thinking_blocks:
+                        content = msg_dict.get('content')
+                        if isinstance(content, str):
+                            msg_dict['content'] = self._strip_think(content)
+
+                    if include_reasoning_context:
+                        if reasoning_family == 'anthropic' and thinking_blocks:
+                            msg_dict['thinking_blocks'] = thinking_blocks
+                        elif reasoning_family in {
+                            'deepseek',
+                            'kimi',
+                            'qwen',
+                            'doubao',
+                            'mimo',
+                            'volcengine',
+                        } and isinstance(reasoning_content, str):
+                            msg_dict['reasoning_content'] = reasoning_content
+
+                    if cleaned_provider_fields:
+                        msg_dict['provider_specific_fields'] = cleaned_provider_fields
+                    else:
+                        msg_dict.pop('provider_specific_fields', None)
+
             if isinstance(content, list):
                 converted_parts = []
                 for part in content:
@@ -421,6 +803,52 @@ class LiteLLMRequester(requester.ProviderAPIRequester):
 
         return content or ''
 
+    @staticmethod
+    def _thinking_blocks_text(thinking_blocks: typing.Any) -> str:
+        if not isinstance(thinking_blocks, list):
+            return ''
+        parts = []
+        for block in thinking_blocks:
+            if isinstance(block, dict):
+                text = block.get('thinking')
+            else:
+                text = getattr(block, 'thinking', None)
+            if isinstance(text, str) and text:
+                parts.append(text)
+        return ''.join(parts)
+
+    @classmethod
+    def _merge_thinking_blocks(
+        cls,
+        current: list[dict[str, typing.Any]],
+        incoming: typing.Any,
+    ) -> list[dict[str, typing.Any]]:
+        """Merge Anthropic thinking block fragments emitted by a stream."""
+        if not isinstance(incoming, list):
+            return current
+        merged = [dict(block) for block in current]
+        for raw_block in incoming:
+            block = cls._as_dict(raw_block)
+            if not block:
+                continue
+            block_type = block.get('type')
+            if block_type == 'redacted_thinking':
+                merged.append(block)
+                continue
+
+            text = block.get('thinking') if isinstance(block.get('thinking'), str) else ''
+            signature = block.get('signature')
+            if merged and merged[-1].get('type') == 'thinking' and not merged[-1].get('signature'):
+                merged[-1]['thinking'] = f'{merged[-1].get("thinking", "")}{text}'
+                if signature:
+                    merged[-1]['signature'] = signature
+            elif merged and signature and merged[-1].get('signature') == signature:
+                if text and text != merged[-1].get('thinking', ''):
+                    merged[-1]['thinking'] = f'{merged[-1].get("thinking", "")}{text}'
+            else:
+                merged.append(block)
+        return merged
+
     @staticmethod
     def _normalize_usage(usage: typing.Any) -> dict:
         """Normalize a LiteLLM/OpenAI usage object into a plain token dict.
@@ -651,7 +1079,13 @@ class LiteLLMRequester(requester.ProviderAPIRequester):
         stream: bool = False,
     ) -> dict:
         """Build common completion arguments for invoke_llm and invoke_llm_stream."""
-        req_messages = self._convert_messages(messages)
+        reasoning_family = self._reasoning_family(model.model_entity.name, model)
+        reasoning_level = self._reasoning_level(model)
+        req_messages = self._convert_messages(
+            messages,
+            reasoning_family=reasoning_family,
+            include_reasoning_context=reasoning_level != 'disabled',
+        )
         model_name = self._build_litellm_model_name(model.model_entity.name)
         api_key = model.provider.token_mgr.get_token()
 
@@ -670,6 +1104,29 @@ class LiteLLMRequester(requester.ProviderAPIRequester):
             args.update(model.model_entity.extra_args)
         args.update(extra_args)
 
+        reasoning_args = self._build_reasoning_args(model)
+        if reasoning_args:
+            conflicts = reasoning.find_reasoning_arg_conflicts(model.model_entity.extra_args)
+            conflicts.extend(reasoning.find_reasoning_arg_conflicts(extra_args))
+            if conflicts:
+                raise errors.RequesterError(
+                    'reasoning_config conflicts with advanced parameters: ' + ', '.join(dict.fromkeys(conflicts))
+                )
+            reasoning_extra_body = reasoning_args.get('extra_body')
+            if isinstance(reasoning_extra_body, dict):
+                existing_extra_body = args.get('extra_body') or {}
+                if not isinstance(existing_extra_body, dict):
+                    raise errors.RequesterError('extra_body must be an object')
+                args.update({key: value for key, value in reasoning_args.items() if key != 'extra_body'})
+                args['extra_body'] = {**existing_extra_body, **reasoning_extra_body}
+            else:
+                args.update(reasoning_args)
+            if 'reasoning_effort' in reasoning_args and self._get_custom_llm_provider() == 'openai':
+                allowed_openai_params = args.get('allowed_openai_params') or []
+                if not isinstance(allowed_openai_params, (list, tuple, set)):
+                    raise errors.RequesterError('allowed_openai_params must be an array')
+                args['allowed_openai_params'] = list(dict.fromkeys([*allowed_openai_params, 'reasoning_effort']))
+
         if funcs:
             tools = await self.ap.tool_mgr.generate_tools_for_openai(funcs)
             if tools:
@@ -699,10 +1156,21 @@ class LiteLLMRequester(requester.ProviderAPIRequester):
 
             content = message_data.get('content', '')
             reasoning_content = message_data.get('reasoning_content', None)
-            message_data['content'] = self._process_thinking_content(content, reasoning_content, remove_think)
+            thinking_blocks = message_data.get('thinking_blocks')
+            if reasoning_content or thinking_blocks:
+                provider_fields = dict(message_data.get('provider_specific_fields') or {})
+                if reasoning_content:
+                    provider_fields['reasoning_content'] = reasoning_content
+                if thinking_blocks:
+                    provider_fields['thinking_blocks'] = thinking_blocks
+                message_data['provider_specific_fields'] = provider_fields
+            display_reasoning = reasoning_content or self._thinking_blocks_text(thinking_blocks) or None
+            message_data['content'] = self._process_thinking_content(content, display_reasoning, remove_think)
 
             if 'reasoning_content' in message_data:
                 del message_data['reasoning_content']
+            if 'thinking_blocks' in message_data:
+                del message_data['thinking_blocks']
 
             message = provider_message.Message(**message_data)
             usage_info = self._extract_usage(response)
@@ -728,6 +1196,9 @@ class LiteLLMRequester(requester.ProviderAPIRequester):
         role = 'assistant'
         tool_call_state: dict[int, dict[str, typing.Any]] = {}
         think_state = _ThinkStripState() if remove_think else None
+        reasoning_started = False
+        reasoning_closed = False
+        thinking_blocks_state: list[dict[str, typing.Any]] = []
 
         try:
             response = await acompletion(**args)
@@ -758,28 +1229,63 @@ class LiteLLMRequester(requester.ProviderAPIRequester):
                 if 'role' in delta and delta['role']:
                     role = delta['role']
 
-                delta_content = delta.get('content', '')
-                reasoning_content = delta.get('reasoning_content', '')
+                delta_content = delta.get('content') or ''
+                reasoning_content = delta.get('reasoning_content') or ''
+                provider_fields = dict(delta.get('provider_specific_fields') or {})
+                raw_thinking_blocks = delta.get('thinking_blocks')
+                if raw_thinking_blocks:
+                    thinking_blocks_state = self._merge_thinking_blocks(thinking_blocks_state, raw_thinking_blocks)
+                    provider_fields['thinking_blocks'] = thinking_blocks_state
+                thinking_blocks_text = self._thinking_blocks_text(raw_thinking_blocks)
+                display_reasoning_content = reasoning_content or thinking_blocks_text
 
                 # Handle reasoning_content based on remove_think flag
                 if reasoning_content:
+                    provider_fields['reasoning_content'] = reasoning_content
                     if remove_think:
-                        # Skip reasoning content when remove_think is True
-                        chunk_idx += 1
-                        continue
+                        delta_content = delta_content or None
                     else:
-                        # Use reasoning_content as the displayed content
-                        delta_content = reasoning_content
+                        # Stream explicit markers so downstream adapters and
+                        # the debug page see the same format as non-streaming
+                        # responses.
+                        if not reasoning_started:
+                            delta_content = '<think>\n'
+                            reasoning_started = True
+                        else:
+                            delta_content = ''
+                        delta_content += display_reasoning_content
+                        if delta.get('content'):
+                            delta_content += f'\n</think>\n{delta.get("content")}'
+                            reasoning_closed = True
+
+                elif display_reasoning_content:
+                    if remove_think:
+                        delta_content = delta_content or None
+                    else:
+                        if not reasoning_started:
+                            delta_content = '<think>\n'
+                            reasoning_started = True
+                        else:
+                            delta_content = ''
+                        delta_content += display_reasoning_content
+                        if delta.get('content'):
+                            delta_content += f'\n</think>\n{delta.get("content")}'
+                            reasoning_closed = True
+
+                elif delta_content and not remove_think and reasoning_started and not reasoning_closed:
+                    delta_content = f'\n</think>\n{delta_content}'
+                    reasoning_closed = True
+
+                if finish_reason and not remove_think and reasoning_started and not reasoning_closed:
+                    delta_content = f'{delta_content}\n</think>\n'
+                    reasoning_closed = True
 
                 if think_state is not None and delta_content:
                     delta_content = think_state.feed(delta_content)
-                    if not delta_content:
-                        chunk_idx += 1
-                        continue
 
                 tool_calls = self._normalize_stream_tool_calls(delta.get('tool_calls'), tool_call_state)
 
-                if chunk_idx == 0 and not delta_content and not tool_calls:
+                if not delta_content and not tool_calls and not provider_fields and not finish_reason:
                     chunk_idx += 1
                     continue
 
@@ -791,13 +1297,20 @@ class LiteLLMRequester(requester.ProviderAPIRequester):
                 }
 
                 # Preserve provider_specific_fields from delta (e.g., Gemini thought_signatures)
-                if delta.get('provider_specific_fields'):
-                    chunk_data['provider_specific_fields'] = delta['provider_specific_fields']
+                if provider_fields:
+                    chunk_data['provider_specific_fields'] = provider_fields
 
                 chunk_data = {k: v for k, v in chunk_data.items() if v is not None}
                 yield provider_message.MessageChunk(**chunk_data)
                 chunk_idx += 1
 
+            if reasoning_started and not reasoning_closed:
+                yield provider_message.MessageChunk(
+                    role=role,
+                    content='\n</think>\n',
+                    is_final=True,
+                )
+
             if think_state is not None:
                 pending_content = think_state.flush()
                 if pending_content:

src/langbot/pkg/provider/runners/localagent.py

@@ -6,6 +6,7 @@ import typing
 from .. import runner
 from ...telemetry import features as telemetry_features
 from ..modelmgr import requester as modelmgr_requester
+from ..modelmgr import reasoning as modelmgr_reasoning
 from ..tools.loaders.native import EXEC_TOOL_NAME
 import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query
 import langbot_plugin.api.entities.builtin.provider.message as provider_message
@@ -60,6 +61,7 @@ class _StreamAccumulator:
         self.msg_idx = 0
         self.accumulated_content = initial_content or ''
         self.last_role = 'assistant'
+        self.provider_specific_fields: dict[str, typing.Any] = {}
         self.msg_sequence = msg_sequence
         self.remove_think = remove_think
         self._think_state = None
@@ -90,10 +92,27 @@ class _StreamAccumulator:
                             name=tool_call.function.name if tool_call.function else '',
                             arguments='',
                         ),
+                        provider_specific_fields=(
+                            dict(tool_call.provider_specific_fields) if tool_call.provider_specific_fields else None
+                        ),
                     )
+                elif tool_call.provider_specific_fields:
+                    existing_fields = self.tool_calls_map[tool_call.id].provider_specific_fields or {}
+                    self.tool_calls_map[tool_call.id].provider_specific_fields = {
+                        **existing_fields,
+                        **tool_call.provider_specific_fields,
+                    }
                 if tool_call.function and tool_call.function.arguments:
                     self.tool_calls_map[tool_call.id].function.arguments += tool_call.function.arguments
 
+        if msg.provider_specific_fields:
+            for key, value in msg.provider_specific_fields.items():
+                if key == 'reasoning_content' and isinstance(value, str):
+                    previous = self.provider_specific_fields.get(key, '')
+                    self.provider_specific_fields[key] = f'{previous}{value}'
+                else:
+                    self.provider_specific_fields[key] = value
+
         if msg.is_final:
             self._flush_think_state()
 
@@ -103,6 +122,7 @@ class _StreamAccumulator:
                 role=self.last_role,
                 content=self._maybe_strip_think(self.accumulated_content),
                 tool_calls=list(self.tool_calls_map.values()) if (self.tool_calls_map and msg.is_final) else None,
+                provider_specific_fields=(self.provider_specific_fields or None) if msg.is_final else None,
                 is_final=msg.is_final,
                 msg_sequence=self.msg_sequence,
             )
@@ -115,6 +135,7 @@ class _StreamAccumulator:
             role=self.last_role,
             content=self._maybe_strip_think(self.accumulated_content),
             tool_calls=list(self.tool_calls_map.values()) if self.tool_calls_map else None,
+            provider_specific_fields=self.provider_specific_fields or None,
             msg_sequence=self.msg_sequence,
         )
 
@@ -233,9 +254,10 @@ class LocalAgentRunner(runner.RequestRunner):
                     execution_context,
                     query.use_llm_model_uuid,
                 )
-                candidates.append(primary)
             except ValueError:
                 self.ap.logger.warning(f'Primary model {query.use_llm_model_uuid} not found')
+            else:
+                candidates.append(LocalAgentRunner._apply_pipeline_reasoning_config(query, primary))
 
         # Fallback models
         fallback_uuids = (query.variables or {}).get('_fallback_model_uuids', [])
@@ -245,12 +267,31 @@ class LocalAgentRunner(runner.RequestRunner):
                     execution_context,
                     fb_uuid,
                 )
-                candidates.append(fb_model)
             except ValueError:
                 self.ap.logger.warning(f'Fallback model {fb_uuid} not found, skipping')
+            else:
+                candidates.append(LocalAgentRunner._apply_pipeline_reasoning_config(query, fb_model))
 
         return candidates
 
+    @staticmethod
+    def _apply_pipeline_reasoning_config(
+        query: pipeline_query.Query,
+        model: modelmgr_requester.RuntimeLLMModel,
+    ) -> modelmgr_requester.RuntimeLLMModel:
+        local_agent_config = query.pipeline_config.get('ai', {}).get('local-agent', {})
+        model_config = local_agent_config.get('model', {})
+        reasoning_by_model = model_config.get('reasoning', {}) if isinstance(model_config, dict) else {}
+        level = (
+            reasoning_by_model.get(model.model_entity.uuid, 'provider_default')
+            if isinstance(reasoning_by_model, dict)
+            else 'provider_default'
+        )
+        reasoning_config = modelmgr_reasoning.normalize_reasoning_config({'level': level})
+        configured_model = copy.copy(model)
+        configured_model.reasoning_config_override = reasoning_config
+        return configured_model
+
     async def _invoke_with_fallback(
         self,
         query: pipeline_query.Query,

src/langbot/templates/metadata/pipeline/ai.yaml

@@ -92,6 +92,7 @@ stages:
         default:
           primary: ''
           fallbacks: []
+          reasoning: {}
       - name: max-round
         label:
           en_US: Max Round