fix(server): OpenCode 2 continuations replay only their own background reply

apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts

@@ -610,6 +610,28 @@ describe("OpenCode2 adapter", () => {
       lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId },
     } as OrchestrationV2AppThread,
   });
+  /**
+   * The continuation turn the orchestrator starts for an offered wake: its
+   * message is the one the offer named, as `ProviderContinuationService` dispatches it.
+   */
+  const continuationTurn = (
+    thread: OrchestrationV2ProviderThread,
+    offer: ProviderContinuationRequest,
+    suffix = "wake",
+  ) => ({
+    ...withLineage(thread),
+    runId: RunId.make(`run:opencode2-adapter:${suffix}`),
+    runOrdinal: 2,
+    providerTurnOrdinal: 2,
+    attemptId: RunAttemptId.make(`attempt:opencode2-adapter:${suffix}`),
+    message: {
+      ...turnInput(thread).message,
+      messageId: offer.messageId!,
+      text: offer.detail ?? "",
+      createdBy: "agent" as const,
+      creationSource: "provider" as const,
+    },
+  });
   const CHILD = "ses_f1485c529ffea4URrYruwEg0Ja";
   /**
    * The single reader of the runtime's events: resolves `attached` once the
@@ -3762,19 +3784,7 @@ describe("OpenCode2 adapter", () => {
       yield* Effect.gen(function* () {
         while (offers.length === 0) yield* Effect.yieldNow;
       }).pipe(Effect.timeout("2 seconds"), Effect.orDie);
-      yield* runtime.startTurn({
-        ...withLineage(thread),
-        runId: RunId.make("run:opencode2-adapter:wake"),
-        runOrdinal: 2,
-        providerTurnOrdinal: 2,
-        attemptId: RunAttemptId.make("attempt:opencode2-adapter:wake"),
-        message: {
-          ...turnInput(thread).message,
-          messageId: MessageId.make("message:opencode2-adapter:wake"),
-          createdBy: "agent" as const,
-          creationSource: "provider" as const,
-        },
-      });
+      yield* runtime.startTurn(continuationTurn(thread, offers[0]!));
       yield* Deferred.await(bothEnded);
       const [first, second] = turns;
       return { runtime, thread, first: first!, second: second! };
@@ -3841,6 +3851,111 @@ describe("OpenCode2 adapter", () => {
     }).pipe(Effect.scoped),
   );
 
+  it.effect("replays only its own background reply into a continuation turn", () =>
+    Effect.gen(function* () {
+      const CHILD_B = "ses_f1485c529ffeBBBBBBBBBBBBBB";
+      const toolB = { sessionID: SESSION, assistantMessageID: "msg_assistant", id: "call-b" };
+      const report = (inboxID: string, child: string, text: string) =>
+        event("session.inbox.enqueued", {
+          inboxID,
+          sessionID: SESSION,
+          item: {
+            type: "synthetic",
+            payload: {
+              text: `<subagent sessionID="${child}" state="completed" description="Sleep">\n${text}\n</subagent>`,
+              description: "Sleep",
+              metadata: {
+                source: "subagent",
+                childID: child,
+                agent: "General",
+                state: "completed",
+              },
+            },
+            delivery: "steer",
+          },
+        });
+      const reply = (assistantMessageID: string, text: string) =>
+        event("session.text.ended", { sessionID: SESSION, assistantMessageID, ordinal: 0, text });
+      const offers: Array<ProviderContinuationRequest> = [];
+      const bothOffered = yield* Deferred.make<void>();
+      const { runtime, thread } = yield* resumed([
+        ...backgroundLaunch(CHILD),
+        // The same turn starts a second background subagent.
+        event("session.tool.input.started", { ...toolB, name: "subagent" }),
+        event("session.tool.called", {
+          ...toolB,
+          name: "subagent",
+          input: { description: "Sleep", prompt: "sleep", background: true },
+          executed: false,
+        }),
+        event("session.created", { ...childCreated(CHILD_B), sessionID: CHILD_B }),
+        event("session.tool.progress", {
+          ...toolB,
+          metadata: { sessionID: CHILD_B, status: "running" },
+        }),
+        event("session.execution.succeeded", { sessionID: SESSION }),
+        // Each subagent's report wakes the parent into a reply OpenCode runs on its own.
+        event("session.execution.succeeded", { sessionID: CHILD }),
+        report("msg_report_a", CHILD, "A_OK"),
+        event("session.execution.started", { sessionID: SESSION }),
+        event("session.inbox.delivered", { sessionID: SESSION, inboxID: "msg_report_a" }),
+        reply("msg_reply_a", "REPLY_A"),
+        event("session.execution.succeeded", { sessionID: SESSION }),
+        event("session.execution.succeeded", { sessionID: CHILD_B }),
+        report("msg_report_b", CHILD_B, "B_OK"),
+        event("session.execution.started", { sessionID: SESSION }),
+        event("session.inbox.delivered", { sessionID: SESSION, inboxID: "msg_report_b" }),
+        reply("msg_reply_b", "REPLY_B"),
+        event("session.execution.succeeded", { sessionID: SESSION }),
+      ]).pipe(
+        Effect.provideService(ProviderContinuationRequests.ProviderContinuationRequests, {
+          offer: (request) =>
+            Effect.suspend(() =>
+              offers.push(request) === 2
+                ? Deferred.succeed(bothOffered, undefined).pipe(Effect.asVoid)
+                : Effect.void,
+            ),
+          take: Effect.never,
+        }),
+      );
+      const collected: Array<ProviderAdapterV2Event> = [];
+      const ended = yield* Deferred.make<void>();
+      yield* runtime.events.pipe(
+        Stream.tap((event) =>
+          Effect.gen(function* () {
+            collected.push(event);
+            if (collected.filter((entry) => entry.type === "turn.terminal").length === 2) {
+              yield* Deferred.succeed(ended, undefined);
+            }
+          }),
+        ),
+        Stream.runDrain,
+        Effect.forkScoped,
+      );
+      yield* runtime.startTurn(withLineage(thread));
+      yield* Deferred.await(bothOffered);
+      // A's continuation was cancelled before it started; B's runs.
+      yield* runtime.startTurn(continuationTurn(thread, offers[1]!, "wake-b"));
+      yield* Deferred.await(ended);
+      const continuationId = (yield* IdAllocator.IdAllocatorV2).derive.providerTurn({
+        driver: OPENCODE_PROVIDER,
+        nativeTurnId: `${SESSION}:attempt:attempt:opencode2-adapter:wake-b`,
+      });
+      assert.deepEqual(
+        collected.flatMap((event) =>
+          event.type === "turn_item.updated" &&
+          event.turnItem.type === "assistant_message" &&
+          event.turnItem.providerTurnId === continuationId
+            ? [event.turnItem.text]
+            : [],
+        ),
+        ["REPLY_B"],
+      );
+      // Nothing is left held for a continuation that will never run.
+      assert.isFalse(yield* runtime.hasPendingBackgroundWork!);
+    }).pipe(Effect.scoped, Effect.provide(IdAllocator.layer)),
+  );
+
   it.effect("ends a follow-up turn a steer joined exactly once", () =>
     Effect.gen(function* () {
       const reportText = `<subagent sessionID="${CHILD}" state="completed" description="Sleep">\nCHILD_OK\n</subagent>`;
@@ -3918,19 +4033,7 @@ describe("OpenCode2 adapter", () => {
       yield* Effect.gen(function* () {
         while (offers.length === 0) yield* Effect.yieldNow;
       }).pipe(Effect.timeout("2 seconds"), Effect.orDie);
-      const followUp = {
-        ...withLineage(thread),
-        runId: RunId.make("run:opencode2-adapter:wake"),
-        runOrdinal: 2,
-        providerTurnOrdinal: 2,
-        attemptId: RunAttemptId.make("attempt:opencode2-adapter:wake"),
-        message: {
-          ...turnInput(thread).message,
-          messageId: MessageId.make("message:opencode2-adapter:wake"),
-          createdBy: "agent" as const,
-          creationSource: "provider" as const,
-        },
-      };
+      const followUp = continuationTurn(thread, offers[0]!);
       yield* runtime.startTurn(followUp);
       const followUpTurnId = (yield* IdAllocator.IdAllocatorV2).derive.providerTurn({
         driver: OPENCODE_PROVIDER,

apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.ts

@@ -49,6 +49,7 @@ import {
   type OrchestrationV2Subagent,
   type OrchestrationV2TurnItem,
   type OrchestrationV2UserInputQuestion,
+  type MessageId,
   type ModelSelection,
   type ProviderApprovalDecision,
   type ProviderInstanceId,
@@ -303,6 +304,8 @@ interface SubagentCall {
  * the continuation turn T3 opens for it takes them.
  */
 interface Wake {
+  /** The message its continuation run starts from, which names the wake in that turn's input. */
+  readonly messageId: MessageId | undefined;
   readonly events: Array<OpenCode2StreamEvent>;
   running: boolean;
   /** The background subagents whose end it answers, as its turn's notification names them. */
@@ -1658,13 +1661,14 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr
      */
     const offerWake = Effect.fnUntraced(function* (state: ThreadState, wake: Wake) {
       const route = state.providerThread.appThreadId;
-      if (route === null) return;
+      if (route === null || wake.messageId === undefined) return;
       const notification = backgroundWorkNotification(wake.reports);
       yield* continuationRequests.offer({
         threadId: route,
         providerThreadId: state.providerThread.id,
         driver,
         detail: wake.detail,
+        messageId: wake.messageId,
         ...(notification === null ? {} : { notification }),
         dispatchIfCurrent: (dispatch) =>
           wake.dropped ? Effect.succeed(Option.none()) : Effect.map(dispatch, Option.some),
@@ -2343,7 +2347,13 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr
     const onWake = Effect.fnUntraced(function* (state: ThreadState) {
       const { delivered, stopped } = yield* takeReports(state);
       if (stopped) return;
+      const route = state.providerThread.appThreadId;
       const wake: Wake = {
+        // Only a thread's wake is offered a continuation; the ordinal is display metadata.
+        messageId:
+          route === null
+            ? undefined
+            : yield* idAllocator.allocate.message({ threadId: route, ordinal: 0 }),
         events: [],
         running: true,
         reports: delivered.map((entry) => entry.report),
@@ -3377,8 +3387,11 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr
 
     /**
      * The continuation turn for an execution OpenCode started on its own: it
-     * takes the oldest one held and ends with it. One already taken (a user
-     * turn joined it) leaves nothing to run, so the turn ends at once.
+     * takes the wake it was dispatched for, named by its message, and ends
+     * with it. Wakes held before that one had their continuation cancelled or
+     * dropped before it started, so nothing will replay them: they are
+     * dropped. One already taken (a user turn joined it) leaves nothing to
+     * run, so the turn ends at once.
      */
     const runWake = Effect.fnUntraced(function* (
       state: ThreadState,
@@ -3387,8 +3400,14 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr
       yield* beginTurn(state, turnInput, false);
       yield* lock.withPermit(
         Effect.gen(function* () {
-          const wake = state.wakes.shift();
-          if (wake === undefined) return yield* finishTurn(state, { status: "completed" });
+          const index = state.wakes.findIndex(
+            (held) => held.messageId === turnInput.message.messageId,
+          );
+          if (index < 0) return yield* finishTurn(state, { status: "completed" });
+          // Trade-off: a continuation the user reordered ahead of an earlier one
+          // drops that one's reply too, so cancelled ones never pin the session.
+          for (const skipped of state.wakes.splice(0, index)) skipped.dropped = true;
+          const wake = state.wakes.shift()!;
           const turn = state.active;
           if (turn !== undefined && wake.after !== undefined) turn.before = wake.after;
           // Its first report is where its history begins, so fork and rollback cut there.

apps/server/src/orchestration-v2/ProviderContinuationRequests.ts

@@ -41,6 +41,12 @@ export interface ProviderContinuationRequest {
    * reach the provider as a real prompt.
    */
   readonly delivery?: "adapter_buffered" | "message_text";
+  /**
+   * The message the continuation run starts from, when the adapter picks it:
+   * its turn input carries this id, so the adapter can tell which of its held
+   * wakes a run was dispatched for. Otherwise the service allocates one.
+   */
+  readonly messageId?: MessageId;
   readonly dispatchIfCurrent?: <A, E, R>(
     effect: Effect.Effect<A, E, R>,
   ) => Effect.Effect<Option.Option<A>, E, R>;

apps/server/src/orchestration-v2/ProviderContinuationService.test.ts

@@ -180,6 +180,25 @@ describe("ProviderContinuationService", () => {
     });
   });
 
+  it.effect("dispatches a wake under the message its adapter named", () => {
+    return Effect.gen(function* () {
+      const dispatched = yield* Queue.unbounded<unknown>();
+      const messageId = MessageId.make("message-provider-continuation-adapter-named");
+      yield* Effect.gen(function* () {
+        const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
+        yield* requests.offer({ ...request(), messageId });
+        const command = (yield* Queue.take(dispatched)) as { readonly messageId: MessageId };
+        // The adapter matches the continuation run's message to the wake it holds.
+        assert.equal(command.messageId, messageId);
+      }).pipe(
+        Effect.provide(
+          layerTest({ dispatched, getThreadRecords: () => Effect.succeed(projection) }),
+        ),
+        Effect.scoped,
+      );
+    });
+  });
+
   it.effect("delivers a message_text wake as a real prompt, not a buffered wake", () => {
     return Effect.gen(function* () {
       const dispatched = yield* Queue.unbounded<unknown>();

apps/server/src/orchestration-v2/ProviderContinuationService.ts

@@ -147,10 +147,12 @@ export const layer = Layer.effectDiscard(
         }
         // The ordinal is display metadata only: allocate.message appends a
         // random UUID, so a stale projection read here cannot collide ids.
-        const messageId = yield* ids.allocate.message({
-          threadId: request.threadId,
-          ordinal: projection.messages.length + 1,
-        });
+        const messageId =
+          request.messageId ??
+          (yield* ids.allocate.message({
+            threadId: request.threadId,
+            ordinal: projection.messages.length + 1,
+          }));
         const commandId = CommandId.make(`provider-continuation:${messageId}`);
         const dispatch = threads.dispatch({
           type: "message.dispatch",