diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts index 15ca24556ad6..7bae7e1c6982 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts @@ -72,11 +72,18 @@ const event = ( type: string, data: Record, label?: string, + version = 1, ): ProviderReplayEntry => ({ type: "emit_inbound", frame: { type: "sdk.event", - event: { id: `evt_${type.replaceAll(".", "")}0000`, created: 1, type, data, ...durable }, + event: { + id: `evt_${type.replaceAll(".", "")}0000`, + created: 1, + type, + data, + durable: { ...durable.durable, version }, + }, }, ...(label === undefined ? {} : { label }), }); @@ -835,47 +842,85 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { }).pipe(Effect.scoped), ); - it.effect("keeps a finished background subagent's queued report as pending work", () => - Effect.gen(function* () { - const { runtime, thread } = yield* resumed([ - ...backgroundLaunch(CHILD), - event("session.execution.succeeded", { sessionID: SESSION }), - event("session.execution.succeeded", { sessionID: CHILD }), - event("session.inbox.enqueued", { - inboxID: "msg_report", - sessionID: SESSION, - item: { - type: "synthetic", - payload: { - text: `\nCHILD_OK\n`, - description: "Sleep", - metadata: { - source: "subagent", - childID: CHILD, - agent: "General", - state: "completed", - }, - }, - delivery: "steer", - }, - }), - event("session.usage.updated", { sessionID: SESSION }), - ]); - const reported = yield* Deferred.make(); - yield* runtime.events.pipe( - Stream.tap((event) => - event.type === "subagent.updated" && event.subagent.status === "completed" - ? Deferred.succeed(reported, undefined) - : Effect.void, - ), - Stream.runDrain, - Effect.forkScoped, - ); - yield* runtime.startTurn(withLineage(thread)); - yield* Deferred.await(reported); - assert.isTrue(yield* runtime.hasPendingBackgroundWork!); - assert.isTrue(yield* runtime.hasPendingBackgroundWorkForThread!(thread)); - }).pipe(Effect.scoped), + it.effect( + "keeps a finished background subagent's queued report pending until its native continuation ends", + () => + Effect.gen(function* () { + const gate = makeProviderReplayGate(["continue"]); + const { runtime, thread } = yield* resumed( + [ + ...nestedLaunch, + event("session.execution.succeeded", { sessionID: DEEP }), + deepReport("completed"), + event("session.model.selected", { + sessionID: SESSION, + model: { providerID: "opencode", id: "big-pickle", variant: "thinking" }, + }), + event("session.execution.started", { sessionID: MIDDLE }, "continue"), + event("session.inbox.delivered", { sessionID: MIDDLE, inboxID: "msg_deep_report" }), + event("session.text.ended", { + sessionID: MIDDLE, + assistantMessageID: "msg_native_continuation", + ordinal: 0, + text: "GRANDCHILD_OK", + }), + event("session.execution.succeeded", { sessionID: MIDDLE }), + event("session.model.selected", { + sessionID: SESSION, + model: { providerID: "opencode", id: "big-pickle", variant: "none" }, + }), + ], + { replayGate: gate }, + ); + const queued = yield* Deferred.make(); + const drained = yield* Deferred.make(); + const timeline: Array = []; + yield* runtime.events.pipe( + Stream.runForEach((event) => + Effect.gen(function* () { + if (event.type === "provider_thread.updated") { + if ( + event.providerThread.nativeThreadRef?.nativeId === MIDDLE && + event.pendingBackgroundWork !== undefined + ) { + timeline.push(event.pendingBackgroundWork); + } + if (event.providerThread.id === thread.id) { + const variant = + event.providerThread.nativeMetadata?.modelSelection?.options?.[0]?.value; + if (variant === "thinking") yield* Deferred.succeed(queued, undefined); + if (variant === "none") yield* Deferred.succeed(drained, undefined); + } + } + if ( + event.type === "provider_turn.updated" && + event.providerTurn.nativeTurnRef?.nativeId === `${MIDDLE}:turn:2` + ) { + timeline.push(event.providerTurn.status); + } + if (event.type === "message.updated" && event.message.text === "GRANDCHILD_OK") { + timeline.push(event.message.text); + } + }), + ), + Effect.forkScoped, + ); + yield* runtime.startTurn(withLineage(thread)); + yield* Deferred.await(queued); + assert.isTrue(yield* runtime.hasPendingBackgroundWork!); + assert.isTrue(yield* runtime.hasPendingBackgroundWorkForThread!(thread)); + assert.strictEqual(timeline.at(-1), true); + timeline.length = 0; + gate.release("continue"); + yield* Deferred.await(drained); + assert.deepEqual( + timeline.filter((entry) => typeof entry === "string"), + ["running", "GRANDCHILD_OK", "completed"], + ); + assert.isAbove(timeline.indexOf(false), timeline.indexOf("completed")); + assert.strictEqual(timeline.at(-1), false); + assert.isFalse(yield* runtime.hasPendingBackgroundWorkForThread!(thread)); + }).pipe(Effect.scoped), ); const returnedRunning = (id: string): ProviderReplayEntry => ({ @@ -1175,6 +1220,9 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { Effect.gen(function* () { const deep: Array = []; const middleTurns = new Map(); + const pendingSnapshots = new Map(); + const variants = new Map(); + let rootVariant: unknown; const terminals: Array = []; const changed = yield* Deferred.make(); let wake = changed; @@ -1182,6 +1230,16 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { Stream.tap((event) => Effect.gen(function* () { if (event.type === "turn.terminal") terminals.push(event.status); + if (event.type === "provider_thread.updated") { + const nativeId = event.providerThread.nativeThreadRef?.nativeId; + if (nativeId != null && event.pendingBackgroundWork !== undefined) { + pendingSnapshots.set(nativeId, event.pendingBackgroundWork); + } + const variant = + event.providerThread.nativeMetadata?.modelSelection?.options?.[0]?.value; + if (nativeId != null) variants.set(nativeId, variant); + if (nativeId === SESSION) rootVariant = variant; + } if (event.type === "subagent.updated" && event.subagent.title === "Deep") { deep.push(event.subagent.status); } @@ -1206,9 +1264,164 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { yield* until(check); }); }); - return { deep, middleTurns, terminals, until }; + return { + deep, + middleTurns, + terminals, + pendingSnapshots, + variants, + get rootVariant() { + return rootVariant; + }, + until, + }; }); + it.effect("clears a nested report cancelled during a held wake", () => + Effect.gen(function* () { + const offers: Array = []; + const gate = makeProviderReplayGate(["drained"]); + const { runtime, thread } = yield* resumed( + [ + ...backgroundLaunch(CHILD), + event("session.execution.succeeded", { sessionID: SESSION }), + event("session.execution.succeeded", { sessionID: CHILD }), + childReport("msg_report", "CHILD_OK"), + event("session.execution.started", { sessionID: SESSION }), + ...nestedLaunch.slice(3, -2), + event( + "session.tool.success", + { + ...toolOf(SESSION, "call-middle"), + content: [{ type: "text", text: "MIDDLE_OK" }], + metadata: { sessionID: MIDDLE, status: "completed", truncated: false }, + executed: false, + }, + undefined, + 2, + ), + event("session.execution.succeeded", { sessionID: DEEP }), + deepReport("completed"), + event("session.inbox.cancelled", { sessionID: MIDDLE, inboxID: "msg_deep_report" }), + event("session.execution.succeeded", { sessionID: SESSION }), + event("session.model.selected", { + sessionID: SESSION, + model: { providerID: "opencode", id: "big-pickle", variant: "thinking" }, + }), + event( + "session.model.selected", + { + sessionID: SESSION, + model: { providerID: "opencode", id: "big-pickle", variant: "none" }, + }, + "drained", + ), + ], + { replayGate: gate }, + ).pipe( + Effect.provideService(ProviderContinuationRequests.ProviderContinuationRequests, { + offer: (request) => + Effect.sync(() => { + offers.push(request); + }), + take: Effect.never, + }), + ); + const watch = yield* watchNested(runtime); + yield* runtime.startTurn(withLineage(thread)); + assert.isTrue(yield* Effect.promise(() => gate.waitForReached("drained"))); + yield* watch.until(() => watch.rootVariant === "thinking"); + assert.lengthOf(offers, 1); + yield* runtime.startTurn(continuationTurn(thread, offers[0]!)); + gate.release("drained"); + yield* watch.until(() => watch.rootVariant === "none"); + assert.isFalse(yield* runtime.hasPendingBackgroundWork!); + assert.strictEqual(watch.pendingSnapshots.get(MIDDLE), false); + }).pipe(Effect.scoped), + ); + + it.effect("clears the caller's pending work when a called session's held wake runs", () => + Effect.gen(function* () { + const OTHER = "ses_other00000000000000000000"; + const offers: Array = []; + const gate = makeProviderReplayGate(["drained"]); + const { runtime, thread } = yield* resumed( + [ + out("session.get", { sessionID: OTHER }), + replyData("session.get", sessionInfo({ id: OTHER })), + out("permission.list", { sessionID: OTHER }), + replyData("permission.list", []), + out("session.form.list", { sessionID: OTHER }), + replyData("session.form.list", []), + event("session.execution.started", { sessionID: OTHER }), + event("session.execution.succeeded", { sessionID: OTHER }), + out("session.prompt", { sessionID: SESSION, text: "" }), + promptAccepted, + event("session.execution.started", { sessionID: SESSION }), + event("session.tool.input.started", { + ...toolOf(SESSION, "call-other"), + name: "subagent", + }), + event("session.tool.called", { + ...toolOf(SESSION, "call-other"), + name: "subagent", + input: { description: "Other", prompt: "again" }, + executed: false, + }), + event("session.tool.progress", { + ...toolOf(SESSION, "call-other"), + metadata: { sessionID: OTHER, status: "running" }, + }), + event("session.tool.success", { + ...toolOf(SESSION, "call-other"), + content: [{ type: "text", text: "OTHER_OK" }], + metadata: { sessionID: OTHER, status: "completed", truncated: false }, + executed: false, + }), + event("session.execution.succeeded", { sessionID: SESSION }), + event( + "session.model.selected", + { + sessionID: OTHER, + model: { providerID: "opencode", id: "big-pickle", variant: "none" }, + }, + "drained", + ), + ], + { replayGate: gate }, + ).pipe( + Effect.provideService(ProviderContinuationRequests.ProviderContinuationRequests, { + offer: (request) => + Effect.sync(() => { + offers.push(request); + }), + take: Effect.never, + }), + ); + const other = yield* runtime.resumeThread({ + providerThread: { + ...thread, + id: ProviderThreadId.make("provider-thread:opencode2-adapter:other"), + nativeThreadRef: { driver: OPENCODE_PROVIDER, nativeId: OTHER, strength: "strong" }, + }, + threadId, + modelSelection: bigPickle, + runtimePolicy: policy(), + }); + const watch = yield* watchNested(runtime); + yield* watch.until(() => offers.length === 1); + yield* runtime.startTurn(withLineage(thread)); + assert.isTrue(yield* Effect.promise(() => gate.waitForReached("drained"))); + // The called session's held wake is its caller's pending work too. + yield* watch.until(() => watch.pendingSnapshots.get(SESSION) === true); + yield* runtime.startTurn(continuationTurn(other, offers[0]!)); + gate.release("drained"); + yield* watch.until(() => watch.variants.get(OTHER) === "none"); + assert.isFalse(watch.pendingSnapshots.get(OTHER)); + assert.isFalse(watch.pendingSnapshots.get(SESSION)); + }).pipe(Effect.scoped), + ); + it.effect("stops a foreground subagent's background subagent after the turn ended", () => Effect.gen(function* () { const offered = yield* Deferred.make(); diff --git a/apps/server/src/orchestration-v2/testkit/OpenCode2PendingSnapshots.evidence.test.ts b/apps/server/src/orchestration-v2/testkit/OpenCode2PendingSnapshots.evidence.test.ts new file mode 100644 index 000000000000..1d7477cc37dc --- /dev/null +++ b/apps/server/src/orchestration-v2/testkit/OpenCode2PendingSnapshots.evidence.test.ts @@ -0,0 +1,159 @@ +import { assert, describe, it } from "@effect/vitest"; +import type { ProviderInstanceId } from "@t3tools/contracts"; +import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import * as Config from "effect/Config"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Stream from "effect/Stream"; + +import { OpenCode2OrchestratorReplayHarness } from "../Adapters/OpenCode2AdapterV2.testkit.ts"; +import * as ProviderAdapterRegistry from "../ProviderAdapterRegistry.ts"; +import { findReplayFixture, runReplayFixture } from "./ProviderReplayFixtures.ts"; + +// Evidence driver for #19228: replays recorded OpenCode 2 fixtures through +// OrchestratorV2 and records the adapter's own event stream, because the +// `pendingBackgroundWork` flag is dropped before it reaches domain events. +// Set EV_TRANSCRIPT to a file path to append the transcript there. + +type Row = { + readonly id: string; + readonly fixture: string; + /** True when every native thread must end on a `pendingBackgroundWork: false` snapshot. */ + readonly requireSettledSnapshots: boolean; +}; + +const ROWS: ReadonlyArray = [ + { id: "R1", fixture: "opencode2_nested_background", requireSettledSnapshots: true }, + { id: "R1-compat", fixture: "opencode2_background", requireSettledSnapshots: true }, + { id: "R2c-compat", fixture: "opencode2_resume_after_restart", requireSettledSnapshots: false }, + { id: "R2c-compat", fixture: "opencode2_interrupt", requireSettledSnapshots: false }, + { id: "R3", fixture: "opencode2_simple", requireSettledSnapshots: false }, + { id: "R3", fixture: "opencode2_tool_call", requireSettledSnapshots: false }, +]; + +const tasksOf = (tasks: ReadonlyArray<{ readonly taskId: string }> | undefined) => + `[${(tasks ?? []).map((task) => task.taskId).join(",")}]`; + +const describeEvent = (event: ProviderAdapter.ProviderAdapterV2Event): string | undefined => { + switch (event.type) { + case "provider_thread.updated": + return `provider_thread.updated native=${event.providerThread.nativeThreadRef?.nativeId ?? "?"} pendingBackgroundWork=${String(event.pendingBackgroundWork)} pendingBackgroundTasks=${tasksOf(event.providerThread.pendingBackgroundTasks)}`; + case "provider_turn.updated": + return `provider_turn.updated native=${event.providerTurn.nativeTurnRef?.nativeId ?? "?"} status=${event.providerTurn.status}`; + case "subagent.updated": + return `subagent.updated title=${event.subagent.title ?? "?"} status=${event.subagent.status}`; + case "turn.terminal": + return `turn.terminal status=${event.status}`; + default: + return undefined; + } +}; + +const tappedHarness = ( + record: (event: ProviderAdapter.ProviderAdapterV2Event) => void, +): typeof OpenCode2OrchestratorReplayHarness => ({ + ...OpenCode2OrchestratorReplayHarness, + makeProviderAdapterRegistryLayer: ( + ...args: Parameters + ) => + Layer.effect( + ProviderAdapterRegistry.ProviderAdapterRegistryV2, + Effect.gen(function* () { + const registry = yield* ProviderAdapterRegistry.ProviderAdapterRegistryV2; + return ProviderAdapterRegistry.ProviderAdapterRegistryV2.of({ + ...registry, + get: (instanceId: ProviderInstanceId) => + registry.get(instanceId).pipe( + Effect.map((adapter) => + ProviderAdapter.ProviderAdapterV2.of({ + ...adapter, + openSession: (input) => + adapter.openSession(input).pipe( + Effect.map((runtime) => ({ + ...runtime, + events: runtime.events.pipe( + Stream.tap((event) => Effect.sync(() => record(event))), + ), + })), + ), + }), + ), + ), + }); + }), + ).pipe( + Layer.provide(OpenCode2OrchestratorReplayHarness.makeProviderAdapterRegistryLayer(...args)), + ), +}); + +const runRow = (row: Row) => + Effect.gen(function* () { + const fixture = findReplayFixture(row.fixture); + const driver = fixture?.providers.find((provider) => provider.driver === "opencode"); + assert.isDefined(fixture); + assert.isDefined(driver); + const events: Array = []; + const lines: Array = []; + const emit = (line: string) => lines.push(line); + emit(`== ${row.id} ${row.fixture} (recorded fixture) ==`); + const exit = yield* runReplayFixture({ + fixtureName: row.fixture, + buildInput: fixture.buildInput, + driver, + harness: tappedHarness((event) => events.push(event)), + }).pipe(Effect.exit); + let index = 0; + for (const event of events) { + const line = describeEvent(event); + if (line !== undefined) emit(` adapter[${index}] ${line}`); + index += 1; + } + const lastSnapshot = new Map(); + let everPending = false; + for (const event of events) { + if (event.type !== "provider_thread.updated") continue; + const nativeId = event.providerThread.nativeThreadRef?.nativeId ?? "?"; + lastSnapshot.set(nativeId, event.pendingBackgroundWork); + if (event.pendingBackgroundWork === true) everPending = true; + } + const finals = [...lastSnapshot].map(([id, flag]) => `${id}=${String(flag)}`).join(" "); + emit(` final snapshot per native thread: ${finals}`); + if (exit._tag === "Success") { + for (const domain of exit.value.domainEvents) { + if (domain.type !== "provider-thread.updated") continue; + emit( + ` domain provider-thread.updated native=${domain.payload.nativeThreadRef?.nativeId ?? "?"} pendingBackgroundTasks=${tasksOf(domain.payload.pendingBackgroundTasks)}`, + ); + } + } + const failures: Array = []; + if (exit._tag === "Failure") failures.push("fixture output assertions failed"); + if (row.requireSettledSnapshots) { + if (!everPending) failures.push("no pendingBackgroundWork=true snapshot was published"); + for (const [id, flag] of lastSnapshot) { + if (flag !== false) failures.push(`${id} ended on pendingBackgroundWork=${String(flag)}`); + } + } else if ([...lastSnapshot.values()].includes(true)) { + failures.push("a native thread ended with pending work"); + } + emit( + ` ${row.id} ${row.fixture}: ${failures.length === 0 ? "PASS" : `FAIL (${failures.join("; ")})`}`, + ); + const transcript = yield* Config.String("EV_TRANSCRIPT").pipe(Config.withDefault("")); + if (transcript !== "") { + const fs = yield* FileSystem.FileSystem; + yield* fs.writeFileString(transcript, `${lines.join("\n")}\n`, { flag: "a" }); + } + return failures; + }).pipe(Effect.provide(NodeServices.layer)); + +describe("OpenCode 2 pending-work snapshots (#19228 evidence)", () => { + it.effect.each(ROWS.map((row) => [`${row.id} ${row.fixture}`, row] as const))("%s", ([, row]) => + runRow(row).pipe( + Effect.map((failures) => assert.deepEqual(failures, [])), + Effect.scoped, + ), + ); +}); diff --git a/packages/provider-core/src/server/ProviderAdapter.ts b/packages/provider-core/src/server/ProviderAdapter.ts index db5a18f77b46..ea8304d84585 100644 --- a/packages/provider-core/src/server/ProviderAdapter.ts +++ b/packages/provider-core/src/server/ProviderAdapter.ts @@ -92,6 +92,8 @@ export const ProviderAdapterV2Event = Schema.Union([ type: Schema.Literal("provider_thread.updated"), driver: ProviderDriverKind, providerThread: OrchestrationV2ProviderThread, + /** Work remaining at this point in the event stream, including native continuations. */ + pendingBackgroundWork: Schema.optional(Schema.Boolean), retainedNativeTurnIds: Schema.optional(Schema.Array(Schema.String)), }), Schema.Struct({ diff --git a/packages/provider-opencode/src/server/v2/adapter.ts b/packages/provider-opencode/src/server/v2/adapter.ts index f837e24e2793..29ed0eed0fbc 100644 --- a/packages/provider-opencode/src/server/v2/adapter.ts +++ b/packages/provider-opencode/src/server/v2/adapter.ts @@ -276,6 +276,7 @@ interface OpenBlock { interface ThreadState { readonly sessionId: string; providerThread: OrchestrationV2ProviderThread; + publishedBackgroundWork?: boolean; readonly providerTurns: Map; active: ActiveTurn | undefined; model: ModelRef | undefined; @@ -851,6 +852,17 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr const hasBackground = (thread: ThreadState) => sessionsOf(thread).some(owesWork); + const emitProviderThread = Effect.fnUntraced(function* (state: ThreadState) { + const pendingBackgroundWork = hasBackground(state); + state.publishedBackgroundWork = pendingBackgroundWork; + yield* emit({ + type: "provider_thread.updated", + driver, + providerThread: state.providerThread, + pendingBackgroundWork, + }); + }); + const setSessionStatus = /* Provider events include streaming deltas; avoid a span for each update. */ Effect.fnUntraced(function* ( @@ -1247,7 +1259,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr result: call.result, }, }); - yield* syncRoster(rootOf(call.state)); + yield* syncRosterAncestors(call.state); }); const syncRoster = Effect.fnUntraced(function* (state: ThreadState) { @@ -1263,6 +1275,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr })); const current = state.providerThread.pendingBackgroundTasks ?? []; if ( + state.publishedBackgroundWork === hasBackground(state) && current.length === roster.length && current.every( (task, index) => @@ -1279,11 +1292,15 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr pendingBackgroundTasks: roster, updatedAt: yield* DateTime.now, }; - yield* emit({ - type: "provider_thread.updated", - driver, - providerThread: state.providerThread, - }); + yield* emitProviderThread(state); + }); + + const syncRosterAncestors = Effect.fnUntraced(function* (state: ThreadState) { + let current: ThreadState | undefined = state; + while (current !== undefined) { + yield* syncRoster(current); + current = current.subagent?.call.state; + } }); const withReportedModel = ( @@ -1394,11 +1411,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr childOwners.set(childId, rootOf(call.state)); call.child = child; yield* emit({ type: "app_thread.created", driver, appThread }); - yield* emit({ - type: "provider_thread.updated", - driver, - providerThread: child.providerThread, - }); + yield* emitProviderThread(child); yield* emitSubagent(call); // A session called again was not made now, so it may hold the rules of // a mode the thread has left. OpenCode applies a rules change to the @@ -1435,11 +1448,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr status: "idle", updatedAt: call.completedAt, }; - yield* emit({ - type: "provider_thread.updated", - driver, - providerThread: child.providerThread, - }); + yield* emitProviderThread(child); } yield* emitSubagent(call); }); @@ -1657,11 +1666,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr status: threadDisposition === "broken" ? "error" : "idle", updatedAt: completedAt, }; - yield* emit({ - type: "provider_thread.updated", - driver, - providerThread: state.providerThread, - }); + yield* emitProviderThread(state); const anyActive = [...threads.values()].some( (candidate) => candidate.subagent === undefined && candidate.active !== undefined, ); @@ -2444,11 +2449,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr const providerThread = withReportedModel(state.providerThread, event.data.model); if (providerThread !== state.providerThread) { state.providerThread = { ...providerThread, updatedAt: yield* DateTime.now }; - yield* emit({ - type: "provider_thread.updated", - driver, - providerThread: state.providerThread, - }); + yield* emitProviderThread(state); } } if ( @@ -2654,10 +2655,12 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr yield* settleCall(call, "interrupted", LOST_BACKGROUND); } } - for (const session of sessionsOf(state)) { + const sessions = sessionsOf(state); + for (const session of sessions) { discardWakes(session, session.wakes.splice(0)); session.reports.clear(); } + for (const session of sessions.toReversed()) yield* syncRoster(session); } // Stopped off the lock this reconcile holds; one still running waits for a Stop. for (const thread of new Set([...strays.values()].map(rootOf))) { @@ -2714,7 +2717,26 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr stream: Stream.Stream, ): Effect.Effect => stream.pipe( - Stream.runForEach((event) => lock.withPermit(handleEvent(event))), + Stream.runForEach((event) => + lock.withPermit( + Effect.gen(function* () { + yield* handleEvent(event); + if ( + event.type === "session.execution.started" || + event.type === "unreadable.execution.started" || + executionEnd(event.type) || + event.type === "unreadable.execution.ended" || + event.type === "session.inbox.enqueued" || + event.type === "session.inbox.delivered" || + event.type === "session.inbox.cancelled" + ) { + const sessionId = sessionOfEvent(event); + const state = sessionId === undefined ? undefined : threads.get(sessionId); + if (state !== undefined) yield* syncRosterAncestors(state); + } + }), + ), + ), Effect.exit, Effect.flatMap(() => Effect.gen(function* () { @@ -3198,11 +3220,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr lastRunOrdinal: turnInput.runOrdinal, updatedAt: startedAt, }; - yield* emit({ - type: "provider_thread.updated", - driver, - providerThread: state.providerThread, - }); + yield* emitProviderThread(state); yield* setSessionStatus("running", null); return turn; }); @@ -3265,6 +3283,8 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr if (wake?.running !== true) return; state.wakes.pop(); yield* replay(wake); + for (const session of sessionsOf(state).toReversed()) yield* syncRoster(session); + yield* syncRosterAncestors(state); }); const runWake = Effect.fnUntraced(function* ( @@ -3292,6 +3312,8 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr } if (wake.running) state.unsettled = false; yield* replay(wake); + for (const session of sessionsOf(state).toReversed()) yield* syncRoster(session); + yield* syncRosterAncestors(state); }), ); }); @@ -3357,6 +3379,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr } const running = state.wakes.some((wake) => wake.running); discardWakes(state, state.wakes.splice(0)); + yield* syncRosterAncestors(state); if (running) { markUnsettled(state); yield* client.session