Merge PR #8717: fix(pi-ai): honor onPayload replacement payloads in openai-completions, bedrock and cursor (@ranxianglei)

packages/ai/CHANGELOG.md

@@ -6,6 +6,7 @@
 
 - Fixed thinking effort selections being ignored for local Qwen 3.8+ models on llama.cpp and vLLM: the Qwen chat-completions dialects only toggled `enable_thinking`, so the chat template always reasoned at its `xhigh` default no matter which level was selected. The encoder now routes the requested effort onto the template's `reasoning_effort` kwarg (`chat_template_kwargs` for both Qwen dialects, plus the top-level field newer llama.cpp builds map natively).
 
+- Fixed OpenAI Completions, Amazon Bedrock, and Cursor providers ignoring `onPayload` replacement payloads. The hook now transforms the actual request body sent upstream on these providers, matching the Anthropic/Gemini/OpenAI Responses replacement contract. `devin-agent` still does not fire the hook (its payload is a protobuf object).
 ## [17.3.7] - 2026-08-17
 
 ### Changed

packages/ai/README.md

@@ -634,7 +634,7 @@ All providers accept the base `StreamOptions` (in addition to provider-specific
 - `headers`: Extra request headers merged on top of model-defined headers
 - `sessionId`: Provider-specific session identifier (prompt caching/routing)
 - `signal`: Abort in-flight requests
-- `onPayload`: Callback invoked with the provider request payload just before sending
+- `onPayload`: Callback invoked with the provider request payload just before sending. Return a replacement payload object (sync or async) to send it instead of the original; return `undefined` to keep the original. The replacement is applied by every provider that fires the hook — all of them except `devin-agent`, whose payload is a protobuf object and does not fire the hook yet.
 
 Example:
 

packages/ai/src/providers/amazon-bedrock.ts

@@ -342,7 +342,7 @@ export const streamBedrock: StreamFunction<"bedrock-converse-stream"> = (
 				if (tc.any || tc.tool) additionalModelRequestFields = undefined;
 			}
 
-			const commandInput: ConverseStreamRequest = {
+			let commandInput: ConverseStreamRequest = {
 				messages: convertedMessages,
 				system: buildSystemPrompt(context.systemPrompt, promptCachePolicy),
 				inferenceConfig: {
@@ -353,7 +353,8 @@ export const streamBedrock: StreamFunction<"bedrock-converse-stream"> = (
 				toolConfig,
 				additionalModelRequestFields,
 			};
-			options?.onPayload?.(commandInput, model);
+			const replacementInput = await options?.onPayload?.(commandInput, model);
+			if (replacementInput !== undefined) commandInput = replacementInput as ConverseStreamRequest;
 
 			const host = `bedrock-runtime.${region}.amazonaws.com`;
 			const url = `https://${host}/model/${encodeURIComponent(model.id)}/converse-stream`;

packages/ai/src/providers/cursor.ts

@@ -615,7 +615,7 @@ export const streamCursor: StreamFunction<"cursor-agent"> = (
 			const blobStore = conversationBlobStores.get(conversationId) ?? new Map<string, Uint8Array>();
 			conversationBlobStores.set(conversationId, blobStore);
 			const cachedState = conversationStateCache.get(conversationId);
-			const { requestBytes, conversationState } = buildGrpcRequest(model, context, options, {
+			const { requestBytes, conversationState } = await buildGrpcRequest(model, context, options, {
 				conversationId,
 				blobStore,
 				conversationState: cachedState,
@@ -4642,7 +4642,7 @@ function extractImages(content: (TextContent | ImageContent)[]) {
 		);
 }
 
-function buildGrpcRequest(
+export async function buildGrpcRequest(
 	model: Model<"cursor-agent">,
 	context: Context,
 	options: CursorOptions | undefined,
@@ -4651,11 +4651,11 @@ function buildGrpcRequest(
 		blobStore: Map<string, Uint8Array>;
 		conversationState?: ConversationStateStructure;
 	},
-): {
+): Promise<{
 	requestBytes: Uint8Array;
 	blobStore: Map<string, Uint8Array>;
 	conversationState: ConversationStateStructure;
-} {
+}> {
 	const blobStore = state.blobStore;
 
 	const systemPromptIds = buildCursorSystemPromptJsons(context.systemPrompt).map(json =>
@@ -4761,7 +4761,7 @@ function buildGrpcRequest(
 		maxMode: cursorMaxMode,
 	});
 
-	const runRequest = create(AgentRunRequestSchema, {
+	let runRequest = create(AgentRunRequestSchema, {
 		conversationState,
 		action,
 		modelDetails,
@@ -4769,14 +4769,18 @@ function buildGrpcRequest(
 		conversationId: state.conversationId,
 	});
 
-	options?.onPayload?.(runRequest, model);
-
-	// Tools are sent later via requestContext (exec handshake)
-
+	// Apply customSystemPrompt BEFORE the hook so the onPayload replacement is the
+	// final word on the wire body — same contract as anthropic, where the hook runs
+	// right before serialization. An extension may inspect or drop it via the
+	// replacement it returns.
 	if (options?.customSystemPrompt) {
 		runRequest.customSystemPrompt = options.customSystemPrompt;
 	}
 
+	// Tools are sent later via requestContext (exec handshake)
+	const replacementRequest = await options?.onPayload?.(runRequest, model);
+	if (replacementRequest !== undefined) runRequest = replacementRequest as typeof runRequest;
+
 	const clientMessage = create(AgentClientMessageSchema, {
 		message: { case: "runRequest", value: runRequest },
 	});

packages/ai/src/providers/openai-completions.ts

@@ -663,7 +663,7 @@ const streamOpenAICompletionsOnce = (
 				: `${trimmedBaseUrl}/chat/completions`;
 			const createCompletionsStream = async (toolStrictModeOverride?: ToolStrictModeOverride) => {
 				const effectiveToolStrictModeOverride = disableStrictTools ? "none" : toolStrictModeOverride;
-				const { params, strictToolsApplied } = buildParams(
+				let { params, strictToolsApplied } = buildParams(
 					model,
 					context,
 					options,
@@ -682,8 +682,9 @@ const streamOpenAICompletionsOnce = (
 					applyOpenAIReasoningEffortFallback(params, requestReasoningEffortFallback);
 				}
 				activeReasoningEffortFallbackKey = reasoningEffortFallbackKey;
+				const replacedParams = await options?.onPayload?.(params, model);
+				if (replacedParams !== undefined) params = replacedParams as typeof params;
 				activeRequestParams = params;
-				options?.onPayload?.(params, model);
 				rawRequestDump = {
 					provider: model.provider,
 					api: output.api,

packages/ai/test/bedrock-on-payload.test.ts

@@ -0,0 +1,74 @@
+// Regression: amazon-bedrock ignored the onPayload replacement return value
+// (fire-and-forget), so the hook could never change the body actually sent
+// upstream. The replacement contract matches anthropic / openai-responses /
+// google: await the hook and use its non-undefined return as the request body.
+import { describe, expect, it, vi } from "bun:test";
+import { streamBedrock } from "@oh-my-pi/pi-ai/providers/amazon-bedrock";
+import type { Context, Model } from "@oh-my-pi/pi-ai/types";
+import { buildModel } from "@oh-my-pi/pi-catalog/build";
+
+function model(): Model<"bedrock-converse-stream"> {
+	return buildModel({
+		id: "us.anthropic.claude-haiku-4-5-20251001-v1:0",
+		name: "haiku",
+		api: "bedrock-converse-stream",
+		provider: "amazon-bedrock",
+		baseUrl: "https://bedrock-runtime.us-east-1.amazonaws.com",
+		reasoning: false,
+		input: ["text"],
+		cost: { input: 5, output: 25, cacheRead: 0.5, cacheWrite: 6.25 },
+		contextWindow: 1_000_000,
+		maxTokens: 128_000,
+	});
+}
+
+const context: Context = {
+	messages: [{ role: "user", content: "hi", timestamp: 0 }],
+};
+
+// Capture the serialized body the provider sends. The response is an empty
+// event stream: the fetch (and thus the body capture) happens before any
+// response parsing, and the stream's outcome is irrelevant to the assertion.
+async function captureSentBody(onPayload: (payload: unknown) => unknown | Promise<unknown>): Promise<Record<string, any>> {
+	const { promise, resolve } = Promise.withResolvers<Record<string, any>>();
+	const fetchMock = vi.fn(async (_input: string | URL | Request, init?: RequestInit) => {
+		const body = init?.body;
+		const text = body instanceof Uint8Array ? new TextDecoder().decode(body) : String(body);
+		resolve(JSON.parse(text));
+		return new Response(
+			new ReadableStream<Uint8Array>({ start(controller) { controller.close(); } }),
+			{ status: 200, headers: { "content-type": "application/vnd.amazon.eventstream" } },
+		);
+	}) as unknown as typeof fetch;
+
+	const stream = streamBedrock(model(), context, { bearerToken: "test-token", fetch: fetchMock, onPayload });
+	void (async () => {
+		try {
+			for await (const _ of stream) {
+				// ignore events
+			}
+		} catch {
+			// empty event stream: stream errors are expected and irrelevant
+		}
+	})();
+
+	return promise;
+}
+
+describe("bedrock onPayload replacement", () => {
+	it("sends an async onPayload replacement body", async () => {
+		const body = await captureSentBody(async payload => ({
+			...(payload as Record<string, unknown>),
+			messages: [{ role: "user", content: [{ text: "replacement" }] }],
+		}));
+
+		expect(body.messages).toEqual([{ role: "user", content: [{ text: "replacement" }] }]);
+		expect(JSON.stringify(body.messages)).not.toContain("hi");
+	}, 10_000);
+
+	it("keeps the original body when onPayload returns undefined", async () => {
+		const body = await captureSentBody(async () => undefined);
+
+		expect(body.messages[0].content[0].text).toBe("hi");
+	}, 10_000);
+});

packages/ai/test/cursor-on-payload.test.ts

@@ -0,0 +1,117 @@
+// Regression: cursor ignored the onPayload replacement return value
+// (fire-and-forget), so the hook could never change the request actually sent
+// upstream. The replacement contract matches anthropic / openai-responses /
+// google: await the hook and use its non-undefined return as the request.
+// buildGrpcRequest is exercised directly (the transport is HTTP/2), and the
+// serialized run request is decoded back from the wire bytes.
+import { describe, expect, it } from "bun:test";
+import { fromBinary } from "@bufbuild/protobuf";
+import { buildGrpcRequest } from "@oh-my-pi/pi-ai/providers/cursor";
+import { AgentClientMessageSchema } from "@oh-my-pi/pi-catalog/discovery/cursor-gen/agent_pb";
+import type { Context, Model } from "@oh-my-pi/pi-ai/types";
+import { buildModel } from "@oh-my-pi/pi-catalog/build";
+
+const model: Model<"cursor-agent"> = buildModel({
+	id: "cursor-composer-2.5",
+	name: "Cursor Composer 2.5",
+	api: "cursor-agent",
+	provider: "cursor",
+	baseUrl: "https://api2.cursor.sh",
+	reasoning: false,
+	input: ["text"],
+	cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 },
+	contextWindow: 200_000,
+	maxTokens: 32_000,
+});
+
+const context: Context = {
+	messages: [{ role: "user", content: "Say hello", timestamp: 0 }],
+};
+
+function decodeRunRequest(requestBytes: Uint8Array): { case: string; value: Record<string, any> } {
+	const decoded = fromBinary(AgentClientMessageSchema, requestBytes);
+	return decoded.message as unknown as { case: string; value: Record<string, any> };
+}
+
+describe("cursor onPayload replacement", () => {
+	it("sends an async onPayload replacement body", async () => {
+		const { requestBytes } = await buildGrpcRequest(
+			model,
+			context,
+			{
+				onPayload: async payload => ({
+					...(payload as Record<string, unknown>),
+					customSystemPrompt: "replacement",
+				}),
+			},
+			{ conversationId: "conv-1", blobStore: new Map() },
+		);
+
+		const message = decodeRunRequest(requestBytes);
+		expect(message.case).toBe("runRequest");
+		expect(message.value.customSystemPrompt).toBe("replacement");
+	});
+
+	it("keeps the original body when onPayload returns undefined", async () => {
+		const { requestBytes } = await buildGrpcRequest(
+			model,
+			context,
+			{ onPayload: async () => undefined },
+			{ conversationId: "conv-1", blobStore: new Map() },
+		);
+
+		const message = decodeRunRequest(requestBytes);
+		expect(message.case).toBe("runRequest");
+		expect(message.value.customSystemPrompt).toBeUndefined();
+	});
+
+	it("applies customSystemPrompt when onPayload returns undefined", async () => {
+		const { requestBytes } = await buildGrpcRequest(
+			model,
+			context,
+			{ customSystemPrompt: "from-options", onPayload: async () => undefined },
+			{ conversationId: "conv-1", blobStore: new Map() },
+		);
+
+		const message = decodeRunRequest(requestBytes);
+		expect(message.value.customSystemPrompt).toBe("from-options");
+	});
+
+	it("lets the onPayload replacement drop customSystemPrompt (replacement is final)", async () => {
+		const { requestBytes } = await buildGrpcRequest(
+			model,
+			context,
+			{
+				customSystemPrompt: "from-options",
+				onPayload: async payload => {
+					const { customSystemPrompt: _dropped, ...rest } = payload as Record<string, unknown>;
+					return rest;
+				},
+			},
+			{ conversationId: "conv-1", blobStore: new Map() },
+		);
+
+		const message = decodeRunRequest(requestBytes);
+		// The hook saw customSystemPrompt already applied (set before the hook) and
+		// returned a replacement that does not carry it — that replacement is final.
+		expect(message.value.customSystemPrompt).toBeUndefined();
+	});
+
+	it("lets the onPayload replacement override customSystemPrompt", async () => {
+		const { requestBytes } = await buildGrpcRequest(
+			model,
+			context,
+			{
+				customSystemPrompt: "from-options",
+				onPayload: async payload => ({
+					...(payload as Record<string, unknown>),
+					customSystemPrompt: "from-hook",
+				}),
+			},
+			{ conversationId: "conv-1", blobStore: new Map() },
+		);
+
+		const message = decodeRunRequest(requestBytes);
+		expect(message.value.customSystemPrompt).toBe("from-hook");
+	});
+});

packages/ai/test/openai-completions-on-payload.test.ts

@@ -0,0 +1,93 @@
+// Regression: openai-completions ignored the onPayload replacement return
+// value (fire-and-forget), so extensions hooking before_provider_request
+// could never transform the body actually sent upstream. The replacement
+// contract matches anthropic / openai-responses / google: await the hook,
+// and use its non-undefined return as the request body.
+import { describe, expect, it } from "bun:test";
+import { streamOpenAICompletions } from "@oh-my-pi/pi-ai/providers/openai-completions";
+import type { Context, FetchImpl, Model } from "@oh-my-pi/pi-ai/types";
+import { getBundledModel } from "@oh-my-pi/pi-catalog/models";
+
+const completionsModel = {
+	...(getBundledModel("openai", "gpt-4o-mini") as Model<"openai-completions">),
+	api: "openai-completions",
+} satisfies Model<"openai-completions">;
+
+function baseContext(): Context {
+	return {
+		messages: [{ role: "user", content: "Say hello", timestamp: Date.now() }],
+	};
+}
+
+function createSseFetch(capture?: (body: unknown) => void): FetchImpl {
+	async function mockFetch(_input: string | URL | Request, init?: RequestInit): Promise<Response> {
+		capture?.(typeof init?.body === "string" ? JSON.parse(init.body) : undefined);
+		const encoder = new TextEncoder();
+		const chunk = (extra: Record<string, unknown>) =>
+			`data: ${JSON.stringify({ id: "chatcmpl-payload", object: "chat.completion.chunk", created: 0, model: completionsModel.id, ...extra })}\n\n`;
+		const sse =
+			chunk({ choices: [{ index: 0, delta: { role: "assistant", content: "ok" } }] }) +
+			chunk({ choices: [{ index: 0, delta: {}, finish_reason: "stop" }] }) +
+			"data: [DONE]\n\n";
+		const stream = new ReadableStream<Uint8Array>({
+			start(controller) {
+				controller.enqueue(encoder.encode(sse));
+				controller.close();
+			},
+		});
+		return new Response(stream, {
+			status: 200,
+			headers: { "content-type": "text/event-stream" },
+		});
+	}
+	return mockFetch as typeof fetch;
+}
+
+type Body = Record<string, any>;
+
+describe("openai-completions onPayload replacement", () => {
+	it("sends an async onPayload replacement body", async () => {
+		let captured: Body | undefined;
+		const result = await streamOpenAICompletions(
+			completionsModel,
+			baseContext(),
+			{
+				apiKey: "test-key",
+				fetch: createSseFetch(body => (captured = body as Body)),
+				onPayload: async payload => ({
+					...(payload as Record<string, unknown>),
+					messages: [{ role: "user", content: "replacement" }],
+				}),
+			},
+		).result();
+
+		expect(result.stopReason).toBe("stop");
+		expect(captured?.messages).toEqual([{ role: "user", content: "replacement" }]);
+		expect(JSON.stringify(captured)).not.toContain("Say hello");
+	}, 10_000);
+
+	it("sends a synchronous onPayload replacement body", async () => {
+		let captured: Body | undefined;
+		await streamOpenAICompletions(completionsModel, baseContext(), {
+			apiKey: "test-key",
+			fetch: createSseFetch(body => (captured = body as Body)),
+			onPayload: payload => ({
+				...(payload as Record<string, unknown>),
+				messages: [{ role: "user", content: "sync-replacement" }],
+			}),
+		}).result();
+
+		expect(captured?.messages).toEqual([{ role: "user", content: "sync-replacement" }]);
+	}, 10_000);
+
+	it("keeps the original body when onPayload returns undefined", async () => {
+		let captured: Body | undefined;
+		await streamOpenAICompletions(completionsModel, baseContext(), {
+			apiKey: "test-key",
+			fetch: createSseFetch(body => (captured = body as Body)),
+			onPayload: async () => undefined,
+		}).result();
+
+		expect(JSON.stringify(captured?.messages)).toContain("Say hello");
+	}, 10_000);
+});