Forward Chat streaming deltas in real time

README.md

@@ -541,6 +541,12 @@ On a connection failure or HTTP 408/429/500/502/503/504, the proxy tries the nex
 
 The entire operation has six attempts and the existing `SKILLCLAW_LLM_REQUEST_TIMEOUT_S` time budget, including backoff and `Retry-After`. Long final quota windows fail promptly. Native Responses and Bedrock forwarding, and the standalone evolve-server LLM client, do not use this list. Restart the proxy after editing configuration.
 
+### Real-time Chat streaming
+
+OpenAI-compatible `/v1/chat/completions` now forwards `stream: true` incrementally, including separate `reasoning_content` and structured tool-call deltas. The completed turn is recorded once; cancellation or incomplete/error streams close the upstream and do not record a successful turn. Failover is allowed before a successful upstream stream starts, never after emitted output. Errors after HTTP headers are sent appear as an SSE `error` object without `[DONE]`.
+
+Streaming requires structured OpenAI tool calls; textual tool-tag repair is available only for buffered requests. Bedrock and Chat-to-Anthropic/Responses conversion retain their existing behavior. Disable buffering in any reverse proxy in front of SkillClaw.
+
 ## Quick Reference
 
 **Operational model** — Every user has their own `skillclaw` process and `~/.skillclaw/config.yaml`. Each group maps to a `group_id`; one evolve server watches that namespace, turns sessions into skills, and writes them back. Joining a group only requires the correct sharing config on the client side.

assets/README_ZH.md

@@ -491,6 +491,12 @@ llm:
 
 整个请求最多尝试六次，并共享现有 `SKILLCLAW_LLM_REQUEST_TIMEOUT_S` 总时限（包括退避和 Retry-After）。最后一个端点的长配额窗口会快速失败。原生 Responses、Bedrock 和独立 evolve-server 的 LLM 调用不使用该列表。修改后重启代理。
 
+### Chat 实时流式输出
+
+OpenAI-compatible `/v1/chat/completions` 的 `stream: true` 现在会实时转发增量，包括独立的 `reasoning_content` 和结构化工具调用分片。完成后只记录一次会话回合；客户端取消、上游错误或不完整流会关闭上游连接，不记录成功回合。仅在成功开启上游流之前允许故障切换，输出后不重放请求。HTTP 响应头发出后的错误以 SSE `error` 对象返回，不发送 `[DONE]`。
+
+流式请求需要上游提供 OpenAI 结构化工具调用；文本工具标签修复仅用于非流式请求。Bedrock、Chat 转 Anthropic/Responses 路径维持原有行为。SkillClaw 前面若有反向代理，需关闭响应缓冲。
+
 ## 快速参考
 
 **运行关系** — 每个用户有自己的 `skillclaw` 进程和 `~/.skillclaw/config.yaml`；每个群组对应一个 `group_id`，由一个 Evolver 持续读取会话、产出技能、写回存储。加入群组只需正确配置共享存储，无需部署服务端。

skillclaw/api_server.py

@@ -745,6 +745,7 @@ def _assemble_streaming_chat_completion(
                 {
                     "role": "assistant",
                     "content_parts": [],
+                    "reasoning_parts": [],
                     "tool_calls": {},
                     "finish_reason": None,
                 },
@@ -753,6 +754,8 @@ def _assemble_streaming_chat_completion(
             if isinstance(delta.get("role"), str):
                 entry["role"] = delta["role"]
 
+            if isinstance(delta.get("reasoning_content"), str):
+                entry["reasoning_parts"].append(delta["reasoning_content"])
             content = delta.get("content")
             if isinstance(content, str):
                 entry["content_parts"].append(content)
@@ -794,6 +797,8 @@ def _assemble_streaming_chat_completion(
             "role": entry["role"],
             "content": "".join(entry["content_parts"]),
         }
+        if entry["reasoning_parts"]:
+            message["reasoning_content"] = "".join(entry["reasoning_parts"])
         if entry["tool_calls"]:
             message["tool_calls"] = [entry["tool_calls"][i] for i in sorted(entry["tool_calls"])]
         choices.append(
@@ -1604,6 +1609,14 @@ async def chat_completions(
             # X-Session-Done or body session_done trigger session close.
 
             stream = bool(body.get("stream", False))
+            if stream and owner.config.llm_provider != "bedrock":
+                if not isinstance(body.get("messages"), list) or not body["messages"]:
+                    raise HTTPException(status_code=400, detail="messages must be a non-empty list")
+                return StreamingResponse(
+                    owner._stream_chat_and_track(body, session_id, turn_type, session_done),
+                    media_type="text/event-stream",
+                    headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
+                )
             result = await owner._handle_request(
                 body,
                 session_id=session_id,
@@ -2313,6 +2326,7 @@ async def _handle_request(
         session_id: str,
         turn_type: str,
         session_done: bool,
+        forwarder=None,
     ) -> dict[str, Any]:
         protocol = str(body.get("_skillclaw_protocol") or "").strip()
         messages = body.get("messages")
@@ -2396,12 +2410,13 @@ def _prompt_len(msgs):
 
         forward_body = {k: v for k, v in body.items() if k not in _NON_STANDARD_BODY_KEYS}
         forward_body["stream"] = False
-        forward_body.pop("stream_options", None)
+        if not forwarder:
+            forward_body.pop("stream_options", None)
         if "model" not in forward_body:
             forward_body["model"] = self._served_model
         forward_body["messages"] = messages  # potentially skill-injected
 
-        output = await self._forward_to_llm(forward_body)
+        output = await (forwarder or self._forward_to_llm)(forward_body)
         output["model"] = forward_body.get("model") or self._served_model
 
         choice = output.get("choices", [{}])[0]
@@ -2422,7 +2437,7 @@ def _prompt_len(msgs):
 
         # Upstream models sometimes emit tool calls as text tags instead of
         # structured `message.tool_calls`; parse and normalize both sources.
-        clean_content, text_tool_calls = _extract_tool_calls_from_text(content)
+        clean_content, text_tool_calls = (content, []) if forwarder else _extract_tool_calls_from_text(content)
         if text_tool_calls:
             content = clean_content
             tool_calls.extend(text_tool_calls)
@@ -2543,6 +2558,49 @@ def _prompt_len(msgs):
     # LLM forwarding                                                       #
     # ------------------------------------------------------------------ #
 
+    async def _stream_chat_and_track(self, body, session_id, turn_type, session_done):
+        from .chat_stream import chat_events
+        from .chat_upstream import forward_chat
+
+        queue = asyncio.Queue(maxsize=16)
+
+        async def forward(prepared):
+            async def collect(response, model):
+                events = []
+                async for event in chat_events(response):
+                    events.append(event)
+                    outgoing = dict(event, model=body.get("model") or self._served_model, session_id=session_id)
+                    await queue.put(outgoing)
+                return _assemble_streaming_chat_completion(events, fallback_model=model)
+
+            return await forward_chat(
+                self.config, prepared, timeout=_llm_request_timeout_seconds(), collect_stream=collect, stream=True
+            )
+
+        async def produce():
+            try:
+                await self._handle_request(body, session_id, turn_type, session_done, forwarder=forward)
+            except Exception as exc:
+                status = exc.status_code if isinstance(exc, HTTPException) else 500
+                await queue.put({"error": {"message": "Chat stream failed", "type": "upstream_error", "code": status}})
+                await queue.put(False)
+            else:
+                await queue.put(True)
+
+        producer = asyncio.create_task(produce())
+        try:
+            while True:
+                item = await queue.get()
+                if isinstance(item, bool):
+                    if item:
+                        yield "data: [DONE]\n\n"
+                    break
+                yield "data: " + json.dumps(item, ensure_ascii=False) + "\n\n"
+        finally:
+            if not producer.done():
+                producer.cancel()
+            await asyncio.gather(producer, return_exceptions=True)
+
     async def _forward_to_llm(self, body: dict[str, Any]) -> dict[str, Any]:
         """Forward to a real LLM API.
 

skillclaw/chat_stream.py

@@ -0,0 +1,54 @@
+"""Validate SSE frames while preserving incremental chat deltas."""
+
+from __future__ import annotations
+
+import json
+
+from fastapi import HTTPException
+
+
+async def chat_events(response):
+    data = []
+    finished = {}
+
+    def parse():
+        payload = "\n".join(data)
+        data.clear()
+        if payload == "[DONE]":
+            if not finished or not all(finished.values()):
+                raise ValueError("Incomplete chat stream")
+            return None
+        event = json.loads(payload)
+        if not isinstance(event, dict) or event.get("error"):
+            raise ValueError("Upstream stream error")
+        choices = event.get("choices")
+        if not isinstance(choices, list):
+            raise ValueError("Invalid chat stream choices")
+        for choice in choices:
+            if not isinstance(choice, dict) or not isinstance(choice.get("delta", {}), dict):
+                raise ValueError("Invalid chat stream delta")
+            index = int(choice.get("index", 0))
+            finished[index] = finished.get(index, False) or bool(choice.get("finish_reason"))
+        return event
+
+    try:
+        async for line in response.aiter_lines():
+            if line.startswith("data:"):
+                data.append(line[5:].lstrip(" "))
+            elif not line and data:
+                event = parse()
+                if event is None:
+                    return
+                yield event
+        if data:
+            event = parse()
+            if event is None:
+                return
+            yield event
+        # Even a finish_reason frame can precede a transport failure. Require
+        # the protocol terminator before committing this turn to session state.
+        raise ValueError("Missing stream terminator")
+    except Exception as exc:
+        # In particular, do not expose a TransportError to failover after
+        # successful response headers: emitted text must never be replayed.
+        raise HTTPException(status_code=502, detail="Upstream chat stream failed or ended incompletely") from exc

skillclaw/chat_upstream.py

@@ -65,6 +65,8 @@ def chat_endpoints(config, requested_model):
 def prepare_chat_request(config, body, endpoint, *, stream=False):
     payload = copy.deepcopy({k: v for k, v in body.items() if k not in {"logprobs", "top_logprobs", "stream_options"}})
     payload.update(model=endpoint["model_id"], stream=stream)
+    if stream and isinstance(body.get("stream_options"), dict):
+        payload["stream_options"] = copy.deepcopy(body["stream_options"])
     headers = {}
     if endpoint["api_key"]:
         headers["Authorization"] = f"Bearer {endpoint['api_key']}"
@@ -97,12 +99,12 @@ def retry_after_seconds(response):
             return None
 
 
-async def forward_chat(config, body, *, timeout, collect_stream):
+async def forward_chat(config, body, *, timeout, collect_stream, stream=False):
     endpoints = chat_endpoints(config, body.get("model", ""))
 
     async def run():
         index = 0
-        url, payload, headers = prepare_chat_request(config, body, endpoints[index])
+        url, payload, headers = prepare_chat_request(config, body, endpoints[index], stream=stream)
         last_status = None
         async with httpx.AsyncClient(timeout=timeout) as client:
             # A shared budget bounds the total attempts, including compatibility
@@ -150,7 +152,7 @@ async def run():
                     raise HTTPException(status_code=502, detail="Upstream LLM returned invalid JSON") from exc
                 if index + 1 < len(endpoints):
                     index += 1
-                    url, payload, headers = prepare_chat_request(config, body, endpoints[index])
+                    url, payload, headers = prepare_chat_request(config, body, endpoints[index], stream=stream)
                     continue
                 if attempt < 5:
                     await asyncio.sleep(delay if delay is not None else min(2**attempt + random.uniform(0, 1), 30))