1import {
2 AuthOrchestrationReadScope,
3 EnvironmentHttpApi,
4 type OrchestrationProjectShell,
5} from "@t3tools/contracts";
6import * as Effect from "effect/Effect";
7import * as Result from "effect/Result";
8import * as Predicate from "effect/Predicate";
9import * as Schema from "effect/Schema";
10import * as HttpApiBuilder from "effect/http-api/HttpApiBuilder";
11import * as SqlClient from "effect/sql/SqlClient";
12
13import {
14 annotateEnvironmentRequest,
15 failEnvironmentInternal,
16 failEnvironmentInvalidRequest,
17 failEnvironmentNotFound,
18 requireEnvironmentScope,
19} from "../auth/http.ts";
20import * as ServerEnvironment from "../environment/ServerEnvironment.ts";
21import { traceLocalHandlerWork } from "../cloud/traceRelayRequest.ts";
22import * as OrchestrationEventStore from "../persistence/OrchestrationEventStore.ts";
23import * as ProjectEnrichmentService from "../project/ProjectEnrichmentService.ts";
24import {
25 buildBoundedThreadProjection,
26 decodeThreadHistoryCursor,
27 InvalidThreadHistoryCursorError,
28 THREAD_HISTORY_SNAPSHOT_ROW_LIMIT,
29 THREAD_HISTORY_PAGE_POLICY,
30 OLDER_THREAD_USER_TURN_LIMIT,
31} from "./threadHistoryPaging.ts";
32import * as ThreadManagementService from "./ThreadManagementService.ts";
33import * as ProjectStore from "./ProjectStore.ts";
34import {
35 loadShellUnlessResumable,
36 loadSubscriberShellSnapshot,
37 projectIdentitiesDigest,
38} from "./ShellStream.ts";
39import { boundedSnapshotResponseFields } from "./ThreadStream.ts";
40import { projectThreadProjectionForWire } from "./WireProjection.ts";
41
42const isInvalidThreadHistoryCursorError = Schema.is(InvalidThreadHistoryCursorError);
43
44const isThreadTranscriptTooLargeError = Schema.is(
45 ThreadManagementService.ThreadTranscriptTooLargeError,
46);
47
48function isThreadNotFound(error: unknown): boolean {
49 return (
50 Predicate.hasProperty(error, "cause") &&
51 Predicate.hasProperty(error.cause, "_tag") &&
52 error.cause._tag === "ProjectionStoreThreadNotFoundError"
53 );
54}
55
56export const layer = HttpApiBuilder.group(
57 EnvironmentHttpApi,
58 "orchestration",
59 Effect.fnUntraced(function* (handlers) {
60 const sql = yield* SqlClient.SqlClient;
61 const environment = yield* ServerEnvironment.ServerEnvironmentIdentity;
62 const threadManagement = yield* ThreadManagementService.ThreadManagementService;
63 const applicationEvents = yield* OrchestrationEventStore.OrchestrationEventStore;
64 const projectStore = yield* ProjectStore.ProjectStoreV2;
65 const projectEnrichment = yield* ProjectEnrichmentService.ProjectEnrichmentService;
66
67 const enrichProjectShells = Effect.fn("http.orchestration.enrichProjectShells")(
68 (projects: ReadonlyArray<OrchestrationProjectShell>) =>
69 Effect.forEach(
70 projects,
71 (project) =>
72 // Use available enrichment only; git probes can take a minute and exceed shell-snapshot budgets.
73 projectEnrichment.getAvailable(project.workspaceRoot).pipe(
74 Effect.map((enrichment) => ({
75 ...project,
76 repositoryIdentity: enrichment.repositoryIdentity,
77 })),
78 ),
79 { concurrency: 16 },
80 ),
81 );
82
83 const loadShellSnapshot = Effect.fn("http.orchestration.loadShellSnapshot")(
84 function* (options: {
85 readonly omitIdleSubagents: boolean;
86 readonly omitOlderSettled: boolean;
87 }) {
88 const { snapshot } = yield* loadSubscriberShellSnapshot({
89 sql,
90 listShellIndex: threadManagement.listActiveShellIndex(),
91 readThreads: (threadIds) =>
92 threadManagement.readShellSnapshot({
93 location: "active",
94 ...(threadIds === undefined ? {} : { threadIds }),
95 }),
96 listProjects: projectStore.listShells(),
97 latestSequence: applicationEvents.latestApplicationSequence,
98 options,
99 });
100 const projects = yield* enrichProjectShells(snapshot.projects);
101 return { ...snapshot, projects, projectIdentities: projectIdentitiesDigest(projects) };
102 },
103 );
104
105 // A client that holds rows names their cursor; a small answer tells it to keep them.
106 const answerShellSnapshot = Effect.fn("http.orchestration.answerShellSnapshot")((query: {
107 readonly omitIdleSubagents?: string;
108 readonly omitOlderSettled?: string;
109 readonly resumeFrom?: string;
110 }) => {
111 const resumeFrom = query.resumeFrom === undefined ? Number.NaN : Number(query.resumeFrom);
112 return loadShellUnlessResumable({
113 events: applicationEvents,
114 resumeFrom: Number.isSafeInteger(resumeFrom) && resumeFrom >= 0 ? resumeFrom : undefined,
115 loadSnapshot: loadShellSnapshot({
116 omitIdleSubagents: query.omitIdleSubagents === "1",
117 omitOlderSettled: query.omitOlderSettled === "1",
118 }),
119 });
120 });
121
122 const loadThreadSnapshot = Effect.fn("http.orchestration.loadThreadSnapshot")(function* (
123 threadId: Parameters<typeof threadManagement.getThreadSnapshot>[0],
124 failureReason:
125 | "orchestration_thread_snapshot_failed"
126 | "orchestration_thread_bounded_snapshot_failed"
127 | "orchestration_thread_history_failed",
128 ) {
129 return yield* threadManagement.getThreadSnapshot(threadId).pipe(
130 Effect.map((snapshot) => ({
131 ...snapshot,
132 projection: projectThreadProjectionForWire(snapshot.projection),
133 })),
134 Effect.catch(
135 Effect.fnUntraced(function* (error) {
136 if (isThreadNotFound(error)) {
137 return yield* failEnvironmentNotFound("thread_not_found");
138 }
139 return yield* failEnvironmentInternal(failureReason, error);
140 }),
141 ),
142 );
143 });
144
145 const loadThreadSnapshotWindow = Effect.fn("http.orchestration.loadThreadSnapshotWindow")(
146 function* (
147 threadId: Parameters<typeof threadManagement.getThreadSnapshot>[0],
148 anchorItemId?: Parameters<
149 typeof threadManagement.getThreadSnapshotWindow
150 >[1]["anchorItemId"],
151 anchorThreadId?: Parameters<
152 typeof threadManagement.getThreadSnapshotWindow
153 >[1]["anchorThreadId"],
154 deferWork = false,
155 ) {
156 return yield* threadManagement
157 .getThreadSnapshotWindow(threadId, {
158 rowLimit: THREAD_HISTORY_SNAPSHOT_ROW_LIMIT,
159 deferWork,
160 userTurnLimit:
161 anchorItemId === undefined
162 ? THREAD_HISTORY_PAGE_POLICY.maxUserTurns
163 : OLDER_THREAD_USER_TURN_LIMIT,
164 ...(anchorItemId === undefined ? {} : { anchorItemId }),
165 ...(anchorThreadId === undefined ? {} : { anchorThreadId }),
166 })
167 .pipe(
168 Effect.map((snapshot) => ({
169 ...snapshot,
170 projection: projectThreadProjectionForWire(snapshot.projection),
171 })),
172 Effect.catch(
173 Effect.fnUntraced(function* (error) {
174 if (isThreadNotFound(error)) {
175 return yield* failEnvironmentNotFound("thread_not_found");
176 }
177 return yield* failEnvironmentInternal("orchestration_thread_history_failed", error);
178 }),
179 ),
180 );
181 },
182 );
183
184 return handlers
185 .handle(
186 "shellSnapshot",
187 Effect.fn("environment.orchestration.shellSnapshot")(function* (args) {
188 yield* annotateEnvironmentRequest(args.endpoint.name);
189 yield* requireEnvironmentScope(AuthOrchestrationReadScope);
190 return yield* answerShellSnapshot(args.query).pipe(
191 traceLocalHandlerWork,
192 Effect.catch((cause) =>
193 failEnvironmentInternal("orchestration_snapshot_failed", cause),
194 ),
195 );
196 }),
197 )
198 .handle(
199 "threadSnapshot",
200 Effect.fn("environment.orchestration.threadSnapshot")(function* (args) {
201 yield* annotateEnvironmentRequest(args.endpoint.name);
202 yield* requireEnvironmentScope(AuthOrchestrationReadScope);
203 const snapshot = yield* loadThreadSnapshot(
204 args.params.threadId,
205 "orchestration_thread_snapshot_failed",
206 ).pipe(traceLocalHandlerWork);
207 return {
208 snapshotSequence: snapshot.snapshotSequence,
209 projection: snapshot.projection,
210 };
211 }),
212 )
213 .handle(
214 "threadTranscript",
215 Effect.fn("environment.orchestration.threadTranscript")(function* (args) {
216 yield* annotateEnvironmentRequest(args.endpoint.name);
217 yield* requireEnvironmentScope(AuthOrchestrationReadScope);
218 return yield* threadManagement.getThreadTranscript(args.params.threadId).pipe(
219 Effect.provideService(ServerEnvironment.ServerEnvironmentIdentity, environment),
220 traceLocalHandlerWork,
221 Effect.catch(
222 Effect.fnUntraced(function* (error) {
223 if (isThreadTranscriptTooLargeError(error)) {
224 return yield* failEnvironmentInvalidRequest("thread_transcript_too_large");
225 }
226 if (isThreadNotFound(error)) {
227 return yield* failEnvironmentNotFound("thread_not_found");
228 }
229 return yield* failEnvironmentInternal(
230 "orchestration_thread_snapshot_failed",
231 error,
232 );
233 }),
234 ),
235 );
236 }),
237 )
238 .handle(
239 "threadBoundedSnapshot",
240 Effect.fn("environment.orchestration.threadBoundedSnapshot")(function* (args) {
241 yield* annotateEnvironmentRequest(args.endpoint.name);
242 yield* requireEnvironmentScope(AuthOrchestrationReadScope);
243 const snapshot = yield* loadThreadSnapshotWindow(
244 args.params.threadId,
245 undefined,
246 undefined,
247 args.query.deferWork === "1",
248 ).pipe(traceLocalHandlerWork);
249 const bounded = buildBoundedThreadProjection({
250 projection: snapshot.projection,
251 snapshotSequence: snapshot.snapshotSequence,
252 });
253 return {
254 snapshotSequence: snapshot.snapshotSequence,
255 ...boundedSnapshotResponseFields({
256 bounded,
257 compactTurnItems: args.query.compactTurnItems === "1",
258 dedupeBodies: args.query.dedupeBodies === "1",
259 compactIds: args.query.compactIds === "1",
260 }),
261 };
262 }),
263 )
264 .handle(
265 "threadHistoryPage",
266 Effect.fn("environment.orchestration.threadHistoryPage")(function* (args) {
267 yield* annotateEnvironmentRequest(args.endpoint.name);
268 yield* requireEnvironmentScope(AuthOrchestrationReadScope);
269 const decoded = Result.try(() => decodeThreadHistoryCursor(args.query.cursor));
270 if (Result.isFailure(decoded)) {
271 const cause = decoded.failure;
272 if (isInvalidThreadHistoryCursorError(cause))
273 return yield* failEnvironmentInvalidRequest("invalid_history_cursor");
274 return yield* failEnvironmentInternal("orchestration_thread_history_failed", cause);
275 }
276 return yield* threadManagement
277 .getThreadHistoryPage(
278 args.params.threadId,
279 args.query.cursor,
280 args.query.throughEntryId,
281 args.query.view === "conversation",
282 args.query.view === "work"
283 ? "work"
284 : args.query.deferWork === "1"
285 ? "deferred"
286 : undefined,
287 )
288 .pipe(
289 Effect.map((page) =>
290 args.query.compactIds === "1" ? { ...page, idTable: true as const } : page,
291 ),
292 traceLocalHandlerWork,
293 Effect.catch(
294 Effect.fnUntraced(function* (cause) {
295 if (isThreadNotFound(cause))
296 return yield* failEnvironmentNotFound("thread_not_found");
297 return yield* failEnvironmentInternal(
298 "orchestration_thread_history_failed",
299 cause,
300 );
301 }),
302 ),
303 );
304 }),
305 );
306 }),
307);