mirror of
https://github.com/anomalyco/opencode.git
synced 2026-08-04 17:39:52 +00:00
fix(core): preserve request bodies across retries
This commit is contained in:
Aiden Cline
parent
4b19e9ce27
commit
eb16388ec6
3 changed files with 138 additions and 38 deletions
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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<Session.ID, SessionRunner.RunError>({
|
||||
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<typeof LLMClient.Service>) =>
|
||||
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<typeof LLMClient.Service>) =>
|
||||
Layer.effect(
|
||||
SessionExecution.Service,
|
||||
Effect.gen(function* () {
|
||||
const sessionRunner = yield* SessionRunner.Service
|
||||
const coordinator = yield* SessionRunCoordinator.make<Session.ID, SessionRunner.RunError>({
|
||||
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<typeof LLMClient.Service>) =>
|
||||
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])
|
||||
}),
|
||||
)
|
||||
})
|
||||
|
|
|
|||
5
packages/www/content/docs/build/plugins.mdx
vendored
5
packages/www/content/docs/build/plugins.mdx
vendored
|
|
@ -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) => {
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue