Merge pull request #3068 from AaronZ345/aaron/feat-openai-chat-api

feat(web): add OpenAI-compatible chat API

agent/chat/service.py

@@ -34,6 +34,7 @@ class ChatService:
         send_chunk_fn: Callable[[dict], None],
         channel_type: str = "",
         agent_id: str = None,
+        request_id: str = None,  # noqa: RUF013
     ):
         """
         Run the agent for *query* and stream results back via *send_chunk_fn*.
@@ -46,6 +47,7 @@ class ChatService:
         :param send_chunk_fn: callable(chunk_data: dict) to send a streaming chunk
         :param channel_type: source channel (e.g. "web", "feishu") for persistence
         :param agent_id: selected agent profile; defaults to the configured default
+        :param request_id: per-request cancellation key; defaults to session scope
         """
         resolved_agent_id = self.agent_bridge._resolve_agent_id(agent_id)
         agent = self.agent_bridge.get_agent(
@@ -250,11 +252,11 @@ class ChatService:
         from agent.protocol.agent_stream import AgentStreamExecutor
 
         # Register a cancel token so /cancel can abort this in-flight run.
-        # IM channels key on session_id (no per-turn request_id here).
+        # API calls can key by request; IM channels remain session scoped.
         from agent.protocol import get_cancel_registry, get_steer_registry
         registry = get_cancel_registry()
         steer_registry = get_steer_registry()
-        cancel_key = (
+        scoped_session_key = (
             self.agent_bridge._cancel_key(
                 resolved_agent_id,
                 session_id,
@@ -263,17 +265,24 @@ class ChatService:
             if session_id
             else None
         )
+        cancel_key = (
+            self.agent_bridge._cancel_key(
+                resolved_agent_id,
+                request_id,
+                self.agent_bridge.agent_registry.default_agent_id,
+            )
+            if request_id
+            else scoped_session_key
+        )
         # Both the token and its session grouping are namespaced: two Agents
         # serving the same session id must not cancel or steer each other.
         cancel_event = (
-            registry.register(cancel_key, session_id=cancel_key)
+            registry.register(cancel_key, session_id=scoped_session_key)
             if cancel_key
             else None
         )
         steer_inbox = (
-            steer_registry.register(cancel_key)
-            if cancel_key
-            else None
+            steer_registry.register(scoped_session_key) if scoped_session_key else None
         )
 
         executor = AgentStreamExecutor(
@@ -290,6 +299,12 @@ class ChatService:
         )
 
         try:
+            if cancel_event is not None and cancel_event.is_set():
+                logger.info(
+                    f"[ChatService] Skipping pre-cancelled run: agent={resolved_agent_id}, "
+                    f"session={session_id}"
+                )
+                return
             response = executor.run_stream(query)
         except Exception:
             # If executor cleared messages (context overflow), sync back
@@ -302,13 +317,13 @@ class ChatService:
             # Clear the mid-run flag so idle scans can review this session again.
             self._mark_run_active(agent, False)
             # Release cancel token to keep the registry bounded.
-            if session_id:
+            if cancel_key:
                 try:
                     registry.unregister(cancel_key)
                 except Exception:
                     pass
-            if cancel_key and steer_inbox is not None:
-                steer_registry.unregister(cancel_key, steer_inbox)
+            if scoped_session_key and steer_inbox is not None:
+                steer_registry.unregister(scoped_session_key, steer_inbox)
 
         # A run that ends without a closing turn_end (e.g. the last turn had no
         # tool calls to flush) must still deliver whatever the send tool uploaded.

channel/web/openai_api.py

@@ -0,0 +1,508 @@
+"""OpenAI-compatible HTTP adapter for CowAgent chat completions."""
+
+from __future__ import annotations
+
+import hmac
+import json
+import queue
+import threading
+import time
+import uuid
+from typing import Callable, Iterator
+
+from common.log import logger
+from config import conf
+
+
+class OpenAIAPIError(Exception):
+    """A public API error with an explicit HTTP status."""
+
+    def __init__(self, status_code: int, message: str, code: str):
+        super().__init__(message)
+        self.status_code = status_code
+        self.message = message
+        self.code = code
+
+
+_SESSION_LOCKS = tuple(threading.Lock() for _ in range(64))
+_STREAM_END = object()
+_STREAM_ERROR = object()
+_FIRST_EVENT_TIMEOUT_SECONDS = 30
+
+
+def _authenticate(authorization: str, external_api_token: str) -> None:
+    token = str(external_api_token or "")
+    if not token:
+        raise OpenAIAPIError(
+            503,
+            "The external chat completions API is disabled.",
+            "api_disabled",
+        )
+    scheme, separator, credential = str(authorization or "").partition(" ")
+    if (
+        not separator
+        or scheme.lower() != "bearer"
+        or not credential
+        or not hmac.compare_digest(credential.strip(), token)
+    ):
+        raise OpenAIAPIError(
+            401, "Invalid authentication credentials.", "invalid_api_key"
+        )
+
+
+def _request_values(payload: dict, completion_id: str):
+    if not isinstance(payload, dict):
+        raise OpenAIAPIError(
+            400, "Request body must be a JSON object.", "invalid_request"
+        )
+
+    model = payload.get("model")
+    if not isinstance(model, str) or not model.strip():
+        raise OpenAIAPIError(
+            400, "'model' must be a non-empty string.", "invalid_request"
+        )
+
+    messages = payload.get("messages")
+    if not isinstance(messages, list) or not messages:
+        raise OpenAIAPIError(
+            400, "'messages' must be a non-empty array.", "invalid_request"
+        )
+
+    query = None
+    for message in messages:
+        if not isinstance(message, dict):
+            raise OpenAIAPIError(
+                400, "Each message must be an object.", "invalid_request"
+            )
+        role = message.get("role")
+        content = message.get("content")
+        if role not in ("system", "user", "assistant"):
+            raise OpenAIAPIError(
+                400, "Only text chat messages are supported.", "invalid_request"
+            )
+        if not isinstance(content, str):
+            raise OpenAIAPIError(
+                400, "Message content must be a string.", "invalid_request"
+            )
+        if role == "user" and content.strip():
+            query = content
+    if query is None:
+        raise OpenAIAPIError(
+            400,
+            "'messages' must contain a non-empty user message.",
+            "invalid_request",
+        )
+
+    conversation_id = payload.get("conversation_id")
+    user = payload.get("user")
+    if conversation_id is not None and (
+        not isinstance(conversation_id, str) or not conversation_id.strip()
+    ):
+        raise OpenAIAPIError(
+            400, "'conversation_id' must be a non-empty string.", "invalid_request"
+        )
+    if user is not None and (not isinstance(user, str) or not user.strip()):
+        raise OpenAIAPIError(
+            400, "'user' must be a non-empty string.", "invalid_request"
+        )
+
+    if conversation_id:
+        session_id = f"openai:conversation:{conversation_id.strip()}"
+    elif user:
+        session_id = f"openai:user:{user.strip()}"
+    else:
+        session_id = f"openai:request:{completion_id}"
+
+    stream = payload.get("stream", False)
+    if not isinstance(stream, bool):
+        raise OpenAIAPIError(400, "'stream' must be a boolean.", "invalid_request")
+    return model.strip(), query, session_id, stream
+
+
+def _base_chunk(
+    completion_id: str,
+    created: int,
+    model: str,
+    delta: dict,
+    finish_reason=None,
+    cow_event: dict | None = None,
+) -> dict:
+    payload = {
+        "id": completion_id,
+        "object": "chat.completion.chunk",
+        "created": created,
+        "model": model,
+        "choices": [
+            {
+                "index": 0,
+                "delta": delta,
+                "finish_reason": finish_reason,
+            }
+        ],
+    }
+    if cow_event is not None:
+        payload["cow_event"] = cow_event
+    return payload
+
+
+def _tool_events(chunk: dict) -> list[dict]:
+    chunk_type = chunk.get("chunk_type")
+    if chunk_type == "tool_start":
+        return [
+            {
+                "type": "tool_start",
+                "id": chunk.get("tool_id"),
+                "name": chunk.get("tool"),
+                "arguments": chunk.get("arguments") or {},
+            }
+        ]
+    if chunk_type == "tool_calls":
+        return [
+            {
+                "type": "tool_result",
+                "id": item.get("id"),
+                "name": item.get("name"),
+                "arguments": item.get("arguments") or {},
+                "result": item.get("result", ""),
+                "status": item.get("status"),
+                "elapsed": item.get("elapsed"),
+            }
+            for item in chunk.get("tool_calls") or []
+        ]
+    if chunk_type in ("subagent_step", "artifact"):
+        event = dict(chunk)
+        event["type"] = event.pop("chunk_type")
+        return [event]
+    return []
+
+
+def _sse_frame(payload: dict) -> str:
+    return f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"
+
+
+def _request_cancel_scope(
+    request_id: str,
+    session_id: str | None = None,
+    agent_id: str | None = None,
+) -> tuple[str, str | None]:
+    """Return the request key and session group used by ChatService."""
+    from bridge.bridge import Bridge
+
+    agent_bridge = Bridge().get_agent_bridge()
+    resolved_agent_id = agent_bridge._resolve_agent_id(agent_id)
+    default_agent_id = agent_bridge.agent_registry.default_agent_id
+    request_key = agent_bridge._cancel_key(
+        resolved_agent_id, request_id, default_agent_id
+    )
+    session_key = (
+        agent_bridge._cancel_key(resolved_agent_id, session_id, default_agent_id)
+        if session_id
+        else None
+    )
+    return request_key, session_key
+
+
+def _request_cancel_key(request_id: str, agent_id: str | None = None) -> str:
+    """Return the Agent-scoped cancellation key for one request."""
+    return _request_cancel_scope(request_id, agent_id=agent_id)[0]
+
+
+def _cancel_agent_request(request_id: str, agent_id: str | None = None) -> bool:
+    """Cancel one request without affecting newer runs in the same session."""
+    from agent.protocol import get_cancel_registry
+
+    return get_cancel_registry().cancel_request(
+        _request_cancel_key(request_id, agent_id)
+    )
+
+
+def _stream_completion(
+    run_chat: Callable,
+    query: str,
+    session_id: str,
+    completion_id: str,
+    created: int,
+    model: str,
+) -> Iterator[str]:
+    from agent.protocol import get_cancel_registry
+
+    registry = get_cancel_registry()
+    cancel_key, scoped_session_key = _request_cancel_scope(completion_id, session_id)
+    registry.register(cancel_key, session_id=scoped_session_key)
+
+    output = queue.Queue(maxsize=256)
+    closed = threading.Event()
+
+    def publish(item) -> None:
+        while not closed.is_set():
+            try:
+                output.put(item, timeout=0.1)
+                return
+            except queue.Full:
+                continue
+
+    def send_chunk(chunk: dict) -> None:
+        chunk_type = chunk.get("chunk_type")
+        if chunk_type == "content" and chunk.get("delta"):
+            publish(
+                _base_chunk(completion_id, created, model, {"content": chunk["delta"]})
+            )
+        elif chunk_type == "reasoning" and chunk.get("delta"):
+            publish(
+                _base_chunk(
+                    completion_id,
+                    created,
+                    model,
+                    {"reasoning_content": chunk["delta"]},
+                    cow_event={"type": "reasoning", "delta": chunk["delta"]},
+                )
+            )
+        else:
+            for event in _tool_events(chunk):
+                publish(_base_chunk(completion_id, created, model, {}, cow_event=event))
+
+    def execute() -> None:
+        try:
+            lock = _SESSION_LOCKS[hash(session_id) % len(_SESSION_LOCKS)]
+            with lock:
+                if closed.is_set():
+                    return
+                run_chat(
+                    query,
+                    session_id,
+                    send_chunk,
+                    channel_type="openai_api",
+                    agent_id=None,
+                    request_id=completion_id,
+                )
+        except Exception:  # noqa: BLE001 - worker boundary becomes an SSE error
+            logger.exception("[OpenAI API] Chat completion failed")
+            publish(_STREAM_ERROR)
+        finally:
+            try:
+                publish(_STREAM_END)
+            finally:
+                registry.unregister(cancel_key)
+
+    worker = threading.Thread(
+        target=execute, name="openai-chat-completion", daemon=True
+    )
+    try:
+        worker.start()
+    except Exception:
+        registry.unregister(cancel_key)
+        raise
+    try:
+        first_item = output.get(timeout=_FIRST_EVENT_TIMEOUT_SECONDS)
+    except queue.Empty as error:
+        closed.set()
+        _cancel_agent_request(completion_id)
+        raise OpenAIAPIError(
+            500, "CowAgent timed out before producing a response.", "timeout"
+        ) from error
+    if first_item is _STREAM_ERROR:
+        closed.set()
+        raise OpenAIAPIError(
+            500, "CowAgent failed to complete the request.", "internal_error"
+        )
+
+    def frames() -> Iterator[str]:
+        finish_reason = "stop"
+        completed = False
+        item = first_item
+        try:
+            yield _sse_frame(
+                _base_chunk(completion_id, created, model, {"role": "assistant"})
+            )
+            while item is not _STREAM_END:
+                if item is _STREAM_ERROR:
+                    finish_reason = "error"
+                    yield _sse_frame(
+                        _base_chunk(
+                            completion_id,
+                            created,
+                            model,
+                            {},
+                            cow_event={
+                                "type": "error",
+                                "message": "CowAgent failed to complete the request.",
+                            },
+                        )
+                    )
+                else:
+                    yield _sse_frame(item)
+                item = output.get()
+            completed = True
+            yield _sse_frame(
+                _base_chunk(
+                    completion_id,
+                    created,
+                    model,
+                    {},
+                    finish_reason=finish_reason,
+                )
+            )
+            yield "data: [DONE]\n\n"
+        finally:
+            closed.set()
+            if not completed:
+                _cancel_agent_request(completion_id)
+
+    return frames()
+
+
+def _non_stream_completion(
+    run_chat: Callable,
+    query: str,
+    session_id: str,
+    completion_id: str,
+    created: int,
+    model: str,
+) -> dict:
+    content = []
+    reasoning = []
+    tool_trace = []
+
+    def send_chunk(chunk: dict) -> None:
+        chunk_type = chunk.get("chunk_type")
+        if chunk_type == "content":
+            content.append(chunk.get("delta") or "")
+        elif chunk_type == "reasoning":
+            reasoning.append(chunk.get("delta") or "")
+        else:
+            tool_trace.extend(_tool_events(chunk))
+
+    try:
+        lock = _SESSION_LOCKS[hash(session_id) % len(_SESSION_LOCKS)]
+        with lock:
+            run_chat(
+                query,
+                session_id,
+                send_chunk,
+                channel_type="openai_api",
+                agent_id=None,
+                request_id=completion_id,
+            )
+    except Exception as error:
+        logger.exception("[OpenAI API] Chat completion failed")
+        raise OpenAIAPIError(
+            500, "CowAgent failed to complete the request.", "internal_error"
+        ) from error
+
+    message = {"role": "assistant", "content": "".join(content)}
+    if reasoning:
+        message["reasoning_content"] = "".join(reasoning)
+    if tool_trace:
+        message["tool_trace"] = tool_trace
+    return {
+        "id": completion_id,
+        "object": "chat.completion",
+        "created": created,
+        "model": model,
+        "choices": [
+            {
+                "index": 0,
+                "message": message,
+                "finish_reason": "stop",
+            }
+        ],
+    }
+
+
+def _encode_stream(stream: Iterator[str]) -> Iterator[bytes]:
+    """Encode SSE frames while propagating client disconnects to the source."""
+    try:
+        for frame in stream:
+            yield frame.encode("utf-8")
+    finally:
+        close = getattr(stream, "close", None)
+        if close is not None:
+            close()
+
+
+def handle_chat_completions(
+    payload: dict,
+    authorization: str,
+    external_api_token: str,
+    run_chat: Callable,
+    created: int | None = None,
+    completion_id: str | None = None,
+):
+    """Validate one request and return a completion dict or SSE iterator."""
+    _authenticate(authorization, external_api_token)
+    completion_id = completion_id or f"chatcmpl-{uuid.uuid4().hex}"
+    created = int(time.time()) if created is None else int(created)
+    model, query, session_id, stream = _request_values(payload, completion_id)
+    if stream:
+        return _stream_completion(
+            run_chat, query, session_id, completion_id, created, model
+        )
+    return _non_stream_completion(
+        run_chat, query, session_id, completion_id, created, model
+    )
+
+
+def _run_chat_service(*args, **kwargs):
+    from agent.chat.service import ChatService
+    from bridge.bridge import Bridge
+
+    return ChatService(Bridge().get_agent_bridge()).run(*args, **kwargs)
+
+
+def _error_body(error: OpenAIAPIError) -> str:
+    return json.dumps(
+        {
+            "error": {
+                "message": error.message,
+                "type": (
+                    "invalid_request_error" if error.status_code == 400 else "api_error"
+                ),
+                "code": error.code,
+            }
+        },
+        ensure_ascii=False,
+    )
+
+
+class OpenAIChatCompletionsHandler:
+    """web.py handler for ``POST /v1/chat/completions``."""
+
+    def POST(self):
+        import web
+
+        try:
+            raw_body = web.data()
+            try:
+                payload = json.loads(raw_body) if raw_body else {}
+            except (TypeError, ValueError) as error:
+                raise OpenAIAPIError(
+                    400, "Request body must be valid JSON.", "invalid_json"
+                ) from error
+
+            result = handle_chat_completions(
+                payload,
+                authorization=web.ctx.env.get("HTTP_AUTHORIZATION", ""),
+                external_api_token=conf().get("external_api_token", ""),
+                run_chat=_run_chat_service,
+            )
+        except OpenAIAPIError as error:
+            statuses = {
+                400: "400 Bad Request",
+                401: "401 Unauthorized",
+                500: "500 Internal Server Error",
+                503: "503 Service Unavailable",
+            }
+            raise web.HTTPError(
+                statuses[error.status_code],
+                {"Content-Type": "application/json; charset=utf-8"},
+                _error_body(error),
+            )
+
+        if isinstance(result, dict):
+            web.header("Content-Type", "application/json; charset=utf-8")
+            return json.dumps(result, ensure_ascii=False)
+
+        web.header("Content-Type", "text/event-stream; charset=utf-8")
+        web.header("Cache-Control", "no-cache")
+        web.header("X-Accel-Buffering", "no")
+        return _encode_stream(result)

channel/web/web_channel.py

@@ -42,6 +42,7 @@ from agent.permission import (
     global_mode as permission_global_mode,
     normalize_mode as permission_normalize_mode,
 )
+from channel.web.openai_api import OpenAIChatCompletionsHandler
 
 IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".gif", ".webp", ".bmp", ".svg"}
 VIDEO_EXTENSIONS = {".mp4", ".webm", ".avi", ".mov", ".mkv"}
@@ -2061,6 +2062,7 @@ class WebChannel(ChatChannel):
             '/stream', 'StreamHandler',
             '/cancel', 'CancelHandler',
             '/chat', 'ChatHandler',
+            '/v1/chat/completions', 'OpenAIChatCompletionsHandler',
             '/config', 'ConfigHandler',
             '/api/models', 'ModelsHandler',
             '/api/channels', 'ChannelsHandler',

config-template.json

@@ -34,6 +34,7 @@
   "wecom_bot_secret": "",
   "web_host": "",
   "web_password": "",
+  "external_api_token": "",
   "agent": true,
   "agent_max_context_tokens": 64000,
   "agent_max_context_turns": 30,

docs/channels/web.mdx

@@ -13,6 +13,7 @@ The Web Console is CowAgent's default channel. It runs automatically once starte
   "web_host": "0.0.0.0",
   "web_port": 9899,
   "web_password": "",
+  "external_api_token": "",
   "enable_thinking": false
 }
 ```
@@ -23,6 +24,7 @@ The Web Console is CowAgent's default channel. It runs automatically once starte
 | `web_host` | Web service listen address. Defaults to `127.0.0.1` (local only); set to `0.0.0.0` for public access and configure a password | `""` |
 | `web_port` | Web service listen port | `9899` |
 | `web_password` | Access password. Leave empty to disable password protection; recommended when listening on `0.0.0.0` | `""` |
+| `external_api_token` | Independent Bearer token for the OpenAI-compatible API. Leave empty to disable the API | `""` |
 | `web_session_expire_days` | Login session validity in days | `30` |
 | `web_file_serve_root` | Root directory the web console can directly read/send files from. Defaults to the user home dir and agent workspace only; set to `/` to allow the whole filesystem | `"~"` |
 | `enable_thinking` | Whether to enable deep thinking mode | `false` |
@@ -40,6 +42,86 @@ After starting the project, visit:
   Ensure the server firewall and security group allow the corresponding port.
 </Note>
 
+## OpenAI-Compatible API
+
+Set `external_api_token` to enable `POST /v1/chat/completions`. This token is
+independent from `web_password` and Web Console login sessions.
+
+The first release accepts text messages and uses CowAgent's configured Agent and
+model. The request `model` is required for OpenAI client compatibility and is
+echoed in the response; it does not select a CowAgent model. The latest non-empty
+user message is submitted to the Agent. Set `conversation_id`, or `user` as a
+fallback, to reuse a stable CowAgent session across requests. Requests without
+either field use an isolated session.
+
+Non-streaming request:
+
+```bash
+curl http://localhost:9899/v1/chat/completions \
+  -H "Authorization: Bearer $TOKEN" \
+  -H "Content-Type: application/json" \
+  -d '{
+    "model": "cowagent",
+    "conversation_id": "example-conversation",
+    "messages": [{"role": "user", "content": "Summarize this workspace."}]
+  }'
+```
+
+The standard response is returned in `choices[0].message.content`. CowAgent adds
+`reasoning_content` and `tool_trace` to the message when those traces are
+available.
+
+Streaming request:
+
+```bash
+curl -N http://localhost:9899/v1/chat/completions \
+  -H "Authorization: Bearer $TOKEN" \
+  -H "Content-Type: application/json" \
+  -d '{
+    "model": "cowagent",
+    "stream": true,
+    "user": "example-user",
+    "messages": [{"role": "user", "content": "Inspect the workspace."}]
+  }'
+```
+
+Streaming content uses standard `chat.completion.chunk` objects and
+`choices[0].delta.content`. Reasoning uses the additive
+`choices[0].delta.reasoning_content` field. Reasoning and tool-process chunks
+also include a top-level `cow_event` object. The stream ends with `data: [DONE]`.
+
+OpenAI Python clients can use the endpoint by setting the base URL:
+
+```python
+from openai import OpenAI
+
+client = OpenAI(
+    base_url="http://localhost:9899/v1",
+    api_key="your-external-api-token",
+)
+response = client.chat.completions.create(
+    model="cowagent",
+    messages=[{"role": "user", "content": "Hello from Python."}],
+    extra_body={"conversation_id": "python-example"},
+)
+print(response.choices[0].message.content)
+```
+
+The API returns `400` for invalid requests, `401` for invalid credentials, and
+`503` when `external_api_token` is not configured. Non-streaming Agent failures
+return a structured JSON `500` response.
+
+For streaming requests, CowAgent waits up to 30 seconds for the Agent to produce
+the first event. An Agent failure before the first event returns a structured
+JSON `500` response because the SSE response has not started. If no first event
+arrives within 30 seconds, the API cancels only that request's in-flight Agent and
+returns the same JSON `500` shape with error code `timeout`.
+
+After the SSE response has started, its HTTP status can no longer change.
+A later Agent failure is therefore emitted as a `cow_event.type=error` chunk,
+followed by a terminal chunk with `finish_reason=error` and then
+`data: [DONE]`.
+
 ## Features
 
 ### Chat Interface

tests/test_chat_service_cancel.py

@@ -0,0 +1,159 @@
+import threading
+from types import SimpleNamespace
+from typing import ClassVar
+
+import pytest
+
+from agent.chat.service import ChatService
+
+
+class _RecordingCancelRegistry:
+    def __init__(self):
+        self.calls = []
+        self.events = {}
+
+    def register(self, request_id, session_id=None):
+        self.calls.append(("register", request_id, session_id))
+        event = self.events.get(request_id)
+        if event is None:
+            event = threading.Event()
+            self.events[request_id] = event
+        return event
+
+    def unregister(self, request_id):
+        self.calls.append(("unregister", request_id))
+        self.events.pop(request_id, None)
+
+
+class _RecordingSteerRegistry:
+    def register(self, session_id):
+        return []
+
+    def unregister(self, session_id, inbox):
+        return None
+
+
+class _FakeExecutor:
+    instances: ClassVar[list] = []
+    run_queries: ClassVar[list] = []
+
+    def __init__(self, *, messages, cancel_event, **kwargs):
+        self.messages = list(messages)
+        self.cancel_event = cancel_event
+        self.__class__.instances.append(self)
+
+    def run_stream(self, query):
+        self.__class__.run_queries.append(query)
+        self.messages.append({"role": "user", "content": query})
+        return ""
+
+
+def _service():
+    agent = SimpleNamespace(
+        model=SimpleNamespace(),
+        tools=[],
+        max_steps=1,
+        messages=[],
+        messages_lock=threading.Lock(),
+        workspace_dir=None,
+        get_full_system_prompt=lambda: "system",
+        _execute_post_process_tools=lambda: None,
+    )
+
+    class FakeBridge:
+        agent_registry = SimpleNamespace(default_agent_id="primary")
+
+        @staticmethod
+        def _resolve_agent_id(agent_id=None):
+            return agent_id or "primary"
+
+        @staticmethod
+        def _cancel_key(agent_id, token, default_agent_id):
+            return token if agent_id == default_agent_id else f"{agent_id}::{token}"
+
+        @staticmethod
+        def get_agent(session_id=None, agent_id=None):
+            return agent
+
+    service = ChatService(FakeBridge())
+    service._build_context = lambda *args, **kwargs: {}
+    service._mark_run_active = lambda *args, **kwargs: None
+    service._note_evolution_turn = lambda *args, **kwargs: None
+    return service
+
+
+@pytest.fixture
+def cancel_runtime(monkeypatch):
+    cancel_registry = _RecordingCancelRegistry()
+    _FakeExecutor.instances = []
+    _FakeExecutor.run_queries = []
+    monkeypatch.setattr("agent.protocol.get_cancel_registry", lambda: cancel_registry)
+    monkeypatch.setattr(
+        "agent.protocol.get_steer_registry", lambda: _RecordingSteerRegistry()
+    )
+    monkeypatch.setattr(
+        "agent.protocol.agent_stream.AgentStreamExecutor", _FakeExecutor
+    )
+    monkeypatch.setattr("config.conf", lambda: {"conversation_persistence": False})
+    return cancel_registry
+
+
+def test_run_registers_and_unregisters_per_request_cancel_key(cancel_runtime):
+    service = _service()
+
+    service.run(
+        "hello",
+        "session-1",
+        lambda chunk: None,
+        agent_id="research",
+        request_id="request-1",
+    )
+
+    assert cancel_runtime.calls == [
+        ("register", "research::request-1", "research::session-1"),
+        ("unregister", "research::request-1"),
+    ]
+    assert _FakeExecutor.instances[0].cancel_event is not None
+
+
+def test_run_without_request_id_remains_session_scoped(cancel_runtime):
+    service = _service()
+
+    service.run(
+        "hello",
+        "session-1",
+        lambda chunk: None,
+        agent_id="research",
+    )
+
+    assert cancel_runtime.calls == [
+        ("register", "research::session-1", "research::session-1"),
+        ("unregister", "research::session-1"),
+    ]
+
+
+def test_run_reuses_pre_cancelled_event_without_executing_or_writing_history(
+    cancel_runtime,
+):
+    service = _service()
+    cancel_event = cancel_runtime.register(
+        "research::request-1", session_id="research::session-1"
+    )
+    cancel_event.set()
+    cancel_runtime.calls.clear()
+
+    service.run(
+        "hello",
+        "session-1",
+        lambda chunk: None,
+        agent_id="research",
+        request_id="request-1",
+    )
+
+    assert cancel_runtime.calls == [
+        ("register", "research::request-1", "research::session-1"),
+        ("unregister", "research::request-1"),
+    ]
+    assert _FakeExecutor.instances[0].cancel_event is cancel_event
+    assert _FakeExecutor.run_queries == []
+    assert _FakeExecutor.instances[0].messages == []

tests/test_openai_chat_api.py

@@ -0,0 +1,995 @@
+import json
+import queue
+import threading
+from pathlib import Path
+
+import pytest
+import web
+
+from channel.web import openai_api, web_channel
+from channel.web.openai_api import (
+    OpenAIAPIError,
+    handle_chat_completions,
+)
+
+
+def _runner(events, calls):
+    def run(
+        query,
+        session_id,
+        send_chunk_fn,
+        channel_type="",
+        agent_id=None,
+        request_id=None,
+    ):
+        calls.append(
+            {
+                "query": query,
+                "session_id": session_id,
+                "channel_type": channel_type,
+                "agent_id": agent_id,
+                "request_id": request_id,
+            }
+        )
+        for event in events:
+            send_chunk_fn(event)
+
+    return run
+
+
+def _http_app(monkeypatch, run_chat, configured_token="secret"):
+    monkeypatch.setattr(
+        openai_api, "conf", lambda: {"external_api_token": configured_token}
+    )
+    monkeypatch.setattr(openai_api, "_run_chat_service", run_chat)
+    return web.application(
+        ("/v1/chat/completions", "OpenAIChatCompletionsHandler"),
+        vars(web_channel),
+        autoreload=False,
+    )
+
+
+def _post(app, payload, authorization="Bearer secret"):
+    return app.request(
+        "/v1/chat/completions",
+        method="POST",
+        data=json.dumps(payload),
+        headers={
+            "Authorization": authorization,
+            "Content-Type": "application/json",
+        },
+    )
+
+
+@pytest.mark.parametrize(
+    ("configured_token", "authorization", "status_code"),
+    [
+        ("", "Bearer secret", 503),
+        ("secret", "", 401),
+        ("secret", "Bearer wrong", 401),
+    ],
+)
+def test_external_api_uses_independent_bearer_auth(
+    configured_token, authorization, status_code
+):
+    with pytest.raises(OpenAIAPIError) as exc_info:
+        handle_chat_completions(
+            {
+                "model": "cowagent",
+                "messages": [{"role": "user", "content": "hello"}],
+            },
+            authorization=authorization,
+            external_api_token=configured_token,
+            run_chat=lambda *args, **kwargs: None,
+        )
+
+    assert exc_info.value.status_code == status_code
+
+
+@pytest.mark.parametrize("scheme", ["bearer", "BEARER", "BeArEr"])
+def test_bearer_scheme_is_case_insensitive(scheme):
+    response = handle_chat_completions(
+        {
+            "model": "cowagent",
+            "messages": [{"role": "user", "content": "hello"}],
+        },
+        authorization=f"{scheme} secret",
+        external_api_token="secret",
+        run_chat=_runner([{"chunk_type": "content", "delta": "Hello"}], []),
+    )
+
+    assert response["choices"][0]["message"]["content"] == "Hello"
+
+
+def test_cancel_agent_request_uses_bridge_scoped_key(monkeypatch):
+    calls = []
+
+    class FakeAgentBridge:
+        agent_registry = type(
+            "FakeAgentRegistry", (), {"default_agent_id": "primary"}
+        )()
+
+        def _resolve_agent_id(self, agent_id=None):
+            calls.append(("resolve", agent_id))
+            return agent_id or "primary"
+
+        @staticmethod
+        def _cancel_key(agent_id, request_id, default_agent_id):
+            calls.append(("scope", agent_id, request_id, default_agent_id))
+            return f"{agent_id}::{request_id}"
+
+    class FakeRegistry:
+        def cancel_request(self, request_id):
+            calls.append(("cancel", request_id))
+            return True
+
+    monkeypatch.setattr(
+        "bridge.bridge.Bridge",
+        lambda: type(
+            "FakeBridge",
+            (),
+            {"get_agent_bridge": lambda self: FakeAgentBridge()},
+        )(),
+    )
+    monkeypatch.setattr("agent.protocol.get_cancel_registry", lambda: FakeRegistry())
+
+    assert openai_api._cancel_agent_request("request-1", agent_id="research") is True
+    assert calls == [
+        ("resolve", "research"),
+        ("scope", "research", "request-1", "primary"),
+        ("cancel", "research::request-1"),
+    ]
+
+
+def test_late_cancel_does_not_cancel_new_request_in_same_session(monkeypatch):
+    from agent.protocol.cancel import CancelTokenRegistry
+
+    registry = CancelTokenRegistry()
+    old_event = registry.register("request-old", session_id="shared-session")
+    registry.unregister("request-old")
+    new_event = registry.register("request-new", session_id="shared-session")
+
+    monkeypatch.setattr("agent.protocol.get_cancel_registry", lambda: registry)
+    monkeypatch.setattr(
+        openai_api,
+        "_request_cancel_key",
+        lambda request_id, agent_id=None: request_id,
+    )
+
+    assert openai_api._cancel_agent_request("request-old") is False
+    assert not old_event.is_set()
+    assert not new_event.is_set()
+
+
+def test_non_streaming_completion_maps_content_reasoning_and_tools():
+    calls = []
+    response = handle_chat_completions(
+        {
+            "model": "cowagent",
+            "conversation_id": "conversation-42",
+            "user": "user-7",
+            "messages": [
+                {"role": "assistant", "content": "previous answer"},
+                {"role": "user", "content": "inspect the workspace"},
+            ],
+        },
+        authorization="Bearer secret",
+        external_api_token="secret",
+        run_chat=_runner(
+            [
+                {"chunk_type": "reasoning", "delta": "Need a file. "},
+                {
+                    "chunk_type": "tool_start",
+                    "tool": "read",
+                    "tool_id": "call-1",
+                    "arguments": {"path": "README.md"},
+                },
+                {
+                    "chunk_type": "tool_calls",
+                    "tool_calls": [
+                        {
+                            "id": "call-1",
+                            "name": "read",
+                            "arguments": {"path": "README.md"},
+                            "result": "CowAgent",
+                            "status": "success",
+                            "elapsed": "0.01s",
+                        }
+                    ],
+                },
+                {"chunk_type": "content", "delta": "The workspace is ready."},
+            ],
+            calls,
+        ),
+        created=1700000000,
+        completion_id="chatcmpl-test",
+    )
+
+    assert calls == [
+        {
+            "query": "inspect the workspace",
+            "session_id": "openai:conversation:conversation-42",
+            "channel_type": "openai_api",
+            "agent_id": None,
+            "request_id": "chatcmpl-test",
+        }
+    ]
+    assert response == {
+        "id": "chatcmpl-test",
+        "object": "chat.completion",
+        "created": 1700000000,
+        "model": "cowagent",
+        "choices": [
+            {
+                "index": 0,
+                "message": {
+                    "role": "assistant",
+                    "content": "The workspace is ready.",
+                    "reasoning_content": "Need a file. ",
+                    "tool_trace": [
+                        {
+                            "type": "tool_start",
+                            "id": "call-1",
+                            "name": "read",
+                            "arguments": {"path": "README.md"},
+                        },
+                        {
+                            "type": "tool_result",
+                            "id": "call-1",
+                            "name": "read",
+                            "arguments": {"path": "README.md"},
+                            "result": "CowAgent",
+                            "status": "success",
+                            "elapsed": "0.01s",
+                        },
+                    ],
+                },
+                "finish_reason": "stop",
+            }
+        ],
+    }
+
+
+def test_streaming_completion_uses_standard_deltas_and_additive_cow_events():
+    calls = []
+    stream = handle_chat_completions(
+        {
+            "model": "cowagent",
+            "stream": True,
+            "user": "user-7",
+            "messages": [{"role": "user", "content": "hello"}],
+        },
+        authorization="Bearer secret",
+        external_api_token="secret",
+        run_chat=_runner(
+            [
+                {"chunk_type": "reasoning", "delta": "Think."},
+                {
+                    "chunk_type": "tool_start",
+                    "tool": "search",
+                    "tool_id": "call-2",
+                    "arguments": {"query": "CowAgent"},
+                },
+                {"chunk_type": "content", "delta": "Hello"},
+                {"chunk_type": "content", "delta": " world"},
+            ],
+            calls,
+        ),
+        created=1700000000,
+        completion_id="chatcmpl-stream",
+    )
+
+    frames = list(stream)
+    assert frames[-1] == "data: [DONE]\n\n"
+    payloads = [
+        json.loads(frame.removeprefix("data: ").removesuffix("\n\n"))
+        for frame in frames[:-1]
+    ]
+    assert payloads[0]["choices"][0]["delta"] == {"role": "assistant"}
+    assert payloads[1]["choices"][0]["delta"] == {"reasoning_content": "Think."}
+    assert payloads[1]["cow_event"] == {
+        "type": "reasoning",
+        "delta": "Think.",
+    }
+    assert payloads[2]["choices"][0]["delta"] == {}
+    assert payloads[2]["cow_event"] == {
+        "type": "tool_start",
+        "id": "call-2",
+        "name": "search",
+        "arguments": {"query": "CowAgent"},
+    }
+    assert [
+        payload["choices"][0]["delta"].get("content") for payload in payloads[3:5]
+    ] == ["Hello", " world"]
+    assert payloads[-1]["choices"][0]["finish_reason"] == "stop"
+    assert calls[0]["session_id"] == "openai:user:user-7"
+
+
+def test_streaming_agent_failure_before_first_event_returns_500():
+    def fail(*args, **kwargs):
+        raise RuntimeError("provider unavailable")
+
+    with pytest.raises(OpenAIAPIError) as exc_info:
+        handle_chat_completions(
+            {
+                "model": "cowagent",
+                "stream": True,
+                "messages": [{"role": "user", "content": "hello"}],
+            },
+            authorization="Bearer secret",
+            external_api_token="secret",
+            run_chat=fail,
+        )
+
+    assert exc_info.value.status_code == 500
+    assert exc_info.value.code == "internal_error"
+    assert "provider unavailable" not in exc_info.value.message
+
+
+def test_streaming_agent_failure_after_first_event_finishes_with_error():
+    def fail_after_content(
+        query,
+        session_id,
+        send_chunk_fn,
+        channel_type="",
+        agent_id=None,
+        request_id=None,
+    ):
+        send_chunk_fn({"chunk_type": "content", "delta": "partial"})
+        raise RuntimeError("provider unavailable")
+
+    stream = handle_chat_completions(
+        {
+            "model": "cowagent",
+            "stream": True,
+            "messages": [{"role": "user", "content": "hello"}],
+        },
+        authorization="Bearer secret",
+        external_api_token="secret",
+        run_chat=fail_after_content,
+        created=1700000000,
+        completion_id="chatcmpl-error",
+    )
+
+    frames = list(stream)
+    payloads = [
+        json.loads(frame.removeprefix("data: ").removesuffix("\n\n"))
+        for frame in frames[:-1]
+    ]
+    assert frames[-1] == "data: [DONE]\n\n"
+    assert payloads[1]["choices"][0]["delta"] == {"content": "partial"}
+    assert payloads[2]["cow_event"] == {
+        "type": "error",
+        "message": "CowAgent failed to complete the request.",
+    }
+    assert payloads[-1]["choices"][0]["finish_reason"] == "error"
+
+
+def test_streaming_returns_after_first_event_without_waiting_for_completion():
+    release_runner = threading.Event()
+    prepared_stream = queue.Queue()
+
+    def run(
+        query,
+        session_id,
+        send_chunk_fn,
+        channel_type="",
+        agent_id=None,
+        request_id=None,
+    ):
+        send_chunk_fn({"chunk_type": "content", "delta": "first"})
+        release_runner.wait()
+        send_chunk_fn({"chunk_type": "content", "delta": "second"})
+
+    def prepare():
+        prepared_stream.put(
+            handle_chat_completions(
+                {
+                    "model": "cowagent",
+                    "stream": True,
+                    "messages": [{"role": "user", "content": "hello"}],
+                },
+                authorization="Bearer secret",
+                external_api_token="secret",
+                run_chat=run,
+            )
+        )
+
+    caller = threading.Thread(target=prepare)
+    caller.start()
+    try:
+        stream = prepared_stream.get(timeout=1)
+    finally:
+        release_runner.set()
+        caller.join(timeout=1)
+
+    frames = list(stream)
+    assert '"content": "first"' in frames[1]
+    assert '"content": "second"' in frames[2]
+
+
+def test_streaming_generator_close_cancels_agent_request_once(monkeypatch):
+    release_runner = threading.Event()
+    runner_started = threading.Event()
+    cancel_calls = []
+
+    def run(
+        query,
+        session_id,
+        send_chunk_fn,
+        channel_type="",
+        agent_id=None,
+        request_id=None,
+    ):
+        send_chunk_fn({"chunk_type": "content", "delta": "first"})
+        runner_started.set()
+        release_runner.wait()
+
+    monkeypatch.setattr(
+        openai_api,
+        "_cancel_agent_request",
+        lambda request_id, agent_id=None: cancel_calls.append((request_id, agent_id)),
+        raising=False,
+    )
+    stream = handle_chat_completions(
+        {
+            "model": "cowagent",
+            "stream": True,
+            "conversation_id": "close-me",
+            "messages": [{"role": "user", "content": "hello"}],
+        },
+        authorization="Bearer secret",
+        external_api_token="secret",
+        run_chat=run,
+        completion_id="chatcmpl-close",
+    )
+
+    try:
+        assert runner_started.wait(timeout=1)
+        next(stream)
+        stream.close()
+    finally:
+        release_runner.set()
+
+    assert cancel_calls == [("chatcmpl-close", None)]
+
+
+def test_streaming_timeout_while_waiting_for_session_lock_skips_run_chat(monkeypatch):
+    from agent.protocol.cancel import CancelTokenRegistry
+
+    registry = CancelTokenRegistry()
+    session_id = "openai:conversation:queued-timeout"
+    session_lock = openai_api._SESSION_LOCKS[
+        hash(session_id) % len(openai_api._SESSION_LOCKS)
+    ]
+    run_calls = []
+    session_writes = []
+    existing_threads = set(threading.enumerate())
+
+    def run(*args, **kwargs):
+        run_calls.append((args, kwargs))
+        session_writes.append(args[1])
+
+    monkeypatch.setattr(openai_api, "_FIRST_EVENT_TIMEOUT_SECONDS", 0.01)
+    monkeypatch.setattr("agent.protocol.get_cancel_registry", lambda: registry)
+    monkeypatch.setattr(
+        openai_api,
+        "_request_cancel_scope",
+        lambda request_id, session_id=None, agent_id=None: (
+            request_id,
+            session_id,
+        ),
+    )
+    monkeypatch.setattr(openai_api, "_cancel_agent_request", lambda *args: False)
+
+    session_lock.acquire()
+    try:
+        with pytest.raises(OpenAIAPIError) as exc_info:
+            handle_chat_completions(
+                {
+                    "model": "cowagent",
+                    "stream": True,
+                    "conversation_id": "queued-timeout",
+                    "messages": [{"role": "user", "content": "hello"}],
+                },
+                authorization="Bearer secret",
+                external_api_token="secret",
+                run_chat=run,
+                completion_id="chatcmpl-queued-timeout",
+            )
+        workers = [
+            thread
+            for thread in threading.enumerate()
+            if thread not in existing_threads
+            and thread.name == "openai-chat-completion"
+        ]
+        assert len(workers) == 1
+    finally:
+        session_lock.release()
+
+    workers[0].join(timeout=1)
+    assert not workers[0].is_alive()
+    assert exc_info.value.code == "timeout"
+    assert registry.get_event("chatcmpl-queued-timeout") is None
+    assert not registry.has_active(session_id)
+    assert run_calls == []
+    assert session_writes == []
+
+
+def test_streaming_timeout_after_closed_check_reuses_pre_registered_cancel_event(
+    monkeypatch,
+):
+    from agent.protocol.cancel import CancelTokenRegistry
+
+    registry = CancelTokenRegistry()
+    entered_run_chat = threading.Event()
+    release_run_chat = threading.Event()
+    response_queue = queue.Queue()
+    observed_events = []
+    agent_runs = []
+    history_writes = []
+    session_id = "openai:conversation:registration-race"
+    completion_id = "chatcmpl-registration-race"
+    request_key = f"research::{completion_id}"
+    session_key = f"research::{session_id}"
+    existing_threads = set(threading.enumerate())
+
+    class FakeAgentBridge:
+        agent_registry = type(
+            "FakeAgentRegistry", (), {"default_agent_id": "primary"}
+        )()
+
+        @staticmethod
+        def _resolve_agent_id(agent_id=None):
+            return agent_id or "research"
+
+        @staticmethod
+        def _cancel_key(agent_id, token, default_agent_id):
+            return token if agent_id == default_agent_id else f"{agent_id}::{token}"
+
+    class FakeBridge:
+        @staticmethod
+        def get_agent_bridge():
+            return FakeAgentBridge()
+
+    def run_chat(*args, **kwargs):
+        entered_run_chat.set()
+        release_run_chat.wait()
+        cancel_event = registry.register(request_key, session_id=session_key)
+        observed_events.append(cancel_event)
+        try:
+            if not cancel_event.is_set():
+                agent_runs.append(kwargs["request_id"])
+                history_writes.append(args[1])
+        finally:
+            registry.unregister(request_key)
+
+    monkeypatch.setattr("bridge.bridge.Bridge", FakeBridge)
+    monkeypatch.setattr("agent.protocol.get_cancel_registry", lambda: registry)
+    monkeypatch.setattr(openai_api, "_FIRST_EVENT_TIMEOUT_SECONDS", 0.05)
+
+    def request():
+        try:
+            handle_chat_completions(
+                {
+                    "model": "cowagent",
+                    "stream": True,
+                    "conversation_id": "registration-race",
+                    "messages": [{"role": "user", "content": "hello"}],
+                },
+                authorization="Bearer secret",
+                external_api_token="secret",
+                run_chat=run_chat,
+                completion_id=completion_id,
+            )
+        except OpenAIAPIError as error:
+            response_queue.put(error)
+
+    caller = threading.Thread(target=request)
+    caller.start()
+    assert entered_run_chat.wait(timeout=1)
+    error = response_queue.get(timeout=1)
+    pre_registered_event = registry.get_event(request_key)
+    session_was_active = registry.has_active(session_key)
+    workers = [
+        thread
+        for thread in threading.enumerate()
+        if thread not in existing_threads and thread.name == "openai-chat-completion"
+    ]
+    assert len(workers) == 1
+    release_run_chat.set()
+    caller.join(timeout=1)
+    workers[0].join(timeout=1)
+
+    assert error.code == "timeout"
+    assert observed_events == [pre_registered_event]
+    assert pre_registered_event is not None
+    assert pre_registered_event.is_set()
+    assert session_was_active
+    assert agent_runs == []
+    assert history_writes == []
+    assert registry.get_event(request_key) is None
+    assert not registry.has_active(session_key)
+
+
+def test_streaming_generator_close_while_waiting_for_session_lock_skips_run_chat(
+    monkeypatch,
+):
+    real_queue = queue.Queue
+    session_id = "openai:conversation:queued-close"
+    session_lock = openai_api._SESSION_LOCKS[
+        hash(session_id) % len(openai_api._SESSION_LOCKS)
+    ]
+    run_calls = []
+    session_writes = []
+    existing_threads = set(threading.enumerate())
+
+    class PrimedQueue:
+        def __init__(self, maxsize):
+            self._queue = real_queue(maxsize=maxsize)
+            self._first_item = {
+                "id": "chatcmpl-queued-close",
+                "object": "chat.completion.chunk",
+                "created": 1700000000,
+                "model": "cowagent",
+                "choices": [
+                    {
+                        "index": 0,
+                        "delta": {"content": "primed"},
+                        "finish_reason": None,
+                    }
+                ],
+            }
+
+        def put(self, item, timeout=None):
+            return self._queue.put(item, timeout=timeout)
+
+        def get(self, timeout=None):
+            if self._first_item is not None:
+                item = self._first_item
+                self._first_item = None
+                return item
+            return self._queue.get(timeout=timeout)
+
+    def run(*args, **kwargs):
+        run_calls.append((args, kwargs))
+        session_writes.append(args[1])
+
+    monkeypatch.setattr(openai_api.queue, "Queue", PrimedQueue)
+    monkeypatch.setattr(openai_api, "_cancel_agent_request", lambda *args: False)
+
+    session_lock.acquire()
+    try:
+        stream = handle_chat_completions(
+            {
+                "model": "cowagent",
+                "stream": True,
+                "conversation_id": "queued-close",
+                "messages": [{"role": "user", "content": "hello"}],
+            },
+            authorization="Bearer secret",
+            external_api_token="secret",
+            run_chat=run,
+            created=1700000000,
+            completion_id="chatcmpl-queued-close",
+        )
+        next(stream)
+        stream.close()
+        workers = [
+            thread
+            for thread in threading.enumerate()
+            if thread not in existing_threads
+            and thread.name == "openai-chat-completion"
+        ]
+        assert len(workers) == 1
+    finally:
+        session_lock.release()
+
+    workers[0].join(timeout=1)
+    assert not workers[0].is_alive()
+    assert run_calls == []
+    assert session_writes == []
+
+
+def test_streaming_normal_completion_does_not_cancel_agent_request(monkeypatch):
+    cancel_calls = []
+    monkeypatch.setattr(
+        openai_api,
+        "_cancel_agent_request",
+        lambda request_id, agent_id=None: cancel_calls.append((request_id, agent_id)),
+        raising=False,
+    )
+    stream = handle_chat_completions(
+        {
+            "model": "cowagent",
+            "stream": True,
+            "messages": [{"role": "user", "content": "hello"}],
+        },
+        authorization="Bearer secret",
+        external_api_token="secret",
+        run_chat=_runner([{"chunk_type": "content", "delta": "Hello"}], []),
+    )
+
+    assert list(stream)[-1] == "data: [DONE]\n\n"
+    assert cancel_calls == []
+
+
+def test_encoded_stream_close_propagates_to_source():
+    close_calls = []
+
+    class Source:
+        def __iter__(self):
+            return self
+
+        def __next__(self):
+            return "data: first\n\n"
+
+        def close(self):
+            close_calls.append("closed")
+
+    source = Source()
+    stream = openai_api._encode_stream(source)
+
+    assert next(stream) == b"data: first\n\n"
+    stream.close()
+
+    assert close_calls == ["closed"]
+
+
+def test_invalid_messages_return_400():
+    with pytest.raises(OpenAIAPIError) as exc_info:
+        handle_chat_completions(
+            {
+                "model": "cowagent",
+                "messages": [{"role": "user", "content": [{"type": "text"}]}],
+            },
+            authorization="Bearer secret",
+            external_api_token="secret",
+            run_chat=lambda *args, **kwargs: None,
+        )
+
+    assert exc_info.value.status_code == 400
+    assert exc_info.value.code == "invalid_request"
+
+
+def test_non_streaming_agent_failure_returns_500():
+    def fail(*args, **kwargs):
+        raise RuntimeError("provider unavailable")
+
+    with pytest.raises(OpenAIAPIError) as exc_info:
+        handle_chat_completions(
+            {
+                "model": "cowagent",
+                "messages": [{"role": "user", "content": "hello"}],
+            },
+            authorization="Bearer secret",
+            external_api_token="secret",
+            run_chat=fail,
+        )
+
+    assert exc_info.value.status_code == 500
+    assert exc_info.value.code == "internal_error"
+    assert "provider unavailable" not in exc_info.value.message
+
+
+@pytest.mark.filterwarnings("ignore:setDaemon\\(\\) is deprecated:DeprecationWarning")
+def test_web_channel_binds_openai_chat_route(monkeypatch):
+    class RoutesCaptured(Exception):
+        pass
+
+    captured = {}
+
+    def capture_application(urls, namespace, autoreload):
+        captured["urls"] = urls
+        captured["namespace"] = namespace
+        captured["autoreload"] = autoreload
+        raise RoutesCaptured
+
+    channel = web_channel.WebChannel()
+    monkeypatch.setattr(channel, "_cleanup_stale_voice_recordings", lambda: None)
+    monkeypatch.setattr(web_channel.web, "application", capture_application)
+    monkeypatch.setattr(
+        web_channel,
+        "conf",
+        lambda: {"web_host": "127.0.0.1", "web_port": 9899},
+    )
+
+    with pytest.raises(RoutesCaptured):
+        channel.startup()
+
+    route_pairs = list(zip(captured["urls"][::2], captured["urls"][1::2]))
+    assert (
+        "/v1/chat/completions",
+        "OpenAIChatCompletionsHandler",
+    ) in route_pairs
+    assert (
+        captured["namespace"]["OpenAIChatCompletionsHandler"]
+        is openai_api.OpenAIChatCompletionsHandler
+    )
+    assert captured["autoreload"] is False
+
+
+def test_http_post_returns_200_json(monkeypatch):
+    app = _http_app(
+        monkeypatch,
+        _runner([{"chunk_type": "content", "delta": "Hello"}], []),
+    )
+
+    response = _post(
+        app,
+        {
+            "model": "cowagent",
+            "messages": [{"role": "user", "content": "hello"}],
+        },
+    )
+
+    assert response.status == "200 OK"
+    assert response.headers["Content-Type"] == "application/json; charset=utf-8"
+    assert json.loads(response.data)["choices"][0]["message"]["content"] == "Hello"
+
+
+def test_http_post_returns_400_for_invalid_json(monkeypatch):
+    app = _http_app(monkeypatch, _runner([], []))
+
+    response = app.request(
+        "/v1/chat/completions",
+        method="POST",
+        data="{",
+        headers={
+            "Authorization": "Bearer secret",
+            "Content-Type": "application/json",
+        },
+    )
+
+    assert response.status == "400 Bad Request"
+    assert response.headers["Content-Type"] == "application/json; charset=utf-8"
+    assert json.loads(response.data) == {
+        "error": {
+            "message": "Request body must be valid JSON.",
+            "type": "invalid_request_error",
+            "code": "invalid_json",
+        }
+    }
+
+
+def test_http_post_returns_401_for_invalid_bearer_token(monkeypatch):
+    app = _http_app(monkeypatch, _runner([], []))
+
+    response = _post(
+        app,
+        {
+            "model": "cowagent",
+            "messages": [{"role": "user", "content": "hello"}],
+        },
+        authorization="Bearer wrong",
+    )
+
+    assert response.status == "401 Unauthorized"
+    assert json.loads(response.data)["error"]["code"] == "invalid_api_key"
+
+
+def test_http_post_returns_503_when_external_api_is_disabled(monkeypatch):
+    app = _http_app(monkeypatch, _runner([], []), configured_token="")
+
+    response = _post(
+        app,
+        {
+            "model": "cowagent",
+            "messages": [{"role": "user", "content": "hello"}],
+        },
+    )
+
+    assert response.status == "503 Service Unavailable"
+    assert json.loads(response.data)["error"]["code"] == "api_disabled"
+
+
+def test_http_stream_returns_500_when_agent_fails_immediately(monkeypatch):
+    def fail(*args, **kwargs):
+        raise RuntimeError("provider unavailable")
+
+    app = _http_app(monkeypatch, fail)
+    response = _post(
+        app,
+        {
+            "model": "cowagent",
+            "stream": True,
+            "messages": [{"role": "user", "content": "hello"}],
+        },
+    )
+
+    assert response.status == "500 Internal Server Error"
+    assert response.headers["Content-Type"] == "application/json; charset=utf-8"
+    assert json.loads(response.data)["error"]["code"] == "internal_error"
+
+
+def test_http_stream_first_event_timeout_returns_500_and_cancels(monkeypatch):
+    release_runner = threading.Event()
+    runner_started = threading.Event()
+    response_queue = queue.Queue()
+    cancel_calls = []
+    request_ids = []
+
+    def block_before_first_event(*args, **kwargs):
+        request_ids.append(kwargs["request_id"])
+        runner_started.set()
+        release_runner.wait()
+
+    monkeypatch.setattr(openai_api, "_FIRST_EVENT_TIMEOUT_SECONDS", 0.01, raising=False)
+    monkeypatch.setattr(
+        openai_api,
+        "_cancel_agent_request",
+        lambda request_id, agent_id=None: cancel_calls.append((request_id, agent_id)),
+        raising=False,
+    )
+    app = _http_app(monkeypatch, block_before_first_event)
+
+    def request():
+        response_queue.put(
+            _post(
+                app,
+                {
+                    "model": "cowagent",
+                    "stream": True,
+                    "conversation_id": "timeout",
+                    "messages": [{"role": "user", "content": "hello"}],
+                },
+            )
+        )
+
+    caller = threading.Thread(target=request)
+    caller.start()
+    try:
+        assert runner_started.wait(timeout=1)
+        response = response_queue.get(timeout=0.5)
+    finally:
+        release_runner.set()
+        caller.join(timeout=1)
+
+    assert response.status == "500 Internal Server Error"
+    assert response.headers["Content-Type"] == "application/json; charset=utf-8"
+    assert json.loads(response.data)["error"] == {
+        "message": "CowAgent timed out before producing a response.",
+        "type": "api_error",
+        "code": "timeout",
+    }
+    assert len(request_ids) == 1
+    assert cancel_calls == [(request_ids[0], None)]
+
+
+def test_http_stream_returns_sse_body_and_done_marker(monkeypatch):
+    app = _http_app(
+        monkeypatch,
+        _runner([{"chunk_type": "content", "delta": "Hello"}], []),
+    )
+
+    response = _post(
+        app,
+        {
+            "model": "cowagent",
+            "stream": True,
+            "messages": [{"role": "user", "content": "hello"}],
+        },
+    )
+
+    body = response.data.decode("utf-8")
+    assert response.status == "200 OK"
+    assert response.headers["Content-Type"] == "text/event-stream; charset=utf-8"
+    assert '"delta": {"role": "assistant"}' in body
+    assert '"delta": {"content": "Hello"}' in body
+    assert '"finish_reason": "stop"' in body
+    assert body.endswith("data: [DONE]\n\n")
+
+
+def test_route_config_and_english_documentation_expose_the_public_api():
+    root = Path(__file__).parents[1]
+    config = json.loads((root / "config-template.json").read_text(encoding="utf-8"))
+    docs = (root / "docs/channels/web.mdx").read_text(encoding="utf-8")
+
+    assert config["external_api_token"] == ""
+    assert "POST /v1/chat/completions" in docs
+    assert "Authorization: Bearer" in docs
+    assert '"stream": true' in docs
+    assert "before the first event" in docs
+    assert "30 seconds" in docs
+    assert "`cow_event.type=error`" in docs
+    assert "`finish_reason=error`" in docs