phase 4.2b: true chunked tool_call streaming through SSE

three layers updated:

  ChatChunk (zhub/brains/base.py) gains optional tool_call_delta
  field. Either delta (text) or tool_call_delta (openai-shape) per
  chunk, never both.

  brain adapters (groq, openai, cerebras) now parse delta.tool_calls
  arrays from upstream SSE and emit one ChatChunk per tool_call delta
  in arrival order. text + finish_reason still emit normally.
  anthropic + ollama unchanged for now (anthropic uses a different
  shape, ollama doesn't natively support function calling).

  publisher SDK (_handle_chat in zhub/client.py) now accepts dicts
  and ChatChunk-shaped objects from chat_handler async-generators in
  addition to plain strings. _serialize_stream_chunk helper picks
  the right WS chat-chunk envelope shape; _chunk_to_text extracts
  just the text fragment for non-streaming accumulation. backward
  compatible: yielding strings still works as today.

  hub streaming branch in chat_completions: forwards tool_call_delta
  as openai-shape SSE chunks (delta.tool_calls). new
  X-Zhub-Stream-Tools mode `auto`: when finish_reason: tool_calls
  arrives, hub auto-resolves accumulated tool_calls in parallel,
  appends role:tool messages, opens a follow-up streaming proxy_chat,
  continues the SSE stream with the new chunks. bounded at 4 hops.
  pre-resolve mode (Phase 4.2) untouched.

modern OpenAI-streaming clients (Cursor agentic mode, Continue,
native openai-py with stream=True) now work fully through zhub
including streamed tool calls.

tests:
  - default mode passes tool_call deltas through (no resolve)
  - auto mode resolves + continues stream with follow-up text
  - pre-resolve mode (Phase 4.2) sanity check
  - bumped cli_up test timeout for slow CI hosts

144/144 pytest now.

spec at docs/superpowers/specs/2026-05-10-zhub-phase-4.2b-tool-call-streaming-design.md.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: Zawwarsami16

README.md

@@ -344,7 +344,7 @@ CI runs the Python suite on 3.10 / 3.11 / 3.12 plus the JS module test on every
 
 | | What |
 |---|---|
-| **4.2b** | True chunked tool_call delta streaming through SSE (today: pre-resolve mode) |
+| **~~4.2b~~** ✅ | True chunked tool_call delta streaming through SSE (default mode passes deltas through; `auto` mode also resolves+continues) |
 | **7.1** | Per-exposure access policies (whitelist of AI names / publisher keys) |
 | **More brains** | Cohere, Mistral, Together, Bedrock, Vertex, vLLM-direct |
 | **MCP resources + prompts** | Surface zhub-served files & prompts to MCP hosts, beyond just tools |

tests/test_cli_up.py

@@ -84,10 +84,13 @@ async def stream(self, messages, *, system=None, temperature=0.7,
 
     stdout_buf: list[str] = []
     api_key = None
-    deadline = time.time() + 12.0
+    deadline = time.time() + 20.0
     try:
         while time.time() < deadline:
-            line = await asyncio.wait_for(proc.stdout.readline(), timeout=4.0)
+            try:
+                line = await asyncio.wait_for(proc.stdout.readline(), timeout=8.0)
+            except asyncio.TimeoutError:
+                continue
             if not line:
                 break
             text = line.decode().rstrip()

tests/test_tool_call_streaming.py

@@ -0,0 +1,274 @@
+"""Phase 4.2b — true chunked tool_call delta streaming through SSE.
+
+Three modes:
+  default               — pass tool_call deltas through verbatim, no resolve
+  X-Zhub-Stream-Tools: pre-resolve   — Phase 4.2 buffer-then-resolve (unchanged)
+  X-Zhub-Stream-Tools: auto          — pass through + auto-resolve when
+                                       finish_reason: tool_calls arrives,
+                                       continue stream with the follow-up
+"""
+
+import asyncio
+import json
+import socket
+import threading
+import time
+
+import pytest
+
+try:
+    import fastapi  # noqa
+    import uvicorn  # noqa
+    import httpx  # noqa
+    DEPS_AVAILABLE = True
+except ImportError:
+    DEPS_AVAILABLE = False
+
+if DEPS_AVAILABLE:
+    from zhub.server import create_app
+from zhub import publish, connect
+
+
+def _free_port() -> int:
+    with socket.socket() as s:
+        s.bind(("", 0))
+        return s.getsockname()[1]
+
+
+@pytest.fixture(scope="module")
+def hub():
+    if not DEPS_AVAILABLE:
+        pytest.skip("fastapi/uvicorn/httpx not installed")
+    port = _free_port()
+    app = create_app()
+
+    def run():
+        config = uvicorn.Config(app, host="127.0.0.1", port=port,
+                                log_level="warning")
+        asyncio.run(uvicorn.Server(config).serve())
+
+    threading.Thread(target=run, daemon=True).start()
+    for _ in range(30):
+        try:
+            with socket.create_connection(("127.0.0.1", port), timeout=0.1):
+                break
+        except OSError:
+            time.sleep(0.1)
+    yield port
+
+
+def _parse_sse(body: str) -> list[dict]:
+    out = []
+    for line in body.splitlines():
+        line = line.rstrip("\r")
+        if not line.startswith("data:"):
+            continue
+        payload = line[5:].strip()
+        if payload == "[DONE]":
+            continue
+        try:
+            out.append(json.loads(payload))
+        except json.JSONDecodeError:
+            continue
+    return out
+
+
+@pytest.mark.asyncio
+async def test_default_mode_passes_tool_call_deltas_through(hub):
+    """Publisher (async-gen) yields a tool_call delta dict, then a finish.
+    Default streaming SSE should contain the tool_call delta + finish_reason
+    without the hub running auto-resolve."""
+
+    async def chat_handler(messages, options):
+        # Yield a tool_call delta, then a finish marker
+        yield {
+            "tool_call_delta": {
+                "index": 0,
+                "id": "call_x",
+                "type": "function",
+                "function": {"name": "do_thing", "arguments": "{\"a\":1}"},
+            },
+        }
+        yield {"done": True, "finish_reason": "tool_calls"}
+
+    pub = publish(
+        name="stream-tc-bot",
+        description="phase 4.2b default test",
+        chat_handler=chat_handler,
+        hub_url=f"ws://127.0.0.1:{hub}",
+    )
+    for _ in range(50):
+        if pub.api_key:
+            break
+        await asyncio.sleep(0.1)
+
+    async with httpx.AsyncClient(timeout=5.0) as c:
+        resp = await c.post(
+            f"http://127.0.0.1:{hub}/{pub.name}/v1/chat/completions",
+            json={"messages": [{"role": "user", "content": "go"}],
+                  "stream": True},
+            headers={"Authorization": f"Bearer {pub.api_key}"},
+        )
+    assert resp.status_code == 200
+    chunks = _parse_sse(resp.text)
+    assert chunks, f"empty SSE: {resp.text!r}"
+
+    # at least one chunk must carry tool_calls in its delta
+    tcs = [c for c in chunks
+           if c.get("choices", [{}])[0].get("delta", {}).get("tool_calls")]
+    assert tcs, f"no tool_call delta SSE chunk found: {chunks!r}"
+    tcd = tcs[0]["choices"][0]["delta"]["tool_calls"][0]
+    assert tcd["function"]["name"] == "do_thing"
+
+    # final chunk must carry finish_reason: tool_calls
+    finish = next(
+        (c["choices"][0].get("finish_reason") for c in reversed(chunks)
+         if c["choices"][0].get("finish_reason")),
+        None,
+    )
+    assert finish == "tool_calls"
+
+
+@pytest.mark.asyncio
+async def test_auto_mode_resolves_and_continues_stream(hub):
+    """Publisher emits tool_call delta + finish:tool_calls on first call,
+    then plain text on the follow-up. Auto mode should auto-invoke the
+    connected capability, append role:tool, re-ask publisher, and stream
+    the follow-up text. Final SSE should carry tool_call deltas THEN the
+    follow-up text deltas."""
+    invoke_n = {"n": 0}
+    call_n = {"n": 0}
+
+    async def chat_handler(messages, options):
+        call_n["n"] += 1
+        if call_n["n"] == 1:
+            yield {
+                "tool_call_delta": {
+                    "index": 0,
+                    "id": "call_a",
+                    "type": "function",
+                    "function": {"name": "auto_thing",
+                                 "arguments": json.dumps({"x": 1})},
+                },
+            }
+            yield {"done": True, "finish_reason": "tool_calls"}
+        else:
+            # follow-up after tool resolution: stream some text
+            for piece in ("ok ", "done"):
+                yield piece
+
+    def thing_handler(args):
+        invoke_n["n"] += 1
+        return {"got": args, "result": "fired"}
+
+    pub = publish(
+        name="auto-tc-bot",
+        description="phase 4.2b auto mode",
+        chat_handler=chat_handler,
+        hub_url=f"ws://127.0.0.1:{hub}",
+    )
+    for _ in range(50):
+        if pub.api_key:
+            break
+        await asyncio.sleep(0.1)
+
+    conn = connect(
+        ai_name=pub.name, api_key=pub.api_key,
+        hub_url=f"ws://127.0.0.1:{hub}",
+        capabilities={"auto_thing": ({"type": "object"}, thing_handler)},
+    )
+    await asyncio.sleep(0.6)
+
+    async with httpx.AsyncClient(timeout=10.0) as c:
+        resp = await c.post(
+            f"http://127.0.0.1:{hub}/{pub.name}/v1/chat/completions",
+            json={"messages": [{"role": "user", "content": "go"}],
+                  "stream": True},
+            headers={
+                "Authorization": f"Bearer {pub.api_key}",
+                "X-Zhub-Stream-Tools": "auto",
+            },
+        )
+    assert resp.status_code == 200
+    chunks = _parse_sse(resp.text)
+    assert chunks
+
+    # Tool_call deltas must be in the early stream
+    tc_chunks = [c for c in chunks
+                 if c.get("choices", [{}])[0].get("delta", {}).get("tool_calls")]
+    assert tc_chunks, f"no tool_call delta seen: {chunks!r}"
+
+    # Follow-up text must arrive after resolution
+    text = "".join(
+        c["choices"][0].get("delta", {}).get("content", "") or ""
+        for c in chunks
+    )
+    assert "ok" in text and "done" in text, f"missing follow-up text: {text!r}"
+
+    # Capability fired exactly once
+    assert invoke_n["n"] == 1
+    # Publisher called twice (initial + follow-up)
+    assert call_n["n"] == 2
+
+    # Final finish_reason should be "stop" (the follow-up's)
+    finish = next(
+        (c["choices"][0].get("finish_reason") for c in reversed(chunks)
+         if c["choices"][0].get("finish_reason")),
+        None,
+    )
+    assert finish == "stop"
+
+
+@pytest.mark.asyncio
+async def test_pre_resolve_mode_still_works(hub):
+    """Phase 4.2 pre-resolve path is untouched."""
+    call_n = {"n": 0}
+
+    def chat_handler(messages, options):
+        call_n["n"] += 1
+        if call_n["n"] == 1:
+            return {
+                "text": "",
+                "tool_calls": [{
+                    "id": "p1",
+                    "type": "function",
+                    "function": {"name": "pre_thing", "arguments": "{}"},
+                }],
+                "finish_reason": "tool_calls",
+            }
+        return "preresolved final"
+
+    pub = publish(
+        name="pre-tc-bot",
+        description="phase 4.2 pre-resolve sanity",
+        chat_handler=chat_handler,
+        hub_url=f"ws://127.0.0.1:{hub}",
+    )
+    for _ in range(50):
+        if pub.api_key:
+            break
+        await asyncio.sleep(0.1)
+
+    conn = connect(
+        ai_name=pub.name, api_key=pub.api_key,
+        hub_url=f"ws://127.0.0.1:{hub}",
+        capabilities={"pre_thing": ({"type": "object"}, lambda a: {"ok": True})},
+    )
+    await asyncio.sleep(0.6)
+
+    async with httpx.AsyncClient(timeout=8.0) as c:
+        resp = await c.post(
+            f"http://127.0.0.1:{hub}/{pub.name}/v1/chat/completions",
+            json={"messages": [{"role": "user", "content": "go"}],
+                  "stream": True},
+            headers={
+                "Authorization": f"Bearer {pub.api_key}",
+                "X-Zhub-Stream-Tools": "pre-resolve",
+            },
+        )
+    assert resp.status_code == 200
+    text = "".join(
+        c["choices"][0].get("delta", {}).get("content", "") or ""
+        for c in _parse_sse(resp.text)
+    )
+    assert "preresolved final" in text

zhub/brains/base.py

@@ -19,16 +19,22 @@ class ChatChunk:
     """One incremental piece of a streaming chat response.
 
     delta: incremental text emitted in this chunk (may be empty when the
-        chunk only signals end-of-stream).
+        chunk only signals end-of-stream or carries a tool_call_delta).
     done: True on the final chunk.
-    finish_reason: standard reason string ("stop", "length", etc.) on the
-        final chunk; None elsewhere.
+    finish_reason: standard reason string ("stop", "tool_calls", "length",
+        etc.) on the final chunk; None elsewhere.
+    tool_call_delta: when set, this chunk represents an incremental
+        OpenAI-shape tool_call delta (Phase 4.2b). Either delta or
+        tool_call_delta will be non-empty, not both.
+        Shape: {index, id?, type?, function: {name?, arguments?}}.
+        Argument fragments concatenate across multiple chunks.
     raw: the underlying API's chunk dict, kept for debugging.
     """
 
     delta: str = ""
     done: bool = False
     finish_reason: Optional[str] = None
+    tool_call_delta: Optional[dict] = None
     raw: Optional[dict] = None
 
 

zhub/brains/cerebras.py

@@ -102,13 +102,18 @@ async def stream(
                 if not choices:
                     continue
                 choice = choices[0]
-                delta = (choice.get("delta") or {}).get("content") or ""
+                delta_obj = choice.get("delta") or {}
+                content = delta_obj.get("content") or ""
                 finish = choice.get("finish_reason")
-                yield ChatChunk(
-                    delta=delta,
-                    done=bool(finish),
-                    finish_reason=finish,
-                    raw=data,
-                )
+                tool_call_deltas = delta_obj.get("tool_calls") or []
+                for tcd in tool_call_deltas:
+                    yield ChatChunk(
+                        delta="", done=False,
+                        tool_call_delta=tcd, raw=data,
+                    )
+                if content:
+                    yield ChatChunk(delta=content, done=False, raw=data)
                 if finish:
+                    yield ChatChunk(delta="", done=True,
+                                    finish_reason=finish, raw=data)
                     return

zhub/brains/groq.py

@@ -100,13 +100,31 @@ async def stream(
                 if not choices:
                     continue
                 choice = choices[0]
-                delta = (choice.get("delta") or {}).get("content") or ""
+                delta_obj = choice.get("delta") or {}
+                content = delta_obj.get("content") or ""
                 finish = choice.get("finish_reason")
-                yield ChatChunk(
-                    delta=delta,
-                    done=bool(finish),
-                    finish_reason=finish,
-                    raw=data,
-                )
+                tool_call_deltas = delta_obj.get("tool_calls") or []
+                # OpenAI-shape: each delta.tool_calls element is one
+                # tool-call's incremental update. Surface each as its own
+                # ChatChunk so downstream sees them in arrival order.
+                for tcd in tool_call_deltas:
+                    yield ChatChunk(
+                        delta="",
+                        done=False,
+                        tool_call_delta=tcd,
+                        raw=data,
+                    )
+                if content:
+                    yield ChatChunk(
+                        delta=content,
+                        done=False,
+                        raw=data,
+                    )
                 if finish:
+                    yield ChatChunk(
+                        delta="",
+                        done=True,
+                        finish_reason=finish,
+                        raw=data,
+                    )
                     return