fix(core): correlate streamed tool calls safely (#205)

* docs(architecture): design provider stream handling

* docs(agent): remove unavailable guard requirement

* fix(core): correlate streamed tool-call fragments

* fix(core): preserve streamed tool-call identity

* fix(core): preserve responses terminal reason precedence

* fix(core): defer tools until stream validation

* fix(soul): avoid replaying published stream failures

* docs(changelog): note safe streamed tool calls

* fix(core): address stream review findings

* fix(core): stop responses streams on error

Author: elkaix

AGENTS.md

@@ -143,33 +143,6 @@ validated, never trusted.
   abstraction, custom logic where native features or existing helpers suffice, and changes a junior
   maintainer would struggle to follow.
 
-## Guardrails: pythinker-guard Skill
-
-**REQUIRED BACKGROUND:** Before committing any changes to Pythinker code, **use the
-`pythinker-guard` skill** to enforce the non-negotiable rules above against time pressure and
-sunk-cost rationalization.
-
-**When to use:** Invoke `pythinker-guard` BEFORE:
-- Committing changes to Pythinker codebase
-- Opening a PR
-- Declaring a feature complete
-
-**What it prevents:** The skill enforces:
-- Surgical changes (no drive-by refactors, reformatting, or cleanup)
-- Explicit error contracts (no bare `except`, no silent failures)
-- Type safety (all new functions typed; `make check` passes)
-- Test-driven development (tests written first, gate passed locally)
-- Fail-closed behavior (errors distinguished and logged, never swallowed)
-
-The skill specifically guards against 5 pressure vectors that trigger violations:
-1. Time scarcity → shortcuts in testing, typing, error handling
-2. Sunk-cost fallacy → skipping types/tests because "we've already built most of it"
-3. Confidence illusion → "it's obvious this works" → silent errors
-4. Proximity heuristic → "we're in the file anyway" → unrelated cleanup
-5. Inversion of priorities → "tests slow us down" → untestable design
-
-See the skill itself for verification checkpoints, hard stops, and escalation triggers.
-
 ## Quick commands
 
 Use these first; they encode the supported local workflow.

CHANGELOG.md

@@ -15,6 +15,8 @@ GitHub Releases page; `0.8.0` is the new starting line.
 
 ## Unreleased
 
+- **Parallel streamed tool calls are now correlated safely.** Interleaved argument chunks stay attached to their indexed calls, malformed or truncated call streams stop before tool execution, and failed attempts are not retried after output has already been shown.
+
 ## 0.58.0 (2026-07-11)
 
 - **Agent-spec loading is more defensive and truthful.** Subagent `path`, `extend`, and

docs/en/release-notes/changelog.md

@@ -17,6 +17,8 @@ GitHub Releases page; `0.8.0` is the new starting line.
 
 ## Unreleased
 
+- **Parallel streamed tool calls are now correlated safely.** Interleaved argument chunks stay attached to their indexed calls, malformed or truncated call streams stop before tool execution, and failed attempts are not retried after output has already been shown.
+
 ## 0.58.0 (2026-07-11)
 
 - **Agent-spec loading is more defensive and truthful.** Subagent `path`, `extend`, and

docs/superpowers/specs/2026-07-14-provider-stream-and-tool-execution-design.md

@@ -15,7 +15,6 @@
 from pythinker_core._generate import GenerateResult, generate
 from pythinker_core.chat_provider import (
     ChatProvider,
-    ChatProviderError,
     StreamedMessagePart,
     TokenUsage,
 )
@@ -74,16 +73,19 @@ async def step(
 
     tool_calls: list[ToolCall] = []
     tool_result_futures: dict[str, ToolResultFuture] = {}
+    tool_callbacks_active = True
 
-    def future_done_callback(future: ToolResultFuture):
+    def future_done_callback(future: ToolResultFuture) -> None:
+        if not tool_callbacks_active:
+            return
         if on_tool_result:
             try:
                 result = future.result()
                 on_tool_result(result)
             except asyncio.CancelledError:
                 return
 
-    async def on_tool_call(tool_call: ToolCall):
+    async def on_tool_call(tool_call: ToolCall) -> None:
         tool_calls.append(tool_call)
         result = toolset.handle(tool_call)
 
@@ -105,12 +107,16 @@ async def on_tool_call(tool_call: ToolCall):
             on_message_part=on_message_part,
             on_tool_call=on_tool_call,
         )
-    except (ChatProviderError, asyncio.CancelledError):
-        # cancel all the futures to avoid hanging tasks
-        for future in tool_result_futures.values():
+    except BaseException:
+        # A later terminal dispatch can fail after earlier work was accepted. Deactivate
+        # publication before touching callbacks/tasks: already-queued callbacks cannot be
+        # retracted by remove_done_callback().
+        tool_callbacks_active = False
+        futures = list(tool_result_futures.values())
+        for future in futures:
             future.remove_done_callback(future_done_callback)
             future.cancel()
-        await asyncio.gather(*tool_result_futures.values(), return_exceptions=True)
+        await asyncio.gather(*futures, return_exceptions=True)
         raise
 
     return StepResult(

packages/pythinker-core/src/pythinker_core/__init__.py

@@ -5,11 +5,13 @@
 
 from pythinker_core.chat_provider import (
     APIEmptyResponseError,
+    APIStreamProtocolError,
     ChatProvider,
     StreamedMessagePart,
     TokenUsage,
 )
-from pythinker_core.message import ContentPart, Message, TextPart, ThinkPart, ToolCall
+from pythinker_core.message import Message, TextPart, ThinkPart, ToolCall
+from pythinker_core.stream_message_assembler import StreamMessageAssembler
 from pythinker_core.tooling import Tool
 from pythinker_core.utils.aio import Callback, callback
 
@@ -46,30 +48,43 @@ async def generate(
         APIEmptyResponseError: If the API returns an empty response.
         ChatProviderError: If any other recognized chat provider error occurs.
     """
-    message = Message(role="assistant", content=[])
-    pending_part: StreamedMessagePart | None = None  # message part that is currently incomplete
+    assembler = StreamMessageAssembler()
+    output_published = False
 
-    logger.trace("Generating with history: {history}", history=history)
+    logger.trace(
+        "Generating with {history_count} history messages and {tool_count} tools",
+        history_count=len(history),
+        tool_count=len(tools),
+    )
     stream = await chat_provider.generate(system_prompt, tools, history)
     async for part in stream:
-        logger.trace("Received part: {part}", part=part)
+        logger.trace("Received stream part: {part_type}", part_type=type(part).__name__)
         if on_message_part:
             await callback(on_message_part, part.model_copy(deep=True))
+            output_published = True
+
+        try:
+            assembler.add(part)
+        except APIStreamProtocolError as error:
+            error.output_published = output_published
+            if error.response_id is None:
+                error.response_id = stream.id
+            raise
+
+    try:
+        message = assembler.finish(
+            response_id=stream.id,
+            finish_reason=stream.finish_reason,
+        )
+    except APIStreamProtocolError as error:
+        error.output_published = output_published
+        if error.response_id is None:
+            error.response_id = stream.id
+        raise
 
-        if pending_part is None:
-            pending_part = part
-        elif not pending_part.merge_in_place(part):  # try merge into the pending part
-            # unmergeable part must push the pending part to the buffer
-            _message_append(message, pending_part)
-            if isinstance(pending_part, ToolCall) and on_tool_call:
-                await callback(on_tool_call, pending_part)
-            pending_part = part
-
-    # end of message
-    if pending_part is not None:
-        _message_append(message, pending_part)
-        if isinstance(pending_part, ToolCall) and on_tool_call:
-            await callback(on_tool_call, pending_part)
+    for tool_call in message.tool_calls or []:
+        if on_tool_call:
+            await callback(on_tool_call, tool_call)
 
     if not message.content and not message.tool_calls:
         raise APIEmptyResponseError("The API returned an empty response.")
@@ -112,16 +127,3 @@ class GenerateResult:
     """The token usage of the generated message."""
     truncated: bool = False
     """True when the response was cut off by the output-token limit (finish_reason 'length')."""
-
-
-def _message_append(message: Message, part: StreamedMessagePart) -> None:
-    match part:
-        case ContentPart():
-            message.content.append(part)
-        case ToolCall():
-            if message.tool_calls is None:
-                message.tool_calls = []
-            message.tool_calls.append(part)
-        case _:
-            # may be an orphaned `ToolCallPart`
-            return

packages/pythinker-core/src/pythinker_core/_generate.py

@@ -157,6 +157,44 @@ def __init__(self, message: str):
         super().__init__(message)
 
 
+type StreamProtocolErrorCategory = Literal[
+    "ambiguous_fragment",
+    "conflicting_identity",
+    "conflicting_name",
+    "missing_name",
+    "mixed_correlation",
+    "orphan_fragment",
+    "terminal_failure",
+    "truncated_tool_call",
+]
+
+
+class APIStreamProtocolError(ChatProviderError):
+    """A provider-neutral streamed-message correlation failure."""
+
+    category: StreamProtocolErrorCategory
+    response_id: str | None
+    stream_index: int | None
+    call_id: str | None
+    output_published: bool
+
+    def __init__(
+        self,
+        category: StreamProtocolErrorCategory,
+        *,
+        response_id: str | None = None,
+        stream_index: int | None = None,
+        call_id: str | None = None,
+        output_published: bool = False,
+    ) -> None:
+        super().__init__(f"Stream protocol error: {category}")
+        self.category = category
+        self.response_id = response_id
+        self.stream_index = stream_index
+        self.call_id = call_id
+        self.output_published = output_published
+
+
 class APIConnectionError(ChatProviderError):
     """The error raised when the API connection fails."""
 

packages/pythinker-core/src/pythinker_core/chat_provider/__init__.py

@@ -457,6 +457,7 @@ async def _convert_stream_response(
         self,
         response: AsyncIterator[ChatCompletionChunk],
     ) -> AsyncIterator[StreamedMessagePart]:
+        started_tool_call_indices: set[int] = set()
         try:
             async for chunk in response:
                 if chunk.id:
@@ -483,24 +484,36 @@ async def _convert_stream_response(
 
                 # convert tool calls
                 for tool_call in delta.tool_calls or []:
-                    if not tool_call.function:
+                    function = tool_call.function
+                    if function is None:
+                        if tool_call.id is not None:
+                            yield ToolCallPart(
+                                stream_index=tool_call.index,
+                                stream_call_id=tool_call.id,
+                            )
                         continue
 
-                    if tool_call.function.name:
+                    if tool_call.index not in started_tool_call_indices:
+                        started_tool_call_indices.add(tool_call.index)
                         yield ToolCall(
-                            id=tool_call.id or str(uuid.uuid4()),
+                            id=tool_call.id or "",
                             function=ToolCall.FunctionBody(
-                                name=tool_call.function.name,
-                                arguments=tool_call.function.arguments,
+                                name=function.name or "",
+                                arguments=function.arguments,
                             ),
+                            stream_index=tool_call.index,
                         )
-                    elif tool_call.function.arguments:
+                    elif (
+                        tool_call.id is not None
+                        or function.name is not None
+                        or function.arguments is not None
+                    ):
                         yield ToolCallPart(
-                            arguments_part=tool_call.function.arguments,
+                            arguments_part=function.arguments,
+                            name_part=function.name,
+                            stream_index=tool_call.index,
+                            stream_call_id=tool_call.id,
                         )
-                    else:
-                        # skip empty tool calls
-                        pass
         except (OpenAIError, httpx.HTTPError) as e:
             raise convert_error(e) from e
 

packages/pythinker-core/src/pythinker_core/chat_provider/pythinker.py

@@ -630,6 +630,7 @@ async def _convert_stream_response(
                                 yield ToolCall(
                                     id=block.id,
                                     function=ToolCall.FunctionBody(name=block.name, arguments=""),
+                                    stream_index=event.index,
                                 )
                             case "server_tool_use" | "web_search_tool_result":
                                 # ignore
@@ -650,7 +651,10 @@ async def _convert_stream_response(
                             case "thinking_delta":
                                 yield ThinkPart(think=delta.thinking)
                             case "input_json_delta":
-                                yield ToolCallPart(arguments_part=delta.partial_json)
+                                yield ToolCallPart(
+                                    arguments_part=delta.partial_json,
+                                    stream_index=event.index,
+                                )
                             case "signature_delta":
                                 yield ThinkPart(think="", encrypted=delta.signature)
                             case "citations_delta":

packages/pythinker-core/src/pythinker_core/contrib/chat_provider/anthropic.py

@@ -305,6 +305,7 @@ async def _convert_stream_response(
         self,
         response: AsyncIterator[ChatCompletionChunk],
     ) -> AsyncIterator[StreamedMessagePart]:
+        started_tool_call_indices: set[int] = set()
         try:
             async for chunk in response:
                 if chunk.id:
@@ -332,24 +333,36 @@ async def _convert_stream_response(
 
                 # convert tool calls
                 for tool_call in delta.tool_calls or []:
-                    if not tool_call.function:
+                    function = tool_call.function
+                    if function is None:
+                        if tool_call.id is not None:
+                            yield ToolCallPart(
+                                stream_index=tool_call.index,
+                                stream_call_id=tool_call.id,
+                            )
                         continue
 
-                    if tool_call.function.name:
+                    if tool_call.index not in started_tool_call_indices:
+                        started_tool_call_indices.add(tool_call.index)
                         yield ToolCall(
-                            id=tool_call.id or str(uuid.uuid4()),
+                            id=tool_call.id or "",
                             function=ToolCall.FunctionBody(
-                                name=tool_call.function.name,
-                                arguments=tool_call.function.arguments,
+                                name=function.name or "",
+                                arguments=function.arguments,
                             ),
+                            stream_index=tool_call.index,
                         )
-                    elif tool_call.function.arguments:
+                    elif (
+                        tool_call.id is not None
+                        or function.name is not None
+                        or function.arguments is not None
+                    ):
                         yield ToolCallPart(
-                            arguments_part=tool_call.function.arguments,
+                            arguments_part=function.arguments,
+                            name_part=function.name,
+                            stream_index=tool_call.index,
+                            stream_call_id=tool_call.id,
                         )
-                    else:
-                        # skip empty tool calls
-                        pass
         except (OpenAIError, httpx.HTTPError) as e:
             raise convert_error(e) from e