import { HttpRecorder } from "@opencode-ai/http-recorder" import * as OpenAIChat from "@opencode-ai/ai/protocols/openai-chat" import { Auth, LLMClient, RequestExecutor } from "@opencode-ai/ai/route" import { Catalog } from "@opencode-ai/core/catalog" import { Database } from "@opencode-ai/core/database/database" import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" import { LayerNodePlatform } from "@opencode-ai/core/effect/app-node-platform" import { LayerNode } from "@opencode-ai/util/effect/layer-node" import { Bus } from "@opencode-ai/core/bus" import { EventTable } from "@opencode-ai/core/event/sql" import { Job } from "@opencode-ai/core/job" import { Permission } from "@opencode-ai/core/permission" import { Agent } from "@opencode-ai/core/agent" import { Config } from "@opencode-ai/core/config" import { Project } from "@opencode-ai/core/project" import { ProjectTable } from "@opencode-ai/core/project/sql" import { AbsolutePath } from "@opencode-ai/core/schema" import { Session } from "@opencode-ai/core/session" import { Snapshot } from "@opencode-ai/core/snapshot" import { SessionCompaction } from "@opencode-ai/core/session/compaction" import { SessionTitle } from "@opencode-ai/core/session/title" import { SessionProjector } from "@opencode-ai/core/session/projector" import { SessionExecution } from "@opencode-ai/core/session/execution" import { SessionRunCoordinator } from "@opencode-ai/core/session/run-coordinator" import { SessionRunner } from "@opencode-ai/core/session/runner" import * as SessionRunnerLLM from "@opencode-ai/core/session/runner/llm" import { SessionRunnerModel } from "@opencode-ai/core/session/runner/model" import { Tool } from "@opencode-ai/core/tool" import { SessionTable } from "@opencode-ai/core/session/sql" import { SessionStore } from "@opencode-ai/core/session/store" import { Location } from "@opencode-ai/core/location" import { InstructionBuiltIns } from "@opencode-ai/core/instructions/builtins" import { InstructionDiscovery } from "@opencode-ai/core/instruction-discovery" import { Instructions } from "@opencode-ai/core/instructions" import { SkillInstructions } from "@opencode-ai/core/skill/instructions" import { ReferenceInstructions } from "@opencode-ai/core/reference/instructions" import { McpInstructions } from "@opencode-ai/core/mcp/instructions" import { PluginSupervisor } from "@opencode-ai/core/plugin/supervisor" import { PluginHooks } from "@opencode-ai/core/plugin/hooks" 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 path from "node:path" import { testEffect } from "./lib/effect" import { agentHost, catalogHost, host } from "./plugin/host" const cassetteName = "session-runner/openai-chat-streams-text" const cassetteDirectory = path.resolve(import.meta.dir, "fixtures/recordings") if (process.env.RECORD === "true") { if (process.env.CI !== undefined) throw new Error("Unset CI before recording HTTP cassettes") HttpRecorder.removeCassetteSync(cassetteName, { directory: cassetteDirectory }) } const cassette = HttpRecorder.layerFetch(cassetteName, { directory: cassetteDirectory }) const executor = RequestExecutor.layer.pipe(Layer.provide(cassette)) const client = LLMClient.layer.pipe(Layer.provide(executor)) const permission = Layer.succeed( Permission.Service, Permission.Service.of({ assert: () => Effect.die("unused"), ask: () => Effect.die("unused"), reply: () => Effect.die("unused"), get: () => Effect.die("unused"), forSession: () => Effect.die("unused"), list: () => Effect.die("unused"), }), ) const model = OpenAIChat.route .with({ endpoint: { baseURL: "https://api.openai.com/v1" }, auth: Auth.bearer(process.env.OPENAI_API_KEY ?? "fixture"), generation: { maxTokens: 20, temperature: 0 }, }) .model({ id: "gpt-4o-mini" }) const models = Layer.mock(SessionRunnerModel.Service)({ resolve: () => Effect.succeed( SessionRunnerModel.resolved(model, { capabilities: { tools: true, input: ["text", "image"], output: ["text"] }, cost: [], }), ), }) const systemContext = Layer.mock(InstructionBuiltIns.Service, { load: () => Effect.succeed(Instructions.empty) }) const instructionContext = Layer.mock(InstructionDiscovery.Service, { load: () => Effect.succeed(Instructions.empty) }) const skillInstructions = Layer.mock(SkillInstructions.Service, { load: () => Effect.succeed(Instructions.empty) }) const referenceInstructions = Layer.mock(ReferenceInstructions.Service, { load: () => Effect.succeed(Instructions.empty), }) const mcpInstructions = Layer.mock(McpInstructions.Service, { load: () => Effect.succeed(Instructions.empty) }) const config = Config.testLayer() const pluginSupervisor = Layer.succeed(PluginSupervisor.Service, PluginSupervisor.Service.of({ flush: Effect.void })) const promptCatalog = Layer.mock(Catalog.Service, { provider: { get: () => Effect.succeed(undefined), all: () => Effect.succeed([]), available: () => Effect.succeed([]), }, model: { get: () => Effect.succeed(undefined), all: () => Effect.succeed([]), available: () => Effect.succeed([]), default: () => Effect.succeed(undefined), 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( AppNodeBuilder.build( LayerNode.group([ Database.node, Bus.node, SessionProjector.node, SessionStore.node, Agent.node, Catalog.node, PluginHooks.node, Tool.node, SessionRunnerModel.node, InstructionBuiltIns.node, InstructionDiscovery.node, SkillInstructions.node, ReferenceInstructions.node, Config.node, Snapshot.node, SessionRunnerLLM.node, Session.node, ]), [ [LayerNodePlatform.llmClient, client], [Permission.node, permission], [Catalog.node, promptCatalog], [SessionRunnerModel.node, models], [InstructionBuiltIns.node, systemContext], [InstructionDiscovery.node, instructionContext], [Location.node, Location.boundNode({ directory: AbsolutePath.make("/project") })], [SkillInstructions.node, skillInstructions], [ReferenceInstructions.node, referenceInstructions], [Config.node, config], [Snapshot.node, Snapshot.noopLayer], [PluginSupervisor.node, pluginSupervisor], [SessionExecution.node, execution], ], ), ) const sessionID = Session.ID.make("ses_runner_recorded") describe("SessionRunnerLLM recorded", () => { it.effect("executes one recorded prompt through the recorded HTTP transport", () => Effect.gen(function* () { 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: (name, callback) => hooks.register("session", name, callback) }, }) 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) yield* db .insert(SessionTable) .values({ id: sessionID, project_id: Project.ID.global, slug: "test", directory: "/project", title: "test", version: "test", }) .onConflictDoNothing() .run() .pipe(Effect.orDie) const session = yield* Session.Service const prompt = yield* session.prompt({ sessionID, text: "Say hello in one short sentence.", resume: false, }) yield* session.resume(sessionID) const messages = yield* session.context(sessionID) expect(messages).toHaveLength(2) expect(messages[0]).toMatchObject({ id: prompt.id, type: "user", text: "Say hello in one short sentence." }) expect(messages[1]).toMatchObject({ type: "assistant", agent: "build", finish: "stop" }) expect(messages[1]?.type === "assistant" ? messages[1].content : []).toMatchObject([ { type: "text", text: "Hello!" }, ]) expect( (yield* db .select({ type: EventTable.type }) .from(EventTable) .where(eq(EventTable.aggregate_id, sessionID)) .orderBy(EventTable.seq) .all()).map((event) => event.type), ).toEqual([ "session.input.admitted.1", "session.instructions.updated.2", "session.input.promoted.1", "session.step.started.1", "session.text.started.1", "session.text.ended.1", "session.step.ended.1", ]) }), ) })