1import { OpenCode, type OpenCodeClient } from "@opencode/client/effect";
2import { OpenCodeEvent } from "@opencode/protocol/groups/event";
3import * as Context from "effect/Context";
4import * as Effect from "effect/Effect";
5import * as Layer from "effect/Layer";
6import * as Option from "effect/Option";
7import * as Redacted from "effect/Redacted";
8import * as Result from "effect/Result";
9import * as Schema from "effect/Schema";
10import * as Stream from "effect/Stream";
11import * as Sse from "effect/encoding/Sse";
12import * as HttpClient from "effect/http/HttpClient";
13import * as HttpClientError from "effect/http/HttpClientError";
14import * as HttpClientRequest from "effect/http/HttpClientRequest";
15import * as HttpClientResponse from "effect/http/HttpClientResponse";
16
17/** OpenCode 2 accepts only HTTP Basic auth, always with this user name. */
18const OPENCODE_USERNAME = "opencode";
19
20/**
21 * A session's execution ended, but this build could not decode the event (a
22 * newer server added a field or a value). Only the type and session survive,
23 * so the turn still ends.
24 */
25export interface OpenCode2UnreadableTerminal {
26 readonly type: "unreadable.execution.ended";
27 readonly executionType: (typeof EXECUTION_ENDS)[number];
28 readonly sessionID: string;
29}
30
31/**
32 * A session's execution started, but this build could not decode the event.
33 * It marks where a new execution begins and never ends a turn.
34 */
35export interface OpenCode2UnreadableStart {
36 readonly type: "unreadable.execution.started";
37 readonly sessionID: string;
38}
39
40/** The events that end an execution; any other undecodable event is skipped. */
41const EXECUTION_ENDS = [
42 "session.execution.succeeded",
43 "session.execution.failed",
44 "session.execution.interrupted",
45] as const;
46const isExecutionEnd = (type: string): type is (typeof EXECUTION_ENDS)[number] =>
47 (EXECUTION_ENDS as ReadonlyArray<string>).includes(type);
48
49/** The server sent nothing, not even a heartbeat, for `SILENT_STREAM_TIMEOUT`. */
50export class OpenCode2SilentStreamError extends Schema.TaggedError<OpenCode2SilentStreamError>()(
51 "OpenCode2SilentStreamError",
52 {},
53) {
54 override get message(): string {
55 return "The OpenCode server stopped sending events.";
56 }
57}
58
59export type OpenCode2StreamEvent =
60 | OpenCodeEvent
61 | OpenCode2UnreadableTerminal
62 | OpenCode2UnreadableStart;
63
64/**
65 * An OpenCode 2 client plus a forward-compatible `/api/event` stream. The
66 * client's own `event.subscribe` fails the whole stream on the first event a
67 * newer server adds, so `events` decodes each frame on its own and skips the
68 * ones this build does not know, except an execution's end or start, which it
69 * keeps as an {@link OpenCode2UnreadableTerminal} or {@link OpenCode2UnreadableStart}.
70 * `events` succeeds once the server accepted the subscription: the stream is
71 * volatile, so callers subscribe before they start work whose events they need.
72 */
73export interface OpenCode2Api {
74 readonly client: OpenCodeClient;
75 readonly events: Effect.Effect<
76 Stream.Stream<
77 OpenCode2StreamEvent,
78 HttpClientError.HttpClientError | Sse.Retry | Sse.SseError | OpenCode2SilentStreamError
79 >,
80 HttpClientError.HttpClientError
81 >;
82}
83
84/** Builds clients for OpenCode 2 servers, one per base URL and password. */
85export class OpenCode2Client extends Context.Service<
86 OpenCode2Client,
87 {
88 readonly connect: (input: {
89 readonly baseUrl: string;
90 readonly password: string | Redacted.Redacted;
91 }) => Effect.Effect<OpenCode2Api>;
92 }
93>()("@t3tools/provider-opencode/server/v2/OpenCode2Client") {}
94
95/**
96 * OpenCode writes a `: heartbeat` comment every 10 to 15 seconds, so this long
97 * without a single byte means the server is stuck, not idle.
98 */
99const SILENT_STREAM_TIMEOUT = "45 seconds";
100
101const decodeEvent = Schema.decodeUnknownResult(Schema.fromJsonString(OpenCodeEvent));
102// Just enough of an event to route it, for frames the full schema rejects.
103const decodeEnvelope = Schema.decodeUnknownOption(
104 Schema.fromJsonString(
105 Schema.Struct({
106 type: Schema.String,
107 data: Schema.optional(Schema.Struct({ sessionID: Schema.optional(Schema.String) })),
108 }),
109 ),
110);
111
112const undecodable = (data: string) =>
113 Effect.gen(function* () {
114 const envelope = decodeEnvelope(data);
115 const type = envelope._tag === "Some" ? envelope.value.type.slice(0, 80) : "<unreadable>";
116 const sessionID = envelope._tag === "Some" ? envelope.value.data?.sessionID : undefined;
117 if (isExecutionEnd(type) && sessionID !== undefined) {
118 yield* Effect.logWarning(
119 "Ended an OpenCode execution from an event this build cannot decode.",
120 {
121 type,
122 },
123 );
124 return Result.succeed<OpenCode2StreamEvent>({
125 type: "unreadable.execution.ended",
126 executionType: type,
127 sessionID,
128 });
129 }
130 if (type === "session.execution.started" && sessionID !== undefined) {
131 yield* Effect.logDebug("Read an OpenCode execution start this build cannot decode.");
132 return Result.succeed<OpenCode2StreamEvent>({
133 type: "unreadable.execution.started",
134 sessionID,
135 });
136 }
137 yield* Effect.logDebug("Skipped an OpenCode event this build cannot decode.", { type });
138 return Result.failVoid;
139 });
140
141/** Subscribes to `/api/event`, then streams every frame this build can route. */
142const readEvents = HttpClient.get("/api/event", {
143 headers: { accept: "text/event-stream" },
144}).pipe(
145 Effect.flatMap(HttpClientResponse.filterStatusOk),
146 Effect.map((response) =>
147 response.stream.pipe(
148 Stream.timeoutOrElse({
149 duration: SILENT_STREAM_TIMEOUT,
150 orElse: () => Stream.fail(new OpenCode2SilentStreamError()),
151 }),
152 Stream.decodeText,
153 Stream.pipeThroughChannel(Sse.decode()),
154 Stream.filterMapEffect((frame) => {
155 const decoded = decodeEvent(frame.data);
156 return Result.isSuccess(decoded)
157 ? Effect.succeed(Result.succeed<OpenCode2StreamEvent>(decoded.success))
158 : undecodable(frame.data);
159 }),
160 ),
161 ),
162);
163
164/**
165 * OpenCode decodes Basic credentials as UTF-8. `HttpClientRequest.basicAuth`
166 * encodes them as Latin-1 (`btoa`), which gets non-ASCII passwords rejected.
167 */
168const basicAuthorization = (password: string | Redacted.Redacted) => {
169 const plain = Redacted.isRedacted(password) ? Redacted.value(password) : password;
170 return `Basic ${Buffer.from(`${OPENCODE_USERNAME}:${plain}`, "utf8").toString("base64")}`;
171};
172
173/** @public Service construction is part of the canonical Effect module API. */
174export const make = Effect.gen(function* () {
175 const httpClient = yield* HttpClient.HttpClient;
176 return OpenCode2Client.of({
177 connect: ({ baseUrl, password }) => {
178 const authenticated = HttpClient.mapRequest(
179 httpClient,
180 HttpClientRequest.setHeader("Authorization", basicAuthorization(password)),
181 );
182 return OpenCode.make({ baseUrl }).pipe(
183 Effect.provideService(HttpClient.HttpClient, authenticated),
184 Effect.map((client) => ({
185 client,
186 events: readEvents.pipe(
187 Effect.provideService(
188 HttpClient.HttpClient,
189 HttpClient.mapRequest(authenticated, HttpClientRequest.prependUrl(baseUrl)),
190 ),
191 ),
192 })),
193 );
194 },
195 });
196});
197
198export const layer = Layer.effect(OpenCode2Client, make);
199
200/**
201 * Streams every item of a cursor-paged OpenCode 2 list. The first request
202 * carries the caller's input (including `order`); later requests send only
203 * the cursor, because OpenCode rejects a cursor combined with `order`.
204 */
205export const paginate = <Input extends { readonly cursor?: unknown }, Item, E, R>(
206 input: Input,
207 list: (
208 input: Input,
209 ) => Effect.Effect<
210 { readonly data: ReadonlyArray<Item>; readonly cursor: { readonly next?: Input["cursor"] } },
211 E,
212 R
213 >,
214): Stream.Stream<Item, E, R> =>
215 Stream.paginate(input, (request) =>
216 list(request).pipe(
217 Effect.map(
218 (page) =>
219 [
220 page.data,
221 page.data.length === 0 || page.cursor.next === undefined
222 ? Option.none()
223 : Option.some({ ...request, order: undefined, cursor: page.cursor.next }),
224 ] as const,
225 ),
226 ),
227 );