1import type { OpencodeClient } from "@opencode-ai/sdk/v2";
2import * as NodeServices from "@effect/platform-node/NodeServices";
3import { ProviderReplayEntry, type ProviderReplayTranscript } from "@t3tools/contracts";
4import * as Effect from "effect/Effect";
5import * as Duration from "effect/Duration";
6import * as Layer from "effect/Layer";
7import * as Schema from "effect/Schema";
8
9import { OpenCodeAdapterV2Driver } from "@t3tools/provider-opencode/server";
10import * as OpenCodeRuntime from "@t3tools/provider-opencode/server/OpenCodeRuntime";
11import * as TestProviderHost from "@t3tools/provider-testing/TestProviderHost";
12import * as ProviderEventLoggers from "@t3tools/provider-core/server/ProviderEventLoggers";
13import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
14import { ProviderAdapterDriverCreateError } from "@t3tools/provider-core/server/adapterDriver";
15import * as ProviderAdapterRegistry from "../ProviderAdapterRegistry.ts";
16import { type OrchestratorV2ProviderReplayHarness } from "../testkit/ProviderReplayHarness.ts";
17import {
18 OPENCODE_DEFAULT_INSTANCE_ID,
19 OPENCODE_PROVIDER,
20 OPENCODE_SDK_PROTOCOL,
21} from "@t3tools/provider-opencode/testing";
22
23const OPENCODE_SDK_REPLAY_PROTOCOL = OPENCODE_SDK_PROTOCOL;
24
25const OpenCodeSdkReplayTranscript = Schema.Struct({
26 provider: Schema.Literal(OPENCODE_PROVIDER),
27 protocol: Schema.Literal(OPENCODE_SDK_REPLAY_PROTOCOL),
28 version: Schema.String,
29 scenario: Schema.String,
30 metadata: Schema.optional(Schema.Record(Schema.String, Schema.Unknown)),
31 entries: Schema.Array(ProviderReplayEntry),
32});
33export type OpenCodeSdkReplayTranscript = typeof OpenCodeSdkReplayTranscript.Type;
34const decodeOpenCodeSdkReplayTranscript = Schema.decodeUnknownEffect(OpenCodeSdkReplayTranscript);
35
36export class OpenCodeReplayTranscriptDecodeError extends Schema.TaggedError<OpenCodeReplayTranscriptDecodeError>()(
37 "OpenCodeReplayTranscriptDecodeError",
38 {
39 driver: Schema.optional(Schema.String),
40 protocol: Schema.optional(Schema.String),
41 scenario: Schema.optional(Schema.String),
42 cause: Schema.Defect(),
43 },
44) {
45 override get message(): string {
46 return `Failed to decode OpenCode replay transcript for scenario ${this.scenario ?? "<unknown>"}.`;
47 }
48}
49
50export class OpenCodeReplayMismatchError extends Schema.TaggedError<OpenCodeReplayMismatchError>()(
51 "OpenCodeReplayMismatchError",
52 {
53 scenario: Schema.String,
54 cursor: Schema.Number,
55 expected: Schema.Unknown,
56 actual: Schema.Unknown,
57 },
58) {
59 override get message(): string {
60 return `OpenCode replay frame mismatch at cursor ${this.cursor} in scenario ${this.scenario}.`;
61 }
62}
63
64export class OpenCodeReplayIncompleteError extends Schema.TaggedError<OpenCodeReplayIncompleteError>()(
65 "OpenCodeReplayIncompleteError",
66 {
67 scenario: Schema.String,
68 cursor: Schema.Number,
69 remaining: Schema.Number,
70 },
71) {
72 override get message(): string {
73 return `OpenCode replay ended with ${this.remaining} unconsumed entries in scenario ${this.scenario}.`;
74 }
75}
76
77export const OpenCodeReplayError = Schema.Union([
78 OpenCodeReplayTranscriptDecodeError,
79 OpenCodeReplayMismatchError,
80 OpenCodeReplayIncompleteError,
81]);
82export type OpenCodeReplayError = typeof OpenCodeReplayError.Type;
83export const OpenCodeOrchestratorReplayHarnessError = Schema.Union([
84 OpenCodeReplayError,
85 ProviderAdapterDriverCreateError,
86]);
87export type OpenCodeOrchestratorReplayHarnessError =
88 typeof OpenCodeOrchestratorReplayHarnessError.Type;
89
90function replayValueMatches(expected: unknown, actual: unknown): boolean {
91 if (expected === "<any>" || expected === "<workspace>") return true;
92 if (Array.isArray(expected)) {
93 return (
94 Array.isArray(actual) &&
95 expected.length === actual.length &&
96 expected.every((entry, index) => replayValueMatches(entry, actual[index]))
97 );
98 }
99 if (typeof expected === "object" && expected !== null) {
100 if (typeof actual !== "object" || actual === null) return false;
101 return Object.entries(expected).every(([key, value]) =>
102 replayValueMatches(value, (actual as Record<string, unknown>)[key]),
103 );
104 }
105 return Object.is(expected, actual);
106}
107
108function frameRecord(frame: unknown): Record<string, unknown> | null {
109 return typeof frame === "object" && frame !== null ? (frame as Record<string, unknown>) : null;
110}
111
112function materializeMessageIds(value: unknown, messageIds: ReadonlyMap<string, string>): unknown {
113 if (Array.isArray(value)) return value.map((entry) => materializeMessageIds(entry, messageIds));
114 const record = frameRecord(value);
115 if (record === null) return value;
116 return Object.fromEntries(
117 Object.entries(record).map(([key, entry]) => [
118 key,
119 typeof entry === "string" &&
120 (key === "id" ||
121 key === "messageID" ||
122 key === "parentID" ||
123 key === "inboxID" ||
124 key === "before")
125 ? (messageIds.get(entry) ?? entry)
126 : materializeMessageIds(entry, messageIds),
127 ]),
128 );
129}
130
131export class OpenCodeReplayController {
132 private cursor = 0;
133 exited = false;
134
135 get finished(): boolean {
136 return this.cursor >= this.transcript.entries.length;
137 }
138 private readonly waiters = new Set<() => void>();
139 private failure: unknown = null;
140 private readonly transcript: ProviderReplayTranscript;
141 private messageIds = new Map<string, string>();
142
143 constructor(transcript: ProviderReplayTranscript) {
144 this.transcript = transcript;
145 }
146
147 async expectOutbound(actual: unknown): Promise<void> {
148 try {
149 const entry = this.transcript.entries[this.cursor];
150 const expectedFrame = entry?.type === "expect_outbound" ? frameRecord(entry.frame) : null;
151 const actualFrame = frameRecord(actual);
152 const messageIds = new Map(this.messageIds);
153 let conflictingMessageId = false;
154 const idKey = expectedFrame?.type === "session.promptAsync" ? "messageID" : "id";
155 if (
156 (expectedFrame?.type === "session.promptAsync" ||
157 expectedFrame?.type === "session.prompt") &&
158 actualFrame?.type === expectedFrame.type
159 ) {
160 const recordedId = frameRecord(expectedFrame.input)?.[idKey];
161 const actualId = frameRecord(actualFrame.input)?.[idKey];
162 if (
163 typeof recordedId === "string" &&
164 typeof actualId === "string" &&
165 recordedId !== "<any>" &&
166 recordedId !== "<workspace>" &&
167 !messageIds.has(recordedId)
168 ) {
169 conflictingMessageId = [...messageIds.values()].includes(actualId);
170 if (!conflictingMessageId) messageIds.set(recordedId, actualId);
171 }
172 }
173 const expected =
174 entry?.type === "expect_outbound"
175 ? materializeMessageIds(entry.frame, messageIds)
176 : (entry ?? null);
177 if (
178 conflictingMessageId ||
179 entry?.type !== "expect_outbound" ||
180 !replayValueMatches(expected, actual)
181 ) {
182 throw new OpenCodeReplayMismatchError({
183 scenario: this.transcript.scenario,
184 cursor: this.cursor,
185 expected,
186 actual,
187 });
188 }
189 this.messageIds = messageIds;
190 this.advance();
191 } catch (cause) {
192 this.fail(cause);
193 throw cause;
194 }
195 }
196
197 async response(operation: string): Promise<unknown> {
198 while (true) {
199 this.throwFailure();
200 const entry = this.transcript.entries[this.cursor];
201 if (entry?.type === "emit_inbound") {
202 const frame = frameRecord(entry.frame);
203 if (frame?.type === "sdk.response" && frame.operation === operation) {
204 if (entry.afterMs !== undefined && entry.afterMs > 0) {
205 await Effect.runPromise(Effect.sleep(Duration.millis(entry.afterMs)));
206 }
207 const data = materializeMessageIds(frame.data, this.messageIds);
208 this.advance();
209 return data;
210 }
211 }
212 if (entry?.type === "runtime_exit") {
213 const mismatch = new OpenCodeReplayMismatchError({
214 scenario: this.transcript.scenario,
215 cursor: this.cursor,
216 expected: { type: "sdk.response", operation },
217 actual: entry,
218 });
219 this.fail(mismatch);
220 throw mismatch;
221 }
222 await this.changed();
223 }
224 }
225
226 async untilEventsDelivered(): Promise<void> {
227 while (true) {
228 this.throwFailure();
229 const entry = this.transcript.entries[this.cursor];
230 if (entry?.type !== "emit_inbound" || frameRecord(entry.frame)?.type !== "sdk.event") return;
231 await this.changed();
232 }
233 }
234
235 /** Waits for events and responses to be consumed so concurrent HTTP requests cannot match earlier answers. */
236 async untilInboundDelivered(): Promise<void> {
237 while (true) {
238 this.throwFailure();
239 if (this.transcript.entries[this.cursor]?.type !== "emit_inbound") return;
240 await this.changed();
241 }
242 }
243
244 async *events(
245 signal?: AbortSignal,
246 beforeEmit?: (label: string | undefined) => Promise<void>,
247 ): AsyncIterable<unknown> {
248 while (true) {
249 if (signal?.aborted === true) return;
250 this.throwFailure();
251 const entry = this.transcript.entries[this.cursor];
252 if (entry?.type === "emit_inbound") {
253 const frame = frameRecord(entry.frame);
254 if (frame?.type === "sdk.event") {
255 if (beforeEmit !== undefined) await beforeEmit(entry.label);
256 if (entry.afterMs !== undefined && entry.afterMs > 0) {
257 await Effect.runPromise(Effect.sleep(Duration.millis(entry.afterMs)));
258 }
259 const event = materializeMessageIds(frame.event, this.messageIds);
260 this.advance();
261 yield event;
262 continue;
263 }
264 }
265 if (entry?.type === "runtime_exit") {
266 this.advance();
267 if (entry.status === "success") {
268 this.exited = true;
269 return;
270 }
271 const mismatch = new OpenCodeReplayMismatchError({
272 scenario: this.transcript.scenario,
273 cursor: this.cursor - 1,
274 expected: { status: "success" },
275 actual: entry,
276 });
277 this.fail(mismatch);
278 throw mismatch;
279 }
280 await this.changed(signal);
281 }
282 }
283
284 assertComplete(): void {
285 while (this.transcript.entries[this.cursor]?.type === "runtime_exit") {
286 const exit = this.transcript.entries[this.cursor];
287 if (exit?.type !== "runtime_exit" || exit.status !== "success") break;
288 this.cursor += 1;
289 }
290 this.throwFailure();
291 if (this.cursor !== this.transcript.entries.length) {
292 throw new OpenCodeReplayIncompleteError({
293 scenario: this.transcript.scenario,
294 cursor: this.cursor,
295 remaining: this.transcript.entries.length - this.cursor,
296 });
297 }
298 }
299
300 private advance(): void {
301 this.cursor += 1;
302 for (const waiter of this.waiters) waiter();
303 this.waiters.clear();
304 }
305
306 private fail(cause: unknown): void {
307 this.failure = cause;
308 for (const waiter of this.waiters) waiter();
309 this.waiters.clear();
310 }
311
312 private throwFailure(): void {
313 if (this.failure !== null) throw this.failure;
314 }
315
316 private changed(signal?: AbortSignal): Promise<void> {
317 return new Promise((resolve) => {
318 let settled = false;
319 const done = () => {
320 if (settled) return;
321 settled = true;
322 signal?.removeEventListener("abort", done);
323 this.waiters.delete(done);
324 resolve();
325 };
326 this.waiters.add(done);
327 signal?.addEventListener("abort", done, { once: true });
328 if (signal?.aborted === true) done();
329 });
330 }
331}
332
333function makeReplayClient(controller: OpenCodeReplayController): OpencodeClient {
334 const request = async (operation: string, input: unknown) => {
335 await controller.expectOutbound({ type: operation, input });
336 return { data: await controller.response(operation) };
337 };
338 return {
339 event: {
340 subscribe: async (_input?: unknown, options?: { readonly signal?: AbortSignal }) => {
341 await controller.expectOutbound({ type: "event.subscribe" });
342 return { stream: controller.events(options?.signal) };
343 },
344 },
345 session: {
346 create: (input: unknown) => request("session.create", input),
347 get: (input: unknown) => request("session.get", input),
348 children: (input: unknown) => request("session.children", input),
349 update: (input: unknown) => request("session.update", input),
350 messages: (input: unknown) => request("session.messages", input),
351 promptAsync: (input: unknown) => request("session.promptAsync", input),
352 abort: (input: unknown) => request("session.abort", input),
353 revert: (input: unknown) => request("session.revert", input),
354 unrevert: (input: unknown) => request("session.unrevert", input),
355 fork: (input: unknown) => request("session.fork", input),
356 },
357 permission: {
358 reply: (input: unknown) => request("permission.reply", input),
359 },
360 question: {
361 reply: (input: unknown) => request("question.reply", input),
362 },
363 mcp: {
364 status: () => request("mcp.status", {}),
365 add: (input: unknown) => request("mcp.add", input),
366 },
367 } as unknown as OpencodeClient;
368}
369
370function layerOpenCodeReplayRuntime(transcript: OpenCodeSdkReplayTranscript) {
371 return Layer.effect(
372 OpenCodeRuntime.OpenCodeRuntime,
373 Effect.gen(function* () {
374 const controller = new OpenCodeReplayController(transcript);
375 yield* Effect.addFinalizer(() =>
376 Effect.sync(() => {
377 controller.assertComplete();
378 }),
379 );
380 const client = makeReplayClient(controller);
381 return OpenCodeRuntime.OpenCodeRuntime.of({
382 startOpenCodeServerProcess: () =>
383 Effect.fail(
384 new OpenCodeRuntime.OpenCodeRuntimeError({
385 operation: "startOpenCodeServerProcess",
386 detail: "OpenCode replay uses an external in-memory SDK boundary.",
387 }),
388 ),
389 connectToOpenCodeServer: () =>
390 Effect.succeed({
391 url: "replay://opencode",
392 version: "0.0.0-replay",
393 exitCode: null,
394 external: true,
395 }),
396 runOpenCodeCommand: () =>
397 Effect.fail(
398 new OpenCodeRuntime.OpenCodeRuntimeError({
399 operation: "runOpenCodeCommand",
400 detail: "OpenCode replay does not execute commands.",
401 }),
402 ),
403 createOpenCodeSdkClient: () => client,
404 loadOpenCodeInventory: () =>
405 Effect.fail(
406 new OpenCodeRuntime.OpenCodeRuntimeError({
407 operation: "loadOpenCodeInventory",
408 detail: "OpenCode replay does not load inventory.",
409 }),
410 ),
411 loadOpenCodeSkills: () =>
412 Effect.fail(
413 new OpenCodeRuntime.OpenCodeRuntimeError({
414 operation: "loadOpenCodeSkills",
415 detail: "OpenCode replay does not load skills.",
416 }),
417 ),
418 } satisfies OpenCodeRuntime.OpenCodeRuntime["Service"]);
419 }),
420 );
421}
422
423function layerOpenCodeProviderAdapterRegistryReplay(transcript: OpenCodeSdkReplayTranscript) {
424 return ProviderAdapterRegistry.layerFromAdaptersEffect(
425 OpenCodeAdapterV2Driver.create({
426 instanceId: OPENCODE_DEFAULT_INSTANCE_ID,
427 environment: [],
428 displayName: undefined,
429 enabled: true,
430 config: { ...OpenCodeAdapterV2Driver.defaultConfig(), serverUrl: "replay://opencode" },
431 }).pipe(Effect.map((adapter) => [adapter])),
432 ).pipe(
433 Layer.provide(
434 Layer.mergeAll(
435 layerOpenCodeReplayRuntime(transcript),
436 TestProviderHost.layer().pipe(Layer.provide(NodeServices.layer)),
437 NodeServices.layer,
438 IdAllocator.layer,
439 Layer.succeed(
440 ProviderEventLoggers.ProviderEventLoggers,
441 ProviderEventLoggers.NoOpProviderEventLoggers,
442 ),
443 ),
444 ),
445 );
446}
447
448function transcriptMetadata(transcript: ProviderReplayTranscript) {
449 return {
450 provider: transcript.provider,
451 protocol: transcript.protocol,
452 scenario: transcript.scenario,
453 };
454}
455
456export const OpenCodeOrchestratorReplayHarness: OrchestratorV2ProviderReplayHarness<
457 OpenCodeSdkReplayTranscript,
458 OpenCodeOrchestratorReplayHarnessError
459> = {
460 driver: OPENCODE_PROVIDER,
461 decodeTranscript: (transcript) =>
462 decodeOpenCodeSdkReplayTranscript(transcript).pipe(
463 Effect.mapError(
464 (cause) =>
465 new OpenCodeReplayTranscriptDecodeError({
466 ...transcriptMetadata(transcript),
467 cause,
468 }),
469 ),
470 ),
471 makeProviderAdapterRegistryLayer: layerOpenCodeProviderAdapterRegistryReplay,
472};