diff --git a/packages/core/src/session/runner/publish-llm-event.ts b/packages/core/src/session/runner/publish-llm-event.ts index 755fbd71c91..69129c1ed74 100644 --- a/packages/core/src/session/runner/publish-llm-event.ts +++ b/packages/core/src/session/runner/publish-llm-event.ts @@ -67,7 +67,9 @@ const hostedContent = (result: ToolResultValue): NonEmptyContent => { * concurrently without a lock. Two rules keep that safe, and every method must preserve * them. (1) Commit state marks synchronously before the first await: never a yield * between a check (`tool.settled`, `stepStarted`, ...) and its mark, so check-and-mark - * stays atomic under cooperative scheduling. (2) Never require a cross-source event + * stays atomic under cooperative scheduling. Provider-event publication remains + * uninterruptible through its writes so cancellation cannot strand a mark without + * its durable event. (2) Never require a cross-source event * order: each publishing fiber is sequential, so per-source order holds by construction, * and consumers fold by id/ordinal rather than global position. */ @@ -528,7 +530,7 @@ export const createLLMEventPublisher = (bus: Pick, inp yield* failAssistant({ type: "provider.unknown", message: event.message }) return } - }) + }, Effect.uninterruptible) const publishTraced = Effect.fn("SessionRunner.publishLLMEvent")(publish) diff --git a/packages/core/test/session-runner-tool-events.test.ts b/packages/core/test/session-runner-tool-events.test.ts index ef28979b921..958a5310744 100644 --- a/packages/core/test/session-runner-tool-events.test.ts +++ b/packages/core/test/session-runner-tool-events.test.ts @@ -1,19 +1,28 @@ import { expect, test } from "bun:test" -import { Cause, Effect, Exit, Schema } from "effect" +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 { RelativePath } from "@opencode-ai/core/schema" +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 { it } from "./lib/effect" +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") @@ -116,6 +125,88 @@ test("provider-executed success derives content and retains provider result stat }) }) +testEffect( + AppNodeBuilder.build(LayerNode.group([Database.node, Bus.node, SessionProjector.node]), [ + [Bus.node, 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() + const release = yield* Deferred.make() + const queued = yield* Deferred.make() + 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))