From 67f329834a11db5328a3dc7907aa255c4a9bcc89 Mon Sep 17 00:00:00 2001 From: elkaix Date: Tue, 14 Jul 2026 19:42:53 -0400 Subject: [PATCH 01/10] docs(architecture): design provider stream handling --- ...ovider-stream-and-tool-execution-design.md | 535 ++++++++++++++++++ 1 file changed, 535 insertions(+) create mode 100644 docs/superpowers/specs/2026-07-14-provider-stream-and-tool-execution-design.md diff --git a/docs/superpowers/specs/2026-07-14-provider-stream-and-tool-execution-design.md b/docs/superpowers/specs/2026-07-14-provider-stream-and-tool-execution-design.md new file mode 100644 index 00000000..481e9c9d --- /dev/null +++ b/docs/superpowers/specs/2026-07-14-provider-stream-and-tool-execution-design.md @@ -0,0 +1,535 @@ +# Provider stream correlation, compatibility profiles, and tool execution design + +**Status:** Approved + +**Date:** 2026-07-14 + +## Purpose + +Pythinker must treat streamed tool calls as correlated protocol objects, not as one sequential stream of mergeable fragments. The current core accumulator holds one pending part. When two tool calls are streamed and their argument fragments interleave, a fragment can attach to the most recently observed call instead of the call identified by the provider's index. A deterministic reproduction leaves the first call empty and concatenates both argument objects onto the second call. + +The same investigation found two adjacent architecture problems. Provider compatibility behavior is distributed across authentication modules, `llm.py`, provider adapters, and tool visibility. Tool execution is concentrated in `PythinkerToolset`, where registry and MCP lifecycle code share a class with parsing, policy, hooks, deduplication, scheduling, telemetry, and cancellation. + +This design addresses the three problems in dependency order: + +1. Correct indexed stream assembly in `pythinker-core`. +2. Concentrate transport compatibility and add clean, independent Z.AI Coding Plan and standard API routes. +3. Extract a batch-oriented tool execution engine without changing execution behavior. + +The exact transcript behind the originally reported intermittent GLM-5.2 symptom is unavailable. The indexed-stream defect and existing Z.AI incompatibilities are proven independently; the design does not claim they are the only possible causes of that report. + +## Goals + +- Correlate tool-call fragments by provider-supplied identity while preserving model call order. +- Prevent any tool side effect until the entire assistant stream has terminated successfully and all calls are structurally complete. +- Fail explicitly on ambiguous, conflicting, orphaned, or truncated tool-call streams. +- Preserve compatibility for simple third-party providers that emit one unindexed call at a time. +- Resolve provider quirks through one typed compatibility profile attached to the active `LLM`. +- Keep provider-specific decisions out of `PythinkerSoul` and tool execution. +- Support Z.AI Coding Plan and standard API credentials simultaneously with explicit identities and no endpoint guessing. +- Preserve exact captured GLM reasoning when preserved thinking is enabled. +- Give tool execution one batch boundary while retaining `PythinkerToolset` as the compatibility facade. +- Keep the three phases independently reviewable and separately releasable. + +## Non-goals + +- Supporting the mainland `open.bigmodel.cn` service. +- Supporting an Anthropic-compatible Z.AI transport. +- Detecting a Z.AI key's plan type or retrying a request against another Z.AI endpoint. +- Migrating old `z-ai` provider entries, model aliases, login names, or environment variables. This integration has no users requiring compatibility, so the new contract is clean rather than transitional. +- Changing approval, hook, deduplication, concurrency, interruption, or tool-result semantics for conforming tools during the execution extraction. A tool that suppresses cancellation is explicitly hardened as an unsupported failure case. +- Starting tools speculatively before stream completion. +- Adding runtime dependencies, raw SSE logging, reasoning logging, or hosted telemetry. +- Claiming the original intermittent symptom is resolved without a redacted live confirmation. + +## External provider contract + +The implementation follows the current Z.AI global API contracts: + +- Coding Plan OpenAI-compatible base URL: `https://api.z.ai/api/coding/paas/v4`. +- Standard API OpenAI-compatible base URL: `https://api.z.ai/api/paas/v4`. +- GLM-5.2 has a 1,000,000-token context window and a documented maximum output of 131,072 tokens. +- `tool_stream` enables streamed function-call deltas on supported GLM models. +- GLM-5.2 maps `low`, `medium`, and `high` reasoning effort to `high`; `xhigh` and `max` map to `max`; `none` and `minimal` skip thinking. +- Preserved thinking requires `clear_thinking: false` and exact, unmodified, correctly ordered replay of prior `reasoning_content`. + +Primary references: + +- +- +- +- + +These are transport contracts, not instructions to trust provider output. Tool names, arguments, stream structure, and model catalogs remain external untrusted input and are validated at their boundaries. + +## Approaches considered + +### Disable tool streaming + +Turning off `tool_stream` or using only non-streaming responses would reduce exposure for one provider, but it would leave the core defect intact for every provider that streams parallel calls. It would also discard supported functionality rather than repair the contract. Rejected. + +### Buffer independently inside each provider adapter + +Every adapter could construct complete calls before yielding them. That localizes wire details but duplicates correlation, ordering, conflict detection, deterministic ID fallback, and malformed-stream policy across OpenAI Chat Completions, Responses, Anthropic-shaped, and first-party transports. Custom providers would still encounter the unsafe generic accumulator. Rejected. + +### Shared stream assembler, typed compatibility profiles, and batch execution + +Adapters preserve wire correlation metadata; one core assembler owns stream-level call state; compatibility profiles own request and replay quirks; one execution engine owns post-generation tool lifecycle. This creates three deep modules with distinct dependencies and failure contracts. Selected. + +## Architecture + +```text +saved provider + model configuration + | + v +ProviderCompatibilityResolver -----> immutable compatibility profile + | | + v v + ChatProvider adapter <------ request/replay policy + | + normalized stream parts + | + v + StreamMessageAssembler + | | + live text/thinking complete ToolCalls + callbacks | + v + ToolExecutionEngine + | | + completion callbacks ordered results + | | + +-------> context growth +``` + +The dependency direction is deliberate: + +- Authentication owns credentials, endpoints, and model discovery. +- Compatibility resolution owns transport behavior for a known provider/model pair. +- Provider adapters translate wire events and message history. +- Stream assembly owns correlation and completion. +- Tool execution owns validated side effects after generation. +- `PythinkerSoul` coordinates request, persistence, interruption, and turn control without knowing provider quirks or tool-internal lifecycle. + +## Phase 1: Correlated stream assembly + +### Stream event contract + +`ToolCall` remains the complete persisted/message-level object. Streaming call starts and `ToolCallPart` fragments carry non-persisted correlation metadata: + +- provider call index when supplied; +- provider call ID when supplied; +- a complete function name on a call start; +- an explicitly identified function-name fragment on a fragment event; +- argument fragment when supplied. + +Correlation metadata is excluded from provider history and session serialization. Non-streamed provider responses continue to yield complete `ToolCall` values. + +Existing providers that emit a complete `ToolCall` start followed by `ToolCallPart` fragments remain supported. OpenAI-shaped adapters must copy `tool_calls[].index` and any available ID onto the normalized stream parts rather than discarding them. + +If a provider omits a call ID, the assembler creates a deterministic message-local surrogate after assembly. It must not use an unseeded UUID. A provider ID that appears later replaces the provisional identity only when it is consistent with the same indexed call. + +### `StreamMessageAssembler` + +A new core module owns message construction. It keeps content assembly separate from an ordered map of in-progress tool calls. + +For each call it records: + +- correlation key; +- provider index or first-seen order; +- stable provider ID or deterministic surrogate; +- complete function name and any name fragments; +- accumulated arguments; +- whether a complete call start was observed. + +Resolution rules are: + +1. Use stream index when present. +2. Otherwise use provider call ID when present. +3. Otherwise use the only currently open legacy call. +4. If more than one call could accept an unindexed fragment, fail as ambiguous. + +A fragment may arrive before its call start when it has an index or ID. It is buffered and must resolve before terminal validation. Repeated complete metadata is accepted only when identical. Name fragments append in arrival order for their correlated call. If a complete name and name fragments both appear, their assembled values must agree; otherwise assembly fails. Conflicting index/ID associations or complete function names fail explicitly. + +An indexed stream returns calls in ascending provider-index order. An ID-only or legacy unindexed stream returns calls in first-seen order. A multi-call stream that mixes indexed and unindexed call starts fails as ambiguous instead of guessing an order. + +### Completion and callback semantics + +`on_message_part` continues receiving defensive copies of normalized text, thinking, and tool stream parts as they arrive. Text and thinking therefore remain live. + +`on_tool_call` fires once per complete call, in model call order, only after successful terminal assembly. `pythinker_core.step` consequently starts no tool task while the provider stream is still open. This intentionally trades speculative overlap for correctness and retry safety. + +The assembler records whether any `on_message_part` callback was invoked during the attempt. That observable-publication state is attached to a terminal protocol error so the retry classifier cannot replay already-published text, thinking, or tool deltas as though the first attempt never happened. + +A successful terminal stream requires: + +- no unresolved keyed fragment; +- no ambiguous legacy fragment; +- one non-empty function name per call; +- stable call identity; +- a finish reason that does not indicate truncation or transport failure. + +An absent argument payload normalizes to `{}`. JSON validity remains an execution-boundary concern so malformed model-generated arguments produce the existing `ToolParseError` and can be shown to the model without retrying generation. + +### Stream failures + +A new typed `APIStreamProtocolError` distinguishes malformed provider streaming from empty output, HTTP failure, and invalid tool JSON. Its safe diagnostic fields are limited to response ID when available, correlation index/ID, and error category; it never includes arguments or reasoning. + +- Ambiguous, conflicting, or orphaned stream state raises `APIStreamProtocolError`. +- A stream ending with `finish_reason="length"` and any tool-call state executes no tools and raises a typed truncation/protocol failure. +- A transport exception cancels assembly and executes no tools. +- The CLI may apply its existing bounded generation retry policy to a stream protocol failure only when no message-part callback published output for that attempt. After any observable publication, the failure is surfaced and is not retried, preventing duplicated or contradictory transcript output. +- Structurally complete invalid JSON is not a retryable stream error; it becomes `ToolParseError` during execution. + +All partial state is released on success, error, or cancellation. + +### Compatibility behavior for custom providers + +The public stream union remains compatible. A custom provider may continue yielding: + +- complete, non-streamed `ToolCall` objects; +- one unindexed `ToolCall` followed by unindexed fragments; +- fully indexed/interleaved calls. + +Only a stream that becomes genuinely ambiguous is newly rejected. Silent cross-call corruption is not a supported compatibility behavior. + +## Phase 2: Provider compatibility profiles and Z.AI routes + +### Profile boundary + +A new `provider_compatibility` module exposes an immutable profile and one resolver. The resolver receives the provider key, provider configuration, and model configuration. Resolution precedence is: + +1. Exact managed provider key. +2. Exact normalized hostname/path and API family for user-defined compatible endpoints. +3. Model-specific behavior only inside an already recognized provider or endpoint family. +4. Conservative API-family defaults for unknown providers. + +A model name alone never activates hosted-provider behavior. A local or third-party model named `glm-*` must not inherit Z.AI credentials, endpoints, thinking replay, or request fields. + +The profile owns only transport compatibility: + +- API family and adapter options; +- output-token parameter and model ceiling; +- tool-result conversion; +- deferred-tool-search capability; +- tool-stream request flags; +- reasoning field and effort mapping; +- thinking enable/disable body; +- reasoning replay requirements; +- provider-specific role support where currently required. + +Authentication, token refresh, catalog retrieval, permissions, approvals, retries, and business policy remain outside the profile. + +`create_llm` remains the transport factory. Its provider-type match instantiates the adapter, while the profile supplies the adapter options and generation overrides. Existing Kimi, DashScope, Qwen, Anthropic-proxy, OpenAI, and related behavior moves behind profiles with behavior-locking tests. Only the approved Z.AI behavior changes. + +The resolved profile is stored on `LLM`. `supports_deferred_tool_search`, tool-result conversion, output capping, thinking-level availability, and replay serialization consult that profile rather than independently reclassifying the provider. + +### Clean Z.AI identities + +The integration defines two independent managed platforms: + +| Route | Platform/model prefix | Provider key | Environment variable | Base URL | +| --- | --- | --- | --- | --- | +| Coding Plan | `z-ai-coding/*` | `managed:z-ai-coding` | `ZAI_CODING_API_KEY` | `https://api.z.ai/api/coding/paas/v4` | +| Standard API | `z-ai-api/*` | `managed:z-ai-api` | `ZAI_API_KEY` | `https://api.z.ai/api/paas/v4` | + +There is no `z-ai/*` compatibility alias, legacy environment fallback, or persisted configuration migration. Interactive and CLI login, refresh, model selection, usage display, and logout use the explicit route identities. Logging into one route does not remove or rewrite the other. The most recently completed login may become the default model according to existing login behavior. + +### Authentication and model discovery + +Each route validates and discovers with its own OpenAI-compatible `/models` endpoint and bearer credential. Discovery state is never shared across routes. + +- Missing or blank credentials fail before network access. +- `401` and `403` are authentication failures; the key and route are not saved. +- Timeout, transport failure, non-auth HTTP failure, or structurally unusable catalog data produces an explicit degraded-login information event and the route's curated built-in catalog. +- An empty but structurally valid catalog also uses the curated catalog and reports degradation. +- GLM-5.2 remains pinned per route when the corresponding endpoint omits it but the route contract supports it. +- Refresh updates and prunes only models owned by that route's provider key. +- Logout removes only the selected route and repairs the default model if necessary. +- No login probe sends a billed chat completion, and no failure retries the credential against the other route. + +### GLM request behavior + +For GLM-5.2 on either route: + +- `reasoning_content` is captured as `ThinkPart` and replayed exactly when present. +- Enabled thinking sends `thinking: {"type": "enabled", "clear_thinking": false}`. +- Disabled or minimal thinking sends exactly `thinking: {"type": "disabled"}` and omits both `clear_thinking` and reasoning effort. +- `low`, `medium`, and `high` send `reasoning_effort: "high"`. +- `xhigh` and `max` send `reasoning_effort: "max"`. +- A new configuration with no explicit thinking preference initializes to `high`; an existing explicit preference is preserved. +- `tool_stream: true` is sent when tools are present. +- The maximum output setting is 131,072 tokens. +- Tool results use the profile's single-text representation. +- Deferred `ToolSearch` remains unavailable. + +Other curated GLM models use their documented context/output limits and binary thinking support. `reasoning_effort` is sent only for a model that supports it, and `tool_stream` is sent only for supported GLM versions. Unknown discovered models do not inherit GLM-5.2-only controls merely from catalog presence. + +Preserved thinking never synthesizes content. An assistant turn with no captured reasoning replays no invented placeholder. Captured reasoning is neither edited nor logged. Request assembly and history compaction may retain or remove complete history turns according to their existing contracts, but they must not truncate or rewrite a retained `ThinkPart`. + +### Provider-neutral callers + +`PythinkerSoul` continues to call `pythinker_core.step` with the active provider and toolset. It contains no Z.AI, GLM, endpoint, replay, or tool-result conversion branch. Tool visibility asks the active profile for the deferred-search capability. Provider adapters receive already-resolved compatibility policy rather than inspecting arbitrary model names in agent code. + +## Phase 3: Batch-oriented tool execution + +### Relationship to the prior Toolset characterization design + +The earlier Toolset characterization design assumed the per-call `handle` interface would remain the execution seam and prohibited extraction based on file size alone. Phase 1 changes that premise: generation now produces a complete call batch before any dispatch, and this design explicitly approves a batch execution boundary. + +This design supersedes only the prior conditional no-go for the private execution-pipeline extraction. The prior numerical thresholds, performance-trigger requirement, and benchmark-backed go/no-go decision record are waived for this extraction because the new complete-batch contract creates a functional seam rather than a line-count or performance optimization. Deterministic fault tests, behavioral characterization, and before/after regression measurements remain mandatory. The prior no-go rules for MCP lifecycle and registry extraction remain unchanged. Phase 3 does not extract MCP lifecycle or the registry. + +### Ownership split + +`PythinkerToolset` remains the caller-facing facade and owns: + +- tool registration and lookup; +- hidden and advertised tool projection; +- runtime visibility policy; +- shared and external tool registration; +- MCP connection, inventory, refresh, publication, and cleanup. + +A private `ToolExecutionEngine` owns: + +- argument parsing and canonicalization; +- same-step and cross-step deduplication state; +- permission and approval enforcement; +- pre-use and post-use hook lifecycle; +- reader/writer scheduling; +- execution lifecycle events; +- tool spans, metrics, and existing telemetry calls; +- exception conversion; +- cancellation and batch settlement; +- repeated-call reminders and execution summary. + +Moved state and logic are deleted from `PythinkerToolset`; the engine is not a forwarding-only file. The read/write gate moves with its sole execution owner. `PythinkerToolset.handle`, `begin_step`, `end_step`, and existing summary properties remain compatibility facades for tests and external callers, but `PythinkerSoul` uses the batch path. + +### Batch contract + +The optional batch protocol accepts: + +- the complete ordered `ToolCall` sequence; +- turn ID; +- step number; +- prior normalized call fingerprints. + +It returns a cancellable batch handle that exposes: + +- completion callbacks as individual calls finish; +- ordered final `ToolResult` values; +- current normalized call fingerprints; +- whether deduplication triggered; +- the consecutive-identical-call count. + +`pythinker_core.step` detects this optional protocol. Toolsets without it retain the existing `handle` dispatch path. The optional execution context and summary do not change provider history or public CLI configuration. + +`PythinkerSoul` passes step context once and consumes the returned summary. It no longer coordinates `begin_step`/`end_step` or reads `PythinkerToolset`-specific execution state. It remains responsible for turn limits, context growth, interrupted-result persistence, and stuck-loop decisions because those are conversation concerns rather than tool invocation concerns. + +### Execution pipeline + +For one batch the engine: + +1. Resolves each tool and parses its JSON arguments. +2. Canonicalizes arguments and records model call order. +3. Classifies same-step and cross-step duplicates. +4. Checks permission and approval policy before side effects. +5. Runs `PreToolUse`; a block result is authoritative. +6. Schedules the call through the existing reader/writer policy. +7. Emits execution-started state at the existing lifecycle point. +8. Invokes the tool with a defensive argument copy. +9. Records existing spans, metrics, logs, and tool-call events. +10. Schedules the existing managed post-use hook. +11. Applies any repeat reminder and resolves exactly one result for the call ID. + +Same-step duplicate calls share the original underlying task and return the same value under their own IDs. They do not duplicate side effects. Final results follow model call order even when completion callbacks arrive in another order. + +### Concurrency and cancellation + +- Tools declaring `supports_parallel=True` remain bounded readers. +- Mutating, unclassified, plugin, and MCP tools remain exclusive by default. +- A queued writer continues blocking new readers and cannot starve. +- No tool invocation is retried automatically. +- Ordinary tool exceptions retain `ToolRuntimeError` conversion and actionable logging. +- Parse, missing-tool, policy, and hook-block outcomes retain their existing typed results. +- `CancelledError` and other control-flow `BaseException` values propagate after owned resources and spans are settled. +- Cancelling a batch cancels pending tasks, waits at most five seconds for settlement, and retains already-completed results for interruption persistence. +- A task still running after that bound raises `ToolCancellationTimeoutError`, marks the engine poisoned, and remains in an engine-owned late-task registry with exception-consuming completion callbacks. A poisoned engine rejects later batches until every late task settles; it never permits a potentially mutating orphan to overlap a new batch. +- Engine cleanup re-cancels and supervises the late-task registry and reports any unresolved count explicitly rather than claiming clean shutdown. +- Post-hook work remains owned by the existing hook engine lifecycle; extraction must not create unsupervised tasks. + +Phase 3 is judged against the Phase 1/2 baseline. The only intentional execution-timing change occurred in Phase 1, when dispatch moved behind successful stream termination. The five-second poison path is an explicit safety hardening for tools that violate the supported cancellation contract; conforming tools retain existing behavior. + +## Failure truthfulness + +- A malformed or truncated tool stream never becomes a successful assistant/tool step. +- A retryable generation failure occurs before tool side effects, so bounded retry cannot duplicate a write; a failed attempt that already published stream output is not retried. +- Invalid model-generated JSON is returned as a parse failure, not tool success and not transport failure. +- Z.AI authentication failure never saves a key or silently tests another paid endpoint. +- Catalog fallback is labeled degraded and cannot be mistaken for live discovery. +- Missing preserved reasoning is never replaced with synthetic text presented as authentic model reasoning. +- Approval, policy, and `PreToolUse` uncertainty continues to fail closed. +- One tool failure does not reorder or erase other completed results. +- Cancellation leaves no unanswered persisted tool call: the soul retains real completed results and writes interruption results only for unfinished calls. An uncooperative task poisons execution and prevents another batch until its possible side effect has settled. + +## Safe diagnostics + +New diagnostics are local and structured. They may include: + +- compatibility profile ID; +- endpoint class, never credential; +- response ID when already safe to log; +- tool-call index or provider call ID; +- protocol error category; +- aggregate call count and completion state. + +They must not include API keys, authorization headers, tool arguments, tool output, reasoning content, full request/response bodies, or raw SSE chunks. This design adds no telemetry destination or event family. + +## Test design + +Tests are written before implementation changes and must fail for the intended reason. + +### Phase 1 regression and contract tests + +- Two indexed calls whose argument fragments alternate reconstruct independently. +- Three-call permutations preserve ascending provider-index order; ID-only calls preserve first-seen order. +- A fragment arriving before its indexed start resolves correctly. +- Stable repeated IDs and complete names are accepted; conflicts fail; correlated name fragments assemble in order. +- Missing provider ID receives a deterministic surrogate. +- One unindexed legacy call remains compatible. +- Two open calls plus an unindexed fragment fail as ambiguous. +- An unresolved orphan fails at terminal validation. +- A terminal `length` with tool-call state executes nothing. +- Transport error and cancellation release all assembly state and execute nothing. +- Text and thinking callbacks remain live while complete-call callbacks wait for terminal assembly. +- Complete-call callbacks fire exactly once and in model order. +- Non-streamed complete tool calls retain behavior. +- OpenAI Chat Completions and every adapter that emits partial tool calls retain correlation metadata. + +The original deterministic reproduction is a required regression fixture. + +### Phase 2 profile and route tests + +- Coding and standard providers, keys, and same-named models coexist. +- Login for one route does not prune or replace the other. +- Each environment variable resolves only its named route. +- Blank credentials, `401`, and `403` do not mutate saved configuration. +- Discovery timeout, transport failure, malformed payload, and empty catalog report degraded fallback. +- Refresh and logout affect only the selected route. +- Request capture asserts exact Coding and standard base URLs and bearer authentication without exposing the key. +- GLM-5.2 off/minimal/high/max bodies match the approved mapping. +- Enabled requests include preserved-thinking and tool-stream controls. +- Disabled requests omit reasoning effort. +- History replay preserves exact reasoning bytes and ordering. +- A local `glm-5.2` on an unrelated endpoint does not resolve a Z.AI profile. +- ToolSearch visibility and tool-result serialization derive from the profile. +- The output-token ceiling and reasoning support are model-specific. +- Existing non-Z.AI compatibility fixtures remain byte- or behavior-equivalent as appropriate. +- Production code, docs, and tests contain no legacy `z-ai/*` route or fallback contract except historical release text if needed. + +No test requires a real key. An optional manual smoke test may use separately configured credentials only when explicitly invoked; it redacts request headers and content. + +### Phase 3 characterization and parity tests + +Characterization precedes movement of production logic. Tests cover: + +- not-found suggestions and malformed JSON; +- permission denial and approval closure; +- authoritative `PreToolUse` block; +- pre-hook and post-hook failure behavior; +- same-step task sharing and per-ID results; +- cross-step reminders and consecutive-repeat counts; +- reader overlap, reader bound, writer exclusion, and writer fairness; +- tool exception conversion and later-call recovery; +- cancellation before admission, during execution, and during batch settlement, including the five-second poisoned-engine path for a cancellation-suppressing tool; +- completion-order callbacks versus model-order results; +- partial completion followed by interruption; +- retry of the same model step without awaiting stale cancelled tasks; +- context rewind with cleared prior fingerprints; +- third-party per-call `handle` fallback; +- no leaked tasks, permits, spans, or hook work. + +Existing Toolset and concurrency tests remain compatibility tests. Tests should assert behavior through the facade and the batch contract rather than private implementation line structure. + +## Acceptance criteria + +### Phase 1 + +- The deterministic interleaving reproduction returns each call with only its own arguments. +- No code path starts a tool before successful terminal stream assembly. +- Ambiguous or conflicting stream state produces a typed failure with no side effect. +- Existing simple and non-streamed providers remain compatible. + +### Phase 2 + +- `z-ai-coding/glm-5.2` and `z-ai-api/glm-5.2` can coexist and select their own credentials/endpoints. +- GLM-5.2 request, replay, stream, and output behavior matches the approved profile. +- Provider-specific behavior is absent from `PythinkerSoul` and tool execution. +- Non-Z.AI compatibility behavior remains locked by tests. +- No migration or endpoint fallback path exists. + +### Phase 3 + +- `PythinkerSoul` performs one batch handoff and consumes one execution summary rather than coordinating per-call Toolset lifecycle. +- Execution state has one owner and is deleted from `PythinkerToolset` except compatibility delegation. +- Approval, hooks, deduplication, concurrency, telemetry, errors, ordering, and cancellation match the characterized baseline. +- Removing the engine would force its state machine back into `PythinkerToolset`; it passes the deletion test. + +## Delivery sequence + +### PR 1: Stream correlation correctness + +Scope: `pythinker-core` message/stream contracts, adapters, assembler, typed error, and tests. Add an `## Unreleased` changelog entry describing corrected parallel streamed tool calls. + +Minimum gate: + +```bash +make check-pythinker-core +make test-pythinker-core +``` + +Run the repository-required broader pre-PR gate before opening the PR. + +### PR 2: Compatibility profiles and dual Z.AI routes + +Scope: compatibility resolver, provider factory integration, clean auth identities, login/refresh/logout, model catalogs, docs, changelog, and focused request/history tests. + +Minimum gates: + +```bash +make check-pythinker-core +make test-pythinker-core +make check-pythinker-code +make test-pythinker-code +``` + +Core gates are required because profile-driven replay and adapter options may change `pythinker-core` contracts. Because this phase depends on terminal correlation before enabling tool streaming, PR 2 must not merge before PR 1. + +### PR 3: Tool execution engine extraction + +Scope: batch protocol, execution engine, Toolset delegation, soul integration, characterization/parity tests, and a changelog entry or an explicitly justified `no-changelog` classification if maintainers determine it is entirely invisible. + +Minimum gates: + +```bash +make check-pythinker-core +make test-pythinker-core +make check-pythinker-code +make test-pythinker-code +``` + +Each PR receives independent code review, the required CodeRabbit review when merging, and the complete repository pre-PR checks for every affected package. Focused passing tests are not a substitute for the full package gates. + +## Rollback + +- PR 1 can be reverted before PR 2. Once PR 2 enables Z.AI tool streaming, PR 2 must be reverted before reverting PR 1. +- PR 2 has no migration to reverse. Before release it is a code-only revert; after a user has configured either new route, rollback must remove the now-unsupported provider/model entries or move the default to another provider before reverting. +- PR 3 can be reverted independently because `PythinkerToolset` retains compatibility facades and provider/history contracts do not depend on the extracted implementation. +- No hidden feature flag selects old versus new behavior. A failed phase is reverted rather than kept as a parallel path. + +## Residual risks + +- The original user-observed GLM transcript is unavailable, so a separate model or prompt-level issue may remain after the deterministic defects are fixed. +- Delaying tool dispatch until stream completion removes speculative overlap. Correctness is mandatory; latency impact should be measured but cannot justify restoring unsafe dispatch. +- Sending the documented maximum output ceiling permits longer paid standard-API responses. It is a ceiling rather than a target, but usage behavior should be observed during the optional smoke test. +- Provider model catalogs and compatibility behavior can change. The route profiles therefore require focused contract tests and authoritative documentation review when updated. +- The execution extraction touches a load-bearing approval and cancellation path. Characterization, one-owner state, and separate delivery are mandatory mitigations. From 2d2fc73c11206af3c0f546e453d8577a7f496dcb Mon Sep 17 00:00:00 2001 From: elkaix Date: Tue, 14 Jul 2026 21:53:29 -0400 Subject: [PATCH 02/10] docs(agent): remove unavailable guard requirement --- AGENTS.md | 27 --------------------------- plips/plip-10-lsp-system.md | 2 +- 2 files changed, 1 insertion(+), 28 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 81716d72..1929beff 100644 --- a/AGENTS.md +++ b/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. diff --git a/plips/plip-10-lsp-system.md b/plips/plip-10-lsp-system.md index c68d13b2..3d93f4ea 100644 --- a/plips/plip-10-lsp-system.md +++ b/plips/plip-10-lsp-system.md @@ -837,7 +837,7 @@ make test-pythinker-code`. | --- | --- | | 0–4 (per phase) | `make check-pythinker-code && make test-pythinker-code` | | Tool list change | rebuild wire-handshake snapshot in `tests_e2e/` (`--inline-snapshot=fix`) | -| Before PR | `## Unreleased` CHANGELOG entry; `pythinker-guard` skill; CodeRabbit green | +| Before PR | `## Unreleased` CHANGELOG entry; C01–C15 review; CodeRabbit green | ## File-by-file checklist (the complete port, in build order) From b3c1287d4cc56421dbedac4d80758ca74fa1134c Mon Sep 17 00:00:00 2001 From: elkaix Date: Tue, 14 Jul 2026 21:59:32 -0400 Subject: [PATCH 03/10] fix(core): correlate streamed tool-call fragments --- .../pythinker_core/chat_provider/__init__.py | 38 ++ .../src/pythinker_core/message.py | 6 +- .../stream_message_assembler.py | 284 +++++++++++++ packages/pythinker-core/tests/test_message.py | 23 ++ .../tests/test_stream_message_assembler.py | 377 ++++++++++++++++++ 5 files changed, 727 insertions(+), 1 deletion(-) create mode 100644 packages/pythinker-core/src/pythinker_core/stream_message_assembler.py create mode 100644 packages/pythinker-core/tests/test_stream_message_assembler.py diff --git a/packages/pythinker-core/src/pythinker_core/chat_provider/__init__.py b/packages/pythinker-core/src/pythinker_core/chat_provider/__init__.py index 7cb88609..440f951d 100644 --- a/packages/pythinker-core/src/pythinker_core/chat_provider/__init__.py +++ b/packages/pythinker-core/src/pythinker_core/chat_provider/__init__.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.""" diff --git a/packages/pythinker-core/src/pythinker_core/message.py b/packages/pythinker-core/src/pythinker_core/message.py index 00c49963..bfcfc415 100644 --- a/packages/pythinker-core/src/pythinker_core/message.py +++ b/packages/pythinker-core/src/pythinker_core/message.py @@ -1,7 +1,7 @@ from abc import ABC from typing import Any, ClassVar, Literal, cast, override -from pydantic import BaseModel, GetCoreSchemaHandler, field_serializer, field_validator +from pydantic import BaseModel, Field, GetCoreSchemaHandler, field_serializer, field_validator from pydantic_core import core_schema from pythinker_core.utils.typing import JsonType @@ -198,6 +198,7 @@ class FunctionBody(BaseModel): """The function body of the tool call.""" extras: dict[str, JsonType] | None = None """Extra information about the tool call.""" + stream_index: int | None = Field(default=None, exclude=True, repr=False) @override def merge_in_place(self, other: Any) -> bool: @@ -215,6 +216,9 @@ class ToolCallPart(BaseModel, MergeableMixin): arguments_part: str | None = None """A part of the arguments of the tool call.""" + name_part: str | None = Field(default=None, exclude=True, repr=False) + stream_index: int | None = Field(default=None, exclude=True, repr=False) + stream_call_id: str | None = Field(default=None, exclude=True, repr=False) @override def merge_in_place(self, other: Any) -> bool: diff --git a/packages/pythinker-core/src/pythinker_core/stream_message_assembler.py b/packages/pythinker-core/src/pythinker_core/stream_message_assembler.py new file mode 100644 index 00000000..66f25515 --- /dev/null +++ b/packages/pythinker-core/src/pythinker_core/stream_message_assembler.py @@ -0,0 +1,284 @@ +from __future__ import annotations + +import hashlib +from dataclasses import dataclass, field + +from pythinker_core.chat_provider import ( + APIStreamProtocolError, + StreamedMessagePart, + StreamProtocolErrorCategory, +) +from pythinker_core.message import ContentPart, Message, ToolCall, ToolCallPart +from pythinker_core.utils.typing import JsonType + +_ALLOWED_TOOL_FINISH_REASONS = { + None, + "stop", + "tool_calls", + "function_call", + "tool_use", + "end_turn", + "stop_sequence", + "completed", +} + + +@dataclass +class _CallState: + stream_index: int | None = None + call_id: str | None = None + started: bool = False + complete_name: str | None = None + name_parts: list[str] = field(default_factory=list[str]) + argument_parts: list[str] = field(default_factory=list[str]) + extras: dict[str, JsonType] | None = None + + +class StreamMessageAssembler: + """Correlate provider-neutral streamed message parts into one assistant message.""" + + def __init__(self) -> None: + self._content: list[ContentPart] = [] + self._calls: list[_CallState] = [] + self._calls_by_index: dict[int, _CallState] = {} + self._calls_by_id: dict[str, _CallState] = {} + + def add(self, part: StreamedMessagePart) -> None: + if isinstance(part, ToolCall): + self._add_call(part) + return + if isinstance(part, ToolCallPart): + self._add_call_part(part) + return + if not self._content or not self._content[-1].merge_in_place(part): + self._content.append(part) + + def finish( + self, + *, + response_id: str | None, + finish_reason: str | None, + ) -> Message: + orphan = next((state for state in self._calls if not state.started), None) + if orphan is not None: + self._raise( + "orphan_fragment", + response_id=response_id, + state=orphan, + ) + + started = [state for state in self._calls if state.started] + if started: + self._validate_finish_reason(finish_reason, response_id=response_id) + + ordered = ( + sorted(started, key=self._indexed_order) + if started and all(state.stream_index is not None for state in started) + else started + ) + calls = [ + self._finalize_call(state, response_id=response_id, order=order) + for order, state in enumerate(ordered) + ] + return Message(role="assistant", content=self._content, tool_calls=calls or None) + + def _add_call(self, call: ToolCall) -> None: + call_id = self._nonblank(call.id) + state = self._find_correlated(call.stream_index, call_id) + if state is None: + self._validate_new_start_mode(call.stream_index) + state = self._new_state() + elif not state.started: + self._validate_new_start_mode(call.stream_index) + + self._bind_index(state, call.stream_index, call_id=call_id) + self._bind_id(state, call_id, stream_index=call.stream_index) + + complete_name = self._nonblank(call.function.name) + if state.complete_name is not None and complete_name is not None: + if state.complete_name != complete_name: + self._raise("conflicting_name", state=state) + elif complete_name is not None: + state.complete_name = complete_name + + if call.function.arguments is not None: + state.argument_parts.append(call.function.arguments) + if state.extras is None: + state.extras = call.extras + state.started = True + + def _add_call_part(self, part: ToolCallPart) -> None: + call_id = self._nonblank(part.stream_call_id) + keyed = part.stream_index is not None or call_id is not None + state = self._find_correlated(part.stream_index, call_id) + + if state is None and keyed: + state = self._new_state() + elif state is None: + started = [candidate for candidate in self._calls if candidate.started] + if len(started) > 1: + raise APIStreamProtocolError("ambiguous_fragment") + if not started: + raise APIStreamProtocolError("orphan_fragment") + state = started[0] + + self._bind_index(state, part.stream_index, call_id=call_id) + self._bind_id(state, call_id, stream_index=part.stream_index) + if part.name_part is not None: + state.name_parts.append(part.name_part) + if part.arguments_part is not None: + state.argument_parts.append(part.arguments_part) + + def _find_correlated(self, stream_index: int | None, call_id: str | None) -> _CallState | None: + indexed = self._calls_by_index.get(stream_index) if stream_index is not None else None + identified = self._calls_by_id.get(call_id) if call_id is not None else None + if indexed is not None and identified is not None and indexed is not identified: + raise APIStreamProtocolError( + "conflicting_identity", + stream_index=stream_index, + call_id=call_id, + ) + if indexed is not None: + return indexed + if identified is not None: + if ( + stream_index is not None + and identified.stream_index is not None + and identified.stream_index != stream_index + ): + raise APIStreamProtocolError( + "conflicting_identity", + stream_index=stream_index, + call_id=call_id, + ) + return identified + return None + + def _new_state(self) -> _CallState: + state = _CallState() + self._calls.append(state) + return state + + def _bind_index( + self, + state: _CallState, + stream_index: int | None, + *, + call_id: str | None, + ) -> None: + if stream_index is None: + return + if state.stream_index is not None and state.stream_index != stream_index: + raise APIStreamProtocolError( + "conflicting_identity", + stream_index=stream_index, + call_id=call_id, + ) + existing = self._calls_by_index.get(stream_index) + if existing is not None and existing is not state: + raise APIStreamProtocolError( + "conflicting_identity", + stream_index=stream_index, + call_id=call_id, + ) + state.stream_index = stream_index + self._calls_by_index[stream_index] = state + + def _bind_id( + self, + state: _CallState, + call_id: str | None, + *, + stream_index: int | None, + ) -> None: + if call_id is None: + return + if state.call_id is not None and state.call_id != call_id: + raise APIStreamProtocolError( + "conflicting_identity", + stream_index=stream_index, + call_id=call_id, + ) + existing = self._calls_by_id.get(call_id) + if existing is not None and existing is not state: + raise APIStreamProtocolError( + "conflicting_identity", + stream_index=stream_index, + call_id=call_id, + ) + state.call_id = call_id + self._calls_by_id[call_id] = state + + def _validate_new_start_mode(self, stream_index: int | None) -> None: + started = [state for state in self._calls if state.started] + if started and (started[0].stream_index is None) != (stream_index is None): + raise APIStreamProtocolError( + "mixed_correlation", + stream_index=stream_index, + ) + + def _validate_finish_reason( + self, finish_reason: str | None, *, response_id: str | None + ) -> None: + normalized = finish_reason.strip().lower() if finish_reason else None + if normalized in _ALLOWED_TOOL_FINISH_REASONS: + return + category: StreamProtocolErrorCategory = ( + "truncated_tool_call" if normalized == "length" else "terminal_failure" + ) + raise APIStreamProtocolError(category, response_id=response_id) + + def _finalize_call( + self, + state: _CallState, + *, + response_id: str | None, + order: int, + ) -> ToolCall: + fragmented_name = "".join(state.name_parts) + if state.complete_name is not None and fragmented_name: + if state.complete_name != fragmented_name: + self._raise("conflicting_name", response_id=response_id, state=state) + name = state.complete_name + else: + name = state.complete_name or fragmented_name + if not name: + self._raise("missing_name", response_id=response_id, state=state) + + arguments = "".join(state.argument_parts) or "{}" + call_id = state.call_id + if call_id is None: + seed = f"{response_id or 'local'}:{order}:{name}:{arguments}" + digest = hashlib.sha256(seed.encode(encoding="utf-8")).hexdigest()[:20] + call_id = f"call_{digest}" + + return ToolCall( + id=call_id, + function=ToolCall.FunctionBody(name=name, arguments=arguments), + extras=state.extras, + ) + + @staticmethod + def _indexed_order(state: _CallState) -> int: + assert state.stream_index is not None + return state.stream_index + + @staticmethod + def _nonblank(value: str | None) -> str | None: + if value is None or not value.strip(): + return None + return value + + @staticmethod + def _raise( + category: StreamProtocolErrorCategory, + *, + response_id: str | None = None, + state: _CallState, + ) -> None: + raise APIStreamProtocolError( + category, + response_id=response_id, + stream_index=state.stream_index, + call_id=state.call_id, + ) diff --git a/packages/pythinker-core/tests/test_message.py b/packages/pythinker-core/tests/test_message.py index 69c8e9fc..8a51ae55 100644 --- a/packages/pythinker-core/tests/test_message.py +++ b/packages/pythinker-core/tests/test_message.py @@ -7,6 +7,7 @@ TextPart, ThinkPart, ToolCall, + ToolCallPart, VideoURLPart, ) @@ -63,6 +64,28 @@ def test_message_with_tool_calls(): assert Message.model_validate(dumped) == message +def test_tool_call_correlation_metadata_is_excluded_from_serialization(): + call = ToolCall( + id="call_1", + function=ToolCall.FunctionBody(name="read", arguments="{}"), + stream_index=3, + ) + part = ToolCallPart( + arguments_part="{}", + name_part="read", + stream_index=3, + stream_call_id="call_1", + ) + + assert call.model_dump() == { + "type": "function", + "id": "call_1", + "function": {"name": "read", "arguments": "{}"}, + "extras": None, + } + assert part.model_dump() == {"arguments_part": "{}"} + + def test_message_with_no_content(): message = Message( role="assistant", diff --git a/packages/pythinker-core/tests/test_stream_message_assembler.py b/packages/pythinker-core/tests/test_stream_message_assembler.py new file mode 100644 index 00000000..db57a1a6 --- /dev/null +++ b/packages/pythinker-core/tests/test_stream_message_assembler.py @@ -0,0 +1,377 @@ +import pytest + +from pythinker_core.chat_provider import APIStreamProtocolError +from pythinker_core.message import TextPart, ToolCall, ToolCallPart +from pythinker_core.stream_message_assembler import StreamMessageAssembler + + +def test_interleaved_indexed_tool_calls_assemble_independently() -> None: + assembler = StreamMessageAssembler() + assembler.add( + ToolCall( + id="call_a", + function=ToolCall.FunctionBody(name="first", arguments=""), + stream_index=0, + ) + ) + assembler.add( + ToolCall( + id="call_b", + function=ToolCall.FunctionBody(name="second", arguments=""), + stream_index=1, + ) + ) + assembler.add(ToolCallPart(arguments_part='{"x":', stream_index=0)) + assembler.add(ToolCallPart(arguments_part='{"y":', stream_index=1)) + assembler.add(ToolCallPart(arguments_part="1}", stream_index=0)) + assembler.add(ToolCallPart(arguments_part="2}", stream_index=1)) + + message = assembler.finish(response_id="resp_1", finish_reason="tool_calls") + + assert message.tool_calls == [ + ToolCall( + id="call_a", + function=ToolCall.FunctionBody(name="first", arguments='{"x":1}'), + ), + ToolCall( + id="call_b", + function=ToolCall.FunctionBody(name="second", arguments='{"y":2}'), + ), + ] + + +def test_indexed_calls_return_in_ascending_index_order() -> None: + assembler = StreamMessageAssembler() + assembler.add( + ToolCall( + id="call_2", + function=ToolCall.FunctionBody(name="second", arguments="{}"), + stream_index=2, + ) + ) + assembler.add( + ToolCall( + id="call_0", + function=ToolCall.FunctionBody(name="first", arguments="{}"), + stream_index=0, + ) + ) + + message = assembler.finish(response_id=None, finish_reason="stop") + + assert [call.id for call in message.tool_calls or []] == ["call_0", "call_2"] + + +def test_id_only_calls_preserve_first_seen_order() -> None: + assembler = StreamMessageAssembler() + assembler.add( + ToolCall(id="second", function=ToolCall.FunctionBody(name="second", arguments="{}")) + ) + assembler.add( + ToolCall(id="first", function=ToolCall.FunctionBody(name="first", arguments="{}")) + ) + + message = assembler.finish(response_id=None, finish_reason="tool_calls") + + assert [call.id for call in message.tool_calls or []] == ["second", "first"] + + +def test_indexed_fragment_before_start_is_buffered() -> None: + assembler = StreamMessageAssembler() + assembler.add(ToolCallPart(arguments_part='{"value":', stream_index=4)) + assembler.add( + ToolCall( + id="late", + function=ToolCall.FunctionBody(name="late_start", arguments="1}"), + stream_index=4, + ) + ) + + message = assembler.finish(response_id=None, finish_reason="completed") + + assert message.tool_calls == [ + ToolCall( + id="late", + function=ToolCall.FunctionBody(name="late_start", arguments='{"value":1}'), + ) + ] + + +def test_repeated_matching_identity_and_complete_name_are_accepted() -> None: + assembler = StreamMessageAssembler() + assembler.add( + ToolCall( + id="same", + function=ToolCall.FunctionBody(name="lookup", arguments=""), + stream_index=0, + ) + ) + assembler.add( + ToolCall( + id="same", + function=ToolCall.FunctionBody(name="lookup", arguments="{}"), + stream_index=0, + ) + ) + + message = assembler.finish(response_id=None, finish_reason="function_call") + + assert message.tool_calls == [ + ToolCall(id="same", function=ToolCall.FunctionBody(name="lookup", arguments="{}")) + ] + + +def test_correlated_name_fragments_assemble_in_order() -> None: + assembler = StreamMessageAssembler() + assembler.add(ToolCallPart(name_part="look", stream_index=0)) + assembler.add( + ToolCall(id="call_1", function=ToolCall.FunctionBody(name="", arguments=""), stream_index=0) + ) + assembler.add(ToolCallPart(name_part="up", arguments_part="{}", stream_index=0)) + + message = assembler.finish(response_id=None, finish_reason="tool_use") + + assert message.tool_calls == [ + ToolCall(id="call_1", function=ToolCall.FunctionBody(name="lookup", arguments="{}")) + ] + + +def test_complete_and_fragmented_name_must_agree() -> None: + assembler = StreamMessageAssembler() + assembler.add( + ToolCall( + id="call_1", + function=ToolCall.FunctionBody(name="lookup", arguments=""), + stream_index=0, + ) + ) + assembler.add(ToolCallPart(name_part="look", stream_index=0)) + assembler.add(ToolCallPart(name_part="up", arguments_part="{}", stream_index=0)) + + message = assembler.finish(response_id=None, finish_reason="end_turn") + + assert message.tool_calls == [ + ToolCall(id="call_1", function=ToolCall.FunctionBody(name="lookup", arguments="{}")) + ] + + +def test_missing_arguments_normalize_to_empty_object() -> None: + assembler = StreamMessageAssembler() + assembler.add( + ToolCall(id="call_1", function=ToolCall.FunctionBody(name="lookup", arguments=None)) + ) + + message = assembler.finish(response_id=None, finish_reason=None) + + assert message.tool_calls == [ + ToolCall(id="call_1", function=ToolCall.FunctionBody(name="lookup", arguments="{}")) + ] + + +def test_missing_provider_id_gets_deterministic_message_local_id() -> None: + first = StreamMessageAssembler() + second = StreamMessageAssembler() + for assembler in (first, second): + assembler.add( + ToolCall(id="", function=ToolCall.FunctionBody(name="lookup", arguments="{}")) + ) + + first_message = first.finish(response_id="response", finish_reason="stop_sequence") + second_message = second.finish(response_id="response", finish_reason="stop_sequence") + + assert first_message.tool_calls == second_message.tool_calls + assert first_message.tool_calls == [ + ToolCall( + id="call_8cbfc88f6a403717b878", + function=ToolCall.FunctionBody(name="lookup", arguments="{}"), + ) + ] + + +def test_late_provider_id_replaces_missing_indexed_identity() -> None: + assembler = StreamMessageAssembler() + assembler.add( + ToolCall(id="", function=ToolCall.FunctionBody(name="lookup", arguments=""), stream_index=0) + ) + assembler.add(ToolCallPart(arguments_part="{}", stream_index=0, stream_call_id="provider_id")) + + message = assembler.finish(response_id=None, finish_reason="tool_calls") + + assert message.tool_calls == [ + ToolCall(id="provider_id", function=ToolCall.FunctionBody(name="lookup", arguments="{}")) + ] + + +def test_conflicting_late_provider_id_is_rejected() -> None: + assembler = StreamMessageAssembler() + assembler.add( + ToolCall(id="", function=ToolCall.FunctionBody(name="lookup", arguments=""), stream_index=0) + ) + assembler.add(ToolCallPart(stream_index=0, stream_call_id="first")) + + with pytest.raises(APIStreamProtocolError) as caught: + assembler.add( + ToolCallPart(arguments_part="secret", stream_index=0, stream_call_id="second") + ) + + assert caught.value.category == "conflicting_identity" + assert "secret" not in str(caught.value) + + +def test_unindexed_fragment_with_two_open_calls_is_ambiguous() -> None: + assembler = StreamMessageAssembler() + assembler.add(ToolCall(id="a", function=ToolCall.FunctionBody(name="first", arguments=""))) + assembler.add(ToolCall(id="b", function=ToolCall.FunctionBody(name="second", arguments=""))) + + with pytest.raises(APIStreamProtocolError) as caught: + assembler.add(ToolCallPart(arguments_part="{}")) + + assert caught.value.category == "ambiguous_fragment" + assert caught.value.call_id is None + assert "{}" not in str(caught.value) + + +def test_conflicting_index_and_id_association_is_rejected() -> None: + assembler = StreamMessageAssembler() + assembler.add( + ToolCall( + id="call_a", + function=ToolCall.FunctionBody(name="first", arguments=""), + stream_index=0, + ) + ) + + with pytest.raises(APIStreamProtocolError) as caught: + assembler.add( + ToolCallPart(arguments_part="private", stream_index=1, stream_call_id="call_a") + ) + + assert caught.value.category == "conflicting_identity" + assert caught.value.stream_index == 1 + assert caught.value.call_id == "call_a" + assert "private" not in str(caught.value) + + +def test_conflicting_complete_and_fragmented_name_is_rejected() -> None: + assembler = StreamMessageAssembler() + assembler.add( + ToolCall( + id="call_1", + function=ToolCall.FunctionBody(name="first", arguments="{}"), + stream_index=0, + ) + ) + assembler.add(ToolCallPart(name_part="second", stream_index=0)) + + with pytest.raises(APIStreamProtocolError) as caught: + assembler.finish(response_id=None, finish_reason="tool_calls") + + assert caught.value.category == "conflicting_name" + + +def test_conflicting_complete_name_is_rejected() -> None: + assembler = StreamMessageAssembler() + assembler.add( + ToolCall( + id="call_1", + function=ToolCall.FunctionBody(name="first", arguments=""), + stream_index=0, + ) + ) + + with pytest.raises(APIStreamProtocolError) as caught: + assembler.add( + ToolCall( + id="call_1", + function=ToolCall.FunctionBody(name="second", arguments="private"), + stream_index=0, + ) + ) + + assert caught.value.category == "conflicting_name" + assert "private" not in str(caught.value) + + +def test_unresolved_keyed_fragment_is_rejected_at_finish() -> None: + assembler = StreamMessageAssembler() + assembler.add(ToolCallPart(arguments_part="private", stream_index=7, stream_call_id="missing")) + + with pytest.raises(APIStreamProtocolError) as caught: + assembler.finish(response_id="response", finish_reason="tool_calls") + + assert caught.value.category == "orphan_fragment" + assert caught.value.response_id == "response" + assert caught.value.stream_index == 7 + assert caught.value.call_id == "missing" + assert "private" not in str(caught.value) + + +def test_mixed_indexed_and_unindexed_multi_call_starts_are_rejected() -> None: + assembler = StreamMessageAssembler() + assembler.add( + ToolCall( + id="indexed", + function=ToolCall.FunctionBody(name="first", arguments="{}"), + stream_index=0, + ) + ) + + with pytest.raises(APIStreamProtocolError) as caught: + assembler.add( + ToolCall(id="legacy", function=ToolCall.FunctionBody(name="second", arguments="{}")) + ) + + assert caught.value.category == "mixed_correlation" + + +def test_empty_function_name_is_rejected() -> None: + assembler = StreamMessageAssembler() + assembler.add(ToolCall(id="call_1", function=ToolCall.FunctionBody(name="", arguments="{}"))) + + with pytest.raises(APIStreamProtocolError) as caught: + assembler.finish(response_id=None, finish_reason="tool_calls") + + assert caught.value.category == "missing_name" + + +@pytest.mark.parametrize( + ("finish_reason", "category"), + [ + ("length", "truncated_tool_call"), + ("content_filter", "terminal_failure"), + ("incomplete", "terminal_failure"), + ("failed", "terminal_failure"), + ("cancelled", "terminal_failure"), + ("network_error", "terminal_failure"), + ("model_context_window_exceeded", "terminal_failure"), + ("pause_turn", "terminal_failure"), + ("refusal", "terminal_failure"), + ("sensitive", "terminal_failure"), + ("provider_new_reason", "terminal_failure"), + ], +) +def test_terminal_failure_reasons_are_rejected(finish_reason: str, category: str) -> None: + assembler = StreamMessageAssembler() + assembler.add( + ToolCall(id="call_1", function=ToolCall.FunctionBody(name="lookup", arguments="private")) + ) + + with pytest.raises(APIStreamProtocolError) as caught: + assembler.finish(response_id="response", finish_reason=finish_reason) + + assert caught.value.category == category + assert caught.value.output_published is False + assert "private" not in str(caught.value) + + +def test_content_parts_assemble_independently_from_tool_calls() -> None: + assembler = StreamMessageAssembler() + assembler.add(TextPart(text="hello ")) + assembler.add( + ToolCall(id="call_1", function=ToolCall.FunctionBody(name="lookup", arguments="{}")) + ) + assembler.add(TextPart(text="world")) + + message = assembler.finish(response_id=None, finish_reason="completed") + + assert message.content == [TextPart(text="hello world")] From 5c4a7abe7dc8d55b6675e96076cddd77545a69f1 Mon Sep 17 00:00:00 2001 From: elkaix Date: Tue, 14 Jul 2026 22:08:49 -0400 Subject: [PATCH 04/10] fix(core): preserve streamed tool-call identity --- .../pythinker_core/chat_provider/pythinker.py | 3 + .../contrib/chat_provider/anthropic.py | 6 +- .../contrib/chat_provider/openai_legacy.py | 3 + .../contrib/chat_provider/openai_responses.py | 46 ++- .../tests/test_stream_tool_call_metadata.py | 324 ++++++++++++++++++ 5 files changed, 364 insertions(+), 18 deletions(-) create mode 100644 packages/pythinker-core/tests/test_stream_tool_call_metadata.py diff --git a/packages/pythinker-core/src/pythinker_core/chat_provider/pythinker.py b/packages/pythinker-core/src/pythinker_core/chat_provider/pythinker.py index e3860e82..40f3db8d 100644 --- a/packages/pythinker-core/src/pythinker_core/chat_provider/pythinker.py +++ b/packages/pythinker-core/src/pythinker_core/chat_provider/pythinker.py @@ -493,10 +493,13 @@ async def _convert_stream_response( name=tool_call.function.name, arguments=tool_call.function.arguments, ), + stream_index=tool_call.index, ) elif tool_call.function.arguments: yield ToolCallPart( arguments_part=tool_call.function.arguments, + stream_index=tool_call.index, + stream_call_id=tool_call.id, ) else: # skip empty tool calls diff --git a/packages/pythinker-core/src/pythinker_core/contrib/chat_provider/anthropic.py b/packages/pythinker-core/src/pythinker_core/contrib/chat_provider/anthropic.py index 21905af2..40c2fa5a 100644 --- a/packages/pythinker-core/src/pythinker_core/contrib/chat_provider/anthropic.py +++ b/packages/pythinker-core/src/pythinker_core/contrib/chat_provider/anthropic.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": diff --git a/packages/pythinker-core/src/pythinker_core/contrib/chat_provider/openai_legacy.py b/packages/pythinker-core/src/pythinker_core/contrib/chat_provider/openai_legacy.py index d34de3c4..c5be7c6b 100644 --- a/packages/pythinker-core/src/pythinker_core/contrib/chat_provider/openai_legacy.py +++ b/packages/pythinker-core/src/pythinker_core/contrib/chat_provider/openai_legacy.py @@ -342,10 +342,13 @@ async def _convert_stream_response( name=tool_call.function.name, arguments=tool_call.function.arguments, ), + stream_index=tool_call.index, ) elif tool_call.function.arguments: yield ToolCallPart( arguments_part=tool_call.function.arguments, + stream_index=tool_call.index, + stream_call_id=tool_call.id, ) else: # skip empty tool calls diff --git a/packages/pythinker-core/src/pythinker_core/contrib/chat_provider/openai_responses.py b/packages/pythinker-core/src/pythinker_core/contrib/chat_provider/openai_responses.py index 39484d80..908bf6d5 100644 --- a/packages/pythinker-core/src/pythinker_core/contrib/chat_provider/openai_responses.py +++ b/packages/pythinker-core/src/pythinker_core/contrib/chat_provider/openai_responses.py @@ -7,8 +7,14 @@ from openai import AsyncStream, OpenAIError from openai.types.responses import ( Response, + ResponseCompletedEvent, + ResponseCreatedEvent, + ResponseFailedEvent, + ResponseFunctionCallArgumentsDeltaEvent, + ResponseIncompleteEvent, ResponseInputItemParam, ResponseInputParam, + ResponseOutputItemAddedEvent, ResponseOutputMessageParam, ResponseOutputTextParam, ResponseStreamEvent, @@ -455,13 +461,16 @@ def _map_audio_url_to_file_content(url: str) -> ResponseInputFileContentParam | def _responses_finish_reason(response: Response) -> str | None: """Map a Responses API terminal state to the loop's OpenAI-compatible finish reason. - The Responses API marks an output-token-capped reply with ``status='incomplete'`` and - ``incomplete_details.reason='max_output_tokens'``; surface that as ``'length'`` so the - loop's truncation recovery fires. Other terminal statuses pass through unchanged. + The Responses API marks an incomplete reply with a machine-readable reason. Normalize + output-token caps and content filtering to the finish reasons used by callers. Other terminal + statuses, including failed and cancelled, pass through unchanged. """ details = response.incomplete_details - if details is not None and details.reason == "max_output_tokens": - return "length" + if details is not None: + if details.reason == "max_output_tokens": + return "length" + if details.reason == "content_filter": + return "content_filter" return response.status @@ -537,11 +546,12 @@ async def _convert_stream_response( """Convert streaming Responses events into message parts.""" try: async for chunk in response: - if chunk.type == "response.output_text.delta": + if isinstance(chunk, ResponseCreatedEvent): + self._id = chunk.response.id + elif chunk.type == "response.output_text.delta": yield TextPart(text=chunk.delta) - elif chunk.type == "response.output_item.added": + elif isinstance(chunk, ResponseOutputItemAddedEvent): item = chunk.item - self._id = item.id if item.type == "function_call": yield ToolCall( id=item.call_id or str(uuid.uuid4()), @@ -549,24 +559,26 @@ async def _convert_stream_response( name=item.name, arguments=item.arguments, ), + stream_index=chunk.output_index, ) elif chunk.type == "response.output_item.done": item = chunk.item - self._id = item.id if item.type == "reasoning": yield ThinkPart(think="", encrypted=item.encrypted_content) - elif chunk.type == "response.function_call_arguments.delta": - yield ToolCallPart(arguments_part=chunk.delta) + elif isinstance(chunk, ResponseFunctionCallArgumentsDeltaEvent): + yield ToolCallPart( + arguments_part=chunk.delta, + stream_index=chunk.output_index, + ) elif chunk.type == "response.reasoning_summary_part.added": yield ThinkPart(think="") elif chunk.type == "response.reasoning_summary_text.delta": yield ThinkPart(think=chunk.delta) - elif chunk.type == "response.completed": - self._usage = chunk.response.usage - self._finish_reason = _responses_finish_reason(chunk.response) - elif chunk.type == "response.incomplete": - # The terminal incomplete event carries the max_output_tokens truncation - # (and final usage/status); kept separate so the event type narrows. + elif isinstance( + chunk, + (ResponseCompletedEvent, ResponseIncompleteEvent, ResponseFailedEvent), + ): + self._id = chunk.response.id self._usage = chunk.response.usage self._finish_reason = _responses_finish_reason(chunk.response) except (OpenAIError, httpx.HTTPError) as e: diff --git a/packages/pythinker-core/tests/test_stream_tool_call_metadata.py b/packages/pythinker-core/tests/test_stream_tool_call_metadata.py new file mode 100644 index 00000000..faf4bd82 --- /dev/null +++ b/packages/pythinker-core/tests/test_stream_tool_call_metadata.py @@ -0,0 +1,324 @@ +from collections.abc import AsyncIterator +from typing import Literal, cast + +import pytest +from anthropic import AsyncStream as AnthropicAsyncStream +from anthropic.types import ( + InputJSONDelta, + MessageDeltaEvent, + MessageDeltaUsage, + RawContentBlockDeltaEvent, + RawContentBlockStartEvent, + RawMessageStreamEvent, + ToolUseBlock, +) +from anthropic.types.raw_message_delta_event import Delta +from openai import AsyncStream +from openai.types.chat import ChatCompletionChunk +from openai.types.responses import ( + Response, + ResponseCreatedEvent, + ResponseFailedEvent, + ResponseFunctionCallArgumentsDeltaEvent, + ResponseFunctionToolCall, + ResponseOutputItemAddedEvent, + ResponseStreamEvent, +) +from openai.types.responses.response import IncompleteDetails + +from pythinker_core.chat_provider.pythinker import PythinkerStreamedMessage +from pythinker_core.contrib.chat_provider.anthropic import AnthropicStreamedMessage +from pythinker_core.contrib.chat_provider.openai_legacy import OpenAILegacyStreamedMessage +from pythinker_core.contrib.chat_provider.openai_responses import OpenAIResponsesStreamedMessage +from pythinker_core.message import ToolCall, ToolCallPart + + +async def _async_events[T](*events: T) -> AsyncIterator[T]: + for event in events: + yield event + + +def _chat_chunk(*, tool_calls: list[dict[str, object]]) -> ChatCompletionChunk: + return ChatCompletionChunk.model_validate( + { + "id": "response_1", + "object": "chat.completion.chunk", + "created": 1, + "model": "provider-model", + "choices": [ + { + "index": 0, + "delta": {"tool_calls": tool_calls}, + "finish_reason": None, + } + ], + } + ) + + +def _response( + *, + response_id: str = "response_1", + status: Literal["completed", "failed", "cancelled", "incomplete"] = "completed", + incomplete_reason: Literal["max_output_tokens", "content_filter"] | None = None, +) -> Response: + return Response( + id=response_id, + created_at=1, + model="gpt-5", + object="response", + output=[], + parallel_tool_calls=True, + tool_choice="auto", + tools=[], + status=status, + incomplete_details=( + IncompleteDetails(reason=incomplete_reason) if incomplete_reason is not None else None + ), + ) + + +async def _collect(stream: object) -> list[ToolCall | ToolCallPart]: + return [part async for part in cast(AsyncIterator[ToolCall | ToolCallPart], stream)] + + +async def test_openai_legacy_preserves_tool_call_index_and_id() -> None: + chunks = _async_events( + _chat_chunk( + tool_calls=[ + { + "index": 0, + "id": "call_1", + "type": "function", + "function": {"name": "read", "arguments": ""}, + } + ] + ), + _chat_chunk( + tool_calls=[ + { + "index": 0, + "id": "call_1", + "type": "function", + "function": {"name": None, "arguments": '{"path":"a.py"}'}, + } + ] + ), + ) + stream = OpenAILegacyStreamedMessage( + cast(AsyncStream[ChatCompletionChunk], chunks), reasoning_key=None + ) + + assert await _collect(stream) == [ + ToolCall( + id="call_1", + function=ToolCall.FunctionBody(name="read", arguments=""), + stream_index=0, + ), + ToolCallPart( + arguments_part='{"path":"a.py"}', + stream_index=0, + stream_call_id="call_1", + ), + ] + + +async def test_pythinker_preserves_tool_call_index_and_id() -> None: + chunks = _async_events( + _chat_chunk( + tool_calls=[ + { + "index": 0, + "id": "call_1", + "type": "function", + "function": {"name": "read", "arguments": ""}, + } + ] + ), + _chat_chunk( + tool_calls=[ + { + "index": 0, + "id": "call_1", + "type": "function", + "function": {"name": None, "arguments": '{"path":"a.py"}'}, + } + ] + ), + ) + stream = PythinkerStreamedMessage(cast(AsyncStream[ChatCompletionChunk], chunks)) + + assert await _collect(stream) == [ + ToolCall( + id="call_1", + function=ToolCall.FunctionBody(name="read", arguments=""), + stream_index=0, + ), + ToolCallPart( + arguments_part='{"path":"a.py"}', + stream_index=0, + stream_call_id="call_1", + ), + ] + + +async def test_openai_responses_uses_output_index_and_semantic_call_id() -> None: + item = ResponseFunctionToolCall( + arguments="", + call_id="call_1", + id="item_1", + name="read", + status="in_progress", + type="function_call", + ) + events = _async_events( + ResponseOutputItemAddedEvent( + item=item, + output_index=3, + sequence_number=1, + type="response.output_item.added", + ), + ResponseFunctionCallArgumentsDeltaEvent( + delta='{"path":"a.py"}', + item_id="item_1", + output_index=3, + sequence_number=2, + type="response.function_call_arguments.delta", + ), + ) + stream = OpenAIResponsesStreamedMessage(cast(AsyncStream[ResponseStreamEvent], events)) + + assert await _collect(stream) == [ + ToolCall( + id="call_1", + function=ToolCall.FunctionBody(name="read", arguments=""), + stream_index=3, + ), + ToolCallPart( + arguments_part='{"path":"a.py"}', + stream_index=3, + stream_call_id=None, + ), + ] + + +async def test_openai_responses_keeps_response_id_separate_from_item_id() -> None: + events = _async_events( + ResponseCreatedEvent( + response=_response(response_id="response_created"), + sequence_number=0, + type="response.created", + ), + ResponseOutputItemAddedEvent( + item=ResponseFunctionToolCall( + arguments="", + call_id="semantic_call", + id="output_item", + name="read", + status="in_progress", + type="function_call", + ), + output_index=0, + sequence_number=1, + type="response.output_item.added", + ), + ResponseFailedEvent( + response=_response(response_id="response_terminal", status="failed"), + sequence_number=2, + type="response.failed", + ), + ) + stream = OpenAIResponsesStreamedMessage(cast(AsyncStream[ResponseStreamEvent], events)) + + parts = await _collect(stream) + + assert isinstance(parts[0], ToolCall) + assert parts[0].id == "semantic_call" + assert stream.id == "response_terminal" + + +class _AnthropicEventStream: + def __init__(self, *events: RawMessageStreamEvent): + self._events = _async_events(*events) + + async def __aenter__(self) -> "_AnthropicEventStream": + return self + + async def __aexit__(self, *_args: object) -> None: + return None + + def __aiter__(self) -> AsyncIterator[RawMessageStreamEvent]: + return self._events + + +async def test_anthropic_preserves_content_block_index() -> None: + start = RawContentBlockStartEvent( + type="content_block_start", + index=4, + content_block=ToolUseBlock( + type="tool_use", + id="call_1", + name="read", + input={}, + ), + ) + delta = RawContentBlockDeltaEvent( + type="content_block_delta", + index=4, + delta=InputJSONDelta( + type="input_json_delta", + partial_json='{"path":"a.py"}', + ), + ) + manager = _AnthropicEventStream(start, delta) + stream = AnthropicStreamedMessage(cast(AnthropicAsyncStream[RawMessageStreamEvent], manager)) + + assert await _collect(stream) == [ + ToolCall( + id="call_1", + function=ToolCall.FunctionBody(name="read", arguments=""), + stream_index=4, + ), + ToolCallPart(arguments_part='{"path":"a.py"}', stream_index=4), + ] + + +@pytest.mark.parametrize( + ("response", "expected_reason"), + [ + (_response(status="incomplete", incomplete_reason="max_output_tokens"), "length"), + (_response(status="incomplete", incomplete_reason="content_filter"), "content_filter"), + (_response(status="failed"), "failed"), + (_response(status="cancelled"), "cancelled"), + ], +) +async def test_responses_normalizes_incomplete_failed_and_cancelled_reasons( + response: Response, expected_reason: str +) -> None: + stream = OpenAIResponsesStreamedMessage(response) + + async for _ in stream: + pass + + assert stream.finish_reason == expected_reason + + +@pytest.mark.parametrize( + ("stop_reason", "expected_reason"), + [("pause_turn", "pause_turn"), ("refusal", "refusal")], +) +async def test_anthropic_exposes_pause_and_refusal_terminal_reasons( + stop_reason: Literal["pause_turn", "refusal"], expected_reason: str +) -> None: + event = MessageDeltaEvent( + type="message_delta", + delta=Delta(stop_reason=stop_reason, stop_sequence=None), + usage=MessageDeltaUsage(output_tokens=1), + ) + manager = _AnthropicEventStream(event) + stream = AnthropicStreamedMessage(cast(AnthropicAsyncStream[RawMessageStreamEvent], manager)) + + async for _ in stream: + pass + + assert stream.finish_reason == expected_reason From 74243085c47c24d16251d42db8f6a09674116b25 Mon Sep 17 00:00:00 2001 From: elkaix Date: Tue, 14 Jul 2026 22:12:40 -0400 Subject: [PATCH 05/10] fix(core): preserve responses terminal reason precedence --- .../contrib/chat_provider/openai_responses.py | 2 +- .../tests/test_stream_tool_call_metadata.py | 18 ++++++++++++++++++ 2 files changed, 19 insertions(+), 1 deletion(-) diff --git a/packages/pythinker-core/src/pythinker_core/contrib/chat_provider/openai_responses.py b/packages/pythinker-core/src/pythinker_core/contrib/chat_provider/openai_responses.py index 908bf6d5..689952bf 100644 --- a/packages/pythinker-core/src/pythinker_core/contrib/chat_provider/openai_responses.py +++ b/packages/pythinker-core/src/pythinker_core/contrib/chat_provider/openai_responses.py @@ -466,7 +466,7 @@ def _responses_finish_reason(response: Response) -> str | None: statuses, including failed and cancelled, pass through unchanged. """ details = response.incomplete_details - if details is not None: + if response.status == "incomplete" and details is not None: if details.reason == "max_output_tokens": return "length" if details.reason == "content_filter": diff --git a/packages/pythinker-core/tests/test_stream_tool_call_metadata.py b/packages/pythinker-core/tests/test_stream_tool_call_metadata.py index faf4bd82..79b61927 100644 --- a/packages/pythinker-core/tests/test_stream_tool_call_metadata.py +++ b/packages/pythinker-core/tests/test_stream_tool_call_metadata.py @@ -303,6 +303,24 @@ async def test_responses_normalizes_incomplete_failed_and_cancelled_reasons( assert stream.finish_reason == expected_reason +@pytest.mark.parametrize( + ("response", "expected_reason"), + [ + (_response(status="failed", incomplete_reason="max_output_tokens"), "failed"), + (_response(status="cancelled", incomplete_reason="content_filter"), "cancelled"), + ], +) +async def test_responses_explicit_status_takes_precedence_over_incomplete_details( + response: Response, expected_reason: str +) -> None: + stream = OpenAIResponsesStreamedMessage(response) + + async for _ in stream: + pass + + assert stream.finish_reason == expected_reason + + @pytest.mark.parametrize( ("stop_reason", "expected_reason"), [("pause_turn", "pause_turn"), ("refusal", "refusal")], From f49a0965887e9e355c64f935f9dcb59fdc2f281c Mon Sep 17 00:00:00 2001 From: elkaix Date: Tue, 14 Jul 2026 22:32:07 -0400 Subject: [PATCH 06/10] fix(core): defer tools until stream validation --- .../src/pythinker_core/__init__.py | 20 +- .../src/pythinker_core/_generate.py | 58 ++-- .../pythinker_core/chat_provider/pythinker.py | 30 +- .../contrib/chat_provider/openai_legacy.py | 30 +- .../pythinker-core/tests/test_generate.py | 220 +++++++++++- packages/pythinker-core/tests/test_step.py | 313 +++++++++++++++++- .../tests/test_stream_tool_call_metadata.py | 202 ++++++++++- 7 files changed, 807 insertions(+), 66 deletions(-) diff --git a/packages/pythinker-core/src/pythinker_core/__init__.py b/packages/pythinker-core/src/pythinker_core/__init__.py index ca78d017..786da3ce 100644 --- a/packages/pythinker-core/src/pythinker_core/__init__.py +++ b/packages/pythinker-core/src/pythinker_core/__init__.py @@ -15,7 +15,6 @@ from pythinker_core._generate import GenerateResult, generate from pythinker_core.chat_provider import ( ChatProvider, - ChatProviderError, StreamedMessagePart, TokenUsage, ) @@ -74,8 +73,11 @@ 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() @@ -83,7 +85,7 @@ def future_done_callback(future: ToolResultFuture): 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( diff --git a/packages/pythinker-core/src/pythinker_core/_generate.py b/packages/pythinker-core/src/pythinker_core/_generate.py index a5c12fb6..4e62957a 100644 --- a/packages/pythinker-core/src/pythinker_core/_generate.py +++ b/packages/pythinker-core/src/pythinker_core/_generate.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,8 +48,8 @@ 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) stream = await chat_provider.generate(system_prompt, tools, history) @@ -55,21 +57,30 @@ async def generate( logger.trace("Received part: {part}", part=part) 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 +123,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 diff --git a/packages/pythinker-core/src/pythinker_core/chat_provider/pythinker.py b/packages/pythinker-core/src/pythinker_core/chat_provider/pythinker.py index 40f3db8d..434fde31 100644 --- a/packages/pythinker-core/src/pythinker_core/chat_provider/pythinker.py +++ b/packages/pythinker-core/src/pythinker_core/chat_provider/pythinker.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,27 +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 diff --git a/packages/pythinker-core/src/pythinker_core/contrib/chat_provider/openai_legacy.py b/packages/pythinker-core/src/pythinker_core/contrib/chat_provider/openai_legacy.py index c5be7c6b..179683f2 100644 --- a/packages/pythinker-core/src/pythinker_core/contrib/chat_provider/openai_legacy.py +++ b/packages/pythinker-core/src/pythinker_core/contrib/chat_provider/openai_legacy.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,27 +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 diff --git a/packages/pythinker-core/tests/test_generate.py b/packages/pythinker-core/tests/test_generate.py index 3240eb87..37b7a793 100644 --- a/packages/pythinker-core/tests/test_generate.py +++ b/packages/pythinker-core/tests/test_generate.py @@ -4,7 +4,11 @@ import pytest from pythinker_core import generate -from pythinker_core.chat_provider import APIEmptyResponseError, StreamedMessagePart +from pythinker_core.chat_provider import ( + APIEmptyResponseError, + APIStreamProtocolError, + StreamedMessagePart, +) from pythinker_core.chat_provider.mock import MockChatProvider from pythinker_core.message import ImageURLPart, TextPart, ThinkPart, ToolCall, ToolCallPart @@ -86,7 +90,145 @@ async def on_tool_call(tool_call: ToolCall): assert output_tool_calls == message.tool_calls -def test_generate_marks_truncated_on_length_finish_reason(): +async def test_tool_callbacks_wait_for_successful_terminal_assembly() -> None: + parts: list[StreamedMessagePart] = [ + TextPart(text="working"), + ToolCall( + id="call_b", + function=ToolCall.FunctionBody(name="second", arguments=""), + stream_index=1, + ), + ToolCall( + id="call_a", + function=ToolCall.FunctionBody(name="first", arguments=""), + stream_index=0, + ), + ToolCallPart(arguments_part="{}", stream_index=1), + ToolCallPart(arguments_part="{}", stream_index=0), + ] + events: list[str] = [] + + async def on_part(part: StreamedMessagePart) -> None: + events.append(f"part:{type(part).__name__}") + + async def on_call(call: ToolCall) -> None: + events.append(f"call:{call.id}") + + result = await generate( + MockChatProvider(parts, finish_reason="tool_calls"), + "", + [], + [], + on_message_part=on_part, + on_tool_call=on_call, + ) + + assert events == [ + "part:TextPart", + "part:ToolCall", + "part:ToolCall", + "part:ToolCallPart", + "part:ToolCallPart", + "call:call_a", + "call:call_b", + ] + assert [call.id for call in result.message.tool_calls or []] == ["call_a", "call_b"] + + +async def test_message_part_callback_receives_defensive_copy() -> None: + source = ToolCall( + id="call_1", + function=ToolCall.FunctionBody(name="read", arguments="{}"), + stream_index=0, + ) + + async def mutate_callback(part: StreamedMessagePart) -> None: + if isinstance(part, ToolCall): + part.function.name = "mutated" + part.stream_index = 99 + + result = await generate( + MockChatProvider([source], finish_reason="tool_calls"), + "", + [], + [], + on_message_part=mutate_callback, + ) + + assert result.message.tool_calls == [ + ToolCall(id="call_1", function=ToolCall.FunctionBody(name="read", arguments="{}")) + ] + + +async def test_protocol_error_without_callback_marks_output_unpublished() -> None: + provider = MockChatProvider( + [ToolCall(id="call_1", function=ToolCall.FunctionBody(name="read", arguments=""))], + finish_reason="length", + ) + + with pytest.raises(APIStreamProtocolError) as caught: + await generate(provider, "", [], []) + + assert caught.value.output_published is False + assert caught.value.category == "truncated_tool_call" + + +async def test_protocol_error_after_successful_callback_marks_output_published() -> None: + provider = MockChatProvider( + [ToolCall(id="call_1", function=ToolCall.FunctionBody(name="read", arguments=""))], + finish_reason="length", + ) + published: list[StreamedMessagePart] = [] + + async def on_part(part: StreamedMessagePart) -> None: + published.append(part) + + with pytest.raises(APIStreamProtocolError) as caught: + await generate(provider, "", [], [], on_message_part=on_part) + + assert published + assert caught.value.output_published is True + assert caught.value.category == "truncated_tool_call" + + +async def test_callback_exception_does_not_become_protocol_error() -> None: + provider = MockChatProvider([TextPart(text="visible")], finish_reason="stop") + + async def fail_callback(_part: StreamedMessagePart) -> None: + raise RuntimeError("callback failed") + + with pytest.raises(RuntimeError, match="callback failed"): + await generate(provider, "", [], [], on_message_part=fail_callback) + + +async def test_add_time_protocol_error_gets_available_response_id() -> None: + provider = MockChatProvider( + [ + ToolCall( + id="call_1", + function=ToolCall.FunctionBody(name="read", arguments=""), + stream_index=0, + ), + ToolCallPart( + arguments_part="private arguments", + stream_index=1, + stream_call_id="call_1", + ), + ], + finish_reason="tool_calls", + ) + + with pytest.raises(APIStreamProtocolError) as caught: + await generate(provider, "", [], []) + + assert caught.value.category == "conflicting_identity" + assert caught.value.response_id == "mock" + assert caught.value.output_published is False + assert "private arguments" not in str(caught.value) + assert "reasoning" not in str(caught.value).lower() + + +def test_text_only_length_still_returns_truncated_result(): """A response cut off by the output-token limit (finish_reason 'length') sets GenerateResult.truncated so the agent loop can detect and recover from truncation.""" chat_provider = MockChatProvider( @@ -115,6 +257,80 @@ def test_generate_not_truncated_on_explicit_stop(): assert result.truncated is False +async def test_length_with_tool_state_raises_protocol_error() -> None: + provider = MockChatProvider( + [ToolCall(id="call_1", function=ToolCall.FunctionBody(name="read", arguments="{}"))], + finish_reason="length", + ) + + with pytest.raises(APIStreamProtocolError) as caught: + await generate(provider, "", [], []) + + assert caught.value.category == "truncated_tool_call" + + +@pytest.mark.parametrize( + "finish_reason", + [ + "content_filter", + "incomplete", + "failed", + "cancelled", + "network_error", + "model_context_window_exceeded", + "pause_turn", + "refusal", + "sensitive", + "provider_future_failure", + ], +) +async def test_each_supported_terminal_failure_reason_rejects_tool_state( + finish_reason: str, +) -> None: + calls: list[ToolCall] = [] + + async def on_call(call: ToolCall) -> None: + calls.append(call) + + provider = MockChatProvider( + [ToolCall(id="call_1", function=ToolCall.FunctionBody(name="read", arguments="{}"))], + finish_reason=finish_reason, + ) + + with pytest.raises(APIStreamProtocolError) as caught: + await generate(provider, "", [], [], on_tool_call=on_call) + + assert caught.value.category == "terminal_failure" + assert calls == [] + + +@pytest.mark.parametrize( + "finish_reason", + [ + None, + "stop", + "tool_calls", + "function_call", + "tool_use", + "end_turn", + "stop_sequence", + "completed", + ], +) +async def test_known_success_and_none_terminal_reasons_allow_complete_calls( + finish_reason: str | None, +) -> None: + provider = MockChatProvider( + [ToolCall(id="call_1", function=ToolCall.FunctionBody(name="read", arguments="{}"))], + finish_reason=finish_reason, + ) + + result = await generate(provider, "", [], []) + + assert [call.id for call in result.message.tool_calls or []] == ["call_1"] + assert result.truncated is False + + def test_generate_think_only_raises_error(): """Think-only response (no text, no tool calls) should raise APIEmptyResponseError.""" chat_provider = MockChatProvider( diff --git a/packages/pythinker-core/tests/test_step.py b/packages/pythinker-core/tests/test_step.py index caa6f884..4600741e 100644 --- a/packages/pythinker-core/tests/test_step.py +++ b/packages/pythinker-core/tests/test_step.py @@ -1,14 +1,125 @@ import asyncio -from typing import override +from collections.abc import AsyncIterator, Sequence +from typing import Self, override + +import pytest from pythinker_core import step -from pythinker_core.chat_provider import StreamedMessagePart +from pythinker_core.chat_provider import ( + APIConnectionError, + APIStreamProtocolError, + StreamedMessage, + StreamedMessagePart, + ThinkingEffort, + TokenUsage, +) from pythinker_core.chat_provider.mock import MockChatProvider -from pythinker_core.message import TextPart, ToolCall -from pythinker_core.tooling import CallableTool, ParametersType, ToolOk, ToolResult, ToolReturnValue +from pythinker_core.message import Message, TextPart, ToolCall, ToolCallPart +from pythinker_core.tooling import ( + CallableTool, + ParametersType, + Tool, + ToolOk, + ToolResult, + ToolResultFuture, + ToolReturnValue, +) +from pythinker_core.tooling.error import ToolParseError from pythinker_core.tooling.simple import SimpleToolset +class _RecordingToolset: + def __init__(self) -> None: + self.handle_count = 0 + + @property + def tools(self) -> list[Tool]: + return [] + + def handle(self, tool_call: ToolCall) -> ToolResult: + self.handle_count += 1 + return ToolResult(tool_call_id=tool_call.id, return_value=ToolOk(output="ok")) + + +class _ScriptedStream: + def __init__( + self, + parts: list[StreamedMessagePart], + *, + finish_reason: str | None = None, + terminal_error: BaseException | None = None, + block_after_parts: bool = False, + ) -> None: + self._parts = parts + self._finish_reason = finish_reason + self._terminal_error = terminal_error + self._block_after_parts = block_after_parts + self.entered_block = asyncio.Event() + self._never = asyncio.Event() + self._iter = self._stream() + + def __aiter__(self) -> AsyncIterator[StreamedMessagePart]: + return self + + async def __anext__(self) -> StreamedMessagePart: + return await self._iter.__anext__() + + async def _stream(self) -> AsyncIterator[StreamedMessagePart]: + for part in self._parts: + yield part + if self._block_after_parts: + self.entered_block.set() + await self._never.wait() + if self._terminal_error is not None: + raise self._terminal_error + + @property + def id(self) -> str: + return "scripted-response" + + @property + def usage(self) -> TokenUsage | None: + return None + + @property + def finish_reason(self) -> str | None: + return self._finish_reason + + +class _StaticProvider: + name = "static" + + def __init__(self, stream: StreamedMessage) -> None: + self._stream = stream + + @property + def model_name(self) -> str: + return "static" + + @property + def thinking_effort(self) -> ThinkingEffort | None: + return None + + async def generate( + self, + system_prompt: str, + tools: Sequence[Tool], + history: Sequence[Message], + ) -> StreamedMessage: + return self._stream + + def with_thinking(self, effort: ThinkingEffort) -> Self: + return self + + +def _tool_call(call_id: str, *, index: int) -> ToolCall: + return ToolCall( + id=call_id, + function=ToolCall.FunctionBody(name="tool", arguments="{}"), + stream_index=index, + ) + + def test_step(): class PlusTool(CallableTool): name: str = "plus" @@ -63,3 +174,197 @@ async def run(): assert output_parts == input_parts assert tool_results == [ToolResult(tool_call_id="plus#123", return_value=ToolOk(output="3"))] assert collected_tool_results == tool_results + + +async def test_structurally_complete_invalid_json_remains_tool_parse_error() -> None: + class NeverRunsTool(CallableTool): + name: str = "never_runs" + description: str = "Must not run when arguments are malformed." + parameters: ParametersType = {"type": "object", "properties": {}} + + @override + async def __call__(self) -> ToolReturnValue: + raise AssertionError("invalid JSON reached tool execution") + + provider = MockChatProvider( + [ + ToolCall( + id="call_1", + function=ToolCall.FunctionBody(name="never_runs", arguments="{not-json"), + ) + ], + finish_reason="tool_calls", + ) + + result = await step(provider, "", SimpleToolset([NeverRunsTool()]), []) + tool_results = await result.tool_results() + + assert len(tool_results) == 1 + assert isinstance(tool_results[0].return_value, ToolParseError) + + +@pytest.mark.parametrize( + "finish_reason", + [ + "length", + "content_filter", + "incomplete", + "failed", + "cancelled", + "network_error", + "model_context_window_exceeded", + "pause_turn", + "refusal", + "sensitive", + ], +) +async def test_terminal_tool_failure_starts_no_tools(finish_reason: str) -> None: + toolset = _RecordingToolset() + callbacks: list[ToolResult] = [] + provider = MockChatProvider([_tool_call("call_1", index=0)], finish_reason=finish_reason) + + with pytest.raises(APIStreamProtocolError): + await step(provider, "", toolset, [], on_tool_result=callbacks.append) + + assert toolset.handle_count == 0 + assert callbacks == [] + + +async def test_malformed_correlation_starts_no_tools_and_does_not_leak_state() -> None: + toolset = _RecordingToolset() + callbacks: list[ToolResult] = [] + malformed = MockChatProvider( + [ + _tool_call("call_1", index=0), + ToolCallPart( + arguments_part="private", + stream_index=1, + stream_call_id="call_1", + ), + ], + finish_reason="tool_calls", + ) + + with pytest.raises(APIStreamProtocolError): + await step(malformed, "", toolset, [], on_tool_result=callbacks.append) + + assert toolset.handle_count == 0 + assert callbacks == [] + + recovered = await step( + MockChatProvider([_tool_call("call_2", index=0)], finish_reason="tool_calls"), + "", + toolset, + [], + ) + assert await recovered.tool_results() == [ + ToolResult(tool_call_id="call_2", return_value=ToolOk(output="ok")) + ] + assert toolset.handle_count == 1 + + +async def test_transport_error_after_partial_call_starts_no_tools() -> None: + toolset = _RecordingToolset() + callbacks: list[ToolResult] = [] + error = APIConnectionError("connection ended") + stream = _ScriptedStream([_tool_call("call_1", index=0)], terminal_error=error) + + with pytest.raises(APIConnectionError) as caught: + await step( + _StaticProvider(stream), + "", + toolset, + [], + on_tool_result=callbacks.append, + ) + + assert caught.value is error + assert toolset.handle_count == 0 + assert callbacks == [] + + +async def test_cancellation_after_partial_call_starts_no_tools() -> None: + toolset = _RecordingToolset() + callbacks: list[ToolResult] = [] + stream = _ScriptedStream([_tool_call("call_1", index=0)], block_after_parts=True) + running = asyncio.create_task( + step( + _StaticProvider(stream), + "", + toolset, + [], + on_tool_result=callbacks.append, + ) + ) + await asyncio.wait_for(stream.entered_block.wait(), timeout=1) + + running.cancel() + with pytest.raises(asyncio.CancelledError): + await running + + assert toolset.handle_count == 0 + assert callbacks == [] + + +class _PendingThenFailingToolset: + def __init__(self) -> None: + self.handle_count = 0 + self.first_future: ToolResultFuture | None = None + + @property + def tools(self) -> list[Tool]: + return [] + + def handle(self, tool_call: ToolCall) -> ToolResult | ToolResultFuture: + self.handle_count += 1 + if self.handle_count == 1: + self.first_future = asyncio.get_running_loop().create_future() + return self.first_future + raise RuntimeError("second dispatch failed") + + +async def test_later_terminal_tool_dispatch_failure_cancels_earlier_future() -> None: + toolset = _PendingThenFailingToolset() + callbacks: list[ToolResult] = [] + provider = MockChatProvider( + [_tool_call("call_1", index=0), _tool_call("call_2", index=1)], + finish_reason="tool_calls", + ) + + with pytest.raises(RuntimeError, match="second dispatch failed"): + await step(provider, "", toolset, [], on_tool_result=callbacks.append) + + assert toolset.first_future is not None + assert toolset.first_future.cancelled() + await asyncio.sleep(0) + assert callbacks == [] + + +class _ImmediateThenCancelledToolset: + def __init__(self) -> None: + self.handle_count = 0 + + @property + def tools(self) -> list[Tool]: + return [] + + def handle(self, tool_call: ToolCall) -> ToolResult: + self.handle_count += 1 + if self.handle_count == 1: + return ToolResult(tool_call_id=tool_call.id, return_value=ToolOk(output="done")) + raise asyncio.CancelledError + + +async def test_dispatch_rollback_suppresses_queued_completed_callback() -> None: + toolset = _ImmediateThenCancelledToolset() + callbacks: list[ToolResult] = [] + provider = MockChatProvider( + [_tool_call("call_1", index=0), _tool_call("call_2", index=1)], + finish_reason="tool_calls", + ) + + with pytest.raises(asyncio.CancelledError): + await step(provider, "", toolset, [], on_tool_result=callbacks.append) + + await asyncio.sleep(0) + assert callbacks == [] diff --git a/packages/pythinker-core/tests/test_stream_tool_call_metadata.py b/packages/pythinker-core/tests/test_stream_tool_call_metadata.py index 79b61927..8d7022c0 100644 --- a/packages/pythinker-core/tests/test_stream_tool_call_metadata.py +++ b/packages/pythinker-core/tests/test_stream_tool_call_metadata.py @@ -1,5 +1,5 @@ -from collections.abc import AsyncIterator -from typing import Literal, cast +from collections.abc import AsyncIterator, Sequence +from typing import Literal, Self, cast import pytest from anthropic import AsyncStream as AnthropicAsyncStream @@ -17,6 +17,7 @@ from openai.types.chat import ChatCompletionChunk from openai.types.responses import ( Response, + ResponseCompletedEvent, ResponseCreatedEvent, ResponseFailedEvent, ResponseFunctionCallArgumentsDeltaEvent, @@ -26,11 +27,14 @@ ) from openai.types.responses.response import IncompleteDetails +from pythinker_core import generate +from pythinker_core.chat_provider import StreamedMessage, StreamedMessagePart, ThinkingEffort from pythinker_core.chat_provider.pythinker import PythinkerStreamedMessage from pythinker_core.contrib.chat_provider.anthropic import AnthropicStreamedMessage from pythinker_core.contrib.chat_provider.openai_legacy import OpenAILegacyStreamedMessage from pythinker_core.contrib.chat_provider.openai_responses import OpenAIResponsesStreamedMessage -from pythinker_core.message import ToolCall, ToolCallPart +from pythinker_core.message import Message, ToolCall, ToolCallPart +from pythinker_core.tooling import Tool async def _async_events[T](*events: T) -> AsyncIterator[T]: @@ -82,6 +86,32 @@ async def _collect(stream: object) -> list[ToolCall | ToolCallPart]: return [part async for part in cast(AsyncIterator[ToolCall | ToolCallPart], stream)] +class _StaticStreamProvider: + name = "static-stream" + + def __init__(self, stream: StreamedMessage) -> None: + self._stream = stream + + @property + def model_name(self) -> str: + return "static-stream" + + @property + def thinking_effort(self) -> ThinkingEffort | None: + return None + + async def generate( + self, + system_prompt: str, + tools: Sequence[Tool], + history: Sequence[Message], + ) -> StreamedMessage: + return self._stream + + def with_thinking(self, effort: ThinkingEffort) -> Self: + return self + + async def test_openai_legacy_preserves_tool_call_index_and_id() -> None: chunks = _async_events( _chat_chunk( @@ -162,6 +192,126 @@ async def test_pythinker_preserves_tool_call_index_and_id() -> None: ] +@pytest.mark.parametrize("adapter", ["openai_legacy", "pythinker"]) +async def test_openai_shaped_missing_id_and_late_name_finalize_deterministically( + adapter: str, +) -> None: + chunks = _async_events( + _chat_chunk( + tool_calls=[ + { + "index": 0, + "id": None, + "type": "function", + "function": {"name": None, "arguments": None}, + } + ] + ), + _chat_chunk( + tool_calls=[ + { + "index": 0, + "id": None, + "type": "function", + "function": {"name": "read", "arguments": '{"path":"a.py"}'}, + } + ] + ), + ) + if adapter == "openai_legacy": + stream: StreamedMessage = OpenAILegacyStreamedMessage( + cast(AsyncStream[ChatCompletionChunk], chunks), reasoning_key=None + ) + else: + stream = PythinkerStreamedMessage(cast(AsyncStream[ChatCompletionChunk], chunks)) + raw_parts: list[ToolCall | ToolCallPart] = [] + + async def on_part(part: StreamedMessagePart) -> None: + if isinstance(part, (ToolCall, ToolCallPart)): + raw_parts.append(part) + + result = await generate( + _StaticStreamProvider(stream), + "", + [], + [], + on_message_part=on_part, + ) + + assert raw_parts == [ + ToolCall( + id="", + function=ToolCall.FunctionBody(name="", arguments=None), + stream_index=0, + ), + ToolCallPart( + arguments_part='{"path":"a.py"}', + name_part="read", + stream_index=0, + stream_call_id=None, + ), + ] + assert result.message.tool_calls == [ + ToolCall( + id="call_fada958acfb8ed05ed05", + function=ToolCall.FunctionBody(name="read", arguments='{"path":"a.py"}'), + ) + ] + + +@pytest.mark.parametrize("adapter", ["openai_legacy", "pythinker"]) +async def test_openai_shaped_late_id_without_function_content_is_preserved( + adapter: str, +) -> None: + chunks = _async_events( + _chat_chunk( + tool_calls=[ + { + "index": 0, + "id": None, + "type": "function", + "function": {"name": "read", "arguments": ""}, + } + ] + ), + _chat_chunk( + tool_calls=[ + { + "index": 0, + "id": "late_call", + "type": "function", + "function": None, + } + ] + ), + _chat_chunk( + tool_calls=[ + { + "index": 0, + "id": None, + "type": "function", + "function": {"name": None, "arguments": "{}"}, + } + ] + ), + ) + if adapter == "openai_legacy": + stream: StreamedMessage = OpenAILegacyStreamedMessage( + cast(AsyncStream[ChatCompletionChunk], chunks), reasoning_key=None + ) + else: + stream = PythinkerStreamedMessage(cast(AsyncStream[ChatCompletionChunk], chunks)) + + result = await generate(_StaticStreamProvider(stream), "", [], []) + + assert result.message.tool_calls == [ + ToolCall( + id="late_call", + function=ToolCall.FunctionBody(name="read", arguments="{}"), + ) + ] + + async def test_openai_responses_uses_output_index_and_semantic_call_id() -> None: item = ResponseFunctionToolCall( arguments="", @@ -237,6 +387,52 @@ async def test_openai_responses_keeps_response_id_separate_from_item_id() -> Non assert stream.id == "response_terminal" +async def test_openai_responses_generate_preserves_response_and_semantic_call_ids() -> None: + events = _async_events( + ResponseCreatedEvent( + response=_response(response_id="response_created"), + sequence_number=0, + type="response.created", + ), + ResponseOutputItemAddedEvent( + item=ResponseFunctionToolCall( + arguments="", + call_id="semantic_call", + id="output_item", + name="read", + status="in_progress", + type="function_call", + ), + output_index=0, + sequence_number=1, + type="response.output_item.added", + ), + ResponseFunctionCallArgumentsDeltaEvent( + delta="{}", + item_id="output_item", + output_index=0, + sequence_number=2, + type="response.function_call_arguments.delta", + ), + ResponseCompletedEvent( + response=_response(response_id="response_terminal", status="completed"), + sequence_number=3, + type="response.completed", + ), + ) + stream = OpenAIResponsesStreamedMessage(cast(AsyncStream[ResponseStreamEvent], events)) + + result = await generate(_StaticStreamProvider(stream), "", [], []) + + assert result.id == "response_terminal" + assert result.message.tool_calls == [ + ToolCall( + id="semantic_call", + function=ToolCall.FunctionBody(name="read", arguments="{}"), + ) + ] + + class _AnthropicEventStream: def __init__(self, *events: RawMessageStreamEvent): self._events = _async_events(*events) From 527b2556a12f57f410e5fd6471c0cf60a9a32868 Mon Sep 17 00:00:00 2001 From: elkaix Date: Tue, 14 Jul 2026 22:37:54 -0400 Subject: [PATCH 07/10] fix(soul): avoid replaying published stream failures --- src/pythinker_code/soul/pythinkersoul.py | 3 + tests/core/test_context.py | 39 ++++- .../core/test_pythinkersoul_retry_recovery.py | 148 ++++++++++++++++++ tests/core/test_wire_message.py | 25 ++- 4 files changed, 208 insertions(+), 7 deletions(-) diff --git a/src/pythinker_code/soul/pythinkersoul.py b/src/pythinker_code/soul/pythinkersoul.py index 7b5c6907..8990ecb8 100644 --- a/src/pythinker_code/soul/pythinkersoul.py +++ b/src/pythinker_code/soul/pythinkersoul.py @@ -18,6 +18,7 @@ APIConnectionError, APIEmptyResponseError, APIStatusError, + APIStreamProtocolError, APITimeoutError, RetryableChatProvider, ThinkingEffort, @@ -2968,6 +2969,8 @@ def _is_retryable_error(exception: BaseException) -> bool: return not bool(getattr(exception, "_pythinker_recovery_exhausted", False)) if isinstance(exception, APIEmptyResponseError): return True + if isinstance(exception, APIStreamProtocolError): + return not exception.output_published if not isinstance(exception, APIStatusError): return False if exception.status_code == 429 and _is_hard_usage_limit(exception): diff --git a/tests/core/test_context.py b/tests/core/test_context.py index dfbe604f..da96475a 100644 --- a/tests/core/test_context.py +++ b/tests/core/test_context.py @@ -8,7 +8,7 @@ from unittest.mock import AsyncMock import pytest -from pythinker_core.message import Message, Role +from pythinker_core.message import Message, Role, ToolCall from pythinker_code.soul.context import Context from pythinker_code.wire.types import TextPart @@ -480,6 +480,43 @@ async def test_write_append_messages_then_restore(tmp_path: Path) -> None: assert ctx2.history[1].role == "assistant" +@pytest.mark.asyncio +async def test_correlated_tool_call_persistence_omits_transport_metadata( + tmp_path: Path, +) -> None: + path = tmp_path / "context.jsonl" + call = ToolCall( + id="call_123", + function=ToolCall.FunctionBody(name="read", arguments='{"path":"a.py"}'), + stream_index=2, + ) + ctx = Context(file_backend=path) + + await ctx.append_message(Message(role="assistant", content=[], tool_calls=[call])) + + persisted = path.read_text(encoding="utf-8") + assert "stream_index" not in persisted + assert "stream_call_id" not in persisted + assert "name_part" not in persisted + record = json.loads(persisted) + assert record["tool_calls"][0]["id"] == "call_123" + assert record["tool_calls"][0]["function"] == { + "name": "read", + "arguments": '{"path":"a.py"}', + } + + restored = Context(file_backend=path) + assert await restored.restore() is True + restored_calls = restored.history[0].tool_calls + assert restored_calls is not None + assert restored_calls == [ + ToolCall( + id="call_123", + function=ToolCall.FunctionBody(name="read", arguments='{"path":"a.py"}'), + ) + ] + + @pytest.mark.asyncio async def test_prepend_then_restore(tmp_path: Path) -> None: """Legacy session migration: prepend system prompt to existing messages, then restore.""" diff --git a/tests/core/test_pythinkersoul_retry_recovery.py b/tests/core/test_pythinkersoul_retry_recovery.py index 07f3e9fa..a09e4b39 100644 --- a/tests/core/test_pythinkersoul_retry_recovery.py +++ b/tests/core/test_pythinkersoul_retry_recovery.py @@ -11,6 +11,7 @@ from pythinker_core.chat_provider import ( APIConnectionError, APIStatusError, + APIStreamProtocolError, StreamedMessagePart, ThinkingEffort, TokenUsage, @@ -308,6 +309,76 @@ def with_thinking(self, effort: ThinkingEffort) -> Self: return self +class StreamProtocolErrorThenSuccessProvider: + name = "stream-protocol-error-then-success" + + def __init__(self) -> None: + self.generate_attempts = 0 + self.error = APIStreamProtocolError( + "orphan_fragment", + response_id="response_safe", + stream_index=0, + call_id="call_safe", + output_published=False, + ) + + @property + def model_name(self) -> str: + return "stream-protocol-error-then-success" + + @property + def thinking_effort(self) -> ThinkingEffort | None: + return None + + async def generate( + self, + system_prompt: str, + tools: Sequence[Tool], + history: Sequence[Message], + ) -> StaticStreamedMessage: + self.generate_attempts += 1 + if self.generate_attempts == 1: + raise self.error + return StaticStreamedMessage([TextPart(text="protocol recovered")]) + + def with_thinking(self, effort: ThinkingEffort) -> Self: + return self + + +class PublishedStreamProtocolErrorProvider: + name = "published-stream-protocol-error" + + def __init__(self) -> None: + self.generate_attempts = 0 + self.error = APIStreamProtocolError( + "truncated_tool_call", + response_id="response_safe", + stream_index=0, + call_id="call_safe", + output_published=True, + ) + + @property + def model_name(self) -> str: + return "published-stream-protocol-error" + + @property + def thinking_effort(self) -> ThinkingEffort | None: + return None + + async def generate( + self, + system_prompt: str, + tools: Sequence[Tool], + history: Sequence[Message], + ) -> StaticStreamedMessage: + self.generate_attempts += 1 + raise self.error + + def with_thinking(self, effort: ThinkingEffort) -> Self: + return self + + class ConnectionThen401ThenSuccessProvider: name = "connection-then-401-then-success" @@ -394,6 +465,83 @@ async def _collect_ui_messages(wire: Wire, seen: list[object]) -> None: return +@pytest.mark.parametrize( + ("published", "expected"), + [(False, True), (True, False)], +) +def test_stream_protocol_retry_depends_on_publication( + published: bool, + expected: bool, +) -> None: + error = APIStreamProtocolError( + "orphan_fragment", + response_id="response_safe", + stream_index=0, + call_id="call_safe", + output_published=published, + ) + + assert PythinkerSoul._is_retryable_error(error) is expected + + +@pytest.mark.asyncio +async def test_unpublished_stream_protocol_error_retries_once_then_succeeds( + runtime: Runtime, + tmp_path: Path, +) -> None: + runtime.config.loop_control.max_retries_per_step = 2 + provider = StreamProtocolErrorThenSuccessProvider() + llm = LLM( + chat_provider=provider, + max_context_size=100_000, + capabilities=set(), + ) + soul, context = _make_soul(runtime, llm, tmp_path) + seen: list[object] = [] + + await run_soul( + soul, + "trigger unpublished protocol retry", + lambda wire: _collect_ui_messages(wire, seen), + asyncio.Event(), + ) + + assert provider.generate_attempts == 2 + retries = [message for message in seen if isinstance(message, StepRetry)] + assert len(retries) == 1 + assert retries[0].error_type == "APIStreamProtocolError" + assert retries[0].status_code is None + assert context.history[-1].extract_text(" ").strip() == "protocol recovered" + + +@pytest.mark.asyncio +async def test_published_stream_protocol_error_surfaces_without_retry( + runtime: Runtime, + tmp_path: Path, +) -> None: + runtime.config.loop_control.max_retries_per_step = 2 + provider = PublishedStreamProtocolErrorProvider() + llm = LLM( + chat_provider=provider, + max_context_size=100_000, + capabilities=set(), + ) + soul, _ = _make_soul(runtime, llm, tmp_path) + seen: list[object] = [] + + with pytest.raises(APIStreamProtocolError) as caught: + await run_soul( + soul, + "trigger published protocol failure", + lambda wire: _collect_ui_messages(wire, seen), + asyncio.Event(), + ) + + assert caught.value is provider.error + assert provider.generate_attempts == 1 + assert [message for message in seen if isinstance(message, StepRetry)] == [] + + @pytest.mark.asyncio async def test_step_retry_recovers_retryable_provider(runtime: Runtime, tmp_path: Path) -> None: runtime.config.loop_control.max_retries_per_step = 2 diff --git a/tests/core/test_wire_message.py b/tests/core/test_wire_message.py index 69c1155a..abacb17a 100644 --- a/tests/core/test_wire_message.py +++ b/tests/core/test_wire_message.py @@ -276,8 +276,10 @@ async def test_wire_message_serde(): msg = ToolCall( id="call_123", function=ToolCall.FunctionBody(name="bash", arguments='{"command": "ls -la"}'), + stream_index=2, ) - assert serialize_wire_message(msg) == snapshot( + serialized_call = serialize_wire_message(msg) + assert serialized_call == snapshot( { "type": "ToolCall", "payload": { @@ -288,13 +290,24 @@ async def test_wire_message_serde(): }, } ) - _test_serde(msg) + semantic_call = deserialize_wire_message(serialized_call) + assert semantic_call == ToolCall( + id="call_123", + function=ToolCall.FunctionBody(name="bash", arguments='{"command": "ls -la"}'), + ) + _test_serde(semantic_call) - msg = ToolCallPart(arguments_part="}") - assert serialize_wire_message(msg) == snapshot( - {"type": "ToolCallPart", "payload": {"arguments_part": "}"}} + msg = ToolCallPart( + arguments_part="}", + name_part=None, + stream_index=2, + stream_call_id="call_123", ) - _test_serde(msg) + serialized_part = serialize_wire_message(msg) + assert serialized_part == snapshot({"type": "ToolCallPart", "payload": {"arguments_part": "}"}}) + semantic_part = deserialize_wire_message(serialized_part) + assert semantic_part == ToolCallPart(arguments_part="}") + _test_serde(semantic_part) msg = ToolResult( tool_call_id="call_123", From 87d6ca7afccb73c118846d086b07205663645e19 Mon Sep 17 00:00:00 2001 From: elkaix Date: Tue, 14 Jul 2026 22:48:40 -0400 Subject: [PATCH 08/10] docs(changelog): note safe streamed tool calls --- CHANGELOG.md | 2 ++ docs/en/release-notes/changelog.md | 2 ++ 2 files changed, 4 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index bde0d26e..b39b3c9f 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/docs/en/release-notes/changelog.md b/docs/en/release-notes/changelog.md index 9c95528f..81076318 100644 --- a/docs/en/release-notes/changelog.md +++ b/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 From 667900eda79863c9102e5b8646a8c962b1f1670d Mon Sep 17 00:00:00 2001 From: elkaix Date: Tue, 14 Jul 2026 23:09:30 -0400 Subject: [PATCH 09/10] fix(core): address stream review findings --- .../src/pythinker_core/_generate.py | 8 +- .../contrib/chat_provider/openai_responses.py | 5 +- packages/pythinker-core/tests/test_step.py | 7 +- .../tests/test_stream_tool_call_metadata.py | 103 +++++++++++++++++- 4 files changed, 116 insertions(+), 7 deletions(-) diff --git a/packages/pythinker-core/src/pythinker_core/_generate.py b/packages/pythinker-core/src/pythinker_core/_generate.py index 4e62957a..aae4d41e 100644 --- a/packages/pythinker-core/src/pythinker_core/_generate.py +++ b/packages/pythinker-core/src/pythinker_core/_generate.py @@ -51,10 +51,14 @@ async def generate( 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 diff --git a/packages/pythinker-core/src/pythinker_core/contrib/chat_provider/openai_responses.py b/packages/pythinker-core/src/pythinker_core/contrib/chat_provider/openai_responses.py index 689952bf..d1b899c4 100644 --- a/packages/pythinker-core/src/pythinker_core/contrib/chat_provider/openai_responses.py +++ b/packages/pythinker-core/src/pythinker_core/contrib/chat_provider/openai_responses.py @@ -9,6 +9,7 @@ Response, ResponseCompletedEvent, ResponseCreatedEvent, + ResponseErrorEvent, ResponseFailedEvent, ResponseFunctionCallArgumentsDeltaEvent, ResponseIncompleteEvent, @@ -554,7 +555,7 @@ async def _convert_stream_response( item = chunk.item if item.type == "function_call": yield ToolCall( - id=item.call_id or str(uuid.uuid4()), + id=item.call_id or "", function=ToolCall.FunctionBody( name=item.name, arguments=item.arguments, @@ -574,6 +575,8 @@ async def _convert_stream_response( yield ThinkPart(think="") elif chunk.type == "response.reasoning_summary_text.delta": yield ThinkPart(think=chunk.delta) + elif isinstance(chunk, ResponseErrorEvent): + self._finish_reason = "failed" elif isinstance( chunk, (ResponseCompletedEvent, ResponseIncompleteEvent, ResponseFailedEvent), diff --git a/packages/pythinker-core/tests/test_step.py b/packages/pythinker-core/tests/test_step.py index 4600741e..94798c74 100644 --- a/packages/pythinker-core/tests/test_step.py +++ b/packages/pythinker-core/tests/test_step.py @@ -70,8 +70,9 @@ async def _stream(self) -> AsyncIterator[StreamedMessagePart]: if self._block_after_parts: self.entered_block.set() await self._never.wait() - if self._terminal_error is not None: - raise self._terminal_error + terminal_error = self._terminal_error + if terminal_error is not None: + raise terminal_error @property def id(self) -> str: @@ -352,7 +353,7 @@ def handle(self, tool_call: ToolCall) -> ToolResult: self.handle_count += 1 if self.handle_count == 1: return ToolResult(tool_call_id=tool_call.id, return_value=ToolOk(output="done")) - raise asyncio.CancelledError + raise asyncio.CancelledError() async def test_dispatch_rollback_suppresses_queued_completed_callback() -> None: diff --git a/packages/pythinker-core/tests/test_stream_tool_call_metadata.py b/packages/pythinker-core/tests/test_stream_tool_call_metadata.py index 8d7022c0..d9722572 100644 --- a/packages/pythinker-core/tests/test_stream_tool_call_metadata.py +++ b/packages/pythinker-core/tests/test_stream_tool_call_metadata.py @@ -19,6 +19,7 @@ Response, ResponseCompletedEvent, ResponseCreatedEvent, + ResponseErrorEvent, ResponseFailedEvent, ResponseFunctionCallArgumentsDeltaEvent, ResponseFunctionToolCall, @@ -28,7 +29,12 @@ from openai.types.responses.response import IncompleteDetails from pythinker_core import generate -from pythinker_core.chat_provider import StreamedMessage, StreamedMessagePart, ThinkingEffort +from pythinker_core.chat_provider import ( + APIStreamProtocolError, + StreamedMessage, + StreamedMessagePart, + ThinkingEffort, +) from pythinker_core.chat_provider.pythinker import PythinkerStreamedMessage from pythinker_core.contrib.chat_provider.anthropic import AnthropicStreamedMessage from pythinker_core.contrib.chat_provider.openai_legacy import OpenAILegacyStreamedMessage @@ -352,6 +358,101 @@ async def test_openai_responses_uses_output_index_and_semantic_call_id() -> None ] +async def test_openai_responses_empty_streamed_call_id_is_deterministic( + monkeypatch: pytest.MonkeyPatch, +) -> None: + def reject_random_id() -> None: + raise AssertionError("streamed Responses call used random ID fallback") + + monkeypatch.setattr( + "pythinker_core.contrib.chat_provider.openai_responses.uuid.uuid4", + reject_random_id, + ) + + async def generate_once() -> ToolCall: + events = _async_events( + ResponseCreatedEvent( + response=_response(response_id="response_1"), + sequence_number=0, + type="response.created", + ), + ResponseOutputItemAddedEvent( + item=ResponseFunctionToolCall( + arguments="{}", + call_id="", + id="output_item", + name="read", + status="completed", + type="function_call", + ), + output_index=3, + sequence_number=1, + type="response.output_item.added", + ), + ResponseCompletedEvent( + response=_response(response_id="response_1"), + sequence_number=2, + type="response.completed", + ), + ) + stream = OpenAIResponsesStreamedMessage(cast(AsyncStream[ResponseStreamEvent], events)) + result = await generate(_StaticStreamProvider(stream), "", [], []) + assert result.message.tool_calls is not None + return result.message.tool_calls[0] + + first = await generate_once() + second = await generate_once() + + assert first == second + assert first.id.startswith("call_") + assert first.function == ToolCall.FunctionBody(name="read", arguments="{}") + + +async def test_openai_responses_error_event_blocks_tool_callback() -> None: + events = _async_events( + ResponseCreatedEvent( + response=_response(response_id="response_created"), + sequence_number=0, + type="response.created", + ), + ResponseOutputItemAddedEvent( + item=ResponseFunctionToolCall( + arguments="{}", + call_id="semantic_call", + id="output_item", + name="read", + status="in_progress", + type="function_call", + ), + output_index=0, + sequence_number=1, + type="response.output_item.added", + ), + ResponseErrorEvent( + code="server_error", + message="provider-private detail", + param=None, + sequence_number=2, + type="error", + ), + ) + stream = OpenAIResponsesStreamedMessage(cast(AsyncStream[ResponseStreamEvent], events)) + callbacks: list[ToolCall] = [] + + with pytest.raises(APIStreamProtocolError) as caught: + await generate( + _StaticStreamProvider(stream), + "", + [], + [], + on_tool_call=callbacks.append, + ) + + assert caught.value.category == "terminal_failure" + assert "provider-private detail" not in str(caught.value) + assert callbacks == [] + + async def test_openai_responses_keeps_response_id_separate_from_item_id() -> None: events = _async_events( ResponseCreatedEvent( From 87c4c1d9a45f68d277df16206434a5028627ee35 Mon Sep 17 00:00:00 2001 From: elkaix Date: Tue, 14 Jul 2026 23:25:16 -0400 Subject: [PATCH 10/10] fix(core): stop responses streams on error --- .../pythinker_core/contrib/chat_provider/openai_responses.py | 1 + .../pythinker-core/tests/test_stream_tool_call_metadata.py | 5 +++++ 2 files changed, 6 insertions(+) diff --git a/packages/pythinker-core/src/pythinker_core/contrib/chat_provider/openai_responses.py b/packages/pythinker-core/src/pythinker_core/contrib/chat_provider/openai_responses.py index d1b899c4..7fb5b893 100644 --- a/packages/pythinker-core/src/pythinker_core/contrib/chat_provider/openai_responses.py +++ b/packages/pythinker-core/src/pythinker_core/contrib/chat_provider/openai_responses.py @@ -577,6 +577,7 @@ async def _convert_stream_response( yield ThinkPart(think=chunk.delta) elif isinstance(chunk, ResponseErrorEvent): self._finish_reason = "failed" + return elif isinstance( chunk, (ResponseCompletedEvent, ResponseIncompleteEvent, ResponseFailedEvent), diff --git a/packages/pythinker-core/tests/test_stream_tool_call_metadata.py b/packages/pythinker-core/tests/test_stream_tool_call_metadata.py index d9722572..93c194ae 100644 --- a/packages/pythinker-core/tests/test_stream_tool_call_metadata.py +++ b/packages/pythinker-core/tests/test_stream_tool_call_metadata.py @@ -435,6 +435,11 @@ async def test_openai_responses_error_event_blocks_tool_callback() -> None: sequence_number=2, type="error", ), + ResponseCompletedEvent( + response=_response(response_id="response_trailing", status="completed"), + sequence_number=3, + type="response.completed", + ), ) stream = OpenAIResponsesStreamedMessage(cast(AsyncStream[ResponseStreamEvent], events)) callbacks: list[ToolCall] = []