fix(core): make provider publication cancellation-safe (#46026)

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<Bus.Interface, "publish">, inp
         yield* failAssistant({ type: "provider.unknown", message: event.message })
         return
     }
-  })
+  }, Effect.uninterruptible)
 
   const publishTraced = Effect.fn("SessionRunner.publishLLMEvent")(publish)
 

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<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))