Open History to see the last commit
1import { assert, describe, it } from "@effect/vitest";
2import type { ProviderInstanceId } from "@t3tools/contracts";
3import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
4import * as NodeServices from "@effect/platform-node/NodeServices";
5import * as Config from "effect/Config";
6import * as Effect from "effect/Effect";
7import * as FileSystem from "effect/FileSystem";
8import * as Layer from "effect/Layer";
9import * as Stream from "effect/Stream";
10
11import { OpenCode2OrchestratorReplayHarness } from "../Adapters/OpenCode2AdapterV2.testkit.ts";
12import * as ProviderAdapterRegistry from "../ProviderAdapterRegistry.ts";
13import { findReplayFixture, runReplayFixture } from "./ProviderReplayFixtures.ts";
14
15// Evidence driver: replays recorded OpenCode 2 fixtures through
16// OrchestratorV2 and records the adapter's own event stream, because the
17// `pendingBackgroundWork` flag is dropped before it reaches domain events.
18// Set EV_TRANSCRIPT to a file path to append the transcript there.
19
20type Row = {
21 readonly id: string;
22 readonly fixture: string;
23 /** True when every native thread must end on a `pendingBackgroundWork: false` snapshot. */
24 readonly requireSettledSnapshots: boolean;
25};
26
27const ROWS: ReadonlyArray<Row> = [
28 { id: "R1", fixture: "opencode2_nested_background", requireSettledSnapshots: true },
29 { id: "R1-compat", fixture: "opencode2_background", requireSettledSnapshots: true },
30 { id: "R2c-compat", fixture: "opencode2_resume_after_restart", requireSettledSnapshots: false },
31 { id: "R2c-compat", fixture: "opencode2_interrupt", requireSettledSnapshots: false },
32 { id: "R3", fixture: "opencode2_simple", requireSettledSnapshots: false },
33 { id: "R3", fixture: "opencode2_tool_call", requireSettledSnapshots: false },
34];
35
36const tasksOf = (tasks: ReadonlyArray<{ readonly taskId: string }> | undefined) =>
37 `[${(tasks ?? []).map((task) => task.taskId).join(",")}]`;
38
39const describeEvent = (event: ProviderAdapter.ProviderAdapterV2Event): string | undefined => {
40 switch (event.type) {
41 case "provider_thread.updated":
42 return `provider_thread.updated native=${event.providerThread.nativeThreadRef?.nativeId ?? "?"} pendingBackgroundWork=${String(event.pendingBackgroundWork)} pendingBackgroundTasks=${tasksOf(event.providerThread.pendingBackgroundTasks)}`;
43 case "provider_turn.updated":
44 return `provider_turn.updated native=${event.providerTurn.nativeTurnRef?.nativeId ?? "?"} status=${event.providerTurn.status}`;
45 case "subagent.updated":
46 return `subagent.updated title=${event.subagent.title ?? "?"} status=${event.subagent.status}`;
47 case "turn.terminal":
48 return `turn.terminal status=${event.status}`;
49 default:
50 return undefined;
51 }
52};
53
54const tappedHarness = (
55 record: (event: ProviderAdapter.ProviderAdapterV2Event) => void,
56): typeof OpenCode2OrchestratorReplayHarness => ({
57 ...OpenCode2OrchestratorReplayHarness,
58 makeProviderAdapterRegistryLayer: (
59 ...args: Parameters<typeof OpenCode2OrchestratorReplayHarness.makeProviderAdapterRegistryLayer>
60 ) =>
61 Layer.effect(
62 ProviderAdapterRegistry.ProviderAdapterRegistryV2,
63 Effect.gen(function* () {
64 const registry = yield* ProviderAdapterRegistry.ProviderAdapterRegistryV2;
65 return ProviderAdapterRegistry.ProviderAdapterRegistryV2.of({
66 ...registry,
67 get: (instanceId: ProviderInstanceId) =>
68 registry.get(instanceId).pipe(
69 Effect.map((adapter) =>
70 ProviderAdapter.ProviderAdapterV2.of({
71 ...adapter,
72 openSession: (input) =>
73 adapter.openSession(input).pipe(
74 Effect.map((runtime) => ({
75 ...runtime,
76 events: runtime.events.pipe(
77 Stream.tap((event) => Effect.sync(() => record(event))),
78 ),
79 })),
80 ),
81 }),
82 ),
83 ),
84 });
85 }),
86 ).pipe(
87 Layer.provide(OpenCode2OrchestratorReplayHarness.makeProviderAdapterRegistryLayer(...args)),
88 ),
89});
90
91const runRow = (row: Row) =>
92 Effect.gen(function* () {
93 const fixture = findReplayFixture(row.fixture);
94 const driver = fixture?.providers.find((provider) => provider.driver === "opencode");
95 assert.isDefined(fixture);
96 assert.isDefined(driver);
97 const events: Array<ProviderAdapter.ProviderAdapterV2Event> = [];
98 const lines: Array<string> = [];
99 const emit = (line: string) => lines.push(line);
100 emit(`== ${row.id} ${row.fixture} (recorded fixture) ==`);
101 const exit = yield* runReplayFixture({
102 fixtureName: row.fixture,
103 buildInput: fixture.buildInput,
104 driver,
105 harness: tappedHarness((event) => events.push(event)),
106 }).pipe(Effect.exit);
107 let index = 0;
108 for (const event of events) {
109 const line = describeEvent(event);
110 if (line !== undefined) emit(` adapter[${index}] ${line}`);
111 index += 1;
112 }
113 const lastSnapshot = new Map<string, boolean | undefined>();
114 let everPending = false;
115 for (const event of events) {
116 if (event.type !== "provider_thread.updated") continue;
117 const nativeId = event.providerThread.nativeThreadRef?.nativeId ?? "?";
118 lastSnapshot.set(nativeId, event.pendingBackgroundWork);
119 if (event.pendingBackgroundWork === true) everPending = true;
120 }
121 const finals = [...lastSnapshot].map(([id, flag]) => `${id}=${String(flag)}`).join(" ");
122 emit(` final snapshot per native thread: ${finals}`);
123 if (exit._tag === "Success") {
124 for (const domain of exit.value.domainEvents) {
125 if (domain.type !== "provider-thread.updated") continue;
126 emit(
127 ` domain provider-thread.updated native=${domain.payload.nativeThreadRef?.nativeId ?? "?"} pendingBackgroundTasks=${tasksOf(domain.payload.pendingBackgroundTasks)}`,
128 );
129 }
130 }
131 const failures: Array<string> = [];
132 if (exit._tag === "Failure") failures.push("fixture output assertions failed");
133 if (row.requireSettledSnapshots) {
134 if (!everPending) failures.push("no pendingBackgroundWork=true snapshot was published");
135 for (const [id, flag] of lastSnapshot) {
136 if (flag !== false) failures.push(`${id} ended on pendingBackgroundWork=${String(flag)}`);
137 }
138 } else if ([...lastSnapshot.values()].includes(true)) {
139 failures.push("a native thread ended with pending work");
140 }
141 emit(
142 ` ${row.id} ${row.fixture}: ${failures.length === 0 ? "PASS" : `FAIL (${failures.join("; ")})`}`,
143 );
144 const transcript = yield* Config.String("EV_TRANSCRIPT").pipe(Config.withDefault(""));
145 if (transcript !== "") {
146 const fs = yield* FileSystem.FileSystem;
147 yield* fs.writeFileString(transcript, `${lines.join("\n")}\n`, { flag: "a" });
148 }
149 return failures;
150 }).pipe(Effect.provide(NodeServices.layer));
151
152describe("OpenCode 2 pending-work snapshots (#19228 evidence)", () => {
153 it.effect.each(ROWS.map((row) => [`${row.id} ${row.fixture}`, row] as const))("%s", ([, row]) =>
154 runRow(row).pipe(
155 Effect.map((failures) => assert.deepEqual(failures, [])),
156 Effect.scoped,
157 ),
158 );
159});