perf(llm): harden request policy and optimize connections, reasoning and prompt caching (#2661)

* fix(llm): centralize request policy and preserve stream diagnostics

* test(llm): dispose SQLite engine after migration regression

* perf(llm): reuse connections and unify reasoning and prompt cache policy

* fix(llm): preserve explicit native prompt cache options

* perf(llm): fix measured TLS setup and streaming reuse overhead

* fix(llm): resolve plugin request-chain regressions found by E2E

---------

Co-authored-by: Hyu <[email protected]>

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

@@ -605,6 +605,7 @@ class MonitoringService(ExecutionDetailsMixin):
         cost: float | None = None,
         error_message: str | None = None,
         message_id: str | None = None,
+        details: dict | None = None,
     ) -> str:
         """Record an LLM call"""
         workspace_uuid = self._require_write_context(context)
@@ -616,7 +617,7 @@ class MonitoringService(ExecutionDetailsMixin):
             'model_name': model_name,
             'input_tokens': input_tokens,
             'output_tokens': output_tokens,
-            'total_tokens': input_tokens + output_tokens,
+            'total_tokens': ((details or {}).get('usage') or {}).get('total_tokens') or (input_tokens + output_tokens),
             'duration': duration,
             'cost': cost,
             'status': status,
@@ -627,6 +628,7 @@ class MonitoringService(ExecutionDetailsMixin):
             'session_id': session_id,
             'error_message': error_message,
             'message_id': message_id,
+            'details': details,
         }
 
         await self.ap.persistence_mgr.execute_async(

src/langbot/pkg/api/mcp/server.py

@@ -165,7 +165,7 @@ class LangBotMCPServer:
             return _dump(ap.box_service.get_recent_errors(context))
 
         @mcp.tool(
-            description='Read bounded Workspace runtime records. Filters follow the monitoring service: bot_ids, pipeline_ids, session_ids, start_time, end_time, mode (all/real/debug), execution_statuses (owning execution status groups), knowledge_base_id, user_query, is_active as supported by the record kind.'
+            description='Read bounded Workspace runtime records. LLM calls include optional details (timing in ms, complete provider usage, structured error); older records have null details. Filters follow the monitoring service: bot_ids, pipeline_ids, session_ids, start_time, end_time, mode (all/real/debug), execution_statuses (owning execution status groups), knowledge_base_id, user_query, is_active as supported by the record kind.'
         )
         async def get_monitoring_records(
             kind: typing.Literal['messages', 'llm_calls', 'tool_calls', 'embedding_calls', 'sessions', 'errors'],
@@ -513,7 +513,13 @@ class LangBotMCPServer:
             context = _authorized(Permission.RESOURCE_VIEW)
             return _dump(await ap.llm_model_service.get_llm_models(context, include_secret=False))
 
-        @mcp.tool(description='Get a single LLM model by UUID.')
+        @mcp.tool(
+            description=(
+                'Get a single LLM model by UUID, including reasoning capabilities. '
+                'Optional budget_tokens min/max describes numeric thinking-budget support; '
+                'reasoning_config and extra_args retain the existing model API shape.'
+            )
+        )
         async def get_llm_model(model_uuid: str) -> str:
             context = _authorized(Permission.RESOURCE_VIEW)
             return _dump(await ap.llm_model_service.get_llm_model(context, model_uuid, include_secret=False))

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

@@ -70,6 +70,8 @@ class MonitoringLLMCall(Base):
     error_message = sqlalchemy.Column(sqlalchemy.Text, nullable=True)
     message_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=True, index=True)  # Associated message ID
 
+    details = sqlalchemy.Column(sqlalchemy.JSON, nullable=True)
+
     __table_args__ = (
         sqlalchemy.Index('ix_monitoring_llm_calls_workspace_timestamp', 'workspace_uuid', 'timestamp'),
         sqlalchemy.Index('ix_monitoring_llm_calls_workspace_session', 'workspace_uuid', 'session_id'),

src/langbot/pkg/persistence/alembic/versions/0038_llm_call_details.py

@@ -0,0 +1,25 @@
+"""Persist additive LLM timing, usage and structured failure diagnostics."""
+
+from alembic import op
+import sqlalchemy as sa
+
+revision = '0038_llm_call_details'
+down_revision = '0037_metadata_text'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    inspector = sa.inspect(op.get_bind())
+    if inspector.has_table('monitoring_llm_calls'):
+        columns = {column['name'] for column in inspector.get_columns('monitoring_llm_calls')}
+        if 'details' not in columns:
+            op.add_column('monitoring_llm_calls', sa.Column('details', sa.JSON(), nullable=True))
+
+
+def downgrade():
+    inspector = sa.inspect(op.get_bind())
+    if inspector.has_table('monitoring_llm_calls'):
+        columns = {column['name'] for column in inspector.get_columns('monitoring_llm_calls')}
+        if 'details' in columns:
+            op.drop_column('monitoring_llm_calls', 'details')

src/langbot/pkg/pipeline/monitoring_helper.py

@@ -380,6 +380,7 @@ class MonitoringHelper:
         cost: float | None = None,
         error_message: str | None = None,
         message_id: str | None = None,
+        details: dict | None = None,
     ):
         """Record LLM call"""
         try:
@@ -400,6 +401,7 @@ class MonitoringHelper:
                 cost=cost,
                 error_message=error_message,
                 message_id=message_id,
+                details=details,
             )
         except Exception as e:
             ap.logger.error(f'Failed to record LLM call: {e}')

src/langbot/pkg/plugin/handler.py

@@ -443,13 +443,19 @@ def _resolve_remove_think(data: dict[str, Any], query: Any | None) -> bool:
     return False
 
 
-def _merge_model_extra_args(model: Any, call_extra_args: Any) -> dict[str, Any]:
+def _merge_model_extra_args(model: Any, call_extra_args: Any, *, llm_policy: bool = False) -> dict[str, Any]:
     """Merge persisted model extra_args with action-level overrides."""
     merged: dict[str, Any] = {}
 
     model_extra_args = getattr(getattr(model, 'model_entity', None), 'extra_args', None)
     if isinstance(model_extra_args, dict):
         merged.update(model_extra_args)
+    if llm_policy:
+        # RuntimeProvider resolves model and call policy layers independently.
+        # Flattening them here makes a model request_policy.retries override a
+        # caller's legacy num_retries=0 (and similarly hides timeout overrides).
+        for key in ('request_policy', 'num_retries', 'max_retries', 'timeout'):
+            merged.pop(key, None)
     if isinstance(call_extra_args, dict):
         merged.update(call_extra_args)
 
@@ -1484,7 +1490,9 @@ class RuntimeConnectionHandler(handler.Handler):
 
             funcs_obj = [resource_tool.LLMTool.model_validate({**func, 'func': _placeholder_func}) for func in funcs]
             query = _resolve_action_query(data, session, self.ap, action_context)
-            effective_extra_args = _merge_model_extra_args(llm_model, extra_args)
+            if query is None:
+                query = _build_plugin_action_query(action_context)
+            effective_extra_args = _merge_model_extra_args(llm_model, extra_args, llm_policy=True)
             remove_think = _resolve_remove_think(data, query)
             effective_funcs = funcs_obj if 'func_call' in (llm_model.model_entity.abilities or []) else []
 
@@ -1580,7 +1588,9 @@ class RuntimeConnectionHandler(handler.Handler):
 
             funcs_obj = [resource_tool.LLMTool.model_validate({**func, 'func': _placeholder_func}) for func in funcs]
             query = _resolve_action_query(data, session, self.ap, action_context)
-            effective_extra_args = _merge_model_extra_args(llm_model, extra_args)
+            if query is None:
+                query = _build_plugin_action_query(action_context)
+            effective_extra_args = _merge_model_extra_args(llm_model, extra_args, llm_policy=True)
             remove_think = _resolve_remove_think(data, query)
             effective_funcs = funcs_obj if 'func_call' in (llm_model.model_entity.abilities or []) else []
 

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

@@ -1,10 +1,19 @@
 """Explicitly safe Codex failures; never construct messages from upstream bodies."""
 
+from .errors import RequesterError
+
 
 class CodexProviderError(ValueError):
     """A known provider failure safe to expose at the HTTP boundary."""
 
     def __init__(self, message: str, status_code: int = 502, error_code: str = 'codex_upstream_failure'):
-        super().__init__(message)
+        ValueError.__init__(self, message)
+        self.code = error_code
+        self.retryable = status_code in (500, 502, 503, 504)
+        self.retry_after = None
+        self.request_id = None
+        self.phase = None
         self.status_code = status_code
         self.error_code = error_code
+
+    as_dict = RequesterError.as_dict

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

@@ -1,5 +1,28 @@
 class RequesterError(Exception):
     """Base class for all Requester errors."""
 
-    def __init__(self, message: str):
+    def __init__(
+        self,
+        message: str,
+        *,
+        code: str = 'provider_error',
+        status_code: int | None = None,
+        retryable: bool = False,
+        retry_after: float | None = None,
+        request_id: str | None = None,
+        phase: str | None = None,
+    ):
         super().__init__('模型请求失败: ' + message)
+        self.code = code
+        self.status_code = status_code
+        self.retryable = retryable
+        self.retry_after = retry_after
+        self.request_id = request_id
+        self.phase = phase
+
+    def as_dict(self) -> dict:
+        """Machine-readable diagnostics without upstream bodies or credentials."""
+        return {
+            key: getattr(self, key)
+            for key in ('code', 'status_code', 'retryable', 'retry_after', 'request_id', 'phase')
+        }

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

@@ -0,0 +1,114 @@
+"""Requester-local connections, retired only after their active leases finish."""
+
+from contextlib import asynccontextmanager
+import asyncio
+import re
+
+import httpx
+
+from .errors import RequesterError
+
+
+class _CompletedSSEStream(httpx.AsyncByteStream):
+    """Finish bounded HTTP framing before SDKs stop at the SSE DONE record."""
+
+    _done = re.compile(rb'\ndata: ?\[DONE\]\r?\n\r?\n')
+
+    def __init__(self, stream):
+        self.stream = stream
+
+    async def __aiter__(self):
+        iterator = self.stream.__aiter__()
+        tail = b'\n'
+        async for chunk in iterator:
+            probe = tail + chunk
+            if self._done.search(probe):
+                # SDK parsers stop at [DONE] before httpcore observes EOF. This
+                # otherwise discards the connection on every successful stream.
+                # Do not drain on cancellation, truncation, or merely a finish
+                # reason: providers can still send usage before the DONE record.
+                try:
+                    async with asyncio.timeout(0.05):
+                        drained = 0
+                        async for extra in iterator:
+                            drained += len(extra)
+                            if drained >= 65536:
+                                break
+                except (TimeoutError, httpx.HTTPError):
+                    pass
+                yield chunk
+                return
+            tail = probe[-32:]
+            yield chunk
+
+    async def aclose(self):
+        await self.stream.aclose()
+
+
+async def _finish_sse_body(response):
+    if response.headers.get('content-type', '').split(';')[0].strip() == 'text/event-stream':
+        response.stream = _CompletedSSEStream(response.stream)
+
+
+class RequesterHTTPPool:
+    def __init__(self, *, follow_redirects=False, drain_sse=False):
+        self.client = None
+        self.active = 0
+        self.retired = False
+        self.follow_redirects = follow_redirects
+        self.drain_sse = drain_sse
+        self._creation_lock = asyncio.Lock()
+
+    @asynccontextmanager
+    async def lease(self):
+        if self.retired:
+            raise RequesterError('Provider has been retired', code='provider_retired')
+        if self.client is None:
+            async with self._creation_lock:
+                if self.retired:
+                    raise RequesterError('Provider has been retired', code='provider_retired')
+                if self.client is None:
+                    # HTTPX otherwise reloads the same CA bundle for its direct
+                    # transport and each environment proxy mount. Load it once,
+                    # off the event loop, preserving HTTPX's TLS/env semantics.
+                    # Cancellation here leaves no open HTTP client or sockets.
+                    ssl_context = await asyncio.to_thread(httpx.create_ssl_context)
+                    if self.retired:
+                        raise RequesterError('Provider has been retired', code='provider_retired')
+                    # Auth and deadlines remain request-local. No pool or TLS
+                    # context is shared across runtime providers.
+                    self.client = httpx.AsyncClient(
+                        verify=ssl_context,
+                        follow_redirects=self.follow_redirects,
+                        limits=httpx.Limits(max_connections=100, max_keepalive_connections=20, keepalive_expiry=30),
+                        event_hooks={'response': [_finish_sse_body]} if self.drain_sse else None,
+                    )
+        self.active += 1
+        try:
+            yield self.client
+        finally:
+            self.active -= 1
+            if self.retired and not self.active:
+                await self._close()
+
+    async def _close(self):
+        client, self.client = self.client, None
+        if client is not None:
+            await client.aclose()
+
+    async def aclose(self):
+        self.retired = True
+        if not self.active:
+            await self._close()
+
+
+@asynccontextmanager
+async def requester_http_client(owner):
+    pool = getattr(owner, '_http_pool', None)
+    if pool is not None:
+        async with pool.lease() as client:
+            yield client
+    else:
+        # Preserve direct/uninitialized requester use by older extensions/tests.
+        async with httpx.AsyncClient(follow_redirects=False) as client:
+            yield client

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

@@ -743,6 +743,7 @@ class ModelManager:
         config = {
             'base_url': provider_entity.base_url,
             'requester_name': provider_entity.requester,
+            'request_policy': self.ap.instance_config.data.get('llm_request_policy', {}),
         }
 
         if provider_entity.requester == 'openai-codex':

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

@@ -0,0 +1,82 @@
+"""Opt-in upstream prompt caching; never cache generated answers locally."""
+
+import copy
+import hashlib
+
+from .errors import RequesterError
+
+
+def _has_breakpoint(value):
+    if isinstance(value, dict):
+        return 'cache_control' in value or any(_has_breakpoint(item) for item in value.values())
+    if isinstance(value, list):
+        return any(_has_breakpoint(item) for item in value)
+    return False
+
+
+def apply_prompt_caching(args, state, *, provider, family):
+    config = args.pop('prompt_caching', None)
+    if config is None:
+        return
+    if not isinstance(config, dict) or set(config) - {'mode', 'retention', 'ttl'}:
+        raise RequesterError('Invalid prompt_caching fields', code='invalid_policy')
+    mode = config.get('mode', 'provider_default')
+    if mode not in ('provider_default', 'openai', 'anthropic'):
+        raise RequesterError('Unsupported prompt_caching mode', code='invalid_policy')
+    if mode == 'provider_default':
+        if set(config) - {'mode'}:
+            raise RequesterError('Cache settings require an explicit mode', code='invalid_policy')
+        return
+    if mode == 'openai':
+        if provider not in ('openai', 'openrouter') or family not in ('openai', 'openrouter'):
+            raise RequesterError('OpenAI caching requires an OpenAI-compatible model', code='invalid_policy')
+        if 'ttl' in config or config.get('retention') not in (None, 'in_memory', '24h'):
+            raise RequesterError('Invalid OpenAI cache retention', code='invalid_policy')
+        body = copy.deepcopy(args.get('extra_body') or {})
+        if not isinstance(body, dict):
+            raise RequesterError('extra_body must be an object', code='invalid_policy')
+        # Bypass LiteLLM's model-specific parameter filter for explicit opt-in
+        # caching fields as well as generated hints. The SDK's extra_body wins
+        # when the same native field appears in both locations.
+        for key in ('prompt_cache_key', 'prompt_cache_retention', 'prompt_cache_options'):
+            if key in args:
+                body.setdefault(key, args.pop(key))
+        # Explicit provider-native settings win. A key is a routing hint only;
+        # it neither enables caching nor guarantees a hit.
+        if (
+            state
+            and state.conversation_id
+            and not any(key in body for key in ('prompt_cache_key', 'prompt_cache_options'))
+        ):
+            scope = f'{state.conversation_id}:{args["model"]}'
+            body['prompt_cache_key'] = 'lb-' + hashlib.sha256(scope.encode()).hexdigest()
+        if config.get('retention') and not any(
+            key in location for key in ('prompt_cache_retention', 'prompt_cache_options') for location in (args, body)
+        ):
+            body['prompt_cache_retention'] = config['retention']
+        if body:
+            args['extra_body'] = body
+        return
+    if family != 'anthropic':
+        raise RequesterError('Anthropic caching requires an Anthropic model', code='invalid_policy')
+    if 'retention' in config or config.get('ttl', '5m') not in ('5m', '1h'):
+        raise RequesterError('Invalid Anthropic cache TTL', code='invalid_policy')
+    # Existing manual or automatic breakpoints are authoritative. Do not exceed
+    # a provider's breakpoint limit or change its TTL ordering.
+    if _has_breakpoint(args):
+        return
+    messages = copy.deepcopy(args['messages'])
+    boundary = None
+    for message in messages:
+        if message.get('role') not in ('system', 'developer'):
+            break
+        content = message.get('content')
+        if isinstance(content, str) and content:
+            message['content'] = [{'type': 'text', 'text': content}]
+        if isinstance(message.get('content'), list):
+            for part in message['content']:
+                if isinstance(part, dict) and part.get('type') == 'text' and part.get('text'):
+                    boundary = part
+    if boundary is not None:
+        boundary['cache_control'] = {'type': 'ephemeral', 'ttl': config.get('ttl', '5m')}
+        args['messages'] = messages

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

@@ -48,20 +48,27 @@ _CONFLICTING_EXTRA_BODY_ARGS = {
 }
 
 
-def normalize_reasoning_config(value: typing.Any) -> dict[str, str]:
+def normalize_reasoning_config(value: typing.Any) -> dict[str, typing.Any]:
     """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'}
+    unknown_fields = set(value) - {'level', 'budget_tokens'}
     if unknown_fields:
         raise ValueError(f'Unsupported reasoning_config fields: {", ".join(sorted(unknown_fields))}')
 
-    level = value.get('level', 'provider_default')
+    budget = value.get('budget_tokens')
+    level = value.get('level', 'enabled' if budget is not None else 'provider_default')
     if level not in REASONING_LEVELS:
         raise ValueError(f'Unsupported reasoning level: {level}')
+    if budget is not None:
+        if type(budget) is not int or budget <= 0:
+            raise ValueError('budget_tokens must be a positive integer')
+        if level != 'enabled':
+            raise ValueError('budget_tokens requires level enabled; do not combine a token budget with an effort level')
+        return {'level': level, 'budget_tokens': budget}
     return {'level': typing.cast(str, level)}
 
 
@@ -69,7 +76,7 @@ def validate_reasoning_config(
     value: typing.Any,
     abilities: typing.Iterable[str] | None,
     extra_args: typing.Any,
-) -> dict[str, str]:
+) -> dict[str, typing.Any]:
     """Validate a model-facing reasoning config and conflicting raw arguments."""
     config = normalize_reasoning_config(value)
     if config['level'] == 'provider_default':
@@ -101,7 +108,13 @@ def validate_reasoning_capabilities(
     model_name: str,
 ) -> None:
     """Ensure an explicit reasoning level can be honored by the requester."""
-    level = normalize_reasoning_config(config)['level']
+    config = normalize_reasoning_config(config)
+    level = config['level']
+    if 'budget_tokens' in config:
+        limits = capabilities.get('budget_tokens')
+        if not isinstance(limits, dict) or not limits['min'] <= config['budget_tokens'] <= limits['max']:
+            raise ValueError(f'Unsupported reasoning token budget for model {model_name}: {config["budget_tokens"]}')
+        return
     if level == 'provider_default':
         return
 
@@ -154,3 +167,17 @@ def model_with_reasoning_level(
     scoped_model = copy.copy(model)
     scoped_model.reasoning_config_override = copy.deepcopy(config)
     return scoped_model
+
+
+def resolve_reasoning_config(model, extra_args=None):
+    """Runner policy wins; a call object replaces, rather than merges, a model policy."""
+    value = getattr(model.model_entity, 'reasoning_config', None)
+    if value is not None and not isinstance(value, (dict, str, int, float, bool, list, tuple)):
+        value = None
+    for args in (model.model_entity.extra_args, extra_args):
+        if isinstance(args, dict) and 'reasoning_config' in args:
+            value = args['reasoning_config']
+    override = getattr(model, 'reasoning_config_override', None)
+    if override is not None:
+        value = override
+    return normalize_reasoning_config(value)

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

@@ -0,0 +1,173 @@
+"""Host-owned LLM deadlines, retries and additive diagnostics.
+
+No SDK wire shape changes: request_policy is a reserved model/call extra argument.
+Context-local state lets adapters report raw events without sharing mutable state.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from contextvars import ContextVar
+from dataclasses import dataclass, fields
+from email.utils import parsedate_to_datetime
+import hashlib
+import json
+import math
+import random
+import time
+
+from .errors import RequesterError
+from .codex_errors import CodexProviderError
+
+STRUCTURED_ERRORS = (RequesterError, CodexProviderError)
+
+
+@dataclass
+class RequestPolicy:
+    retries: int = 0
+    total_timeout: float = 300
+    first_event_timeout: float = 120
+    stream_idle_timeout: float = 120
+    connect_timeout: float = 10
+    read_timeout: float = 120
+    write_timeout: float = 30
+    pool_timeout: float = 10
+    backoff_base: float = 0.5
+    backoff_max: float = 8
+    attribution: bool = True
+    conversation_header: str | None = None
+
+    @classmethod
+    def resolve(cls, *layers):
+        values = {}
+        names = {field.name for field in fields(cls)}
+        for layer in layers:
+            if not isinstance(layer, dict):
+                continue
+            # Preserve the legacy per-model/call retry knob, including zero.
+            if 'max_retries' in layer:
+                values['retries'] = layer['max_retries']
+            if 'num_retries' in layer:
+                values['retries'] = layer['num_retries']
+            if isinstance(layer.get('timeout'), (int, float)):
+                values['read_timeout'] = layer['timeout']
+                values['first_event_timeout'] = layer['timeout']
+                values['stream_idle_timeout'] = layer['timeout']
+            policy = layer.get('request_policy', {})
+            if not isinstance(policy, dict) or set(policy) - names:
+                raise RequesterError('Invalid request_policy fields', code='invalid_policy')
+            values.update(policy)
+        policy = cls(**values)
+        for field in fields(cls):
+            value = getattr(policy, field.name)
+            if field.name == 'conversation_header':
+                if value not in (None, 'x-session-id', 'x-litellm-session-id', 'x-langbot-conversation-id'):
+                    raise RequesterError('Unsupported conversation header', code='invalid_policy')
+            elif field.name == 'attribution':
+                if not isinstance(value, bool):
+                    raise RequesterError('attribution must be boolean', code='invalid_policy')
+            elif field.name == 'retries':
+                if type(value) is not int or not 0 <= value <= 5:
+                    raise RequesterError('retries must be an integer from 0 to 5', code='invalid_policy')
+            elif (
+                isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value) or value <= 0
+            ):
+                raise RequesterError(f'{field.name} must be finite and positive', code='invalid_policy')
+        return policy
+
+
+class CallState:
+    def __init__(self, policy, conversation_id=None):
+        self.policy = policy
+        self.conversation_id = conversation_id
+        self.start = time.perf_counter()
+        self.attempt_start = self.start
+        self.metrics = {'attempts': 0, 'retry_wait_ms': 0}
+        self.raw_started = False
+        self.error = None
+        self.usage = None
+        self.prepared = {}
+
+    def mark(self, name):
+        self.metrics.setdefault(name, round((time.perf_counter() - self.start) * 1000, 3))
+
+    async def wait(self, operation, phase, limit):
+        remaining = self.policy.total_timeout - (time.perf_counter() - self.start)
+        if phase == 'first_event':
+            limit = min(limit, self.policy.first_event_timeout - (time.perf_counter() - self.attempt_start))
+        token = current_call.set(self)
+        try:
+            # timeout(0) only cancels at a scheduling point. Buffered SDK
+            # responses can complete synchronously, so enforce expired budgets
+            # before starting another operation as well.
+            if min(limit, remaining) <= 0:
+                raise TimeoutError
+            async with asyncio.timeout(max(0, min(limit, remaining))):
+                return await operation()
+        except TimeoutError as exc:
+            raise RequesterError(
+                'LLM request timed out', code='timeout', retryable=True, phase='total' if remaining <= limit else phase
+            ) from exc
+        finally:
+            current_call.reset(token)
+
+    async def retry(self, error):
+        self.error = error.as_dict() if isinstance(error, STRUCTURED_ERRORS) else {'code': 'provider_error'}
+        self.metrics.setdefault('attempt_errors', []).append(dict(self.error))
+        if not isinstance(error, STRUCTURED_ERRORS) or not error.retryable or self.raw_started:
+            return False
+        if self.metrics['attempts'] > self.policy.retries:
+            return False
+        delay = random.uniform(
+            0, min(self.policy.backoff_max, self.policy.backoff_base * 2 ** (self.metrics['attempts'] - 1))
+        )
+        delay = max(delay, error.retry_after or 0)
+        remaining = self.policy.total_timeout - (time.perf_counter() - self.start)
+        if delay >= remaining:
+            return False
+        started = time.perf_counter()
+        await asyncio.sleep(delay)
+        self.metrics['retry_wait_ms'] += round((time.perf_counter() - started) * 1000, 3)
+        return True
+
+
+current_call: ContextVar[CallState | None] = ContextVar('llm_call', default=None)
+
+
+def opaque_conversation_id(query, context):
+    if query is None:
+        return None
+    session = getattr(query, 'session', None)
+    session_id = getattr(session, 'session_id', None)
+    if not isinstance(session_id, (str, int)):
+        launcher = getattr(query, 'launcher_id', None)
+        if not isinstance(launcher, (str, int)):
+            return None
+        launcher_type = getattr(query, 'launcher_type', '')
+        session_id = [str(getattr(launcher_type, 'value', launcher_type)), str(launcher)]
+    scope = [context.workspace_uuid, query.bot_uuid, query.pipeline_uuid, session_id]
+    return 'lb-' + hashlib.sha256(json.dumps(scope, ensure_ascii=True).encode()).hexdigest()
+
+
+def error_metadata(exc):
+    status = getattr(exc, 'status_code', None)
+    status = status if isinstance(status, int) else None
+    response = getattr(exc, 'response', None)
+    headers = getattr(response, 'headers', {})
+    headers = headers if hasattr(headers, 'get') else {}
+    retry_after = None
+    raw = headers.get('retry-after')
+    if raw:
+        try:
+            retry_after = float(raw)
+        except (ValueError, TypeError):
+            try:
+                retry_after = parsedate_to_datetime(raw).timestamp() - time.time()
+            except (ValueError, TypeError, OverflowError):
+                pass
+    if retry_after is not None:
+        retry_after = max(0, retry_after) if math.isfinite(retry_after) else None
+    request_id = headers.get('x-request-id') or headers.get('request-id')
+    if not isinstance(request_id, str):
+        request_id = None
+    return {'status_code': status, 'retry_after': retry_after, 'request_id': request_id[:256] if request_id else None}

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

@@ -14,6 +14,7 @@ from ...workspace.errors import WorkspaceInvariantError
 import langbot_plugin.api.entities.builtin.resource.tool as resource_tool
 from . import token
 from . import reasoning
+from . import request_policy
 import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query
 import langbot_plugin.api.entities.builtin.provider.message as provider_message
 
@@ -101,6 +102,48 @@ class RuntimeProvider:
             raise WorkspaceInvariantError('LLM invocation requires an ExecutionContext when query is absent')
         return execution_context
 
+    def _call_state(self, query, model, extra_args, context):
+        config = getattr(self.requester, 'requester_cfg', {})
+        policy = request_policy.RequestPolicy.resolve(config, model.model_entity.extra_args, extra_args)
+        if query is not None:
+            if query.variables is None:
+                query.variables = {}
+            query.variables.pop(LLM_USAGE_QUERY_VARIABLE, None)
+            query.variables.pop(STREAM_USAGE_QUERY_VARIABLE, None)
+        return request_policy.CallState(policy, request_policy.opaque_conversation_id(query, context))
+
+    async def _record_call(self, query, model, state, status, error):
+        state.mark('total_ms')
+        if query is None:
+            return
+        stream_usage = query.variables.pop(STREAM_USAGE_QUERY_VARIABLE, None)
+        usage = state.usage if state.usage is not None else stream_usage
+        if usage is not None:
+            _store_llm_usage(query, usage)
+        details = {'timing': dict(state.metrics), 'usage': usage, 'error': error}
+        query.variables['_llm_diagnostics'] = details
+        try:
+            from ...pipeline import monitoring_helper
+
+            await monitoring_helper.MonitoringHelper.record_llm_call(
+                ap=self.requester.ap,
+                query=query,
+                bot_id=query.bot_uuid or 'unknown',
+                bot_name=query.variables.get('_monitoring_bot_name', 'Unknown'),
+                pipeline_id=query.pipeline_uuid or 'unknown',
+                pipeline_name=query.variables.get('_monitoring_pipeline_name', 'Unknown'),
+                model_name=model.model_entity.name,
+                input_tokens=(usage or {}).get('prompt_tokens', 0) or 0,
+                output_tokens=(usage or {}).get('completion_tokens', 0) or 0,
+                duration_ms=int(state.metrics['total_ms']),
+                status=status,
+                error_message=error.get('message') if error else None,
+                message_id=query.variables.get('_monitoring_message_id'),
+                details=details,
+            )
+        except Exception as exc:
+            self.requester.ap.logger.error(f'[Monitoring] Failed to record LLM call: {exc}')
+
     async def invoke_llm(
         self,
         query: pipeline_query.Query | None,
@@ -111,79 +154,46 @@ class RuntimeProvider:
         remove_think: bool = False,
         execution_context: ExecutionContext | None = None,
     ) -> provider_message.Message:
-        """Bridge method for invoking LLM with monitoring"""
-        invocation_context = self._resolve_llm_execution_context(query, execution_context)
-        self._validate_invocation(model, invocation_context)
-        # Start timing for monitoring
-        start_time = time.time()
-        input_tokens = 0
-        output_tokens = 0
-        status = 'success'
-        error_message = None
-
+        """Preserve the legacy Message return while owning retries and deadlines."""
+        context = self._resolve_llm_execution_context(query, execution_context)
+        self._validate_invocation(model, context)
+        state = self._call_state(query, model, extra_args, context)
+        status, error = 'success', None
         try:
-            # Call the underlying requester
-            result = await self.requester.invoke_llm(
-                query=query,
-                model=model,
-                messages=messages,
-                funcs=funcs,
-                extra_args=extra_args,
-                remove_think=remove_think,
-            )
-
-            # Try to extract token usage if the requester returns it
-            # For requesters that return tuple (message, usage_info)
-            if isinstance(result, tuple):
-                msg, usage_info = result
-                if usage_info:
-                    _store_llm_usage(query, usage_info)
-                    input_tokens = usage_info.get('prompt_tokens', 0)
-                    output_tokens = usage_info.get('completion_tokens', 0)
-                return msg
-            else:
-                return result
-
-        except Exception as e:
+            while True:
+                state.metrics['attempts'] += 1
+                state.attempt_start = time.perf_counter()
+                try:
+                    result = await state.wait(
+                        lambda: self.requester.invoke_llm(
+                            query=query,
+                            model=model,
+                            messages=messages,
+                            funcs=funcs,
+                            extra_args=extra_args,
+                            remove_think=remove_think,
+                        ),
+                        'response',
+                        state.policy.total_timeout,
+                    )
+                    state.mark('response_completed_ms')
+                    if isinstance(result, tuple):
+                        message, state.usage = result
+                        return message
+                    return result
+                except Exception as exc:
+                    if not await state.retry(exc):
+                        raise
+        except (asyncio.CancelledError, GeneratorExit):
+            status, error = 'error', {'code': 'cancelled', 'message': 'LLM invocation cancelled'}
+            raise
+        except Exception as exc:
             status = 'error'
-            error_message = str(e)
+            error = exc.as_dict() if isinstance(exc, request_policy.STRUCTURED_ERRORS) else {'code': 'provider_error'}
+            error['message'] = str(exc)
             raise
         finally:
-            # Record LLM call monitoring data (only if query is provided)
-            if query is not None:
-                duration_ms = int((time.time() - start_time) * 1000)
-
-                # Import monitoring helper
-                try:
-                    from ...pipeline import monitoring_helper
-
-                    # Get monitoring metadata from query variables
-                    if query.variables:
-                        bot_name = query.variables.get('_monitoring_bot_name', 'Unknown')
-                        pipeline_name = query.variables.get('_monitoring_pipeline_name', 'Unknown')
-                        message_id = query.variables.get('_monitoring_message_id')
-                    else:
-                        bot_name = 'Unknown'
-                        pipeline_name = 'Unknown'
-                        message_id = None
-
-                    await monitoring_helper.MonitoringHelper.record_llm_call(
-                        ap=self.requester.ap,
-                        query=query,
-                        bot_id=query.bot_uuid or 'unknown',
-                        bot_name=bot_name,
-                        pipeline_id=query.pipeline_uuid or 'unknown',
-                        pipeline_name=pipeline_name,
-                        model_name=model.model_entity.name,
-                        input_tokens=input_tokens,
-                        output_tokens=output_tokens,
-                        duration_ms=duration_ms,
-                        status=status,
-                        error_message=error_message,
-                        message_id=message_id,
-                    )
-                except Exception as monitor_err:
-                    self.requester.ap.logger.error(f'[Monitoring] Failed to record LLM call: {monitor_err}')
+            await self._record_call(query, model, state, status, error)
 
     async def invoke_llm_stream(
         self,
@@ -194,85 +204,66 @@ class RuntimeProvider:
         extra_args: dict[str, typing.Any] = {},
         remove_think: bool = False,
         execution_context: ExecutionContext | None = None,
-    ) -> provider_message.MessageChunk:
-        """Bridge method for invoking LLM stream with monitoring"""
-        invocation_context = self._resolve_llm_execution_context(query, execution_context)
-        self._validate_invocation(model, invocation_context)
-        # Start timing for monitoring
-        start_time = time.time()
-        status = 'success'
-        error_message = None
-        input_tokens = 0
-        output_tokens = 0
-
+    ) -> typing.AsyncIterator[provider_message.MessageChunk]:
+        """Never retry after a raw upstream event or a legacy adapter chunk."""
+        context = self._resolve_llm_execution_context(query, execution_context)
+        self._validate_invocation(model, context)
+        state = self._call_state(query, model, extra_args, context)
+        status, error = 'success', None
         try:
-            # Stream the response
-            async with aclosing(
-                self.requester.invoke_llm_stream(
-                    query=query,
-                    model=model,
-                    messages=messages,
-                    funcs=funcs,
-                    extra_args=extra_args,
-                    remove_think=remove_think,
-                )
-            ) as owned_stream:
-                async for chunk in owned_stream:
-                    yield chunk
-            # Extract usage from stream if available (stored by LiteLLM requester)
-            if query:
-                if query.variables is None:
-                    query.variables = {}
-                if STREAM_USAGE_QUERY_VARIABLE in query.variables:
-                    usage_info = query.variables[STREAM_USAGE_QUERY_VARIABLE]
-                    _store_llm_usage(query, usage_info)
-                    input_tokens = usage_info.get('prompt_tokens', 0)
-                    output_tokens = usage_info.get('completion_tokens', 0)
-                    del query.variables[STREAM_USAGE_QUERY_VARIABLE]
+            while True:
+                state.metrics['attempts'] += 1
+                state.attempt_start = time.perf_counter()
+                try:
+                    async with aclosing(
+                        self.requester.invoke_llm_stream(
+                            query=query,
+                            model=model,
+                            messages=messages,
+                            funcs=funcs,
+                            extra_args=extra_args,
+                            remove_think=remove_think,
+                        )
+                    ) as stream:
+                        while True:
+                            try:
+                                chunk = await state.wait(
+                                    lambda: anext(stream),
+                                    'total'
+                                    if getattr(self.requester, 'reports_raw_timing', False) is True
+                                    else ('stream_idle' if state.raw_started else 'first_event'),
+                                    state.policy.total_timeout
+                                    if getattr(self.requester, 'reports_raw_timing', False) is True
+                                    else (
+                                        state.policy.stream_idle_timeout
+                                        if state.raw_started
+                                        else state.policy.first_event_timeout
+                                    ),
+                                )
+                            except StopAsyncIteration:
+                                break
+                            if chunk is None:
+                                continue
+                            state.raw_started = True
+                            state.mark('first_event_ms')
+                            if chunk.content:
+                                state.mark('first_visible_chunk_ms')
+                            yield chunk
+                    state.mark('response_completed_ms')
+                    return
+                except Exception as exc:
+                    if not await state.retry(exc):
+                        raise
         except (asyncio.CancelledError, GeneratorExit):
-            status = 'error'
-            error_message = 'Streaming generation cancelled by consumer'
+            status, error = 'error', {'code': 'cancelled', 'message': 'LLM stream cancelled'}
             raise
-        except Exception as e:
+        except Exception as exc:
             status = 'error'
-            error_message = str(e)
+            error = exc.as_dict() if isinstance(exc, request_policy.STRUCTURED_ERRORS) else {'code': 'provider_error'}
+            error['message'] = str(exc)
             raise
         finally:
-            # Record LLM call monitoring data (only if query is provided)
-            if query is not None:
-                duration_ms = int((time.time() - start_time) * 1000)
-
-                # Import monitoring helper
-                try:
-                    from ...pipeline import monitoring_helper
-
-                    # Get monitoring metadata from query variables
-                    if query.variables:
-                        bot_name = query.variables.get('_monitoring_bot_name', 'Unknown')
-                        pipeline_name = query.variables.get('_monitoring_pipeline_name', 'Unknown')
-                        message_id = query.variables.get('_monitoring_message_id')
-                    else:
-                        bot_name = 'Unknown'
-                        pipeline_name = 'Unknown'
-                        message_id = None
-
-                    await monitoring_helper.MonitoringHelper.record_llm_call(
-                        ap=self.requester.ap,
-                        query=query,
-                        bot_id=query.bot_uuid or 'unknown',
-                        bot_name=bot_name,
-                        pipeline_id=query.pipeline_uuid or 'unknown',
-                        pipeline_name=pipeline_name,
-                        model_name=model.model_entity.name,
-                        input_tokens=input_tokens,
-                        output_tokens=output_tokens,
-                        duration_ms=duration_ms,
-                        status=status,
-                        error_message=error_message,
-                        message_id=message_id,
-                    )
-                except Exception as monitor_err:
-                    self.requester.ap.logger.error(f'[Monitoring] Failed to record LLM stream call: {monitor_err}')
+            await self._record_call(query, model, state, status, error)
 
     async def invoke_embedding(
         self,
@@ -388,7 +379,7 @@ class RuntimeLLMModel:
     provider: RuntimeProvider
     """提供商实例"""
 
-    reasoning_config_override: dict[str, str] | None
+    reasoning_config_override: dict[str, typing.Any] | None
     """Request-scoped reasoning policy supplied by the active pipeline."""
 
     def __init__(
@@ -396,7 +387,7 @@ class RuntimeLLMModel:
         execution_context: ExecutionContext,
         model_entity: persistence_model.LLMModel,
         provider: RuntimeProvider,
-        reasoning_config_override: dict[str, str] | None = None,
+        reasoning_config_override: dict[str, typing.Any] | None = None,
     ):
         _ensure_same_execution_scope(provider.execution_context, execution_context, resource='LLM model')
         if model_entity.workspace_uuid != execution_context.workspace_uuid:

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

@@ -7,12 +7,16 @@ import json
 import secrets
 import time
 from collections import OrderedDict
+from contextlib import AsyncExitStack, aclosing
 
 import httpx
 import langbot
 import langbot_plugin.api.entities.builtin.provider.message as pm
 
 from .. import requester, reasoning
+from ..request_policy import current_call
+from ..http_pool import RequesterHTTPPool, requester_http_client
+from ..errors import RequesterError
 from ..codex_auth import BASE_URL, CodexAuth, LOGIN_REQUIRED
 from ..codex_errors import CodexProviderError
 
@@ -21,7 +25,8 @@ async def sse_events(response):
     """Decode SSE records, including CRLF, comments, and multiline data."""
     data = []
     size = 0
-    async for line in response.aiter_lines():
+    lines = response.aiter_lines()
+    async for line in lines:
         if not line:
             if data:
                 text = '\n'.join(data)
@@ -33,6 +38,18 @@ async def sse_events(response):
                         raise ValueError
                 except ValueError:
                     raise ValueError('Codex returned an invalid stream event') from None
+                if event.get('type') == 'response.completed':
+                    # Finish the HTTP body so a healthy connection can return
+                    # to the pool. Never wait indefinitely for a broken SSE
+                    # server after it has already sent the terminal result.
+                    try:
+                        async with asyncio.timeout(0.05):
+                            async for _ in lines:
+                                pass
+                    except (TimeoutError, httpx.HTTPError):
+                        pass
+                    yield event
+                    return
                 yield event
             data, size = [], 0
         elif line.startswith('data:'):
@@ -85,6 +102,7 @@ def _tool(item):
 def _usage(response):
     usage = response.get('usage') or {}
     return {
+        **usage,
         'prompt_tokens': usage.get('input_tokens', 0),
         'completion_tokens': usage.get('output_tokens', 0),
         'total_tokens': usage.get('total_tokens', usage.get('input_tokens', 0) + usage.get('output_tokens', 0)),
@@ -94,7 +112,10 @@ def _usage(response):
 
 
 class CodexRequester(requester.ProviderAPIRequester):
+    reports_raw_timing = True
+
     async def initialize(self):
+        self._http_pool = RequesterHTTPPool()
         self.auth = CodexAuth(self.ap)
         self.workspace = self.requester_cfg['workspace_uuid']
         self.provider = self.requester_cfg['provider_uuid']
@@ -104,6 +125,8 @@ class CodexRequester(requester.ProviderAPIRequester):
 
     async def aclose(self):
         self._replay.clear()
+        if getattr(self, '_http_pool', None) is not None:
+            await self._http_pool.aclose()
 
     def get_reasoning_capabilities(self, model):
         return {
@@ -184,6 +207,8 @@ class CodexRequester(requester.ProviderAPIRequester):
 
     def _body(self, query, model, messages, funcs, extra_args, tokens):
         args = {**(model.model_entity.extra_args or {}), **(extra_args or {})}
+        args.pop('request_policy', None)
+        args.pop('reasoning_config', None)
         # Never permit credentials, transport overrides, store/history or arbitrary
         # SDK kwargs to be smuggled through model advanced parameters.
         allowed = {'reasoning', 'text', 'parallel_tool_calls', 'tool_choice'}
@@ -235,13 +260,11 @@ class CodexRequester(requester.ProviderAPIRequester):
             'stream': True,
             'include': ['reasoning.encrypted_content'],
         }
-        reasoning_config = getattr(model, 'reasoning_config_override', None)
-        if reasoning_config is None:
-            reasoning_config = getattr(model.model_entity, 'reasoning_config', None)
-        level = reasoning.normalize_reasoning_config(reasoning_config)['level']
+        reasoning_config = reasoning.resolve_reasoning_config(model, extra_args)
+        level = reasoning_config['level']
         if level != 'provider_default':
             reasoning.validate_reasoning_capabilities(
-                {'level': level}, self.get_reasoning_capabilities(model), model.model_entity.name
+                reasoning_config, self.get_reasoning_capabilities(model), model.model_entity.name
             )
             body['reasoning'] = {'effort': level, 'summary': 'auto'}
         if funcs:
@@ -259,13 +282,39 @@ class CodexRequester(requester.ProviderAPIRequester):
 
     async def _events(self, query, model, messages, funcs, extra_args):
         tokens = await self.auth.access(self.workspace, self.provider)
+        state = current_call.get()
+        timeout = 120
+        if state:
+            p = state.policy
+            timeout = httpx.Timeout(
+                connect=p.connect_timeout, read=p.read_timeout, write=p.write_timeout, pool=p.pool_timeout
+            )
         try:
-            async with asyncio.timeout(300), httpx.AsyncClient(timeout=120, follow_redirects=False) as client:
+            async with (
+                asyncio.timeout(None if state else 300),
+                requester_http_client(self) as client,
+            ):
                 for attempt in range(2):
                     body = self._body(query, model, messages, funcs, extra_args, tokens)
-                    async with client.stream(
-                        'POST', BASE_URL + '/responses', json=body, headers=self._headers(tokens, stream=True)
-                    ) as response:
+                    if state:
+                        state.mark('request_prepared_ms')
+                        state.mark('request_started_ms')
+                    async with AsyncExitStack() as stack:
+                        response_context = client.stream(
+                            'POST',
+                            BASE_URL + '/responses',
+                            json=body,
+                            headers=self._headers(tokens, stream=True),
+                            timeout=timeout,
+                        )
+                        if state:
+                            response = await state.wait(
+                                lambda: stack.enter_async_context(response_context),
+                                'first_event',
+                                state.policy.first_event_timeout,
+                            )
+                        else:
+                            response = await stack.enter_async_context(response_context)
                         if response.status_code == 401 and attempt == 0:
                             tokens = await self.auth.access(
                                 self.workspace, self.provider, rejected_token=tokens['access_token']
@@ -273,85 +322,117 @@ class CodexRequester(requester.ProviderAPIRequester):
                             continue
                         if response.status_code != 200:
                             raise await self._response_error(response)
-                        async for event in sse_events(response):
+                        if state:
+                            state.mark('response_open_ms')
+                        events = sse_events(response)
+                        while True:
+                            try:
+                                if state:
+                                    event = await state.wait(
+                                        lambda: anext(events),
+                                        'stream_idle' if state.raw_started else 'first_event',
+                                        state.policy.stream_idle_timeout
+                                        if state.raw_started
+                                        else state.policy.first_event_timeout,
+                                    )
+                                else:
+                                    event = await anext(events)
+                            except StopAsyncIteration:
+                                break
+                            if state:
+                                state.raw_started = True
+                                state.mark('first_event_ms')
+                                kind = event.get('type', '')
+                                if kind == 'response.output_text.delta' and event.get('delta'):
+                                    state.mark('first_content_ms')
+                                elif 'reasoning' in kind and event.get('delta'):
+                                    state.mark('first_reasoning_ms')
                             yield event, tokens
                         return
-        except (httpx.HTTPError, TimeoutError):
-            raise ValueError('ChatGPT Codex network error or timeout. Please retry.') from None
+        except (httpx.HTTPError, TimeoutError) as exc:
+            raise RequesterError(
+                'ChatGPT Codex network error or timeout. Please retry.',
+                code='timeout' if isinstance(exc, (httpx.TimeoutException, TimeoutError)) else 'connection',
+                retryable=True,
+            ) from None
 
     async def _chunks(self, query, model, messages, funcs, extra_args, remove_think, usage_out):
         text = ''
         seen_calls = set()
         output_items = {}
         response_id = None
-        async for event, tokens in self._events(query, model, messages, funcs, extra_args):
-            kind = event.get('type')
-            response = event.get('response') or {}
-            response_id = response.get('id') or response_id
-            if kind in ('error', 'response.failed', 'response.incomplete'):
-                raise CodexProviderError('ChatGPT Codex response failed or was incomplete. Please retry.')
-            if kind == 'response.output_text.delta':
-                delta = event.get('delta', '')
-                text += delta
-                yield pm.MessageChunk(role='assistant', content=delta, resp_message_id=response_id)
-            elif kind in ('response.reasoning_summary_text.delta', 'response.reasoning_text.delta'):
-                if not remove_think:
-                    yield pm.MessageChunk(
-                        role='assistant',
-                        content='',
-                        provider_specific_fields={'reasoning_content': event.get('delta', '')},
-                    )
-            elif kind == 'response.output_item.done':
-                item = event.get('item') or {}
-                output_items[event.get('output_index', len(output_items))] = item
-                if item.get('type') == 'function_call' and item.get('call_id') not in seen_calls:
-                    seen_calls.add(item.get('call_id'))
-                    yield pm.MessageChunk(role='assistant', content='', tool_calls=[_tool(item)])
-            elif kind in ('response.completed', 'response.done'):
-                if response.get('status') not in (None, 'completed'):
-                    raise CodexProviderError('ChatGPT Codex response was not completed')
-                output = response.get('output') or [output_items[k] for k in sorted(output_items)]
-                for item in output:
+        async with aclosing(self._events(query, model, messages, funcs, extra_args)) as events:
+            async for event, tokens in events:
+                kind = event.get('type')
+                response = event.get('response') or {}
+                response_id = response.get('id') or response_id
+                if kind in ('error', 'response.failed', 'response.incomplete'):
+                    raise CodexProviderError('ChatGPT Codex response failed or was incomplete. Please retry.')
+                if kind == 'response.output_text.delta':
+                    delta = event.get('delta', '')
+                    text += delta
+                    yield pm.MessageChunk(role='assistant', content=delta, resp_message_id=response_id)
+                elif kind in ('response.reasoning_summary_text.delta', 'response.reasoning_text.delta'):
+                    if not remove_think:
+                        yield pm.MessageChunk(
+                            role='assistant',
+                            content='',
+                            provider_specific_fields={'reasoning_content': event.get('delta', '')},
+                        )
+                elif kind == 'response.output_item.done':
+                    item = event.get('item') or {}
+                    output_items[event.get('output_index', len(output_items))] = item
                     if item.get('type') == 'function_call' and item.get('call_id') not in seen_calls:
                         seen_calls.add(item.get('call_id'))
                         yield pm.MessageChunk(role='assistant', content='', tool_calls=[_tool(item)])
-                # Some servers send only the terminal output, without text deltas.
-                final_text = ''.join(
-                    p.get('text', '')
-                    for item in output
-                    if item.get('type') == 'message'
-                    for p in item.get('content', [])
-                    if p.get('type') == 'output_text'
-                )
-                if not text and final_text:
-                    text = final_text
-                    yield pm.MessageChunk(role='assistant', content=text, resp_message_id=response_id)
-                usage_out.update(_usage(response))
-                if query is not None:
-                    if query.variables is None:
-                        query.variables = {}
-                    query.variables[requester.STREAM_USAGE_QUERY_VARIABLE] = dict(usage_out)
-                fields = None
-                if query is not None and output:
-                    handle = secrets.token_urlsafe(24)
-                    self._replay[handle] = (self._scope(query, model, tokens), time.time() + 3600, output)
-                    while len(self._replay) > 64:
-                        self._replay.popitem(last=False)
-                    fields = {'codex_replay_id': handle}
-                yield pm.MessageChunk(
-                    role='assistant',
-                    content='',
-                    all_content=text,
-                    is_final=True,
-                    resp_message_id=response_id,
-                    provider_specific_fields=fields,
-                )
-                return
+                elif kind in ('response.completed', 'response.done'):
+                    if response.get('status') not in (None, 'completed'):
+                        raise CodexProviderError('ChatGPT Codex response was not completed')
+                    output = response.get('output') or [output_items[k] for k in sorted(output_items)]
+                    for item in output:
+                        if item.get('type') == 'function_call' and item.get('call_id') not in seen_calls:
+                            seen_calls.add(item.get('call_id'))
+                            yield pm.MessageChunk(role='assistant', content='', tool_calls=[_tool(item)])
+                    # Some servers send only the terminal output, without text deltas.
+                    final_text = ''.join(
+                        p.get('text', '')
+                        for item in output
+                        if item.get('type') == 'message'
+                        for p in item.get('content', [])
+                        if p.get('type') == 'output_text'
+                    )
+                    if not text and final_text:
+                        text = final_text
+                        yield pm.MessageChunk(role='assistant', content=text, resp_message_id=response_id)
+                    usage_out.update(_usage(response))
+                    if current_call.get():
+                        current_call.get().usage = dict(usage_out)
+                    if query is not None:
+                        if query.variables is None:
+                            query.variables = {}
+                        query.variables[requester.STREAM_USAGE_QUERY_VARIABLE] = dict(usage_out)
+                    fields = None
+                    if query is not None and output:
+                        handle = secrets.token_urlsafe(24)
+                        self._replay[handle] = (self._scope(query, model, tokens), time.time() + 3600, output)
+                        while len(self._replay) > 64:
+                            self._replay.popitem(last=False)
+                        fields = {'codex_replay_id': handle}
+                    yield pm.MessageChunk(
+                        role='assistant',
+                        content='',
+                        all_content=text,
+                        is_final=True,
+                        resp_message_id=response_id,
+                        provider_specific_fields=fields,
+                    )
+                    return
         raise CodexProviderError('ChatGPT Codex stream ended before completion. Please retry.')
 
     async def invoke_llm_stream(self, query, model, messages, funcs=None, extra_args=None, remove_think=False):
-        async for chunk in self._chunks(query, model, messages, funcs, extra_args, remove_think, {}):
-            yield chunk
+        async with aclosing(self._chunks(query, model, messages, funcs, extra_args, remove_think, {})) as chunks:
+            async for chunk in chunks:
+                yield chunk
 
     async def invoke_llm(self, query, model, messages, funcs=None, extra_args=None, remove_think=False):
         usage = {}
@@ -376,11 +457,12 @@ class CodexRequester(requester.ProviderAPIRequester):
     async def scan_models(self, api_key=None):
         tokens = await self.auth.access(self.workspace, self.provider)
         try:
-            async with asyncio.timeout(90), httpx.AsyncClient(timeout=30, follow_redirects=False) as client:
+            async with asyncio.timeout(90), requester_http_client(self) as client:
                 for attempt in range(2):
                     response = await client.get(
                         BASE_URL + '/models',
                         params={'client_version': langbot.__version__},
+                        timeout=30,
                         headers=self._headers(tokens),
                     )
                     if response.status_code == 401 and attempt == 0:

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

@@ -3,10 +3,23 @@
 from __future__ import annotations
 
 import typing
+import copy
+import time
+import os
+from collections import OrderedDict
+from contextlib import asynccontextmanager
+from openai import AsyncOpenAI
+from ..http_pool import RequesterHTTPPool
+from ..prompt_caching import apply_prompt_caching
 import asyncio
+import httpx
+from urllib.parse import urlsplit
+from ..request_policy import current_call, RequestPolicy, error_metadata
 
 import litellm
 from litellm import acompletion, aembedding, arerank
+from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
+from litellm.secret_managers.main import get_secret, get_secret_bool
 
 from .. import errors, reasoning, requester
 from ....utils import httpclient
@@ -15,6 +28,37 @@ import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query
 import langbot_plugin.api.entities.builtin.provider.message as provider_message
 
 
+class _LeasedHTTPHandler(AsyncHTTPHandler):
+    """Adapt a leased transport without LiteLLM's hidden connection-error retry."""
+
+    def __init__(self, client, timeout):
+        self.client = client
+        self.timeout = timeout
+        self.event_hooks = None
+        self.client_alias = 'langbot'
+
+    async def post(self, url, *, timeout=None, stream=False, logging_obj=None, **kwargs):
+        if isinstance(kwargs.get('data'), (str, bytes)):
+            kwargs['content'] = kwargs.pop('data')
+        request = self.client.build_request('POST', url, timeout=timeout or self.timeout, **kwargs)
+        response = await self.client.send(request, stream=stream)
+        try:
+            response.raise_for_status()
+        except BaseException:
+            await response.aclose()
+            raise
+        return response
+
+    async def close(self):
+        pass  # The requester's lease owns the transport.
+
+    async def __aexit__(self, *args):
+        pass
+
+    def __del__(self):
+        pass
+
+
 class _ThinkStripState:
     """Drop explicit think blocks, buffering only possible split tag prefixes.
 
@@ -131,6 +175,8 @@ class _ThinkStripState:
 class LiteLLMRequester(requester.ProviderAPIRequester):
     """LiteLLM unified API requester supporting chat, embedding, and rerank."""
 
+    reports_raw_timing = True
+
     _EMBEDDING_MODEL_HINTS = ('embedding', 'embed', 'bge-', 'e5-', 'm3e', 'gte-', 'text-embedding')
     _RERANK_MODEL_HINTS = ('rerank', 're-rank', 're_rank')
     _QWEN_DEDICATED_THINKING_MODELS = frozenset(
@@ -179,9 +225,83 @@ class LiteLLMRequester(requester.ProviderAPIRequester):
 
     async def initialize(self):
         """Initialize LiteLLM client settings."""
-        # LiteLLM doesn't require explicit client initialization
-        # Configuration is passed per-request via litellm params
-        pass
+        self._http_pool = RequesterHTTPPool(
+            follow_redirects=True, drain_sse=self._get_custom_llm_provider() == 'openai'
+        )
+
+    async def aclose(self):
+        if getattr(self, '_http_pool', None) is not None:
+            await self._http_pool.aclose()
+        if hasattr(self, '_metadata_cache'):
+            self._metadata_cache.clear()
+
+    @asynccontextmanager
+    async def _client_args(self, args):
+        pool = getattr(self, '_http_pool', None)
+        provider = args.get('custom_llm_provider') or self._get_custom_llm_provider()
+        if pool is None or provider not in ('openai', 'openrouter') or args.get('client') is not None:
+            yield args
+            return
+        # Preserve explicitly customized LiteLLM transports and TLS settings.
+        if (
+            any(args.get(key) is not None for key in ('shared_session', 'http_client', 'ssl_verify', 'ssl_certificate'))
+            or getattr(litellm, 'aclient_session', None) is not None
+            or getattr(litellm, 'ssl_verify', True) is not True
+            or getattr(litellm, 'ssl_certificate', None)
+            or getattr(litellm, 'force_ipv4', False)
+            or any(name in os.environ for name in ('SSL_VERIFY', 'SSL_CERTIFICATE'))
+        ):
+            yield args
+            return
+        if not args['model'].startswith(f'{provider}/'):
+            yield args  # A caller redirected the model; let LiteLLM resolve its transport.
+            return
+        model_name = args['model'].removeprefix(f'{provider}/')
+        if provider == 'openrouter' and model_name in litellm.open_ai_chat_completion_models:
+            yield args  # Preserve LiteLLM's alternate bare-model routing.
+            return
+        transport_started = time.perf_counter()
+        async with pool.lease() as client:
+            state = current_call.get()
+            if state:
+                state.metrics['transport_setup_ms'] = round(
+                    state.metrics.get('transport_setup_ms', 0) + (time.perf_counter() - transport_started) * 1000, 3
+                )
+            if provider == 'openrouter' or get_secret_bool('EXPERIMENTAL_OPENAI_BASE_LLM_HTTP_HANDLER'):
+                yield {**args, 'client': _LeasedHTTPHandler(client, args.get('timeout', 120))}
+                return
+            # A fresh lightweight SDK wrapper keeps rotating keys, organization
+            # and deadlines request-local; only its HTTP transport is reused.
+            sdk = AsyncOpenAI(
+                api_key=args.get('api_key') or litellm.api_key or litellm.openai_key or get_secret('OPENAI_API_KEY'),
+                base_url=(
+                    args.get('api_base')
+                    or litellm.api_base
+                    or get_secret('OPENAI_BASE_URL')
+                    or get_secret('OPENAI_API_BASE')
+                    or 'https://api.openai.com/v1'
+                ),
+                http_client=client,
+                organization=args.get('organization') or litellm.organization,
+                timeout=args.get('timeout', 120),
+                max_retries=0,
+            )
+            yield {**args, 'client': sdk}
+
+    def _cached_metadata(self, key, compute):
+        if not hasattr(self, '_metadata_cache'):
+            self._metadata_cache = OrderedDict()
+        cache = self._metadata_cache
+        cached = cache.get(key)
+        if cached is not None and time.monotonic() - cached[0] < 300:
+            cache.move_to_end(key)
+            return copy.deepcopy(cached[1])
+        value = compute()
+        cache[key] = (time.monotonic(), copy.deepcopy(value))
+        cache.move_to_end(key)
+        while len(cache) > 128:
+            cache.popitem(last=False)
+        return value
 
     def _build_litellm_model_name(self, model_name: str, custom_llm_provider: str | None = None) -> str:
         """Build LiteLLM model name with provider prefix if needed."""
@@ -197,8 +317,21 @@ class LiteLLMRequester(requester.ProviderAPIRequester):
     def _get_custom_llm_provider(self) -> str | None:
         return self.requester_cfg.get('custom_llm_provider') or None
 
+    def _uses_openrouter_protocol(self) -> bool:
+        # OpenRouter's shipped metadata also serves OpenAI-compatible non-chat
+        # operations. Select gateway chat semantics independently of it.
+        return (
+            self._get_custom_llm_provider() == 'openrouter'
+            or self.requester_cfg.get('requester_name') == 'openrouter-chat-completions'
+        )
+
     def _safe_litellm_bool_helper(self, helper_name: str, model_name: str) -> bool:
         """Call a LiteLLM boolean capability helper without letting metadata gaps fail requests."""
+        return self._cached_metadata(
+            ('bool', helper_name, model_name), lambda: self._uncached_litellm_bool_helper(helper_name, model_name)
+        )
+
+    def _uncached_litellm_bool_helper(self, helper_name, model_name):
         helper = getattr(litellm, helper_name, None)
         if not callable(helper):
             return False
@@ -297,6 +430,9 @@ class LiteLLMRequester(requester.ProviderAPIRequester):
         return None
 
     def _safe_context_length(self, model_name: str) -> int | None:
+        return self._cached_metadata(('model', model_name), lambda: self._uncached_model_info(model_name))
+
+    def _uncached_model_info(self, model_name):
         helper = getattr(litellm, 'get_model_info', None)
         if not callable(helper):
             return self._known_context_length_fallback(model_name)
@@ -561,6 +697,20 @@ class LiteLLMRequester(requester.ProviderAPIRequester):
         }
         if family == 'qwen' and 'disabled' in capabilities['levels'] and 'enabled' not in capabilities['levels']:
             capabilities['legacy_levels'] = ['enabled']
+        if self._uses_openrouter_protocol():
+            capabilities['budget_tokens'] = {'min': 1, 'max': 32768}
+        elif family == 'anthropic' and not self._is_anthropic_adaptive_model(normalized_name):
+            if normalized_name.rsplit('/', 1)[-1].startswith(
+                ('claude-3-7-', 'claude-sonnet-4', 'claude-opus-4', 'claude-haiku-4')
+            ):
+                capabilities['budget_tokens'] = {'min': 1024, 'max': 32768}
+        elif family == 'qwen' and self._supports_qwen_thinking_budget(model_name):
+            capabilities['budget_tokens'] = {'min': 1, 'max': 32768}
+        elif family == 'gemini' and 'gemini-2.5' in normalized_name:
+            capabilities['budget_tokens'] = {
+                'min': 128 if 'pro' in normalized_name else (512 if 'flash-lite' in normalized_name else 1),
+                'max': 32768 if 'pro' in normalized_name else 24576,
+            }
         return capabilities
 
     def _build_reasoning_args(self, model: requester.RuntimeLLMModel) -> dict[str, typing.Any]:
@@ -568,14 +718,21 @@ class LiteLLMRequester(requester.ProviderAPIRequester):
         if level == 'provider_default':
             return {}
 
-        config = {'level': level}
+        config = reasoning.normalize_reasoning_config(self._reasoning_config_value(model))
         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
+            raise errors.RequesterError(str(exc), code='invalid_policy') from exc
 
         family = self._reasoning_family(model.model_entity.name, model)
+        budget = config.get('budget_tokens')
+        if budget is not None:
+            if self._uses_openrouter_protocol():
+                return {'extra_body': {'reasoning': {'max_tokens': budget}}}
+            if family == 'qwen':
+                return {'extra_body': {'enable_thinking': True, 'thinking_budget': budget}}
+            return {'thinking': {'type': 'enabled', 'budget_tokens': budget}}
         if level == 'disabled':
             if family in {'deepseek', 'kimi', 'mimo', 'doubao'}:
                 return {'extra_body': {'thinking': {'type': 'disabled'}}}
@@ -1027,7 +1184,7 @@ class LiteLLMRequester(requester.ProviderAPIRequester):
         if include_retry_params:
             if self.requester_cfg.get('drop_params'):
                 args['drop_params'] = self.requester_cfg['drop_params']
-            if self.requester_cfg.get('num_retries'):
+            if self.requester_cfg.get('num_retries') is not None:
                 args['num_retries'] = self.requester_cfg['num_retries']
             if self.requester_cfg.get('api_version'):
                 args['api_version'] = self.requester_cfg['api_version']
@@ -1035,24 +1192,102 @@ class LiteLLMRequester(requester.ProviderAPIRequester):
 
     def _handle_litellm_error(self, e: Exception) -> None:
         """Convert LiteLLM exceptions to RequesterError. Never returns, always raises."""
-        # Check more specific exceptions first (they inherit from base exceptions)
-        if isinstance(e, litellm.ContextWindowExceededError):
-            raise errors.RequesterError(f'上下文长度超限: {str(e)}')
-        if isinstance(e, litellm.BadRequestError):
-            raise errors.RequesterError(f'请求参数错误: {str(e)}')
-        if isinstance(e, litellm.AuthenticationError):
-            raise errors.RequesterError(f'API key 无效: {str(e)}')
-        if isinstance(e, litellm.NotFoundError):
-            raise errors.RequesterError(f'模型或路径无效: {str(e)}')
-        if isinstance(e, litellm.RateLimitError):
-            raise errors.RequesterError(f'请求过于频繁或余额不足: {str(e)}')
-        if isinstance(e, litellm.Timeout):
-            raise errors.RequesterError(f'请求超时: {str(e)}')
-        if isinstance(e, litellm.APIConnectionError):
-            raise errors.RequesterError(f'连接错误: {str(e)}')
-        if isinstance(e, litellm.APIError):
-            raise errors.RequesterError(f'API 错误: {str(e)}')
-        raise errors.RequesterError(f'未知错误: {str(e)}')
+        if isinstance(e, errors.RequesterError):
+            raise e
+        metadata = error_metadata(e)
+        mappings = (
+            (litellm.ContextWindowExceededError, 'context_length', '上下文长度超限', False),
+            (litellm.AuthenticationError, 'authentication', 'API key 无效', False),
+            (litellm.NotFoundError, 'not_found', '模型或路径无效', False),
+            (litellm.RateLimitError, 'rate_limit', '请求过于频繁或余额不足', True),
+            (litellm.Timeout, 'timeout', '请求超时', True),
+            (litellm.APIConnectionError, 'connection', '连接错误', True),
+            (litellm.BadRequestError, 'bad_request', '请求参数错误', False),
+        )
+        for exception_type, code, message, retryable in mappings:
+            if isinstance(e, exception_type):
+                body = getattr(e, 'body', None)
+                body = body if isinstance(body, dict) else {}
+                upstream_error = body.get('error', body)
+                upstream_code = upstream_error.get('code') if isinstance(upstream_error, dict) else None
+                if upstream_code in ('insufficient_quota', 'quota_exceeded', 'billing_hard_limit_reached'):
+                    code, retryable = 'quota_exceeded', False
+                raise errors.RequesterError(message, code=code, retryable=retryable, **metadata) from e
+        status = metadata['status_code']
+        raise errors.RequesterError(
+            'API 错误' if isinstance(e, litellm.APIError) else '未知错误',
+            code='upstream_error',
+            retryable=status in (408, 409, 429, 500, 502, 503, 504),
+            **metadata,
+        ) from e
+
+    def _apply_transport(self, args, model, extra_args):
+        state = current_call.get()
+        policy = (
+            state.policy
+            if state
+            else RequestPolicy.resolve(self.requester_cfg, model.model_entity.extra_args, extra_args)
+        )
+        args.pop('request_policy', None)
+        if state:
+            # The Host owns retries; do not multiply attempts in nested SDKs.
+            args['num_retries'] = 0
+            args['max_retries'] = 0
+            args['timeout'] = httpx.Timeout(
+                connect=policy.connect_timeout,
+                read=policy.read_timeout,
+                write=policy.write_timeout,
+                pool=policy.pool_timeout,
+            )
+            state.mark('request_prepared_ms')
+        base = args.get('api_base', '')
+        is_openrouter = urlsplit(base).hostname == 'openrouter.ai' or (
+            not base and self._get_custom_llm_provider() == 'openrouter'
+        )
+        headers = dict(args.get('extra_headers') or {})
+        existing = {key.lower() for key in headers}
+        if policy.attribution and is_openrouter:
+            if 'http-referer' not in existing:
+                headers['HTTP-Referer'] = 'https://langbot.app'
+            if not {'x-title', 'x-openrouter-title'} & existing:
+                headers['X-OpenRouter-Title'] = 'LangBot'
+        conversation_header = policy.conversation_header or ('x-session-id' if is_openrouter else None)
+        if state and state.conversation_id and conversation_header and conversation_header not in existing:
+            headers[conversation_header] = state.conversation_id
+        if headers:
+            args['extra_headers'] = headers
+
+    async def _completion(self, args):
+        state = current_call.get()
+        if state:
+            state.mark('request_started_ms')
+            result = await state.wait(
+                lambda: acompletion(**args),
+                'first_event' if args.get('stream') else 'response',
+                state.policy.first_event_timeout if args.get('stream') else state.policy.total_timeout,
+            )
+            state.mark('response_open_ms')
+            return result
+        return await acompletion(**args)
+
+    async def _raw_chunks(self, response):
+        state = current_call.get()
+        iterator = response.__aiter__()
+        while True:
+            try:
+                if state:
+                    chunk = await state.wait(
+                        lambda: anext(iterator),
+                        'stream_idle' if state.raw_started else 'first_event',
+                        state.policy.stream_idle_timeout if state.raw_started else state.policy.first_event_timeout,
+                    )
+                    state.raw_started = True
+                    state.mark('first_event_ms')
+                else:
+                    chunk = await anext(iterator)
+            except StopAsyncIteration:
+                return
+            yield chunk
 
     async def _build_completion_args(
         self,
@@ -1063,14 +1298,31 @@ class LiteLLMRequester(requester.ProviderAPIRequester):
         stream: bool = False,
     ) -> dict:
         """Build common completion arguments for invoke_llm and invoke_llm_stream."""
+        preparation_start = time.perf_counter()
+        state = current_call.get()
+        cache_preparation = state is not None and state.policy.retries > 0
+        scoped_model = copy.copy(model)
+        try:
+            scoped_model.reasoning_config_override = reasoning.resolve_reasoning_config(model, extra_args)
+        except ValueError as exc:
+            raise errors.RequesterError(str(exc), code='invalid_policy') from exc
+        model = scoped_model
         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)
+        message_key = ('messages', id(messages), reasoning_family, reasoning_level != 'disabled')
+        if cache_preparation and message_key in state.prepared:
+            req_messages = copy.deepcopy(state.prepared[message_key])
+            state.metrics['preparation_cache_hits'] = state.metrics.get('preparation_cache_hits', 0) + 1
+        else:
+            req_messages = self._convert_messages(
+                messages, reasoning_family=reasoning_family, include_reasoning_context=reasoning_level != 'disabled'
+            )
+            if cache_preparation:
+                state.prepared[message_key] = copy.deepcopy(req_messages)
+        # OpenRouter shares requester metadata with OpenAI-compatible non-chat
+        # operations, but chat needs its own adapter to preserve cache markers.
+        chat_provider = 'openrouter' if self._uses_openrouter_protocol() else self._get_custom_llm_provider()
+        model_name = self._build_litellm_model_name(model.model_entity.name, chat_provider)
         api_key = model.provider.token_mgr.get_token()
 
         args = {
@@ -1083,10 +1335,14 @@ class LiteLLMRequester(requester.ProviderAPIRequester):
             args['stream_options'] = {'include_usage': True}
         self._build_common_args(args)
 
+        if chat_provider == 'openrouter':
+            args['custom_llm_provider'] = chat_provider
+
         # Apply model-level extra_args first, then call-level extra_args
         if model.model_entity.extra_args:
             args.update(model.model_entity.extra_args)
         args.update(extra_args)
+        args.pop('reasoning_config', None)
 
         reasoning_args = self._build_reasoning_args(model)
         if reasoning_args:
@@ -1111,8 +1367,23 @@ class LiteLLMRequester(requester.ProviderAPIRequester):
                     raise errors.RequesterError('allowed_openai_params must be an array')
                 args['allowed_openai_params'] = list(dict.fromkeys([*allowed_openai_params, 'reasoning_effort']))
 
+        budget = model.reasoning_config_override.get('budget_tokens')
+        if budget is not None and reasoning_family == 'anthropic' and not self._uses_openrouter_protocol():
+            output_limit = args.get('max_completion_tokens', args.get('max_tokens'))
+            if type(output_limit) is not int or output_limit <= budget:
+                raise errors.RequesterError(
+                    'Anthropic budget_tokens requires explicit max_tokens greater than the budget',
+                    code='invalid_policy',
+                )
+
         if funcs:
-            tools = await self.ap.tool_mgr.generate_tools_for_openai(funcs)
+            tool_key = ('tools', id(funcs))
+            if cache_preparation and tool_key in state.prepared:
+                tools = copy.deepcopy(state.prepared[tool_key])
+            else:
+                tools = copy.deepcopy(await self.ap.tool_mgr.generate_tools_for_openai(funcs))
+                if cache_preparation:
+                    state.prepared[tool_key] = copy.deepcopy(tools)
             if tools:
                 args['tools'] = tools
                 args.setdefault('tool_choice', 'auto')
@@ -1121,6 +1392,15 @@ class LiteLLMRequester(requester.ProviderAPIRequester):
                 # and imports proxy-only dependencies such as FastAPI.
                 args['_skip_mcp_handler'] = True
 
+        cache_family = reasoning_family
+        if cache_family == 'openrouter':
+            cache_family = self._infer_reasoning_family_from_model_name(model.model_entity.name) or cache_family
+        apply_prompt_caching(args, state, provider=self._get_custom_llm_provider(), family=cache_family)
+        self._apply_transport(args, model, extra_args)
+        if state:
+            state.metrics['preparation_ms'] = round(
+                state.metrics.get('preparation_ms', 0) + (time.perf_counter() - preparation_start) * 1000, 3
+            )
         return args
 
     async def count_tokens(
@@ -1142,7 +1422,7 @@ class LiteLLMRequester(requester.ProviderAPIRequester):
             count_args['tool_choice'] = args['tool_choice']
 
         try:
-            tokens = litellm.token_counter(**count_args)
+            tokens = await asyncio.to_thread(litellm.token_counter, **count_args)
         except Exception as e:
             self._handle_litellm_error(e)
 
@@ -1162,38 +1442,41 @@ class LiteLLMRequester(requester.ProviderAPIRequester):
         """Invoke LLM and return message with usage info."""
         args = await self._build_completion_args(model, messages, funcs, extra_args, stream=False)
 
-        try:
-            response = await acompletion(**args)
+        async with self._client_args(args) as args:
+            try:
+                response = await self._completion(args)
 
-            message_data = response.choices[0].message.model_dump()
-            if 'role' not in message_data or message_data['role'] is None:
-                message_data['role'] = 'assistant'
+                message_data = response.choices[0].message.model_dump()
+                if 'role' not in message_data or message_data['role'] is None:
+                    message_data['role'] = 'assistant'
 
-            content = message_data.get('content', '')
-            reasoning_content = message_data.get('reasoning_content', None)
-            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)
+                content = message_data.get('content', '')
+                reasoning_content = message_data.get('reasoning_content', None)
+                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']
+                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)
+                message = provider_message.Message(**message_data)
+                usage_info = self._extract_usage(response)
+                if current_call.get():
+                    current_call.get().usage = usage_info
 
-            return message, usage_info
+                return message, usage_info
 
-        except Exception as e:
-            self._handle_litellm_error(e)
+            except Exception as e:
+                self._handle_litellm_error(e)
 
     async def invoke_llm_stream(
         self,
@@ -1216,141 +1499,149 @@ class LiteLLMRequester(requester.ProviderAPIRequester):
         thinking_blocks_state: list[dict[str, typing.Any]] = []
         response = None
 
-        try:
-            response = await acompletion(**args)
-            async for chunk in response:
-                # Capture usage whenever a chunk carries it.
-                #
-                # Important: many OpenAI-compatible gateways (e.g. new-api) and
-                # providers send the final usage payload in a chunk that STILL
-                # contains a (empty-delta) choice, not an empty `choices` list.
-                # The previous implementation only captured usage when `choices`
-                # was empty, so streamed calls always recorded 0 tokens.
-                # We therefore capture usage independently of `choices`, and then
-                # fall through to also process any content this chunk may carry.
-                if getattr(chunk, 'usage', None):
-                    usage_info = self._normalize_usage(chunk.usage)
-                    if query is not None:
-                        if query.variables is None:
-                            query.variables = {}
-                        query.variables[requester.STREAM_USAGE_QUERY_VARIABLE] = usage_info
+        async with self._client_args(args) as args:
+            try:
+                response = await self._completion(args)
+                async for chunk in self._raw_chunks(response):
+                    # Capture usage whenever a chunk carries it.
+                    #
+                    # Important: many OpenAI-compatible gateways (e.g. new-api) and
+                    # providers send the final usage payload in a chunk that STILL
+                    # contains a (empty-delta) choice, not an empty `choices` list.
+                    # The previous implementation only captured usage when `choices`
+                    # was empty, so streamed calls always recorded 0 tokens.
+                    # We therefore capture usage independently of `choices`, and then
+                    # fall through to also process any content this chunk may carry.
+                    if getattr(chunk, 'usage', None):
+                        if current_call.get():
+                            current_call.get().usage = self._extract_usage(chunk)
+                        usage_info = self._normalize_usage(chunk.usage)
+                        if query is not None:
+                            if query.variables is None:
+                                query.variables = {}
+                            query.variables[requester.STREAM_USAGE_QUERY_VARIABLE] = usage_info
 
-                if not hasattr(chunk, 'choices') or not chunk.choices:
-                    continue
+                    if not hasattr(chunk, 'choices') or not chunk.choices:
+                        continue
 
-                choice = chunk.choices[0]
-                delta = choice.delta.model_dump() if hasattr(choice, 'delta') else {}
-                finish_reason = getattr(choice, 'finish_reason', None)
+                    choice = chunk.choices[0]
+                    delta = choice.delta.model_dump() if hasattr(choice, 'delta') else {}
+                    finish_reason = getattr(choice, 'finish_reason', None)
 
-                if 'role' in delta and delta['role']:
-                    role = delta['role']
+                    if 'role' in delta and delta['role']:
+                        role = delta['role']
 
-                delta_content = delta.get('content') or ''
-                reasoning_content = delta.get('reasoning_content') or ''
-                provider_fields = dict(delta.get('provider_specific_fields') or {})
-                if finish_reason:
-                    # Preserve an explicit provider stop even when the final delta is empty.
-                    provider_fields['finish_reason'] = finish_reason
-                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
+                    state = current_call.get()
+                    if state and delta.get('content'):
+                        state.mark('first_content_ms')
+                    if state and (delta.get('reasoning_content') or delta.get('thinking_blocks')):
+                        state.mark('first_reasoning_ms')
+                    delta_content = delta.get('content') or ''
+                    reasoning_content = delta.get('reasoning_content') or ''
+                    provider_fields = dict(delta.get('provider_specific_fields') or {})
+                    if finish_reason:
+                        # Preserve an explicit provider stop even when the final delta is empty.
+                        provider_fields['finish_reason'] = finish_reason
+                    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:
-                        delta_content = delta_content or None
-                    else:
-                        # 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
+                    # Handle reasoning_content based on remove_think flag
+                    if reasoning_content:
+                        provider_fields['reasoning_content'] = reasoning_content
+                        if remove_think:
+                            delta_content = delta_content or None
                         else:
-                            delta_content = ''
-                        delta_content += display_reasoning_content
-                        if delta.get('content'):
-                            delta_content += f'\n</think>\n{delta.get("content")}'
-                            reasoning_closed = True
+                            # 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
+                    elif display_reasoning_content:
+                        if remove_think:
+                            delta_content = delta_content or None
                         else:
-                            delta_content = ''
-                        delta_content += display_reasoning_content
-                        if delta.get('content'):
-                            delta_content += f'\n</think>\n{delta.get("content")}'
-                            reasoning_closed = True
+                            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
+                    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 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 think_state is not None and delta_content:
+                        delta_content = think_state.feed(delta_content)
 
-                tool_calls = self._normalize_stream_tool_calls(delta.get('tool_calls'), tool_call_state)
+                    tool_calls = self._normalize_stream_tool_calls(delta.get('tool_calls'), tool_call_state)
 
-                if not delta_content and not tool_calls and not provider_fields and not finish_reason:
+                    if not delta_content and not tool_calls and not provider_fields and not finish_reason:
+                        chunk_idx += 1
+                        continue
+
+                    chunk_data: dict[str, typing.Any] = {
+                        'role': role,
+                        'content': delta_content if delta_content else None,
+                        'tool_calls': tool_calls,
+                        'is_final': bool(finish_reason),
+                    }
+
+                    # Preserve provider_specific_fields from delta (e.g., Gemini thought_signatures)
+                    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
-                    continue
 
-                chunk_data: dict[str, typing.Any] = {
-                    'role': role,
-                    'content': delta_content if delta_content else None,
-                    'tool_calls': tool_calls,
-                    'is_final': bool(finish_reason),
-                }
-
-                # Preserve provider_specific_fields from delta (e.g., Gemini thought_signatures)
-                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:
+                if reasoning_started and not reasoning_closed:
                     yield provider_message.MessageChunk(
                         role=role,
-                        content=pending_content,
+                        content='\n</think>\n',
                         is_final=True,
                     )
 
-        except Exception as e:
-            self._handle_litellm_error(e)
+                if think_state is not None:
+                    pending_content = think_state.flush()
+                    if pending_content:
+                        yield provider_message.MessageChunk(
+                            role=role,
+                            content=pending_content,
+                            is_final=True,
+                        )
 
-        finally:
-            # Closing a downstream async generator must release the provider's
-            # HTTP stream immediately, including cancellation while at yield.
-            close = getattr(response, 'aclose', None)
-            if callable(close):
-                try:
-                    await asyncio.wait_for(close(), timeout=5.0)
-                except Exception as exc:
-                    self.ap.logger.debug(f'Failed to close LLM stream: {type(exc).__name__}')
+            except Exception as e:
+                self._handle_litellm_error(e)
+
+            finally:
+                # Closing a downstream async generator must release the provider's
+                # HTTP stream immediately, including cancellation while at yield.
+                close = getattr(response, 'aclose', None)
+                if callable(close):
+                    try:
+                        await asyncio.wait_for(close(), timeout=5.0)
+                    except Exception as exc:
+                        self.ap.logger.debug(f'Failed to close LLM stream: {type(exc).__name__}')
 
     async def invoke_embedding(
         self,

src/langbot/templates/config.yaml

@@ -458,3 +458,20 @@ space:
     # Complete execution records are batched as {"records": [...]}; records are
     # removed from the buffer only once the sender accepted them.
     telemetry_flush_seconds: 180
+
+# Host-owned LLM policy. Models/calls may override through extra_args.request_policy.
+llm_request_policy:
+    retries: 0
+    total_timeout: 300
+    first_event_timeout: 120
+    stream_idle_timeout: 120
+    connect_timeout: 10
+    read_timeout: 120
+    write_timeout: 30
+    pool_timeout: 10
+    backoff_base: 0.5
+    backoff_max: 8
+    attribution: true
+    # Optional: x-session-id, x-litellm-session-id, x-langbot-conversation-id.
+    # OpenRouter uses x-session-id automatically; other providers default to none.
+    conversation_header: null