authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-10-10 22:35:03-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-10 22:35:03-07:00
logcb7d0739389a602d81000bfb5aeb6ebea9e853bb
tree1d2a034c0caf62adabf7e4f1ca7785a02945d754
parentb4542c5418e727db5c5b578e3dc2b39e915fd0bc
signature Signed by PGP key B5690EEEBB952194

fix(server): OpenCode 2 continuations end after a reconnect took a Stop's end (#14744)

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

3 files changed, 409 insertions(+), 16 deletions(-)

apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts+353-3
...@@ -48,6 +48,10 @@ import {...@@ -48,6 +48,10 @@ import {
48 t3McpServerName,48 t3McpServerName,
49} from "@t3tools/provider-opencode/testing";49} from "@t3tools/provider-opencode/testing";
50import { openCode2ReplayRuntime } from "./OpenCode2AdapterV2.testkit.ts";50import { openCode2ReplayRuntime } from "./OpenCode2AdapterV2.testkit.ts";
51import {
52 makeProviderReplayGate,
53 type ProviderReplayGate,
54} from "@t3tools/provider-testing/replayGate";
5155
52const SESSION = "ses_f148ca2deffeJcwCnRQtb0YFNX";56const SESSION = "ses_f148ca2deffeJcwCnRQtb0YFNX";
53const WORK = "/work/opencode2";57const WORK = "/work/opencode2";
...@@ -64,12 +68,18 @@ const reply = (operation: string, data: unknown): ProviderReplayEntry => ({...@@ -64,12 +68,18 @@ const reply = (operation: string, data: unknown): ProviderReplayEntry => ({
64 frame: { type: "sdk.response", operation, data },68 frame: { type: "sdk.response", operation, data },
65});69});
66const replyData = (operation: string, data: unknown) => reply(operation, { data });70const replyData = (operation: string, data: unknown) => reply(operation, { data });
67const event = (type: string, data: Record<string, unknown>): ProviderReplayEntry => ({71const event = (
72 type: string,
73 data: Record<string, unknown>,
74 /** Holds the event until a test's replay gate releases this label. */
75 label?: string,
76): ProviderReplayEntry => ({
68 type: "emit_inbound",77 type: "emit_inbound",
69 frame: {78 frame: {
70 type: "sdk.event",79 type: "sdk.event",
71 event: { id: `evt_${type.replaceAll(".", "")}0000`, created: 1, type, data, ...durable },80 event: { id: `evt_${type.replaceAll(".", "")}0000`, created: 1, type, data, ...durable },
72 },81 },
82 ...(label === undefined ? {} : { label }),
73});83});
74const durable = { durable: { aggregateID: SESSION, seq: 1, version: 1 } };84const durable = { durable: { aggregateID: SESSION, seq: 1, version: 1 } };
7585
...@@ -316,7 +326,11 @@ const colorForm = {...@@ -316,7 +326,11 @@ const colorForm = {
316 */326 */
317const resumed = (327const resumed = (
318 entries: ReadonlyArray<ProviderReplayEntry>,328 entries: ReadonlyArray<ProviderReplayEntry>,
319 options?: { readonly external?: boolean; readonly supervised?: boolean },329 options?: {
330 readonly external?: boolean;
331 readonly supervised?: boolean;
332 readonly replayGate?: ProviderReplayGate;
333 },
320) =>334) =>
321 Effect.gen(function* () {335 Effect.gen(function* () {
322 const runtime = yield* openCode2ReplayRuntime(336 const runtime = yield* openCode2ReplayRuntime(
...@@ -335,7 +349,10 @@ const resumed = (...@@ -335,7 +349,10 @@ const resumed = (
335 : []),349 : []),
336 ...entries,350 ...entries,
337 ]),351 ]),
338 options?.external === undefined ? undefined : { external: options.external },352 {
353 ...(options?.external === undefined ? {} : { external: options.external }),
354 ...(options?.replayGate === undefined ? {} : { replayGate: options.replayGate }),
355 },
339 );356 );
340 const thread = yield* runtime.resumeThread({357 const thread = yield* runtime.resumeThread({
341 providerThread: providerThread(yield* DateTime.now),358 providerThread: providerThread(yield* DateTime.now),
...@@ -1724,6 +1741,88 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => {...@@ -1724,6 +1741,88 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => {
1724 }).pipe(Effect.scoped, Effect.provide(TestClock.layer())),1741 }).pipe(Effect.scoped, Effect.provide(TestClock.layer())),
1725 );1742 );
17261743
1744 it.effect("still checks the server when a failed prompt's execution starts late", () =>
1745 Effect.gen(function* () {
1746 const offered = yield* Deferred.make<void>();
1747 const { runtime, thread } = yield* resumed([
1748 out("session.prompt", { sessionID: SESSION, text: "<any>" }),
1749 // The request failed, but the server took the prompt: its execution
1750 // starts only after the turn already failed.
1751 reply("session.prompt", {
1752 status: 502,
1753 body: { _tag: "UnknownError", message: "bad gateway" },
1754 }),
1755 event("session.execution.started", { sessionID: SESSION }),
1756 // That run is still going, so the next turn stops it and fails.
1757 out("session.active"),
1758 reply("session.active", { data: { [SESSION]: { type: "running" } } }),
1759 out("session.interrupt", { sessionID: SESSION }),
1760 reply("session.interrupt", { interrupted: true }),
1761 ]).pipe(
1762 Effect.provideService(ProviderContinuationRequests.ProviderContinuationRequests, {
1763 offer: () => Deferred.succeed(offered, undefined).pipe(Effect.asVoid),
1764 take: Effect.never,
1765 }),
1766 );
1767 const ended = yield* terminals(runtime, 2);
1768 yield* runtime.startTurn(turnInput(thread)).pipe(Effect.ignore);
1769 // The late start has been read.
1770 yield* Deferred.await(offered);
1771 yield* runtime.startTurn(secondTurn(thread));
1772 const [first, second] = yield* Fiber.join(ended);
1773 assert.equal(first?.status, "failed");
1774 assert.equal(second?.failure?.message, OPENCODE_2_STILL_STOPPING);
1775 }).pipe(Effect.scoped, Effect.provide(TestClock.layer())),
1776 );
1777
1778 it.effect("still checks the server when a failed prompt starts after a reconnect", () =>
1779 Effect.gen(function* () {
1780 const offered = yield* Deferred.make<void>();
1781 const gate = makeProviderReplayGate(["drop"]);
1782 const { runtime, thread } = yield* resumed(
1783 [
1784 out("session.prompt", { sessionID: SESSION, text: "<any>" }),
1785 reply("session.prompt", {
1786 status: 502,
1787 body: { _tag: "UnknownError", message: "bad gateway" },
1788 }),
1789 // The stream drops before the prompt's execution starts, and the
1790 // reconnect finds the session active: the prompt's run is still to come.
1791 event("session.usage.updated", { sessionID: SESSION }, "drop"),
1792 { type: "runtime_exit", status: "success" } as const,
1793 out("event.subscribe"),
1794 out("session.active"),
1795 replyData("session.active", { [SESSION]: { type: "running" } }),
1796 out("permission.list", { sessionID: SESSION }),
1797 replyData("permission.list", []),
1798 out("session.form.list", { sessionID: SESSION }),
1799 replyData("session.form.list", []),
1800 // Its own start: that run is still going, so the next turn stops it and fails.
1801 event("session.execution.started", { sessionID: SESSION }),
1802 out("session.active"),
1803 reply("session.active", { data: { [SESSION]: { type: "running" } } }),
1804 out("session.interrupt", { sessionID: SESSION }),
1805 reply("session.interrupt", { interrupted: true }),
1806 ],
1807 { replayGate: gate },
1808 ).pipe(
1809 Effect.provideService(ProviderContinuationRequests.ProviderContinuationRequests, {
1810 offer: () => Deferred.succeed(offered, undefined).pipe(Effect.asVoid),
1811 take: Effect.never,
1812 }),
1813 );
1814 const ended = yield* terminals(runtime, 2);
1815 yield* runtime.startTurn(turnInput(thread)).pipe(Effect.ignore);
1816 gate.release("drop");
1817 // The late start has been read.
1818 yield* Deferred.await(offered);
1819 yield* runtime.startTurn(secondTurn(thread));
1820 const [first, second] = yield* Fiber.join(ended);
1821 assert.equal(first?.status, "failed");
1822 assert.equal(second?.failure?.message, OPENCODE_2_STILL_STOPPING);
1823 }).pipe(Effect.scoped, Effect.provide(TestClock.layer())),
1824 );
1825
1727 it.effect("prompts again without a check after the server refused a prompt", () =>1826 it.effect("prompts again without a check after the server refused a prompt", () =>
1728 Effect.gen(function* () {1827 Effect.gen(function* () {
1729 const { runtime, thread } = yield* resumed([1828 const { runtime, thread } = yield* resumed([
...@@ -1764,6 +1863,257 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => {...@@ -1764,6 +1863,257 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => {
1764 }).pipe(Effect.scoped, Effect.provide(TestClock.layer())),1863 }).pipe(Effect.scoped, Effect.provide(TestClock.layer())),
1765 );1864 );
17661865
1866 it.effect("ends a background reply's continuation after a lost stream took a Stop's end", () =>
1867 Effect.gen(function* () {
1868 const offered = yield* Deferred.make<void>();
1869 const gate = makeProviderReplayGate(["drop", "reply"]);
1870 const { runtime, thread } = yield* resumed(
1871 [
1872 ...backgroundLaunch(CHILD),
1873 // Restarting the turn stops only the parent, and the server never answers.
1874 out("session.interrupt", { sessionID: SESSION }),
1875 reply("session.interrupt", "<hang>"),
1876 // The restarted turn finds the parent still stopping and fails.
1877 out("session.active"),
1878 replyData("session.active", { [SESSION]: { type: "running" } }),
1879 out("session.interrupt", { sessionID: SESSION }),
1880 reply("session.interrupt", { interrupted: true }),
1881 // The stream drops before the parent's end; the reconnect finds it idle.
1882 event("session.usage.updated", { sessionID: SESSION }, "drop"),
1883 { type: "runtime_exit", status: "success" } as const,
1884 out("event.subscribe"),
1885 out("session.active"),
1886 replyData("session.active", { [CHILD]: { type: "running" } }),
1887 out("permission.list", { sessionID: CHILD }),
1888 replyData("permission.list", []),
1889 out("session.form.list", { sessionID: CHILD }),
1890 replyData("session.form.list", []),
1891 // The subagent ends and OpenCode answers its report on its own.
1892 event("session.execution.succeeded", { sessionID: CHILD }),
1893 event("session.inbox.enqueued", {
1894 inboxID: "msg_report",
1895 sessionID: SESSION,
1896 item: {
1897 type: "synthetic",
1898 payload: {
1899 text: `<subagent sessionID="${CHILD}" state="completed" description="Sleep">\nCHILD_OK\n</subagent>`,
1900 description: "Sleep",
1901 metadata: {
1902 source: "subagent",
1903 childID: CHILD,
1904 agent: "General",
1905 state: "completed",
1906 },
1907 },
1908 delivery: "steer",
1909 },
1910 }),
1911 event("session.execution.started", { sessionID: SESSION }),
1912 event(
1913 "session.text.ended",
1914 {
1915 sessionID: SESSION,
1916 assistantMessageID: "msg_followup",
1917 ordinal: 0,
1918 text: "CHILD_OK",
1919 },
1920 "reply",
1921 ),
1922 event("session.execution.succeeded", { sessionID: SESSION }),
1923 // The subagent's session runs again: its turn marks that the reply's end was handled.
1924 event("session.execution.started", { sessionID: CHILD }),
1925 ],
1926 { replayGate: gate },
1927 ).pipe(
1928 Effect.provideService(ProviderContinuationRequests.ProviderContinuationRequests, {
1929 offer: () => Deferred.succeed(offered, undefined).pipe(Effect.asVoid),
1930 take: Effect.never,
1931 }),
1932 );
1933 // The single reader of the runtime's events.
1934 const attached = yield* Deferred.make<void>();
1935 const collected = yield* runtime.events.pipe(
1936 Stream.tap((event) =>
1937 event.type === "subagent.updated" && event.subagent.childThreadId !== null
1938 ? Deferred.succeed(attached, undefined)
1939 : Effect.void,
1940 ),
1941 Stream.takeUntil(
1942 (event) =>
1943 event.type === "provider_turn.updated" &&
1944 event.providerTurn.nativeTurnRef?.nativeId?.startsWith(`${CHILD}:turn:`) === true,
1945 ),
1946 Stream.runCollect,
1947 Effect.forkScoped,
1948 );
1949 yield* runtime.startTurn(withLineage(thread));
1950 yield* Deferred.await(attached);
1951 const interrupt = yield* runtime
1952 .interruptTurn({ providerThread: thread, providerTurnId: yield* providerTurnId })
1953 .pipe(Effect.forkScoped);
1954 yield* TestClock.adjust("11 seconds");
1955 yield* Fiber.join(interrupt);
1956 yield* runtime.startTurn({ ...secondTurn(thread), appThread: withLineage(thread).appThread });
1957 gate.release("drop");
1958 yield* Deferred.await(offered);
1959 yield* runtime.startTurn({
1960 ...withLineage(thread),
1961 runId: RunId.make("run:opencode2-adapter:wake"),
1962 runOrdinal: 3,
1963 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 },
1971 });
1972 gate.release("reply");
1973 // The reply's own end finishes its continuation: the lost Stop's end does not take it.
1974 assert.deepEqual(
1975 [...(yield* Fiber.join(collected))].flatMap((event) =>
1976 event.type === "turn.terminal" ? [event.status] : [],
1977 ),
1978 ["interrupted", "failed", "completed"],
1979 );
1980 }).pipe(Effect.scoped, Effect.provide(TestClock.layer())),
1981 );
1982
1983 /**
1984 * A prompt fails without a clear rejection, OpenCode never took it, and a
1985 * background subagent's report then wakes the parent into a reply. With
1986 * `endedFirst` that reply ends before its continuation takes it.
1987 */
1988 const replyAfterUnclearFailure = (endedFirst: boolean) =>
1989 Effect.gen(function* () {
1990 const offered = yield* Deferred.make<void>();
1991 const gate = makeProviderReplayGate(endedFirst ? ["wake"] : ["wake", "reply"]);
1992 const { runtime, thread } = yield* resumed(
1993 [
1994 ...backgroundLaunch(CHILD),
1995 event("session.execution.succeeded", { sessionID: SESSION }),
1996 out("session.update", { sessionID: CHILD, permissions: t3Rules }),
1997 reply("session.update", null),
1998 out("session.prompt", { sessionID: SESSION, text: "<any>" }),
1999 reply("session.prompt", {
2000 status: 502,
2001 body: { _tag: "UnknownError", message: "bad gateway" },
2002 }),
2003 // The subagent ends and OpenCode answers its report on its own.
2004 event("session.execution.succeeded", { sessionID: CHILD }, "wake"),
2005 event("session.inbox.enqueued", {
2006 inboxID: "msg_report",
2007 sessionID: SESSION,
2008 item: {
2009 type: "synthetic",
2010 payload: {
2011 text: `<subagent sessionID="${CHILD}" state="completed" description="Sleep">\nCHILD_OK\n</subagent>`,
2012 description: "Sleep",
2013 metadata: {
2014 source: "subagent",
2015 childID: CHILD,
2016 agent: "General",
2017 state: "completed",
2018 },
2019 },
2020 delivery: "steer",
2021 },
2022 }),
2023 event("session.execution.started", { sessionID: SESSION }),
2024 event(
2025 "session.text.ended",
2026 {
2027 sessionID: SESSION,
2028 assistantMessageID: "msg_followup",
2029 ordinal: 0,
2030 text: "CHILD_OK",
2031 },
2032 endedFirst ? undefined : "reply",
2033 ),
2034 event("session.execution.succeeded", { sessionID: SESSION }),
2035 // The subagent's session runs again: a turn of its own, after the reply's end.
2036 event("session.execution.started", { sessionID: CHILD }),
2037 ],
2038 { replayGate: gate },
2039 ).pipe(
2040 Effect.provideService(ProviderContinuationRequests.ProviderContinuationRequests, {
2041 offer: () => Deferred.succeed(offered, undefined).pipe(Effect.asVoid),
2042 take: Effect.never,
2043 }),
2044 );
2045 // The single reader of the runtime's events.
2046 const launched = yield* Deferred.make<void>();
2047 const childRan = yield* Deferred.make<void>();
2048 const ended: Array<ProviderAdapter.ProviderAdapterV2Event> = [];
2049 const allEnded = yield* Deferred.make<void>();
2050 yield* runtime.events.pipe(
2051 Stream.tap((event) =>
2052 Effect.gen(function* () {
2053 if (
2054 event.type === "provider_turn.updated" &&
2055 event.providerTurn.nativeTurnRef?.nativeId?.startsWith(`${CHILD}:turn:`) === true
2056 ) {
2057 yield* Deferred.succeed(childRan, undefined);
2058 }
2059 if (event.type !== "turn_item.updated" && event.type !== "turn.terminal") return;
2060 ended.push(event);
2061 if (event.type === "turn.terminal") yield* Deferred.succeed(launched, undefined);
2062 if (ended.filter((entry) => entry.type === "turn.terminal").length === 3) {
2063 yield* Deferred.succeed(allEnded, undefined);
2064 }
2065 }),
2066 ),
2067 Stream.runDrain,
2068 Effect.forkScoped,
2069 );
2070 yield* runtime.startTurn(withLineage(thread));
2071 yield* Deferred.await(launched);
2072 yield* runtime
2073 .startTurn({ ...secondTurn(thread), appThread: withLineage(thread).appThread })
2074 .pipe(Effect.ignore);
2075 gate.release("wake");
2076 yield* Deferred.await(offered);
2077 // Everything up to the subagent's next run has been read: the reply ended.
2078 if (endedFirst) yield* Deferred.await(childRan);
2079 yield* runtime.startTurn({
2080 ...withLineage(thread),
2081 runId: RunId.make("run:opencode2-adapter:wake"),
2082 runOrdinal: 3,
2083 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 },
2091 });
2092 if (!endedFirst) gate.release("reply");
2093 // The reply's own end finishes its continuation: the failed prompt's mark does not take it.
2094 yield* Deferred.await(allEnded);
2095 assert.deepEqual(
2096 ended.flatMap((event) => (event.type === "turn.terminal" ? [event.status] : [])),
2097 ["completed", "failed", "completed"],
2098 );
2099 assert.include(
2100 ended.flatMap((event) =>
2101 event.type === "turn_item.updated" && event.turnItem.type === "assistant_message"
2102 ? [event.turnItem.text]
2103 : [],
2104 ),
2105 "CHILD_OK",
2106 );
2107 }).pipe(Effect.scoped, Effect.provide(TestClock.layer()));
2108
2109 it.effect("ends a background reply's continuation after a prompt failed unclearly", () =>
2110 replyAfterUnclearFailure(false),
2111 );
2112
2113 it.effect("ends a finished background reply's continuation after a prompt failed unclearly", () =>
2114 replyAfterUnclearFailure(true),
2115 );
2116
1767 it.effect("does not report a Stop the server says did nothing", () =>2117 it.effect("does not report a Stop the server says did nothing", () =>
1768 Effect.gen(function* () {2118 Effect.gen(function* () {
1769 const { runtime, thread } = yield* resumed([2119 const { runtime, thread } = yield* resumed([
apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.testkit.ts+5-1
...@@ -286,7 +286,11 @@ function layerRegistry(...@@ -286,7 +286,11 @@ function layerRegistry(
286 */286 */
287export const openCode2ReplayRuntime = (287export const openCode2ReplayRuntime = (
288 entries: ReadonlyArray<ProviderReplayEntry>,288 entries: ReadonlyArray<ProviderReplayEntry>,
289 options?: { readonly external?: boolean; readonly borrowers?: { current: number } },289 options?: {
290 readonly external?: boolean;
291 readonly borrowers?: { current: number };
292 readonly replayGate?: ProviderReplayGate;
293 },
290) =>294) =>
291 Effect.gen(function* () {295 Effect.gen(function* () {
292 const adapter = yield* makeReplayAdapter(296 const adapter = yield* makeReplayAdapter(
packages/provider-opencode/src/server/v2/adapter.ts+51-12
...@@ -340,6 +340,12 @@ interface ThreadState {...@@ -340,6 +340,12 @@ interface ThreadState {
340 * so the next execution end belongs to that run; it clears this and ends no turn.340 * so the next execution end belongs to that run; it clears this and ends no turn.
341 */341 */
342 unsettled: boolean;342 unsettled: boolean;
343 /**
344 * Whether the run `unsettled` waits on had started when it was marked. A
345 * prompt whose request failed may start its run late; that start is the
346 * run's own, and only a start after it means its end was lost.
347 */
348 unsettledStarted: boolean;
343 /** The session's location, where its agents' path rules are read. */349 /** The session's location, where its agents' path rules are read. */
344 directory: string;350 directory: string;
345 /**351 /**
...@@ -923,6 +929,11 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr...@@ -923,6 +929,11 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr
923 const emit = (event: ProviderAdapter.ProviderAdapterV2Event) =>929 const emit = (event: ProviderAdapter.ProviderAdapterV2Event) =>
924 Queue.offer(events, event).pipe(Effect.asVoid);930 Queue.offer(events, event).pipe(Effect.asVoid);
925 const ownerOf = (sessionId: string) => childOwners.get(sessionId) ?? threads.get(sessionId);931 const ownerOf = (sessionId: string) => childOwners.get(sessionId) ?? threads.get(sessionId);
932 /** Marks a run T3 ended while OpenCode may still run it (see `unsettled`). */
933 const markUnsettled = (state: ThreadState) => {
934 state.unsettled = true;
935 state.unsettledStarted = busy.has(state.sessionId);
936 };
926937
927 const newThreadState = (938 const newThreadState = (
928 sessionId: string,939 sessionId: string,
...@@ -936,6 +947,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr...@@ -936,6 +947,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr
936 active: undefined,947 active: undefined,
937 model: undefined,948 model: undefined,
938 unsettled: false,949 unsettled: false,
950 unsettledStarted: false,
939 directory,951 directory,
940 agent: "build",952 agent: "build",
941 title: undefined,953 title: undefined,
...@@ -1838,7 +1850,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr...@@ -1838,7 +1850,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr
1838 reason,1850 reason,
1839 });1851 });
1840 if (state.active === undefined) return;1852 if (state.active === undefined) return;
1841 state.unsettled = true;1853 markUnsettled(state);
1842 yield* finishTurn(state, {1854 yield* finishTurn(state, {
1843 status: "failed",1855 status: "failed",
1844 failure: makeProviderFailure({1856 failure: makeProviderFailure({
...@@ -1935,7 +1947,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr...@@ -1935,7 +1947,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr
1935 const [permissions, forms] = listed.value;1947 const [permissions, forms] = listed.value;
1936 if (permissions.length === 0 && forms.length === 0) return;1948 if (permissions.length === 0 && forms.length === 0) return;
1937 // The stopped run's end is not the next turn's.1949 // The stopped run's end is not the next turn's.
1938 state.unsettled = true;1950 markUnsettled(state);
1939 for (const request of permissions) {1951 for (const request of permissions) {
1940 yield* stopStaleRequest(state.sessionId, { type: "permission", id: request.id });1952 yield* stopStaleRequest(state.sessionId, { type: "permission", id: request.id });
1941 }1953 }
...@@ -2120,7 +2132,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr...@@ -2120,7 +2132,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr
2120 // goes on. A subagent's session has no Stop end to skip.2132 // goes on. A subagent's session has no Stop end to skip.
2121 const asker = threads.get(form.sessionID) ?? state;2133 const asker = threads.get(form.sessionID) ?? state;
2122 if (!cancelled) return yield* abandonRequest(asker, "form cancel failed");2134 if (!cancelled) return yield* abandonRequest(asker, "form cancel failed");
2123 if (asker.subagent === undefined) asker.unsettled = true;2135 if (asker.subagent === undefined) markUnsettled(asker);
2124 yield* finishTurn(asker, {2136 yield* finishTurn(asker, {
2125 status: "failed",2137 status: "failed",
2126 failure: makeProviderFailure({2138 failure: makeProviderFailure({
...@@ -2358,7 +2370,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr...@@ -2358,7 +2370,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr
2358 delivered.every((entry) => state.stoppedChildren.has(entry.childId));2370 delivered.every((entry) => state.stoppedChildren.has(entry.childId));
2359 for (const entry of delivered) state.stoppedChildren.delete(entry.childId);2371 for (const entry of delivered) state.stoppedChildren.delete(entry.childId);
2360 if (stopped) {2372 if (stopped) {
2361 state.unsettled = true;2373 markUnsettled(state);
2362 yield* client.session2374 yield* client.session
2363 .interrupt({ sessionID: Session.ID.make(state.sessionId) })2375 .interrupt({ sessionID: Session.ID.make(state.sessionId) })
2364 .pipe(Effect.timeout(INTERRUPT_TIMEOUT), Effect.ignore({ log: true }));2376 .pipe(Effect.timeout(INTERRUPT_TIMEOUT), Effect.ignore({ log: true }));
...@@ -2484,6 +2496,16 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr...@@ -2484,6 +2496,16 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr
2484 event.type === "session.execution.started" ||2496 event.type === "session.execution.started" ||
2485 event.type === "unreadable.execution.started"2497 event.type === "unreadable.execution.started"
2486 ) {2498 ) {
2499 // A session runs one execution at a time, each opening with `started`
2500 // on this ordered stream. Once the run `unsettled` waits on has
2501 // started, a later start means that run is over: an end not seen by
2502 // now was lost with a dropped stream, and the next end is this
2503 // execution's own. A run not yet started when it was marked starts here.
2504 const state = threads.get(sessionId);
2505 if (state?.unsettled === true) {
2506 if (state.unsettledStarted) state.unsettled = false;
2507 else state.unsettledStarted = true;
2508 }
2487 busy.add(sessionId);2509 busy.add(sessionId);
2488 } else if (event.type === "unreadable.execution.ended" || executionEnd(event.type)) {2510 } else if (event.type === "unreadable.execution.ended" || executionEnd(event.type)) {
2489 busy.delete(sessionId);2511 busy.delete(sessionId);
...@@ -2515,14 +2537,20 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr...@@ -2515,14 +2537,20 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr
2515 return yield* route(event, sessionId);2537 return yield* route(event, sessionId);
2516 });2538 });
25172539
2518 /** Handles one event for its session, as the turn running there sees it. */2540 /**
2541 * Handles one event for its session, as the turn running there sees it.
2542 * `held` is set for a held execution's events as its turn replays them.
2543 */
2519 const route = Effect.fnUntraced(function* (2544 const route = Effect.fnUntraced(function* (
2520 event: OpenCode2Client.OpenCode2StreamEvent,2545 event: OpenCode2Client.OpenCode2StreamEvent,
2521 sessionId: string | undefined,2546 sessionId: string | undefined,
2547 held = false,
2522 ) {2548 ) {
2523 // The end of the run a timed-out Stop left behind; no turn is its own.2549 // The end of the run a timed-out Stop left behind; no turn is its own.
2550 // A held execution's end is its own: any such run was over, or was this
2551 // execution, before it started.
2524 const ended =2552 const ended =
2525 event.type === "unreadable.execution.ended" || executionEnd(event.type)2553 !held && (event.type === "unreadable.execution.ended" || executionEnd(event.type))
2526 ? threads.get(sessionId ?? "")2554 ? threads.get(sessionId ?? "")
2527 : undefined;2555 : undefined;
2528 if (ended?.unsettled === true) {2556 if (ended?.unsettled === true) {
...@@ -2777,10 +2805,18 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr...@@ -2777,10 +2805,18 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr
2777 const background = [...threads.values()].filter(2805 const background = [...threads.values()].filter(
2778 (state) => state.subagent === undefined && hasBackground(state),2806 (state) => state.subagent === undefined && hasBackground(state),
2779 );2807 );
2780 if (running.length === 0 && background.length === 0 && busy.size === 0) return;2808 const marked = [...threads.values()].some((state) => state.unsettled);
2809 if (running.length === 0 && background.length === 0 && busy.size === 0 && !marked) return;
2781 const active = yield* client.session.active();2810 const active = yield* client.session.active();
2782 // An execution that ended while the stream was down never said so.2811 // An execution that ended while the stream was down never said so.
2783 for (const sessionId of busy) if (!(sessionId in active)) busy.delete(sessionId);2812 for (const sessionId of busy) if (!(sessionId in active)) busy.delete(sessionId);
2813 // A run `unsettled` waits on is over once its session is idle: any start
2814 // from here is a new execution's, even if the run's own start was lost in
2815 // the gap. A session still active may not have started that run yet, so
2816 // its next start may still be the run's own; its end clears the mark.
2817 for (const state of threads.values()) {
2818 if (state.unsettled && !(state.sessionId in active)) state.unsettledStarted = true;
2819 }
2784 for (const [sessionId, state] of running) {2820 for (const [sessionId, state] of running) {
2785 const turn = state.active;2821 const turn = state.active;
2786 if (turn === undefined) continue;2822 if (turn === undefined) continue;
...@@ -3403,7 +3439,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr...@@ -3403,7 +3439,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr
3403 /** Feeds a held execution's events to the turn now running on its session. */3439 /** Feeds a held execution's events to the turn now running on its session. */
3404 const replay = Effect.fnUntraced(function* (wake: Wake) {3440 const replay = Effect.fnUntraced(function* (wake: Wake) {
3405 wake.dropped = true;3441 wake.dropped = true;
3406 for (const event of wake.events) yield* route(event, sessionOfEvent(event));3442 for (const event of wake.events) yield* route(event, sessionOfEvent(event), true);
3407 });3443 });
34083444
3409 /**3445 /**
...@@ -3438,6 +3474,9 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr...@@ -3438,6 +3474,9 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr
3438 if (turn !== undefined && wake.first !== undefined) {3474 if (turn !== undefined && wake.first !== undefined) {
3439 turn.providerTurn = { ...turn.providerTurn, nativeTurnRef: ref(wake.first, "weak") };3475 turn.providerTurn = { ...turn.providerTurn, nativeTurnRef: ref(wake.first, "weak") };
3440 }3476 }
3477 // The execution still runs, so the run a mark waits on is this one or
3478 // was over before it started: its end, still to come, is this turn's.
3479 if (wake.running) state.unsettled = false;
3441 yield* replay(wake);3480 yield* replay(wake);
3442 }),3481 }),
3443 );3482 );
...@@ -3520,7 +3559,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr...@@ -3520,7 +3559,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr
3520 const running = state.wakes.some((wake) => wake.running);3559 const running = state.wakes.some((wake) => wake.running);
3521 for (const wake of state.wakes.splice(0)) wake.dropped = true;3560 for (const wake of state.wakes.splice(0)) wake.dropped = true;
3522 if (running) {3561 if (running) {
3523 state.unsettled = true;3562 markUnsettled(state);
3524 yield* client.session3563 yield* client.session
3525 .interrupt({ sessionID: Session.ID.make(state.sessionId) })3564 .interrupt({ sessionID: Session.ID.make(state.sessionId) })
3526 .pipe(Effect.timeout(INTERRUPT_TIMEOUT), Effect.ignore({ log: true }));3565 .pipe(Effect.timeout(INTERRUPT_TIMEOUT), Effect.ignore({ log: true }));
...@@ -3896,7 +3935,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr...@@ -3896,7 +3935,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr
3896 ? Effect.gen(function* () {3935 ? Effect.gen(function* () {
3897 // Without a clear rejection the server may have taken the3936 // Without a clear rejection the server may have taken the
3898 // prompt, so the next turn checks before it prompts again.3937 // prompt, so the next turn checks before it prompts again.
3899 if (!CLEAR_PROMPT_REJECTIONS.has(cause._tag)) state.unsettled = true;3938 if (!CLEAR_PROMPT_REJECTIONS.has(cause._tag)) markUnsettled(state);
3900 yield* finishTurn(state, {3939 yield* finishTurn(state, {
3901 status: "failed",3940 status: "failed",
3902 failure: makeProviderFailure({ cause, class: "provider_error" }),3941 failure: makeProviderFailure({ cause, class: "provider_error" }),
...@@ -4004,13 +4043,13 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr...@@ -4004,13 +4043,13 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr
4004 ),4043 ),
4005 );4044 );
4006 if (reply._tag === "None") {4045 if (reply._tag === "None") {
4007 state.unsettled = true;4046 markUnsettled(state);
4008 return yield* lock.withPermit(finishTurn(state, { status: "interrupted" }));4047 return yield* lock.withPermit(finishTurn(state, { status: "interrupted" }));
4009 }4048 }
4010 // Between the executions a late steer spans nothing runs yet. The4049 // Between the executions a late steer spans nothing runs yet. The
4011 // steers are taken back, and a run they already started is skipped.4050 // steers are taken back, and a run they already started is skipped.
4012 if (!reply.value.interrupted && state.active === turn && turn.heldEnd !== undefined) {4051 if (!reply.value.interrupted && state.active === turn && turn.heldEnd !== undefined) {
4013 state.unsettled = true;4052 markUnsettled(state);
4014 yield* lock.withPermit(finishTurn(state, { status: "interrupted" }));4053 yield* lock.withPermit(finishTurn(state, { status: "interrupted" }));
4015 return yield* cancelStrandedSteers(state);4054 return yield* cancelStrandedSteers(state);
4016 }4055 }