From eb16388ec6659a827eed390bcec86394e5e8f760 Mon Sep 17 00:00:00 2001 From: Aiden Cline Date: Mon, 3 Aug 2026 23:34:57 -0500 Subject: [PATCH] fix(core): preserve request bodies across retries --- packages/core/src/session/model-request.ts | 2 +- .../core/test/session-runner-recorded.test.ts | 169 ++++++++++++++---- packages/www/content/docs/build/plugins.mdx | 5 +- 3 files changed, 138 insertions(+), 38 deletions(-) diff --git a/packages/core/src/session/model-request.ts b/packages/core/src/session/model-request.ts index 9c66f79fc01..09b1815d396 100644 --- a/packages/core/src/session/model-request.ts +++ b/packages/core/src/session/model-request.ts @@ -231,7 +231,7 @@ export const layer = Layer.effect( if (input.body) sent = HttpClientRequest.bodyUint8Array( sent, - new Uint8Array(yield* Effect.promise(() => input.arrayBuffer())), + new Uint8Array(yield* Effect.promise(() => input.clone().arrayBuffer())), input.headers.get("content-type") ?? undefined, ) const response = yield* handler(sent) diff --git a/packages/core/test/session-runner-recorded.test.ts b/packages/core/test/session-runner-recorded.test.ts index ad735569ccb..df6a9b68733 100644 --- a/packages/core/test/session-runner-recorded.test.ts +++ b/packages/core/test/session-runner-recorded.test.ts @@ -42,6 +42,7 @@ import { SystemPromptPlugin } from "@opencode-ai/core/plugin/system-prompt" import { describe, expect } from "bun:test" import { eq } from "drizzle-orm" import { Effect, Layer, Stream } from "effect" +import { HttpClient, HttpClientResponse } from "effect/unstable/http" import path from "node:path" import { testEffect } from "./lib/effect" import { agentHost, catalogHost, host } from "./plugin/host" @@ -105,37 +106,39 @@ const promptCatalog = Layer.mock(Catalog.Service, { small: () => Effect.succeed(undefined), }, }) -const runnerLayer = AppNodeBuilder.build(SessionRunnerLLM.node, [ - [Snapshot.node, Snapshot.noopLayer], - [LayerNodePlatform.llmClient, client], - [SessionRunnerModel.node, models], - [InstructionBuiltIns.node, systemContext], - [InstructionDiscovery.node, instructionContext], - [Location.node, Location.boundNode({ directory: AbsolutePath.make("/project") })], - [SkillInstructions.node, skillInstructions], - [ReferenceInstructions.node, referenceInstructions], - [McpInstructions.node, mcpInstructions], - [Config.node, config], - [Permission.node, permission], - [PluginSupervisor.node, pluginSupervisor], -]) -const execution = Layer.effect( - SessionExecution.Service, - Effect.gen(function* () { - const sessionRunner = yield* SessionRunner.Service - const coordinator = yield* SessionRunCoordinator.make({ - drain: (sessionID, force) => sessionRunner.drain({ sessionID, force }), - }) - return SessionExecution.Service.of({ - active: coordinator.active, - resume: coordinator.run, - wake: coordinator.wake, - interrupt: coordinator.interrupt, - awaitIdle: coordinator.awaitIdle, - }) - }), -).pipe(Layer.provide(runnerLayer)) -const it = testEffect( +const runnerLayer = (llmClient: Layer.Layer) => + AppNodeBuilder.build(SessionRunnerLLM.node, [ + [Snapshot.node, Snapshot.noopLayer], + [LayerNodePlatform.llmClient, llmClient], + [SessionRunnerModel.node, models], + [InstructionBuiltIns.node, systemContext], + [InstructionDiscovery.node, instructionContext], + [Location.node, Location.boundNode({ directory: AbsolutePath.make("/project") })], + [SkillInstructions.node, skillInstructions], + [ReferenceInstructions.node, referenceInstructions], + [McpInstructions.node, mcpInstructions], + [Config.node, config], + [Permission.node, permission], + [PluginSupervisor.node, pluginSupervisor], + ]) +const execution = (llmClient: Layer.Layer) => + Layer.effect( + SessionExecution.Service, + Effect.gen(function* () { + const sessionRunner = yield* SessionRunner.Service + const coordinator = yield* SessionRunCoordinator.make({ + drain: (sessionID, force) => sessionRunner.drain({ sessionID, force }), + }) + return SessionExecution.Service.of({ + active: coordinator.active, + resume: coordinator.run, + wake: coordinator.wake, + interrupt: coordinator.interrupt, + awaitIdle: coordinator.awaitIdle, + }) + }), + ).pipe(Layer.provide(runnerLayer(llmClient))) +const testLayer = (llmClient: Layer.Layer) => AppNodeBuilder.build( LayerNode.group([ Database.node, @@ -157,7 +160,7 @@ const it = testEffect( Session.node, ]), [ - [LayerNodePlatform.llmClient, client], + [LayerNodePlatform.llmClient, llmClient], [Permission.node, permission], [Catalog.node, promptCatalog], [SessionRunnerModel.node, models], @@ -169,10 +172,10 @@ const it = testEffect( [Config.node, config], [Snapshot.node, Snapshot.noopLayer], [PluginSupervisor.node, pluginSupervisor], - [SessionExecution.node, execution], + [SessionExecution.node, execution(llmClient)], ], - ), -) + ) +const it = testEffect(testLayer(client)) const sessionID = Session.ID.make("ses_runner_recorded") describe("SessionRunnerLLM recorded", () => { @@ -253,3 +256,99 @@ describe("SessionRunnerLLM recorded", () => { }), ) }) + +describe("SessionModelRequest HTTP bridge", () => { + const bodies: Uint8Array[] = [] + const methods: string[] = [] + const response = [ + 'data: {"id":"chatcmpl_test","object":"chat.completion.chunk","created":0,"model":"gpt-4o-mini","choices":[{"index":0,"delta":{"role":"assistant","content":"Hello!"},"finish_reason":null}]}', + 'data: {"id":"chatcmpl_test","object":"chat.completion.chunk","created":0,"model":"gpt-4o-mini","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}', + "data: [DONE]", + "", + ].join("\n\n") + const transport = Layer.succeed( + HttpClient.HttpClient, + HttpClient.make((request) => + Effect.sync(() => { + if (request.body._tag !== "Uint8Array") throw new Error(`Unexpected request body: ${request.body._tag}`) + methods.push(request.method) + bodies.push(request.body.body.slice()) + return HttpClientResponse.fromWeb( + request, + new Response(response, { headers: { "content-type": "text/event-stream" } }), + ) + }), + ), + ) + const retryIt = testEffect( + testLayer(LLMClient.layer.pipe(Layer.provide(RequestExecutor.layer.pipe(Layer.provide(transport))))), + ) + + retryIt.effect("lets an Effect plugin send the same POST Request twice", () => + Effect.gen(function* () { + bodies.length = 0 + methods.length = 0 + const agents = yield* Agent.Service + const catalog = yield* Catalog.Service + const hooks = yield* PluginHooks.Service + yield* agents.transform((draft) => + draft.update(Agent.ID.make("build"), (agent) => { + agent.mode = "primary" + agent.permissions.push({ action: "execute", resource: "*", effect: "deny" }) + }), + ) + const pluginHost = host({ + agent: agentHost(agents), + catalog: catalogHost(catalog), + session: { + hook: (...registration: SessionHookRegistration) => { + if (registration[0] === "context") return hooks.register("session", "context", registration[1]) + const middleware = registration[1] + return hooks.register("session", "http", (event) => + Effect.sync(() => { + const next = event.request + event.request = (request) => middleware(event, request, next) + }), + ) + }, + }, + }) + yield* pluginHost.session.hook("http", (_context, request, next) => + Effect.gen(function* () { + yield* next(request).pipe(Effect.flatMap((response) => Effect.promise(() => response.text()))) + return yield* next(request) + }), + ) + yield* Effect.forEach(SystemPromptPlugin.Plugins, (plugin) => plugin.effect(pluginHost), { discard: true }) + const { db } = yield* Database.Service + yield* db + .insert(ProjectTable) + .values({ id: Project.ID.global, worktree: AbsolutePath.make("/project"), sandboxes: [] }) + .onConflictDoNothing() + .run() + .pipe(Effect.orDie) + const retrySessionID = Session.ID.make("ses_model_request_http_retry") + yield* db + .insert(SessionTable) + .values({ + id: retrySessionID, + project_id: Project.ID.global, + slug: "test", + directory: "/project", + title: "test", + version: "test", + }) + .run() + .pipe(Effect.orDie) + const session = yield* Session.Service + yield* session.prompt({ sessionID: retrySessionID, text: "Say hello.", resume: false }) + + yield* session.resume(retrySessionID) + + expect(methods).toEqual(["POST", "POST"]) + expect(bodies).toHaveLength(2) + expect(bodies[0]?.byteLength).toBeGreaterThan(0) + expect(bodies[1]).toEqual(bodies[0]) + }), + ) +}) diff --git a/packages/www/content/docs/build/plugins.mdx b/packages/www/content/docs/build/plugins.mdx index 9ad2c24f2a9..3900d9b0215 100644 --- a/packages/www/content/docs/build/plugins.mdx +++ b/packages/www/content/docs/build/plugins.mdx @@ -177,7 +177,7 @@ and plugin options. | `ctx.integration` | `list`, `get`, `connect`, `attempt`, `transform`, `reload`, and connection lookup/resolution | | `ctx.plugin` | `list` currently active plugin IDs | | `ctx.reference` | `list`, `transform`, `reload` | -| `ctx.session` | `create`, `get`, `prompt`, `command`, `rename`, `synthetic`, `interrupt`, `wait`, `hook`, and `http` | +| `ctx.session` | `create`, `get`, `prompt`, `command`, `rename`, `synthetic`, `interrupt`, `wait`, and `hook` | | `ctx.skill` | `list`, `transform`, `reload` | | `ctx.tool` | `transform` and `hook` | | `ctx.aisdk` | `hook` | @@ -251,7 +251,8 @@ Runtime hooks intercept live operations: | `ctx.tool.hook("execute.after", callback)` | Terminal `result` on success or `error` on failure | HTTP middleware can modify requests, inspect responses, retry, or return a -response without calling the provider: +response without calling the provider. It applies to native models; AI SDK +models do not currently pass through this hook. ```ts await ctx.session.hook("http", async (event, request, next) => {