1import { assert, describe, it } from "@effect/vitest";
2import {
3 AuthOrchestrationReadScope,
4 AuthSessionId,
5 EnvironmentAuthenticatedAuth,
6 EnvironmentAuthenticatedPrincipal,
7 EnvironmentHttpApi,
8 EnvironmentId,
9 EventId,
10 MessageId,
11 NodeId,
12 ORCHESTRATION_PROTOCOL_HEADER,
13 ORCHESTRATION_PROTOCOL_VERSION,
14 OrchestrationV2ThreadBoundedSnapshot,
15 OrchestrationV2ThreadStreamItem,
16 ProjectId,
17 ProviderDriverKind,
18 ProviderInstanceId,
19 RunId,
20 ThreadId,
21 TurnItemId,
22 type OrchestrationV2DomainEvent,
23 type OrchestrationV2ThreadProjection,
24 type OrchestrationV2TurnItem,
25} from "@t3tools/contracts";
26import { boundedSnapshotProjection } from "@t3tools/shared/orchestrationV2BoundedSnapshot";
27import * as NodeHttpPlatform from "@effect/platform-node/NodeHttpPlatform";
28import * as NodeServices from "@effect/platform-node/NodeServices";
29import * as DateTime from "effect/DateTime";
30import * as Effect from "effect/Effect";
31import * as Layer from "effect/Layer";
32import * as Schema from "effect/Schema";
33import * as Stream from "effect/Stream";
34import * as HttpApi from "effect/http-api/HttpApi";
35import * as HttpApiBuilder from "effect/http-api/HttpApiBuilder";
36import { Etag, HttpRouter } from "effect/http";
37
38import * as ServerEnvironment from "../environment/ServerEnvironment.ts";
39import * as SqlitePersistence from "../persistence/Sqlite.ts";
40import * as OrchestrationEventStore from "../persistence/OrchestrationEventStore.ts";
41import * as ProjectEnrichmentService from "../project/ProjectEnrichmentService.ts";
42import { subscribeOrchestrationV2Thread } from "../ws.ts";
43import * as OrchestrationHttp from "./http.ts";
44import * as ProjectionStore from "./ProjectionStore.ts";
45import * as ProjectStore from "./ProjectStore.ts";
46import * as ThreadManagementService from "./ThreadManagementService.ts";
47
48const decodeBounded = Schema.decodeUnknownSync(
49 Schema.fromJsonString(Schema.toCodecJson(OrchestrationV2ThreadBoundedSnapshot)),
50);
51const decodeStreamItem = Schema.decodeUnknownSync(
52 Schema.toCodecJson(OrchestrationV2ThreadStreamItem),
53);
54const encodeStreamItem = Schema.encodeSync(Schema.toCodecJson(OrchestrationV2ThreadStreamItem));
55
56const provider = ProviderInstanceId.make("codex");
57const modelSelection = { instanceId: provider, model: "gpt-5.4" };
58const projectId = ProjectId.make("project:compact-transport");
59const PARENT = ThreadId.make("thread:compact-transport:parent");
60const FORK = ThreadId.make("thread:compact-transport:fork");
61const LONG = ThreadId.make("thread:compact-transport:long");
62const parentRun = RunId.make("run:compact-transport:parent");
63const longRun = RunId.make("run:compact-transport:long");
64
65const store = Layer.mergeAll(
66 ProjectionStore.layer.pipe(Layer.provideMerge(SqlitePersistence.layerMemory)),
67 SqlitePersistence.layerMemory,
68);
69const management = Layer.unwrap(
70 Effect.gen(function* () {
71 const projections = yield* ProjectionStore.ProjectionStoreV2;
72 return Layer.mock(ThreadManagementService.ThreadManagementService)({
73 ensureLegacyTranscript: () => Effect.void,
74 getThreadSnapshot: (id) => projections.getThreadSnapshot(id).pipe(Effect.orDie),
75 getThreadSnapshotWindow: (id, options) =>
76 projections.getThreadSnapshotWindow(id, options).pipe(Effect.orDie),
77 streamStoredEventsFrom: () => Stream.empty,
78 });
79 }),
80);
81
82class OrchestrationApi extends HttpApi.make("environment").add(
83 EnvironmentHttpApi.groups.orchestration,
84) {}
85const auth = Layer.succeed(EnvironmentAuthenticatedAuth, (effect) =>
86 effect.pipe(
87 Effect.provideService(EnvironmentAuthenticatedPrincipal, {
88 sessionId: AuthSessionId.make("compact-transport"),
89 subject: "compact-transport",
90 method: "browser-session-cookie",
91 scopes: new Set([AuthOrchestrationReadScope]),
92 }),
93 ),
94);
95
96function threadCreated(id: ThreadId, at: DateTime.Utc, forkOf?: RunId): OrchestrationV2DomainEvent {
97 return {
98 id: EventId.make(`event:${id}:created`),
99 type: "thread.created",
100 threadId: id,
101 occurredAt: at,
102 payload: {
103 createdBy: "user",
104 creationSource: "web",
105 id,
106 projectId,
107 title: String(id),
108 providerInstanceId: provider,
109 modelSelection,
110 runtimeMode: "full-access",
111 interactionMode: "default",
112 branch: null,
113 worktreePath: null,
114 activeProviderThreadId: null,
115 lineage:
116 forkOf === undefined
117 ? { parentThreadId: null, relationshipToParent: null, rootThreadId: id }
118 : { parentThreadId: PARENT, relationshipToParent: "fork", rootThreadId: PARENT },
119 forkedFrom: forkOf === undefined ? null : { type: "run", threadId: PARENT, runId: forkOf },
120 createdAt: at,
121 updatedAt: at,
122 archivedAt: null,
123 settledOverride: null,
124 settledAt: null,
125 lastVisitedAt: null,
126 deletedAt: null,
127 },
128 };
129}
130
131function runCreated(
132 threadId: ThreadId,
133 runId: RunId,
134 at: DateTime.Utc,
135): OrchestrationV2DomainEvent {
136 const rootNodeId = NodeId.make(`node:${runId}`);
137 return {
138 id: EventId.make(`event:${runId}:created`),
139 type: "run.created",
140 threadId,
141 runId,
142 nodeId: rootNodeId,
143 driver: ProviderDriverKind.make("codex"),
144 occurredAt: at,
145 payload: {
146 id: runId,
147 threadId,
148 ordinal: 1,
149 providerInstanceId: provider,
150 modelSelection,
151 providerThreadId: null,
152 userMessageId: MessageId.make(`message:${runId}`),
153 rootNodeId,
154 activeAttemptId: null,
155 status: "completed",
156 requestedAt: at,
157 startedAt: at,
158 completedAt: at,
159 checkpointId: null,
160 contextHandoffId: null,
161 },
162 };
163}
164
165function itemUpdated(
166 threadId: ThreadId,
167 runId: RunId | null,
168 key: string,
169 ordinal: number,
170 at: DateTime.Utc,
171 type:
172 | "command_execution"
173 | "run_interrupt_request"
174 | "run_interrupt_result" = "command_execution",
175): OrchestrationV2DomainEvent {
176 const base = {
177 id: TurnItemId.make(`turn-item:${threadId}:${key}`),
178 threadId,
179 runId,
180 nodeId: null,
181 providerThreadId: null,
182 providerTurnId: null,
183 nativeItemRef: null,
184 parentItemId: null,
185 ordinal,
186 status: "completed" as const,
187 title: key,
188 startedAt: at,
189 completedAt: at,
190 updatedAt: at,
191 };
192 const payload: OrchestrationV2TurnItem =
193 type === "command_execution"
194 ? { ...base, type, input: `echo ${key}`, output: `output ${key}`, exitCode: 0 }
195 : { ...base, type, message: key };
196 return {
197 id: EventId.make(`event:${threadId}:${key}`),
198 type: "turn-item.updated",
199 threadId,
200 ...(runId === null ? {} : { runId }),
201 occurredAt: at,
202 payload,
203 };
204}
205
206const seed = Effect.gen(function* () {
207 const projections = yield* ProjectionStore.ProjectionStoreV2;
208 const at = yield* DateTime.now;
209 const events: OrchestrationV2DomainEvent[] = [
210 threadCreated(PARENT, at),
211 runCreated(PARENT, parentRun, at),
212 ...[1, 2, 3].map((n) => itemUpdated(PARENT, parentRun, `parent-${n}`, n, at)),
213 threadCreated(FORK, at, parentRun),
214 ...[1, 2, 3, 4].map((n) => itemUpdated(FORK, null, `fork-${n}`, n, at)),
215 threadCreated(LONG, at),
216 runCreated(LONG, longRun, at),
217 // The request sits far outside the recent window; its result is inside it.
218 itemUpdated(LONG, longRun, "request", 1, at, "run_interrupt_request"),
219 ...Array.from({ length: 90 }, (_, index) =>
220 itemUpdated(LONG, longRun, `tool-${index}`, index + 2, at),
221 ),
222 itemUpdated(LONG, longRun, "result", 92, at, "run_interrupt_result"),
223 ];
224 for (const event of events) yield* projections.apply(event);
225});
226
227// Bounded subscriptions without afterSequence never read the event store, and
228// thread routes never touch projects.
229const TestLayer = Layer.mergeAll(
230 management,
231 Layer.succeed(ServerEnvironment.ServerEnvironmentIdentity, {
232 getEnvironmentId: Effect.succeed(EnvironmentId.make("compact-transport")),
233 }),
234 Layer.effectDiscard(seed),
235 Layer.mock(OrchestrationEventStore.OrchestrationEventStore)({}),
236 Layer.mock(ProjectStore.ProjectStoreV2)({}),
237 Layer.mock(ProjectEnrichmentService.ProjectEnrichmentService)({}),
238).pipe(Layer.provideMerge(store));
239
240const withHttp = <A>(use: (get: (path: string) => Promise<Response>) => Promise<A>) =>
241 Effect.gen(function* () {
242 const context = yield* Effect.context<Layer.Success<typeof TestLayer>>();
243 const appLayer = HttpApiBuilder.layer(OrchestrationApi).pipe(
244 Layer.provide(OrchestrationHttp.layer),
245 Layer.provide(auth),
246 Layer.provide(Layer.succeedContext(context)),
247 Layer.provide(NodeHttpPlatform.layer),
248 Layer.provide(Etag.layerWeak),
249 Layer.provide(NodeServices.layer),
250 );
251 return yield* Effect.acquireUseRelease(
252 Effect.sync(() => HttpRouter.toWebHandler(appLayer, { disableLogger: true })),
253 ({ handler }) =>
254 Effect.promise(() =>
255 use((path) =>
256 handler(
257 new Request(`http://environment.test${path}`, {
258 headers: {
259 [ORCHESTRATION_PROTOCOL_HEADER]: String(ORCHESTRATION_PROTOCOL_VERSION),
260 },
261 }),
262 ),
263 ),
264 ),
265 ({ dispose }) => Effect.promise(() => dispose()),
266 );
267 });
268
269const firstSnapshot = (input: Parameters<typeof subscribeOrchestrationV2Thread>[0]) =>
270 subscribeOrchestrationV2Thread(input).pipe(
271 Effect.flatMap((stream) => Stream.runHead(stream)),
272 Effect.map((head) => {
273 if (head._tag !== "Some" || head.value.kind !== "snapshot") throw new Error("No snapshot");
274 // Through the wire codec, as a client receives it.
275 const decoded = decodeStreamItem(JSON.parse(JSON.stringify(encodeStreamItem(head.value))));
276 if (decoded.kind !== "snapshot") throw new Error("No snapshot");
277 return { raw: head.value, decoded };
278 }),
279 );
280
281function itemIds(projection: OrchestrationV2ThreadProjection): string[] {
282 return projection.turnItems.map((item) => String(item.id));
283}
284
285it.layer(TestLayer)("compact bounded snapshot transport", (it) => {
286 describe.each([
287 ["fork", FORK],
288 ["long run with a retained interrupt request", LONG],
289 ])("%s", (_, threadId) => {
290 it.effect("HTTP: legacy, ignored and opted-in requests restore to the same snapshot", () =>
291 withHttp(async (get) => {
292 const base = `/api/orchestration/threads/${encodeURIComponent(threadId)}/bounded`;
293 const legacyBody = await (await get(base)).text();
294 const ignoredBody = await (await get(`${base}?compactTurnItems=yes&unknown=1`)).text();
295 const compactResponse = await get(`${base}?compactTurnItems=1`);
296 assert.strictEqual(compactResponse.status, 200);
297 const compactBody = await compactResponse.text();
298
299 // Requests without the exact opt-in get the unchanged representation.
300 assert.notInclude(legacyBody, "turnItemsOmitLocalVisible");
301 assert.strictEqual(ignoredBody, legacyBody);
302 assert.isBelow(compactBody.length, legacyBody.length);
303
304 const legacy = decodeBounded(legacyBody);
305 const compact = decodeBounded(compactBody);
306 assert.isTrue(compact.turnItemsOmitLocalVisible);
307 const { turnItemsOmitLocalVisible: _marker, projection: _p, ...compactRest } = compact;
308 const { projection: _legacyProjection, ...legacyRest } = legacy;
309 assert.deepStrictEqual(compactRest, legacyRest);
310 assert.deepStrictEqual(boundedSnapshotProjection(compact), legacy.projection);
311 return { legacy, compact };
312 }).pipe(
313 Effect.tap(({ legacy }) =>
314 Effect.sync(() => {
315 if (threadId === FORK) {
316 const visibility = legacy.projection.visibleTurnItems.map((row) => row.visibility);
317 assert.deepStrictEqual(visibility, [
318 "inherited",
319 "inherited",
320 "inherited",
321 "synthetic",
322 "local",
323 "local",
324 "local",
325 "local",
326 ]);
327 } else {
328 const ids = itemIds(legacy.projection);
329 assert.isTrue(legacy.hasMoreHistory);
330 // Window rows first, then the retained request outside the window.
331 assert.strictEqual(ids.at(-1), `turn-item:${LONG}:request`);
332 assert.isFalse(
333 legacy.projection.visibleTurnItems.some((row) => row.sourceItemId === ids.at(-1)),
334 );
335 }
336 }),
337 ),
338 ),
339 );
340
341 it.effect("WS: bounded snapshots stay unchanged unless the client opts in", () =>
342 Effect.gen(function* () {
343 const legacy = yield* firstSnapshot({ threadId, acceptBoundedSnapshot: true });
344 const compact = yield* firstSnapshot({
345 threadId,
346 acceptBoundedSnapshot: true,
347 acceptCompactTurnItems: true,
348 });
349 // Full snapshots ignore the compact opt-in entirely.
350 const full = yield* firstSnapshot({ threadId, acceptCompactTurnItems: true });
351
352 assert.isUndefined(legacy.decoded.turnItemsOmitLocalVisible);
353 assert.isUndefined(full.decoded.turnItemsOmitLocalVisible);
354 assert.isUndefined(full.decoded.historyCursor);
355 assert.isTrue(compact.decoded.turnItemsOmitLocalVisible);
356 assert.isBelow(
357 compact.decoded.projection.turnItems.length,
358 legacy.decoded.projection.turnItems.length,
359 );
360 assert.deepStrictEqual(
361 boundedSnapshotProjection(compact.decoded),
362 legacy.decoded.projection,
363 );
364 assert.deepStrictEqual(
365 {
366 ...compact.decoded,
367 projection: legacy.decoded.projection,
368 turnItemsOmitLocalVisible: undefined,
369 },
370 { ...legacy.decoded, turnItemsOmitLocalVisible: undefined },
371 );
372 }),
373 );
374 });
375});