1import * as NodeServices from "@effect/platform-node/NodeServices";
2import { assert, it } from "@effect/vitest";
3import * as Effect from "effect/Effect";
4import * as Fiber from "effect/Fiber";
5import * as FileSystem from "effect/FileSystem";
6import * as Layer from "effect/Layer";
7import * as Schema from "effect/Schema";
8import * as Stream from "effect/Stream";
9import { TestClock } from "effect/testing";
10import { HttpClient, HttpClientResponse } from "effect/http";
11import { describe } from "vite-plus/test";
12
13import * as OpenCode2Client from "@t3tools/provider-opencode/server/v2/OpenCode2Client";
14
15const RECORDING = new URL(
16 "../orchestration-v2/testkit/fixtures/opencode2_simple/opencode_transcript.ndjson",
17 import.meta.url,
18);
19const Entry = Schema.Struct({ type: Schema.String, frame: Schema.optional(Schema.Unknown) });
20const decodeEntry = Schema.decodeUnknownSync(Schema.fromJsonString(Entry));
21const encodeJson = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown));
22
23/** The recorded `simple` turn's events, as a newer server would send them. */
24const newerServerStream = Effect.gen(function* () {
25 const fs = yield* FileSystem.FileSystem;
26 const recorded = (yield* fs.readFileString(RECORDING.pathname))
27 .split("\n")
28 .filter((line) => line.length > 0)
29 .map((line) => decodeEntry(line))
30 .flatMap((entry) => {
31 const frame = entry.frame as { readonly type?: string; readonly event?: unknown } | undefined;
32 return frame?.type === "sdk.event" ? [frame.event as Record<string, unknown>] : [];
33 });
34 const executionStarted = recorded.findIndex(
35 (event) => event.type === "session.execution.started",
36 );
37 const [first] = recorded;
38 return [
39 ...recorded.slice(0, executionStarted),
40 // An event type this build has never heard of.
41 {
42 id: "evt_0000000000newtype",
43 created: 1,
44 type: "session.hologram.projected",
45 data: { beams: 3 },
46 },
47 // A known event carrying a field added after 2.0.18.
48 {
49 ...recorded[executionStarted],
50 data: { ...(recorded[executionStarted]!.data as object), lane: "fast" },
51 },
52 ...recorded.slice(executionStarted + 1),
53 first,
54 ];
55});
56
57const layerServing = (events: ReadonlyArray<unknown>) =>
58 Layer.succeed(
59 HttpClient.HttpClient,
60 HttpClient.make((request) =>
61 Effect.succeed(
62 HttpClientResponse.fromWeb(
63 request,
64 new Response(events.map((event) => `data: ${encodeJson(event)}\n\n`).join(""), {
65 headers: { "content-type": "text/event-stream" },
66 }),
67 ),
68 ),
69 ),
70 );
71
72const connect = Effect.gen(function* () {
73 const opencode = yield* OpenCode2Client.OpenCode2Client;
74 return yield* opencode.connect({ baseUrl: "http://127.0.0.1:4096", password: "secret" });
75});
76
77/** Sends `events`, then keeps the connection open without another byte, like a frozen server. */
78const layerServingThenSilent = (events: ReadonlyArray<unknown>) =>
79 Layer.succeed(
80 HttpClient.HttpClient,
81 HttpClient.make((request) =>
82 Effect.succeed(
83 HttpClientResponse.fromWeb(
84 request,
85 new Response(
86 new ReadableStream<Uint8Array>({
87 start(controller) {
88 for (const event of events) {
89 controller.enqueue(new TextEncoder().encode(`data: ${encodeJson(event)}\n\n`));
90 }
91 },
92 }),
93 { headers: { "content-type": "text/event-stream" } },
94 ),
95 ),
96 ),
97 ),
98 );
99
100describe("OpenCode2Client events", () => {
101 it.effect("fails a stream that stops sending anything, even heartbeats", () =>
102 Effect.gen(function* () {
103 const { events } = yield* connect.pipe(
104 Effect.provide(
105 OpenCode2Client.layer.pipe(
106 Layer.provide(
107 layerServingThenSilent([{ id: "evt_1", type: "server.connected", data: {} }]),
108 ),
109 ),
110 ),
111 );
112 const drained = yield* (yield* events).pipe(Stream.runDrain, Effect.flip, Effect.forkChild);
113 // Let the first frame arrive and the timer start before the clock moves.
114 for (let i = 0; i < 20; i++) yield* Effect.yieldNow;
115 yield* TestClock.adjust("46 seconds");
116 const failure = yield* Fiber.join(drained);
117 assert.strictEqual(failure._tag, "OpenCode2SilentStreamError");
118 }).pipe(Effect.provide(TestClock.layer())),
119 );
120
121 it.effect(
122 "keeps an execution's end and start it cannot decode, and skips other unknown events",
123 () =>
124 Effect.gen(function* () {
125 const { events } = yield* connect.pipe(
126 Effect.provide(
127 OpenCode2Client.layer.pipe(
128 Layer.provide(
129 layerServing([
130 {
131 id: "evt_1",
132 created: 1,
133 type: "session.hologram.projected",
134 data: { sessionID: "ses_x" },
135 },
136 // A start is kept as a marker; it never ends anything.
137 {
138 id: "evt_0",
139 created: 1,
140 type: "session.execution.started",
141 data: { sessionID: "ses_x", lane: 7 },
142 durable: "not-an-envelope",
143 },
144 {
145 id: "evt_2",
146 created: 1,
147 type: "session.execution.interrupted",
148 data: { sessionID: "ses_x", reason: "budget" },
149 durable: { aggregateID: "ses_x", seq: 1, version: 1 },
150 },
151 ]),
152 ),
153 ),
154 ),
155 );
156 const received = yield* (yield* events).pipe(Stream.runCollect);
157 assert.deepStrictEqual(received, [
158 { type: "unreadable.execution.started", sessionID: "ses_x" },
159 {
160 type: "unreadable.execution.ended",
161 executionType: "session.execution.interrupted",
162 sessionID: "ses_x",
163 },
164 ]);
165 }),
166 );
167
168 it.effect("skips events a newer server adds and still sees the turn end", () =>
169 Effect.gen(function* () {
170 const events = yield* newerServerStream;
171 const { events: subscribe } = yield* connect.pipe(
172 Effect.provide(OpenCode2Client.layer.pipe(Layer.provide(layerServing(events)))),
173 );
174 const types = yield* (yield* subscribe).pipe(
175 Stream.map((event) => event.type),
176 Stream.runCollect,
177 );
178
179 assert.notInclude(types, "session.hologram.projected");
180 assert.include(types, "session.execution.started");
181 assert.include(types, "session.execution.succeeded");
182 assert.strictEqual(types.length, events.length - 1);
183 }).pipe(Effect.provide(NodeServices.layer)),
184 );
185
186 it.effect("the client's own subscription fails the whole stream on the same events", () =>
187 Effect.gen(function* () {
188 const events = yield* newerServerStream;
189 const { client } = yield* connect.pipe(
190 Effect.provide(OpenCode2Client.layer.pipe(Layer.provide(layerServing(events)))),
191 );
192 const failure = yield* client.event.subscribe().pipe(Stream.runDrain, Effect.flip);
193 assert.strictEqual(failure._tag, "ClientError");
194 }).pipe(Effect.provide(NodeServices.layer)),
195 );
196});