mirror of
https://github.com/anomalyco/opencode.git
synced 2026-09-07 20:34:46 +00:00
36ac35a7c8
Replace exposed layer graph assembly with opaque declarations, checked substitutions, and lifetime-aware compilation. Preserve deep replacement, ordered startup, and Effect-owned resource lifetimes; migrate callers and verify source and published package contracts.
627 lines
25 KiB
TypeScript
627 lines
25 KiB
TypeScript
import { expect, test } from "bun:test"
|
|
import { Cause, Deferred, Effect, Exit, Fiber, Schema } from "effect"
|
|
import { eq } from "drizzle-orm"
|
|
import { LLMEvent } from "@opencode-ai/ai"
|
|
import { Money } from "@opencode-ai/schema/money"
|
|
import { Bus } from "@opencode-ai/core/bus"
|
|
import { Database } from "@opencode-ai/core/database/database"
|
|
import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
|
|
import { EventTable } from "@opencode-ai/core/event/sql"
|
|
import { Event } from "@opencode-ai/schema/event"
|
|
import { Agent } from "@opencode-ai/core/agent"
|
|
import { SessionEvent } from "@opencode-ai/core/session/event"
|
|
import { SessionMessage } from "@opencode-ai/core/session/message"
|
|
import { Session } from "@opencode-ai/core/session"
|
|
import { SessionProjector } from "@opencode-ai/core/session/projector"
|
|
import { SessionMessageTable, SessionTable } from "@opencode-ai/core/session/sql"
|
|
import { Project } from "@opencode-ai/core/project"
|
|
import { ProjectTable } from "@opencode-ai/core/project/sql"
|
|
import { Model } from "@opencode-ai/core/model"
|
|
import { Provider } from "@opencode-ai/core/provider"
|
|
import { AbsolutePath, RelativePath } from "@opencode-ai/core/schema"
|
|
import { Snapshot } from "@opencode-ai/core/snapshot"
|
|
import { createLLMEventPublisher } from "@opencode-ai/core/session/runner/publish-llm-event"
|
|
import { LayerNode } from "@opencode-ai/util/effect/layer-node"
|
|
import { it, testEffect } from "./lib/effect"
|
|
import { TestClock } from "effect/testing"
|
|
|
|
const sessionID = Session.ID.make("ses_tool_event_test")
|
|
const base64 = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAAB"
|
|
|
|
const capture = (
|
|
providerMetadataKey = "anthropic",
|
|
options?: { readonly interruptProgress?: boolean; readonly beforeTextDelta?: Effect.Effect<void> },
|
|
) => {
|
|
const published: Array<{ readonly type: string; readonly data: unknown }> = []
|
|
const bus: Pick<Bus.Interface, "publish"> = {
|
|
publish: (definition, data) => {
|
|
const publish = Effect.sync(() => {
|
|
const event = { id: Event.ID.create(), type: definition.type, data } as Event.Payload<typeof definition>
|
|
published.push({
|
|
type: definition.durable ? Bus.versionedType(definition.type, definition.durable.version) : definition.type,
|
|
data,
|
|
})
|
|
return event
|
|
})
|
|
if (definition.type === SessionEvent.Text.Delta.type && options?.beforeTextDelta)
|
|
return options.beforeTextDelta.pipe(Effect.andThen(publish))
|
|
return definition.type === SessionEvent.Tool.Progress.type && options?.interruptProgress
|
|
? publish.pipe(Effect.andThen(Effect.interrupt))
|
|
: publish
|
|
},
|
|
}
|
|
return {
|
|
published,
|
|
publisher: createLLMEventPublisher(bus, {
|
|
sessionID,
|
|
agent: Agent.ID.make("build"),
|
|
model: {
|
|
id: Model.ID.make("model"),
|
|
providerID: Provider.ID.opencode,
|
|
},
|
|
providerMetadataKey,
|
|
assistantMessageID: SessionMessage.ID.create(),
|
|
}),
|
|
}
|
|
}
|
|
|
|
const call = LLMEvent.toolCall({ id: "call-image", name: "read", input: { path: "pixel.png" } })
|
|
const hostedResult = LLMEvent.toolResult({
|
|
id: "call-image",
|
|
name: "read",
|
|
result: {
|
|
type: "content",
|
|
value: [
|
|
{ type: "text", text: "Image read successfully" },
|
|
{ type: "file", uri: `data:image/png;base64,${base64}`, mime: "image/png", name: "pixel.png" },
|
|
],
|
|
},
|
|
})
|
|
|
|
test("local tool success serializes media base64 once through canonical content", async () => {
|
|
const { published, publisher } = capture()
|
|
await Effect.runPromise(publisher.publish(call))
|
|
await Effect.runPromise(
|
|
publisher.toolExecution(call.id, call.name, {
|
|
output: { type: "media", mime: "image/png" },
|
|
content: [
|
|
{ type: "text", text: "Image read successfully" },
|
|
{ type: "file", uri: `data:image/png;base64,${base64}`, mime: "image/png", name: "pixel.png" },
|
|
],
|
|
}),
|
|
)
|
|
|
|
const success = published.find((event) => event.type === "session.tool.success.2")
|
|
expect(success).toBeDefined()
|
|
const serialized = JSON.stringify(success)
|
|
expect(serialized.split(base64)).toHaveLength(2)
|
|
expect(success?.data).not.toHaveProperty("result")
|
|
expect(success?.data).not.toHaveProperty("output")
|
|
|
|
expect(success?.data).toMatchObject({
|
|
content: [
|
|
{ type: "text", text: "Image read successfully" },
|
|
{ type: "file", uri: `data:image/png;base64,${base64}`, mime: "image/png" },
|
|
],
|
|
})
|
|
})
|
|
|
|
test("provider-executed success derives content and retains provider result state", async () => {
|
|
const { published, publisher } = capture()
|
|
await Effect.runPromise(publisher.publish(LLMEvent.toolCall({ ...call, providerExecuted: true })))
|
|
await Effect.runPromise(
|
|
publisher.publish(
|
|
LLMEvent.toolResult({
|
|
...hostedResult,
|
|
providerExecuted: true,
|
|
providerMetadata: { anthropic: { result: { type: "content", value: [] } } },
|
|
}),
|
|
),
|
|
)
|
|
const success = published.find((event) => event.type === "session.tool.success.2")
|
|
expect(success?.data).not.toHaveProperty("result")
|
|
expect(success?.data).toMatchObject({
|
|
executed: true,
|
|
content: [
|
|
{ type: "text", text: "Image read successfully" },
|
|
{ type: "file", uri: `data:image/png;base64,${base64}`, mime: "image/png" },
|
|
],
|
|
resultState: { result: { type: "content" } },
|
|
})
|
|
})
|
|
|
|
testEffect(
|
|
AppNodeBuilder.build(LayerNode.group([Database.node, Bus.node, SessionProjector.node]), [
|
|
Bus.node.replace(Bus.configured({ persist: true })),
|
|
]),
|
|
).effect("commits a hosted tool result when cancellation races with the aggregate lock", () =>
|
|
Effect.gen(function* () {
|
|
const database = yield* Database.Service
|
|
const bus = yield* Bus.Service
|
|
const held = yield* Deferred.make<void>()
|
|
const release = yield* Deferred.make<void>()
|
|
const queued = yield* Deferred.make<void>()
|
|
const assistantMessageID = SessionMessage.ID.create()
|
|
yield* database.db
|
|
.insert(ProjectTable)
|
|
.values({ id: Project.ID.global, worktree: AbsolutePath.make("/project"), sandboxes: [] })
|
|
.run()
|
|
yield* database.db
|
|
.insert(SessionTable)
|
|
.values({ id: sessionID, project_id: Project.ID.global, slug: "publish", directory: "/project", version: "test" })
|
|
.run()
|
|
const publisher = createLLMEventPublisher(
|
|
{
|
|
publish: (definition, data, options) =>
|
|
(definition.type === SessionEvent.Tool.Success.type ? Deferred.succeed(queued, undefined) : Effect.void).pipe(
|
|
Effect.andThen(bus.publish(definition, data, options)),
|
|
),
|
|
},
|
|
{
|
|
sessionID,
|
|
assistantMessageID,
|
|
agent: Agent.defaultID,
|
|
model: { id: Model.ID.make("test-model"), providerID: Provider.ID.opencode },
|
|
providerMetadataKey: "openai",
|
|
},
|
|
)
|
|
yield* publisher.publish(LLMEvent.toolCall({ ...call, providerExecuted: true }))
|
|
yield* Effect.acquireRelease(
|
|
bus.listen((event) =>
|
|
event.type === SessionEvent.Renamed.type
|
|
? Deferred.succeed(held, undefined).pipe(Effect.andThen(Deferred.await(release)))
|
|
: Effect.void,
|
|
),
|
|
(unsubscribe) => unsubscribe,
|
|
)
|
|
// Listener delivery holds the aggregate lock after this unrelated event commits.
|
|
const holder = yield* bus
|
|
.publish(SessionEvent.Renamed, { sessionID, title: "Hold publication" })
|
|
.pipe(Effect.forkScoped({ startImmediately: true }))
|
|
yield* Deferred.await(held)
|
|
const publication = yield* publisher
|
|
.publish(LLMEvent.toolResult({ ...hostedResult, providerExecuted: true }))
|
|
.pipe(Effect.forkScoped({ startImmediately: true }))
|
|
yield* Effect.addFinalizer(() => Deferred.succeed(release, undefined))
|
|
yield* Deferred.await(queued)
|
|
const cancellation = yield* Fiber.interrupt(publication).pipe(Effect.forkChild({ startImmediately: true }))
|
|
yield* Effect.yieldNow
|
|
|
|
expect(cancellation.pollUnsafe()).toBeUndefined()
|
|
yield* Deferred.succeed(release, undefined)
|
|
yield* Fiber.join(holder)
|
|
yield* Fiber.join(cancellation)
|
|
expect(Exit.hasInterrupts(yield* Fiber.await(publication))).toBe(true)
|
|
expect(yield* publisher.failUnsettledTools({ type: "aborted", message: "Interrupted" })).toBe(false)
|
|
|
|
const events = yield* database.db
|
|
.select({ type: EventTable.type })
|
|
.from(EventTable)
|
|
.where(eq(EventTable.aggregate_id, sessionID))
|
|
.all()
|
|
expect(events.filter((event) => event.type === "session.tool.success.2")).toHaveLength(1)
|
|
expect(events.some((event) => event.type === "session.tool.failed.2")).toBe(false)
|
|
const message = yield* database.db
|
|
.select()
|
|
.from(SessionMessageTable)
|
|
.where(eq(SessionMessageTable.id, assistantMessageID))
|
|
.get()
|
|
expect(message?.data).toMatchObject({
|
|
content: [{ type: "tool", id: call.id, executed: true, state: { status: "completed" } }],
|
|
})
|
|
}),
|
|
)
|
|
|
|
test("interrupted progress metadata remains in the terminal failure snapshot", async () => {
|
|
const { published, publisher } = capture("anthropic", { interruptProgress: true })
|
|
await Effect.runPromise(publisher.publish(call))
|
|
const exit = await Effect.runPromiseExit(publisher.progress(call.id, { phase: "visible" }))
|
|
expect(Exit.isFailure(exit) && Cause.hasInterruptsOnly(exit.cause)).toBe(true)
|
|
await Effect.runPromise(publisher.failUnsettledTools({ type: "aborted", message: "interrupted" }))
|
|
|
|
expect(published.find((event) => event.type === "session.tool.failed.2")?.data).toMatchObject({
|
|
metadata: { phase: "visible" },
|
|
})
|
|
})
|
|
|
|
test("interrupted subagent failures expose their existing child session to the model", async () => {
|
|
const { published, publisher } = capture("anthropic", { interruptProgress: true })
|
|
const subagent = LLMEvent.toolCall({
|
|
id: "call-subagent",
|
|
name: "subagent",
|
|
input: { agent: "general", description: "Recover child", prompt: "Continue working" },
|
|
})
|
|
await Effect.runPromise(publisher.publish(subagent))
|
|
await Effect.runPromiseExit(publisher.progress(subagent.id, { sessionID: "ses_existing_child", status: "running" }))
|
|
await Effect.runPromise(publisher.failUnsettledTools({ type: "aborted", message: "Tool execution interrupted" }))
|
|
|
|
expect(published.find((event) => event.type === "session.tool.failed.2")?.data).toMatchObject({
|
|
error: { type: "aborted", message: "Tool execution interrupted (sessionID: ses_existing_child)" },
|
|
metadata: { sessionID: "ses_existing_child", status: "running" },
|
|
})
|
|
})
|
|
|
|
test("interrupted non-subagent failures do not expose their progress session IDs", async () => {
|
|
const { published, publisher } = capture()
|
|
await Effect.runPromise(publisher.publish(call))
|
|
await Effect.runPromise(publisher.progress(call.id, { sessionID: "ses_private", status: "running" }))
|
|
await Effect.runPromise(publisher.failUnsettledTools({ type: "aborted", message: "Tool execution interrupted" }))
|
|
|
|
expect(published.find((event) => event.type === "session.tool.failed.2")?.data).toMatchObject({
|
|
error: { type: "aborted", message: "Tool execution interrupted" },
|
|
metadata: { sessionID: "ses_private", status: "running" },
|
|
})
|
|
})
|
|
|
|
test("local failure metadata completes the progress snapshot", async () => {
|
|
const { published, publisher } = capture()
|
|
await Effect.runPromise(publisher.publish(call))
|
|
await Effect.runPromise(publisher.progress(call.id, { phase: "running", provider: "old" }))
|
|
await Effect.runPromise(
|
|
publisher.failTool(call.id, { type: "tool.execution", message: "failed" }, { provider: "exa" }),
|
|
)
|
|
|
|
expect(published.find((event) => event.type === "session.tool.failed.2")?.data).toMatchObject({
|
|
metadata: { phase: "running", provider: "exa" },
|
|
})
|
|
})
|
|
|
|
test("failure snapshot retains canonical progress above the default byte limit", async () => {
|
|
const { published, publisher } = capture("anthropic", { interruptProgress: true })
|
|
await Effect.runPromise(publisher.publish(call))
|
|
const detail = "x".repeat(60 * 1024)
|
|
await Effect.runPromiseExit(publisher.progress(call.id, { detail }))
|
|
await Effect.runPromise(publisher.failUnsettledTools({ type: "aborted", message: "interrupted" }))
|
|
|
|
expect(published.find((event) => event.type === "session.tool.failed.2")?.data).toMatchObject({
|
|
metadata: { detail },
|
|
})
|
|
})
|
|
|
|
test("failure before progress omits partial output fields", async () => {
|
|
const { published, publisher } = capture()
|
|
await Effect.runPromise(publisher.publish(call))
|
|
await Effect.runPromise(publisher.failUnsettledTools({ type: "aborted", message: "interrupted" }))
|
|
|
|
const failed = published.find((event) => event.type === "session.tool.failed.2")?.data
|
|
expect(failed).not.toHaveProperty("content")
|
|
expect(failed).not.toHaveProperty("metadata")
|
|
})
|
|
|
|
test("provider metadata is flattened using the route key", async () => {
|
|
const { published, publisher } = capture()
|
|
await Effect.runPromise(
|
|
publisher.publish(
|
|
LLMEvent.reasoningStart({ id: "reasoning", providerMetadata: { anthropic: { signature: "signed" } } }),
|
|
),
|
|
)
|
|
|
|
expect(published.find((event) => event.type === "session.reasoning.started.1")?.data).toMatchObject({
|
|
state: { signature: "signed" },
|
|
})
|
|
})
|
|
|
|
test("reasoning state from start, empty delta, and end is merged", async () => {
|
|
const { published, publisher } = capture()
|
|
await Effect.runPromise(
|
|
publisher.publish(
|
|
LLMEvent.reasoningStart({ id: "reasoning", providerMetadata: { anthropic: { blockType: "thinking" } } }),
|
|
),
|
|
)
|
|
await Effect.runPromise(
|
|
publisher.publish(
|
|
LLMEvent.reasoningDelta({
|
|
id: "reasoning",
|
|
text: "",
|
|
providerMetadata: { anthropic: { signature: "signed" }, gateway: { traceID: "trace" } },
|
|
}),
|
|
),
|
|
)
|
|
await Effect.runPromise(
|
|
publisher.publish(
|
|
LLMEvent.reasoningEnd({ id: "reasoning", providerMetadata: { anthropic: { stopReason: "tool_use" } } }),
|
|
),
|
|
)
|
|
|
|
expect(published.find((event) => event.type === "session.reasoning.ended.1")?.data).toMatchObject({
|
|
state: { blockType: "thinking", signature: "signed", stopReason: "tool_use" },
|
|
})
|
|
})
|
|
|
|
for (const kind of ["text", "reasoning"] as const) {
|
|
it.effect(`publishes the trailing ${kind} chunk while the provider is paused`, () =>
|
|
Effect.gen(function* () {
|
|
const { published, publisher } = capture()
|
|
yield* publisher.publish(LLMEvent[kind === "text" ? "textStart" : "reasoningStart"]({ id: kind }))
|
|
yield* publisher.publish(
|
|
LLMEvent[kind === "text" ? "textDelta" : "reasoningDelta"]({ id: kind, text: "Running the comm" }),
|
|
)
|
|
yield* TestClock.adjust("100 millis")
|
|
expect(
|
|
published.filter((event) => event.type === `session.${kind}.delta`).map((event) => event.data),
|
|
).toMatchObject([{ delta: "Running the comm" }])
|
|
|
|
yield* publisher.publish(LLMEvent[kind === "text" ? "textDelta" : "reasoningDelta"]({ id: kind, text: "and." }))
|
|
yield* TestClock.adjust("100 millis")
|
|
expect(
|
|
published.filter((event) => event.type === `session.${kind}.delta`).map((event) => event.data),
|
|
).toMatchObject([{ delta: "Running the comm" }, { delta: "and." }])
|
|
expect(published.some((event) => event.type === `session.${kind}.ended.1`)).toBe(false)
|
|
yield* publisher.flush()
|
|
}),
|
|
)
|
|
}
|
|
|
|
it.effect("batches text deltas and flushes pending text before the terminal event", () =>
|
|
Effect.gen(function* () {
|
|
const { published, publisher } = capture()
|
|
yield* Effect.forEach(
|
|
[
|
|
LLMEvent.textStart({ id: "text" }),
|
|
LLMEvent.textDelta({ id: "text", text: "one" }),
|
|
LLMEvent.textDelta({ id: "text", text: " two" }),
|
|
LLMEvent.textDelta({ id: "text", text: " three" }),
|
|
],
|
|
publisher.publish,
|
|
{ discard: true },
|
|
)
|
|
|
|
expect(published.filter((event) => event.type === "session.text.delta")).toHaveLength(0)
|
|
yield* TestClock.adjust("99 millis")
|
|
expect(published.filter((event) => event.type === "session.text.delta")).toHaveLength(0)
|
|
yield* TestClock.adjust("1 millis")
|
|
expect(published.filter((event) => event.type === "session.text.delta").map((event) => event.data)).toMatchObject([
|
|
{ delta: "one two three" },
|
|
])
|
|
|
|
yield* publisher.publish(LLMEvent.textDelta({ id: "text", text: " four" }))
|
|
yield* publisher.publish(LLMEvent.textDelta({ id: "text", text: " five" }))
|
|
yield* publisher.publish(LLMEvent.textEnd({ id: "text" }))
|
|
expect(published.slice(-2).map((event) => event.type)).toEqual(["session.text.delta", "session.text.ended.1"])
|
|
expect(published.at(-2)?.data).toMatchObject({ delta: " four five" })
|
|
const count = published.length
|
|
yield* TestClock.adjust("1 second")
|
|
expect(published).toHaveLength(count)
|
|
}),
|
|
)
|
|
|
|
it.effect("retains new chunks and orders text-end behind an in-flight timer publication", () =>
|
|
Effect.gen(function* () {
|
|
const entered = yield* Deferred.make<void>()
|
|
const release = yield* Deferred.make<void>()
|
|
const { published, publisher } = capture("anthropic", {
|
|
beforeTextDelta: Deferred.succeed(entered, undefined).pipe(Effect.andThen(Deferred.await(release))),
|
|
})
|
|
yield* publisher.publish(LLMEvent.textStart({ id: "text" }))
|
|
yield* publisher.publish(LLMEvent.textDelta({ id: "text", text: "comm" }))
|
|
yield* TestClock.adjust("100 millis")
|
|
yield* Deferred.await(entered)
|
|
yield* publisher.publish(LLMEvent.textDelta({ id: "text", text: "and." }))
|
|
const ending = yield* publisher
|
|
.publish(LLMEvent.textEnd({ id: "text" }))
|
|
.pipe(Effect.forkChild({ startImmediately: true }))
|
|
expect(published.some((event) => event.type === "session.text.ended.1")).toBe(false)
|
|
yield* Deferred.succeed(release, undefined)
|
|
yield* Fiber.join(ending)
|
|
expect(published.slice(-3)).toMatchObject([
|
|
{ type: "session.text.delta", data: { delta: "comm" } },
|
|
{ type: "session.text.delta", data: { delta: "and." } },
|
|
{ type: "session.text.ended.1", data: { text: "command." } },
|
|
])
|
|
}),
|
|
)
|
|
|
|
it.effect("batches reasoning deltas and flushes pending reasoning before the terminal event", () =>
|
|
Effect.gen(function* () {
|
|
const { published, publisher } = capture()
|
|
yield* Effect.forEach(
|
|
[
|
|
LLMEvent.reasoningStart({ id: "reasoning" }),
|
|
LLMEvent.reasoningDelta({ id: "reasoning", text: "one" }),
|
|
LLMEvent.reasoningDelta({ id: "reasoning", text: " two" }),
|
|
LLMEvent.reasoningDelta({ id: "reasoning", text: " three" }),
|
|
LLMEvent.reasoningEnd({ id: "reasoning" }),
|
|
],
|
|
publisher.publish,
|
|
{ discard: true },
|
|
)
|
|
|
|
expect(
|
|
published.filter((event) => event.type === "session.reasoning.delta").map((event) => event.data),
|
|
).toMatchObject([{ delta: "one two three" }])
|
|
expect(published.slice(-2).map((event) => event.type)).toEqual([
|
|
"session.reasoning.delta",
|
|
"session.reasoning.ended.1",
|
|
])
|
|
}),
|
|
)
|
|
|
|
test("authoritative end values replace accumulated deltas in the durable ended events", async () => {
|
|
const { published, publisher } = capture()
|
|
await Effect.runPromise(
|
|
Effect.forEach(
|
|
[
|
|
LLMEvent.textStart({ id: "text" }),
|
|
LLMEvent.textDelta({ id: "text", text: "Hel" }),
|
|
LLMEvent.textEnd({ id: "text", text: "Hello!" }),
|
|
LLMEvent.reasoningStart({ id: "reasoning" }),
|
|
LLMEvent.reasoningDelta({ id: "reasoning", text: "Thin" }),
|
|
LLMEvent.reasoningEnd({ id: "reasoning", text: "Thinking done." }),
|
|
],
|
|
publisher.publish,
|
|
{ discard: true },
|
|
),
|
|
)
|
|
|
|
expect(published.find((event) => event.type === "session.text.ended.1")?.data).toMatchObject({ text: "Hello!" })
|
|
expect(published.find((event) => event.type === "session.reasoning.ended.1")?.data).toMatchObject({
|
|
text: "Thinking done.",
|
|
})
|
|
})
|
|
|
|
test("tool input deltas are accumulated without being published", async () => {
|
|
const { published, publisher } = capture()
|
|
await Effect.runPromise(
|
|
Effect.forEach(
|
|
[
|
|
LLMEvent.toolInputStart({ id: "call", name: "read" }),
|
|
LLMEvent.toolInputDelta({ id: "call", name: "read", text: '{"path":' }),
|
|
LLMEvent.toolInputDelta({ id: "call", name: "read", text: '"file.txt"}' }),
|
|
LLMEvent.toolInputEnd({ id: "call", name: "read" }),
|
|
],
|
|
publisher.publish,
|
|
{ discard: true },
|
|
),
|
|
)
|
|
|
|
expect(published.some((event) => event.type === "session.tool.input.delta")).toBe(false)
|
|
expect(published.find((event) => event.type === "session.tool.input.ended.1")?.data).toMatchObject({
|
|
text: '{"path":"file.txt"}',
|
|
})
|
|
})
|
|
|
|
test("provider-executed tool metadata is flattened using the route key", async () => {
|
|
const { published, publisher } = capture("openai")
|
|
await Effect.runPromise(
|
|
publisher.publish(
|
|
LLMEvent.toolCall({
|
|
id: "hosted",
|
|
name: "web_search",
|
|
input: { query: "Effect" },
|
|
providerExecuted: true,
|
|
providerMetadata: { openai: { itemId: "call" } },
|
|
}),
|
|
),
|
|
)
|
|
await Effect.runPromise(
|
|
publisher.publish(
|
|
LLMEvent.toolResult({
|
|
id: "hosted",
|
|
name: "web_search",
|
|
result: { type: "json", value: { found: true } },
|
|
providerExecuted: true,
|
|
providerMetadata: { openai: { itemId: "result" } },
|
|
}),
|
|
),
|
|
)
|
|
|
|
expect(published.find((event) => event.type === "session.tool.called.1")?.data).toMatchObject({
|
|
state: { itemId: "call" },
|
|
})
|
|
expect(published.find((event) => event.type === "session.tool.success.2")?.data).toMatchObject({
|
|
resultState: { itemId: "result" },
|
|
})
|
|
})
|
|
|
|
test("binary failure emits no success event", async () => {
|
|
const { published, publisher } = capture()
|
|
await Effect.runPromise(publisher.publish(call))
|
|
await Effect.runPromise(publisher.failTool(call.id, { type: "tool.execution", message: "Cannot read binary file" }))
|
|
expect(published.some((event) => event.type === "session.tool.success.2")).toBe(false)
|
|
expect(published.some((event) => event.type === "session.tool.failed.2")).toBe(true)
|
|
})
|
|
|
|
test("success event data can carry provider-executed result state", () => {
|
|
const decoded = Schema.decodeUnknownSync(SessionEvent.Tool.Success.data)({
|
|
sessionID,
|
|
assistantMessageID: SessionMessage.ID.create(),
|
|
id: "call-old",
|
|
content: [{ type: "file", uri: `data:image/png;base64,${base64}`, mime: "image/png" }],
|
|
executed: true,
|
|
resultState: {
|
|
result: {
|
|
type: "content",
|
|
value: [{ type: "file", uri: `data:image/png;base64,${base64}`, mime: "image/png" }],
|
|
},
|
|
},
|
|
})
|
|
expect(decoded.resultState).toMatchObject({ result: { type: "content" } })
|
|
})
|
|
|
|
test("step finish records settlement without publishing step ended", async () => {
|
|
const { published, publisher } = capture()
|
|
await Effect.runPromise(publisher.publish(LLMEvent.stepStart({ index: 0 })))
|
|
await Effect.runPromise(publisher.publish(LLMEvent.stepFinish({ index: 0, reason: { normalized: "stop" } })))
|
|
|
|
expect(published.map((event) => event.type)).toEqual(["session.step.started.1"])
|
|
expect(publisher.record().finish).toMatchObject({ finish: "stop" })
|
|
})
|
|
|
|
test("content-filter finish retains failure evidence until step closeout", async () => {
|
|
const { published, publisher } = capture()
|
|
await Effect.runPromise(publisher.publish(LLMEvent.stepStart({ index: 0 })))
|
|
await Effect.runPromise(
|
|
publisher.publish(
|
|
LLMEvent.stepFinish({
|
|
index: 0,
|
|
reason: { normalized: "content-filter", raw: "refusal" },
|
|
providerMetadata: {
|
|
anthropic: {
|
|
stopDetails: { type: "refusal", category: "safety", explanation: "Blocked" },
|
|
},
|
|
},
|
|
usage: {
|
|
nonCachedInputTokens: 8,
|
|
outputTokens: 3,
|
|
reasoningTokens: 1,
|
|
},
|
|
}),
|
|
),
|
|
)
|
|
|
|
expect(published.map((event) => event.type)).toEqual(["session.step.started.1"])
|
|
const settlement = publisher.record().finish
|
|
expect(settlement).toMatchObject({
|
|
finish: "content-filter",
|
|
rawFinish: "refusal",
|
|
providerState: {
|
|
stopDetails: { type: "refusal", category: "safety", explanation: "Blocked" },
|
|
},
|
|
tokens: { input: 8, output: 2, reasoning: 1 },
|
|
})
|
|
if (!settlement) throw new Error("Expected content-filter settlement")
|
|
await Effect.runPromise(
|
|
publisher.publishStepFailure({
|
|
cost: Money.USD.make(1.25),
|
|
tokens: settlement.tokens,
|
|
snapshot: Snapshot.ID.make("tree-end"),
|
|
files: [RelativePath.make("src/changed.ts")],
|
|
}),
|
|
)
|
|
expect(published.map((event) => event.type)).toEqual(["session.step.started.1", "session.step.failed.1"])
|
|
expect(published.at(-1)?.data).toMatchObject({
|
|
error: { type: "provider.content-filter", message: "Provider blocked the response" },
|
|
finish: "content-filter",
|
|
rawFinish: "refusal",
|
|
providerState: {
|
|
stopDetails: { type: "refusal", category: "safety", explanation: "Blocked" },
|
|
},
|
|
cost: 1.25,
|
|
tokens: { input: 8, output: 2, reasoning: 1 },
|
|
snapshot: "tree-end",
|
|
files: ["src/changed.ts"],
|
|
})
|
|
})
|
|
|
|
test("content-filter finish preserves partial streamed text and never ends the step successfully", async () => {
|
|
const { published, publisher } = capture()
|
|
await Effect.runPromise(
|
|
Effect.forEach(
|
|
[
|
|
LLMEvent.stepStart({ index: 0 }),
|
|
LLMEvent.textStart({ id: "text" }),
|
|
LLMEvent.textDelta({ id: "text", text: "Partial" }),
|
|
LLMEvent.stepFinish({ index: 0, reason: { normalized: "content-filter" } }),
|
|
],
|
|
(event) => publisher.publish(event),
|
|
{ discard: true },
|
|
),
|
|
)
|
|
await Effect.runPromise(publisher.publishStepFailure())
|
|
|
|
expect(published.some((event) => event.type === "session.step.ended.1")).toBe(false)
|
|
expect(published.find((event) => event.type === "session.text.ended.1")?.data).toMatchObject({ text: "Partial" })
|
|
expect(published.find((event) => event.type === "session.step.failed.1")?.data).toMatchObject({
|
|
error: { type: "provider.content-filter" },
|
|
})
|
|
})
|