feat(work-items): run remote comments through continuations

src/features/Org2Cloud/SessionConversation/conversationTurnRunner.test.ts

@@ -241,8 +241,11 @@ describe("conversation turn continuation", () => {
       ],
     ];
     const onRunnerReady = vi.fn();
+    const onTurnAccepted = vi.fn();
 
-    const result = await runConversationTurn(params({ onRunnerReady }));
+    const result = await runConversationTurn(
+      params({ onRunnerReady, onTurnAccepted })
+    );
 
     expect(result).toMatchObject({
       runnerSessionId: "fresh-runner",
@@ -259,6 +262,10 @@ describe("conversation turn continuation", () => {
       "intent-1",
       "intent-1"
     );
+    expect(onTurnAccepted).toHaveBeenCalledWith("fresh-runner", "intent-1");
+    expect(onRunnerReady.mock.invocationCallOrder[0]).toBeLessThan(
+      onTurnAccepted.mock.invocationCallOrder[0]
+    );
     expect(loadContinuation("scope", "root")).toMatchObject({
       continuationSessionId: "fresh-runner",
       established: true,
@@ -340,6 +347,49 @@ describe("conversation turn continuation", () => {
     expect(state.cleaned).toContain("runner-dead");
   });
 
+  it("keeps a durably prepared resume on transport ambiguity", async () => {
+    saveContinuation("scope", "root", {
+      continuationSessionId: "runner-prepared",
+      readThroughPlaneSeq: 10,
+      established: true,
+      agentDefinitionId: "agent-a",
+    });
+    state.rejectNextSend = true;
+
+    await expect(
+      runConversationTurn(params({ preserveRunnerOnTransportFailure: true }))
+    ).rejects.toThrow("session cannot accept turns");
+
+    expect(SessionService.create).not.toHaveBeenCalled();
+    expect(state.cleaned).not.toContain("runner-prepared");
+    expect(loadContinuation("scope", "root")).toMatchObject({
+      continuationSessionId: "runner-prepared",
+      established: true,
+    });
+  });
+
+  it("keeps a fresh prepared runner recoverable when its first send is ambiguous", async () => {
+    state.rejectNextSend = true;
+    const onTurnAccepted = vi.fn();
+
+    await expect(
+      runConversationTurn(
+        params({
+          preserveRunnerOnTransportFailure: true,
+          onTurnAccepted,
+        })
+      )
+    ).rejects.toThrow("session cannot accept turns");
+
+    expect(onTurnAccepted).not.toHaveBeenCalled();
+    expect(state.cleaned).not.toContain("fresh-runner");
+    expect(loadContinuation("scope", "root")).toMatchObject({
+      continuationSessionId: "fresh-runner",
+      established: false,
+      bootstrapTurnIntentId: "intent-1",
+    });
+  });
+
   it("deletes an unestablished runner before accepting a different intent", async () => {
     saveContinuation("scope", "root", {
       continuationSessionId: "runner-pending",

src/features/Org2Cloud/SessionConversation/conversationTurnRunner.ts

@@ -199,6 +199,12 @@ export interface RunConversationTurnParams {
   executionScopeKey: string;
   /** Stable logical id for durable redelivery; minted for ordinary chat. */
   turnIntentId?: string;
+  /**
+   * A durable caller may bind this runner before transport send. In that
+   * mode a send error must escape to the caller instead of silently rolling
+   * to a different runner under the same fenced claim.
+   */
+  preserveRunnerOnTransportFailure?: boolean;
   /**
    * Called when the reusable runner and exact runtime intent are known.
    */
@@ -207,6 +213,11 @@ export interface RunConversationTurnParams {
     turnId: string,
     turnIntentId: string
   ) => void | Promise<void>;
+  /** Called after the adapter accepts the exact runtime intent. */
+  onTurnAccepted?: (
+    runnerSessionId: string,
+    turnIntentId: string
+  ) => void | Promise<void>;
   /**
    * Fires after push #1 (the user's message row) lands on the plane — the
    * composer unblocks here; the agent tail streams in later under the same
@@ -449,6 +460,7 @@ async function runConversationTurnSerialized(
         accountId: decision.record.accountId,
       });
     } catch (error) {
+      if (params.preserveRunnerOnTransportFailure) throw error;
       log.warn("continuation send rejected; rolling to a fresh runner", error);
       clearContinuation(params.executionScopeKey, params.rootSessionId);
       await cleanupConversationRunnerBestEffort(
@@ -461,6 +473,10 @@ async function runConversationTurnSerialized(
         userRowAlreadyPushed: true,
       });
     }
+    await params.onTurnAccepted?.(
+      decision.record.continuationSessionId,
+      io.turnIntentId
+    );
     return settleEpisode(params, io, {
       key,
       runnerSessionId: decision.record.continuationSessionId,
@@ -633,8 +649,10 @@ async function dispatchBootstrapEpisode(
       accountId: episode.accountId,
     });
   } catch (error) {
-    clearContinuation(params.executionScopeKey, params.rootSessionId);
-    await cleanupConversationRunnerBestEffort(episode.runnerSessionId);
+    if (!params.preserveRunnerOnTransportFailure) {
+      clearContinuation(params.executionScopeKey, params.rootSessionId);
+      await cleanupConversationRunnerBestEffort(episode.runnerSessionId);
+    }
     throw error;
   }
   if (
@@ -647,6 +665,7 @@ async function dispatchBootstrapEpisode(
   ) {
     log.warn("continuation acceptance could not be persisted");
   }
+  await params.onTurnAccepted?.(episode.runnerSessionId, io.turnIntentId);
   return settleEpisode(params, io, {
     key,
     runnerSessionId: episode.runnerSessionId,

src/features/Org2Cloud/SessionConversation/useWorkItemConversationTurnBridge.ts

@@ -0,0 +1,19 @@
+import { useEffect } from "react";
+
+import { installWorkItemConversationTurnBridge } from "./workItemConversationTurnBridge";
+
+/** Mount once so background Work Item dispatches always find a listener. */
+export function useWorkItemConversationTurnBridge(): void {
+  useEffect(() => {
+    let disposed = false;
+    let unlisten: (() => void) | null = null;
+    void installWorkItemConversationTurnBridge().then((dispose) => {
+      if (disposed) dispose();
+      else unlisten = dispose;
+    });
+    return () => {
+      disposed = true;
+      unlisten?.();
+    };
+  }, []);
+}