1import * as Path from "effect/Path";
2import { PROVIDER_SEND_TURN_MAX_IMAGE_BYTES } from "@t3tools/contracts";
3import { OpenCode, type OpenCodeClient } from "@opencode/client/effect";
4import { OpenCodeEvent } from "@opencode/protocol/groups/event";
5import * as Context from "effect/Context";
6import * as Effect from "effect/Effect";
7import * as Layer from "effect/Layer";
8import * as Option from "effect/Option";
9import * as Redacted from "effect/Redacted";
10import * as Result from "effect/Result";
11import * as Schema from "effect/Schema";
12import type * as Scope from "effect/Scope";
13import * as Stream from "effect/Stream";
14import * as Sse from "effect/encoding/Sse";
15import * as HttpClient from "effect/http/HttpClient";
16import * as HttpClientError from "effect/http/HttpClientError";
17import * as HttpClientRequest from "effect/http/HttpClientRequest";
18import * as HttpClientResponse from "effect/http/HttpClientResponse";
19
20import { readServerUsageLimits } from "./usageLimits.ts";
21
22/** OpenCode 2 accepts only HTTP Basic auth, always with this user name. */
23const OPENCODE_USERNAME = "opencode";
24
25export interface OpenCode2UnreadableTerminal {
26 readonly type: "unreadable.execution.ended";
27 readonly executionType: (typeof EXECUTION_ENDS)[number];
28 readonly sessionID: string;
29}
30
31export interface OpenCode2UnreadableStart {
32 readonly type: "unreadable.execution.started";
33 readonly sessionID: string;
34}
35
36const EXECUTION_ENDS = [
37 "session.execution.succeeded",
38 "session.execution.failed",
39 "session.execution.interrupted",
40] as const;
41const isExecutionEnd = (type: string): type is (typeof EXECUTION_ENDS)[number] =>
42 (EXECUTION_ENDS as ReadonlyArray<string>).includes(type);
43
44export class OpenCode2SilentStreamError extends Schema.TaggedError<OpenCode2SilentStreamError>()(
45 "OpenCode2SilentStreamError",
46 {},
47) {
48 override get message(): string {
49 return "The OpenCode server stopped sending events.";
50 }
51}
52
53export type OpenCode2StreamEvent =
54 | OpenCodeEvent
55 | OpenCode2UnreadableTerminal
56 | OpenCode2UnreadableStart;
57
58const ShellJobs = Schema.Struct({
59 data: Schema.Array(
60 Schema.Struct({
61 kind: Schema.Literal("shell"),
62 shellID: Schema.String,
63 sessionID: Schema.String,
64 command: Schema.String,
65 status: Schema.Literals(["running", "completed", "error", "cancelled"]),
66 }),
67 ),
68});
69export class OpenCode2ImageError extends Schema.TaggedError<OpenCode2ImageError>()(
70 "OpenCode2ImageError",
71 { cause: Schema.Defect() },
72) {}
73
74const IMAGE_MIME: Readonly<Record<string, string>> = {
75 ".png": "image/png",
76 ".jpg": "image/jpeg",
77 ".jpeg": "image/jpeg",
78 ".webp": "image/webp",
79 ".gif": "image/gif",
80};
81export const imagePreviewMime = (extension: string) => IMAGE_MIME[extension.toLowerCase()];
82
83const NativeImageJob = Schema.Struct({
84 kind: Schema.Literal("image"),
85 id: Schema.String,
86 sessionID: Schema.String,
87 status: Schema.Literals(["running", "completed", "error", "cancelled"]),
88 path: Schema.String,
89 model: Schema.String,
90});
91const decodeJobs = Schema.decodeUnknownEffect(
92 Schema.Struct({ data: Schema.Array(Schema.Unknown) }),
93);
94const decodeImageJob = Schema.decodeUnknownOption(NativeImageJob);
95
96/** events resolves on subscription acceptance; subscribe before work because the stream is volatile. */
97export interface OpenCode2Api {
98 readonly client: OpenCodeClient;
99 readonly usageLimits: (
100 directory: string,
101 ) => Effect.Effect<import("@t3tools/contracts").ServerProviderUsageLimits | undefined>;
102 readonly shellJobs?: (
103 sessionID: string,
104 ) => Effect.Effect<typeof ShellJobs.Type, HttpClientError.HttpClientError | Schema.SchemaError>;
105 readonly imageJobs: (
106 sessionID: string,
107 ) => Effect.Effect<ReadonlyArray<typeof NativeImageJob.Type>, OpenCode2ImageError>;
108 readonly readImage: (
109 path: string,
110 directory: string,
111 ) => Effect.Effect<{ uri: string; mime: string }, OpenCode2ImageError>;
112 readonly events: Effect.Effect<
113 Stream.Stream<
114 OpenCode2StreamEvent,
115 HttpClientError.HttpClientError | Sse.Retry | Sse.SseError | OpenCode2SilentStreamError
116 >,
117 HttpClientError.HttpClientError,
118 Scope.Scope
119 >;
120}
121
122export class OpenCode2Client extends Context.Service<
123 OpenCode2Client,
124 {
125 readonly connect: (input: {
126 readonly baseUrl: string;
127 readonly password: string | Redacted.Redacted;
128 }) => Effect.Effect<OpenCode2Api>;
129 }
130>()("@t3tools/provider-opencode/server/v2/OpenCode2Client") {}
131
132// Heartbeats arrive every 10-15 seconds; longer silence means stuck, not idle.
133const SILENT_STREAM_TIMEOUT = "45 seconds";
134const SUBSCRIPTION_TIMEOUT = "10 seconds";
135
136const decodeEvent = Schema.decodeUnknownResult(Schema.fromJsonString(OpenCodeEvent));
137const decodeEnvelope = Schema.decodeUnknownOption(
138 Schema.fromJsonString(
139 Schema.Struct({
140 type: Schema.String,
141 data: Schema.optional(Schema.Struct({ sessionID: Schema.optional(Schema.String) })),
142 }),
143 ),
144);
145
146// Rejected stream frames can be frequent when the server introduces new event types.
147const undecodable = Effect.fnUntraced(function* (data: string) {
148 const envelope = decodeEnvelope(data);
149 const type = envelope._tag === "Some" ? envelope.value.type.slice(0, 80) : "<unreadable>";
150 const sessionID = envelope._tag === "Some" ? envelope.value.data?.sessionID : undefined;
151 if (isExecutionEnd(type) && sessionID !== undefined) {
152 yield* Effect.logWarning(
153 "Ended an OpenCode execution from an event this build cannot decode.",
154 {
155 type,
156 },
157 );
158 return Result.succeed<OpenCode2StreamEvent>({
159 type: "unreadable.execution.ended",
160 executionType: type,
161 sessionID,
162 });
163 }
164 if (type === "session.execution.started" && sessionID !== undefined) {
165 yield* Effect.logDebug("Read an OpenCode execution start this build cannot decode.");
166 return Result.succeed<OpenCode2StreamEvent>({
167 type: "unreadable.execution.started",
168 sessionID,
169 });
170 }
171 yield* Effect.logDebug("Skipped an OpenCode event this build cannot decode.", { type });
172 return Result.failVoid;
173});
174
175const readEvents = HttpClient.HttpClient.pipe(
176 // Recovery subscribes before reading history, so its stream may never be consumed.
177 Effect.flatMap((client) =>
178 HttpClient.withScope(client).get("/api/event", {
179 headers: { accept: "text/event-stream" },
180 }),
181 ),
182 Effect.flatMap(HttpClientResponse.filterStatusOk),
183 Effect.map((response) =>
184 response.stream.pipe(
185 Stream.timeoutOrElse({
186 duration: SILENT_STREAM_TIMEOUT,
187 orElse: () => Stream.fail(new OpenCode2SilentStreamError()),
188 }),
189 Stream.decodeText,
190 Stream.pipeThroughChannel(Sse.decode()),
191 Stream.filterMapEffect((frame) => {
192 const decoded = decodeEvent(frame.data);
193 return Result.isSuccess(decoded)
194 ? Effect.succeed(Result.succeed<OpenCode2StreamEvent>(decoded.success))
195 : undecodable(frame.data);
196 }),
197 ),
198 ),
199);
200
201// OpenCode requires UTF-8 Basic credentials; HttpClientRequest.basicAuth uses Latin-1.
202const basicAuthorization = (password: string | Redacted.Redacted) => {
203 const plain = Redacted.isRedacted(password) ? Redacted.value(password) : password;
204 return `Basic ${Buffer.from(`${OPENCODE_USERNAME}:${plain}`, "utf8").toString("base64")}`;
205};
206
207const MESSAGE_LIST_URL = /\/api\/session\/[^/?#]+\/message(?:[?#]|$)/;
208
209/** Reads only the page's shape; each message stays opaque for the real schema to judge. */
210const MessagePage = Schema.fromJsonString(
211 Schema.StructWithRest(Schema.Struct({ data: Schema.Array(Schema.Unknown) }), [
212 Schema.Record(Schema.String, Schema.Unknown),
213 ]),
214);
215const decodePage = Schema.decodeUnknownOption(MessagePage);
216const encodePage = Schema.encodeSync(MessagePage);
217
218const CompactionWithRecent = Schema.Struct({
219 type: Schema.Literal("compaction"),
220 status: Schema.Literals(["running", "completed"]),
221 metadata: Schema.optionalKey(Schema.Unknown),
222});
223const isCompactionWithRecent = Schema.is(CompactionWithRecent);
224const isLegacyMetadata = Schema.is(Schema.Struct({ recent: Schema.String }));
225
226const fillCompactionRecent = (message: unknown) =>
227 isCompactionWithRecent(message) && !("recent" in message)
228 ? { ...message, recent: isLegacyMetadata(message.metadata) ? message.metadata.recent : "" }
229 : message;
230
231/**
232 * OpenCode 2.1 dropped a compaction's `recent`, which the bundled 2.0.28 schema requires.
233 * Migrated pre-2.1 compactions keep it under `metadata.recent`; native 2.1 ones have none.
234 * T3 never reads it, so fill the gap; every other field still validates downstream.
235 */
236const fillCompactionRecents = HttpClient.transformResponse(
237 Effect.fnUntraced(function* (
238 pending: Effect.Effect<HttpClientResponse.HttpClientResponse, HttpClientError.HttpClientError>,
239 ) {
240 const response = yield* pending;
241 const { request } = response;
242 if (
243 request.method !== "GET" ||
244 response.status !== 200 ||
245 !MESSAGE_LIST_URL.test(request.url)
246 ) {
247 return response;
248 }
249 const text = yield* response.text;
250 const page = decodePage(text);
251 // The response caches its body, so the untouched one still decodes downstream.
252 if (Option.isNone(page)) return response;
253 const filled = encodePage({ ...page.value, data: page.value.data.map(fillCompactionRecent) });
254 // The filled body is longer than the one the server measured.
255 const { "content-length": _, ...headers } = response.headers;
256 return HttpClientResponse.fromWeb(
257 request,
258 new Response(filled, { status: response.status, headers }),
259 );
260 }),
261);
262
263/** @public */
264export const make = Effect.gen(function* () {
265 const httpClient = yield* HttpClient.HttpClient;
266 const paths = yield* Path.Path.pipe(Effect.provide(Path.layer));
267 return OpenCode2Client.of({
268 connect: Effect.fn("OpenCode2Client.connect")(function ({ baseUrl, password }) {
269 const authenticated = HttpClient.mapRequest(
270 httpClient,
271 HttpClientRequest.setHeader("Authorization", basicAuthorization(password)),
272 );
273 return OpenCode.make({ baseUrl }).pipe(
274 Effect.provideService(HttpClient.HttpClient, fillCompactionRecents(authenticated)),
275 Effect.map((client) => ({
276 client,
277 usageLimits: (directory) =>
278 readServerUsageLimits(directory).pipe(
279 Effect.provideService(
280 HttpClient.HttpClient,
281 HttpClient.mapRequest(authenticated, HttpClientRequest.prependUrl(baseUrl)),
282 ),
283 ),
284 shellJobs: Effect.fn("OpenCode2Client.shellJobs")(function (sessionID: string) {
285 return authenticated
286 .get(`${baseUrl}/api/experimental/job`, {
287 urlParams: { sessionID, kind: "shell", descendants: "true" },
288 })
289 .pipe(
290 Effect.flatMap(HttpClientResponse.filterStatusOk),
291 Effect.flatMap(HttpClientResponse.schemaBodyJson(ShellJobs)),
292 );
293 }),
294 imageJobs: (sessionID: string) =>
295 authenticated
296 .get(`${baseUrl}/api/experimental/job`, {
297 urlParams: { sessionID, kind: "image" },
298 })
299 .pipe(
300 Effect.flatMap(HttpClientResponse.filterStatusOk),
301 Effect.flatMap((response) => response.json),
302 Effect.flatMap(decodeJobs),
303 Effect.map((page) =>
304 page.data.flatMap((item) => {
305 const job = decodeImageJob(item);
306 return Option.isSome(job) && job.value.sessionID === sessionID
307 ? [job.value]
308 : [];
309 }),
310 ),
311 Effect.mapError((cause) => new OpenCode2ImageError({ cause })),
312 ),
313 readImage: Effect.fn("OpenCode2Client.readImage")(
314 function* (path: string, directory: string) {
315 const mime = imagePreviewMime(paths.extname(path));
316 if (mime === undefined)
317 return yield* new OpenCode2ImageError({ cause: "Unsupported image format" });
318 const relative = paths.relative(directory, path);
319 // OpenCode confines reads to the requested location, including its own temp images.
320 const outside =
321 relative === ".." ||
322 relative.startsWith(`..${paths.sep}`) ||
323 paths.isAbsolute(relative);
324 const location = outside ? paths.dirname(path) : directory;
325 const file = outside ? paths.basename(path) : relative;
326 const response = yield* authenticated
327 .get(`${baseUrl}/api/fs/read/${encodeURIComponent(file)}`, {
328 urlParams: { "location[directory]": location },
329 })
330 .pipe(Effect.flatMap(HttpClientResponse.filterStatusOk));
331 const bytes = yield* response.arrayBuffer;
332 if (bytes.byteLength > PROVIDER_SEND_TURN_MAX_IMAGE_BYTES)
333 return yield* new OpenCode2ImageError({ cause: "Image exceeds preview limit" });
334 return { uri: `data:${mime};base64,${Buffer.from(bytes).toString("base64")}`, mime };
335 },
336 Effect.timeout("10 seconds"),
337 Effect.mapError((cause) => new OpenCode2ImageError({ cause })),
338 ),
339 events: readEvents.pipe(
340 Effect.provideService(
341 HttpClient.HttpClient,
342 HttpClient.mapRequest(authenticated, HttpClientRequest.prependUrl(baseUrl)).pipe(
343 HttpClient.timeout(SUBSCRIPTION_TIMEOUT),
344 ),
345 ),
346 ),
347 })),
348 );
349 }),
350 });
351});
352
353export const layer = Layer.effect(OpenCode2Client, make);
354
355/** Send input only initially; cursor pages must omit order or OpenCode rejects them. */
356export const paginate = <Input extends { readonly cursor?: unknown }, Item, E, R>(
357 input: Input,
358 list: (
359 input: Input,
360 ) => Effect.Effect<
361 { readonly data: ReadonlyArray<Item>; readonly cursor: { readonly next?: Input["cursor"] } },
362 E,
363 R
364 >,
365): Stream.Stream<Item, E, R> =>
366 Stream.paginate(input, (request) =>
367 list(request).pipe(
368 Effect.map(
369 (page) =>
370 [
371 page.data,
372 page.data.length === 0 || page.cursor.next === undefined
373 ? Option.none()
374 : Option.some({ ...request, order: undefined, cursor: page.cursor.next }),
375 ] as const,
376 ),
377 ),
378 );