fix(server): background subagents sharing an OpenCode 2 child session all finish

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

@@ -878,6 +878,310 @@ describe("OpenCode2 adapter", () => {
     }).pipe(Effect.scoped),
   );
 
+  /** Another call in the same turn that resumes `child`, by default a background one. */
+  const backgroundResume = (
+    child: string,
+    { id = "call-again", description = "Again", prompt = "again", background = true } = {},
+  ): ReadonlyArray<ProviderReplayEntry> => {
+    const tool = { sessionID: SESSION, assistantMessageID: "msg_assistant", id };
+    return [
+      event("session.tool.input.started", { ...tool, name: "subagent" }),
+      event("session.tool.called", {
+        ...tool,
+        name: "subagent",
+        input: { description, prompt, background, sessionID: child },
+        executed: false,
+      }),
+      event("session.tool.progress", {
+        ...tool,
+        metadata: background ? { sessionID: child, status: "running" } : { sessionID: child },
+      }),
+      // The child was announced for the first call, so its rules are written again.
+      out("session.update", { sessionID: child, permissions: "<any>" }),
+      reply("session.update", null),
+    ];
+  };
+  /** A subagent's report queued for the parent, and named only in its text. */
+  const childReport = (inboxID: string, child: string, description: string) =>
+    event("session.inbox.enqueued", {
+      inboxID,
+      sessionID: SESSION,
+      item: {
+        type: "synthetic",
+        payload: {
+          text: `<subagent sessionID="${child}" state="completed" description="${description}">\nOK\n</subagent>`,
+          metadata: { source: "subagent", childID: child, agent: "General", state: "completed" },
+        },
+        delivery: "steer",
+      },
+    });
+  /** A child prompt OpenCode queues for one call. */
+  const childPrompt = (inboxID: string, child: string, text: string) =>
+    event("session.inbox.enqueued", {
+      inboxID,
+      sessionID: child,
+      item: { type: "user", payload: { text }, delivery: "steer" },
+    });
+  /**
+   * Reported after every event the test cares about: the session moves to
+   * another model, which the adapter announces.
+   */
+  const lastEvent = event("session.model.selected", {
+    sessionID: SESSION,
+    model: { id: "gpt-5", providerID: "openai", variant: "default" },
+  });
+  /** Records each subagent's latest status by its title, and the last event's arrival. */
+  const watchSubagents = (runtime: ProviderAdapterV2SessionRuntime) =>
+    Effect.gen(function* () {
+      const status = new Map<string, string>();
+      const done = { current: false };
+      let wake = yield* Deferred.make<void>();
+      yield* runtime.events.pipe(
+        Stream.tap((event) =>
+          Effect.gen(function* () {
+            if (event.type === "subagent.updated") {
+              status.set(event.subagent.title ?? "", event.subagent.status);
+            }
+            if (
+              event.type === "provider_thread.updated" &&
+              event.providerThread.nativeThreadRef?.nativeId === SESSION &&
+              event.providerThread.nativeMetadata?.modelSelection?.model === "openai/gpt-5"
+            ) {
+              done.current = true;
+            }
+            yield* Deferred.succeed(wake, undefined);
+          }),
+        ),
+        Stream.runDrain,
+        Effect.forkScoped,
+      );
+      const until = (check: () => boolean): Effect.Effect<void> =>
+        Effect.suspend(() => {
+          if (check()) return Effect.void;
+          return Effect.gen(function* () {
+            wake = yield* Deferred.make<void>();
+            if (check()) return;
+            yield* Deferred.await(wake);
+            yield* until(check);
+          });
+        });
+      return { status, until, done: () => done.current };
+    });
+
+  it.effect("settles the background subagent a report names when calls share a child", () =>
+    Effect.gen(function* () {
+      const { runtime, thread } = yield* resumed([
+        ...backgroundLaunch(CHILD),
+        ...backgroundResume(CHILD),
+        event("session.execution.succeeded", { sessionID: SESSION }),
+        // The resumed call's report comes first and names only that call.
+        childReport("msg_report_again", CHILD, "Again"),
+        event("session.usage.updated", { sessionID: SESSION }),
+        event("session.inbox.delivered", { sessionID: SESSION, inboxID: "msg_report_again" }),
+        childReport("msg_report_sleep", CHILD, "Sleep"),
+        event("session.inbox.delivered", { sessionID: SESSION, inboxID: "msg_report_sleep" }),
+        lastEvent,
+      ]);
+      const watch = yield* watchSubagents(runtime);
+      yield* runtime.startTurn(withLineage(thread));
+      yield* watch.until(() => watch.status.get("Again") === "completed");
+      assert.equal(watch.status.get("Sleep"), "running");
+      yield* watch.until(watch.done);
+      assert.equal(watch.status.get("Sleep"), "completed");
+      assert.isFalse(yield* runtime.hasPendingBackgroundWork!);
+    }).pipe(Effect.scoped),
+  );
+
+  it.effect("settles every background call answered by one run of their shared child", () =>
+    Effect.gen(function* () {
+      const { runtime, thread } = yield* resumed([
+        ...backgroundLaunch(CHILD),
+        childPrompt(
+          "msg_child_sleep",
+          CHILD,
+          "You are a subagent spawned by another session.\nsleep",
+        ),
+        event("session.execution.started", { sessionID: CHILD }),
+        event("session.inbox.delivered", { sessionID: CHILD, inboxID: "msg_child_sleep" }),
+        // The second call's prompt is steered into the child's running execution.
+        ...backgroundResume(CHILD),
+        childPrompt("msg_child_again", CHILD, "again"),
+        event("session.inbox.delivered", { sessionID: CHILD, inboxID: "msg_child_again" }),
+        event("session.execution.succeeded", { sessionID: SESSION }),
+        event("session.execution.succeeded", { sessionID: CHILD }),
+        // One execution answered both prompts, so only one report comes.
+        childReport("msg_report", CHILD, "Sleep"),
+        event("session.inbox.delivered", { sessionID: SESSION, inboxID: "msg_report" }),
+        lastEvent,
+      ]);
+      const watch = yield* watchSubagents(runtime);
+      yield* runtime.startTurn(withLineage(thread));
+      yield* watch.until(watch.done);
+      assert.equal(watch.status.get("Sleep"), "completed");
+      assert.equal(watch.status.get("Again"), "completed");
+      assert.isFalse(yield* runtime.hasPendingBackgroundWork!);
+    }).pipe(Effect.scoped),
+  );
+
+  it.effect("keeps a background call whose prompt the shared child has not taken yet", () =>
+    Effect.gen(function* () {
+      const { runtime, thread } = yield* resumed([
+        ...backgroundLaunch(CHILD),
+        // The second call names the child before the first prompt is delivered.
+        ...backgroundResume(CHILD),
+        childPrompt(
+          "msg_child_sleep",
+          CHILD,
+          "You are a subagent spawned by another session.\nsleep",
+        ),
+        event("session.execution.started", { sessionID: CHILD }),
+        event("session.inbox.delivered", { sessionID: CHILD, inboxID: "msg_child_sleep" }),
+        event("session.execution.succeeded", { sessionID: SESSION }),
+        event("session.execution.succeeded", { sessionID: CHILD }),
+        // The second prompt is only queued once the first execution is over.
+        childPrompt("msg_child_again", CHILD, "again"),
+        childReport("msg_report", CHILD, "Sleep"),
+        lastEvent,
+      ]);
+      const watch = yield* watchSubagents(runtime);
+      yield* runtime.startTurn(withLineage(thread));
+      yield* watch.until(watch.done);
+      assert.equal(watch.status.get("Sleep"), "completed");
+      assert.equal(watch.status.get("Again"), "running");
+      assert.isTrue(yield* runtime.hasPendingBackgroundWork!);
+    }).pipe(Effect.scoped),
+  );
+
+  it.effect("matches a shared child's prompt to the call that sent exactly that prompt", () =>
+    Effect.gen(function* () {
+      const { runtime, thread } = yield* resumed([
+        ...backgroundLaunch(CHILD),
+        // Both calls name the child before either prompt is queued; one prompt ends the other.
+        ...backgroundResume(CHILD, { prompt: "go on" }),
+        // The rules written for the first resume already match, so none follow.
+        ...backgroundResume(CHILD, {
+          id: "call-third",
+          description: "Third",
+          prompt: "check, go on",
+        }).slice(0, -2),
+        childPrompt(
+          "msg_child_sleep",
+          CHILD,
+          "You are a subagent spawned by another session.\nsleep",
+        ),
+        event("session.execution.started", { sessionID: CHILD }),
+        event("session.inbox.delivered", { sessionID: CHILD, inboxID: "msg_child_sleep" }),
+        childPrompt("msg_child_third", CHILD, "check, go on"),
+        event("session.inbox.delivered", { sessionID: CHILD, inboxID: "msg_child_third" }),
+        event("session.execution.succeeded", { sessionID: SESSION }),
+        event("session.execution.succeeded", { sessionID: CHILD }),
+        childReport("msg_report", CHILD, "Sleep"),
+        lastEvent,
+      ]);
+      const watch = yield* watchSubagents(runtime);
+      yield* runtime.startTurn(withLineage(thread));
+      yield* watch.until(watch.done);
+      assert.equal(watch.status.get("Sleep"), "completed");
+      assert.equal(watch.status.get("Third"), "completed");
+      assert.equal(watch.status.get("Again"), "running");
+    }).pipe(Effect.scoped),
+  );
+
+  it.effect("keeps a shared child's runs apart across a lost stream", () =>
+    Effect.gen(function* () {
+      const { runtime, thread } = yield* resumed([
+        ...backgroundLaunch(CHILD),
+        childPrompt(
+          "msg_child_sleep",
+          CHILD,
+          "You are a subagent spawned by another session.\nsleep",
+        ),
+        event("session.execution.started", { sessionID: CHILD }),
+        event("session.inbox.delivered", { sessionID: CHILD, inboxID: "msg_child_sleep" }),
+        ...backgroundResume(CHILD),
+        childPrompt("msg_child_again", CHILD, "again"),
+        event("session.execution.succeeded", { sessionID: SESSION }),
+        // The child's run ends and the next one starts while the stream is down.
+        { type: "runtime_exit", status: "success" } as const,
+        out("event.subscribe"),
+        out("session.active"),
+        replyData("session.active", { [CHILD]: { type: "running" } }),
+        // The child's running turn reads back what it missed.
+        out("message.list", { sessionID: CHILD, order: "desc", limit: "50" }),
+        reply("message.list", { data: [], cursor: {} }),
+        out("permission.list", { sessionID: CHILD }),
+        replyData("permission.list", []),
+        out("session.form.list", { sessionID: CHILD }),
+        replyData("session.form.list", []),
+        event("session.inbox.delivered", { sessionID: CHILD, inboxID: "msg_child_again" }),
+        event("session.execution.succeeded", { sessionID: CHILD }),
+        childReport("msg_report", CHILD, "Again"),
+        lastEvent,
+      ]);
+      const watch = yield* watchSubagents(runtime);
+      yield* runtime.startTurn(withLineage(thread));
+      yield* watch.until(watch.done);
+      assert.equal(watch.status.get("Again"), "completed");
+      assert.equal(watch.status.get("Sleep"), "running");
+    }).pipe(Effect.scoped),
+  );
+
+  it.effect("fails a background call when the foreground call sharing its run fails", () =>
+    Effect.gen(function* () {
+      const tool = { sessionID: SESSION, assistantMessageID: "msg_assistant", id: "call-again" };
+      const { runtime, thread } = yield* resumed([
+        ...backgroundLaunch(CHILD),
+        childPrompt(
+          "msg_child_sleep",
+          CHILD,
+          "You are a subagent spawned by another session.\nsleep",
+        ),
+        event("session.execution.started", { sessionID: CHILD }),
+        event("session.inbox.delivered", { sessionID: CHILD, inboxID: "msg_child_sleep" }),
+        // A foreground call resumes the child while it still runs.
+        event("session.tool.input.started", { ...tool, name: "subagent" }),
+        event("session.tool.called", {
+          ...tool,
+          name: "subagent",
+          input: { description: "Again", prompt: "again", sessionID: CHILD },
+          executed: false,
+        }),
+        event("session.tool.progress", { ...tool, metadata: { sessionID: CHILD } }),
+        out("session.update", { sessionID: CHILD, permissions: "<any>" }),
+        reply("session.update", null),
+        childPrompt("msg_child_again", CHILD, "again"),
+        event("session.inbox.delivered", { sessionID: CHILD, inboxID: "msg_child_again" }),
+        // The run fails; the foreground call reports it and no report follows.
+        event("session.execution.failed", {
+          sessionID: CHILD,
+          error: { type: "unknown", message: "boom" },
+        }),
+        // `session.tool.failed` is at version 2 of its durable schema.
+        {
+          type: "emit_inbound",
+          frame: {
+            type: "sdk.event",
+            event: {
+              id: "evt_sessiontoolfailed0001",
+              created: 1,
+              type: "session.tool.failed",
+              data: { ...tool, error: { type: "unknown", message: "boom" }, executed: true },
+              durable: { aggregateID: SESSION, seq: 1, version: 2 },
+            },
+          },
+        },
+        event("session.execution.succeeded", { sessionID: SESSION }),
+        lastEvent,
+      ]);
+      const watch = yield* watchSubagents(runtime);
+      yield* runtime.startTurn(withLineage(thread));
+      yield* watch.until(watch.done);
+      assert.equal(watch.status.get("Again"), "failed");
+      assert.equal(watch.status.get("Sleep"), "failed");
+      assert.isFalse(yield* runtime.hasPendingBackgroundWork!);
+    }).pipe(Effect.scoped),
+  );
+
   it.effect("leaves no background work pending once a reconnect finds its subagent gone", () =>
     Effect.gen(function* () {
       // The subagent ends while the stream is down: its end, its report and