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