authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-10-10 22:46:25-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-10 22:46:25-07:00
logae5fd184288aa600f5afbaaf53eeabc5610f9036
treebb448665a988ae8a356c48c0eb11a784cda8a11e
parentcb7d0739389a602d81000bfb5aeb6ebea9e853bb
signature Signed by PGP key B5690EEEBB952194

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

Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

5 files changed, 861 insertions(+), 64 deletions(-)

apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts+700-52
......@@ -715,6 +715,28 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => {
715715 lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId },
716716 } as OrchestrationV2AppThread,
717717 });
718 /**
719 * The continuation turn the orchestrator starts for an offered wake: its
720 * message is the one the offer named, as `ProviderContinuationService` dispatches it.
721 */
722 const continuationTurn = (
723 thread: OrchestrationV2ProviderThread,
724 offer: ProviderContinuationRequests.ProviderContinuationRequest,
725 suffix = "wake",
726 ) => ({
727 ...withLineage(thread),
728 runId: RunId.make(`run:opencode2-adapter:${suffix}`),
729 runOrdinal: 2,
730 providerTurnOrdinal: 2,
731 attemptId: RunAttemptId.make(`attempt:opencode2-adapter:${suffix}`),
732 message: {
733 ...turnInput(thread).message,
734 messageId: offer.messageId!,
735 text: offer.detail ?? "",
736 createdBy: "agent" as const,
737 creationSource: "provider" as const,
738 },
739 });
718740 const CHILD = "ses_f1485c529ffea4URrYruwEg0Ja";
719741 /**
720742 * The single reader of the runtime's events: resolves `attached` once the
......@@ -1865,7 +1887,8 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => {
18651887
18661888 it.effect("ends a background reply's continuation after a lost stream took a Stop's end", () =>
18671889 Effect.gen(function* () {
1868 const offered = yield* Deferred.make<void>();
1890 const offered =
1891 yield* Deferred.make<ProviderContinuationRequests.ProviderContinuationRequest>();
18691892 const gate = makeProviderReplayGate(["drop", "reply"]);
18701893 const { runtime, thread } = yield* resumed(
18711894 [
......@@ -1926,7 +1949,7 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => {
19261949 { replayGate: gate },
19271950 ).pipe(
19281951 Effect.provideService(ProviderContinuationRequests.ProviderContinuationRequests, {
1929 offer: () => Deferred.succeed(offered, undefined).pipe(Effect.asVoid),
1952 offer: (request) => Deferred.succeed(offered, request).pipe(Effect.asVoid),
19301953 take: Effect.never,
19311954 }),
19321955 );
......@@ -1955,26 +1978,133 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => {
19551978 yield* Fiber.join(interrupt);
19561979 yield* runtime.startTurn({ ...secondTurn(thread), appThread: withLineage(thread).appThread });
19571980 gate.release("drop");
1958 yield* Deferred.await(offered);
19591981 yield* runtime.startTurn({
1960 ...withLineage(thread),
1961 runId: RunId.make("run:opencode2-adapter:wake"),
1982 ...continuationTurn(thread, yield* Deferred.await(offered)),
19621983 runOrdinal: 3,
19631984 providerTurnOrdinal: 3,
1964 attemptId: RunAttemptId.make("attempt:opencode2-adapter:wake"),
1965 message: {
1966 ...turnInput(thread).message,
1967 messageId: MessageId.make("message:opencode2-adapter:wake"),
1968 createdBy: "agent" as const,
1969 creationSource: "provider" as const,
1970 },
19711985 });
19721986 gate.release("reply");
19731987 // The reply's own end finishes its continuation: the lost Stop's end does not take it.
1988 const events = [...(yield* Fiber.join(collected))];
19741989 assert.deepEqual(
1975 [...(yield* Fiber.join(collected))].flatMap((event) =>
1976 event.type === "turn.terminal" ? [event.status] : [],
1990 events.flatMap((event) => (event.type === "turn.terminal" ? [event.status] : [])),
1991 ["interrupted", "failed", "completed"],
1992 );
1993 // The continuation replayed the reply.
1994 assert.include(
1995 events.flatMap((event) =>
1996 event.type === "turn_item.updated" && event.turnItem.type === "assistant_message"
1997 ? [event.turnItem.text]
1998 : [],
1999 ),
2000 "CHILD_OK",
2001 );
2002 }).pipe(Effect.scoped, Effect.provide(TestClock.layer())),
2003 );
2004
2005 it.effect("lets a prompt join a background reply after a lost stream took a Stop's end", () =>
2006 Effect.gen(function* () {
2007 const offered = yield* Deferred.make<void>();
2008 const gate = makeProviderReplayGate(["drop"]);
2009 const { runtime, thread } = yield* resumed(
2010 [
2011 ...backgroundLaunch(CHILD),
2012 // Restarting the turn stops only the parent, and the server never answers.
2013 out("session.interrupt", { sessionID: SESSION }),
2014 reply("session.interrupt", "<hang>"),
2015 // The restarted turn finds the parent still stopping and fails.
2016 out("session.active"),
2017 replyData("session.active", { [SESSION]: { type: "running" } }),
2018 out("session.interrupt", { sessionID: SESSION }),
2019 reply("session.interrupt", { interrupted: true }),
2020 // The stream drops before the parent's end; the reconnect finds it idle.
2021 event("session.usage.updated", { sessionID: SESSION }, "drop"),
2022 { type: "runtime_exit", status: "success" } as const,
2023 out("event.subscribe"),
2024 out("session.active"),
2025 replyData("session.active", { [CHILD]: { type: "running" } }),
2026 out("permission.list", { sessionID: CHILD }),
2027 replyData("permission.list", []),
2028 out("session.form.list", { sessionID: CHILD }),
2029 replyData("session.form.list", []),
2030 // The subagent ends and OpenCode answers its report on its own.
2031 event("session.execution.succeeded", { sessionID: CHILD }),
2032 event("session.inbox.enqueued", {
2033 inboxID: "msg_report",
2034 sessionID: SESSION,
2035 item: {
2036 type: "synthetic",
2037 payload: {
2038 text: `<subagent sessionID="${CHILD}" state="completed" description="Sleep">\nCHILD_OK\n</subagent>`,
2039 description: "Sleep",
2040 metadata: {
2041 source: "subagent",
2042 childID: CHILD,
2043 agent: "General",
2044 state: "completed",
2045 },
2046 },
2047 delivery: "steer",
2048 },
2049 }),
2050 event("session.execution.started", { sessionID: SESSION }),
2051 // A prompt sent while that reply runs joins it, with no check first.
2052 out("session.prompt", { sessionID: SESSION, text: "<any>" }),
2053 promptAccepted,
2054 event("session.text.ended", {
2055 sessionID: SESSION,
2056 assistantMessageID: "msg_followup",
2057 ordinal: 0,
2058 text: "CHILD_OK",
2059 }),
2060 event("session.execution.succeeded", { sessionID: SESSION }),
2061 ],
2062 { replayGate: gate },
2063 ).pipe(
2064 Effect.provideService(ProviderContinuationRequests.ProviderContinuationRequests, {
2065 offer: () => Deferred.succeed(offered, undefined).pipe(Effect.asVoid),
2066 take: Effect.never,
2067 }),
2068 );
2069 // The single reader of the runtime's events.
2070 const attached = yield* Deferred.make<void>();
2071 const ended = yield* runtime.events.pipe(
2072 Stream.tap((event) =>
2073 event.type === "subagent.updated" && event.subagent.childThreadId !== null
2074 ? Deferred.succeed(attached, undefined)
2075 : Effect.void,
19772076 ),
2077 Stream.filter(
2078 (
2079 event,
2080 ): event is Extract<ProviderAdapter.ProviderAdapterV2Event, { type: "turn.terminal" }> =>
2081 event.type === "turn.terminal",
2082 ),
2083 Stream.take(3),
2084 Stream.runCollect,
2085 Effect.forkScoped,
2086 );
2087 yield* runtime.startTurn(withLineage(thread));
2088 yield* Deferred.await(attached);
2089 const interrupt = yield* runtime
2090 .interruptTurn({ providerThread: thread, providerTurnId: yield* providerTurnId })
2091 .pipe(Effect.forkScoped);
2092 yield* TestClock.adjust("11 seconds");
2093 yield* Fiber.join(interrupt);
2094 yield* runtime.startTurn({ ...secondTurn(thread), appThread: withLineage(thread).appThread });
2095 gate.release("drop");
2096 // The reply has started.
2097 yield* Deferred.await(offered);
2098 yield* runtime.startTurn({
2099 ...withLineage(thread),
2100 runId: RunId.make("run:opencode2-adapter:3"),
2101 runOrdinal: 3,
2102 providerTurnOrdinal: 3,
2103 attemptId: RunAttemptId.make("attempt:opencode2-adapter:3"),
2104 });
2105 // The reply's start cleared the Stop's mark, so its end finishes the joined turn.
2106 assert.deepEqual(
2107 [...(yield* Fiber.join(ended))].map((terminal) => terminal.status),
19782108 ["interrupted", "failed", "completed"],
19792109 );
19802110 }).pipe(Effect.scoped, Effect.provide(TestClock.layer())),
......@@ -1987,7 +2117,8 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => {
19872117 */
19882118 const replyAfterUnclearFailure = (endedFirst: boolean) =>
19892119 Effect.gen(function* () {
1990 const offered = yield* Deferred.make<void>();
2120 const offered =
2121 yield* Deferred.make<ProviderContinuationRequests.ProviderContinuationRequest>();
19912122 const gate = makeProviderReplayGate(endedFirst ? ["wake"] : ["wake", "reply"]);
19922123 const { runtime, thread } = yield* resumed(
19932124 [
......@@ -2038,7 +2169,7 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => {
20382169 { replayGate: gate },
20392170 ).pipe(
20402171 Effect.provideService(ProviderContinuationRequests.ProviderContinuationRequests, {
2041 offer: () => Deferred.succeed(offered, undefined).pipe(Effect.asVoid),
2172 offer: (request) => Deferred.succeed(offered, request).pipe(Effect.asVoid),
20422173 take: Effect.never,
20432174 }),
20442175 );
......@@ -2073,21 +2204,12 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => {
20732204 .startTurn({ ...secondTurn(thread), appThread: withLineage(thread).appThread })
20742205 .pipe(Effect.ignore);
20752206 gate.release("wake");
2076 yield* Deferred.await(offered);
20772207 // Everything up to the subagent's next run has been read: the reply ended.
20782208 if (endedFirst) yield* Deferred.await(childRan);
20792209 yield* runtime.startTurn({
2080 ...withLineage(thread),
2081 runId: RunId.make("run:opencode2-adapter:wake"),
2210 ...continuationTurn(thread, yield* Deferred.await(offered)),
20822211 runOrdinal: 3,
20832212 providerTurnOrdinal: 3,
2084 attemptId: RunAttemptId.make("attempt:opencode2-adapter:wake"),
2085 message: {
2086 ...turnInput(thread).message,
2087 messageId: MessageId.make("message:opencode2-adapter:wake"),
2088 createdBy: "agent" as const,
2089 creationSource: "provider" as const,
2090 },
20912213 });
20922214 if (!endedFirst) gate.release("reply");
20932215 // The reply's own end finishes its continuation: the failed prompt's mark does not take it.
......@@ -4378,19 +4500,7 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => {
43784500 yield* Effect.gen(function* () {
43794501 while (offers.length === 0) yield* Effect.yieldNow;
43804502 }).pipe(Effect.timeout("2 seconds"), Effect.orDie);
4381 yield* runtime.startTurn({
4382 ...withLineage(thread),
4383 runId: RunId.make("run:opencode2-adapter:wake"),
4384 runOrdinal: 2,
4385 providerTurnOrdinal: 2,
4386 attemptId: RunAttemptId.make("attempt:opencode2-adapter:wake"),
4387 message: {
4388 ...turnInput(thread).message,
4389 messageId: MessageId.make("message:opencode2-adapter:wake"),
4390 createdBy: "agent" as const,
4391 creationSource: "provider" as const,
4392 },
4393 });
4503 yield* runtime.startTurn(continuationTurn(thread, offers[0]!));
43944504 yield* Deferred.await(bothEnded);
43954505 const [first, second] = turns;
43964506 return { runtime, thread, first: first!, second: second! };
......@@ -4457,6 +4567,556 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => {
44574567 }).pipe(Effect.scoped),
44584568 );
44594569
4570 it.effect("keeps a subagent a stopped background reply started off the next turn", () =>
4571 Effect.gen(function* () {
4572 const CHILD_B = "ses_f1485c529ffeBBBBBBBBBBBBBB";
4573 const SPAWNED = "ses_f1485c529ffeDDDDDDDDDDDDDD";
4574 const NESTED = "ses_f1485c529ffeNNNNNNNNNNNNNN";
4575 const toolB = { sessionID: SESSION, assistantMessageID: "msg_assistant", id: "call-b" };
4576 const spawn = { sessionID: SESSION, assistantMessageID: "msg_reply", id: "call-spawn" };
4577 const offers: Array<ProviderContinuationRequests.ProviderContinuationRequest> = [];
4578 const gate = makeProviderReplayGate(["settled"]);
4579 const { runtime, thread } = yield* resumed(
4580 [
4581 ...backgroundLaunch(CHILD),
4582 // The same turn starts a second background subagent, which runs on.
4583 event("session.tool.input.started", { ...toolB, name: "subagent" }),
4584 event("session.tool.called", {
4585 ...toolB,
4586 name: "subagent",
4587 input: { description: "Sleep", prompt: "sleep", background: true },
4588 executed: false,
4589 }),
4590 event("session.created", { ...childCreated(CHILD_B), sessionID: CHILD_B }),
4591 event("session.tool.progress", {
4592 ...toolB,
4593 metadata: { sessionID: CHILD_B, status: "running" },
4594 }),
4595 event("session.execution.succeeded", { sessionID: SESSION }),
4596 event("session.execution.started", { sessionID: CHILD_B }),
4597 // The first subagent's report wakes the parent into a reply T3 holds
4598 // for its continuation.
4599 event("session.execution.succeeded", { sessionID: CHILD }),
4600 event("session.inbox.enqueued", {
4601 inboxID: "msg_report",
4602 sessionID: SESSION,
4603 item: {
4604 type: "synthetic",
4605 payload: {
4606 text: `<subagent sessionID="${CHILD}" state="completed" description="Sleep">\nCHILD_OK\n</subagent>`,
4607 description: "Sleep",
4608 metadata: {
4609 source: "subagent",
4610 childID: CHILD,
4611 agent: "General",
4612 state: "completed",
4613 },
4614 },
4615 delivery: "steer",
4616 },
4617 }),
4618 event("session.execution.started", { sessionID: SESSION }),
4619 // The reply starts a background subagent of its own, named only by
4620 // the reply's held events.
4621 event("session.tool.input.started", { ...spawn, name: "subagent" }),
4622 event("session.tool.called", {
4623 ...spawn,
4624 name: "subagent",
4625 input: { description: "Sleep", prompt: "sleep", background: true },
4626 executed: false,
4627 }),
4628 event("session.created", { ...childCreated(SPAWNED), sessionID: SPAWNED }),
4629 event("session.tool.progress", {
4630 ...spawn,
4631 metadata: { sessionID: SPAWNED, status: "running" },
4632 }),
4633 event("session.execution.succeeded", { sessionID: SESSION }),
4634 // The second subagent's text, shown live, marks that the reply has ended.
4635 event("session.text.ended", {
4636 sessionID: CHILD_B,
4637 assistantMessageID: "msg_child_b",
4638 ordinal: 0,
4639 text: "B_WORKING",
4640 }),
4641 // The user's Stop stops the second subagent and drops the held reply.
4642 out("session.interrupt", { sessionID: CHILD_B }),
4643 reply("session.interrupt", { interrupted: true }),
4644 event("session.execution.interrupted", { sessionID: CHILD_B }),
4645 // The subagent only the dropped reply named is stopped too; that Stop
4646 // fails, so it runs on.
4647 out("session.interrupt", { sessionID: SPAWNED }),
4648 reply("session.interrupt", {
4649 status: 500,
4650 body: { _tag: "UnknownError", message: "interrupt failed" },
4651 }),
4652 // An unrelated prompt runs while that subagent, and a subagent it
4653 // starts, ask for permissions, which no turn shows: they are refused.
4654 out("session.prompt", { sessionID: SESSION, text: "<any>" }),
4655 promptAccepted,
4656 event("session.execution.started", { sessionID: SESSION }),
4657 event("permission.asked", { ...shellAsk.data, id: "per_spawned", sessionID: SPAWNED }),
4658 out("permission.reply", {
4659 sessionID: SPAWNED,
4660 requestID: "per_spawned",
4661 decision: "reject",
4662 }),
4663 reply("permission.reply", null),
4664 event("session.created", {
4665 ...childCreated(NESTED),
4666 sessionID: NESTED,
4667 parentID: SPAWNED,
4668 }),
4669 event("permission.asked", { ...shellAsk.data, id: "per_nested", sessionID: NESTED }),
4670 out("permission.reply", {
4671 sessionID: NESTED,
4672 requestID: "per_nested",
4673 decision: "reject",
4674 }),
4675 reply("permission.reply", null),
4676 event("session.text.ended", {
4677 sessionID: SESSION,
4678 assistantMessageID: "msg_unrelated",
4679 ordinal: 0,
4680 text: "UNRELATED",
4681 }),
4682 event("session.execution.succeeded", { sessionID: SESSION }),
4683 // The next Stop tries the subagent again, and the one it started, and reaches both.
4684 out("session.interrupt", { sessionID: SPAWNED }),
4685 reply("session.interrupt", { interrupted: true }),
4686 out("session.interrupt", { sessionID: NESTED }),
4687 reply("session.interrupt", { interrupted: true }),
4688 // OpenCode reports the stopped subagent to the thread's session and
4689 // wakes it; that execution answers no turn, so it is stopped and no
4690 // continuation is offered for it.
4691 event("session.execution.interrupted", { sessionID: SPAWNED }),
4692 event("session.inbox.enqueued", {
4693 inboxID: "msg_report_spawned",
4694 sessionID: SESSION,
4695 item: {
4696 type: "synthetic",
4697 payload: {
4698 text: `<subagent sessionID="${SPAWNED}" state="cancelled" description="Sleep">\n</subagent>`,
4699 description: "Sleep",
4700 metadata: {
4701 source: "subagent",
4702 childID: SPAWNED,
4703 agent: "General",
4704 state: "cancelled",
4705 },
4706 },
4707 delivery: "steer",
4708 },
4709 }),
4710 event("session.execution.started", { sessionID: SESSION }),
4711 out("session.interrupt", { sessionID: SESSION }),
4712 reply("session.interrupt", { interrupted: true }),
4713 event("session.execution.interrupted", { sessionID: SESSION }, "settled"),
4714 ],
4715 { replayGate: gate },
4716 ).pipe(
4717 Effect.provideService(ProviderContinuationRequests.ProviderContinuationRequests, {
4718 offer: (request) => Effect.sync(() => void offers.push(request)),
4719 take: Effect.never,
4720 }),
4721 );
4722 const collected: Array<ProviderAdapter.ProviderAdapterV2Event> = [];
4723 const replyEnded = yield* Deferred.make<void>();
4724 const bothEnded = yield* Deferred.make<void>();
4725 yield* runtime.events.pipe(
4726 Stream.tap((event) =>
4727 Effect.gen(function* () {
4728 collected.push(event);
4729 if (event.type === "turn_item.updated" && event.turnItem.type === "assistant_message") {
4730 if (event.turnItem.text === "B_WORKING")
4731 yield* Deferred.succeed(replyEnded, undefined);
4732 }
4733 if (collected.filter((entry) => entry.type === "turn.terminal").length === 2) {
4734 yield* Deferred.succeed(bothEnded, undefined);
4735 }
4736 }),
4737 ),
4738 Stream.runDrain,
4739 Effect.forkScoped,
4740 );
4741 yield* runtime.startTurn(withLineage(thread));
4742 yield* Deferred.await(replyEnded);
4743 assert.lengthOf(offers, 1);
4744 // The user stops the thread before the reply's continuation runs.
4745 yield* runtime.interruptTurn({
4746 providerThread: thread,
4747 providerTurnId: yield* providerTurnId,
4748 requestRuntimeRestart: true,
4749 });
4750 yield* runtime.startTurn({ ...secondTurn(thread), appThread: withLineage(thread).appThread });
4751 yield* Deferred.await(bothEnded);
4752 // A later Stop reaches both subagents; once stopped they are not tried
4753 // again (the replay fails on any further request).
4754 for (let stops = 0; stops < 2; stops++) {
4755 yield* runtime.interruptTurn({
4756 providerThread: thread,
4757 providerTurnId: yield* providerTurnId,
4758 requestRuntimeRestart: true,
4759 });
4760 }
4761 // The stopped subagent's report wakes nothing T3 runs.
4762 yield* Effect.promise(() => gate.waitForReached("settled"));
4763 gate.release("settled");
4764 assert.lengthOf(offers, 1);
4765 // The unrelated turn shows no request it never made.
4766 assert.deepEqual(
4767 collected.flatMap((event) =>
4768 event.type === "runtime_request.updated"
4769 ? [`${event.runtimeRequest.nativeRequestRef?.nativeId}:${event.runtimeRequest.status}`]
4770 : [],
4771 ),
4772 [],
4773 );
4774 }).pipe(Effect.scoped),
4775 );
4776
4777 it.effect(
4778 "keeps every subagent a stopped background reply's subagent started off the next turn",
4779 () =>
4780 Effect.gen(function* () {
4781 const CHILD_B = "ses_f1485c529ffeBBBBBBBBBBBBBB";
4782 const SPAWNED = "ses_f1485c529ffeDDDDDDDDDDDDDD";
4783 const NESTED = "ses_f1485c529ffeNNNNNNNNNNNNNN";
4784 const DEEP = "ses_f1485c529ffeEEEEEEEEEEEEEE";
4785 const toolB = { sessionID: SESSION, assistantMessageID: "msg_assistant", id: "call-b" };
4786 const spawn = { sessionID: SESSION, assistantMessageID: "msg_reply", id: "call-spawn" };
4787 const { runtime, thread } = yield* resumed([
4788 ...backgroundLaunch(CHILD),
4789 // The same turn starts a second background subagent, which runs on.
4790 event("session.tool.input.started", { ...toolB, name: "subagent" }),
4791 event("session.tool.called", {
4792 ...toolB,
4793 name: "subagent",
4794 input: { description: "Sleep", prompt: "sleep", background: true },
4795 executed: false,
4796 }),
4797 event("session.created", { ...childCreated(CHILD_B), sessionID: CHILD_B }),
4798 event("session.tool.progress", {
4799 ...toolB,
4800 metadata: { sessionID: CHILD_B, status: "running" },
4801 }),
4802 event("session.execution.succeeded", { sessionID: SESSION }),
4803 event("session.execution.started", { sessionID: CHILD_B }),
4804 // The first subagent's report wakes the parent into a reply T3 holds.
4805 event("session.execution.succeeded", { sessionID: CHILD }),
4806 event("session.inbox.enqueued", {
4807 inboxID: "msg_report",
4808 sessionID: SESSION,
4809 item: {
4810 type: "synthetic",
4811 payload: {
4812 text: `<subagent sessionID="${CHILD}" state="completed" description="Sleep">\nCHILD_OK\n</subagent>`,
4813 description: "Sleep",
4814 metadata: {
4815 source: "subagent",
4816 childID: CHILD,
4817 agent: "General",
4818 state: "completed",
4819 },
4820 },
4821 delivery: "steer",
4822 },
4823 }),
4824 event("session.execution.started", { sessionID: SESSION }),
4825 // The held reply starts a subagent, which starts one, which starts another.
4826 event("session.tool.input.started", { ...spawn, name: "subagent" }),
4827 event("session.tool.called", {
4828 ...spawn,
4829 name: "subagent",
4830 input: { description: "Sleep", prompt: "sleep", background: true },
4831 executed: false,
4832 }),
4833 event("session.created", { ...childCreated(SPAWNED), sessionID: SPAWNED }),
4834 event("session.tool.progress", {
4835 ...spawn,
4836 metadata: { sessionID: SPAWNED, status: "running" },
4837 }),
4838 event("session.created", {
4839 ...childCreated(NESTED),
4840 sessionID: NESTED,
4841 parentID: SPAWNED,
4842 }),
4843 event("session.created", { ...childCreated(DEEP), sessionID: DEEP, parentID: NESTED }),
4844 event("session.execution.succeeded", { sessionID: SESSION }),
4845 // The second subagent's text, shown live, marks that the reply has ended.
4846 event("session.text.ended", {
4847 sessionID: CHILD_B,
4848 assistantMessageID: "msg_child_b",
4849 ordinal: 0,
4850 text: "B_WORKING",
4851 }),
4852 // The user's Stop stops the second subagent and drops the held reply,
4853 // then each subagent under the reply, the deepest of which runs on.
4854 out("session.interrupt", { sessionID: CHILD_B }),
4855 reply("session.interrupt", { interrupted: true }),
4856 event("session.execution.interrupted", { sessionID: CHILD_B }),
4857 out("session.interrupt", { sessionID: SPAWNED }),
4858 reply("session.interrupt", { interrupted: true }),
4859 out("session.interrupt", { sessionID: NESTED }),
4860 reply("session.interrupt", { interrupted: true }),
4861 out("session.interrupt", { sessionID: DEEP }),
4862 reply("session.interrupt", {
4863 status: 500,
4864 body: { _tag: "UnknownError", message: "interrupt failed" },
4865 }),
4866 // An unrelated prompt runs while the deepest asks for a permission,
4867 // which no turn shows: it is refused.
4868 out("session.prompt", { sessionID: SESSION, text: "<any>" }),
4869 promptAccepted,
4870 event("session.execution.started", { sessionID: SESSION }),
4871 event("permission.asked", { ...shellAsk.data, id: "per_deep", sessionID: DEEP }),
4872 out("permission.reply", { sessionID: DEEP, requestID: "per_deep", decision: "reject" }),
4873 reply("permission.reply", null),
4874 event("session.execution.succeeded", { sessionID: SESSION }),
4875 ]);
4876 const collected: Array<ProviderAdapter.ProviderAdapterV2Event> = [];
4877 const replyEnded = yield* Deferred.make<void>();
4878 const bothEnded = yield* Deferred.make<void>();
4879 yield* runtime.events.pipe(
4880 Stream.tap((event) =>
4881 Effect.gen(function* () {
4882 collected.push(event);
4883 if (
4884 event.type === "turn_item.updated" &&
4885 event.turnItem.type === "assistant_message" &&
4886 event.turnItem.text === "B_WORKING"
4887 ) {
4888 yield* Deferred.succeed(replyEnded, undefined);
4889 }
4890 if (collected.filter((entry) => entry.type === "turn.terminal").length === 2) {
4891 yield* Deferred.succeed(bothEnded, undefined);
4892 }
4893 }),
4894 ),
4895 Stream.runDrain,
4896 Effect.forkScoped,
4897 );
4898 yield* runtime.startTurn(withLineage(thread));
4899 yield* Deferred.await(replyEnded);
4900 yield* runtime.interruptTurn({
4901 providerThread: thread,
4902 providerTurnId: yield* providerTurnId,
4903 requestRuntimeRestart: true,
4904 });
4905 yield* runtime.startTurn({
4906 ...secondTurn(thread),
4907 appThread: withLineage(thread).appThread,
4908 });
4909 yield* Deferred.await(bothEnded);
4910 // The unrelated turn shows no request it never made.
4911 assert.deepEqual(
4912 collected.flatMap((event) =>
4913 event.type === "runtime_request.updated"
4914 ? [
4915 `${event.runtimeRequest.nativeRequestRef?.nativeId}:${event.runtimeRequest.status}`,
4916 ]
4917 : [],
4918 ),
4919 [],
4920 );
4921 }).pipe(Effect.scoped),
4922 );
4923
4924 it.effect("ends a continuation while a skipped reply's subagent does not answer its Stop", () =>
4925 Effect.gen(function* () {
4926 const CHILD_B = "ses_f1485c529ffeBBBBBBBBBBBBBB";
4927 const SPAWNED = "ses_f1485c529ffeDDDDDDDDDDDDDD";
4928 const toolB = { sessionID: SESSION, assistantMessageID: "msg_assistant", id: "call-b" };
4929 const spawn = { sessionID: SESSION, assistantMessageID: "msg_reply_a", id: "call-spawn" };
4930 const report = (inboxID: string, child: string) =>
4931 event("session.inbox.enqueued", {
4932 inboxID,
4933 sessionID: SESSION,
4934 item: {
4935 type: "synthetic",
4936 payload: {
4937 text: `<subagent sessionID="${child}" state="completed" description="Sleep">\nOK\n</subagent>`,
4938 description: "Sleep",
4939 metadata: {
4940 source: "subagent",
4941 childID: child,
4942 agent: "General",
4943 state: "completed",
4944 },
4945 },
4946 delivery: "steer",
4947 },
4948 });
4949 const offers: Array<ProviderContinuationRequests.ProviderContinuationRequest> = [];
4950 const bothOffered = yield* Deferred.make<void>();
4951 const { runtime, thread } = yield* resumed([
4952 ...backgroundLaunch(CHILD),
4953 event("session.tool.input.started", { ...toolB, name: "subagent" }),
4954 event("session.tool.called", {
4955 ...toolB,
4956 name: "subagent",
4957 input: { description: "Sleep", prompt: "sleep", background: true },
4958 executed: false,
4959 }),
4960 event("session.created", { ...childCreated(CHILD_B), sessionID: CHILD_B }),
4961 event("session.tool.progress", {
4962 ...toolB,
4963 metadata: { sessionID: CHILD_B, status: "running" },
4964 }),
4965 event("session.execution.succeeded", { sessionID: SESSION }),
4966 // Reply A starts a subagent of its own, named only by its held events.
4967 event("session.execution.succeeded", { sessionID: CHILD }),
4968 report("msg_report_a", CHILD),
4969 event("session.execution.started", { sessionID: SESSION }),
4970 event("session.tool.input.started", { ...spawn, name: "subagent" }),
4971 event("session.tool.called", {
4972 ...spawn,
4973 name: "subagent",
4974 input: { description: "Sleep", prompt: "sleep", background: true },
4975 executed: false,
4976 }),
4977 event("session.created", { ...childCreated(SPAWNED), sessionID: SPAWNED }),
4978 event("session.execution.succeeded", { sessionID: SESSION }),
4979 event("session.execution.succeeded", { sessionID: CHILD_B }),
4980 report("msg_report_b", CHILD_B),
4981 event("session.execution.started", { sessionID: SESSION }),
4982 event("session.text.ended", {
4983 sessionID: SESSION,
4984 assistantMessageID: "msg_reply_b",
4985 ordinal: 0,
4986 text: "REPLY_B",
4987 }),
4988 event("session.execution.succeeded", { sessionID: SESSION }),
4989 // B's continuation skips reply A and stops its subagent, which never answers.
4990 out("session.interrupt", { sessionID: SPAWNED }),
4991 reply("session.interrupt", "<hang>"),
4992 ]).pipe(
4993 Effect.provideService(ProviderContinuationRequests.ProviderContinuationRequests, {
4994 offer: (request) =>
4995 Effect.suspend(() =>
4996 offers.push(request) === 2
4997 ? Deferred.succeed(bothOffered, undefined).pipe(Effect.asVoid)
4998 : Effect.void,
4999 ),
5000 take: Effect.never,
5001 }),
5002 );
5003 const ended = yield* terminals(runtime, 2);
5004 yield* runtime.startTurn(withLineage(thread));
5005 yield* Deferred.await(bothOffered);
5006 yield* runtime.startTurn(continuationTurn(thread, offers[1]!, "wake-b"));
5007 // B's reply replays and ends its turn while that Stop still waits.
5008 assert.deepEqual(
5009 [...(yield* Fiber.join(ended))].map((terminal) => terminal.status),
5010 ["completed", "completed"],
5011 );
5012 }).pipe(Effect.scoped),
5013 );
5014
5015 it.effect("replays only its own background reply into a continuation turn", () =>
5016 Effect.gen(function* () {
5017 const CHILD_B = "ses_f1485c529ffeBBBBBBBBBBBBBB";
5018 const toolB = { sessionID: SESSION, assistantMessageID: "msg_assistant", id: "call-b" };
5019 const report = (inboxID: string, child: string, text: string) =>
5020 event("session.inbox.enqueued", {
5021 inboxID,
5022 sessionID: SESSION,
5023 item: {
5024 type: "synthetic",
5025 payload: {
5026 text: `<subagent sessionID="${child}" state="completed" description="Sleep">\n${text}\n</subagent>`,
5027 description: "Sleep",
5028 metadata: {
5029 source: "subagent",
5030 childID: child,
5031 agent: "General",
5032 state: "completed",
5033 },
5034 },
5035 delivery: "steer",
5036 },
5037 });
5038 const reply = (assistantMessageID: string, text: string) =>
5039 event("session.text.ended", { sessionID: SESSION, assistantMessageID, ordinal: 0, text });
5040 const offers: Array<ProviderContinuationRequests.ProviderContinuationRequest> = [];
5041 const bothOffered = yield* Deferred.make<void>();
5042 const { runtime, thread } = yield* resumed([
5043 ...backgroundLaunch(CHILD),
5044 // The same turn starts a second background subagent.
5045 event("session.tool.input.started", { ...toolB, name: "subagent" }),
5046 event("session.tool.called", {
5047 ...toolB,
5048 name: "subagent",
5049 input: { description: "Sleep", prompt: "sleep", background: true },
5050 executed: false,
5051 }),
5052 event("session.created", { ...childCreated(CHILD_B), sessionID: CHILD_B }),
5053 event("session.tool.progress", {
5054 ...toolB,
5055 metadata: { sessionID: CHILD_B, status: "running" },
5056 }),
5057 event("session.execution.succeeded", { sessionID: SESSION }),
5058 // Each subagent's report wakes the parent into a reply OpenCode runs on its own.
5059 event("session.execution.succeeded", { sessionID: CHILD }),
5060 report("msg_report_a", CHILD, "A_OK"),
5061 event("session.execution.started", { sessionID: SESSION }),
5062 event("session.inbox.delivered", { sessionID: SESSION, inboxID: "msg_report_a" }),
5063 reply("msg_reply_a", "REPLY_A"),
5064 event("session.execution.succeeded", { sessionID: SESSION }),
5065 event("session.execution.succeeded", { sessionID: CHILD_B }),
5066 report("msg_report_b", CHILD_B, "B_OK"),
5067 event("session.execution.started", { sessionID: SESSION }),
5068 event("session.inbox.delivered", { sessionID: SESSION, inboxID: "msg_report_b" }),
5069 reply("msg_reply_b", "REPLY_B"),
5070 event("session.execution.succeeded", { sessionID: SESSION }),
5071 ]).pipe(
5072 Effect.provideService(ProviderContinuationRequests.ProviderContinuationRequests, {
5073 offer: (request) =>
5074 Effect.suspend(() =>
5075 offers.push(request) === 2
5076 ? Deferred.succeed(bothOffered, undefined).pipe(Effect.asVoid)
5077 : Effect.void,
5078 ),
5079 take: Effect.never,
5080 }),
5081 );
5082 const collected: Array<ProviderAdapter.ProviderAdapterV2Event> = [];
5083 const ended = yield* Deferred.make<void>();
5084 yield* runtime.events.pipe(
5085 Stream.tap((event) =>
5086 Effect.gen(function* () {
5087 collected.push(event);
5088 if (collected.filter((entry) => entry.type === "turn.terminal").length === 2) {
5089 yield* Deferred.succeed(ended, undefined);
5090 }
5091 }),
5092 ),
5093 Stream.runDrain,
5094 Effect.forkScoped,
5095 );
5096 yield* runtime.startTurn(withLineage(thread));
5097 yield* Deferred.await(bothOffered);
5098 // A's continuation was cancelled before it started; B's runs.
5099 yield* runtime.startTurn(continuationTurn(thread, offers[1]!, "wake-b"));
5100 yield* Deferred.await(ended);
5101 const continuationId = (yield* IdAllocator.IdAllocatorV2).derive.providerTurn({
5102 driver: OPENCODE_PROVIDER,
5103 nativeTurnId: `${SESSION}:attempt:attempt:opencode2-adapter:wake-b`,
5104 });
5105 assert.deepEqual(
5106 collected.flatMap((event) =>
5107 event.type === "turn_item.updated" &&
5108 event.turnItem.type === "assistant_message" &&
5109 event.turnItem.providerTurnId === continuationId
5110 ? [event.turnItem.text]
5111 : [],
5112 ),
5113 ["REPLY_B"],
5114 );
5115 // Nothing is left held for a continuation that will never run.
5116 assert.isFalse(yield* runtime.hasPendingBackgroundWork!);
5117 }).pipe(Effect.scoped, Effect.provide(IdAllocator.layer)),
5118 );
5119
44605120 it.effect("ends a follow-up turn a steer joined exactly once", () =>
44615121 Effect.gen(function* () {
44625122 const reportText = `<subagent sessionID="${CHILD}" state="completed" description="Sleep">\nCHILD_OK\n</subagent>`;
......@@ -4534,19 +5194,7 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => {
45345194 yield* Effect.gen(function* () {
45355195 while (offers.length === 0) yield* Effect.yieldNow;
45365196 }).pipe(Effect.timeout("2 seconds"), Effect.orDie);
4537 const followUp = {
4538 ...withLineage(thread),
4539 runId: RunId.make("run:opencode2-adapter:wake"),
4540 runOrdinal: 2,
4541 providerTurnOrdinal: 2,
4542 attemptId: RunAttemptId.make("attempt:opencode2-adapter:wake"),
4543 message: {
4544 ...turnInput(thread).message,
4545 messageId: MessageId.make("message:opencode2-adapter:wake"),
4546 createdBy: "agent" as const,
4547 creationSource: "provider" as const,
4548 },
4549 };
5197 const followUp = continuationTurn(thread, offers[0]!);
45505198 yield* runtime.startTurn(followUp);
45515199 const followUpTurnId = (yield* IdAllocator.IdAllocatorV2).derive.providerTurn({
45525200 driver: OPENCODE_PROVIDER,
apps/server/src/orchestration-v2/ProviderContinuationService.test.ts+19
......@@ -180,6 +180,25 @@ describe("ProviderContinuationService", () => {
180180 });
181181 });
182182
183 it.effect("dispatches a wake under the message its adapter named", () => {
184 return Effect.gen(function* () {
185 const dispatched = yield* Queue.unbounded<unknown>();
186 const messageId = MessageId.make("message-provider-continuation-adapter-named");
187 yield* Effect.gen(function* () {
188 const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
189 yield* requests.offer({ ...request(), messageId });
190 const command = (yield* Queue.take(dispatched)) as { readonly messageId: MessageId };
191 // The adapter matches the continuation run's message to the wake it holds.
192 assert.equal(command.messageId, messageId);
193 }).pipe(
194 Effect.provide(
195 layerTest({ dispatched, getThreadRecords: () => Effect.succeed(projection) }),
196 ),
197 Effect.scoped,
198 );
199 });
200 });
201
183202 it.effect("delivers a message_text wake as a real prompt, not a buffered wake", () => {
184203 return Effect.gen(function* () {
185204 const dispatched = yield* Queue.unbounded<unknown>();
apps/server/src/orchestration-v2/ProviderContinuationService.ts+6-4
......@@ -147,10 +147,12 @@ export const layer = Layer.effectDiscard(
147147 }
148148 // The ordinal is display metadata only: allocate.message appends a
149149 // random UUID, so a stale projection read here cannot collide ids.
150 const messageId = yield* ids.allocate.message({
151 threadId: request.threadId,
152 ordinal: projection.messages.length + 1,
153 });
150 const messageId =
151 request.messageId ??
152 (yield* ids.allocate.message({
153 threadId: request.threadId,
154 ordinal: projection.messages.length + 1,
155 }));
154156 const commandId = CommandId.make(`provider-continuation:${messageId}`);
155157 const dispatch = threads.dispatch({
156158 type: "message.dispatch",
packages/provider-core/src/server/ProviderContinuationRequests.ts+6
......@@ -41,6 +41,12 @@ export interface ProviderContinuationRequest {
4141 * reach the provider as a real prompt.
4242 */
4343 readonly delivery?: "adapter_buffered" | "message_text";
44 /**
45 * The message the continuation run starts from, when the adapter picks it:
46 * its turn input carries this id, so the adapter can tell which of its held
47 * wakes a run was dispatched for. Otherwise the service allocates one.
48 */
49 readonly messageId?: MessageId;
4450 readonly dispatchIfCurrent?: <A, E, R>(
4551 effect: Effect.Effect<A, E, R>,
4652 ) => Effect.Effect<Option.Option<A>, E, R>;
packages/provider-opencode/src/server/v2/adapter.ts+130-8
......@@ -49,6 +49,7 @@ import {
4949 type OrchestrationV2Subagent,
5050 type OrchestrationV2TurnItem,
5151 type OrchestrationV2UserInputQuestion,
52 type MessageId,
5253 type ModelSelection,
5354 type ProviderApprovalDecision,
5455 type ProviderInstanceId,
......@@ -305,6 +306,8 @@ interface SubagentCall {
305306 * the continuation turn T3 opens for it takes them.
306307 */
307308interface Wake {
309 /** The message its continuation run starts from, which names the wake in that turn's input. */
310 readonly messageId: MessageId | undefined;
308311 readonly events: Array<OpenCode2Client.OpenCode2StreamEvent>;
309312 running: boolean;
310313 /** The background subagents whose end it answers, as its turn's notification names them. */
......@@ -317,6 +320,11 @@ interface Wake {
317320 readonly after: string | undefined;
318321 /** The first report it delivered, where fork and rollback cut before its turn. */
319322 readonly first: string | undefined;
323 /**
324 * Subagent sessions OpenCode announced while it was held: their calls are
325 * in its events, so only its replay names them.
326 */
327 readonly children: Set<string>;
320328}
321329
322330interface OpenBlock {
......@@ -907,6 +915,11 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr
907915 const childOwners = new Map<string, ThreadState>();
908916 // What OpenCode announced about a subagent's session, until its call names it.
909917 const announced = new Map<string, EventOf<"session.created">["data"]>();
918 // Subagent sessions a dropped held execution started, and the subagents
919 // they start, by the session the first of them was called from. Nothing
920 // names them, so they no longer route to its thread; they stay here until
921 // a Stop of that thread stops them or they are gone.
922 const strays = new Map<string, ThreadState>();
910923 // Sessions of these threads with an execution running, seen on the stream.
911924 const busy = new Set<string>();
912925 const pending = new Map<RuntimeRequestId, PendingRequest>();
......@@ -1687,13 +1700,14 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr
16871700 */
16881701 const offerWake = Effect.fnUntraced(function* (state: ThreadState, wake: Wake) {
16891702 const route = state.providerThread.appThreadId;
1690 if (route === null) return;
1703 if (route === null || wake.messageId === undefined) return;
16911704 const notification = backgroundWorkNotification(wake.reports);
16921705 yield* continuationRequests.offer({
16931706 threadId: route,
16941707 providerThreadId: state.providerThread.id,
16951708 driver,
16961709 detail: wake.detail,
1710 messageId: wake.messageId,
16971711 ...(notification === null ? {} : { notification }),
16981712 dispatchIfCurrent: (dispatch) =>
16991713 wake.dropped ? Effect.succeed(Option.none()) : Effect.map(dispatch, Option.some),
......@@ -2153,6 +2167,15 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr
21532167 | { readonly type: "form"; readonly form: NativeForm },
21542168 ) {
21552169 const asking = asked.type === "permission" ? asked.request.sessionID : asked.form.sessionID;
2170 // A stray belongs to no turn: its request is refused, which ends its run.
2171 if (strays.has(asking)) {
2172 return yield* stopStaleRequest(
2173 asking,
2174 asked.type === "permission"
2175 ? { type: "permission", id: asked.request.id }
2176 : { type: "form", id: asked.form.id },
2177 );
2178 }
21562179 const state = ownerOf(asking);
21572180 if (state === undefined) return;
21582181 const target = requestTurn(asking);
......@@ -2381,7 +2404,13 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr
23812404 const onWake = Effect.fnUntraced(function* (state: ThreadState) {
23822405 const { delivered, stopped } = yield* takeReports(state);
23832406 if (stopped) return;
2407 const route = state.providerThread.appThreadId;
23842408 const wake: Wake = {
2409 // Only a thread's wake is offered a continuation; the ordinal is display metadata.
2410 messageId:
2411 route === null
2412 ? undefined
2413 : yield* idAllocator.allocate.message({ threadId: route, ordinal: 0 }),
23852414 events: [],
23862415 running: true,
23872416 reports: delivered.map((entry) => entry.report),
......@@ -2389,6 +2418,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr
23892418 dropped: false,
23902419 after: delivered.at(-1)?.inboxId,
23912420 first: delivered[0]?.inboxId,
2421 children: new Set(),
23922422 };
23932423 state.wakes.push(wake);
23942424 yield* Effect.logInfo("OpenCode started a turn on its own; asking for a continuation.", {
......@@ -2408,6 +2438,8 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr
24082438 ) {
24092439 const childId = stringField(payload.metadata, "childID");
24102440 if (stringField(payload.metadata, "source") !== "subagent" || childId === undefined) return;
2441 // A stray that reports has ended: no Stop needs to reach it.
2442 strays.delete(childId);
24112443 const call = [...state.calls.values()].find(
24122444 (candidate) => candidate.child?.sessionId === childId,
24132445 );
......@@ -2513,10 +2545,17 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr
25132545 }
25142546 // A subagent's session: named on its parent's `subagent` call just after.
25152547 if (event.type === "session.created" && event.data.parentID !== undefined) {
2548 // A stray's subagent is a stray too: no call of the thread's names it.
2549 const strayCaller = strays.get(event.data.parentID);
2550 if (strayCaller !== undefined) {
2551 strays.set(event.data.sessionID, strayCaller);
2552 return;
2553 }
25162554 const owner = ownerOf(event.data.parentID);
25172555 if (owner !== undefined) {
25182556 childOwners.set(event.data.sessionID, owner);
25192557 announced.set(event.data.sessionID, event.data);
2558 holderOf(event.data.parentID)?.wakes.at(-1)?.children.add(event.data.sessionID);
25202559 }
25212560 return;
25222561 }
......@@ -2806,7 +2845,15 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr
28062845 (state) => state.subagent === undefined && hasBackground(state),
28072846 );
28082847 const marked = [...threads.values()].some((state) => state.unsettled);
2809 if (running.length === 0 && background.length === 0 && busy.size === 0 && !marked) return;
2848 if (
2849 running.length === 0 &&
2850 background.length === 0 &&
2851 busy.size === 0 &&
2852 !marked &&
2853 strays.size === 0
2854 ) {
2855 return;
2856 }
28102857 const active = yield* client.session.active();
28112858 // An execution that ended while the stream was down never said so.
28122859 for (const sessionId of busy) if (!(sessionId in active)) busy.delete(sessionId);
......@@ -2817,6 +2864,8 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr
28172864 for (const state of threads.values()) {
28182865 if (state.unsettled && !(state.sessionId in active)) state.unsettledStarted = true;
28192866 }
2867 // A stray that stopped meanwhile runs nothing to stop.
2868 for (const childId of strays.keys()) if (!(childId in active)) strays.delete(childId);
28202869 for (const [sessionId, state] of running) {
28212870 const turn = state.active;
28222871 if (turn === undefined) continue;
......@@ -2848,10 +2897,14 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr
28482897 }
28492898 // A subagent's session holds reports of its own background subagents.
28502899 for (const session of sessionsOf(state)) {
2851 for (const wake of session.wakes.splice(0)) wake.dropped = true;
2900 discardWakes(session, session.wakes.splice(0));
28522901 session.reports.clear();
28532902 }
28542903 }
2904 // Stopped off the lock this reconcile holds; one still running waits for a Stop.
2905 for (const thread of new Set([...strays.values()].map(rootOf))) {
2906 yield* Effect.forkIn(stopStrays(thread), sessionScope);
2907 }
28552908 // A request asked while the stream was down was never shown, and the
28562909 // run waits on it; one T3 shows that OpenCode no longer lists was
28572910 // answered elsewhere or dropped with its execution. A session that
......@@ -3442,6 +3495,62 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr
34423495 for (const event of wake.events) yield* route(event, sessionOfEvent(event), true);
34433496 });
34443497
3498 /**
3499 * Drops held executions no turn will take. The subagents one started are
3500 * named only by its replay, so they stop routing to the thread at once and
3501 * become strays for {@link stopStrays}. A subagent that registered its own
3502 * call outside the held execution is the thread's, and is left alone.
3503 */
3504 const discardWakes = (caller: ThreadState, wakes: ReadonlyArray<Wake>) => {
3505 const retired = new Set<string>();
3506 const retire = (childId: string) => {
3507 const info = announced.get(childId);
3508 if (info === undefined) return;
3509 announced.delete(childId);
3510 childOwners.delete(childId);
3511 strays.set(childId, caller);
3512 retired.add(childId);
3513 // OpenCode reports its end to the session that called it; that
3514 // report answers no turn, so the execution it wakes is stopped. A
3515 // deeper one reports to a stray, which T3 does not track.
3516 threads.get(info.parentID ?? "")?.stoppedChildren.add(childId);
3517 };
3518 for (const wake of wakes) {
3519 wake.dropped = true;
3520 for (const childId of wake.children) retire(childId);
3521 }
3522 // Their own subagents, at any depth, were announced live but no call
3523 // of the thread's names them either.
3524 for (let grew = retired.size > 0; grew;) {
3525 grew = false;
3526 for (const [childId, info] of announced) {
3527 if (info.parentID === undefined || !retired.has(info.parentID)) continue;
3528 retire(childId);
3529 grew = true;
3530 }
3531 }
3532 };
3533
3534 /**
3535 * Stops a thread's strays one by one, as a Stop stops background
3536 * subagents, and never under `lock`: one that does not answer would
3537 * otherwise hold up every event. A stray is forgotten once stopped or
3538 * gone; one still running waits for the next Stop.
3539 */
3540 const stopStrays = (thread: ThreadState) =>
3541 Effect.forEach(
3542 [...strays].flatMap(([childId, caller]) => (rootOf(caller) === thread ? [childId] : [])),
3543 (childId) =>
3544 client.session.interrupt({ sessionID: Session.ID.make(childId) }).pipe(
3545 Effect.asVoid,
3546 Effect.catchTags({ SessionNotFoundError: () => Effect.void }),
3547 Effect.timeout(INTERRUPT_TIMEOUT),
3548 Effect.tap(() => Effect.sync(() => strays.delete(childId))),
3549 Effect.ignore({ log: true }),
3550 ),
3551 { discard: true },
3552 );
3553
34453554 /**
34463555 * A user turn that starts while OpenCode runs an execution on its own: the
34473556 * prompt joins that execution, so the turn takes it, and its continuation
......@@ -3456,8 +3565,11 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr
34563565
34573566 /**
34583567 * The continuation turn for an execution OpenCode started on its own: it
3459 * takes the oldest one held and ends with it. One already taken (a user
3460 * turn joined it) leaves nothing to run, so the turn ends at once.
3568 * takes the wake it was dispatched for, named by its message, and ends
3569 * with it. Wakes held before that one had their continuation cancelled or
3570 * dropped before it started, so nothing will replay them: they are
3571 * dropped. One already taken (a user turn joined it) leaves nothing to
3572 * run, so the turn ends at once.
34613573 */
34623574 const runWake = Effect.fnUntraced(function* (
34633575 state: ThreadState,
......@@ -3466,8 +3578,16 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr
34663578 yield* beginTurn(state, turnInput, false);
34673579 yield* lock.withPermit(
34683580 Effect.gen(function* () {
3469 const wake = state.wakes.shift();
3470 if (wake === undefined) return yield* finishTurn(state, { status: "completed" });
3581 const index = state.wakes.findIndex(
3582 (held) => held.messageId === turnInput.message.messageId,
3583 );
3584 if (index < 0) return yield* finishTurn(state, { status: "completed" });
3585 // Queued continuations can't be reordered (they carry a notification),
3586 // so an earlier wake here lost its continuation to a cancel or a failed
3587 // dispatch. Dropping it keeps that wake from pinning the session.
3588 discardWakes(state, state.wakes.splice(0, index));
3589 yield* Effect.forkIn(stopStrays(state), sessionScope);
3590 const wake = state.wakes.shift()!;
34713591 const turn = state.active;
34723592 if (turn !== undefined && wake.after !== undefined) turn.before = wake.after;
34733593 // Its first report is where its history begins, so fork and rollback cut there.
......@@ -3557,7 +3677,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr
35573677 }
35583678 // A held wake no turn will take: its execution is stopped, not replayed.
35593679 const running = state.wakes.some((wake) => wake.running);
3560 for (const wake of state.wakes.splice(0)) wake.dropped = true;
3680 discardWakes(state, state.wakes.splice(0));
35613681 if (running) {
35623682 markUnsettled(state);
35633683 yield* client.session
......@@ -3566,6 +3686,8 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr
35663686 }
35673687 }),
35683688 );
3689 // Subagents of held executions this or an earlier Stop dropped.
3690 yield* stopStrays(state);
35693691 });
35703692
35713693 /**