1import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
2import * as NodeServices from "@effect/platform-node/NodeServices";
3import {
4 ProviderInstanceId,
5 ProviderReplayEntry,
6 ProviderSessionId,
7 ThreadId,
8 type ProviderReplayTranscript,
9} from "@t3tools/contracts";
10import * as Effect from "effect/Effect";
11import * as FileSystem from "effect/FileSystem";
12import * as Layer from "effect/Layer";
13import * as Schema from "effect/Schema";
14import * as HttpClient from "effect/http/HttpClient";
15import * as HttpClientError from "effect/http/HttpClientError";
16import * as HttpClientResponse from "effect/http/HttpClientResponse";
17import * as UrlParams from "effect/http/UrlParams";
18
19import * as OpenCode2Client from "@t3tools/provider-opencode/server/v2/OpenCode2Client";
20import * as OpenCode2Server from "@t3tools/provider-opencode/server/v2/OpenCode2Server";
21import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
22import * as ProviderHost from "@t3tools/provider-core/server/ProviderHost";
23import { resolveAttachmentPath } from "../../attachmentStore.ts";
24import * as ProviderAdapterRegistry from "../ProviderAdapterRegistry.ts";
25import type { ProviderReplayGate } from "@t3tools/provider-testing/replayGate";
26import type { OrchestratorV2ProviderReplayHarness } from "../testkit/ProviderReplayHarness.ts";
27import * as TestProviderHost from "@t3tools/provider-testing/TestProviderHost";
28import { OPENCODE_PROVIDER } from "@t3tools/provider-opencode/testing";
29import {
30 OpenCodeReplayController,
31 OpenCodeReplayTranscriptDecodeError,
32} from "./OpenCodeAdapterV2.testkit.ts";
33import * as OpenCode2Adapter from "@t3tools/provider-opencode/server/v2/adapter";
34
35export const OPENCODE2_HTTP_PROTOCOL = "opencode2-http.sse" as const;
36const BASE_URL = "http://opencode2.replay";
37const decodeJson = Schema.decodeUnknownSync(Schema.fromJsonString(Schema.Unknown));
38const encodeJson = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown));
39
40const OpenCode2ReplayTranscript = Schema.Struct({
41 provider: Schema.Literal(OPENCODE_PROVIDER),
42 protocol: Schema.Literal(OPENCODE2_HTTP_PROTOCOL),
43 version: Schema.String,
44 scenario: Schema.String,
45 metadata: Schema.optional(Schema.Record(Schema.String, Schema.Unknown)),
46 entries: Schema.Array(ProviderReplayEntry),
47});
48type OpenCode2ReplayTranscript = typeof OpenCode2ReplayTranscript.Type;
49const decodeOpenCode2ReplayTranscript = Schema.decodeUnknownEffect(OpenCode2ReplayTranscript);
50
51const operationOf = (
52 method: string,
53 path: string,
54 query: Record<string, unknown>,
55 body: unknown,
56) => {
57 if (method === "GET" && path === "/api/experimental/job" && query.kind === "image")
58 return { type: "imageJobs", input: query };
59 if (method === "GET" && path.startsWith("/api/fs/read/"))
60 return { type: "image.read", input: { path, ...query } };
61 const rpc = /^\/api\/rpc\/([^/]+)\/([^/]+)$/.exec(path);
62 if (method === "POST" && rpc !== null)
63 return {
64 type: "rpc.call",
65 input: { rpcID: rpc[1], method: rpc[2], ...query, ...(body as object) },
66 };
67 if (method === "GET" && path === "/api/experimental/job")
68 return { type: "job.list", input: query };
69 const shell = /^\/api\/shell(?:\/([^/]+)(\/output)?)?$/.exec(path);
70 if (shell !== null) {
71 const input = { ...(shell[1] === undefined ? {} : { id: shell[1] }), ...query };
72 return {
73 type: `shell.${method === "DELETE" ? "remove" : shell[2] ? "output" : shell[1] ? "get" : "list"}`,
74 input,
75 };
76 }
77 const session = /^\/api\/session\/([^/]+)(\/.*)?$/.exec(path);
78 if (method === "GET" && path === "/api/event") return { type: "event.subscribe" };
79 if (method === "GET" && path === "/api/model") return { type: "model.list", input: query };
80 if (method === "GET" && path === "/api/agent") return { type: "agent.list", input: query };
81 if (method === "POST" && path === "/api/session") return { type: "session.create", input: body };
82 if (method === "GET" && path === "/api/session/active") return { type: "session.active" };
83 if (method === "GET" && path === "/api/command") return { type: "command.list", input: query };
84 const mcp = /^\/api\/experimental\/mcp\/([^/]+)$/.exec(path);
85 if (mcp !== null && method === "PUT") {
86 return { type: "mcp.add", input: { server: mcp[1], ...query, ...(body as object) } };
87 }
88 if (mcp !== null && method === "DELETE") {
89 return { type: "mcp.remove", input: { server: mcp[1], ...query } };
90 }
91 const entry = /^\/api\/experimental\/session\/([^/]+)\/instructions\/entries\/([^/]+)$/.exec(
92 path,
93 );
94 if (entry !== null && method === "PUT") {
95 return {
96 type: "session.instructions.entry.put",
97 input: { sessionID: entry[1], key: entry[2], ...(body as object) },
98 };
99 }
100 if (method === "GET" && path === "/api/skill") return { type: "skill.list", input: query };
101 if (session !== null) {
102 const [, sessionID, rest = ""] = session;
103 // The client sends absent optional fields as `null`; recordings omit them.
104 const fields = Object.entries((body ?? {}) as Record<string, unknown>).filter(
105 ([, value]) => value !== null,
106 );
107 const input = { sessionID, ...query, ...Object.fromEntries(fields) };
108 if (method === "GET" && rest === "") return { type: "session.get", input };
109 if (method === "DELETE" && rest === "") return { type: "session.remove", input };
110 if (method === "PATCH" && rest === "") return { type: "session.update", input };
111 if (method === "POST" && rest === "/prompt") return { type: "session.prompt", input };
112 if (method === "POST" && rest === "/command") return { type: "session.command", input };
113 if (method === "POST" && rest === "/compact") return { type: "session.compact", input };
114 if (method === "POST" && rest === "/interrupt") return { type: "session.interrupt", input };
115 if (method === "POST" && rest === "/model") return { type: "session.switchModel", input };
116 if (method === "POST" && rest === "/agent") return { type: "session.switchAgent", input };
117 if (method === "POST" && rest === "/move") return { type: "session.move", input };
118 if (method === "POST" && rest === "/synthetic") return { type: "session.synthetic", input };
119 if (method === "POST" && rest === "/fork") return { type: "session.fork", input };
120 if (method === "POST" && rest === "/revert/stage") {
121 return { type: "session.revert.stage", input };
122 }
123 if (method === "POST" && rest === "/revert/commit") {
124 return { type: "session.revert.commit", input };
125 }
126 if (method === "DELETE" && rest === "/revert") return { type: "session.revert.clear", input };
127 const inbox = /^\/inbox\/([^/]+)$/.exec(rest);
128 if (method === "DELETE" && inbox !== null) {
129 const inboxID = decodeURIComponent(inbox[1] ?? "");
130 return { type: "session.inbox.cancel", input: { ...input, inboxID } };
131 }
132 if (method === "GET" && rest === "/message") return { type: "message.list", input };
133 if (method === "GET" && rest === "/permission") return { type: "permission.list", input };
134 if (method === "GET" && rest === "/form") return { type: "session.form.list", input };
135 const form = /^\/form\/([^/]+)(\/reply)?$/.exec(rest);
136 if (form !== null) {
137 const formInput = { ...input, formID: form[1] };
138 if (method === "POST" && form[2] !== undefined) {
139 return { type: "session.form.reply", input: formInput };
140 }
141 if (method === "DELETE") return { type: "session.form.cancel", input: formInput };
142 }
143 const permission = /^\/permission\/([^/]+)\/reply$/.exec(rest);
144 if (method === "POST" && permission !== null) {
145 return { type: "permission.reply", input: { ...input, requestID: permission[1] } };
146 }
147 }
148 return { type: `${method} ${path}`, input: { ...query, body } };
149};
150
151const replayHttpClient = (
152 controller: OpenCodeReplayController,
153 replayGate: ProviderReplayGate | undefined,
154 onPrompt: ((input: unknown) => void) | undefined,
155 beforeResponse: ((path: string) => Effect.Effect<void>) | undefined,
156) =>
157 HttpClient.make((request, url) =>
158 Effect.tryPromise({
159 try: async () => {
160 if (controller.exited && controller.finished) {
161 throw new Error("The replayed OpenCode server has exited.");
162 }
163 const raw =
164 request.body._tag === "Uint8Array"
165 ? new TextDecoder().decode(request.body.body)
166 : undefined;
167 const query = UrlParams.toRecord(UrlParams.fromInput(url.searchParams));
168 const operation = operationOf(
169 request.method,
170 url.pathname,
171 query,
172 raw === undefined ? undefined : decodeJson(raw),
173 );
174 // HTTP streams and requests use separate connections; deliver prior recorded inbound data before matching requests.
175 if (operation.type !== "event.subscribe") await controller.untilInboundDelivered();
176 if (operation.type === "session.prompt") onPrompt?.(operation.input);
177 await controller.expectOutbound(operation);
178 if (operation.type === "event.subscribe") {
179 controller.exited = false;
180 const encoder = new TextEncoder();
181 const events = controller.events(undefined, replayGate?.beforeEmit);
182 const frames = events[Symbol.asyncIterator]();
183 const body = new ReadableStream<Uint8Array>({
184 async pull(stream) {
185 const next = await frames.next();
186 if (next.done === true) stream.close();
187 else stream.enqueue(encoder.encode(`data: ${encodeJson(next.value)}\n\n`));
188 },
189 });
190 return new Response(body, { headers: { "content-type": "text/event-stream" } });
191 }
192 // Recorded responses are the raw HTTP bodies; `null` is an empty 204.
193 const body = await controller.response(operation.type);
194 if (typeof body === "object" && body !== null && "imageBase64" in body) {
195 return new Response(Buffer.from(String(body.imageBase64), "base64"), {
196 headers: { "content-type": "image/png" },
197 });
198 }
199 if (body === "<hang>") return new Promise<Response>(() => {});
200 if (typeof body === "object" && body !== null && "status" in body && "body" in body) {
201 return new Response(encodeJson(body.body), {
202 status: Number(body.status),
203 headers: { "content-type": "application/json" },
204 });
205 }
206 return body === null
207 ? new Response(null, { status: 204 })
208 : new Response(encodeJson(body), { headers: { "content-type": "application/json" } });
209 },
210 catch: (cause) =>
211 new HttpClientError.HttpClientError({
212 reason: new HttpClientError.TransportError({ request, cause }),
213 }),
214 }).pipe(
215 Effect.map((response) => HttpClientResponse.fromWeb(request, response)),
216 Effect.tap(() => beforeResponse?.(url.pathname) ?? Effect.void),
217 ),
218 );
219
220export const replayServer = (
221 transcript: ProviderReplayTranscript,
222 options?: {
223 readonly external?: boolean;
224 readonly processExit?: Effect.Effect<void>;
225 readonly beforeResponse?: (path: string) => Effect.Effect<void>;
226 readonly onPrompt?: (input: unknown) => void;
227 readonly replayGate?: ProviderReplayGate;
228 readonly borrowers?: { current: number };
229 },
230) =>
231 Effect.gen(function* () {
232 const controller = new OpenCodeReplayController(transcript);
233 yield* Effect.addFinalizer(() =>
234 Effect.sync(() => {
235 options?.replayGate?.releaseAll();
236 controller.assertComplete();
237 }),
238 );
239 const opencode = yield* OpenCode2Client.make.pipe(
240 Effect.provideService(
241 HttpClient.HttpClient,
242 replayHttpClient(
243 controller,
244 options?.replayGate,
245 options?.onPrompt,
246 options?.beforeResponse,
247 ),
248 ),
249 );
250 const connection = {
251 ...(yield* opencode.connect({ baseUrl: BASE_URL, password: "replay" })),
252 url: BASE_URL,
253 version: transcript.version,
254 external: options?.external ?? false,
255 ...(options?.external ? {} : { processExit: options?.processExit ?? Effect.never }),
256 };
257 const borrowers = options?.borrowers;
258 return OpenCode2Server.OpenCode2Server.of({
259 withConnection: (use) =>
260 borrowers === undefined
261 ? use(connection)
262 : Effect.acquireUseRelease(
263 Effect.sync(() => {
264 borrowers.current += 1;
265 }),
266 () => use(connection),
267 () =>
268 Effect.sync(() => {
269 borrowers.current -= 1;
270 }),
271 ),
272 forget: Effect.void,
273 });
274 });
275
276const makeReplayAdapter = (
277 transcript: ProviderReplayTranscript,
278 options?: {
279 readonly external?: boolean;
280 readonly processExit?: Effect.Effect<void>;
281 readonly beforeResponse?: (path: string) => Effect.Effect<void>;
282 readonly onPrompt?: (input: unknown) => void;
283 readonly replayGate?: ProviderReplayGate;
284 readonly borrowers?: { current: number };
285 },
286) =>
287 Effect.gen(function* () {
288 const server = yield* replayServer(transcript, options);
289 return yield* OpenCode2Adapter.make(ProviderInstanceId.make("opencode")).pipe(
290 Effect.provideService(OpenCode2Server.OpenCode2Server, server),
291 );
292 });
293
294const layerReplayHost = TestProviderHost.layer().pipe(
295 Layer.provide(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)),
296);
297
298const layerReplayHostWithAttachments = (
299 attachmentsDir: string,
300 resolveExtractedTextPath?: (path: string) => Effect.Effect<string | undefined>,
301) =>
302 Layer.effect(
303 ProviderHost.ProviderHost,
304 Effect.gen(function* () {
305 const host = yield* ProviderHost.ProviderHost;
306 return ProviderHost.ProviderHost.of({
307 ...host,
308 paths: { ...host.paths, attachmentsDir },
309 resolveAttachmentPath: (attachment) =>
310 resolveAttachmentPath({ attachmentsDir, attachment }),
311 ...(resolveExtractedTextPath === undefined ? {} : { resolveExtractedTextPath }),
312 });
313 }),
314 ).pipe(Layer.provide(layerReplayHost));
315
316function layerRegistry(
317 transcript: OpenCode2ReplayTranscript,
318 options?: { readonly replayGate?: ProviderReplayGate },
319) {
320 return Layer.unwrap(
321 makeReplayAdapter(transcript, { external: true, ...options }).pipe(
322 Effect.map((adapter) => ProviderAdapterRegistry.layerFromAdapters([adapter])),
323 ),
324 ).pipe(
325 Layer.provide(
326 Layer.mergeAll(layerReplayHost, IdAllocator.layer, NodeServices.layer, NodeCrypto.layer),
327 ),
328 );
329}
330
331export const openCode2ReplayRuntime = (
332 entries: ReadonlyArray<ProviderReplayEntry>,
333 options?: {
334 readonly external?: boolean;
335 readonly processExit?: Effect.Effect<void>;
336 readonly beforeResponse?: (path: string) => Effect.Effect<void>;
337 readonly version?: string;
338 readonly onPrompt?: (input: unknown) => void;
339 readonly borrowers?: { current: number };
340 readonly replayGate?: ProviderReplayGate;
341 readonly attachmentsDir?: string;
342 readonly resolveExtractedTextPath?: (path: string) => Effect.Effect<string | undefined>;
343 readonly fileSystem?: FileSystem.FileSystem;
344 },
345) =>
346 Effect.gen(function* () {
347 const fileSystem = options?.fileSystem ?? (yield* FileSystem.FileSystem);
348 const adapter = yield* makeReplayAdapter(
349 {
350 provider: OPENCODE_PROVIDER,
351 protocol: OPENCODE2_HTTP_PROTOCOL,
352 version: options?.version ?? "2.0.18",
353 scenario: "opencode2_adapter",
354 entries,
355 },
356 options,
357 ).pipe(Effect.provideService(FileSystem.FileSystem, fileSystem));
358 return yield* adapter.openSession({
359 threadId: ThreadId.make("thread:opencode2-adapter"),
360 providerSessionId: ProviderSessionId.make("provider-session:opencode2-adapter"),
361 modelSelection: {
362 instanceId: ProviderInstanceId.make("opencode"),
363 model: "opencode/big-pickle",
364 },
365 runtimePolicy: {
366 runtimeMode: "full-access",
367 interactionMode: "default",
368 cwd: "/work/opencode2",
369 },
370 });
371 }).pipe(
372 Effect.provide(
373 Layer.mergeAll(
374 options?.attachmentsDir === undefined
375 ? layerReplayHost
376 : layerReplayHostWithAttachments(
377 options.attachmentsDir,
378 options.resolveExtractedTextPath,
379 ),
380 IdAllocator.layer,
381 NodeServices.layer,
382 NodeCrypto.layer,
383 ),
384 ),
385 );
386
387export const OpenCode2OrchestratorReplayHarness: OrchestratorV2ProviderReplayHarness<
388 OpenCode2ReplayTranscript,
389 OpenCodeReplayTranscriptDecodeError
390> = {
391 driver: OPENCODE_PROVIDER,
392 decodeTranscript: (transcript: ProviderReplayTranscript) =>
393 decodeOpenCode2ReplayTranscript(transcript).pipe(
394 Effect.mapError(
395 (cause) =>
396 new OpenCodeReplayTranscriptDecodeError({
397 driver: transcript.provider,
398 protocol: transcript.protocol,
399 scenario: transcript.scenario,
400 cause,
401 }),
402 ),
403 ),
404 makeProviderAdapterRegistryLayer: layerRegistry,
405};