1import {
2 AuthOrchestrationReadScope,
3 EnvironmentHttpApi,
4 type OrchestrationProjectShell,
5} from "@t3tools/contracts";
6import * as Effect from "effect/Effect";
7import * as Predicate from "effect/Predicate";
8import * as Schema from "effect/Schema";
9import * as HttpApiBuilder from "effect/http-api/HttpApiBuilder";
10import * as SqlClient from "effect/sql/SqlClient";
11
12import {
13 annotateEnvironmentRequest,
14 failEnvironmentInternal,
15 failEnvironmentInvalidRequest,
16 failEnvironmentNotFound,
17 requireEnvironmentScope,
18} from "../auth/http.ts";
19import * as ServerEnvironment from "../environment/ServerEnvironment.ts";
20import * as OrchestrationEventStore from "../persistence/OrchestrationEventStore.ts";
21import * as ProjectEnrichmentService from "../project/ProjectEnrichmentService.ts";
22import {
23 buildBoundedThreadProjection,
24 decodeThreadHistoryCursor,
25 InvalidThreadHistoryCursorError,
26 THREAD_HISTORY_SNAPSHOT_ROW_LIMIT,
27 THREAD_HISTORY_PAGE_POLICY,
28 OLDER_THREAD_USER_TURN_LIMIT,
29} from "./threadHistoryPaging.ts";
30import * as ThreadManagementService from "./ThreadManagementService.ts";
31import * as ProjectStore from "./ProjectStore.ts";
32import { buildActiveShellSnapshot, loadShellSnapshotParts } from "./ShellStream.ts";
33import { boundedSnapshotResponseFields } from "./ThreadStream.ts";
34import { projectThreadProjectionForWire } from "./WireProjection.ts";
35
36const isThreadTranscriptTooLargeError = Schema.is(
37 ThreadManagementService.ThreadTranscriptTooLargeError,
38);
39
40function isThreadNotFound(error: unknown): boolean {
41 return (
42 Predicate.hasProperty(error, "cause") &&
43 Predicate.hasProperty(error.cause, "_tag") &&
44 error.cause._tag === "ProjectionStoreThreadNotFoundError"
45 );
46}
47
48/**
49 * Serves orchestration V2 snapshots over HTTP so clients can load the
50 * (potentially large) shell and thread projections off the socket; gzip
51 * compressible and cacheable — and then resume the WebSocket subscription via
52 * `afterSequence`.
53 */
54export const layer = HttpApiBuilder.group(
55 EnvironmentHttpApi,
56 "orchestration",
57 Effect.fnUntraced(function* (handlers) {
58 const sql = yield* SqlClient.SqlClient;
59 const environment = yield* ServerEnvironment.ServerEnvironmentIdentity;
60 const threadManagement = yield* ThreadManagementService.ThreadManagementService;
61 const applicationEvents = yield* OrchestrationEventStore.OrchestrationEventStore;
62 const projectStore = yield* ProjectStore.ProjectStoreV2;
63 const projectEnrichment = yield* ProjectEnrichmentService.ProjectEnrichmentService;
64
65 const enrichProjectShells = Effect.fn("http.orchestration.enrichProjectShells")(
66 (projects: ReadonlyArray<OrchestrationProjectShell>) =>
67 Effect.forEach(
68 projects,
69 (project) =>
70 // Use immediately available enrichment only. Awaiting git-backed
71 // identity resolution can exceed the client shell-snapshot budget
72 // (ProcessRunner allows probes up to one minute). Background workers
73 // plus the WS enrichment subscription fill in repositoryIdentity.
74 projectEnrichment.getAvailable(project.workspaceRoot).pipe(
75 Effect.map((enrichment) => ({
76 ...project,
77 repositoryIdentity: enrichment.repositoryIdentity,
78 })),
79 ),
80 { concurrency: 16 },
81 ),
82 );
83
84 const loadShellSnapshot = Effect.fn("http.orchestration.loadShellSnapshot")(function* () {
85 const base = buildActiveShellSnapshot(
86 yield* loadShellSnapshotParts({
87 sql,
88 readThreads: threadManagement.readShellSnapshot({ location: "active" }),
89 listProjects: projectStore.listShells(),
90 latestSequence: applicationEvents.latestApplicationSequence,
91 }),
92 );
93 const projects = yield* enrichProjectShells(base.projects);
94 return { ...base, projects };
95 });
96
97 const loadThreadSnapshot = Effect.fn("http.orchestration.loadThreadSnapshot")(function* (
98 threadId: Parameters<typeof threadManagement.getThreadSnapshot>[0],
99 failureReason:
100 | "orchestration_thread_snapshot_failed"
101 | "orchestration_thread_bounded_snapshot_failed"
102 | "orchestration_thread_history_failed",
103 ) {
104 return yield* threadManagement.getThreadSnapshot(threadId).pipe(
105 Effect.map((snapshot) => ({
106 ...snapshot,
107 projection: projectThreadProjectionForWire(snapshot.projection),
108 })),
109 Effect.catch(
110 Effect.fnUntraced(function* (error) {
111 if (isThreadNotFound(error)) {
112 return yield* failEnvironmentNotFound("thread_not_found");
113 }
114 return yield* failEnvironmentInternal(failureReason, error);
115 }),
116 ),
117 );
118 });
119
120 const loadThreadSnapshotWindow = Effect.fn("http.orchestration.loadThreadSnapshotWindow")(
121 function* (
122 threadId: Parameters<typeof threadManagement.getThreadSnapshot>[0],
123 anchorItemId?: Parameters<
124 typeof threadManagement.getThreadSnapshotWindow
125 >[1]["anchorItemId"],
126 anchorThreadId?: Parameters<
127 typeof threadManagement.getThreadSnapshotWindow
128 >[1]["anchorThreadId"],
129 ) {
130 return yield* threadManagement
131 .getThreadSnapshotWindow(threadId, {
132 rowLimit: THREAD_HISTORY_SNAPSHOT_ROW_LIMIT,
133 userTurnLimit:
134 anchorItemId === undefined
135 ? THREAD_HISTORY_PAGE_POLICY.maxUserTurns
136 : OLDER_THREAD_USER_TURN_LIMIT,
137 ...(anchorItemId === undefined ? {} : { anchorItemId }),
138 ...(anchorThreadId === undefined ? {} : { anchorThreadId }),
139 })
140 .pipe(
141 Effect.map((snapshot) => ({
142 ...snapshot,
143 projection: projectThreadProjectionForWire(snapshot.projection),
144 })),
145 Effect.catch(
146 Effect.fnUntraced(function* (error) {
147 if (isThreadNotFound(error)) {
148 return yield* failEnvironmentNotFound("thread_not_found");
149 }
150 return yield* failEnvironmentInternal("orchestration_thread_history_failed", error);
151 }),
152 ),
153 );
154 },
155 );
156
157 return handlers
158 .handle(
159 "shellSnapshot",
160 Effect.fn("environment.orchestration.shellSnapshot")(function* (args) {
161 yield* annotateEnvironmentRequest(args.endpoint.name);
162 yield* requireEnvironmentScope(AuthOrchestrationReadScope);
163 return yield* loadShellSnapshot().pipe(
164 Effect.catch((cause) =>
165 failEnvironmentInternal("orchestration_snapshot_failed", cause),
166 ),
167 );
168 }),
169 )
170 .handle(
171 "threadSnapshot",
172 Effect.fn("environment.orchestration.threadSnapshot")(function* (args) {
173 yield* annotateEnvironmentRequest(args.endpoint.name);
174 yield* requireEnvironmentScope(AuthOrchestrationReadScope);
175 const snapshot = yield* loadThreadSnapshot(
176 args.params.threadId,
177 "orchestration_thread_snapshot_failed",
178 );
179 return {
180 snapshotSequence: snapshot.snapshotSequence,
181 projection: snapshot.projection,
182 };
183 }),
184 )
185 .handle(
186 "threadTranscript",
187 Effect.fn("environment.orchestration.threadTranscript")(function* (args) {
188 yield* annotateEnvironmentRequest(args.endpoint.name);
189 yield* requireEnvironmentScope(AuthOrchestrationReadScope);
190 return yield* threadManagement.getThreadTranscript(args.params.threadId).pipe(
191 Effect.provideService(ServerEnvironment.ServerEnvironmentIdentity, environment),
192 Effect.catch(
193 Effect.fnUntraced(function* (error) {
194 if (isThreadTranscriptTooLargeError(error)) {
195 return yield* failEnvironmentInvalidRequest("thread_transcript_too_large");
196 }
197 if (isThreadNotFound(error)) {
198 return yield* failEnvironmentNotFound("thread_not_found");
199 }
200 return yield* failEnvironmentInternal(
201 "orchestration_thread_snapshot_failed",
202 error,
203 );
204 }),
205 ),
206 );
207 }),
208 )
209 .handle(
210 "threadBoundedSnapshot",
211 Effect.fn("environment.orchestration.threadBoundedSnapshot")(function* (args) {
212 yield* annotateEnvironmentRequest(args.endpoint.name);
213 yield* requireEnvironmentScope(AuthOrchestrationReadScope);
214 const snapshot = yield* loadThreadSnapshotWindow(args.params.threadId);
215 const bounded = buildBoundedThreadProjection({
216 projection: snapshot.projection,
217 snapshotSequence: snapshot.snapshotSequence,
218 });
219 return {
220 snapshotSequence: snapshot.snapshotSequence,
221 ...boundedSnapshotResponseFields({
222 bounded,
223 compactTurnItems: args.query.compactTurnItems === "1",
224 }),
225 };
226 }),
227 )
228 .handle(
229 "threadHistoryPage",
230 Effect.fn("environment.orchestration.threadHistoryPage")(function* (args) {
231 yield* annotateEnvironmentRequest(args.endpoint.name);
232 yield* requireEnvironmentScope(AuthOrchestrationReadScope);
233 try {
234 decodeThreadHistoryCursor(args.query.cursor);
235 } catch (cause) {
236 if (cause instanceof InvalidThreadHistoryCursorError)
237 return yield* failEnvironmentInvalidRequest("invalid_history_cursor");
238 return yield* failEnvironmentInternal("orchestration_thread_history_failed", cause);
239 }
240 return yield* threadManagement
241 .getThreadHistoryPage(
242 args.params.threadId,
243 args.query.cursor,
244 args.query.throughEntryId,
245 args.query.view === "conversation",
246 )
247 .pipe(
248 Effect.catch(
249 Effect.fnUntraced(function* (cause) {
250 if (isThreadNotFound(cause))
251 return yield* failEnvironmentNotFound("thread_not_found");
252 return yield* failEnvironmentInternal(
253 "orchestration_thread_history_failed",
254 cause,
255 );
256 }),
257 ),
258 );
259 }),
260 );
261 }),
262);