phase 4.2: pre-resolve streaming mode for tool calls

new opt-in header X-Zhub-Stream-Tools: pre-resolve. when set with
stream:true, the hub runs the full non-streaming auto-resolve loop
internally (same one phase 1.8 uses), then emits the resolved final
text as a single SSE chunk + done. trades stream-latency for tool
correctness — useful when the brain may emit tool_calls and you want
the resolved answer arriving over SSE instead of as a non-stream JSON.

default streaming behavior unchanged: text chunks forwarded as SSE,
no tool resolution (matches today's path). header is purely additive.

implementation: extracted the auto-resolve body into a single helper
_run_autoresolve_loop(ai_name, ...) → (final_text, finish_reason).
both the existing non-streaming path and the new pre-resolve path call
it. zero duplication of the schema-validation + parallel-gather +
unwrap dance.

true chunked tool_call delta passthrough (where each function-args
fragment streams to the client as it arrives) is phase 4.2b — needs
brain-adapter + publisher-sdk changes to surface tool_call deltas
through the WS chat-chunk envelope shape.

tests:
- stream:true + pre-resolve header → SSE response with the AUTO-
  RESOLVED text containing both the publisher's final output and the
  tool result the connected handler returned (1)
- stream:true + no header → today's behavior preserved (1)

122/122 pytest now.

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

Author: Zawwarsami16

CLAUDE.md

@@ -86,8 +86,9 @@ Tests: `pytest -v`. The e2e tests spin up the hub in-process and run the full pu
 - **Phase 3.0b** ✅ — `X-Zhub-Entity-Hint` header on 4xx/5xx responses pointing at `/entity/errors/<code>`. Closes the entity loop: any AI hitting an error gets a self-debug pointer.
 - **Phase 4.0** ✅ — `zhub/brains/` package: `BrainAdapter` ABC + four streaming adapters (Ollama, Groq, OpenAI, Cerebras). `detect()` walks them in priority order. `examples/multi_brain_publisher.py` exposes `--brain auto|ollama|groq|openai|cerebras` so the brain underneath any zhub publisher is one CLI flag away from a swap. External clients (Pocket/Loki/curl/MCP) see no change; key stays stable across brain swaps via persistence.
 - **Phase 4.1** ✅ — Entity v2: operator-extensible. `POST/GET /entity/extend` and `DELETE /entity/extend/{id}` (auth: any registered publisher's bearer key). Extensions persist in SQLite (`entity_extensions` table), surface inline in `/entity/<section>` and at the title-matched code under `/entity/errors/<code>`, and live alongside shipped recipes (shipped wins on canonical conflicts). Caps: 8KB per body, 200 per hub. Each hub now grows its own institutional memory.
+- **Phase 4.2** ✅ — Pre-resolve streaming mode for tool calls. Header `X-Zhub-Stream-Tools: pre-resolve` + `stream:true` runs the full non-streaming auto-resolve loop internally, then emits the resolved final text as one SSE chunk + done. Trades stream-latency for tool-call correctness in streaming mode. The non-streaming auto-resolve loop is now a shared helper (`_run_autoresolve_loop`) used by both code paths. True per-token tool_call delta passthrough = future Phase 4.2b (needs brain-adapter + publisher-SDK changes to surface tool_call deltas).
 
-**Next (not started):** tool streaming via SSE (Phase 1.8c — would need to detect tool_calls during stream and pause), multi-tier API keys, real ZAI integration via `zai_publish.py`.
+**Next (not started):** real ZAI integration via `zai_publish.py`, multi-tier API keys, full tool_call streaming (4.2b), MCP resources/prompts surface.
 
 ## 6. File layout (what's where)
 

tests/test_stream_tools_preresolve.py

@@ -0,0 +1,197 @@
+"""Phase 4.2 — pre-resolve mode for tool calls in streaming responses.
+
+When client sends `stream:true` AND `X-Zhub-Stream-Tools: pre-resolve`,
+the hub runs the non-streaming auto-resolve path internally so any
+tool_calls the publisher emits are resolved before the SSE stream
+starts. The final text is emitted as one SSE chunk + done.
+
+Default streaming (no header) keeps today's behavior: text chunks
+forwarded as SSE, tool_calls (if any) ignored on the streaming path.
+"""
+
+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 stream_hub_port():
+    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]:
+    """Pull all `data: {...}` JSON chunks out of an SSE response, skipping
+    `[DONE]`."""
+    out: list[dict] = []
+    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_stream_with_preresolve_header_runs_tool_calls(stream_hub_port):
+    """Publisher emits tool_calls on first turn, plain text on second.
+    Connected client exposes the capability. Stream + pre-resolve header
+    should auto-resolve and surface the final text in the SSE stream."""
+    hub_ws = f"ws://127.0.0.1:{stream_hub_port}"
+    hub_http = f"http://127.0.0.1:{stream_hub_port}"
+    invoke_n = {"n": 0}
+    call_n = {"n": 0}
+
+    def chat_handler(messages, options):
+        call_n["n"] += 1
+        if call_n["n"] == 1:
+            return {
+                "text": "",
+                "tool_calls": [{
+                    "id": "c1",
+                    "type": "function",
+                    "function": {"name": "do_thing",
+                                 "arguments": json.dumps({"x": 1})},
+                }],
+                "finish_reason": "tool_calls",
+            }
+        last_tool = next(
+            (m for m in reversed(messages) if m.get("role") == "tool"),
+            None,
+        )
+        return f"final answer using tool: {last_tool['content']}"
+
+    def thing_handler(args):
+        invoke_n["n"] += 1
+        return {"result": "tool-fired", "got": args}
+
+    pub = publish(
+        name="stream-tool-bot",
+        description="stream pre-resolve test",
+        chat_handler=chat_handler,
+        hub_url=hub_ws,
+    )
+    for _ in range(50):
+        if pub.api_key:
+            break
+        await asyncio.sleep(0.1)
+    assert pub.api_key
+
+    conn = connect(
+        ai_name=pub.name, api_key=pub.api_key, hub_url=hub_ws,
+        capabilities={"do_thing": ({"type": "object"}, thing_handler)},
+    )
+    await asyncio.sleep(0.6)
+
+    async with httpx.AsyncClient(timeout=10.0) as client:
+        resp = await client.post(
+            f"{hub_http}/{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, resp.text
+    assert resp.headers.get("content-type", "").startswith("text/event-stream")
+
+    chunks = _parse_sse(resp.text)
+    assert chunks, f"no SSE chunks parsed: {resp.text!r}"
+    accumulated = "".join(
+        c["choices"][0].get("delta", {}).get("content", "") or ""
+        for c in chunks
+    )
+    assert "final answer using tool" in accumulated, accumulated
+    assert "tool-fired" in accumulated
+    # The last chunk should carry finish_reason
+    finish = next(
+        (c["choices"][0].get("finish_reason") for c in reversed(chunks)
+         if c["choices"][0].get("finish_reason")),
+        None,
+    )
+    assert finish == "stop"
+    assert invoke_n["n"] == 1
+    assert call_n["n"] == 2
+
+
+@pytest.mark.asyncio
+async def test_stream_without_header_keeps_current_behavior(stream_hub_port):
+    """No header → publisher just streams its text. Tool calls (if any)
+    flow through whatever today's path supports — for a plain-text
+    publisher, the SSE stream just has the text."""
+    hub_ws = f"ws://127.0.0.1:{stream_hub_port}"
+    hub_http = f"http://127.0.0.1:{stream_hub_port}"
+
+    def chat_handler(messages, options):
+        return "hello world"
+
+    pub = publish(
+        name="stream-plain-bot",
+        description="plain stream test",
+        chat_handler=chat_handler,
+        hub_url=hub_ws,
+    )
+    for _ in range(50):
+        if pub.api_key:
+            break
+        await asyncio.sleep(0.1)
+
+    async with httpx.AsyncClient(timeout=5.0) as client:
+        resp = await client.post(
+            f"{hub_http}/{pub.name}/v1/chat/completions",
+            json={"messages": [{"role": "user", "content": "hi"}],
+                  "stream": True},
+            headers={"Authorization": f"Bearer {pub.api_key}"},
+        )
+    assert resp.status_code == 200
+    chunks = _parse_sse(resp.text)
+    accumulated = "".join(
+        c["choices"][0].get("delta", {}).get("content", "") or ""
+        for c in chunks
+    )
+    assert "hello world" in accumulated

zhub/entity.md

@@ -193,9 +193,14 @@ returns final text. You get one round-trip externally even if the LLM
 took multiple internal turns.
 
 ### **Streaming for long responses**
-Add `"stream": true`. Hub returns OpenAI SSE chunks. Note: tool calls
-aren't streamed today (Phase 1.8c) — non-streaming path auto-resolves
-them; streaming path returns them in the final chunk.
+Add `"stream": true`. Hub returns OpenAI SSE chunks. Tool calls in
+streaming mode: by default the SSE stream just forwards text chunks
+(no tool resolution). Opt in to **pre-resolve mode** with
+`X-Zhub-Stream-Tools: pre-resolve` and the hub will run the full
+non-streaming auto-resolve loop internally, then emit the resolved
+final text as one SSE chunk + done. Trades stream-latency for tool
+correctness; useful when the brain might emit `tool_calls` and you
+want the resolved answer over SSE.
 
 ### **Cross-hub federation**
 Configure `ZHUB_PEERS=http://hub-b.example.com,http://hub-c.example.com`

zhub/server.py

@@ -887,6 +887,100 @@ async def manifest(ai_name: str) -> JSONResponse:
         ]
         return JSONResponse(m)
 
+    async def _run_autoresolve_loop(
+        ai_name: str,
+        initial_messages: list[dict[str, Any]],
+        model: str,
+        temperature: float,
+        max_tokens: int,
+        tools: Optional[list[dict[str, Any]]],
+        tool_choice: Any,
+        tool_resolve_mode: str,
+        max_iters: int = 4,
+    ) -> tuple[str, Optional[str]]:
+        """Run the same tool-call auto-resolve loop the non-streaming chat
+        path uses, returning (final_text, finish_reason). Used by Phase 4.2's
+        pre-resolve streaming mode so streaming clients can opt into proper
+        tool-call resolution. Side-effects: capability invocations + metrics
+        bumps happen as the loop runs."""
+        running_messages = list(initial_messages)
+        iters = 0
+        while True:
+            try:
+                response = await hub.proxy_chat(
+                    ai_name, running_messages, model, temperature, max_tokens,
+                    tools=tools or None, tool_choice=tool_choice,
+                )
+            except LookupError:
+                raise HTTPException(404, "AI offline")
+            except asyncio.TimeoutError:
+                raise HTTPException(504, "AI did not respond in time")
+
+            tool_calls = response.get("tool_calls") or []
+            if not tool_calls or tool_resolve_mode == "client" or iters >= max_iters:
+                return (response.get("text", "") or "",
+                        response.get("finish_reason"))
+
+            running_messages = running_messages + [{
+                "role": "assistant",
+                "content": response.get("text", "") or None,
+                "tool_calls": tool_calls,
+            }]
+
+            async def _resolve_one(tc: Any) -> dict[str, Any]:
+                fn = tc.get("function", {}) if isinstance(tc, dict) else {}
+                cap_name = fn.get("name", "")
+                args_raw = fn.get("arguments", "{}")
+                try:
+                    args = json.loads(args_raw) if isinstance(args_raw, str) else args_raw
+                except (ValueError, TypeError):
+                    args = {}
+                conn_id = hub.find_capability_connection(ai_name, cap_name)
+                if conn_id is None:
+                    tool_result: Any = {"error": f"capability '{cap_name}' not connected"}
+                else:
+                    schema = hub.find_capability_schema(ai_name, cap_name)
+                    val_errors = validate_schema(args, schema) if schema else []
+                    if val_errors:
+                        return {
+                            "tool_call_id": tc.get("id") if isinstance(tc, dict) else None,
+                            "name": cap_name,
+                            "args": args,
+                            "result": {"error": "validation failed: " + "; ".join(val_errors)},
+                        }
+                    try:
+                        relayed = await hub.invoke_capability(
+                            ai_name, conn_id, cap_name, args,
+                        )
+                        if isinstance(relayed, dict) and "ok" in relayed:
+                            if relayed.get("ok"):
+                                tool_result = relayed.get("result")
+                                if tool_result is None:
+                                    tool_result = {"ok": True}
+                            else:
+                                tool_result = {"error": relayed.get("error") or "invoke failed"}
+                        else:
+                            tool_result = relayed
+                    except Exception as e:
+                        tool_result = {"error": str(e)}
+                return {
+                    "tool_call_id": tc.get("id") if isinstance(tc, dict) else None,
+                    "name": cap_name,
+                    "args": args,
+                    "result": tool_result,
+                }
+
+            resolved = await asyncio.gather(*(_resolve_one(tc) for tc in tool_calls))
+            for entry in resolved:
+                hub.bump(ai_name, "tool_calls_resolved")
+                running_messages.append({
+                    "role": "tool",
+                    "tool_call_id": entry["tool_call_id"],
+                    "name": entry["name"],
+                    "content": json.dumps(entry["result"]),
+                })
+            iters += 1
+
     @app.post("/{ai_name}/v1/invoke")
     async def invoke_capability_http(ai_name: str, request: Request):
         """Direct HTTP invocation of a connected client's capability.
@@ -1007,6 +1101,59 @@ async def chat_completions(ai_name: str, request: Request):
         tool_choice = body.get("tool_choice")
 
         if stream:
+            stream_tools_mode = request.headers.get("x-zhub-stream-tools", "").lower()
+            if stream_tools_mode == "pre-resolve":
+                # Phase 4.2: run the full non-streaming auto-resolve loop,
+                # then emit the final text as a single SSE chunk + done.
+                # Trades stream-latency for tool-call correctness when the
+                # client has opted in via header.
+                tool_resolve_mode = request.headers.get("x-zhub-tool-resolve", "auto").lower()
+                final_text, final_finish = await _run_autoresolve_loop(
+                    ai_name=ai_name,
+                    initial_messages=messages,
+                    model=model,
+                    temperature=temperature,
+                    max_tokens=max_tokens,
+                    tools=merged_tools or None,
+                    tool_choice=tool_choice,
+                    tool_resolve_mode=tool_resolve_mode,
+                )
+
+                async def preresolved_stream():
+                    created = int(time.time())
+                    completion_id = "chatcmpl-" + new_request_id()[:16]
+                    sse = {
+                        "id": completion_id,
+                        "object": "chat.completion.chunk",
+                        "created": created,
+                        "model": model,
+                        "choices": [{
+                            "index": 0,
+                            "delta": {"role": "assistant", "content": final_text},
+                            "finish_reason": None,
+                        }],
+                    }
+                    yield f"data: {json.dumps(sse)}\n\n"
+                    sse_done = {
+                        "id": completion_id,
+                        "object": "chat.completion.chunk",
+                        "created": created,
+                        "model": model,
+                        "choices": [{
+                            "index": 0,
+                            "delta": {},
+                            "finish_reason": final_finish or "stop",
+                        }],
+                    }
+                    yield f"data: {json.dumps(sse_done)}\n\n"
+                    yield "data: [DONE]\n\n"
+
+                return StreamingResponse(
+                    preresolved_stream(),
+                    media_type="text/event-stream",
+                    headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
+                )
+
             try:
                 response = await hub.proxy_chat(
                     ai_name, messages, model, temperature, max_tokens, stream=True,