1import {
2 AuthSessionId,
3 type AuthEnvironmentScope,
4 EnvironmentAuthenticatedAuth,
5 EnvironmentAuthenticatedPrincipal,
6 EnvironmentHttpApi,
7 EnvironmentId,
8 EventId,
9 ORCHESTRATION_PROTOCOL_HEADER,
10 ORCHESTRATION_PROTOCOL_VERSION_TEXT,
11 OrchestrationV2AppThread,
12 OrchestrationV2Run,
13 OrchestrationV2TurnItem,
14 OrchestrationV2ThreadTranscript,
15 PROVIDER_SEND_TURN_MAX_FILE_BYTES,
16 ThreadId,
17} from "@t3tools/contracts";
18import { expect, it } from "@effect/vitest";
19import { threadTranscriptHeader } from "@t3tools/shared/threadTranscript";
20import * as Effect from "effect/Effect";
21import * as Context from "effect/Context";
22import * as DateTime from "effect/DateTime";
23import * as Layer from "effect/Layer";
24import * as Schema from "effect/Schema";
25import * as SqlClient from "effect/sql/SqlClient";
26import * as Statement from "effect/sql/Statement";
27import * as HttpRouter from "effect/http/HttpRouter";
28import * as HttpPlatform from "effect/http/HttpPlatform";
29import * as Etag from "effect/http/Etag";
30import * as NodeServices from "@effect/platform-node/NodeServices";
31import * as HttpApi from "effect/http-api/HttpApi";
32import * as HttpApiBuilder from "effect/http-api/HttpApiBuilder";
33import * as ServerEnvironment from "../environment/ServerEnvironment.ts";
34import * as SqlitePersistence from "../persistence/Sqlite.ts";
35import * as ServerSettings from "../serverSettings.ts";
36
37import * as OrchestrationEventStore from "../persistence/OrchestrationEventStore.ts";
38import { ProjectEnrichmentService } from "../project/ProjectEnrichmentService.ts";
39import * as OrchestrationHttp from "./http.ts";
40import { EventSinkV2 } from "./EventSink.ts";
41import * as ProjectStore from "./ProjectStore.ts";
42import * as CommandReceiptStore from "./CommandReceiptStore.ts";
43import * as ThreadManagementService from "./ThreadManagementService.ts";
44import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts";
45import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts";
46
47class TranscriptTestApi extends HttpApi.make("environment").add(
48 EnvironmentHttpApi.groups.orchestration,
49) {}
50const decodeTranscript = Schema.decodeUnknownEffect(
51 Schema.toCodecJson(OrchestrationV2ThreadTranscript),
52);
53
54const now = DateTime.makeUnsafe("2026-09-29T00:00:00.000Z");
55const environmentId = EnvironmentId.make("source-بيئة");
56const output = "Full tool output. ".repeat(4_000);
57const thread = Schema.decodeSync(OrchestrationV2AppThread)({
58 id: "source-thread",
59 projectId: "source-project",
60 title: "Source 界\nconversation",
61 providerInstanceId: "codex",
62 modelSelection: { instanceId: "codex", model: "gpt-5.4" },
63 runtimeMode: "full-access",
64 interactionMode: "default",
65 branch: null,
66 worktreePath: null,
67 activeProviderThreadId: null,
68 lineage: { rootThreadId: "source-thread", parentThreadId: null, relationshipToParent: null },
69 forkedFrom: null,
70 createdBy: "user",
71 creationSource: "web",
72 createdAt: now,
73 updatedAt: now,
74 archivedAt: null,
75 settledOverride: null,
76 settledAt: null,
77 lastVisitedAt: null,
78 deletedAt: null,
79});
80const items = Schema.decodeUnknownSync(Schema.Array(OrchestrationV2TurnItem))(
81 Array.from({ length: 200 }, (_, position) => ({
82 id: `item-${position}`,
83 threadId: "source-thread",
84 runId: null,
85 nodeId: null,
86 providerThreadId: null,
87 providerTurnId: null,
88 nativeItemRef: null,
89 parentItemId: null,
90 ordinal: position,
91 type: "command_execution",
92 status: "completed",
93 title: "Read notes",
94 input: "cat notes.md",
95 output: position === 0 ? output : "Done",
96 startedAt: now,
97 completedAt: now,
98 updatedAt: now,
99 })),
100);
101
102const rolledBackRun = Schema.decodeSync(OrchestrationV2Run)({
103 id: "run:rolled-back",
104 threadId: "source-thread",
105 ordinal: 1,
106 providerInstanceId: "codex",
107 modelSelection: { instanceId: "codex", model: "gpt-5.4" },
108 providerThreadId: null,
109 userMessageId: "message:rolled-back",
110 rootNodeId: null,
111 activeAttemptId: null,
112 status: "rolled_back",
113 requestedAt: now,
114 startedAt: now,
115 completedAt: now,
116 checkpointId: null,
117 contextHandoffId: null,
118});
119
120// Serves the orchestration HTTP routes over real stores; `allowed` gates the read scope.
121const serveOrchestrationHttp = (allowed: () => boolean) =>
122 Effect.gen(function* () {
123 const runtime = ProviderReplayHarness.layerWithRegistry(
124 { name: "thread-transcript" },
125 ProviderAdapterRegistry.layerFromAdapters([]),
126 { runEffectWorker: false },
127 );
128 const services = yield* Layer.build(
129 Layer.mergeAll(
130 runtime,
131 ThreadManagementService.layer.pipe(
132 Layer.provide(
133 Layer.mergeAll(runtime, CommandReceiptStore.layer, ServerSettings.layerTest({})),
134 ),
135 ),
136 OrchestrationEventStore.layer,
137 ProjectStore.layer,
138 ).pipe(Layer.provideMerge(SqlitePersistence.layerMemory)),
139 );
140 const eventSink = Context.get(services, EventSinkV2);
141 yield* eventSink.write({
142 events: [
143 {
144 id: EventId.make("create-source"),
145 type: "thread.created",
146 threadId: thread.id,
147 occurredAt: now,
148 payload: thread,
149 },
150 ],
151 });
152 const routes = HttpApiBuilder.layer(TranscriptTestApi).pipe(
153 Layer.provide(OrchestrationHttp.layer),
154 Layer.provide(
155 Layer.succeed(ServerEnvironment.ServerEnvironmentIdentity, {
156 getEnvironmentId: Effect.succeed(environmentId),
157 }),
158 ),
159 Layer.provide(
160 Layer.succeed(EnvironmentAuthenticatedAuth, (effect) =>
161 Effect.suspend(() =>
162 effect.pipe(
163 Effect.provideService(EnvironmentAuthenticatedPrincipal, {
164 sessionId: AuthSessionId.make("test-session"),
165 subject: "test",
166 method: "bearer-access-token",
167 scopes: new Set<AuthEnvironmentScope>(allowed() ? ["orchestration:read"] : []),
168 }),
169 ),
170 ),
171 ),
172 ),
173 Layer.provide(Layer.succeedContext(services)),
174 Layer.provide(Layer.mock(ProjectEnrichmentService)({})),
175 Layer.provide(HttpPlatform.layer),
176 Layer.provide(Etag.layerWeak),
177 Layer.provide(NodeServices.layer),
178 );
179 const { handler, dispose } = HttpRouter.toWebHandler(routes, { disableLogger: true });
180 yield* Effect.addFinalizer(() => Effect.promise(dispose));
181 return { services, eventSink, handler };
182 });
183
184it.effect("exports full history with size, scope, and missing-thread checks", () =>
185 Effect.gen(function* () {
186 let allowed = true;
187 const { services, eventSink, handler } = yield* serveOrchestrationHttp(() => allowed);
188 const read = (threadId: ThreadId) =>
189 Effect.promise(() =>
190 handler(
191 new Request(`http://localhost/api/orchestration/threads/${threadId}/transcript`, {
192 headers: { [ORCHESTRATION_PROTOCOL_HEADER]: ORCHESTRATION_PROTOCOL_VERSION_TEXT },
193 }),
194 ),
195 );
196
197 const readTranscriptError = Context.get(
198 services,
199 ThreadManagementService.ThreadManagementService,
200 )
201 .getThreadTranscript(thread.id)
202 .pipe(
203 Effect.provideService(ServerEnvironment.ServerEnvironmentIdentity, {
204 getEnvironmentId: Effect.succeed(environmentId),
205 }),
206 Effect.flip,
207 );
208
209 const emptyResponse = yield* read(thread.id);
210 expect(emptyResponse.status).toBe(200);
211 expect(
212 (yield* decodeTranscript(yield* Effect.promise(() => emptyResponse.json()))).items,
213 ).toEqual([]);
214 yield* eventSink.write({
215 events: items.map((item) => ({
216 id: EventId.make(`create-${item.id}`),
217 type: "turn-item.updated" as const,
218 threadId: thread.id,
219 occurredAt: now,
220 payload: item,
221 })),
222 });
223
224 const response = yield* read(thread.id);
225 expect(response.status).toBe(200);
226 const transcript = yield* decodeTranscript(yield* Effect.promise(() => response.json()));
227 expect(transcript.items).toHaveLength(200);
228 expect(transcript.items[0]?.item).toMatchObject({ output });
229 expect(transcript.items[0]?.visibility).toBe("local");
230 expect(transcript.items[199]?.position).toBe(199);
231
232 const emptyOutputRows = transcript.items.map((row, position) =>
233 position === 0 ? { ...row, item: { ...row.item, output: "" } } : row,
234 );
235 const header = threadTranscriptHeader(environmentId, transcript);
236 const baseSize =
237 Buffer.byteLength(header, "utf8") +
238 emptyOutputRows.reduce(
239 (size, row) => size + Buffer.byteLength(`${JSON.stringify(row)}\n`, "utf8"),
240 0,
241 );
242 const boundaryOutput = "界" + "x".repeat(PROVIDER_SEND_TURN_MAX_FILE_BYTES - baseSize - 3);
243 for (const [suffix, expectedStatus] of [
244 ["", 200],
245 ["x", 400],
246 ] as const) {
247 yield* eventSink.write({
248 events: [
249 {
250 id: EventId.make(`boundary-${expectedStatus}`),
251 type: "turn-item.updated",
252 threadId: thread.id,
253 occurredAt: now,
254 payload: {
255 ...items.find((item) => item.type === "command_execution")!,
256 output: boundaryOutput + suffix,
257 },
258 },
259 ],
260 });
261 const response = yield* read(thread.id);
262 expect(response.status).toBe(expectedStatus);
263 if (expectedStatus === 200) continue;
264 expect(yield* readTranscriptError).toBeInstanceOf(
265 ThreadManagementService.ThreadTranscriptTooLargeError,
266 );
267 expect(yield* Effect.promise(() => response.json())).toMatchObject({
268 _tag: "EnvironmentRequestInvalidError",
269 reason: "thread_transcript_too_large",
270 });
271 }
272
273 const sql = Context.get(services, SqlClient.SqlClient);
274 const commandItemId = items.find((item) => item.type === "command_execution")!.id;
275 const setCommandPayload = (payload: Statement.Fragment) => sql`
276 UPDATE orchestration_v2_projection_turn_items SET payload_json = ${payload}
277 WHERE turn_item_id = ${commandItemId}
278 `;
279 const fill = (bytes: number) => sql`printf('%.*c', ${bytes}, 'x')`;
280 // Stored bytes the export drops do not count, and neither does a row SQLite cannot parse:
281 // only what the export keeps can reject it.
282 yield* setCommandPayload(sql`json_set(
283 payload_json, '$.output', 'x',
284 '$.droppedLegacyField', ${fill(PROVIDER_SEND_TURN_MAX_FILE_BYTES + 1)}
285 )`);
286 expect((yield* read(thread.id)).status).toBe(200);
287 const deepField = sql`'"droppedDeepField":' || printf('%.*c', 1001, '[') || printf('%.*c', 1001, ']') || ','`;
288 yield* setCommandPayload(
289 sql`replace(payload_json, '"droppedLegacyField":', ${deepField} || '"droppedLegacyField":')`,
290 );
291 expect((yield* read(thread.id)).status).toBe(200);
292 // Tool output over the limit rejects the export, here as the content array a dynamic tool
293 // stores. Its only read is the size.
294 yield* sql`
295 UPDATE orchestration_v2_projection_turn_items SET type = 'dynamic_tool'
296 WHERE turn_item_id = ${commandItemId}
297 `;
298 yield* setCommandPayload(sql`json_set(
299 replace(payload_json, ${deepField}, ''), '$.type', 'dynamic_tool', '$.toolName', NULL,
300 '$.output',
301 json_array(json_object('type', 'text', 'text', ${fill(PROVIDER_SEND_TURN_MAX_FILE_BYTES + 1)}))
302 )`);
303 const reads: Array<string> = [];
304 expect(
305 yield* readTranscriptError.pipe(
306 Effect.provideService(Statement.CurrentTransformer, (statement) => {
307 reads.push(statement.compile()[0]);
308 return Effect.succeed(statement);
309 }),
310 ),
311 ).toBeInstanceOf(ThreadManagementService.ThreadTranscriptTooLargeError);
312 expect(reads).toEqual([expect.stringContaining("octet_length(payload_json)")]);
313 expect((yield* read(thread.id)).status).toBe(400);
314 // The export leaves a rolled back run out, so its output cannot reject it.
315 yield* eventSink.write({
316 events: [
317 {
318 id: EventId.make("create-rolled-back-run"),
319 type: "run.created",
320 threadId: thread.id,
321 occurredAt: now,
322 payload: rolledBackRun,
323 },
324 ],
325 });
326 yield* sql`
327 UPDATE orchestration_v2_projection_turn_items SET run_id = ${rolledBackRun.id}
328 WHERE turn_item_id = ${commandItemId}
329 `;
330 yield* setCommandPayload(sql`json_set(payload_json, '$.runId', ${rolledBackRun.id})`);
331 const withoutRolledBackRun = yield* read(thread.id);
332 expect(withoutRolledBackRun.status).toBe(200);
333 expect(
334 (yield* decodeTranscript(yield* Effect.promise(() => withoutRolledBackRun.json()))).items,
335 ).toHaveLength(199);
336
337 expect((yield* read(ThreadId.make("missing"))).status).toBe(404);
338 allowed = false;
339 expect((yield* read(thread.id)).status).toBe(403);
340 }),
341);
342
343it.effect("leaves older settled threads out of the HTTP shell only when asked", () =>
344 Effect.gen(function* () {
345 const { eventSink, handler } = yield* serveOrchestrationHttp(() => true);
346 yield* eventSink.write({
347 events: Array.from({ length: 26 }, (_, index) => {
348 const id = ThreadId.make(`settled-${index}`);
349 return {
350 id: EventId.make(`create-${id}`),
351 type: "thread.created" as const,
352 threadId: id,
353 occurredAt: now,
354 payload: {
355 ...thread,
356 id,
357 lineage: { ...thread.lineage, rootThreadId: id },
358 settledOverride: "settled" as const,
359 settledAt: DateTime.makeUnsafe(Date.UTC(2026, 8, 1 + index)),
360 },
361 };
362 }),
363 });
364 const shell = (query: string) =>
365 Effect.promise(async () => {
366 const response = await handler(
367 new Request(`http://localhost/api/orchestration/shell${query}`, {
368 headers: { [ORCHESTRATION_PROTOCOL_HEADER]: ORCHESTRATION_PROTOCOL_VERSION_TEXT },
369 }),
370 );
371 return (await response.json()) as {
372 readonly snapshotSequence: number;
373 readonly projectIdentities?: string;
374 readonly threads: ReadonlyArray<{ readonly id: string }>;
375 readonly olderSettled?: { readonly counts: ReadonlyArray<unknown> };
376 };
377 });
378
379 const full = yield* shell("");
380 expect(full.threads).toHaveLength(27);
381 expect(full.olderSettled).toBeUndefined();
382 const cut = yield* shell("?omitOlderSettled=1");
383 expect(cut.threads.map((row) => row.id).toSorted()).toEqual(
384 [thread.id, ...Array.from({ length: 25 }, (_, index) => `settled-${index + 1}`)].toSorted(),
385 );
386 expect(cut.olderSettled?.counts).toEqual([{ projectId: thread.projectId, count: 1 }]);
387
388 // A client that holds rows the socket can replay from keeps them; any other cursor reloads.
389 expect(full.projectIdentities).toBeTypeOf("string");
390 expect(yield* shell(`?resumeFrom=${full.snapshotSequence}`)).toEqual({ resumable: true });
391 expect(yield* shell(`?resumeFrom=${full.snapshotSequence - 1}`)).toEqual({ resumable: true });
392 for (const cursor of [full.snapshotSequence + 1, "x"]) {
393 expect((yield* shell(`?omitOlderSettled=1&resumeFrom=${cursor}`)).threads).toHaveLength(26);
394 }
395 }),
396);