import { NodeFileSystem } from "@effect/platform-node" import { expect } from "bun:test" import { APICallError } from "ai" import { Cause, Effect, Exit, Fiber, Layer, ServiceMap, Stream } from "effect" import path from "path" import type { Agent } from "../../src/agent/agent" import { Agent as AgentSvc } from "../../src/agent/agent" import { Bus } from "../../src/bus" import { Config } from "../../src/config/config" import { Permission } from "../../src/permission" import { Plugin } from "../../src/plugin" import { Provider } from "../../src/provider/provider" import { ModelID, ProviderID } from "../../src/provider/schema" import { Session } from "../../src/session" import { LLM } from "../../src/session/llm" import { MessageV2 } from "../../src/session/message-v2" import { SessionProcessor } from "../../src/session/processor" import { MessageID, PartID, SessionID } from "../../src/session/schema" import { SessionStatus } from "../../src/session/status" import { Snapshot } from "../../src/snapshot" import { Log } from "../../src/util/log" import * as CrossSpawnSpawner from "../../src/effect/cross-spawn-spawner" import { provideTmpdirInstance, provideTmpdirServer } from "../fixture/fixture" import { testEffect } from "../lib/effect" import { TestLLMServer } from "../lib/llm-server" Log.init({ print: false }) const ref = { providerID: ProviderID.make("test"), modelID: ModelID.make("test-model"), } const cfg = { provider: { test: { name: "Test", id: "test", env: [], npm: "@ai-sdk/openai-compatible", models: { "test-model": { id: "test-model", name: "Test Model", attachment: false, reasoning: false, temperature: false, tool_call: true, release_date: "2025-01-01", limit: { context: 100000, output: 10000 }, cost: { input: 0, output: 0 }, options: {}, }, }, options: { apiKey: "test-key", baseURL: "http://localhost:1/v1", }, }, }, } function providerCfg(url: string) { return { ...cfg, provider: { ...cfg.provider, test: { ...cfg.provider.test, options: { ...cfg.provider.test.options, baseURL: url, }, }, }, } } function agent(): Agent.Info { return { name: "build", mode: "primary", options: {}, permission: [{ permission: "*", pattern: "*", action: "allow" }], } } function defer() { let resolve!: (value: T | PromiseLike) => void const promise = new Promise((done) => { resolve = done }) return { promise, resolve } } const user = Effect.fn("TestSession.user")(function* (sessionID: SessionID, text: string) { const session = yield* Session.Service const msg = yield* session.updateMessage({ id: MessageID.ascending(), role: "user", sessionID, agent: "build", model: ref, time: { created: Date.now() }, }) yield* session.updatePart({ id: PartID.ascending(), messageID: msg.id, sessionID, type: "text", text, }) return msg }) const assistant = Effect.fn("TestSession.assistant")(function* ( sessionID: SessionID, parentID: MessageID, root: string, ) { const session = yield* Session.Service const msg: MessageV2.Assistant = { id: MessageID.ascending(), role: "assistant", sessionID, mode: "build", agent: "build", path: { cwd: root, root }, cost: 0, tokens: { total: 0, input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 }, }, modelID: ref.modelID, providerID: ref.providerID, parentID, time: { created: Date.now() }, finish: "end_turn", } yield* session.updateMessage(msg) return msg }) const status = SessionStatus.layer.pipe(Layer.provideMerge(Bus.layer)) const infra = Layer.mergeAll(NodeFileSystem.layer, CrossSpawnSpawner.defaultLayer) const deps = Layer.mergeAll( Session.defaultLayer, Snapshot.defaultLayer, AgentSvc.defaultLayer, Permission.layer, Plugin.defaultLayer, Config.defaultLayer, LLM.defaultLayer, status, ).pipe(Layer.provideMerge(infra)) const env = Layer.mergeAll( TestLLMServer.layer, SessionProcessor.layer.pipe(Layer.provideMerge(deps)), ) const it = testEffect(env) // --------------------------------------------------------------------------- // TestLLM kept only for the reasoning test // TODO: reasoning events not available via OpenAI-compatible SSE // --------------------------------------------------------------------------- type Script = Stream.Stream | ((input: LLM.StreamInput) => Stream.Stream) class TestLLM extends ServiceMap.Service< TestLLM, { readonly push: (stream: Script) => Effect.Effect readonly reply: (...items: LLM.Event[]) => Effect.Effect readonly calls: Effect.Effect readonly inputs: Effect.Effect } >()("@test/SessionProcessorLLM") {} function reasoningUsage(input = 1, output = 1, total = input + output) { return { inputTokens: input, outputTokens: output, totalTokens: total, inputTokenDetails: { noCacheTokens: undefined, cacheReadTokens: undefined, cacheWriteTokens: undefined, }, outputTokenDetails: { textTokens: undefined, reasoningTokens: undefined, }, } } const reasoningLlm = Layer.unwrap( Effect.gen(function* () { const queue: Script[] = [] const inputs: LLM.StreamInput[] = [] let calls = 0 const push = Effect.fn("TestLLM.push")((item: Script) => { queue.push(item) return Effect.void }) const reply = Effect.fn("TestLLM.reply")((...items: LLM.Event[]) => push(Stream.make(...items))) return Layer.mergeAll( Layer.succeed( LLM.Service, LLM.Service.of({ stream: (input) => { calls += 1 inputs.push(input) const item = queue.shift() ?? Stream.empty return typeof item === "function" ? item(input) : item }, }), ), Layer.succeed( TestLLM, TestLLM.of({ push, reply, calls: Effect.sync(() => calls), inputs: Effect.sync(() => [...inputs]), }), ), ) }), ) const reasoningDeps = Layer.mergeAll( Session.defaultLayer, Snapshot.defaultLayer, AgentSvc.defaultLayer, Permission.layer, Plugin.defaultLayer, Config.defaultLayer, status, reasoningLlm, ).pipe(Layer.provideMerge(infra)) const reasoningEnv = SessionProcessor.layer.pipe(Layer.provideMerge(reasoningDeps)) const reasoningIt = testEffect(reasoningEnv) function reasoningModel(context: number): Provider.Model { return { id: "test-model", providerID: "test", name: "Test", limit: { context, output: 10 }, cost: { input: 0, output: 0, cache: { read: 0, write: 0 } }, capabilities: { toolcall: true, attachment: false, reasoning: false, temperature: true, input: { text: true, image: false, audio: false, video: false }, output: { text: true, image: false, audio: false, video: false }, }, api: { npm: "@ai-sdk/anthropic" }, options: {}, } as Provider.Model } // --------------------------------------------------------------------------- // Tests // --------------------------------------------------------------------------- it.live("session.processor effect tests capture llm input cleanly", () => provideTmpdirServer( ({ dir, llm }) => Effect.gen(function* () { const processors = yield* SessionProcessor.Service const session = yield* Session.Service yield* llm.text("hello") const chat = yield* session.create({}) const parent = yield* user(chat.id, "hi") const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) const mdl = yield* Effect.promise(() => Provider.getModel(ref.providerID, ref.modelID)) const handle = yield* processors.create({ assistantMessage: msg, sessionID: chat.id, model: mdl, }) const input = { user: { id: parent.id, sessionID: chat.id, role: "user", time: parent.time, agent: parent.agent, model: { providerID: ref.providerID, modelID: ref.modelID }, } satisfies MessageV2.User, sessionID: chat.id, model: mdl, agent: agent(), system: [], messages: [{ role: "user", content: "hi" }], tools: {}, } satisfies LLM.StreamInput const value = yield* handle.process(input) const parts = yield* Effect.promise(() => MessageV2.parts(msg.id)) const calls = yield* llm.calls expect(value).toBe("continue") expect(calls).toBe(1) expect(parts.some((part) => part.type === "text" && part.text === "hello")).toBe(true) }), { git: true, config: (url) => providerCfg(url) }, ), ) it.live("session.processor effect tests stop after token overflow requests compaction", () => provideTmpdirServer( ({ dir, llm }) => Effect.gen(function* () { const processors = yield* SessionProcessor.Service const session = yield* Session.Service yield* llm.text("after", { usage: { input: 100, output: 0 } }) const chat = yield* session.create({}) const parent = yield* user(chat.id, "compact") const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) const base = yield* Effect.promise(() => Provider.getModel(ref.providerID, ref.modelID)) const mdl = { ...base, limit: { context: 20, output: 10 } } const handle = yield* processors.create({ assistantMessage: msg, sessionID: chat.id, model: mdl, }) const value = yield* handle.process({ user: { id: parent.id, sessionID: chat.id, role: "user", time: parent.time, agent: parent.agent, model: { providerID: ref.providerID, modelID: ref.modelID }, } satisfies MessageV2.User, sessionID: chat.id, model: mdl, agent: agent(), system: [], messages: [{ role: "user", content: "compact" }], tools: {}, }) const parts = yield* Effect.promise(() => MessageV2.parts(msg.id)) expect(value).toBe("compact") expect(parts.some((part) => part.type === "text")).toBe(false) expect(parts.some((part) => part.type === "step-finish")).toBe(true) }), { git: true, config: (url) => providerCfg(url) }, ), ) // TODO: reasoning events not available via OpenAI-compatible SSE reasoningIt.live("session.processor effect tests reset reasoning state across retries", () => provideTmpdirInstance( (dir) => Effect.gen(function* () { const test = yield* TestLLM const processors = yield* SessionProcessor.Service const session = yield* Session.Service yield* test.push( Stream.make( { type: "start" }, { type: "reasoning-start", id: "r" }, { type: "reasoning-delta", id: "r", text: "one" }, ).pipe( Stream.concat( Stream.fail( new APICallError({ message: "boom", url: "https://example.com/v1/chat/completions", requestBodyValues: {}, statusCode: 503, responseHeaders: { "retry-after-ms": "0" }, responseBody: '{"error":"boom"}', isRetryable: true, }), ), ), ), ) yield* test.reply( { type: "start" }, { type: "reasoning-start", id: "r" }, { type: "reasoning-delta", id: "r", text: "two" }, { type: "reasoning-end", id: "r" }, { type: "finish-step", finishReason: "stop", rawFinishReason: "stop", response: { id: "res", modelId: "test-model", timestamp: new Date() }, providerMetadata: undefined, usage: reasoningUsage(), }, { type: "finish", finishReason: "stop", rawFinishReason: "stop", totalUsage: reasoningUsage() }, ) const chat = yield* session.create({}) const parent = yield* user(chat.id, "reason") const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) const mdl = reasoningModel(100) const handle = yield* processors.create({ assistantMessage: msg, sessionID: chat.id, model: mdl, }) const value = yield* handle.process({ user: { id: parent.id, sessionID: chat.id, role: "user", time: parent.time, agent: parent.agent, model: { providerID: ref.providerID, modelID: ref.modelID }, } satisfies MessageV2.User, sessionID: chat.id, model: mdl, agent: agent(), system: [], messages: [{ role: "user", content: "reason" }], tools: {}, }) const parts = yield* Effect.promise(() => MessageV2.parts(msg.id)) const reasoning = parts.filter( (part): part is MessageV2.ReasoningPart => part.type === "reasoning", ) expect(value).toBe("continue") expect(yield* test.calls).toBe(2) expect(reasoning.some((part) => part.text === "two")).toBe(true) expect(reasoning.some((part) => part.text === "onetwo")).toBe(false) }), { git: true }, ), ) it.live("session.processor effect tests do not retry unknown json errors", () => provideTmpdirServer( ({ dir, llm }) => Effect.gen(function* () { const processors = yield* SessionProcessor.Service const session = yield* Session.Service yield* llm.error(500, { error: { message: "no_kv_space" } }) const chat = yield* session.create({}) const parent = yield* user(chat.id, "json") const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) const mdl = yield* Effect.promise(() => Provider.getModel(ref.providerID, ref.modelID)) const handle = yield* processors.create({ assistantMessage: msg, sessionID: chat.id, model: mdl, }) const value = yield* handle.process({ user: { id: parent.id, sessionID: chat.id, role: "user", time: parent.time, agent: parent.agent, model: { providerID: ref.providerID, modelID: ref.modelID }, } satisfies MessageV2.User, sessionID: chat.id, model: mdl, agent: agent(), system: [], messages: [{ role: "user", content: "json" }], tools: {}, }) expect(value).toBe("stop") expect(yield* llm.calls).toBe(1) expect(handle.message.error?.name).toBe("UnknownError") }), { git: true, config: (url) => providerCfg(url) }, ), ) it.live("session.processor effect tests retry recognized structured json errors", () => provideTmpdirServer( ({ dir, llm }) => Effect.gen(function* () { const processors = yield* SessionProcessor.Service const session = yield* Session.Service yield* llm.error(429, { type: "error", error: { type: "too_many_requests" } }) yield* llm.text("after") const chat = yield* session.create({}) const parent = yield* user(chat.id, "retry json") const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) const mdl = yield* Effect.promise(() => Provider.getModel(ref.providerID, ref.modelID)) const handle = yield* processors.create({ assistantMessage: msg, sessionID: chat.id, model: mdl, }) const value = yield* handle.process({ user: { id: parent.id, sessionID: chat.id, role: "user", time: parent.time, agent: parent.agent, model: { providerID: ref.providerID, modelID: ref.modelID }, } satisfies MessageV2.User, sessionID: chat.id, model: mdl, agent: agent(), system: [], messages: [{ role: "user", content: "retry json" }], tools: {}, }) const parts = yield* Effect.promise(() => MessageV2.parts(msg.id)) expect(value).toBe("continue") expect(yield* llm.calls).toBe(2) expect(parts.some((part) => part.type === "text" && part.text === "after")).toBe(true) expect(handle.message.error).toBeUndefined() }), { git: true, config: (url) => providerCfg(url) }, ), ) it.live("session.processor effect tests publish retry status updates", () => provideTmpdirServer( ({ dir, llm }) => Effect.gen(function* () { const processors = yield* SessionProcessor.Service const session = yield* Session.Service const bus = yield* Bus.Service yield* llm.error(503, { error: "boom" }) yield* llm.text("") const chat = yield* session.create({}) const parent = yield* user(chat.id, "retry") const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) const mdl = yield* Effect.promise(() => Provider.getModel(ref.providerID, ref.modelID)) const states: number[] = [] const off = yield* bus.subscribeCallback(SessionStatus.Event.Status, (evt) => { if (evt.properties.sessionID !== chat.id) return if (evt.properties.status.type === "retry") states.push(evt.properties.status.attempt) }) const handle = yield* processors.create({ assistantMessage: msg, sessionID: chat.id, model: mdl, }) const value = yield* handle.process({ user: { id: parent.id, sessionID: chat.id, role: "user", time: parent.time, agent: parent.agent, model: { providerID: ref.providerID, modelID: ref.modelID }, } satisfies MessageV2.User, sessionID: chat.id, model: mdl, agent: agent(), system: [], messages: [{ role: "user", content: "retry" }], tools: {}, }) off() expect(value).toBe("continue") expect(yield* llm.calls).toBe(2) expect(states).toStrictEqual([1]) }), { git: true, config: (url) => providerCfg(url) }, ), ) it.live("session.processor effect tests compact on structured context overflow", () => provideTmpdirServer( ({ dir, llm }) => Effect.gen(function* () { const processors = yield* SessionProcessor.Service const session = yield* Session.Service yield* llm.error(400, { type: "error", error: { code: "context_length_exceeded" } }) const chat = yield* session.create({}) const parent = yield* user(chat.id, "compact json") const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) const mdl = yield* Effect.promise(() => Provider.getModel(ref.providerID, ref.modelID)) const handle = yield* processors.create({ assistantMessage: msg, sessionID: chat.id, model: mdl, }) const value = yield* handle.process({ user: { id: parent.id, sessionID: chat.id, role: "user", time: parent.time, agent: parent.agent, model: { providerID: ref.providerID, modelID: ref.modelID }, } satisfies MessageV2.User, sessionID: chat.id, model: mdl, agent: agent(), system: [], messages: [{ role: "user", content: "compact json" }], tools: {}, }) expect(value).toBe("compact") expect(yield* llm.calls).toBe(1) expect(handle.message.error).toBeUndefined() }), { git: true, config: (url) => providerCfg(url) }, ), ) it.live("session.processor effect tests mark pending tools as aborted on cleanup", () => provideTmpdirServer( ({ dir, llm }) => Effect.gen(function* () { const processors = yield* SessionProcessor.Service const session = yield* Session.Service yield* llm.tool("bash", { cmd: "pwd" }) yield* llm.hang const chat = yield* session.create({}) const parent = yield* user(chat.id, "tool abort") const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) const mdl = yield* Effect.promise(() => Provider.getModel(ref.providerID, ref.modelID)) const handle = yield* processors.create({ assistantMessage: msg, sessionID: chat.id, model: mdl, }) const run = yield* handle .process({ user: { id: parent.id, sessionID: chat.id, role: "user", time: parent.time, agent: parent.agent, model: { providerID: ref.providerID, modelID: ref.modelID }, } satisfies MessageV2.User, sessionID: chat.id, model: mdl, agent: agent(), system: [], messages: [{ role: "user", content: "tool abort" }], tools: {}, }) .pipe(Effect.forkChild) yield* llm.wait(1) yield* Effect.sleep("100 millis") yield* Fiber.interrupt(run) const exit = yield* Fiber.await(run) if (Exit.isFailure(exit) && Cause.hasInterruptsOnly(exit.cause)) { yield* handle.abort() } const parts = yield* Effect.promise(() => MessageV2.parts(msg.id)) const tool = parts.find((part): part is MessageV2.ToolPart => part.type === "tool") expect(Exit.isFailure(exit)).toBe(true) if (Exit.isFailure(exit)) { expect(Cause.hasInterruptsOnly(exit.cause)).toBe(true) } expect(yield* llm.calls).toBe(1) expect(tool?.state.status).toBe("error") if (tool?.state.status === "error") { expect(tool.state.error).toBe("Tool execution aborted") expect(tool.state.time.end).toBeDefined() } }), { git: true, config: (url) => providerCfg(url) }, ), ) it.live("session.processor effect tests record aborted errors and idle state", () => provideTmpdirServer( ({ dir, llm }) => Effect.gen(function* () { const seen = defer() const processors = yield* SessionProcessor.Service const session = yield* Session.Service const bus = yield* Bus.Service const sts = yield* SessionStatus.Service yield* llm.hang const chat = yield* session.create({}) const parent = yield* user(chat.id, "abort") const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) const mdl = yield* Effect.promise(() => Provider.getModel(ref.providerID, ref.modelID)) const errs: string[] = [] const off = yield* bus.subscribeCallback(Session.Event.Error, (evt) => { if (evt.properties.sessionID !== chat.id) return if (!evt.properties.error) return errs.push(evt.properties.error.name) seen.resolve() }) const handle = yield* processors.create({ assistantMessage: msg, sessionID: chat.id, model: mdl, }) const run = yield* handle .process({ user: { id: parent.id, sessionID: chat.id, role: "user", time: parent.time, agent: parent.agent, model: { providerID: ref.providerID, modelID: ref.modelID }, } satisfies MessageV2.User, sessionID: chat.id, model: mdl, agent: agent(), system: [], messages: [{ role: "user", content: "abort" }], tools: {}, }) .pipe(Effect.forkChild) yield* llm.wait(1) yield* Fiber.interrupt(run) const exit = yield* Fiber.await(run) if (Exit.isFailure(exit) && Cause.hasInterruptsOnly(exit.cause)) { yield* handle.abort() } yield* Effect.promise(() => seen.promise) const stored = yield* Effect.promise(() => MessageV2.get({ sessionID: chat.id, messageID: msg.id })) const state = yield* sts.get(chat.id) off() expect(Exit.isFailure(exit)).toBe(true) if (Exit.isFailure(exit)) { expect(Cause.hasInterruptsOnly(exit.cause)).toBe(true) } expect(handle.message.error?.name).toBe("MessageAbortedError") expect(stored.info.role).toBe("assistant") if (stored.info.role === "assistant") { expect(stored.info.error?.name).toBe("MessageAbortedError") } expect(state).toMatchObject({ type: "idle" }) expect(errs).toContain("MessageAbortedError") }), { git: true, config: (url) => providerCfg(url) }, ), ) it.live("session.processor effect tests mark interruptions aborted without manual abort", () => provideTmpdirServer( ({ dir, llm }) => Effect.gen(function* () { const processors = yield* SessionProcessor.Service const session = yield* Session.Service const sts = yield* SessionStatus.Service yield* llm.hang const chat = yield* session.create({}) const parent = yield* user(chat.id, "interrupt") const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) const mdl = yield* Effect.promise(() => Provider.getModel(ref.providerID, ref.modelID)) const handle = yield* processors.create({ assistantMessage: msg, sessionID: chat.id, model: mdl, }) const run = yield* handle .process({ user: { id: parent.id, sessionID: chat.id, role: "user", time: parent.time, agent: parent.agent, model: { providerID: ref.providerID, modelID: ref.modelID }, } satisfies MessageV2.User, sessionID: chat.id, model: mdl, agent: agent(), system: [], messages: [{ role: "user", content: "interrupt" }], tools: {}, }) .pipe(Effect.forkChild) yield* llm.wait(1) yield* Fiber.interrupt(run) const exit = yield* Fiber.await(run) const stored = yield* Effect.promise(() => MessageV2.get({ sessionID: chat.id, messageID: msg.id })) const state = yield* sts.get(chat.id) expect(Exit.isFailure(exit)).toBe(true) expect(handle.message.error?.name).toBe("MessageAbortedError") expect(stored.info.role).toBe("assistant") if (stored.info.role === "assistant") { expect(stored.info.error?.name).toBe("MessageAbortedError") } expect(state).toMatchObject({ type: "idle" }) }), { git: true, config: (url) => providerCfg(url) }, ), )