1/**
2 * Replays an OpenCode 2 transcript at the HTTP boundary: the real
3 * `@opencode/client` and the adapter's event reader run against an
4 * `HttpClient` that answers from the transcript, so request encoding and
5 * response decoding are both exercised.
6 */
7import * as NodeServices from "@effect/platform-node/NodeServices";
8import {
9 ProviderInstanceId,
10 ProviderReplayEntry,
11 ProviderSessionId,
12 ThreadId,
13 type ProviderReplayTranscript,
14} from "@t3tools/contracts";
15import * as Effect from "effect/Effect";
16import * as Layer from "effect/Layer";
17import * as Schema from "effect/Schema";
18import * as HttpClient from "effect/http/HttpClient";
19import * as HttpClientError from "effect/http/HttpClientError";
20import * as HttpClientResponse from "effect/http/HttpClientResponse";
21import * as UrlParams from "effect/http/UrlParams";
22
23import * as OpenCode2Client from "@t3tools/provider-opencode/server/v2/OpenCode2Client";
24import * as OpenCode2Server from "@t3tools/provider-opencode/server/v2/OpenCode2Server";
25import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
26import * as ProviderAdapterRegistry from "../ProviderAdapterRegistry.ts";
27import type { ProviderReplayGate } from "@t3tools/provider-testing/replayGate";
28import type { OrchestratorV2ProviderReplayHarness } from "../testkit/ProviderReplayHarness.ts";
29import * as TestProviderHost from "@t3tools/provider-testing/TestProviderHost";
30import { OPENCODE_PROVIDER } from "@t3tools/provider-opencode/testing";
31import {
32 OpenCodeReplayController,
33 OpenCodeReplayTranscriptDecodeError,
34} from "./OpenCodeAdapterV2.testkit.ts";
35import * as OpenCode2Adapter from "@t3tools/provider-opencode/server/v2/adapter";
36
37export const OPENCODE2_HTTP_PROTOCOL = "opencode2-http.sse" as const;
38const BASE_URL = "http://opencode2.replay";
39const decodeJson = Schema.decodeUnknownSync(Schema.fromJsonString(Schema.Unknown));
40const encodeJson = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown));
41
42const OpenCode2ReplayTranscript = Schema.Struct({
43 provider: Schema.Literal(OPENCODE_PROVIDER),
44 protocol: Schema.Literal(OPENCODE2_HTTP_PROTOCOL),
45 version: Schema.String,
46 scenario: Schema.String,
47 metadata: Schema.optional(Schema.Record(Schema.String, Schema.Unknown)),
48 entries: Schema.Array(ProviderReplayEntry),
49});
50type OpenCode2ReplayTranscript = typeof OpenCode2ReplayTranscript.Type;
51const decodeOpenCode2ReplayTranscript = Schema.decodeUnknownEffect(OpenCode2ReplayTranscript);
52
53/** Names a request the way the recordings do: the client operation and its JSON input. */
54const operationOf = (
55 method: string,
56 path: string,
57 query: Record<string, unknown>,
58 body: unknown,
59) => {
60 const session = /^\/api\/session\/([^/]+)(\/.*)?$/.exec(path);
61 if (method === "GET" && path === "/api/event") return { type: "event.subscribe" };
62 if (method === "GET" && path === "/api/model") return { type: "model.list", input: query };
63 if (method === "GET" && path === "/api/agent") return { type: "agent.list", input: query };
64 if (method === "POST" && path === "/api/session") return { type: "session.create", input: body };
65 if (method === "GET" && path === "/api/session/active") return { type: "session.active" };
66 if (method === "GET" && path === "/api/command") return { type: "command.list", input: query };
67 const mcp = /^\/api\/experimental\/mcp\/([^/]+)$/.exec(path);
68 if (mcp !== null && method === "PUT") {
69 return { type: "mcp.add", input: { server: mcp[1], ...query, ...(body as object) } };
70 }
71 if (mcp !== null && method === "DELETE") {
72 return { type: "mcp.remove", input: { server: mcp[1], ...query } };
73 }
74 const entry = /^\/api\/experimental\/session\/([^/]+)\/instructions\/entries\/([^/]+)$/.exec(
75 path,
76 );
77 if (entry !== null && method === "PUT") {
78 return {
79 type: "session.instructions.entry.put",
80 input: { sessionID: entry[1], key: entry[2], ...(body as object) },
81 };
82 }
83 if (method === "GET" && path === "/api/skill") return { type: "skill.list", input: query };
84 if (session !== null) {
85 const [, sessionID, rest = ""] = session;
86 // The client sends absent optional fields as `null`; recordings omit them.
87 const fields = Object.entries((body ?? {}) as Record<string, unknown>).filter(
88 ([, value]) => value !== null,
89 );
90 const input = { sessionID, ...query, ...Object.fromEntries(fields) };
91 if (method === "GET" && rest === "") return { type: "session.get", input };
92 if (method === "DELETE" && rest === "") return { type: "session.remove", input };
93 if (method === "PATCH" && rest === "") return { type: "session.update", input };
94 if (method === "POST" && rest === "/prompt") return { type: "session.prompt", input };
95 if (method === "POST" && rest === "/command") return { type: "session.command", input };
96 if (method === "POST" && rest === "/compact") return { type: "session.compact", input };
97 if (method === "POST" && rest === "/interrupt") return { type: "session.interrupt", input };
98 if (method === "POST" && rest === "/model") return { type: "session.switchModel", input };
99 if (method === "POST" && rest === "/agent") return { type: "session.switchAgent", input };
100 if (method === "POST" && rest === "/move") return { type: "session.move", input };
101 if (method === "POST" && rest === "/synthetic") return { type: "session.synthetic", input };
102 if (method === "POST" && rest === "/fork") return { type: "session.fork", input };
103 if (method === "POST" && rest === "/revert/stage") {
104 return { type: "session.revert.stage", input };
105 }
106 if (method === "POST" && rest === "/revert/commit") {
107 return { type: "session.revert.commit", input };
108 }
109 if (method === "DELETE" && rest === "/revert") return { type: "session.revert.clear", input };
110 const inbox = /^\/inbox\/([^/]+)$/.exec(rest);
111 if (method === "DELETE" && inbox !== null) {
112 // T3's steer ids carry `:`, which the path encodes.
113 const inboxID = decodeURIComponent(inbox[1] ?? "");
114 return { type: "session.inbox.cancel", input: { ...input, inboxID } };
115 }
116 if (method === "GET" && rest === "/message") return { type: "message.list", input };
117 if (method === "GET" && rest === "/permission") return { type: "permission.list", input };
118 if (method === "GET" && rest === "/form") return { type: "session.form.list", input };
119 const form = /^\/form\/([^/]+)(\/reply)?$/.exec(rest);
120 if (form !== null) {
121 const formInput = { ...input, formID: form[1] };
122 if (method === "POST" && form[2] !== undefined) {
123 return { type: "session.form.reply", input: formInput };
124 }
125 if (method === "DELETE") return { type: "session.form.cancel", input: formInput };
126 }
127 const permission = /^\/permission\/([^/]+)\/reply$/.exec(rest);
128 if (method === "POST" && permission !== null) {
129 return { type: "permission.reply", input: { ...input, requestID: permission[1] } };
130 }
131 }
132 return { type: `${method} ${path}`, input: { ...query, body } };
133};
134
135/**
136 * An `HttpClient` that answers every request from the transcript. A successful
137 * `runtime_exit` ends the event stream: the server has stopped. When entries
138 * follow it, the server was restarted and the next `event.subscribe` opens its
139 * new stream; a transcript that ends there stays down, so every later request
140 * fails to connect as it would against a stopped server.
141 */
142const replayHttpClient = (
143 controller: OpenCodeReplayController,
144 replayGate: ProviderReplayGate | undefined,
145) =>
146 HttpClient.make((request, url) =>
147 Effect.tryPromise({
148 try: async () => {
149 if (controller.exited && controller.finished) {
150 throw new Error("The replayed OpenCode server has exited.");
151 }
152 const raw =
153 request.body._tag === "Uint8Array"
154 ? new TextDecoder().decode(request.body.body)
155 : undefined;
156 const query = UrlParams.toRecord(UrlParams.fromInput(url.searchParams));
157 const operation = operationOf(
158 request.method,
159 url.pathname,
160 query,
161 raw === undefined ? undefined : decodeJson(raw),
162 );
163 // The event stream and requests are separate connections: a request
164 // is matched only once what was recorded before it was delivered.
165 if (operation.type !== "event.subscribe") await controller.untilInboundDelivered();
166 await controller.expectOutbound(operation);
167 if (operation.type === "event.subscribe") {
168 controller.exited = false;
169 const encoder = new TextEncoder();
170 const events = controller.events(undefined, replayGate?.beforeEmit);
171 const frames = events[Symbol.asyncIterator]();
172 const body = new ReadableStream<Uint8Array>({
173 async pull(stream) {
174 const next = await frames.next();
175 if (next.done === true) stream.close();
176 else stream.enqueue(encoder.encode(`data: ${encodeJson(next.value)}\n\n`));
177 },
178 });
179 return new Response(body, { headers: { "content-type": "text/event-stream" } });
180 }
181 // Recorded responses are the raw HTTP bodies; `null` is an empty 204.
182 // `{ status, body }` replays an error status and `"<hang>"` never answers.
183 const body = await controller.response(operation.type);
184 if (body === "<hang>") return new Promise<Response>(() => {});
185 if (typeof body === "object" && body !== null && "status" in body && "body" in body) {
186 return new Response(encodeJson(body.body), {
187 status: Number(body.status),
188 headers: { "content-type": "application/json" },
189 });
190 }
191 return body === null
192 ? new Response(null, { status: 204 })
193 : new Response(encodeJson(body), { headers: { "content-type": "application/json" } });
194 },
195 catch: (cause) =>
196 new HttpClientError.HttpClientError({
197 reason: new HttpClientError.TransportError({ request, cause }),
198 }),
199 }).pipe(Effect.map((response) => HttpClientResponse.fromWeb(request, response))),
200 );
201
202/**
203 * An OpenCode 2 server that answers from `transcript`, checking at scope close
204 * that the transcript ran out.
205 */
206export const replayServer = (
207 transcript: ProviderReplayTranscript,
208 options?: {
209 readonly external?: boolean;
210 readonly replayGate?: ProviderReplayGate;
211 /** Counts the connections currently lent out, as the server owner's borrowers. */
212 readonly borrowers?: { current: number };
213 },
214) =>
215 Effect.gen(function* () {
216 const controller = new OpenCodeReplayController(transcript);
217 yield* Effect.addFinalizer(() =>
218 Effect.sync(() => {
219 options?.replayGate?.releaseAll();
220 controller.assertComplete();
221 }),
222 );
223 const opencode = yield* OpenCode2Client.make.pipe(
224 Effect.provideService(
225 HttpClient.HttpClient,
226 replayHttpClient(controller, options?.replayGate),
227 ),
228 );
229 const connection = {
230 ...(yield* opencode.connect({ baseUrl: BASE_URL, password: "replay" })),
231 url: BASE_URL,
232 version: transcript.version,
233 external: options?.external ?? false,
234 };
235 const borrowers = options?.borrowers;
236 return OpenCode2Server.OpenCode2Server.of({
237 withConnection: (use) =>
238 borrowers === undefined
239 ? use(connection)
240 : Effect.acquireUseRelease(
241 Effect.sync(() => {
242 borrowers.current += 1;
243 }),
244 () => use(connection),
245 () =>
246 Effect.sync(() => {
247 borrowers.current -= 1;
248 }),
249 ),
250 });
251 });
252
253/** The 2.x adapter over a replayed server. */
254const makeReplayAdapter = (
255 transcript: ProviderReplayTranscript,
256 options?: {
257 readonly external?: boolean;
258 readonly replayGate?: ProviderReplayGate;
259 /** Counts the connections currently lent out, as the server owner's borrowers. */
260 readonly borrowers?: { current: number };
261 },
262) =>
263 Effect.gen(function* () {
264 const server = yield* replayServer(transcript, options);
265 return yield* OpenCode2Adapter.make(ProviderInstanceId.make("opencode")).pipe(
266 Effect.provideService(OpenCode2Server.OpenCode2Server, server),
267 );
268 });
269
270const layerReplayHost = TestProviderHost.layer().pipe(Layer.provide(NodeServices.layer));
271
272function layerRegistry(
273 transcript: OpenCode2ReplayTranscript,
274 options?: { readonly replayGate?: ProviderReplayGate },
275) {
276 return Layer.unwrap(
277 makeReplayAdapter(transcript, { external: true, ...options }).pipe(
278 Effect.map((adapter) => ProviderAdapterRegistry.layerFromAdapters([adapter])),
279 ),
280 ).pipe(Layer.provide(Layer.mergeAll(layerReplayHost, IdAllocator.layer, NodeServices.layer)));
281}
282
283/**
284 * An open 2.x session runtime whose server answers from `entries`, for
285 * adapter-level tests of failures a real server cannot produce on demand.
286 */
287export const openCode2ReplayRuntime = (
288 entries: ReadonlyArray<ProviderReplayEntry>,
289 options?: {
290 readonly external?: boolean;
291 readonly borrowers?: { current: number };
292 readonly replayGate?: ProviderReplayGate;
293 },
294) =>
295 Effect.gen(function* () {
296 const adapter = yield* makeReplayAdapter(
297 {
298 provider: OPENCODE_PROVIDER,
299 protocol: OPENCODE2_HTTP_PROTOCOL,
300 version: "2.0.18",
301 scenario: "opencode2_adapter",
302 entries,
303 },
304 options,
305 );
306 return yield* adapter.openSession({
307 threadId: ThreadId.make("thread:opencode2-adapter"),
308 providerSessionId: ProviderSessionId.make("provider-session:opencode2-adapter"),
309 modelSelection: {
310 instanceId: ProviderInstanceId.make("opencode"),
311 model: "opencode/big-pickle",
312 },
313 // Opened where the adapter tests' threads run, as T3 opens a session for its first thread.
314 runtimePolicy: {
315 runtimeMode: "full-access",
316 interactionMode: "default",
317 cwd: "/work/opencode2",
318 },
319 });
320 }).pipe(Effect.provide(Layer.mergeAll(layerReplayHost, IdAllocator.layer, NodeServices.layer)));
321
322export const OpenCode2OrchestratorReplayHarness: OrchestratorV2ProviderReplayHarness<
323 OpenCode2ReplayTranscript,
324 OpenCodeReplayTranscriptDecodeError
325> = {
326 driver: OPENCODE_PROVIDER,
327 decodeTranscript: (transcript: ProviderReplayTranscript) =>
328 decodeOpenCode2ReplayTranscript(transcript).pipe(
329 Effect.mapError(
330 (cause) =>
331 new OpenCodeReplayTranscriptDecodeError({
332 driver: transcript.provider,
333 protocol: transcript.protocol,
334 scenario: transcript.scenario,
335 cause,
336 }),
337 ),
338 ),
339 makeProviderAdapterRegistryLayer: layerRegistry,
340};