From 5cabd2b72e20912530e5874ae2fff0dec635988e Mon Sep 17 00:00:00 2001 From: Onur Solmaz <2453968+osolmaz@users.noreply.github.com> Date: Wed, 19 Aug 2026 00:16:09 +0300 Subject: [PATCH] Revert "fix(providers): report request acceptance consistently (#125807)" (#126021) This reverts commit f5e9622fc982a4c4c403c2460459f13c2636b914. --- docs/plugins/sdk-provider-plugins.md | 2 - docs/plugins/sdk-subpaths.md | 1 - .../mantle-anthropic.runtime.test.ts | 6 - .../mantle-anthropic.runtime.ts | 2 - .../amazon-bedrock/stream.runtime.test.ts | 21 +-- extensions/amazon-bedrock/stream.runtime.ts | 8 +- .../anthropic-vertex/stream-runtime.test.ts | 20 --- extensions/anthropic-vertex/stream-runtime.ts | 2 - extensions/google/transport-stream.test.ts | 152 ------------------ extensions/google/transport-stream.ts | 142 ++++------------ extensions/ollama/src/stream-runtime.test.ts | 26 --- extensions/ollama/src/stream.runtime.ts | 52 +++++- package.json | 4 - packages/agent-core/src/agent.ts | 5 - packages/ai/src/providers/anthropic.ts | 11 +- .../ai/src/providers/google-shared.test.ts | 37 ----- packages/ai/src/providers/google-shared.ts | 13 +- packages/ai/src/providers/mistral.test.ts | 142 ---------------- packages/ai/src/providers/mistral.ts | 45 ++---- ...openai-chatgpt-responses-streaming.test.ts | 51 ------ ...-chatgpt-responses.encrypted-retry.test.ts | 41 ----- .../openai-chatgpt-responses.retry.test.ts | 6 - .../src/providers/openai-chatgpt-responses.ts | 19 +-- .../ai/src/providers/openai-completions.ts | 4 +- .../src/providers/openai-responses-shared.ts | 12 +- packages/ai/src/providers/simple-options.ts | 1 - .../anthropic-transport-stream.test.ts | 68 +------- .../transports/anthropic-transport-stream.ts | 29 ++-- ...ompletions-transport.response-hook.test.ts | 58 +------ .../openai-completions-transport.ts | 5 +- .../src/transports/openai-responses-client.ts | 10 +- .../openai-responses-websocket-client.test.ts | 16 -- .../src/transports/openai-transport-shared.ts | 18 +-- ...transport-stream-shared.acceptance.test.ts | 144 ----------------- .../src/transports/transport-stream-shared.ts | 136 +++------------- packages/llm-core/src/types.ts | 18 +-- packages/plugin-sdk/package.json | 4 - packages/plugin-sdk/src/provider-lifecycle.ts | 2 - scripts/lib/plugin-sdk-entrypoints.json | 1 - scripts/plugin-sdk-surface-report.mts | 9 +- ...attempt.model-diagnostic-lifecycle.test.ts | 49 +----- .../run/attempt.model-diagnostic-lifecycle.ts | 27 +--- src/plugin-sdk/llm.ts | 1 - src/plugin-sdk/provider-lifecycle.ts | 7 - 44 files changed, 165 insertions(+), 1262 deletions(-) delete mode 100644 packages/ai/src/transports/transport-stream-shared.acceptance.test.ts delete mode 100644 packages/plugin-sdk/src/provider-lifecycle.ts delete mode 100644 src/plugin-sdk/provider-lifecycle.ts diff --git a/docs/plugins/sdk-provider-plugins.md b/docs/plugins/sdk-provider-plugins.md index ed134dc52f31..9b24294591cd 100644 --- a/docs/plugins/sdk-provider-plugins.md +++ b/docs/plugins/sdk-provider-plugins.md @@ -716,8 +716,6 @@ catalog, API-key auth, and dynamic model resolution. | `validateReplayTurns` | Strict replay-turn validation before the embedded runner | | `onModelSelected` | Post-selection callback (e.g. telemetry) | - Custom `createStreamFn` transports must report provider acceptance before exposing the first stream event. Import the lifecycle helpers from `openclaw/plugin-sdk/provider-lifecycle`. Use `notifyProviderHttpResponse` when the transport owns a real `Response`, or `notifyProviderHttpMetadata` when an SDK exposes the real status and headers without the response body. Both helpers also run the compatibility `onResponse` callback. Use `notifyProviderStreamOpened` when an SDK returns an open stream but hides HTTP metadata; it never invents status or headers. Stream wrappers must forward `onProviderAccepted` and `onResponse` unchanged. - Runtime fallback notes: - `normalizeConfig` resolves one owning plugin per provider id (bundled providers first, then the matched runtime plugin) and calls only that hook - there is no scan across other providers. Google's own `normalizeConfig` hook is what normalizes `google` / `google-vertex` / `google-antigravity` config entries; it is not a separate core fallback. diff --git a/docs/plugins/sdk-subpaths.md b/docs/plugins/sdk-subpaths.md index a040a3b979fd..4aab3766115e 100644 --- a/docs/plugins/sdk-subpaths.md +++ b/docs/plugins/sdk-subpaths.md @@ -149,7 +149,6 @@ are private-local. | `plugin-sdk/provider-catalog-runtime` | Provider catalog augmentation runtime hook and plugin-provider registry seams for contract tests | | `plugin-sdk/provider-catalog-shared` | Private-local after July 2026; `findCatalogTemplate`, `buildSingleProviderApiKeyCatalog`, `buildManifestModelProviderConfig`, `supportsNativeStreamingUsageCompat`, `applyProviderNativeStreamingUsageCompat` | | `plugin-sdk/provider-http` | Private-local after July 2026; Generic provider HTTP/endpoint capability helpers, provider HTTP errors, and audio transcription multipart form helpers | - | `plugin-sdk/provider-lifecycle` | Supported provider request-acceptance types and helpers for real HTTP metadata and metadata-free SDK or WebSocket streams | | `plugin-sdk/provider-binary-stream` | Direct-reader bounded binary streams with fitting-prefix delivery and explicit overflow/release errors | | `plugin-sdk/provider-web-fetch-contract` | Private-local after July 2026; Narrow web-fetch config/selection contract helpers such as `enablePluginInConfig` and `WebFetchProviderPlugin` | | `plugin-sdk/provider-web-fetch` | Private-local after July 2026; Web-fetch provider registration/cache helpers | diff --git a/extensions/amazon-bedrock-mantle/mantle-anthropic.runtime.test.ts b/extensions/amazon-bedrock-mantle/mantle-anthropic.runtime.test.ts index ac6144c6ef32..6fa24510593e 100644 --- a/extensions/amazon-bedrock-mantle/mantle-anthropic.runtime.test.ts +++ b/extensions/amazon-bedrock-mantle/mantle-anthropic.runtime.test.ts @@ -60,13 +60,9 @@ describe("createMantleAnthropicStreamFn", () => { const context = { messages: [] }; const deps = createTestDeps(); deps.stream.mockReturnValue(stream as never); - const onProviderAccepted = vi.fn(); - const onResponse = vi.fn(); const result = createMantleAnthropicStreamFn(deps)(model, context, { apiKey: "bedrock-bearer-token", - onProviderAccepted, - onResponse, headers: { "X-Caller": "caller-header", }, @@ -91,8 +87,6 @@ describe("createMantleAnthropicStreamFn", () => { "bedrock-bearer-token", ); expect(streamOptions.thinkingEnabled).toBe(false); - expect(streamOptions.onProviderAccepted).toBe(onProviderAccepted); - expect(streamOptions.onResponse).toBe(onResponse); }); it("omits unsupported Opus 4.7 sampling and reasoning overrides", () => { diff --git a/extensions/amazon-bedrock-mantle/mantle-anthropic.runtime.ts b/extensions/amazon-bedrock-mantle/mantle-anthropic.runtime.ts index 076c97bfded0..afaed5a55890 100644 --- a/extensions/amazon-bedrock-mantle/mantle-anthropic.runtime.ts +++ b/extensions/amazon-bedrock-mantle/mantle-anthropic.runtime.ts @@ -136,8 +136,6 @@ function buildMantleAnthropicBaseOptions( cacheRetention: options?.cacheRetention, sessionId: options?.sessionId, onPayload: options?.onPayload, - onProviderAccepted: options?.onProviderAccepted, - onResponse: options?.onResponse, maxRetryDelayMs: options?.maxRetryDelayMs, metadata: options?.metadata, }; diff --git a/extensions/amazon-bedrock/stream.runtime.test.ts b/extensions/amazon-bedrock/stream.runtime.test.ts index 1f2f29af7756..11e5c0c84455 100644 --- a/extensions/amazon-bedrock/stream.runtime.test.ts +++ b/extensions/amazon-bedrock/stream.runtime.test.ts @@ -119,17 +119,12 @@ describe("Bedrock stream client lifecycle", () => { yield { messageStop: { stopReason: BedrockStopReason.END_TURN } }; } const send = vi.spyOn(BedrockRuntimeClient.prototype, "send").mockResolvedValue({ - $metadata: { httpStatusCode: 200, requestId: "bedrock-request-1" }, + $metadata: { httpStatusCode: 200 }, stream: successfulStream(), } as never); const destroy = vi.spyOn(BedrockRuntimeClient.prototype, "destroy"); - const onProviderAccepted = vi.fn(); - const onResponse = vi.fn(); - const resultPromise = streamBedrockForTest(bedrockModel({}), context, { - onProviderAccepted, - onResponse, - }).result(); + const resultPromise = streamBedrockForTest(bedrockModel({}), context).result(); await streamBlocked; expect(destroy).not.toHaveBeenCalled(); @@ -137,18 +132,6 @@ describe("Bedrock stream client lifecycle", () => { const result = await resultPromise; expect(result.stopReason).toBe("stop"); - expect(onProviderAccepted).toHaveBeenCalledWith( - { - kind: "http_response", - status: 200, - headers: { "x-amzn-requestid": "bedrock-request-1" }, - }, - expect.objectContaining({ provider: "amazon-bedrock" }), - ); - expect(onResponse).toHaveBeenCalledWith( - { status: 200, headers: { "x-amzn-requestid": "bedrock-request-1" } }, - expect.objectContaining({ provider: "amazon-bedrock" }), - ); expectDestroyedClient(send, destroy); }); diff --git a/extensions/amazon-bedrock/stream.runtime.ts b/extensions/amazon-bedrock/stream.runtime.ts index c3297319cfe8..9a95c7ec513b 100644 --- a/extensions/amazon-bedrock/stream.runtime.ts +++ b/extensions/amazon-bedrock/stream.runtime.ts @@ -55,7 +55,6 @@ import { type ToolResultMessage, } from "openclaw/plugin-sdk/llm"; import { canonicalizeBase64 } from "openclaw/plugin-sdk/media-runtime"; -import { notifyProviderHttpMetadata } from "openclaw/plugin-sdk/provider-lifecycle"; import { resolveClaudeFable5ModelIdentity, resolveClaudeModelIdentity, @@ -284,11 +283,10 @@ const streamBedrock: StreamFunction<"bedrock-converse-stream", BedrockOptions> = if (response.$metadata.requestId) { responseHeaders["x-amzn-requestid"] = response.$metadata.requestId; } - await notifyProviderHttpMetadata({ - options, - response: { status: response.$metadata.httpStatusCode, headers: responseHeaders }, + await options?.onResponse?.( + { status: response.$metadata.httpStatusCode, headers: responseHeaders }, model, - }); + ); } let sawMessageStop = false; diff --git a/extensions/anthropic-vertex/stream-runtime.test.ts b/extensions/anthropic-vertex/stream-runtime.test.ts index 60532aad9054..2cd043fba24a 100644 --- a/extensions/anthropic-vertex/stream-runtime.test.ts +++ b/extensions/anthropic-vertex/stream-runtime.test.ts @@ -541,26 +541,6 @@ describe("createAnthropicVertexStreamFn", () => { expect(transportOptions).not.toHaveProperty("temperature"); }); - it("forwards provider acceptance hooks to the shared Anthropic transport", () => { - const { deps, streamAnthropicMock } = createStreamDeps(); - const streamFn = createAnthropicVertexStreamFn("vertex-project", "us-east5", undefined, deps); - const onProviderAccepted = vi.fn(); - const onResponse = vi.fn(); - - void streamFn( - makeModel({ id: "claude-sonnet-4-6" }), - { messages: [] }, - { - onProviderAccepted, - onResponse, - }, - ); - - const transportOptions = streamTransportOptions(streamAnthropicMock); - expect(transportOptions.onProviderAccepted).toBe(onProviderAccepted); - expect(transportOptions.onResponse).toBe(onResponse); - }); - it("keeps already-budgeted cache_control markers intact when forwarding payload hooks", async () => { const { deps, streamAnthropicMock } = createStreamDeps(); const onPayload = vi.fn(async (payload: unknown) => payload); diff --git a/extensions/anthropic-vertex/stream-runtime.ts b/extensions/anthropic-vertex/stream-runtime.ts index c920ac63683a..572e4e523ddd 100644 --- a/extensions/anthropic-vertex/stream-runtime.ts +++ b/extensions/anthropic-vertex/stream-runtime.ts @@ -231,8 +231,6 @@ export function createAnthropicVertexStreamFn( // cache boundary and budgets all cache_control markers; re-applying the // payload policy here marked the uncached suffix and breached the 4-marker cap. onPayload: options?.onPayload, - onProviderAccepted: options?.onProviderAccepted, - onResponse: options?.onResponse, maxRetryDelayMs: options?.maxRetryDelayMs, metadata: options?.metadata, }; diff --git a/extensions/google/transport-stream.test.ts b/extensions/google/transport-stream.test.ts index 0b06f6ce6827..7c6f00eb4d65 100644 --- a/extensions/google/transport-stream.test.ts +++ b/extensions/google/transport-stream.test.ts @@ -630,58 +630,6 @@ describe("google transport stream", () => { expect(guardedFetchMock).not.toHaveBeenCalled(); }); - it("reports the real HTTP response before consuming Gemini SSE output", async () => { - mockGoogleTextResponse(); - const onProviderAccepted = vi.fn(); - const onResponse = vi.fn(); - - const result = await runGeminiStreamResult({ - options: { onProviderAccepted, onResponse }, - }); - - expect(result.stopReason).toBe("stop"); - expect(onProviderAccepted).toHaveBeenCalledWith( - { - kind: "http_response", - status: 200, - headers: { "content-type": "text/event-stream" }, - }, - expect.objectContaining({ provider: "google" }), - ); - expect(onResponse).toHaveBeenCalledWith( - { status: 200, headers: { "content-type": "text/event-stream" } }, - expect.objectContaining({ provider: "google" }), - ); - }); - - it("reports rejected HTTP responses without marking them accepted", async () => { - guardedFetchMock.mockResolvedValueOnce( - new Response('{"error":{"message":"rate limited"}}', { - status: 429, - headers: { - "content-type": "application/json", - "x-request-id": "req-rejected", - }, - }), - ); - const onProviderAccepted = vi.fn(); - const onResponse = vi.fn(); - - const result = await runGeminiStreamResult({ - options: { onProviderAccepted, onResponse }, - }); - - expect(result.stopReason).toBe("error"); - expect(onProviderAccepted).not.toHaveBeenCalled(); - expect(onResponse).toHaveBeenCalledWith( - { - status: 429, - headers: expect.objectContaining({ "x-request-id": "req-rejected" }), - }, - expect.objectContaining({ provider: "google" }), - ); - }); - it("uses the guarded fetch transport and parses Gemini SSE output", async () => { guardedFetchMock.mockResolvedValueOnce( buildSseResponse([ @@ -1471,106 +1419,6 @@ describe("google transport stream", () => { }, ); - it("does not retry when a slow provider acceptance callback rejects", async () => { - vi.stubEnv("OPENCLAW_GOOGLE_GEMINI_FIRST_RESPONSE_RETRY_MS", "10"); - let cancelCalled = false; - guardedFetchMock.mockResolvedValueOnce( - buildOpenRawSseResponse({ - sse: 'data: {"candidates":[{"finishReason":"STOP"}]}\n\n', - onCancel: () => { - cancelCalled = true; - }, - }), - ); - - const result = await runGeminiStreamResult({ - model: buildGeminiModel({ id: "gemini-3.1-pro-preview" }), - options: { - reasoning: "high", - onProviderAccepted: async () => { - await new Promise((resolve) => { - setTimeout(resolve, 25); - }); - throw new Error("acceptance callback failed"); - }, - }, - }); - - expect(result).toMatchObject({ - stopReason: "error", - errorMessage: "acceptance callback failed", - }); - expect(guardedFetchMock).toHaveBeenCalledOnce(); - expect(cancelCalled).toBe(true); - }); - - it.each(["onProviderAccepted", "onResponse"] as const)( - "aborts a pending %s callback without retrying", - async (hookName) => { - vi.stubEnv("OPENCLAW_GOOGLE_GEMINI_FIRST_RESPONSE_RETRY_MS", "1000"); - const controller = new AbortController(); - const cancel = vi.fn(); - guardedFetchMock.mockResolvedValueOnce( - buildOpenRawSseResponse({ - sse: 'data: {"candidates":[{"finishReason":"STOP"}]}\n\n', - onCancel: cancel, - }), - ); - let markHookStarted!: () => void; - const hookStarted = new Promise((resolve) => { - markHookStarted = resolve; - }); - const hook = vi.fn(() => { - markHookStarted(); - return new Promise(() => {}); - }); - - const resultPromise = runGeminiStreamResult({ - model: buildGeminiModel({ id: "gemini-3.1-pro-preview" }), - options: { - reasoning: "high", - signal: controller.signal, - [hookName]: hook, - }, - }); - await hookStarted; - controller.abort( - Object.assign(new Error("operator canceled the request"), { - code: "OPERATOR_CANCELLED", - }), - ); - - await expect(resultPromise).resolves.toMatchObject({ - stopReason: "aborted", - errorCode: "OPERATOR_CANCELLED", - errorMessage: "operator canceled the request", - }); - expect(hook).toHaveBeenCalledOnce(); - expect(guardedFetchMock).toHaveBeenCalledOnce(); - expect(cancel).toHaveBeenCalledOnce(); - }, - ); - - it("does not count provider acceptance callback time against the retry deadline", async () => { - vi.stubEnv("OPENCLAW_GOOGLE_GEMINI_FIRST_RESPONSE_RETRY_MS", "10"); - mockGoogleTextResponse("accepted"); - - const result = await runGeminiStreamResult({ - model: buildGeminiModel({ id: "gemini-3.1-pro-preview" }), - options: { - reasoning: "high", - onProviderAccepted: async () => { - await new Promise((resolve) => { - setTimeout(resolve, 25); - }); - }, - }, - }); - - expect(result.content).toEqual([{ type: "text", text: "accepted" }]); - expect(guardedFetchMock).toHaveBeenCalledOnce(); - }); - it("keeps oversized-video shedding in the Gemini 3 retry payload", async () => { vi.stubEnv("OPENCLAW_GOOGLE_GEMINI_FIRST_RESPONSE_RETRY_MS", "10"); guardedFetchMock diff --git a/extensions/google/transport-stream.ts b/extensions/google/transport-stream.ts index d69ca669a00e..84499320a02a 100644 --- a/extensions/google/transport-stream.ts +++ b/extensions/google/transport-stream.ts @@ -24,7 +24,6 @@ import { providerOperationRetryConfig, resolveProviderRequestHeaders, } from "openclaw/plugin-sdk/provider-http"; -import { notifyProviderHttpResponse } from "openclaw/plugin-sdk/provider-lifecycle"; import { buildGuardedModelFetch, coerceTransportToolCallArguments, @@ -1042,43 +1041,10 @@ function buildGoogleGemini3FirstResponseRetryParams(params: { function createChildSignal(parent: AbortSignal | undefined, timeoutMs: number) { const controller = new AbortController(); let timedOut = false; - let remainingMs = timeoutMs; - let deadlineStartedAt: number | undefined; let timeout: ReturnType | undefined; const abortFromParent = () => { controller.abort(parent?.reason); }; - const abortForTimeout = () => { - timedOut = true; - timeout = undefined; - deadlineStartedAt = undefined; - controller.abort(new Error("Google Gemini first response retry deadline reached")); - }; - const startDeadline = () => { - if (timeout || controller.signal.aborted || timeoutMs <= 0) { - return; - } - if (remainingMs <= 0) { - abortForTimeout(); - return; - } - deadlineStartedAt = Date.now(); - timeout = setTimeout(abortForTimeout, remainingMs); - timeout.unref?.(); - }; - const clearDeadline = () => { - if (timeout) { - clearTimeout(timeout); - timeout = undefined; - } - deadlineStartedAt = undefined; - }; - const pauseDeadline = () => { - if (deadlineStartedAt !== undefined) { - remainingMs = Math.max(0, remainingMs - (Date.now() - deadlineStartedAt)); - } - clearDeadline(); - }; if (parent) { if (parent.aborted) { abortFromParent(); @@ -1086,12 +1052,22 @@ function createChildSignal(parent: AbortSignal | undefined, timeoutMs: number) { parent.addEventListener("abort", abortFromParent, { once: true }); } } - startDeadline(); + if (timeoutMs > 0) { + timeout = setTimeout(() => { + timedOut = true; + controller.abort(new Error("Google Gemini first response retry deadline reached")); + }, timeoutMs); + timeout.unref?.(); + } + const clearDeadline = () => { + if (timeout) { + clearTimeout(timeout); + timeout = undefined; + } + }; return { signal: controller.signal, timedOut: () => timedOut, - pauseDeadline, - resumeDeadline: startDeadline, clearDeadline, cleanup: () => { clearDeadline(); @@ -1128,20 +1104,6 @@ type GoogleSseAttempt = } | { type: "timeout" }; -async function notifyGoogleTransportHttpResponse( - model: GoogleTransportModel, - options: GoogleTransportOptions | undefined, - response: Response, - signal?: AbortSignal, -): Promise { - await notifyProviderHttpResponse({ - options, - response, - model: canonicalGoogleModel(model), - signal, - }); -} - async function openGoogleSseAttempt(params: { guardedFetch: ReturnType; url: string; @@ -1151,64 +1113,44 @@ async function openGoogleSseAttempt(params: { parentSignal?: AbortSignal; firstResponseTimeoutMs: number; errorPrefix: string; - model: GoogleTransportModel; - options: GoogleTransportOptions | undefined; }): Promise { const attemptSignal = params.firstResponseTimeoutMs > 0 ? createChildSignal(params.parentSignal, params.firstResponseTimeoutMs) : undefined; const signal = attemptSignal?.signal ?? params.parentSignal; - const handleTimedOperationError = (error: unknown): GoogleSseAttempt => { - attemptSignal?.cleanup(); - if (attemptSignal?.timedOut() && !params.parentSignal?.aborted) { - return { type: "timeout" }; - } - throw error; - }; - let response: Response; try { - response = await params.guardedFetch(params.url, { + const response = await params.guardedFetch(params.url, { method: "POST", headers: params.headers, body: serializeGoogleRequest(params.request, params.videoSlots), signal, }); - } catch (error) { - return handleTimedOperationError(error); - } - attemptSignal?.pauseDeadline(); - try { - await notifyGoogleTransportHttpResponse(params.model, params.options, response, signal); - } catch (error) { - attemptSignal?.cleanup(); - throw error; - } - if (!response.ok) { - attemptSignal?.cleanup(); - throw await createProviderHttpError(response, params.errorPrefix); - } - attemptSignal?.resumeDeadline(); - const chunks = parseGoogleSseChunks(response, signal); - const iterator = chunks[Symbol.asyncIterator](); - let first: IteratorResult; - try { - first = await iterator.next(); - } catch (error) { - return handleTimedOperationError(error); - } - attemptSignal?.clearDeadline(); - if (first.done) { + if (!response.ok) { + throw await createProviderHttpError(response, params.errorPrefix); + } + const chunks = parseGoogleSseChunks(response, signal); + const iterator = chunks[Symbol.asyncIterator](); + const first = await iterator.next(); + attemptSignal?.clearDeadline(); + if (first.done) { + return { + type: "ready", + chunks: iteratorToAsyncGenerator(iterator, attemptSignal?.cleanup), + }; + } return { type: "ready", + firstChunk: first.value, chunks: iteratorToAsyncGenerator(iterator, attemptSignal?.cleanup), }; + } catch (error) { + attemptSignal?.cleanup(); + if (attemptSignal?.timedOut() && !params.parentSignal?.aborted) { + return { type: "timeout" }; + } + throw error; } - return { - type: "ready", - firstChunk: first.value, - chunks: iteratorToAsyncGenerator(iterator, attemptSignal?.cleanup), - }; } async function openGoogleSseChunks(params: { @@ -1232,12 +1174,6 @@ async function openGoogleSseChunks(params: { body: serializeGoogleRequest(params.request, params.videoSlots), signal: params.options?.signal, }); - await notifyGoogleTransportHttpResponse( - params.model, - params.options, - response, - params.options?.signal, - ); if (!response.ok) { throw await createProviderHttpError(response, errorPrefix); } @@ -1255,12 +1191,6 @@ async function openGoogleSseChunks(params: { body: serializeGoogleRequest(params.request, params.videoSlots), signal: params.options?.signal, }); - await notifyGoogleTransportHttpResponse( - params.model, - params.options, - response, - params.options?.signal, - ); if (!response.ok) { throw await createProviderHttpError(response, errorPrefix); } @@ -1279,8 +1209,6 @@ async function openGoogleSseChunks(params: { parentSignal: params.options?.signal, firstResponseTimeoutMs: retryMs, errorPrefix, - model: params.model, - options: params.options, }); if (firstAttempt.type === "ready") { return firstAttempt; @@ -1301,8 +1229,6 @@ async function openGoogleSseChunks(params: { parentSignal: params.options?.signal, firstResponseTimeoutMs: 0, errorPrefix, - model: params.model, - options: params.options, }); if (retryAttempt.type === "timeout") { throw new Error("Google Gemini first response retry timed out unexpectedly"); diff --git a/extensions/ollama/src/stream-runtime.test.ts b/extensions/ollama/src/stream-runtime.test.ts index 06d12b3ec43d..27d70ab05fbc 100644 --- a/extensions/ollama/src/stream-runtime.test.ts +++ b/extensions/ollama/src/stream-runtime.test.ts @@ -456,32 +456,6 @@ describe("createConfiguredOllamaCompatStreamWrapper", () => { }, ); - it("reports the real HTTP response before consuming native Ollama output", async () => { - await withSuccessfulOllamaFetch(async () => { - const onProviderAccepted = vi.fn(); - const onResponse = vi.fn(); - const stream = await createOllamaTestStream({ - baseUrl: "http://ollama-host:11434", - options: { onProviderAccepted, onResponse }, - }); - - await collectStreamEvents(stream); - - expect(onProviderAccepted).toHaveBeenCalledWith( - { - kind: "http_response", - status: 200, - headers: { "content-type": "application/x-ndjson" }, - }, - expect.objectContaining({ provider: "custom-ollama" }), - ); - expect(onResponse).toHaveBeenCalledWith( - { status: 200, headers: { "content-type": "application/x-ndjson" } }, - expect.objectContaining({ provider: "custom-ollama" }), - ); - }); - }); - it("passes resolved provider request timeouts to native Ollama chat fetches", async () => { await withMockNdjsonFetch( [ diff --git a/extensions/ollama/src/stream.runtime.ts b/extensions/ollama/src/stream.runtime.ts index 0a449ef1d6f8..2db5ebe1410a 100644 --- a/extensions/ollama/src/stream.runtime.ts +++ b/extensions/ollama/src/stream.runtime.ts @@ -19,7 +19,6 @@ import { createAssistantMessageEventStream } from "openclaw/plugin-sdk/llm"; import type { ProviderRuntimeModel } from "openclaw/plugin-sdk/plugin-entry"; import { isNonSecretApiKeyMarker } from "openclaw/plugin-sdk/provider-auth"; import { readResponseTextLimited } from "openclaw/plugin-sdk/provider-http"; -import { notifyProviderHttpResponse } from "openclaw/plugin-sdk/provider-lifecycle"; import { createPlainTextToolCallCompatWrapper } from "openclaw/plugin-sdk/provider-stream-shared"; import { createSubsystemLogger } from "openclaw/plugin-sdk/runtime-env"; import { fetchWithSsrFGuard } from "openclaw/plugin-sdk/ssrf-runtime"; @@ -76,6 +75,36 @@ function throwIfOllamaStreamAborted(signal?: AbortSignal): void { } } +async function runOllamaResponseHook(params: { + hook: (() => void | Promise) | undefined; + signal: AbortSignal | undefined; +}): Promise { + const { hook, signal } = params; + if (!hook) { + return; + } + throwIfOllamaStreamAborted(signal); + if (!signal) { + await hook(); + return; + } + let onAbort: (() => void) | undefined; + try { + await Promise.race([ + Promise.resolve().then(hook), + new Promise((_resolve, reject) => { + onAbort = () => reject(new Error("Request was aborted")); + signal.addEventListener("abort", onAbort, { once: true }); + }), + ]); + } finally { + if (onAbort) { + signal.removeEventListener("abort", onAbort); + } + } + throwIfOllamaStreamAborted(signal); +} + function createOllamaStreamCooperativeScheduler( signal?: AbortSignal, ): OllamaStreamCooperativeScheduler { @@ -1026,7 +1055,26 @@ function createRawOllamaStreamFn( }); try { - await notifyProviderHttpResponse({ options, response, model }); + const responseHook = options?.onResponse; + try { + await runOllamaResponseHook({ + hook: responseHook + ? () => + responseHook( + { + status: response.status, + headers: Object.fromEntries(response.headers.entries()), + }, + model, + ) + : undefined, + signal: options?.signal, + }); + } catch (error) { + // A pending body cancel must not stall release or the terminal error. + void response.body?.cancel().catch(() => undefined); + throw error; + } if (!response.ok) { const errorText = await readResponseTextLimited( response, diff --git a/package.json b/package.json index e33abe467d6f..c3e1249db5f5 100644 --- a/package.json +++ b/package.json @@ -1319,10 +1319,6 @@ "./plugin-sdk/provider-http": { "default": "./dist/plugin-sdk/provider-http.js" }, - "./plugin-sdk/provider-lifecycle": { - "types": "./dist/plugin-sdk/provider-lifecycle.d.ts", - "default": "./dist/plugin-sdk/provider-lifecycle.js" - }, "./plugin-sdk/provider-binary-stream": { "types": "./dist/plugin-sdk/provider-binary-stream.d.ts", "default": "./dist/plugin-sdk/provider-binary-stream.js" diff --git a/packages/agent-core/src/agent.ts b/packages/agent-core/src/agent.ts index 124b819c95b3..a28896bb5779 100644 --- a/packages/agent-core/src/agent.ts +++ b/packages/agent-core/src/agent.ts @@ -118,8 +118,6 @@ export interface AgentOptions { getApiKey?: (provider: string) => Promise | string | undefined; /** Inspect the provider payload before it is sent. */ onPayload?: SimpleStreamOptions["onPayload"]; - /** Observe when the provider accepts the request. */ - onProviderAccepted?: SimpleStreamOptions["onProviderAccepted"]; /** Inspect the provider response after it returns. */ onResponse?: SimpleStreamOptions["onResponse"]; /** Hook that may short-circuit or alter a tool call before execution. */ @@ -231,7 +229,6 @@ export class Agent { public streamFn: StreamFn; public getApiKey?: (provider: string) => Promise | string | undefined; public onPayload?: SimpleStreamOptions["onPayload"]; - public onProviderAccepted?: SimpleStreamOptions["onProviderAccepted"]; public onResponse?: SimpleStreamOptions["onResponse"]; public beforeToolCall?: ( context: BeforeToolCallContext, @@ -273,7 +270,6 @@ export class Agent { this.streamFn = resolveAgentCoreStreamFn(options.runtime, options.streamFn); this.getApiKey = options.getApiKey; this.onPayload = options.onPayload; - this.onProviderAccepted = options.onProviderAccepted; this.onResponse = options.onResponse; this.beforeToolCall = options.beforeToolCall; this.resolveDeferredTool = options.resolveDeferredTool; @@ -525,7 +521,6 @@ export class Agent { ), sessionId: this.sessionId, onPayload: this.onPayload, - onProviderAccepted: this.onProviderAccepted, onResponse: this.onResponse, transport: this.transport, thinkingBudgets: this.thinkingBudgets, diff --git a/packages/ai/src/providers/anthropic.ts b/packages/ai/src/providers/anthropic.ts index bdc6c6ec7851..e21a997c62bc 100644 --- a/packages/ai/src/providers/anthropic.ts +++ b/packages/ai/src/providers/anthropic.ts @@ -29,10 +29,7 @@ import { type AnthropicCompactionBlock, } from "../transports/anthropic-compaction-replay.js"; import { applyAnthropicCacheControlToMessages } from "../transports/anthropic-payload-policy.js"; -import { - notifyProviderHttpResponse, - transportAbortError, -} from "../transports/transport-stream-shared.js"; +import { transportAbortError } from "../transports/transport-stream-shared.js"; import { MALFORMED_STREAMING_FRAGMENT_ERROR_MESSAGE } from "../transports/transport-utils.js"; import type { AnthropicMessagesCompat, @@ -52,6 +49,7 @@ import type { } from "../types.js"; import { createDeferredEventBuffer } from "../utils/deferred-event-buffer.js"; import { AssistantMessageEventStream } from "../utils/event-stream.js"; +import { headersToRecord } from "../utils/headers.js"; import { parseJsonWithRepair, parseStreamingJson } from "../utils/json-parse.js"; import { notifyLlmRequestActivity } from "../utils/llm-request-activity.js"; import { projectProviderError } from "../utils/provider-error.js"; @@ -424,7 +422,10 @@ export const streamAnthropic: StreamFunction<"anthropic-messages", AnthropicComp const response = await client.messages .create({ ...params, stream: true }, sdkRequestOptions) .asResponse(); - await notifyProviderHttpResponse({ options: requestOptions, response, model }); + await requestOptions?.onResponse?.( + { status: response.status, headers: headersToRecord(response.headers) }, + model, + ); type Block = (ThinkingContent | TextContent | (ToolCall & { partialJson: string })) & { index: number; diff --git a/packages/ai/src/providers/google-shared.test.ts b/packages/ai/src/providers/google-shared.test.ts index 76d3228d7bc0..3c9412d00f36 100644 --- a/packages/ai/src/providers/google-shared.test.ts +++ b/packages/ai/src/providers/google-shared.test.ts @@ -590,43 +590,6 @@ describe("consumeGoogleGenerateContentStream", () => { }); describe("runGoogleGenerateContentLifecycle", () => { - it("reports SDK stream acceptance without fabricated HTTP metadata", async () => { - const onProviderAccepted = vi.fn(); - - const { result } = await runGoogleFixture( - [googleResponse({ parts: [{ text: "ok" }], finishReason: FinishReason.STOP })], - { options: { onProviderAccepted } }, - ); - - expect(result.stopReason).toBe("stop"); - expect(onProviderAccepted).toHaveBeenCalledWith({ kind: "provider_stream_opened" }, model); - }); - - it("closes an unread SDK stream without waiting when acceptance fails", async () => { - const close = vi.fn(() => new Promise>(() => {})); - const googleStream = { - next: vi.fn(), - return: close, - throw: vi.fn(), - [Symbol.asyncIterator]() { - return this; - }, - } as unknown as AsyncGenerator; - - const { result } = await runGoogleFixture([], { - options: { - onProviderAccepted: () => Promise.reject(new Error("acceptance callback failed")), - }, - generateContentStream: async () => googleStream, - }); - - expect(result).toMatchObject({ - stopReason: "error", - errorMessage: "acceptance callback failed", - }); - expect(close).toHaveBeenCalledOnce(); - }); - it.each(["google-generative-ai", "google-vertex"] as const)( "rejects an unfinished %s stream instead of silently completing partial output", async (api) => { diff --git a/packages/ai/src/providers/google-shared.ts b/packages/ai/src/providers/google-shared.ts index cb0952df9a50..b8df401bc97e 100644 --- a/packages/ai/src/providers/google-shared.ts +++ b/packages/ai/src/providers/google-shared.ts @@ -19,7 +19,6 @@ import { googleFlashSupportsMinimalThinking } from "../transports/google-thinkin import { assignTransportErrorDetails, coerceTransportToolCallArguments, - notifyProviderStreamOpened, transportAbortError, } from "../transports/transport-stream-shared.js"; import type { @@ -422,7 +421,7 @@ export async function runGoogleGenerateContentLifecycle stream: AssistantMessageEventStream; model: Model; output: AssistantMessage; - options?: Pick; + options?: Pick; createClient: () => GoogleGenerateContentClient; buildParams: () => GenerateContentParameters; nextToolCallId: (name: string | undefined) => string; @@ -437,16 +436,8 @@ export async function runGoogleGenerateContentLifecycle requestParams = nextParams as GenerateContentParameters; } const googleStream = await client.models.generateContentStream(requestParams); - const googleIterator = googleStream[Symbol.asyncIterator](); - try { - await notifyProviderStreamOpened({ options, model }); - } catch (error) { - // Cleanup is best effort; callback failure must not wait on an unread SDK stream. - void Promise.resolve(googleIterator.return?.()).catch(() => undefined); - throw error; - } await consumeGoogleGenerateContentStream({ - chunks: { [Symbol.asyncIterator]: () => googleIterator }, + chunks: googleStream, model, output, stream, diff --git a/packages/ai/src/providers/mistral.test.ts b/packages/ai/src/providers/mistral.test.ts index e6eb0a76632a..c34a47b331bb 100644 --- a/packages/ai/src/providers/mistral.test.ts +++ b/packages/ai/src/providers/mistral.test.ts @@ -9,7 +9,6 @@ const mistralMockState = vi.hoisted(() => ({ payloads: [] as unknown[], requestOptions: [] as unknown[], randomUUIDs: [] as string[], - requestThroughHttpClient: false, streamError: new Error("stop before network") as unknown, streamResult: undefined as unknown, })); @@ -28,10 +27,7 @@ vi.mock("@mistralai/mistralai", async () => { return { ...actual, Mistral: class MockMistral { - private readonly config: unknown; - constructor(config: unknown) { - this.config = config; mistralMockState.configs.push(config); } @@ -39,19 +35,6 @@ vi.mock("@mistralai/mistralai", async () => { stream: vi.fn(async (payload: unknown, requestOptions: unknown) => { mistralMockState.payloads.push(payload); mistralMockState.requestOptions.push(requestOptions); - if (mistralMockState.requestThroughHttpClient) { - const httpClient = ( - this.config as { - httpClient?: { request(request: Request): Promise }; - } - ).httpClient; - const response = await httpClient?.request(new Request("https://api.mistral.ai/chat")); - if (response && !response.ok) { - throw Object.assign(new Error(`Mistral HTTP ${response.status}`), { - statusCode: response.status, - }); - } - } if (mistralMockState.streamResult !== undefined) { return mistralMockState.streamResult; } @@ -236,7 +219,6 @@ describe("Mistral provider", () => { mistralMockState.payloads = []; mistralMockState.requestOptions = []; mistralMockState.randomUUIDs = []; - mistralMockState.requestThroughHttpClient = false; mistralMockState.streamError = new Error("stop before network"); mistralMockState.streamResult = undefined; }); @@ -245,130 +227,6 @@ describe("Mistral provider", () => { configureAiTransportHost({}); }); - it("reports the real HTTP response captured by the Mistral HTTPClient hook", async () => { - mistralMockState.requestThroughHttpClient = true; - mistralMockState.streamResult = { - async *[Symbol.asyncIterator]() { - yield { - data: { - id: "resp-http-ack", - model: "mistral-large-latest", - choices: [{ finishReason: "stop", delta: { content: "ok" } }], - }, - }; - }, - }; - const hostFetch = vi.fn( - async () => - new Response("stream", { - status: 200, - headers: { - "content-type": "text/event-stream", - "x-mistral-request-id": "req-1", - }, - }), - ); - configureAiTransportHost({ buildModelFetch: () => hostFetch }); - const onProviderAccepted = vi.fn(); - const onResponse = vi.fn(); - - const result = await runSimpleMistralFixture(context, { - onProviderAccepted, - onResponse, - }); - - expect(result.stopReason).toBe("stop"); - expect(onProviderAccepted).toHaveBeenCalledWith( - { - kind: "http_response", - status: 200, - headers: expect.objectContaining({ - "content-type": "text/event-stream", - "x-mistral-request-id": "req-1", - }), - }, - expect.objectContaining({ provider: "mistral" }), - ); - expect(onResponse).toHaveBeenCalledWith( - { - status: 200, - headers: expect.objectContaining({ "x-mistral-request-id": "req-1" }), - }, - expect.objectContaining({ provider: "mistral" }), - ); - expect(hostFetch).toHaveBeenCalledOnce(); - }); - - it("cancels an unread Mistral stream when provider acceptance fails", async () => { - mistralMockState.requestThroughHttpClient = true; - const cancel = vi.fn(async () => undefined); - mistralMockState.streamResult = { - cancel, - async *[Symbol.asyncIterator]() { - yield { - data: { - id: "resp-http-ack", - model: "mistral-large-latest", - choices: [{ finishReason: "stop", delta: { content: "ok" } }], - }, - }; - }, - }; - configureAiTransportHost({ - buildModelFetch: () => async () => new Response("stream", { status: 200 }), - }); - const hookError = new Error("acceptance callback failed"); - - const result = await runSimpleMistralFixture(context, { - onProviderAccepted: () => Promise.reject(hookError), - }); - - expect(result).toMatchObject({ - stopReason: "error", - errorMessage: "acceptance callback failed", - }); - expect(cancel).toHaveBeenCalledWith(hookError); - }); - - it("reports a rejected HTTP response without marking it accepted", async () => { - mistralMockState.requestThroughHttpClient = true; - const hostFetch = vi.fn( - async () => - new Response("rate limited", { - status: 429, - headers: { "x-mistral-request-id": "req-rejected" }, - }), - ); - configureAiTransportHost({ buildModelFetch: () => hostFetch }); - const onProviderAccepted = vi.fn(); - const onResponse = vi.fn(); - - const result = await runSimpleMistralFixture(context, { - onProviderAccepted, - onResponse, - }); - - expect(result.stopReason).toBe("error"); - expect(onProviderAccepted).not.toHaveBeenCalled(); - expect(onResponse).toHaveBeenCalledWith( - { - status: 429, - headers: expect.objectContaining({ "x-mistral-request-id": "req-rejected" }), - }, - expect.objectContaining({ provider: "mistral" }), - ); - expect(hostFetch).toHaveBeenCalledOnce(); - }); - - it("does not report acceptance when SDK stream setup fails", async () => { - const onProviderAccepted = vi.fn(); - - const result = await runSimpleMistralFixture(context, { onProviderAccepted }); - - expect(result.stopReason).toBe("error"); - expect(onProviderAccepted).not.toHaveBeenCalled(); - }); - it("forwards simple stop sequences to Mistral stop", async () => { const result = await runSimpleMistralFixture(context, { stop: ["STOP"], diff --git a/packages/ai/src/providers/mistral.ts b/packages/ai/src/providers/mistral.ts index a3490dcc504a..a9e28c9764de 100644 --- a/packages/ai/src/providers/mistral.ts +++ b/packages/ai/src/providers/mistral.ts @@ -13,10 +13,7 @@ import { getEnvApiKey } from "../env-api-keys.js"; import { getAiTransportHost } from "../host.js"; import { calculateCost, clampThinkingLevel } from "../model-utils.js"; import { transformProviderMessages as transformMessages } from "../provider-transcript-transform.js"; -import { - notifyProviderHttpResponse, - transportAbortError, -} from "../transports/transport-stream-shared.js"; +import { transportAbortError } from "../transports/transport-stream-shared.js"; import type { AssistantMessage, Context, @@ -141,28 +138,24 @@ export const streamMistral: StreamFunction<"mistral-conversations", MistralOptio throw new Error(`No API key for provider: ${model.provider}`); } - const boundedFetcher = createBoundedMistralFetcher( - MISTRAL_STREAM_BODY_MAX_BYTES, - getAiTransportHost().buildModelFetch(model) ?? fetch, - ); - let mistralResponse: Response | undefined; - let reportedResponse: Response | undefined; - const httpClient = new HTTPClient({ fetcher: boundedFetcher }); - httpClient.addHook("response", async (response) => { - mistralResponse = response; - if (!response.ok) { - await notifyProviderHttpResponse({ options, response, model }); - reportedResponse = response; - } - }); // Intentionally per-request: avoids shared SDK mutable state across concurrent consumers. const mistral = new Mistral({ apiKey, serverURL: model.baseUrl, // Bound the streamed Mistral response body at 16 MiB so a hostile or - // malfunctioning endpoint cannot exhaust memory. The HTTPClient is the - // SDK's public fetch and response-hook boundary for every chat.stream attempt. - httpClient, + // malfunctioning endpoint cannot exhaust memory. The fetcher is + // injected via the SDK's `HTTPClient` (see + // `@mistralai/mistralai/lib/sdks.ts` `ClientSDK` constructor: when + // `httpClient` is passed, `ClientSDK.#httpClient` is set from it and + // every `chat.stream` / `complete` call routes through + // `HTTPClient.request` → `this.fetcher(req)`). + // Mistral accepts HTTPClient.fetcher, so compose guarded egress with the byte cap. + httpClient: new HTTPClient({ + fetcher: createBoundedMistralFetcher( + MISTRAL_STREAM_BODY_MAX_BYTES, + getAiTransportHost().buildModelFetch(model) ?? fetch, + ), + }), }); const normalizeMistralToolCallId = createMistralToolCallIdNormalizer(); @@ -185,16 +178,6 @@ export const streamMistral: StreamFunction<"mistral-conversations", MistralOptio headers, signal: options?.signal, }); - if (mistralResponse && mistralResponse !== reportedResponse) { - try { - await notifyProviderHttpResponse({ options, response: mistralResponse, model }); - } catch (error) { - // The SDK EventStream owns the locked response reader after chat.stream resolves. - // Cancellation is best effort and must not delay the lifecycle callback failure. - void mistralStream.cancel(error).catch(() => undefined); - throw error; - } - } stream.push({ type: "start", partial: output }); await consumeChatStream(model, output, stream, mistralStream); diff --git a/packages/ai/src/providers/openai-chatgpt-responses-streaming.test.ts b/packages/ai/src/providers/openai-chatgpt-responses-streaming.test.ts index 237e7e6df18d..4ad10c424edf 100644 --- a/packages/ai/src/providers/openai-chatgpt-responses-streaming.test.ts +++ b/packages/ai/src/providers/openai-chatgpt-responses-streaming.test.ts @@ -151,57 +151,6 @@ describe("OpenAI ChatGPT Responses inference streaming", () => { }); }); - it("reports acceptance before the default WebSocket stream starts", async () => { - class AcceptedWebSocket extends EventTarget { - constructor() { - super(); - queueMicrotask(() => this.dispatchEvent(new Event("open"))); - } - - send(): void { - queueMicrotask(() => { - this.dispatchEvent( - Object.assign(new Event("message"), { - data: JSON.stringify({ - type: "response.completed", - response: { - id: "resp_ws_accepted", - status: "completed", - output: [], - usage: { input_tokens: 5, output_tokens: 3, total_tokens: 8 }, - }, - }), - }), - ); - }); - } - - close(): void {} - } - - const order: string[] = []; - const onProviderAccepted = vi.fn(async () => { - order.push("accepted"); - }); - const fetchMock = vi.fn(); - vi.stubGlobal("WebSocket", AcceptedWebSocket); - vi.stubGlobal("fetch", fetchMock); - - const stream = streamOpenAICodexResponses(model, context, { - apiKey: createJwt({ - "https://api.openai.com/auth": { chatgpt_account_id: "acct-1" }, - }), - onProviderAccepted, - }); - for await (const event of stream) { - order.push(event.type); - } - - expect(order).toEqual(["accepted", "start", "done"]); - expect(onProviderAccepted).toHaveBeenCalledWith({ kind: "provider_stream_opened" }, model); - expect(fetchMock).not.toHaveBeenCalled(); - }); - it("emits an error for a content-filtered incomplete WebSocket turn", async () => { class ContentFilteredWebSocket extends EventTarget { constructor() { diff --git a/packages/ai/src/providers/openai-chatgpt-responses.encrypted-retry.test.ts b/packages/ai/src/providers/openai-chatgpt-responses.encrypted-retry.test.ts index 53191570d3f0..0a95003e94b0 100644 --- a/packages/ai/src/providers/openai-chatgpt-responses.encrypted-retry.test.ts +++ b/packages/ai/src/providers/openai-chatgpt-responses.encrypted-retry.test.ts @@ -544,47 +544,6 @@ describe("ChatGPT Responses encrypted replay recovery", () => { expect(onCompactionRejected).toHaveBeenCalledOnce(); }); - it("WebSocket commits stripped compaction before a provider acceptance callback fails", async () => { - const context = createReplayContext("compaction"); - const onCompactionRejected = vi.fn(); - const observations: ResponsesPromptObservation[] = []; - const scripted = installScriptedWebSocket([ - { events: [invalidEncryptedEvent()] }, - { events: [completionEvent("resp_ws_hook_failure")] }, - ]); - const onProviderAccepted = vi.fn(async () => { - if (scripted.requests.length >= 2) { - throw new Error("acceptance callback failed"); - } - }); - const options = createObservedOptions( - { - apiKey: createJwt(), - transport: "websocket" as const, - onCompactionRejected, - onProviderAccepted, - ...REPLAY_IDENTITY, - }, - observations, - ); - - const result = await streamOpenAICodexResponses(model, context, options).result(); - - expect(result).toMatchObject({ - stopReason: "error", - errorMessage: "acceptance callback failed", - providerReplay: { type: "openai-responses-compaction-suppression" }, - }); - expect(scripted.requests).toHaveLength(2); - expect(hasInputType(requireItem(scripted.requests, 0), "compaction")).toBe(true); - expect(hasInputType(requireItem(scripted.requests, 1), "compaction")).toBe(false); - expect(observations.map((entry) => entry.payloadVariant)).toEqual([ - "initial", - "compaction-stripped", - ]); - expect(onCompactionRejected).toHaveBeenCalledOnce(); - }); - it("WebSocket preserves compaction when reasoning-stripped recovery succeeds", async () => { const context = createReplayContext("mixed"); const observations: ResponsesPromptObservation[] = []; diff --git a/packages/ai/src/providers/openai-chatgpt-responses.retry.test.ts b/packages/ai/src/providers/openai-chatgpt-responses.retry.test.ts index 5de4abdcccf9..5a3256e4ff25 100644 --- a/packages/ai/src/providers/openai-chatgpt-responses.retry.test.ts +++ b/packages/ai/src/providers/openai-chatgpt-responses.retry.test.ts @@ -92,13 +92,9 @@ describe("streamOpenAICodexResponses retry classification", () => { return 0 as unknown as ReturnType; }); - const onProviderAccepted = vi.fn(); - const onResponse = vi.fn(); const options = { apiKey: jwt, transport: "sse" as const, - onProviderAccepted, - onResponse, }; responsesPromptObserver.set(options, (observation) => observations.push(observation)); @@ -110,8 +106,6 @@ describe("streamOpenAICodexResponses retry classification", () => { expect(result.stopReason).toBe("error"); expect(fetchMock).toHaveBeenCalledTimes(2); - expect(onProviderAccepted).not.toHaveBeenCalled(); - expect(onResponse.mock.calls.map(([response]) => response.status)).toEqual([503, 401]); expect(observations).toHaveLength(2); expect(observations.every((entry) => entry.egress === "native-codex-sse")).toBe(true); expect(observations.every((entry) => entry.payloadVariant === "initial")).toBe(true); diff --git a/packages/ai/src/providers/openai-chatgpt-responses.ts b/packages/ai/src/providers/openai-chatgpt-responses.ts index d10563e7a66b..81c7ab1a5ddd 100644 --- a/packages/ai/src/providers/openai-chatgpt-responses.ts +++ b/packages/ai/src/providers/openai-chatgpt-responses.ts @@ -33,12 +33,8 @@ import { type ResponsesEncryptedContentAttempt, } from "../transports/openai-responses-replay-internal.js"; import { processResponsesStream } from "../transports/openai-responses-stream-internal.js"; +import { createOpenAIResponseHook } from "../transports/openai-transport-shared.js"; import { - createOpenAIProviderAcceptanceHook, - createOpenAIResponseHook, -} from "../transports/openai-transport-shared.js"; -import { - notifyProviderStreamOpened, transportAbortError, withProviderResponseHook, } from "../transports/transport-stream-shared.js"; @@ -363,11 +359,9 @@ export const streamOpenAICodexResponses: StreamFunction< output, stream, model, - () => { - websocketStarted = true; - }, () => { commitSemanticAttempt(activeAttempt); + websocketStarted = true; }, requestOptions, firstEventAbort.abort, @@ -588,7 +582,7 @@ export const streamOpenAICodexResponses: StreamFunction< stream: mapCodexEvents(parseOpenAIChatGptResponsesSse(response)), signal: firstEventAbort.signal, abort: firstEventAbort.abort, - hook: createOpenAIProviderAcceptanceHook(options, response, model), + hook: createOpenAIResponseHook(options?.onResponse, response, model), onReady: () => stream.push({ type: "start", partial: output }), }); await processResponsesStream(hookedResponseStream, output, stream, model, { @@ -1459,17 +1453,13 @@ async function* startWebSocketOutputOnFirstEvent( events: AsyncIterable, output: AssistantMessage, stream: AssistantMessageEventStream, - onFirstProviderEvent: () => void, - onProviderAccepted: () => Promise, onStart: () => void, ): AsyncGenerator { let started = false; for await (const event of events) { if (!started) { started = true; - onFirstProviderEvent(); onStart(); - await onProviderAccepted(); stream.push({ type: "start", partial: output }); } yield event; @@ -1483,7 +1473,6 @@ async function processWebSocketStream( output: AssistantMessage, stream: AssistantMessageEventStream, model: Model<"openai-chatgpt-responses">, - onFirstProviderEvent: () => void, onStart: () => void, options?: OpenAICodexResponsesOptions, abortFirstEventStream?: (reason: Error) => void, @@ -1520,8 +1509,6 @@ async function processWebSocketStream( mapCodexEvents(parseWebSocket(socket, options?.signal)), output, stream, - onFirstProviderEvent, - () => notifyProviderStreamOpened({ options, model }), onStart, ), output, diff --git a/packages/ai/src/providers/openai-completions.ts b/packages/ai/src/providers/openai-completions.ts index 20cab4f377ce..e007e755fd36 100644 --- a/packages/ai/src/providers/openai-completions.ts +++ b/packages/ai/src/providers/openai-completions.ts @@ -19,7 +19,7 @@ import { } from "../transports/openai-completions-compat.js"; import { resolveOpenAIReasoningEffortMap } from "../transports/openai-reasoning-compat.js"; import { - createOpenAIProviderAcceptanceHook, + createOpenAIResponseHook, isOpenAICompletionsThinkingEnabled, parseOpenAICompletionsUsage, readOpenAICompletionsContentDeltas, @@ -184,7 +184,7 @@ export const streamOpenAICompletions: StreamFunction< stream: openaiStream, signal: firstEventAbort.signal, abort: firstEventAbort.abort, - hook: createOpenAIProviderAcceptanceHook(options, response, model), + hook: createOpenAIResponseHook(options?.onResponse, response, model), onReady: () => stream.push({ type: "start", partial: output }), }); diff --git a/packages/ai/src/providers/openai-responses-shared.ts b/packages/ai/src/providers/openai-responses-shared.ts index 9058b4d4a422..a3cfdd65e4b7 100644 --- a/packages/ai/src/providers/openai-responses-shared.ts +++ b/packages/ai/src/providers/openai-responses-shared.ts @@ -18,7 +18,7 @@ import { convertProviderResponsesMessages, } from "../transports/openai-responses-replay-internal.js"; import { processResponsesStream } from "../transports/openai-responses-stream-internal.js"; -import { createOpenAIProviderAcceptanceHook } from "../transports/openai-transport-shared.js"; +import { createOpenAIResponseHook } from "../transports/openai-transport-shared.js"; import { transportAbortError, withProviderResponseHook, @@ -92,13 +92,7 @@ type ResponsesStreamClient = { type ResponsesLifecycleStreamOptions = Pick< StreamOptions, - | "signal" - | "timeoutMs" - | "maxRetries" - | "onPayload" - | "onProviderAccepted" - | "onResponse" - | "sessionId" + "signal" | "timeoutMs" | "maxRetries" | "onPayload" | "onResponse" | "sessionId" > & Pick & FirstStreamEventInternalOptions; @@ -302,7 +296,7 @@ export async function runResponsesStreamLifecycle(params: { stream: openaiStream, signal: firstEventAbort.signal, abort: firstEventAbort.abort, - hook: createOpenAIProviderAcceptanceHook(options, response, model), + hook: createOpenAIResponseHook(options?.onResponse, response, model), onReady: () => stream.push({ type: "start", partial: output }), }); diff --git a/packages/ai/src/providers/simple-options.ts b/packages/ai/src/providers/simple-options.ts index c6e852f344a7..e0ae64ac9a74 100644 --- a/packages/ai/src/providers/simple-options.ts +++ b/packages/ai/src/providers/simple-options.ts @@ -31,7 +31,6 @@ export function buildBaseOptions( promptCacheKey: options?.promptCacheKey, headers: options?.headers, onPayload: options?.onPayload, - onProviderAccepted: options?.onProviderAccepted, onResponse: options?.onResponse, timeoutMs: options?.timeoutMs, firstEventTimeoutMs: firstEventOptions?.firstEventTimeoutMs, diff --git a/packages/ai/src/transports/anthropic-transport-stream.test.ts b/packages/ai/src/transports/anthropic-transport-stream.test.ts index c365b1cc48dd..f4940e358f37 100644 --- a/packages/ai/src/transports/anthropic-transport-stream.test.ts +++ b/packages/ai/src/transports/anthropic-transport-stream.test.ts @@ -1129,25 +1129,18 @@ describe("anthropic transport stream", () => { ), ); - const onProviderAccepted = vi.fn(); - const onResponse = vi.fn(); const result = await runTransportStream( makeAnthropicTransportModel(), { messages: [{ role: "user", content: "hello" }], } as AnthropicStreamContext, - { apiKey: "test-api-key", onProviderAccepted, onResponse } as AnthropicStreamOptions, + { apiKey: "test-api-key" } as AnthropicStreamOptions, ); expect(result.stopReason).toBe("error"); expect(result.errorMessage).toBe( 'HTTP 429: {"type":"error","error":{"type":"rate_limit_error","message":"Number of request tokens exceeded the per-minute rate limit."}}; Retry-After: 30 seconds', ); - expect(onProviderAccepted).not.toHaveBeenCalled(); - expect(onResponse).toHaveBeenCalledWith( - { status: 429, headers: { "content-type": "text/plain;charset=UTF-8", "retry-after": "30" } }, - expect.objectContaining({ provider: "anthropic" }), - ); }); it("bounds streamed Anthropic error responses without content-length", async () => { @@ -3801,33 +3794,6 @@ describe("anthropic transport stream", () => { expect(cancelReason).toBe(abortReason); }); - it("cancels an unread SSE body when provider acceptance fails", async () => { - let cancelCalled = false; - guardedFetchMock.mockResolvedValueOnce( - createOpenRawSseResponse({ - body: "", - onCancel: () => { - cancelCalled = true; - }, - }), - ); - - const result = await runTransportStream( - makeAnthropicTransportModel(), - { messages: [{ role: "user", content: "hello" }] } as AnthropicStreamContext, - { - apiKey: "sk-ant-api", - onProviderAccepted: () => Promise.reject(new Error("acceptance callback failed")), - } as AnthropicStreamOptions, - ); - - expect(result).toMatchObject({ - stopReason: "error", - errorMessage: "acceptance callback failed", - }); - await vi.waitFor(() => expect(cancelCalled).toBe(true)); - }); - it("cancels open SSE bodies when Anthropic stream consumers throw", async () => { let cancelCalled = false; guardedFetchMock.mockResolvedValueOnce( @@ -4367,12 +4333,10 @@ describe("anthropic transport stream", () => { ]), ); const streamFn = createAnthropicMessagesTransportStreamFn(); - const onProviderAccepted = vi.fn(); - const onResponse = vi.fn(); const stream = streamFn( makeAnthropicTransportModel(), { messages: [{ role: "user", content: "hi" }] } as AnthropicStreamContext, - { apiKey: "sk-ant-api", onProviderAccepted, onResponse } as AnthropicStreamOptions, + { apiKey: "sk-ant-api" } as AnthropicStreamOptions, ); const eventTypes: string[] = []; @@ -4383,18 +4347,6 @@ describe("anthropic transport stream", () => { const startIndex = eventTypes.indexOf("start"); expect(startIndex).toBeGreaterThanOrEqual(0); expect(eventTypes.slice(0, startIndex).some((t) => t === "error")).toBe(false); - expect(onProviderAccepted).toHaveBeenCalledWith( - { - kind: "http_response", - status: 200, - headers: { "content-type": "text/event-stream" }, - }, - expect.objectContaining({ provider: "anthropic" }), - ); - expect(onResponse).toHaveBeenCalledWith( - { status: 200, headers: { "content-type": "text/event-stream" } }, - expect.objectContaining({ provider: "anthropic" }), - ); }); it("emits error without a preceding start event when SSE error arrives before message_start", async () => { @@ -4409,12 +4361,10 @@ describe("anthropic transport stream", () => { ), ); const streamFn = createAnthropicMessagesTransportStreamFn(); - const onProviderAccepted = vi.fn(); - const onResponse = vi.fn(); const stream = streamFn( makeAnthropicTransportModel(), { messages: [{ role: "user", content: "hi" }] } as AnthropicStreamContext, - { apiKey: "sk-ant-api", onProviderAccepted, onResponse } as AnthropicStreamOptions, + { apiKey: "sk-ant-api" } as AnthropicStreamOptions, ); const eventTypes: string[] = []; @@ -4426,18 +4376,6 @@ describe("anthropic transport stream", () => { // surfaces the SSE error as an explicit "error" event or silently ends the // stream (a timing artefact of synchronous mock SSE delivery). expect(eventTypes).not.toContain("start"); - expect(onProviderAccepted).toHaveBeenCalledWith( - { - kind: "http_response", - status: 200, - headers: { "content-type": "text/event-stream" }, - }, - expect.objectContaining({ provider: "anthropic" }), - ); - expect(onResponse).toHaveBeenCalledWith( - { status: 200, headers: { "content-type": "text/event-stream" } }, - expect.objectContaining({ provider: "anthropic" }), - ); }); }); /* oxlint-disable max-lines -- TODO: split this grandfathered oversized file. */ diff --git a/packages/ai/src/transports/anthropic-transport-stream.ts b/packages/ai/src/transports/anthropic-transport-stream.ts index 4b564b55c9a7..bba288c39ae6 100644 --- a/packages/ai/src/transports/anthropic-transport-stream.ts +++ b/packages/ai/src/transports/anthropic-transport-stream.ts @@ -107,7 +107,6 @@ import { failTransportStream, finalizeTransportStream, mergeTransportHeaders, - notifyProviderHttpResponse, sanitizeNonEmptyTransportPayloadText, sanitizeTransportPayloadText, transportAbortError, @@ -144,10 +143,7 @@ type AnthropicMessagesClient = { stream( params: Record, options?: { signal?: AbortSignal }, - ): Promise<{ - response: Response; - stream: AsyncIterable> | Iterable>; - }>; + ): AsyncIterable>; }; }; @@ -745,7 +741,7 @@ function createAnthropicMessagesClient(params: { const url = resolveAnthropicMessagesUrl(params.baseURL); return { messages: { - async stream(body: Record, options?: { signal?: AbortSignal }) { + async *stream(body: Record, options?: { signal?: AbortSignal }) { const headers = mergeTransportHeaders( { "content-type": "application/json", @@ -761,10 +757,14 @@ function createAnthropicMessagesClient(params: { body: JSON.stringify(body), signal: options?.signal, }); - return { - response, - stream: response.body ? parseAnthropicSseBody(response.body, options?.signal) : [], - }; + if (!response.ok) { + const detail = await readAnthropicMessagesErrorBodySnippet(response); + throw new Error(formatAnthropicMessagesHttpError(response, detail)); + } + if (!response.body) { + return; + } + yield* parseAnthropicSseBody(response.body, options?.signal); }, }, }; @@ -1094,8 +1094,6 @@ function resolveAnthropicTransportOptions( sessionId: options?.sessionId, headers: options?.headers, onPayload: options?.onPayload, - onProviderAccepted: options?.onProviderAccepted, - onResponse: options?.onResponse, maxRetryDelayMs: options?.maxRetryDelayMs, metadata: options?.metadata, interleavedThinking: options?.interleavedThinking, @@ -1193,15 +1191,10 @@ export function createAnthropicMessagesTransportStreamFn(): StreamFn { params = nextParams as Record; } applyClaudeRequestContract(params, model); - const { response, stream: anthropicStream } = await client.messages.stream( + const anthropicStream = client.messages.stream( { ...params, stream: true }, transportOptions.signal ? { signal: transportOptions.signal } : undefined, ); - await notifyProviderHttpResponse({ options: transportOptions, response, model }); - if (!response.ok) { - const detail = await readAnthropicMessagesErrorBodySnippet(response); - throw new Error(formatAnthropicMessagesHttpError(response, detail)); - } const blocks = output.content; const blockIndexes = new Map(); const compactionCapture = createCompactionCapture(output, model, transportOptions); diff --git a/packages/ai/src/transports/openai-completions-transport.response-hook.test.ts b/packages/ai/src/transports/openai-completions-transport.response-hook.test.ts index a505065956d9..a5a4c30e9189 100644 --- a/packages/ai/src/transports/openai-completions-transport.response-hook.test.ts +++ b/packages/ai/src/transports/openai-completions-transport.response-hook.test.ts @@ -116,21 +116,12 @@ describe.each([ const hookCompleted = new Promise((resolve) => { continueHook = resolve; }); - const onProviderAccepted = vi.fn>( - (acceptance) => { - order.push(`accepted:${acceptance.kind}`); - }, - ); const onResponse = vi.fn>(async () => { order.push("hook:start"); await hookCompleted; order.push("hook:end"); }); - const stream = createStream(model, context, { - apiKey: "fixture-token", - onProviderAccepted, - onResponse, - }); + const stream = createStream(model, context, { apiKey: "fixture-token", onResponse }); const consume = (async () => { for await (const event of stream) { order.push(event.type); @@ -138,19 +129,7 @@ describe.each([ })(); await vi.waitFor(() => expect(onResponse).toHaveBeenCalledOnce()); - expect(order).toEqual(["accepted:http_response", "hook:start"]); - expect(onProviderAccepted).toHaveBeenCalledWith( - { - kind: "http_response", - status: 202, - headers: { - "content-type": "text/event-stream", - "x-ratelimit-remaining-requests": "42", - "x-request-id": "req_observable", - }, - }, - model, - ); + expect(order).toEqual(["hook:start"]); expect(onResponse).toHaveBeenCalledWith( { status: 202, @@ -166,12 +145,7 @@ describe.each([ continueHook(); await consume; expect((await stream.result()).stopReason).toBe("stop"); - expect(order.slice(0, 4)).toEqual([ - "accepted:http_response", - "hook:start", - "hook:end", - "start", - ]); + expect(order.slice(0, 3)).toEqual(["hook:start", "hook:end", "start"]); }); it.each(["throw", "reject"] as const)( @@ -203,32 +177,6 @@ describe.each([ }, ); - it("preserves an acceptance hook failure and closes the unread request", async () => { - const lifecycle = installResponse(); - const hookError = new Error("provider acceptance hook failed"); - const onProviderAccepted = vi.fn(() => Promise.reject(hookError)); - const onResponse = vi.fn(); - const stream = createStream(model, context, { - apiKey: "fixture-token", - onProviderAccepted, - onResponse, - }); - const eventTypes: string[] = []; - - for await (const event of stream) { - eventTypes.push(event.type); - } - - expect(await stream.result()).toMatchObject({ - stopReason: "error", - errorMessage: "provider acceptance hook failed", - }); - expect(onProviderAccepted).toHaveBeenCalledOnce(); - expect(onResponse).not.toHaveBeenCalled(); - expect(eventTypes).toEqual(["error"]); - expect(lifecycle.requestAborted).toHaveBeenCalledOnce(); - }); - it("applies the first-event timeout while the hook is pending", async () => { const lifecycle = installResponse(); const onFirstEventTimeout = vi.fn(); diff --git a/packages/ai/src/transports/openai-completions-transport.ts b/packages/ai/src/transports/openai-completions-transport.ts index 885c5b29f7d0..cd50c705b7b2 100644 --- a/packages/ai/src/transports/openai-completions-transport.ts +++ b/packages/ai/src/transports/openai-completions-transport.ts @@ -27,7 +27,6 @@ import { resolveCodeModeResponsesVisibleToolNames, } from "./openai-transport-params.js"; import { - createOpenAIProviderAcceptanceHook, createOpenAIResponseHook, type MutableAssistantOutput, type OpenAIModeModel, @@ -272,9 +271,7 @@ export function createOpenAICompletionsTransportStreamFn(): StreamFn { stream: responseStream, signal: firstEventAbort.signal, abort: firstEventAbort.abort, - hook: options?.onProviderAccepted - ? createOpenAIProviderAcceptanceHook(options, response, model) - : createOpenAIResponseHook(options?.onResponse, response, model), + hook: createOpenAIResponseHook(options?.onResponse, response, model), onReady: () => stream.push({ type: "start", partial: output }), }); await processCompletionsStream(hookedResponseStream, output, model, stream, { diff --git a/packages/ai/src/transports/openai-responses-client.ts b/packages/ai/src/transports/openai-responses-client.ts index 36585f1c4441..fa9e6ce94521 100644 --- a/packages/ai/src/transports/openai-responses-client.ts +++ b/packages/ai/src/transports/openai-responses-client.ts @@ -70,14 +70,13 @@ import { isOpenAICodexResponsesModel, resolveCodeModeResponsesVisibleToolNames, } from "./openai-transport-params.js"; -import { createOpenAIProviderAcceptanceHook, log } from "./openai-transport-shared.js"; +import { createOpenAIResponseHook, log } from "./openai-transport-shared.js"; import { sanitizeResponsesImagePayload } from "./responses-image-payload-sanitizer.js"; import { createWritableTransportEventStream, failTransportStream, finalizeTransportStream, mergeTransportMetadata, - notifyProviderStreamOpened, transportAbortError, withProviderResponseHook, } from "./transport-stream-shared.js"; @@ -449,7 +448,7 @@ function createResponsesTransportExecutor(config: ResponsesTransportExecutorOpti stream: observeResponsesStream(rawResponseStream, model, requestStartedAt), signal: firstEvent.signal, abort: firstEvent.abort, - hook: createOpenAIProviderAcceptanceHook(options, response, model), + hook: createOpenAIResponseHook(options?.onResponse, response, model), onReady: () => { emitModelTransportDebug( log, @@ -504,13 +503,8 @@ function createResponsesTransportExecutor(config: ResponsesTransportExecutorOpti ); responseStream = { async *[Symbol.asyncIterator]() { - let providerAccepted = false; try { for await (const event of websocket.stream) { - if (!providerAccepted) { - providerAccepted = true; - await notifyProviderStreamOpened({ options, model }); - } startStream(); yield event; } diff --git a/packages/ai/src/transports/openai-responses-websocket-client.test.ts b/packages/ai/src/transports/openai-responses-websocket-client.test.ts index 1ca8de466d5a..a1a61f687656 100644 --- a/packages/ai/src/transports/openai-responses-websocket-client.test.ts +++ b/packages/ai/src/transports/openai-responses-websocket-client.test.ts @@ -3,7 +3,6 @@ import { type AssistantMessage, type Context, type Model, - type StreamOptions, } from "@openclaw/llm-core"; import { WebSocketError } from "openai/resources/responses/internal-base.js"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; @@ -269,7 +268,6 @@ async function run( headers?: Record; observations?: ResponsesPromptObservation[]; onCompactionRejected?: () => void; - onProviderAccepted?: NonNullable; } = {}, ): Promise { const options = { @@ -280,7 +278,6 @@ async function run( timeoutMs: overrides.timeoutMs, headers: overrides.headers, onCompactionRejected: overrides.onCompactionRejected, - onProviderAccepted: overrides.onProviderAccepted, }; if (overrides.observations) { responsesPromptObserver.set(options, (observation) => @@ -344,19 +341,6 @@ describe("native OpenAI Responses WebSocket client integration", () => { configureAiTransportHost(initialHost); }); - it("reports WebSocket acceptance without fabricated HTTP metadata", async () => { - transportState.responseBatches.push([message(completedEvent("resp_accepted", "ok"))]); - const onProviderAccepted = vi.fn(); - - const result = await run( - { messages: [userMessage("hello", 1)], tools: [] }, - { onProviderAccepted }, - ); - - expect(result.stopReason).toBe("stop"); - expect(onProviderAccepted).toHaveBeenCalledWith({ kind: "provider_stream_opened" }, model); - }); - it("continues past provider-only output metadata with one socket and only new input", async () => { transportState.responseBatches.push( [message(completedEvent("resp_1", "first answer"))], diff --git a/packages/ai/src/transports/openai-transport-shared.ts b/packages/ai/src/transports/openai-transport-shared.ts index ca7d9311d659..4be60f9ff045 100644 --- a/packages/ai/src/transports/openai-transport-shared.ts +++ b/packages/ai/src/transports/openai-transport-shared.ts @@ -15,7 +15,7 @@ import type { BaseOpenAIStreamOptions } from "../provider-options.js"; /** Shared options, usage shape, cache identity, ordering, and stream scheduling for OpenAI APIs. */ import { clampOpenAIPromptCacheKey } from "../providers/openai-prompt-cache.js"; import { headersToRecord } from "../utils/headers.js"; -import { notifyProviderHttpMetadata, transportAbortError } from "./transport-stream-shared.js"; +import { transportAbortError } from "./transport-stream-shared.js"; export { sortPromptCacheToolsByName as sortTransportToolsByName } from "../utils/prompt-cache-stability.js"; @@ -230,22 +230,6 @@ export function createOpenAIResponseHook( : undefined; } -export function createOpenAIProviderAcceptanceHook( - options: Pick | undefined, - response: Response, - model: Model, -): (() => void | Promise) | undefined { - if (!options?.onProviderAccepted) { - return createOpenAIResponseHook(options?.onResponse, response, model); - } - return () => - notifyProviderHttpMetadata({ - options, - response: { status: response.status, headers: headersToRecord(response.headers) }, - model, - }); -} - type ModelStreamCooperativeScheduler = { afterEvent: () => Promise; }; diff --git a/packages/ai/src/transports/transport-stream-shared.acceptance.test.ts b/packages/ai/src/transports/transport-stream-shared.acceptance.test.ts deleted file mode 100644 index 6644450953a4..000000000000 --- a/packages/ai/src/transports/transport-stream-shared.acceptance.test.ts +++ /dev/null @@ -1,144 +0,0 @@ -import type { Model, StreamOptions } from "@openclaw/llm-core"; -import { describe, expect, it, vi } from "vitest"; -import { - notifyProviderHttpResponse, - notifyProviderStreamOpened, -} from "./transport-stream-shared.js"; - -const model = { id: "acceptance-test", provider: "test" } as Model; - -describe("notifyProviderHttpResponse", () => { - it.each(["onProviderAccepted", "onResponse"] as const)( - "cancels an unread response when %s fails", - async (hookName) => { - const hookError = new Error(`${hookName} failed`); - const cancel = vi.fn(); - const response = new Response( - new ReadableStream({ - cancel, - }), - { status: 200 }, - ); - const options: StreamOptions = { - [hookName]: vi.fn(() => Promise.reject(hookError)), - }; - - await expect(notifyProviderHttpResponse({ options, response, model })).rejects.toBe( - hookError, - ); - - expect(cancel).toHaveBeenCalledWith(hookError); - }, - ); - - it("does not wait for unread response cancellation after a callback fails", async () => { - const hookError = new Error("acceptance failed"); - let markCancelStarted!: () => void; - const cancelStarted = new Promise((resolve) => { - markCancelStarted = resolve; - }); - const response = new Response( - new ReadableStream({ - cancel() { - markCancelStarted(); - return new Promise(() => {}); - }, - }), - { status: 200 }, - ); - const notification = notifyProviderHttpResponse({ - options: { onProviderAccepted: () => Promise.reject(hookError) }, - response, - model, - }); - - await cancelStarted; - await expect(notification).rejects.toBe(hookError); - }); - - it("reports a rejected HTTP response without marking it accepted", async () => { - const onProviderAccepted = vi.fn(); - const onResponse = vi.fn(); - const options: StreamOptions = { onProviderAccepted, onResponse }; - const response = new Response("rejected", { status: 429 }); - - await notifyProviderHttpResponse({ options, response, model }); - - expect(onProviderAccepted).not.toHaveBeenCalled(); - expect(onResponse).toHaveBeenCalledWith(expect.objectContaining({ status: 429 }), model); - }); - - it("does not resume response handling when a callback aborts its signal", async () => { - const controller = new AbortController(); - const abortReason = Object.assign(new Error("operator canceled"), { - code: "OPERATOR_CANCELLED", - }); - const cancel = vi.fn(); - const response = new Response(new ReadableStream({ cancel }), { status: 200 }); - - await expect( - notifyProviderHttpResponse({ - options: { - signal: controller.signal, - onProviderAccepted: () => controller.abort(abortReason), - }, - response, - model, - }), - ).rejects.toBe(abortReason); - expect(cancel).toHaveBeenCalledWith(abortReason); - }); - - it("uses the option signal to abort a pending HTTP acceptance callback", async () => { - const controller = new AbortController(); - const abortReason = Object.assign(new Error("operator canceled"), { - code: "OPERATOR_CANCELLED", - }); - const cancel = vi.fn(); - const response = new Response(new ReadableStream({ cancel }), { status: 200 }); - let markHookStarted!: () => void; - const hookStarted = new Promise((resolve) => { - markHookStarted = resolve; - }); - const options: StreamOptions = { - signal: controller.signal, - onProviderAccepted: () => { - markHookStarted(); - return new Promise(() => {}); - }, - }; - - const notification = notifyProviderHttpResponse({ options, response, model }); - await hookStarted; - controller.abort(abortReason); - - await expect(notification).rejects.toBe(abortReason); - expect(cancel).toHaveBeenCalledWith(abortReason); - }); -}); - -describe("notifyProviderStreamOpened", () => { - it("uses the option signal to abort a pending SDK acceptance callback", async () => { - const controller = new AbortController(); - const abortReason = Object.assign(new Error("operator canceled"), { - code: "OPERATOR_CANCELLED", - }); - let markHookStarted!: () => void; - const hookStarted = new Promise((resolve) => { - markHookStarted = resolve; - }); - const options: StreamOptions = { - signal: controller.signal, - onProviderAccepted: () => { - markHookStarted(); - return new Promise(() => {}); - }, - }; - - const notification = notifyProviderStreamOpened({ options, model }); - await hookStarted; - controller.abort(abortReason); - - await expect(notification).rejects.toBe(abortReason); - }); -}); diff --git a/packages/ai/src/transports/transport-stream-shared.ts b/packages/ai/src/transports/transport-stream-shared.ts index ed05517aad22..f14ec518bc9a 100644 --- a/packages/ai/src/transports/transport-stream-shared.ts +++ b/packages/ai/src/transports/transport-stream-shared.ts @@ -3,16 +3,9 @@ * * Sanitizes provider payloads, merges metadata, and formats streamed assistant events. */ -import type { - AssistantMessage, - Model, - ProviderResponse, - StreamOptions, - Usage, -} from "@openclaw/llm-core"; +import type { AssistantMessage, Usage } from "@openclaw/llm-core"; import { asNonArrayRecord, asOptionalRecord } from "@openclaw/normalization-core/record-coerce"; import { createAssistantMessageEventStream } from "../utils/event-stream.js"; -import { headersToRecord } from "../utils/headers.js"; import { projectProviderError, type ProviderErrorProjection } from "../utils/provider-error.js"; import { sanitizeSurrogates } from "../utils/sanitize-unicode.js"; @@ -130,115 +123,6 @@ export function transportAbortError(signal?: AbortSignal): Error { : new Error("Request was aborted"); } -type ProviderAcceptanceOptions = Pick< - StreamOptions, - "onProviderAccepted" | "onResponse" | "signal" ->; - -async function awaitProviderLifecycleCallback( - callback: (() => void | Promise) | undefined, - signal?: AbortSignal, -): Promise { - if (signal?.aborted) { - throw transportAbortError(signal); - } - if (!callback) { - return; - } - const callbackPromise = Promise.resolve().then(callback); - if (!signal) { - await callbackPromise; - return; - } - let onAbort: (() => void) | undefined; - try { - await Promise.race([ - callbackPromise, - new Promise((_resolve, reject) => { - onAbort = () => reject(transportAbortError(signal)); - signal.addEventListener("abort", onAbort, { once: true }); - }), - ]); - } finally { - if (onAbort) { - signal.removeEventListener("abort", onAbort); - } - } - if (signal.aborted) { - throw transportAbortError(signal); - } -} - -/** Report observed HTTP metadata; rejected responses use only onResponse. */ -export async function notifyProviderHttpMetadata(params: { - options?: ProviderAcceptanceOptions; - response: ProviderResponse; - model: Model; - signal?: AbortSignal; -}): Promise { - if (!params.options?.onProviderAccepted && !params.options?.onResponse) { - return; - } - const { status, headers } = params.response; - const signal = params.signal ?? params.options?.signal; - const accepted = status >= 200 && status < 300; - await awaitProviderLifecycleCallback( - accepted && params.options.onProviderAccepted - ? () => - params.options?.onProviderAccepted?.( - { kind: "http_response", status, headers }, - params.model, - ) - : undefined, - signal, - ); - await awaitProviderLifecycleCallback( - params.options.onResponse - ? () => params.options?.onResponse?.({ status, headers }, params.model) - : undefined, - signal, - ); -} - -/** Report a real HTTP response before body consumption. */ -export async function notifyProviderHttpResponse(params: { - options?: ProviderAcceptanceOptions; - response: Response; - model: Model; - signal?: AbortSignal; -}): Promise { - try { - await notifyProviderHttpMetadata({ - options: params.options, - response: { - status: params.response.status, - headers: headersToRecord(params.response.headers), - }, - model: params.model, - signal: params.signal, - }); - } catch (error) { - // Cancellation is best-effort cleanup; a stalled body must not retain the request owner - // or delay the callback failure that made the body unreadable. - void params.response.body?.cancel(error).catch(() => undefined); - throw error; - } -} - -/** Report an accepted SDK stream when the SDK does not expose HTTP metadata. */ -export async function notifyProviderStreamOpened(params: { - options?: Pick; - model: Model; - signal?: AbortSignal; -}): Promise { - await awaitProviderLifecycleCallback( - params.options?.onProviderAccepted - ? () => params.options?.onProviderAccepted?.({ kind: "provider_stream_opened" }, params.model) - : undefined, - params.signal ?? params.options?.signal, - ); -} - /** Run a provider-response hook before start/body consumption inside the first-event deadline. */ export function withProviderResponseHook(params: { stream?: AsyncIterable; @@ -249,11 +133,27 @@ export function withProviderResponseHook(params: { }): AsyncIterable { return { async *[Symbol.asyncIterator]() { + let onAbort: (() => void) | undefined; try { - await awaitProviderLifecycleCallback(params.hook, params.signal); + if (params.signal.aborted) { + throw transportAbortError(params.signal); + } + if (params.hook) { + await Promise.race([ + Promise.resolve().then(params.hook), + new Promise((_resolve, reject) => { + onAbort = () => reject(transportAbortError(params.signal)); + params.signal.addEventListener("abort", onAbort, { once: true }); + }), + ]); + } } catch (error) { params.abort(error instanceof Error ? error : new Error(String(error))); throw error; + } finally { + if (onAbort) { + params.signal.removeEventListener("abort", onAbort); + } } if (params.signal.aborted) { throw transportAbortError(params.signal); diff --git a/packages/llm-core/src/types.ts b/packages/llm-core/src/types.ts index ea270816aa1c..e2b23b95cf67 100644 --- a/packages/llm-core/src/types.ts +++ b/packages/llm-core/src/types.ts @@ -64,15 +64,6 @@ export interface ProviderResponse { headers: Record; } -/** Evidence that a text provider accepted a request. */ -export type ProviderAcceptance = - | { - kind: "http_response"; - status: number; - headers: Record; - } - | { kind: "provider_stream_opened" }; - /** Request options shared by text streaming providers. */ export interface StreamOptions { temperature?: number; @@ -122,13 +113,8 @@ export interface StreamOptions { */ onPayload?: (payload: unknown, model: Model) => MaybePromise; /** - * Optional callback invoked after the provider accepts the request and before - * its body stream is consumed. HTTP metadata is included only when the transport sees it. - */ - onProviderAccepted?: (acceptance: ProviderAcceptance, model: Model) => void | Promise; - /** - * Optional compatibility callback invoked after a transport receives a real - * HTTP response and before its body stream is consumed. + * Optional callback invoked after an HTTP response is received and before + * its body stream is consumed. */ onResponse?: (response: ProviderResponse, model: Model) => void | Promise; /** diff --git a/packages/plugin-sdk/package.json b/packages/plugin-sdk/package.json index ae971747b8ac..73067a3855e3 100644 --- a/packages/plugin-sdk/package.json +++ b/packages/plugin-sdk/package.json @@ -136,10 +136,6 @@ "types": "./dist/src/plugin-sdk/provider-http.d.ts", "default": "./src/provider-http.ts" }, - "./provider-lifecycle": { - "types": "./dist/src/plugin-sdk/provider-lifecycle.d.ts", - "default": "./src/provider-lifecycle.ts" - }, "./provider-model-shared": { "types": "./dist/src/plugin-sdk/provider-model-shared.d.ts", "default": "./src/provider-model-shared.ts" diff --git a/packages/plugin-sdk/src/provider-lifecycle.ts b/packages/plugin-sdk/src/provider-lifecycle.ts deleted file mode 100644 index d3ee2a734e17..000000000000 --- a/packages/plugin-sdk/src/provider-lifecycle.ts +++ /dev/null @@ -1,2 +0,0 @@ -/** Workspace facade for the public provider acceptance lifecycle API. */ -export * from "../../../src/plugin-sdk/provider-lifecycle.js"; diff --git a/scripts/lib/plugin-sdk-entrypoints.json b/scripts/lib/plugin-sdk-entrypoints.json index a235dfbb4438..8243b649a022 100644 --- a/scripts/lib/plugin-sdk-entrypoints.json +++ b/scripts/lib/plugin-sdk-entrypoints.json @@ -287,7 +287,6 @@ "provider-entry", "provider-env-vars", "provider-http", - "provider-lifecycle", "provider-binary-stream", "provider-model-types", "provider-model-shared", diff --git a/scripts/plugin-sdk-surface-report.mts b/scripts/plugin-sdk-surface-report.mts index 7e825a186ceb..7497f0ed981f 100644 --- a/scripts/plugin-sdk-surface-report.mts +++ b/scripts/plugin-sdk-surface-report.mts @@ -195,8 +195,7 @@ export function readPluginSdkSurfaceBudgets(env: NodeJS.ProcessEnv = process.env // +1: canonical Computer Use wire contract and node-host provider seam. // -1: retire the deprecated messaging-targets subpath. // +2: bounded provider streams and read-only SecretRef resolution. - // +1: supported provider request-acceptance lifecycle seam. - 147, + 146, env, ), publicExports: readPluginSdkSurfaceBudgetEnv( @@ -301,8 +300,7 @@ export function readPluginSdkSurfaceBudgets(env: NodeJS.ProcessEnv = process.env // +1: account-scoped model catalog discovery for native agent harnesses. // +2: shared delegation policy (mode resolver + section builder) so harness // runtimes render the same guidance instead of diverging prompt copies. - // +5: provider acceptance receipt/response types and three lifecycle helpers. - 4339, + 4334, env, ), publicFunctionExports: readPluginSdkSurfaceBudgetEnv( @@ -387,8 +385,7 @@ export function readPluginSdkSurfaceBudgets(env: NodeJS.ProcessEnv = process.env // +1: canonical sensitive-URL redactor so plugin CLI errors never print URL userinfo. // +2: shared delegation policy (mode resolver + section builder) so harness // runtimes render the same guidance instead of diverging prompt copies. - // +3: provider acceptance lifecycle helpers for HTTP and metadata-free streams. - 2580, + 2577, env, ), publicDeprecatedExports: readPluginSdkSurfaceBudgetEnv( diff --git a/src/agents/embedded-agent-runner/run/attempt.model-diagnostic-lifecycle.test.ts b/src/agents/embedded-agent-runner/run/attempt.model-diagnostic-lifecycle.test.ts index 321eae14306b..e2cbb8005ef4 100644 --- a/src/agents/embedded-agent-runner/run/attempt.model-diagnostic-lifecycle.test.ts +++ b/src/agents/embedded-agent-runner/run/attempt.model-diagnostic-lifecycle.test.ts @@ -172,7 +172,7 @@ describe("wrapStreamFnWithDiagnosticModelCallEvents lifecycle", () => { expect(events[0]?.status).toBeUndefined(); }); - it("records legacy response status without inferring provider acceptance", async () => { + it("records provider response status and preserves the original response callback", async () => { const originalOnResponse = vi.fn(async () => undefined); const wrapped = wrapStreamFnWithDiagnosticModelCallEvents( (( @@ -212,56 +212,9 @@ describe("wrapStreamFnWithDiagnosticModelCallEvents lifecycle", () => { type: "provider.request", ok: true, status: 200, - attributes: { - providerAccepted: false, - }, }); }); - it("records provider acceptance when an SDK hides HTTP metadata", async () => { - const originalOnProviderAccepted = vi.fn(async () => undefined); - const wrapped = wrapStreamFnWithDiagnosticModelCallEvents( - (( - model: Parameters[0], - _context: Parameters[1], - options: Parameters[2], - ) => { - return options?.onProviderAccepted?.({ kind: "provider_stream_opened" }, model); - }) as unknown as StreamFn, - { - runId: "run-timeline-accepted", - provider: "google", - model: "gemini-2.5-pro", - api: "google-generative-ai", - trace: createDiagnosticTraceContext(), - nextCallId: () => "call-timeline-accepted", - }, - ); - - const events = await collectProviderTimelineEvents(async () => { - await wrapped( - { id: "gemini-2.5-pro" } as never, - {} as never, - { onProviderAccepted: originalOnProviderAccepted } as never, - ); - }); - - expect(originalOnProviderAccepted).toHaveBeenCalledWith( - { kind: "provider_stream_opened" }, - { id: "gemini-2.5-pro" }, - ); - expect(events).toHaveLength(1); - expect(events[0]).toMatchObject({ - type: "provider.request", - ok: true, - attributes: { - providerAccepted: true, - providerAcceptanceKind: "provider_stream_opened", - }, - }); - expect(events[0]?.status).toBeUndefined(); - }); - it("writes Unicode-safe bounded attributes to the provider timeline JSONL", async () => { const modelPrefix = "m".repeat(255); const exactBoundary = "b".repeat(256); diff --git a/src/agents/embedded-agent-runner/run/attempt.model-diagnostic-lifecycle.ts b/src/agents/embedded-agent-runner/run/attempt.model-diagnostic-lifecycle.ts index 6548b980d377..e3dcb3b35a5a 100644 --- a/src/agents/embedded-agent-runner/run/attempt.model-diagnostic-lifecycle.ts +++ b/src/agents/embedded-agent-runner/run/attempt.model-diagnostic-lifecycle.ts @@ -1,4 +1,3 @@ -import type { ProviderAcceptance } from "@openclaw/llm-core"; import { isPromiseLike } from "@openclaw/normalization-core/promise-like"; import { truncateUtf16Safe } from "@openclaw/normalization-core/utf16-slice"; import { fireAndForgetBoundedHook } from "../../../hooks/fire-and-forget.js"; @@ -86,7 +85,6 @@ export type ModelCallUsage = NonNullable< >; export type ModelCallObservationState = { requestPayloadBytes?: number; - providerAcceptanceKind?: ProviderAcceptance["kind"]; responseStatus?: number; responseStreamBytes: number; timeToFirstByteMs?: number; @@ -156,7 +154,6 @@ function emitProviderRequestTimelineEvent( durationMs: number, ok: boolean, responseStatus: number | undefined, - providerAcceptanceKind: ModelCallObservationState["providerAcceptanceKind"], ): void { const provider = boundedTimelineAttribute(eventBase.provider); const model = boundedTimelineAttribute(eventBase.model); @@ -177,8 +174,6 @@ function emitProviderRequestTimelineEvent( ...(model ? { model } : {}), ...(api ? { api } : {}), ...(transport ? { transport } : {}), - providerAccepted: providerAcceptanceKind !== undefined, - ...(providerAcceptanceKind ? { providerAcceptanceKind } : {}), }, }); } @@ -296,7 +291,6 @@ function emitModelCallCompleted( durationMs, true, observer.state.responseStatus, - observer.state.providerAcceptanceKind, ); emitCoreModelRequestEndedDiagnosticEvent( { @@ -335,14 +329,7 @@ function emitModelCallError( const errorStatus = diagnosticHttpStatusCode(err); const responseStatus = observer.state.responseStatus ?? (errorStatus === undefined ? undefined : Number(errorStatus)); - emitProviderRequestTimelineEvent( - eventBase, - startedAt, - durationMs, - false, - responseStatus, - observer.state.providerAcceptanceKind, - ); + emitProviderRequestTimelineEvent(eventBase, startedAt, durationMs, false, responseStatus); emitCoreModelRequestEndedDiagnosticEvent( { type: "model.call.error", @@ -373,7 +360,6 @@ function withDiagnosticRequestContext( ): ModelCallStreamOptions { const traceparent = formatPropagatedDiagnosticTraceparent(trace); const originalOnPayload = options?.onPayload; - const originalOnProviderAccepted = options?.onProviderAccepted; const originalOnResponse = options?.onResponse; const onPayload: NonNullable["onPayload"] = (payload, model) => { if (!originalOnPayload) { @@ -390,16 +376,6 @@ function withDiagnosticRequestContext( observer.assignRequestPayloadBytes(result ?? payload); return result; }; - const onProviderAccepted: NonNullable["onProviderAccepted"] = ( - acceptance, - model, - ) => { - observer.state.providerAcceptanceKind = acceptance.kind; - if (acceptance.kind === "http_response") { - observer.state.responseStatus = acceptance.status; - } - return originalOnProviderAccepted?.(acceptance, model); - }; const onResponse: NonNullable["onResponse"] = (response, model) => { // Retrying providers can expose several responses; the terminal request status // is the latest response observed before the model call completes or fails. @@ -422,7 +398,6 @@ function withDiagnosticRequestContext( requestId: callId, ...((options?.headers || traceparent) && { headers }), onPayload, - onProviderAccepted, onResponse, }; } diff --git a/src/plugin-sdk/llm.ts b/src/plugin-sdk/llm.ts index bc6b6e5e6f35..e034a2846656 100644 --- a/src/plugin-sdk/llm.ts +++ b/src/plugin-sdk/llm.ts @@ -37,7 +37,6 @@ export type { Message, Model, ModelThinkingLevel, - ProviderAcceptance, ProviderResponse, ProviderStreamOptions, SimpleStreamOptions, diff --git a/src/plugin-sdk/provider-lifecycle.ts b/src/plugin-sdk/provider-lifecycle.ts deleted file mode 100644 index b320e7a59cc2..000000000000 --- a/src/plugin-sdk/provider-lifecycle.ts +++ /dev/null @@ -1,7 +0,0 @@ -/** Public provider request-acceptance lifecycle types and helpers. */ -export type { ProviderAcceptance, ProviderResponse } from "@openclaw/llm-core"; -export { - notifyProviderHttpMetadata, - notifyProviderHttpResponse, - notifyProviderStreamOpened, -} from "@openclaw/ai/transports";