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
131/**
132 * Replays one transcript at a client boundary: each outbound call must match the
133 * next `expect_outbound`, then answers with its `sdk.response`; `sdk.event`
134 * frames feed the event stream. Shared by the 1.x SDK and 2.x HTTP transports.
135 */
136export class OpenCodeReplayController {
137 private cursor = 0;
138 /** Set once the event stream reached a successful `runtime_exit`: the server is gone. */
139 exited = false;
140
141 /** Whether every entry has been used, so a stopped server is not coming back. */
142 get finished(): boolean {
143 return this.cursor >= this.transcript.entries.length;
144 }
145 private readonly waiters = new Set<() => void>();
146 private failure: unknown = null;
147 private readonly transcript: ProviderReplayTranscript;
148 private messageIds = new Map<string, string>();
149
150 constructor(transcript: ProviderReplayTranscript) {
151 this.transcript = transcript;
152 }
153
154 async expectOutbound(actual: unknown): Promise<void> {
155 try {
156 const entry = this.transcript.entries[this.cursor];
157 const expectedFrame = entry?.type === "expect_outbound" ? frameRecord(entry.frame) : null;
158 const actualFrame = frameRecord(actual);
159 const messageIds = new Map(this.messageIds);
160 let conflictingMessageId = false;
161 // A message id the client picks (1.x prompts, 2.x steers) replaces the
162 // recorded one everywhere after it, so replies and events carry the live id.
163 const idKey = expectedFrame?.type === "session.promptAsync" ? "messageID" : "id";
164 if (
165 (expectedFrame?.type === "session.promptAsync" ||
166 expectedFrame?.type === "session.prompt") &&
167 actualFrame?.type === expectedFrame.type
168 ) {
169 const recordedId = frameRecord(expectedFrame.input)?.[idKey];
170 const actualId = frameRecord(actualFrame.input)?.[idKey];
171 if (
172 typeof recordedId === "string" &&
173 typeof actualId === "string" &&
174 recordedId !== "<any>" &&
175 recordedId !== "<workspace>" &&
176 !messageIds.has(recordedId)
177 ) {
178 conflictingMessageId = [...messageIds.values()].includes(actualId);
179 if (!conflictingMessageId) messageIds.set(recordedId, actualId);
180 }
181 }
182 const expected =
183 entry?.type === "expect_outbound"
184 ? materializeMessageIds(entry.frame, messageIds)
185 : (entry ?? null);
186 if (
187 conflictingMessageId ||
188 entry?.type !== "expect_outbound" ||
189 !replayValueMatches(expected, actual)
190 ) {
191 throw new OpenCodeReplayMismatchError({
192 scenario: this.transcript.scenario,
193 cursor: this.cursor,
194 expected,
195 actual,
196 });
197 }
198 this.messageIds = messageIds;
199 this.advance();
200 } catch (cause) {
201 this.fail(cause);
202 throw cause;
203 }
204 }
205
206 async response(operation: string): Promise<unknown> {
207 while (true) {
208 this.throwFailure();
209 const entry = this.transcript.entries[this.cursor];
210 if (entry?.type === "emit_inbound") {
211 const frame = frameRecord(entry.frame);
212 if (frame?.type === "sdk.response" && frame.operation === operation) {
213 if (entry.afterMs !== undefined && entry.afterMs > 0) {
214 await Effect.runPromise(Effect.sleep(Duration.millis(entry.afterMs)));
215 }
216 const data = materializeMessageIds(frame.data, this.messageIds);
217 this.advance();
218 return data;
219 }
220 }
221 if (entry?.type === "runtime_exit") {
222 const mismatch = new OpenCodeReplayMismatchError({
223 scenario: this.transcript.scenario,
224 cursor: this.cursor,
225 expected: { type: "sdk.response", operation },
226 actual: entry,
227 });
228 this.fail(mismatch);
229 throw mismatch;
230 }
231 await this.changed();
232 }
233 }
234
235 /**
236 * Resolves once the server events recorded before the next entry have been
237 * delivered. For transports whose events and requests travel separately, a
238 * request is matched at its recorded point instead of racing those events.
239 */
240 async untilEventsDelivered(): Promise<void> {
241 while (true) {
242 this.throwFailure();
243 const entry = this.transcript.entries[this.cursor];
244 if (entry?.type !== "emit_inbound" || frameRecord(entry.frame)?.type !== "sdk.event") return;
245 await this.changed();
246 }
247 }
248
249 /**
250 * Like {@link untilEventsDelivered}, but also waits for recorded responses to
251 * be taken: over HTTP a request can start while an earlier one's answer is
252 * still in flight, and must not be matched against that answer.
253 */
254 async untilInboundDelivered(): Promise<void> {
255 while (true) {
256 this.throwFailure();
257 if (this.transcript.entries[this.cursor]?.type !== "emit_inbound") return;
258 await this.changed();
259 }
260 }
261
262 async *events(
263 signal?: AbortSignal,
264 beforeEmit?: (label: string | undefined) => Promise<void>,
265 ): AsyncIterable<unknown> {
266 while (true) {
267 if (signal?.aborted === true) return;
268 this.throwFailure();
269 const entry = this.transcript.entries[this.cursor];
270 if (entry?.type === "emit_inbound") {
271 const frame = frameRecord(entry.frame);
272 if (frame?.type === "sdk.event") {
273 // A scenario can hold an event until it has done what happened meanwhile.
274 if (beforeEmit !== undefined) await beforeEmit(entry.label);
275 if (entry.afterMs !== undefined && entry.afterMs > 0) {
276 await Effect.runPromise(Effect.sleep(Duration.millis(entry.afterMs)));
277 }
278 const event = materializeMessageIds(frame.event, this.messageIds);
279 this.advance();
280 yield event;
281 continue;
282 }
283 }
284 if (entry?.type === "runtime_exit") {
285 this.advance();
286 if (entry.status === "success") {
287 this.exited = true;
288 return;
289 }
290 const mismatch = new OpenCodeReplayMismatchError({
291 scenario: this.transcript.scenario,
292 cursor: this.cursor - 1,
293 expected: { status: "success" },
294 actual: entry,
295 });
296 this.fail(mismatch);
297 throw mismatch;
298 }
299 await this.changed(signal);
300 }
301 }
302
303 assertComplete(): void {
304 while (this.transcript.entries[this.cursor]?.type === "runtime_exit") {
305 const exit = this.transcript.entries[this.cursor];
306 if (exit?.type !== "runtime_exit" || exit.status !== "success") break;
307 this.cursor += 1;
308 }
309 this.throwFailure();
310 if (this.cursor !== this.transcript.entries.length) {
311 throw new OpenCodeReplayIncompleteError({
312 scenario: this.transcript.scenario,
313 cursor: this.cursor,
314 remaining: this.transcript.entries.length - this.cursor,
315 });
316 }
317 }
318
319 private advance(): void {
320 this.cursor += 1;
321 for (const waiter of this.waiters) waiter();
322 this.waiters.clear();
323 }
324
325 private fail(cause: unknown): void {
326 this.failure = cause;
327 for (const waiter of this.waiters) waiter();
328 this.waiters.clear();
329 }
330
331 private throwFailure(): void {
332 if (this.failure !== null) throw this.failure;
333 }
334
335 private changed(signal?: AbortSignal): Promise<void> {
336 return new Promise((resolve) => {
337 let settled = false;
338 const done = () => {
339 if (settled) return;
340 settled = true;
341 signal?.removeEventListener("abort", done);
342 this.waiters.delete(done);
343 resolve();
344 };
345 this.waiters.add(done);
346 signal?.addEventListener("abort", done, { once: true });
347 if (signal?.aborted === true) done();
348 });
349 }
350}
351
352function makeReplayClient(controller: OpenCodeReplayController): OpencodeClient {
353 const request = async (operation: string, input: unknown) => {
354 await controller.expectOutbound({ type: operation, input });
355 return { data: await controller.response(operation) };
356 };
357 return {
358 event: {
359 subscribe: async (_input?: unknown, options?: { readonly signal?: AbortSignal }) => {
360 await controller.expectOutbound({ type: "event.subscribe" });
361 return { stream: controller.events(options?.signal) };
362 },
363 },
364 session: {
365 create: (input: unknown) => request("session.create", input),
366 get: (input: unknown) => request("session.get", input),
367 children: (input: unknown) => request("session.children", input),
368 update: (input: unknown) => request("session.update", input),
369 messages: (input: unknown) => request("session.messages", input),
370 promptAsync: (input: unknown) => request("session.promptAsync", input),
371 abort: (input: unknown) => request("session.abort", input),
372 revert: (input: unknown) => request("session.revert", input),
373 unrevert: (input: unknown) => request("session.unrevert", input),
374 fork: (input: unknown) => request("session.fork", input),
375 },
376 permission: {
377 reply: (input: unknown) => request("permission.reply", input),
378 },
379 question: {
380 reply: (input: unknown) => request("question.reply", input),
381 },
382 mcp: {
383 status: () => request("mcp.status", {}),
384 add: (input: unknown) => request("mcp.add", input),
385 },
386 } as unknown as OpencodeClient;
387}
388
389function layerOpenCodeReplayRuntime(transcript: OpenCodeSdkReplayTranscript) {
390 return Layer.effect(
391 OpenCodeRuntime.OpenCodeRuntime,
392 Effect.gen(function* () {
393 const controller = new OpenCodeReplayController(transcript);
394 yield* Effect.addFinalizer(() =>
395 Effect.sync(() => {
396 controller.assertComplete();
397 }),
398 );
399 const client = makeReplayClient(controller);
400 return OpenCodeRuntime.OpenCodeRuntime.of({
401 startOpenCodeServerProcess: () =>
402 Effect.fail(
403 new OpenCodeRuntime.OpenCodeRuntimeError({
404 operation: "startOpenCodeServerProcess",
405 detail: "OpenCode replay uses an external in-memory SDK boundary.",
406 }),
407 ),
408 connectToOpenCodeServer: () =>
409 Effect.succeed({
410 url: "replay://opencode",
411 version: "0.0.0-replay",
412 exitCode: null,
413 external: true,
414 }),
415 runOpenCodeCommand: () =>
416 Effect.fail(
417 new OpenCodeRuntime.OpenCodeRuntimeError({
418 operation: "runOpenCodeCommand",
419 detail: "OpenCode replay does not execute commands.",
420 }),
421 ),
422 createOpenCodeSdkClient: () => client,
423 loadOpenCodeInventory: () =>
424 Effect.fail(
425 new OpenCodeRuntime.OpenCodeRuntimeError({
426 operation: "loadOpenCodeInventory",
427 detail: "OpenCode replay does not load inventory.",
428 }),
429 ),
430 loadInventoryFromCli: () =>
431 Effect.fail(
432 new OpenCodeRuntime.OpenCodeRuntimeError({
433 operation: "loadInventoryFromCli",
434 detail: "OpenCode replay does not load inventory.",
435 }),
436 ),
437 loadOpenCodeSkills: () =>
438 Effect.fail(
439 new OpenCodeRuntime.OpenCodeRuntimeError({
440 operation: "loadOpenCodeSkills",
441 detail: "OpenCode replay does not load skills.",
442 }),
443 ),
444 loadSkillsFromCli: () =>
445 Effect.fail(
446 new OpenCodeRuntime.OpenCodeRuntimeError({
447 operation: "loadSkillsFromCli",
448 detail: "OpenCode replay does not load skills.",
449 }),
450 ),
451 } satisfies OpenCodeRuntime.OpenCodeRuntime["Service"]);
452 }),
453 );
454}
455
456function layerOpenCodeProviderAdapterRegistryReplay(transcript: OpenCodeSdkReplayTranscript) {
457 return ProviderAdapterRegistry.layerFromDrivers({
458 drivers: [OpenCodeAdapterV2Driver],
459 configMap: {
460 [OPENCODE_DEFAULT_INSTANCE_ID]: {
461 driver: OPENCODE_PROVIDER,
462 config: { serverUrl: "replay://opencode" },
463 },
464 },
465 }).pipe(
466 Layer.provide(
467 Layer.mergeAll(
468 layerOpenCodeReplayRuntime(transcript),
469 TestProviderHost.layer().pipe(Layer.provide(NodeServices.layer)),
470 NodeServices.layer,
471 IdAllocator.layer,
472 Layer.succeed(
473 ProviderEventLoggers.ProviderEventLoggers,
474 ProviderEventLoggers.NoOpProviderEventLoggers,
475 ),
476 ),
477 ),
478 );
479}
480
481function transcriptMetadata(transcript: ProviderReplayTranscript) {
482 return {
483 provider: transcript.provider,
484 protocol: transcript.protocol,
485 scenario: transcript.scenario,
486 };
487}
488
489export const OpenCodeOrchestratorReplayHarness: OrchestratorV2ProviderReplayHarness<
490 OpenCodeSdkReplayTranscript,
491 OpenCodeOrchestratorReplayHarnessError
492> = {
493 driver: OPENCODE_PROVIDER,
494 decodeTranscript: (transcript) =>
495 decodeOpenCodeSdkReplayTranscript(transcript).pipe(
496 Effect.mapError(
497 (cause) =>
498 new OpenCodeReplayTranscriptDecodeError({
499 ...transcriptMetadata(transcript),
500 cause,
501 }),
502 ),
503 ),
504 makeProviderAdapterRegistryLayer: layerOpenCodeProviderAdapterRegistryReplay,
505};