1import { ThreadId } from "@t3tools/contracts";
2import { describe, expect, it } from "@effect/vitest";
3import * as Effect from "effect/Effect";
4import * as Layer from "effect/Layer";
5
6import { PrimaryConnectionTarget, type PreparedConnection } from "../connection/model.ts";
7import * as RpcHttp from "../rpc/http.ts";
8import * as BoundedThreadSnapshotHttp from "./boundedThreadSnapshotHttp.ts";
9import * as ThreadSnapshotLoader from "./threadSnapshotHttp.ts";
10
11const TARGET = new PrimaryConnectionTarget({
12 environmentId: "environment-bounded" as never,
13 label: "Bounded",
14 httpBaseUrl: "https://environment.example.test",
15 wsBaseUrl: "wss://environment.example.test",
16});
17
18const PREPARED: PreparedConnection = {
19 environmentId: TARGET.environmentId,
20 label: TARGET.label,
21 httpBaseUrl: TARGET.httpBaseUrl,
22 socketUrl: TARGET.wsBaseUrl,
23 httpAuthorization: null,
24 target: TARGET,
25};
26
27const THREAD_ID = ThreadId.make("thread-bounded");
28const NOW = "2026-06-20T00:00:00.000Z";
29
30const FULL_SNAPSHOT_BODY = {
31 snapshotSequence: 9,
32 projection: {
33 thread: {
34 createdBy: "user",
35 creationSource: "web",
36 id: String(THREAD_ID),
37 projectId: "project-1",
38 title: "Full fallback thread",
39 providerInstanceId: "codex",
40 modelSelection: { instanceId: "codex", model: "gpt-5" },
41 runtimeMode: "full-access",
42 interactionMode: "default",
43 branch: null,
44 worktreePath: null,
45 activeProviderThreadId: null,
46 lineage: {
47 parentThreadId: null,
48 relationshipToParent: null,
49 rootThreadId: String(THREAD_ID),
50 },
51 forkedFrom: null,
52 createdAt: NOW,
53 updatedAt: NOW,
54 archivedAt: null,
55 deletedAt: null,
56 settledOverride: null,
57 settledAt: null,
58 },
59 runs: [],
60 attempts: [],
61 nodes: [],
62 subagents: [],
63 providerSessions: [],
64 providerThreads: [],
65 providerTurns: [],
66 runtimeRequests: [],
67 messages: [],
68 plans: [],
69 turnItems: [],
70 checkpointScopes: [],
71 checkpoints: [],
72 contextHandoffs: [],
73 contextTransfers: [],
74 visibleTurnItems: [],
75 updatedAt: NOW,
76 },
77};
78
79const AT = "2026-06-20T00:00:00.000Z";
80const command = (id: string, ordinal: number) => ({
81 id,
82 type: "command_execution",
83 threadId: String(THREAD_ID),
84 runId: null,
85 nodeId: null,
86 providerThreadId: null,
87 providerTurnId: null,
88 nativeItemRef: null,
89 parentItemId: null,
90 ordinal,
91 status: "completed",
92 title: null,
93 input: id,
94 exitCode: 0,
95 startedAt: AT,
96 completedAt: AT,
97 updatedAt: AT,
98});
99const LOCAL_ITEMS = [command("local-1", 1), command("local-2", 2)];
100const DEPENDENCY = command("older-dependency", 0);
101const VISIBLE = LOCAL_ITEMS.map((item, position) => ({
102 position,
103 visibility: "local",
104 sourceThreadId: String(THREAD_ID),
105 sourceItemId: item.id,
106 item,
107}));
108const boundedBody = (compact: boolean) => ({
109 snapshotSequence: 12,
110 projection: {
111 ...FULL_SNAPSHOT_BODY.projection,
112 turnItems: compact ? [DEPENDENCY] : [...LOCAL_ITEMS, DEPENDENCY],
113 visibleTurnItems: VISIBLE,
114 },
115 historyCursor: "bounded-cursor",
116 hasMoreHistory: true,
117 latestLocalTurnOrdinal: 2,
118 payloadBudgetExceeded: false,
119 ...(compact ? { turnItemsOmitLocalVisible: true } : {}),
120});
121
122describe("boundedThreadSnapshotLoader", () => {
123 it.effect.each([
124 ["an older server that ignores the opt-in", false],
125 ["a server that sends compact turnItems", true],
126 ] as const)("requests compact turnItems and loads %s", ([, compact]) => {
127 const urls: string[] = [];
128 const fetchFn = ((input: RequestInfo | URL) => {
129 urls.push(String(input));
130 return Promise.resolve(Response.json(boundedBody(compact)));
131 }) satisfies typeof fetch;
132
133 return Effect.gen(function* () {
134 const loader = yield* ThreadSnapshotLoader.ThreadSnapshotLoader;
135 const result = yield* loader.load(PREPARED, THREAD_ID);
136 expect(new URL(urls[0]!).searchParams.get("compactTurnItems")).toBe("1");
137 expect(result._tag).toBe("present");
138 if (result._tag !== "present") return;
139 expect(result.snapshot.projection.turnItems.map((item) => String(item.id))).toEqual([
140 "local-1",
141 "local-2",
142 "older-dependency",
143 ]);
144 expect("turnItemsOmitLocalVisible" in result.snapshot).toBe(false);
145 expect(result.history).toEqual({
146 historyCursor: "bounded-cursor",
147 hasMoreHistory: true,
148 latestLocalTurnOrdinal: 2,
149 });
150 }).pipe(
151 Effect.provide(
152 Layer.provide(BoundedThreadSnapshotHttp.layer, RpcHttp.layerRemoteHttpClient(fetchFn)),
153 ),
154 );
155 });
156
157 it.effect("falls back to full HTTP snapshot when bounded returns a plain route 404", () => {
158 const fetchFn = ((input: RequestInfo | URL) => {
159 const url = String(input);
160 if (url.includes(`/api/orchestration/threads/${THREAD_ID}/bounded`)) {
161 // Generic HTML/route 404: not a structured thread_not_found payload.
162 return Promise.resolve(
163 new Response("Not Found", {
164 status: 404,
165 headers: { "content-type": "text/plain" },
166 }),
167 );
168 }
169 expect(url).toContain(`/api/orchestration/threads/${THREAD_ID}`);
170 expect(url.includes("/bounded")).toBe(false);
171 return Promise.resolve(
172 new Response(JSON.stringify(FULL_SNAPSHOT_BODY), {
173 status: 200,
174 headers: { "content-type": "application/json" },
175 }),
176 );
177 }) satisfies typeof fetch;
178
179 return Effect.gen(function* () {
180 const loader = yield* ThreadSnapshotLoader.ThreadSnapshotLoader;
181 const result = yield* loader.load(PREPARED, THREAD_ID);
182 expect(result._tag).toBe("present");
183 if (result._tag === "present") {
184 expect(result.snapshot.snapshotSequence).toBe(9);
185 expect(result.snapshot.projection.thread.title).toBe("Full fallback thread");
186 expect(result.history).toBeUndefined();
187 }
188 }).pipe(
189 Effect.provide(
190 Layer.provide(BoundedThreadSnapshotHttp.layer, RpcHttp.layerRemoteHttpClient(fetchFn)),
191 ),
192 );
193 });
194
195 it.effect("treats structured EnvironmentResourceNotFoundError from bounded as missing", () => {
196 const fetchFn = ((input: RequestInfo | URL) => {
197 const url = String(input);
198 expect(url).toContain(`/api/orchestration/threads/${THREAD_ID}/bounded`);
199 return Promise.resolve(
200 Response.json(
201 {
202 _tag: "EnvironmentResourceNotFoundError",
203 code: "not_found",
204 reason: "thread_not_found",
205 traceId: "trace-bounded-missing",
206 },
207 { status: 404 },
208 ),
209 );
210 }) satisfies typeof fetch;
211
212 return Effect.gen(function* () {
213 const loader = yield* ThreadSnapshotLoader.ThreadSnapshotLoader;
214 const result = yield* loader.load(PREPARED, THREAD_ID);
215 expect(result).toEqual({ _tag: "missing" });
216 }).pipe(
217 Effect.provide(
218 Layer.provide(BoundedThreadSnapshotHttp.layer, RpcHttp.layerRemoteHttpClient(fetchFn)),
219 ),
220 );
221 });
222
223 it.effect("treats structured missing from full fallback as missing", () => {
224 let fullCalls = 0;
225 const fetchFn = ((input: RequestInfo | URL) => {
226 const url = String(input);
227 if (url.includes("/bounded")) {
228 return Promise.resolve(
229 new Response("Not Found", {
230 status: 404,
231 headers: { "content-type": "text/plain" },
232 }),
233 );
234 }
235 fullCalls += 1;
236 return Promise.resolve(
237 Response.json(
238 {
239 _tag: "EnvironmentResourceNotFoundError",
240 code: "not_found",
241 reason: "thread_not_found",
242 traceId: "trace-full-missing",
243 },
244 { status: 404 },
245 ),
246 );
247 }) satisfies typeof fetch;
248
249 return Effect.gen(function* () {
250 const loader = yield* ThreadSnapshotLoader.ThreadSnapshotLoader;
251 const result = yield* loader.load(PREPARED, THREAD_ID);
252 expect(result).toEqual({ _tag: "missing" });
253 expect(fullCalls).toBe(1);
254 }).pipe(
255 Effect.provide(
256 Layer.provide(BoundedThreadSnapshotHttp.layer, RpcHttp.layerRemoteHttpClient(fetchFn)),
257 ),
258 );
259 });
260
261 it.effect(
262 "uses socket fallback directly when the bounded endpoint has a transient failure",
263 () => {
264 let fullCalls = 0;
265 const fetchFn = ((input: RequestInfo | URL) => {
266 const url = String(input);
267 if (url.includes("/bounded")) {
268 return Promise.resolve(new Response("Service Unavailable", { status: 503 }));
269 }
270 fullCalls += 1;
271 return Promise.resolve(
272 new Response(JSON.stringify(FULL_SNAPSHOT_BODY), {
273 status: 200,
274 headers: { "content-type": "application/json" },
275 }),
276 );
277 }) satisfies typeof fetch;
278
279 return Effect.gen(function* () {
280 const loader = yield* ThreadSnapshotLoader.ThreadSnapshotLoader;
281 const result = yield* loader.load(PREPARED, THREAD_ID);
282 expect(result).toEqual({ _tag: "unavailable" });
283 expect(fullCalls).toBe(0);
284 }).pipe(
285 Effect.provide(
286 Layer.provide(BoundedThreadSnapshotHttp.layer, RpcHttp.layerRemoteHttpClient(fetchFn)),
287 ),
288 );
289 },
290 );
291});