| 12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202 |
- import { NodeFileSystem } from "@effect/platform-node"
- import { expect, spyOn } from "bun:test"
- import { Cause, Effect, Exit, Fiber, Layer } from "effect"
- import path from "path"
- import z from "zod"
- import type { Agent } from "../../src/agent/agent"
- import { Agent as AgentSvc } from "../../src/agent/agent"
- import { Bus } from "../../src/bus"
- import { Command } from "../../src/command"
- import { Config } from "../../src/config/config"
- import { FileTime } from "../../src/file/time"
- import { LSP } from "../../src/lsp"
- import { MCP } from "../../src/mcp"
- import { Permission } from "../../src/permission"
- import { Plugin } from "../../src/plugin"
- import { Provider as ProviderSvc } from "../../src/provider/provider"
- import type { 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 { AppFileSystem } from "../../src/filesystem"
- import { SessionCompaction } from "../../src/session/compaction"
- import { SessionProcessor } from "../../src/session/processor"
- import { SessionPrompt } from "../../src/session/prompt"
- import { MessageID, PartID, SessionID } from "../../src/session/schema"
- import { SessionStatus } from "../../src/session/status"
- import { Shell } from "../../src/shell/shell"
- import { Snapshot } from "../../src/snapshot"
- import { TaskTool } from "../../src/tool/task"
- import { ToolRegistry } from "../../src/tool/registry"
- import { Truncate } from "../../src/tool/truncate"
- 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"),
- }
- function defer<T>() {
- let resolve!: (value: T | PromiseLike<T>) => void
- const promise = new Promise<T>((done) => {
- resolve = done
- })
- return { promise, resolve }
- }
- function withSh<A, E, R>(fx: () => Effect.Effect<A, E, R>) {
- return Effect.acquireUseRelease(
- Effect.sync(() => {
- const prev = process.env.SHELL
- process.env.SHELL = "/bin/sh"
- Shell.preferred.reset()
- return prev
- }),
- () => fx(),
- (prev) =>
- Effect.sync(() => {
- if (prev === undefined) delete process.env.SHELL
- else process.env.SHELL = prev
- Shell.preferred.reset()
- }),
- )
- }
- function toolPart(parts: MessageV2.Part[]) {
- return parts.find((part): part is MessageV2.ToolPart => part.type === "tool")
- }
- type CompletedToolPart = MessageV2.ToolPart & { state: MessageV2.ToolStateCompleted }
- type ErrorToolPart = MessageV2.ToolPart & { state: MessageV2.ToolStateError }
- function completedTool(parts: MessageV2.Part[]) {
- const part = toolPart(parts)
- expect(part?.state.status).toBe("completed")
- return part?.state.status === "completed" ? (part as CompletedToolPart) : undefined
- }
- function errorTool(parts: MessageV2.Part[]) {
- const part = toolPart(parts)
- expect(part?.state.status).toBe("error")
- return part?.state.status === "error" ? (part as ErrorToolPart) : undefined
- }
- const mcp = Layer.succeed(
- MCP.Service,
- MCP.Service.of({
- status: () => Effect.succeed({}),
- clients: () => Effect.succeed({}),
- tools: () => Effect.succeed({}),
- prompts: () => Effect.succeed({}),
- resources: () => Effect.succeed({}),
- add: () => Effect.succeed({ status: { status: "disabled" as const } }),
- connect: () => Effect.void,
- disconnect: () => Effect.void,
- getPrompt: () => Effect.succeed(undefined),
- readResource: () => Effect.succeed(undefined),
- startAuth: () => Effect.die("unexpected MCP auth in prompt-effect tests"),
- authenticate: () => Effect.die("unexpected MCP auth in prompt-effect tests"),
- finishAuth: () => Effect.die("unexpected MCP auth in prompt-effect tests"),
- removeAuth: () => Effect.void,
- supportsOAuth: () => Effect.succeed(false),
- hasStoredTokens: () => Effect.succeed(false),
- getAuthStatus: () => Effect.succeed("not_authenticated" as const),
- }),
- )
- const lsp = Layer.succeed(
- LSP.Service,
- LSP.Service.of({
- init: () => Effect.void,
- status: () => Effect.succeed([]),
- hasClients: () => Effect.succeed(false),
- touchFile: () => Effect.void,
- diagnostics: () => Effect.succeed({}),
- hover: () => Effect.succeed(undefined),
- definition: () => Effect.succeed([]),
- references: () => Effect.succeed([]),
- implementation: () => Effect.succeed([]),
- documentSymbol: () => Effect.succeed([]),
- workspaceSymbol: () => Effect.succeed([]),
- prepareCallHierarchy: () => Effect.succeed([]),
- incomingCalls: () => Effect.succeed([]),
- outgoingCalls: () => Effect.succeed([]),
- }),
- )
- const filetime = Layer.succeed(
- FileTime.Service,
- FileTime.Service.of({
- read: () => Effect.void,
- get: () => Effect.succeed(undefined),
- assert: () => Effect.void,
- withLock: (_filepath, fn) => Effect.promise(fn),
- }),
- )
- const status = SessionStatus.layer.pipe(Layer.provideMerge(Bus.layer))
- const infra = Layer.mergeAll(NodeFileSystem.layer, CrossSpawnSpawner.defaultLayer)
- function makeHttp() {
- const deps = Layer.mergeAll(
- Session.defaultLayer,
- Snapshot.defaultLayer,
- LLM.defaultLayer,
- AgentSvc.defaultLayer,
- Command.defaultLayer,
- Permission.layer,
- Plugin.defaultLayer,
- Config.defaultLayer,
- ProviderSvc.defaultLayer,
- filetime,
- lsp,
- mcp,
- AppFileSystem.defaultLayer,
- status,
- ).pipe(Layer.provideMerge(infra))
- const registry = ToolRegistry.layer.pipe(Layer.provideMerge(deps))
- const trunc = Truncate.layer.pipe(Layer.provideMerge(deps))
- const proc = SessionProcessor.layer.pipe(Layer.provideMerge(deps))
- const compact = SessionCompaction.layer.pipe(Layer.provideMerge(proc), Layer.provideMerge(deps))
- return Layer.mergeAll(
- TestLLMServer.layer,
- SessionPrompt.layer.pipe(
- Layer.provideMerge(compact),
- Layer.provideMerge(proc),
- Layer.provideMerge(registry),
- Layer.provideMerge(trunc),
- Layer.provideMerge(deps),
- ),
- )
- }
- const it = testEffect(makeHttp())
- const unix = process.platform !== "win32" ? it.live : it.live.skip
- // Config that registers a custom "test" provider with a "test-model" model
- // so Provider.getModel("test", "test-model") succeeds inside the loop.
- 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,
- },
- },
- },
- }
- }
- const user = Effect.fn("test.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 seed = Effect.fn("test.seed")(function* (sessionID: SessionID, opts?: { finish?: string }) {
- const session = yield* Session.Service
- const msg = yield* user(sessionID, "hello")
- const assistant: MessageV2.Assistant = {
- id: MessageID.ascending(),
- role: "assistant",
- parentID: msg.id,
- sessionID,
- mode: "build",
- agent: "build",
- cost: 0,
- path: { cwd: "/tmp", root: "/tmp" },
- tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } },
- modelID: ref.modelID,
- providerID: ref.providerID,
- time: { created: Date.now() },
- ...(opts?.finish ? { finish: opts.finish } : {}),
- }
- yield* session.updateMessage(assistant)
- yield* session.updatePart({
- id: PartID.ascending(),
- messageID: assistant.id,
- sessionID,
- type: "text",
- text: "hi there",
- })
- return { user: msg, assistant }
- })
- const addSubtask = (sessionID: SessionID, messageID: MessageID, model = ref) =>
- Effect.gen(function* () {
- const session = yield* Session.Service
- yield* session.updatePart({
- id: PartID.ascending(),
- messageID,
- sessionID,
- type: "subtask",
- prompt: "look into the cache key path",
- description: "inspect bug",
- agent: "general",
- model,
- })
- })
- const boot = Effect.fn("test.boot")(function* (input?: { title?: string }) {
- const prompt = yield* SessionPrompt.Service
- const sessions = yield* Session.Service
- const chat = yield* sessions.create(input ?? { title: "Pinned" })
- return { prompt, sessions, chat }
- })
- // Loop semantics
- it.live("loop exits immediately when last assistant has stop finish", () =>
- provideTmpdirServer(
- Effect.fnUntraced(function* ({ llm }) {
- const prompt = yield* SessionPrompt.Service
- const sessions = yield* Session.Service
- const chat = yield* sessions.create({ title: "Pinned" })
- yield* seed(chat.id, { finish: "stop" })
- const result = yield* prompt.loop({ sessionID: chat.id })
- expect(result.info.role).toBe("assistant")
- if (result.info.role === "assistant") expect(result.info.finish).toBe("stop")
- expect(yield* llm.calls).toBe(0)
- }),
- { git: true, config: providerCfg },
- ),
- )
- it.live("loop calls LLM and returns assistant message", () =>
- provideTmpdirServer(
- Effect.fnUntraced(function* ({ llm }) {
- const prompt = yield* SessionPrompt.Service
- const sessions = yield* Session.Service
- const chat = yield* sessions.create({
- title: "Pinned",
- permission: [{ permission: "*", pattern: "*", action: "allow" }],
- })
- yield* prompt.prompt({
- sessionID: chat.id,
- agent: "build",
- noReply: true,
- parts: [{ type: "text", text: "hello" }],
- })
- yield* llm.text("world")
- const result = yield* prompt.loop({ sessionID: chat.id })
- expect(result.info.role).toBe("assistant")
- const parts = result.parts.filter((p) => p.type === "text")
- expect(parts.some((p) => p.type === "text" && p.text === "world")).toBe(true)
- expect(yield* llm.hits).toHaveLength(1)
- }),
- { git: true, config: providerCfg },
- ),
- )
- it.live("static loop returns assistant text through local provider", () =>
- provideTmpdirServer(
- Effect.fnUntraced(function* ({ llm }) {
- const session = yield* Effect.promise(() =>
- Session.create({
- title: "Prompt provider",
- permission: [{ permission: "*", pattern: "*", action: "allow" }],
- }),
- )
- yield* Effect.promise(() =>
- SessionPrompt.prompt({
- sessionID: session.id,
- agent: "build",
- noReply: true,
- parts: [{ type: "text", text: "hello" }],
- }),
- )
- yield* llm.text("world")
- const result = yield* Effect.promise(() => SessionPrompt.loop({ sessionID: session.id }))
- expect(result.info.role).toBe("assistant")
- expect(result.parts.some((part) => part.type === "text" && part.text === "world")).toBe(true)
- expect(yield* llm.hits).toHaveLength(1)
- expect(yield* llm.pending).toBe(0)
- }),
- { git: true, config: providerCfg },
- ),
- )
- it.live("static loop consumes queued replies across turns", () =>
- provideTmpdirServer(
- Effect.fnUntraced(function* ({ llm }) {
- const session = yield* Effect.promise(() =>
- Session.create({
- title: "Prompt provider turns",
- permission: [{ permission: "*", pattern: "*", action: "allow" }],
- }),
- )
- yield* Effect.promise(() =>
- SessionPrompt.prompt({
- sessionID: session.id,
- agent: "build",
- noReply: true,
- parts: [{ type: "text", text: "hello one" }],
- }),
- )
- yield* llm.text("world one")
- const first = yield* Effect.promise(() => SessionPrompt.loop({ sessionID: session.id }))
- expect(first.info.role).toBe("assistant")
- expect(first.parts.some((part) => part.type === "text" && part.text === "world one")).toBe(true)
- yield* Effect.promise(() =>
- SessionPrompt.prompt({
- sessionID: session.id,
- agent: "build",
- noReply: true,
- parts: [{ type: "text", text: "hello two" }],
- }),
- )
- yield* llm.text("world two")
- const second = yield* Effect.promise(() => SessionPrompt.loop({ sessionID: session.id }))
- expect(second.info.role).toBe("assistant")
- expect(second.parts.some((part) => part.type === "text" && part.text === "world two")).toBe(true)
- expect(yield* llm.hits).toHaveLength(2)
- expect(yield* llm.pending).toBe(0)
- }),
- { git: true, config: providerCfg },
- ),
- )
- it.live("loop continues when finish is tool-calls", () =>
- provideTmpdirServer(
- Effect.fnUntraced(function* ({ llm }) {
- const prompt = yield* SessionPrompt.Service
- const sessions = yield* Session.Service
- const session = yield* sessions.create({
- title: "Pinned",
- permission: [{ permission: "*", pattern: "*", action: "allow" }],
- })
- yield* prompt.prompt({
- sessionID: session.id,
- agent: "build",
- noReply: true,
- parts: [{ type: "text", text: "hello" }],
- })
- yield* llm.tool("first", { value: "first" })
- yield* llm.text("second")
- const result = yield* prompt.loop({ sessionID: session.id })
- expect(yield* llm.calls).toBe(2)
- expect(result.info.role).toBe("assistant")
- if (result.info.role === "assistant") {
- expect(result.parts.some((part) => part.type === "text" && part.text === "second")).toBe(true)
- expect(result.info.finish).toBe("stop")
- }
- }),
- { git: true, config: providerCfg },
- ),
- )
- it.live("failed subtask preserves metadata on error tool state", () =>
- provideTmpdirServer(
- Effect.fnUntraced(function* ({ llm }) {
- const prompt = yield* SessionPrompt.Service
- const sessions = yield* Session.Service
- const chat = yield* sessions.create({ title: "Pinned" })
- yield* llm.tool("task", {
- description: "inspect bug",
- prompt: "look into the cache key path",
- subagent_type: "general",
- })
- yield* llm.text("done")
- const msg = yield* user(chat.id, "hello")
- yield* addSubtask(chat.id, msg.id)
- const result = yield* prompt.loop({ sessionID: chat.id })
- expect(result.info.role).toBe("assistant")
- expect(yield* llm.calls).toBe(2)
- const msgs = yield* MessageV2.filterCompactedEffect(chat.id)
- const taskMsg = msgs.find((item) => item.info.role === "assistant" && item.info.agent === "general")
- expect(taskMsg?.info.role).toBe("assistant")
- if (!taskMsg || taskMsg.info.role !== "assistant") return
- const tool = errorTool(taskMsg.parts)
- if (!tool) return
- expect(tool.state.error).toContain("Tool execution failed")
- expect(tool.state.metadata).toBeDefined()
- expect(tool.state.metadata?.sessionId).toBeDefined()
- expect(tool.state.metadata?.model).toEqual({
- providerID: ProviderID.make("test"),
- modelID: ModelID.make("missing-model"),
- })
- }),
- {
- git: true,
- config: (url) => ({
- ...providerCfg(url),
- agent: {
- general: {
- model: "test/missing-model",
- },
- },
- }),
- },
- ),
- )
- it.live(
- "loop sets status to busy then idle",
- () =>
- provideTmpdirServer(
- Effect.fnUntraced(function* ({ llm }) {
- const prompt = yield* SessionPrompt.Service
- const sessions = yield* Session.Service
- const status = yield* SessionStatus.Service
- yield* llm.hang
- const chat = yield* sessions.create({})
- yield* user(chat.id, "hi")
- const fiber = yield* prompt.loop({ sessionID: chat.id }).pipe(Effect.forkChild)
- yield* llm.wait(1)
- expect((yield* status.get(chat.id)).type).toBe("busy")
- yield* prompt.cancel(chat.id)
- yield* Fiber.await(fiber)
- expect((yield* status.get(chat.id)).type).toBe("idle")
- }),
- { git: true, config: providerCfg },
- ),
- 3_000,
- )
- // Cancel semantics
- it.live(
- "cancel interrupts loop and resolves with an assistant message",
- () =>
- provideTmpdirServer(
- Effect.fnUntraced(function* ({ llm }) {
- const prompt = yield* SessionPrompt.Service
- const sessions = yield* Session.Service
- const chat = yield* sessions.create({ title: "Pinned" })
- yield* seed(chat.id)
- yield* llm.hang
- yield* user(chat.id, "more")
- const fiber = yield* prompt.loop({ sessionID: chat.id }).pipe(Effect.forkChild)
- yield* llm.wait(1)
- yield* prompt.cancel(chat.id)
- const exit = yield* Fiber.await(fiber)
- expect(Exit.isSuccess(exit)).toBe(true)
- if (Exit.isSuccess(exit)) {
- expect(exit.value.info.role).toBe("assistant")
- }
- }),
- { git: true, config: providerCfg },
- ),
- 3_000,
- )
- it.live(
- "cancel records MessageAbortedError on interrupted process",
- () =>
- provideTmpdirServer(
- Effect.fnUntraced(function* ({ llm }) {
- const prompt = yield* SessionPrompt.Service
- const sessions = yield* Session.Service
- const chat = yield* sessions.create({ title: "Pinned" })
- yield* llm.hang
- yield* user(chat.id, "hello")
- const fiber = yield* prompt.loop({ sessionID: chat.id }).pipe(Effect.forkChild)
- yield* llm.wait(1)
- yield* prompt.cancel(chat.id)
- const exit = yield* Fiber.await(fiber)
- expect(Exit.isSuccess(exit)).toBe(true)
- if (Exit.isSuccess(exit)) {
- const info = exit.value.info
- if (info.role === "assistant") {
- expect(info.error?.name).toBe("MessageAbortedError")
- }
- }
- }),
- { git: true, config: providerCfg },
- ),
- 3_000,
- )
- it.live(
- "cancel finalizes subtask tool state",
- () =>
- provideTmpdirInstance(
- (dir) =>
- Effect.gen(function* () {
- const ready = defer<void>()
- const aborted = defer<void>()
- const init = spyOn(TaskTool, "init").mockImplementation(async () => ({
- description: "task",
- parameters: z.object({
- description: z.string(),
- prompt: z.string(),
- subagent_type: z.string(),
- task_id: z.string().optional(),
- command: z.string().optional(),
- }),
- execute: async (_args, ctx) => {
- ready.resolve()
- ctx.abort.addEventListener("abort", () => aborted.resolve(), { once: true })
- await new Promise<void>(() => {})
- return {
- title: "",
- metadata: {
- sessionId: SessionID.make("task"),
- model: ref,
- },
- output: "",
- }
- },
- }))
- yield* Effect.addFinalizer(() => Effect.sync(() => init.mockRestore()))
- const { prompt, chat } = yield* boot()
- const msg = yield* user(chat.id, "hello")
- yield* addSubtask(chat.id, msg.id)
- const fiber = yield* prompt.loop({ sessionID: chat.id }).pipe(Effect.forkChild)
- yield* Effect.promise(() => ready.promise)
- yield* prompt.cancel(chat.id)
- yield* Effect.promise(() => aborted.promise)
- const exit = yield* Fiber.await(fiber)
- expect(Exit.isSuccess(exit)).toBe(true)
- const msgs = yield* MessageV2.filterCompactedEffect(chat.id)
- const taskMsg = msgs.find((item) => item.info.role === "assistant" && item.info.agent === "general")
- expect(taskMsg?.info.role).toBe("assistant")
- if (!taskMsg || taskMsg.info.role !== "assistant") return
- const tool = toolPart(taskMsg.parts)
- expect(tool?.type).toBe("tool")
- if (!tool) return
- expect(tool.state.status).not.toBe("running")
- expect(taskMsg.info.time.completed).toBeDefined()
- expect(taskMsg.info.finish).toBeDefined()
- }),
- { git: true, config: cfg },
- ),
- 30_000,
- )
- it.live(
- "cancel with queued callers resolves all cleanly",
- () =>
- provideTmpdirServer(
- Effect.fnUntraced(function* ({ llm }) {
- const prompt = yield* SessionPrompt.Service
- const sessions = yield* Session.Service
- const chat = yield* sessions.create({ title: "Pinned" })
- yield* llm.hang
- yield* user(chat.id, "hello")
- const a = yield* prompt.loop({ sessionID: chat.id }).pipe(Effect.forkChild)
- yield* llm.wait(1)
- const b = yield* prompt.loop({ sessionID: chat.id }).pipe(Effect.forkChild)
- yield* Effect.sleep(50)
- yield* prompt.cancel(chat.id)
- const [exitA, exitB] = yield* Effect.all([Fiber.await(a), Fiber.await(b)])
- expect(Exit.isSuccess(exitA)).toBe(true)
- expect(Exit.isSuccess(exitB)).toBe(true)
- if (Exit.isSuccess(exitA) && Exit.isSuccess(exitB)) {
- expect(exitA.value.info.id).toBe(exitB.value.info.id)
- }
- }),
- { git: true, config: providerCfg },
- ),
- 3_000,
- )
- // Queue semantics
- it.live("concurrent loop callers get same result", () =>
- provideTmpdirInstance(
- (dir) =>
- Effect.gen(function* () {
- const { prompt, chat } = yield* boot()
- yield* seed(chat.id, { finish: "stop" })
- const [a, b] = yield* Effect.all([prompt.loop({ sessionID: chat.id }), prompt.loop({ sessionID: chat.id })], {
- concurrency: "unbounded",
- })
- expect(a.info.id).toBe(b.info.id)
- expect(a.info.role).toBe("assistant")
- yield* prompt.assertNotBusy(chat.id)
- }),
- { git: true },
- ),
- )
- it.live(
- "concurrent loop callers all receive same error result",
- () =>
- provideTmpdirServer(
- Effect.fnUntraced(function* ({ llm }) {
- const prompt = yield* SessionPrompt.Service
- const sessions = yield* Session.Service
- const chat = yield* sessions.create({ title: "Pinned" })
- yield* llm.fail("boom")
- yield* user(chat.id, "hello")
- const [a, b] = yield* Effect.all([prompt.loop({ sessionID: chat.id }), prompt.loop({ sessionID: chat.id })], {
- concurrency: "unbounded",
- })
- expect(a.info.id).toBe(b.info.id)
- expect(a.info.role).toBe("assistant")
- }),
- { git: true, config: providerCfg },
- ),
- 3_000,
- )
- it.live(
- "prompt submitted during an active run is included in the next LLM input",
- () =>
- provideTmpdirServer(
- Effect.fnUntraced(function* ({ llm }) {
- const gate = defer<void>()
- const prompt = yield* SessionPrompt.Service
- const sessions = yield* Session.Service
- const chat = yield* sessions.create({ title: "Pinned" })
- yield* llm.hold("first", gate.promise)
- yield* llm.text("second")
- const a = yield* prompt
- .prompt({
- sessionID: chat.id,
- agent: "build",
- model: ref,
- parts: [{ type: "text", text: "first" }],
- })
- .pipe(Effect.forkChild)
- yield* llm.wait(1)
- const id = MessageID.ascending()
- const b = yield* prompt
- .prompt({
- sessionID: chat.id,
- messageID: id,
- agent: "build",
- model: ref,
- parts: [{ type: "text", text: "second" }],
- })
- .pipe(Effect.forkChild)
- yield* Effect.promise(async () => {
- const end = Date.now() + 5000
- while (Date.now() < end) {
- const msgs = await Effect.runPromise(sessions.messages({ sessionID: chat.id }))
- if (msgs.some((msg) => msg.info.role === "user" && msg.info.id === id)) return
- await new Promise((done) => setTimeout(done, 20))
- }
- throw new Error("timed out waiting for second prompt to save")
- })
- gate.resolve()
- const [ea, eb] = yield* Effect.all([Fiber.await(a), Fiber.await(b)])
- expect(Exit.isSuccess(ea)).toBe(true)
- expect(Exit.isSuccess(eb)).toBe(true)
- expect(yield* llm.calls).toBe(2)
- const msgs = yield* sessions.messages({ sessionID: chat.id })
- const assistants = msgs.filter((msg) => msg.info.role === "assistant")
- expect(assistants).toHaveLength(2)
- const last = assistants.at(-1)
- if (!last || last.info.role !== "assistant") throw new Error("expected second assistant")
- expect(last.info.parentID).toBe(id)
- expect(last.parts.some((part) => part.type === "text" && part.text === "second")).toBe(true)
- const inputs = yield* llm.inputs
- expect(inputs).toHaveLength(2)
- expect(JSON.stringify(inputs.at(-1)?.messages)).toContain("second")
- }),
- { git: true, config: providerCfg },
- ),
- 3_000,
- )
- it.live(
- "assertNotBusy throws BusyError when loop running",
- () =>
- provideTmpdirServer(
- Effect.fnUntraced(function* ({ llm }) {
- const prompt = yield* SessionPrompt.Service
- const sessions = yield* Session.Service
- yield* llm.hang
- const chat = yield* sessions.create({})
- yield* user(chat.id, "hi")
- const fiber = yield* prompt.loop({ sessionID: chat.id }).pipe(Effect.forkChild)
- yield* llm.wait(1)
- const exit = yield* prompt.assertNotBusy(chat.id).pipe(Effect.exit)
- expect(Exit.isFailure(exit)).toBe(true)
- if (Exit.isFailure(exit)) {
- expect(Cause.squash(exit.cause)).toBeInstanceOf(Session.BusyError)
- }
- yield* prompt.cancel(chat.id)
- yield* Fiber.await(fiber)
- }),
- { git: true, config: providerCfg },
- ),
- 3_000,
- )
- it.live("assertNotBusy succeeds when idle", () =>
- provideTmpdirInstance(
- (dir) =>
- Effect.gen(function* () {
- const prompt = yield* SessionPrompt.Service
- const sessions = yield* Session.Service
- const chat = yield* sessions.create({})
- const exit = yield* prompt.assertNotBusy(chat.id).pipe(Effect.exit)
- expect(Exit.isSuccess(exit)).toBe(true)
- }),
- { git: true },
- ),
- )
- // Shell semantics
- it.live(
- "shell rejects with BusyError when loop running",
- () =>
- provideTmpdirServer(
- Effect.fnUntraced(function* ({ llm }) {
- const prompt = yield* SessionPrompt.Service
- const sessions = yield* Session.Service
- const chat = yield* sessions.create({ title: "Pinned" })
- yield* llm.hang
- yield* user(chat.id, "hi")
- const fiber = yield* prompt.loop({ sessionID: chat.id }).pipe(Effect.forkChild)
- yield* llm.wait(1)
- const exit = yield* prompt.shell({ sessionID: chat.id, agent: "build", command: "echo hi" }).pipe(Effect.exit)
- expect(Exit.isFailure(exit)).toBe(true)
- if (Exit.isFailure(exit)) {
- expect(Cause.squash(exit.cause)).toBeInstanceOf(Session.BusyError)
- }
- yield* prompt.cancel(chat.id)
- yield* Fiber.await(fiber)
- }),
- { git: true, config: providerCfg },
- ),
- 3_000,
- )
- unix("shell captures stdout and stderr in completed tool output", () =>
- provideTmpdirInstance(
- (dir) =>
- Effect.gen(function* () {
- const { prompt, chat } = yield* boot()
- const result = yield* prompt.shell({
- sessionID: chat.id,
- agent: "build",
- command: "printf out && printf err >&2",
- })
- expect(result.info.role).toBe("assistant")
- const tool = completedTool(result.parts)
- if (!tool) return
- expect(tool.state.output).toContain("out")
- expect(tool.state.output).toContain("err")
- expect(tool.state.metadata.output).toContain("out")
- expect(tool.state.metadata.output).toContain("err")
- yield* prompt.assertNotBusy(chat.id)
- }),
- { git: true, config: cfg },
- ),
- )
- unix("shell completes a fast command on the preferred shell", () =>
- provideTmpdirInstance(
- (dir) =>
- Effect.gen(function* () {
- const { prompt, chat } = yield* boot()
- const result = yield* prompt.shell({
- sessionID: chat.id,
- agent: "build",
- command: "pwd",
- })
- expect(result.info.role).toBe("assistant")
- const tool = completedTool(result.parts)
- if (!tool) return
- expect(tool.state.input.command).toBe("pwd")
- expect(tool.state.output).toContain(dir)
- expect(tool.state.metadata.output).toContain(dir)
- yield* prompt.assertNotBusy(chat.id)
- }),
- { git: true, config: cfg },
- ),
- )
- unix("shell lists files from the project directory", () =>
- provideTmpdirInstance(
- (dir) =>
- Effect.gen(function* () {
- const { prompt, chat } = yield* boot()
- yield* Effect.promise(() => Bun.write(path.join(dir, "README.md"), "# e2e\n"))
- const result = yield* prompt.shell({
- sessionID: chat.id,
- agent: "build",
- command: "command ls",
- })
- expect(result.info.role).toBe("assistant")
- const tool = completedTool(result.parts)
- if (!tool) return
- expect(tool.state.input.command).toBe("command ls")
- expect(tool.state.output).toContain("README.md")
- expect(tool.state.metadata.output).toContain("README.md")
- yield* prompt.assertNotBusy(chat.id)
- }),
- { git: true, config: cfg },
- ),
- )
- unix("shell captures stderr from a failing command", () =>
- provideTmpdirInstance(
- (dir) =>
- Effect.gen(function* () {
- const { prompt, chat } = yield* boot()
- const result = yield* prompt.shell({
- sessionID: chat.id,
- agent: "build",
- command: "command -v __nonexistent_cmd_e2e__ || echo 'not found' >&2; exit 1",
- })
- expect(result.info.role).toBe("assistant")
- const tool = completedTool(result.parts)
- if (!tool) return
- expect(tool.state.output).toContain("not found")
- expect(tool.state.metadata.output).toContain("not found")
- yield* prompt.assertNotBusy(chat.id)
- }),
- { git: true, config: cfg },
- ),
- )
- unix(
- "shell updates running metadata before process exit",
- () =>
- withSh(() =>
- provideTmpdirInstance(
- (dir) =>
- Effect.gen(function* () {
- const { prompt, chat } = yield* boot()
- const fiber = yield* prompt
- .shell({ sessionID: chat.id, agent: "build", command: "printf first && sleep 0.2 && printf second" })
- .pipe(Effect.forkChild)
- yield* Effect.promise(async () => {
- const start = Date.now()
- while (Date.now() - start < 5000) {
- const msgs = await MessageV2.filterCompacted(MessageV2.stream(chat.id))
- const taskMsg = msgs.find((item) => item.info.role === "assistant")
- const tool = taskMsg ? toolPart(taskMsg.parts) : undefined
- if (tool?.state.status === "running" && tool.state.metadata?.output.includes("first")) return
- await new Promise((done) => setTimeout(done, 20))
- }
- throw new Error("timed out waiting for running shell metadata")
- })
- const exit = yield* Fiber.await(fiber)
- expect(Exit.isSuccess(exit)).toBe(true)
- }),
- { git: true, config: cfg },
- ),
- ),
- 30_000,
- )
- it.live(
- "loop waits while shell runs and starts after shell exits",
- () =>
- provideTmpdirServer(
- Effect.fnUntraced(function* ({ llm }) {
- const prompt = yield* SessionPrompt.Service
- const sessions = yield* Session.Service
- const chat = yield* sessions.create({
- title: "Pinned",
- permission: [{ permission: "*", pattern: "*", action: "allow" }],
- })
- yield* llm.text("after-shell")
- const sh = yield* prompt
- .shell({ sessionID: chat.id, agent: "build", command: "sleep 0.2" })
- .pipe(Effect.forkChild)
- yield* Effect.sleep(50)
- const loop = yield* prompt.loop({ sessionID: chat.id }).pipe(Effect.forkChild)
- yield* Effect.sleep(50)
- expect(yield* llm.calls).toBe(0)
- yield* Fiber.await(sh)
- const exit = yield* Fiber.await(loop)
- expect(Exit.isSuccess(exit)).toBe(true)
- if (Exit.isSuccess(exit)) {
- expect(exit.value.info.role).toBe("assistant")
- expect(exit.value.parts.some((part) => part.type === "text" && part.text === "after-shell")).toBe(true)
- }
- expect(yield* llm.calls).toBe(1)
- }),
- { git: true, config: providerCfg },
- ),
- 3_000,
- )
- it.live(
- "shell completion resumes queued loop callers",
- () =>
- provideTmpdirServer(
- Effect.fnUntraced(function* ({ llm }) {
- const prompt = yield* SessionPrompt.Service
- const sessions = yield* Session.Service
- const chat = yield* sessions.create({
- title: "Pinned",
- permission: [{ permission: "*", pattern: "*", action: "allow" }],
- })
- yield* llm.text("done")
- const sh = yield* prompt
- .shell({ sessionID: chat.id, agent: "build", command: "sleep 0.2" })
- .pipe(Effect.forkChild)
- yield* Effect.sleep(50)
- const a = yield* prompt.loop({ sessionID: chat.id }).pipe(Effect.forkChild)
- const b = yield* prompt.loop({ sessionID: chat.id }).pipe(Effect.forkChild)
- yield* Effect.sleep(50)
- expect(yield* llm.calls).toBe(0)
- yield* Fiber.await(sh)
- const [ea, eb] = yield* Effect.all([Fiber.await(a), Fiber.await(b)])
- expect(Exit.isSuccess(ea)).toBe(true)
- expect(Exit.isSuccess(eb)).toBe(true)
- if (Exit.isSuccess(ea) && Exit.isSuccess(eb)) {
- expect(ea.value.info.id).toBe(eb.value.info.id)
- expect(ea.value.info.role).toBe("assistant")
- }
- expect(yield* llm.calls).toBe(1)
- }),
- { git: true, config: providerCfg },
- ),
- 3_000,
- )
- unix(
- "cancel interrupts shell and resolves cleanly",
- () =>
- withSh(() =>
- provideTmpdirInstance(
- (dir) =>
- Effect.gen(function* () {
- const { prompt, chat } = yield* boot()
- const sh = yield* prompt
- .shell({ sessionID: chat.id, agent: "build", command: "sleep 30" })
- .pipe(Effect.forkChild)
- yield* Effect.sleep(50)
- yield* prompt.cancel(chat.id)
- const status = yield* SessionStatus.Service
- expect((yield* status.get(chat.id)).type).toBe("idle")
- const busy = yield* prompt.assertNotBusy(chat.id).pipe(Effect.exit)
- expect(Exit.isSuccess(busy)).toBe(true)
- const exit = yield* Fiber.await(sh)
- expect(Exit.isSuccess(exit)).toBe(true)
- if (Exit.isSuccess(exit)) {
- expect(exit.value.info.role).toBe("assistant")
- const tool = completedTool(exit.value.parts)
- if (tool) {
- expect(tool.state.output).toContain("User aborted the command")
- }
- }
- }),
- { git: true, config: cfg },
- ),
- ),
- 30_000,
- )
- unix(
- "cancel persists aborted shell result when shell ignores TERM",
- () =>
- withSh(() =>
- provideTmpdirInstance(
- (dir) =>
- Effect.gen(function* () {
- const { prompt, chat } = yield* boot()
- const sh = yield* prompt
- .shell({ sessionID: chat.id, agent: "build", command: "trap '' TERM; sleep 30" })
- .pipe(Effect.forkChild)
- yield* Effect.sleep(50)
- yield* prompt.cancel(chat.id)
- const exit = yield* Fiber.await(sh)
- expect(Exit.isSuccess(exit)).toBe(true)
- if (Exit.isSuccess(exit)) {
- expect(exit.value.info.role).toBe("assistant")
- const tool = completedTool(exit.value.parts)
- if (tool) {
- expect(tool.state.output).toContain("User aborted the command")
- }
- }
- }),
- { git: true, config: cfg },
- ),
- ),
- 30_000,
- )
- unix(
- "cancel interrupts loop queued behind shell",
- () =>
- provideTmpdirInstance(
- (dir) =>
- Effect.gen(function* () {
- const { prompt, chat } = yield* boot()
- const sh = yield* prompt
- .shell({ sessionID: chat.id, agent: "build", command: "sleep 30" })
- .pipe(Effect.forkChild)
- yield* Effect.sleep(50)
- const loop = yield* prompt.loop({ sessionID: chat.id }).pipe(Effect.forkChild)
- yield* Effect.sleep(50)
- yield* prompt.cancel(chat.id)
- const exit = yield* Fiber.await(loop)
- expect(Exit.isSuccess(exit)).toBe(true)
- yield* Fiber.await(sh)
- }),
- { git: true, config: cfg },
- ),
- 30_000,
- )
- unix(
- "shell rejects when another shell is already running",
- () =>
- withSh(() =>
- provideTmpdirInstance(
- (dir) =>
- Effect.gen(function* () {
- const { prompt, chat } = yield* boot()
- const a = yield* prompt
- .shell({ sessionID: chat.id, agent: "build", command: "sleep 30" })
- .pipe(Effect.forkChild)
- yield* Effect.sleep(50)
- const exit = yield* prompt
- .shell({ sessionID: chat.id, agent: "build", command: "echo hi" })
- .pipe(Effect.exit)
- expect(Exit.isFailure(exit)).toBe(true)
- if (Exit.isFailure(exit)) {
- expect(Cause.squash(exit.cause)).toBeInstanceOf(Session.BusyError)
- }
- yield* prompt.cancel(chat.id)
- yield* Fiber.await(a)
- }),
- { git: true, config: cfg },
- ),
- ),
- 30_000,
- )
|