1// @effect-diagnostics nodeBuiltinImport:off - Adapts dev-tui's Unix socket protocol.
2import {
3 DevServerError,
4 isProviderNativeSubagentThread,
5 type OrchestrationV2AppThread,
6 type DevServerStatus,
7 type DevServerTarget,
8 type ThreadId,
9} from "@t3tools/contracts";
10import { backgroundWorkHoldsCompletion } from "@t3tools/shared/orchestrationV2PendingBackgroundWork";
11import * as HostProcess from "@t3tools/shared/HostProcess";
12import * as Context from "effect/Context";
13import * as Effect from "effect/Effect";
14import * as FileSystem from "effect/FileSystem";
15import * as Layer from "effect/Layer";
16import * as Option from "effect/Option";
17import * as Path from "effect/Path";
18import * as Schema from "effect/Schema";
19import { FetchHttpClient, HttpClient } from "effect/http";
20import * as NodeNet from "node:net";
21
22import * as ProjectionStore from "../orchestration-v2/ProjectionStore.ts";
23import * as ProjectStore from "../orchestration-v2/ProjectStore.ts";
24import * as ProcessRunner from "../processRunner.ts";
25import { withWorkspaceLease } from "./workspaceLease.ts";
26
27const RuntimeStatus = Schema.Struct({
28 root: Schema.String,
29 dataDir: Schema.String,
30 environment: Schema.String,
31 stopping: Schema.Boolean,
32 services: Schema.Record(
33 Schema.String,
34 Schema.Struct({
35 status: Schema.String,
36 url: Schema.NullOr(Schema.String),
37 }),
38 ),
39});
40const decodeStatus = Schema.decodeUnknownSync(Schema.fromJsonString(RuntimeStatus));
41const decodeFrame = Schema.decodeUnknownSync(
42 Schema.fromJsonString(
43 Schema.Struct({
44 out: Schema.optional(Schema.String),
45 code: Schema.optional(Schema.Int),
46 }),
47 ),
48);
49
50function isParked(
51 thread: Pick<
52 OrchestrationV2AppThread,
53 "archivedAt" | "deletedAt" | "settledOverride" | "settledAt"
54 >,
55): boolean {
56 return (
57 thread.deletedAt !== null ||
58 thread.archivedAt !== null ||
59 thread.settledOverride === "settled" ||
60 (thread.settledOverride !== "active" && thread.settledAt !== null)
61 );
62}
63
64const liveRunStatuses = new Set(["queued", "preparing", "starting", "running", "waiting"]);
65
66export class DevServer extends Context.Service<
67 DevServer,
68 {
69 readonly status: (target: DevServerTarget) => Effect.Effect<DevServerStatus, DevServerError>;
70 readonly start: (target: DevServerTarget) => Effect.Effect<DevServerStatus, DevServerError>;
71 readonly stopUnused: (threadId: ThreadId) => Effect.Effect<void, DevServerError>;
72 }
73>()("t3/workspace/DevServer") {}
74
75const make = Effect.gen(function* () {
76 const fs = yield* FileSystem.FileSystem;
77 const http = HttpClient.withScope(yield* HttpClient.HttpClient);
78 const path = yield* Path.Path;
79 const projects = yield* ProjectStore.ProjectStoreV2;
80 const threads = yield* ProjectionStore.ProjectionStoreV2;
81 const processRunner = yield* ProcessRunner.ProcessRunner;
82 const home = yield* HostProcess.HomeDirectory;
83 const environment = yield* HostProcess.Environment;
84 const devTuiDirectory = path.join(home, "dev", "dev-tui");
85
86 const workspace = Effect.fn("DevServer.workspace")(function* (target: DevServerTarget) {
87 const thread =
88 target._tag === "Thread"
89 ? yield* threads
90 .getThread(target.threadId)
91 .pipe(Effect.mapError((cause) => new DevServerError({ reason: "workspace", cause })))
92 : null;
93 const projectId = target._tag === "Project" ? target.projectId : thread?.projectId;
94 if (!projectId) return yield* new DevServerError({ reason: "workspace" });
95 const project = yield* projects
96 .get(projectId)
97 .pipe(Effect.mapError((cause) => new DevServerError({ reason: "workspace", cause })));
98 if (Option.isNone(project)) return yield* new DevServerError({ reason: "workspace" });
99 const selectedRoot =
100 target._tag === "Project"
101 ? target.workspacePath
102 : (thread?.worktreePath ?? project.value.workspaceRoot);
103 return yield* fs
104 .realPath(selectedRoot)
105 .pipe(Effect.mapError((cause) => new DevServerError({ reason: "workspace", cause })));
106 });
107
108 const canStart = Effect.fn("DevServer.canStart")(function* (root: string) {
109 return yield* Effect.all([
110 fs.exists(path.join(root, "dev.sh")),
111 fs.exists(path.join(root, "dev.config.ts")),
112 fs.exists(path.join(devTuiDirectory, "dev.sh")),
113 ]).pipe(
114 Effect.map((present) => present.every(Boolean)),
115 Effect.orElseSucceed(() => false),
116 );
117 });
118
119 const request = (root: string, command: ReadonlyArray<string>) =>
120 Effect.callback<string | null, DevServerError>((resume) => {
121 const socket = NodeNet.connect(path.join(root, ".dev-tui", "ipc.sock"));
122 let frameBuffer = "";
123 let output = "";
124 let settled = false;
125 const finish = (result: Effect.Effect<string | null, DevServerError>) => {
126 if (settled) return;
127 settled = true;
128 socket.destroy();
129 resume(result);
130 };
131 socket.setEncoding("utf8");
132 socket.once("connect", () => socket.write(`${JSON.stringify(command)}\n`));
133 socket.once("error", (cause: NodeJS.ErrnoException) => {
134 finish(
135 cause.code === "ENOENT" || cause.code === "ECONNREFUSED"
136 ? Effect.succeed(null)
137 : Effect.fail(new DevServerError({ reason: "status", cause })),
138 );
139 });
140 socket.on("data", (chunk: string) => {
141 if (socket.bytesRead > 128 * 1024) {
142 finish(Effect.fail(new DevServerError({ reason: "status" })));
143 return;
144 }
145 frameBuffer += chunk;
146 const lines = frameBuffer.split("\n");
147 frameBuffer = lines.pop() ?? "";
148 try {
149 for (const line of lines) {
150 if (!line) continue;
151 const frame = decodeFrame(line);
152 output += frame.out ?? "";
153 if (frame.code !== undefined) {
154 finish(
155 frame.code === 0
156 ? Effect.succeed(output)
157 : Effect.fail(new DevServerError({ reason: "status" })),
158 );
159 return;
160 }
161 }
162 } catch (cause) {
163 finish(Effect.fail(new DevServerError({ reason: "status", cause })));
164 }
165 });
166 socket.once("close", () => finish(Effect.fail(new DevServerError({ reason: "status" }))));
167 return Effect.sync(() => {
168 socket.destroy();
169 });
170 }).pipe(
171 Effect.timeoutOrElse({
172 duration: "2 seconds",
173 orElse: () => Effect.fail(new DevServerError({ reason: "status" })),
174 }),
175 );
176
177 const probe = Effect.fn("DevServer.probe")(function* (root: string) {
178 const output = yield* request(root, ["status", "--json"]);
179 if (output === null) return null;
180 const runtime = yield* Effect.try(() => decodeStatus(output)).pipe(
181 Effect.mapError((cause) => new DevServerError({ reason: "status", cause })),
182 );
183 const [runtimeRoot, dataDir, expectedDataDir] = yield* Effect.all([
184 fs.realPath(runtime.root),
185 fs.realPath(runtime.dataDir),
186 fs.realPath(path.join(root, ".dev-tui")),
187 ]).pipe(Effect.mapError((cause) => new DevServerError({ reason: "status", cause })));
188 if (runtimeRoot !== root || dataDir !== expectedDataDir) {
189 return yield* new DevServerError({ reason: "status" });
190 }
191 return runtime;
192 });
193
194 const read = Effect.fn("DevServer.read")(function* (root: string) {
195 const [runtime, launchable] = yield* Effect.all([probe(root), canStart(root)]);
196 const services = Object.entries(runtime?.services ?? {}).map(([id, service]) => ({
197 id,
198 ...service,
199 }));
200 const client = runtime?.services.client;
201 let frontendUrl: string | null = null;
202 const clientUrl = client?.url;
203 if (client?.status === "ready" && clientUrl && !runtime?.stopping) {
204 const url = yield* Effect.try(() => new URL(clientUrl)).pipe(
205 Effect.mapError((cause) => new DevServerError({ reason: "status", cause })),
206 );
207 if (
208 url.protocol !== "http:" ||
209 !["localhost", "127.0.0.1", "[::1]"].includes(url.hostname) ||
210 url.username ||
211 url.password
212 ) {
213 return yield* new DevServerError({ reason: "status" });
214 }
215 const reachable = yield* http.get(url.href).pipe(
216 Effect.map((response) => response.status >= 200 && response.status < 400),
217 Effect.provideService(FetchHttpClient.RequestInit, { redirect: "manual" }),
218 Effect.timeout("3 seconds"),
219 Effect.scoped,
220 Effect.orElseSucceed(() => false),
221 );
222 if (reachable) frontendUrl = url.href;
223 }
224 return {
225 workspaceRoot: root,
226 canStart: launchable,
227 environment: runtime?.environment ?? null,
228 stopping: runtime?.stopping ?? false,
229 services,
230 frontendUrl,
231 } satisfies DevServerStatus;
232 });
233
234 const canonicalPath = (directory: string) =>
235 fs
236 .realPath(directory)
237 .pipe(
238 Effect.catch((cause) =>
239 cause.reason._tag === "NotFound"
240 ? Effect.succeed(null)
241 : Effect.fail(new DevServerError({ reason: "workspace", cause })),
242 ),
243 );
244
245 const unused = Effect.fn("DevServer.unused")(function* (threadId: ThreadId, root: string) {
246 const [snapshot, projectShells, targetRecords] = yield* Effect.all([
247 threads.getShellSnapshot(),
248 projects.listShells(),
249 threads.getThreadRecords(threadId, ["runs", "runtimeRequests"]),
250 ]).pipe(Effect.mapError((cause) => new DevServerError({ reason: "workspace", cause })));
251 const target = targetRecords.thread;
252 if (
253 !isParked(target) ||
254 targetRecords.runs.some((run) => liveRunStatuses.has(run.status)) ||
255 targetRecords.runtimeRequests.some((request) => request.status === "pending")
256 )
257 return false;
258 const candidates = [...snapshot.threads, ...snapshot.archivedThreads];
259 const owners = new Map(candidates.map((thread) => [thread.id, thread]));
260 const projectRoots = new Map(
261 projectShells.map((project) => [project.id, project.workspaceRoot]),
262 );
263 const targetDirectory = target.worktreePath ?? projectRoots.get(target.projectId);
264 if (targetDirectory === undefined || (yield* canonicalPath(targetDirectory)) !== root)
265 return false;
266 const canonicalRoots = new Map<string, string | null>();
267 for (const thread of candidates) {
268 const directory = thread.worktreePath ?? projectRoots.get(thread.projectId);
269 if (directory === undefined) return false;
270 if (!canonicalRoots.has(directory))
271 canonicalRoots.set(directory, yield* canonicalPath(directory));
272 const resolved = canonicalRoots.get(directory);
273 if (resolved == null) continue;
274 const relative = path.relative(root, resolved);
275 if (relative === ".." || relative.startsWith(`..${path.sep}`) || path.isAbsolute(relative))
276 continue;
277 if (
278 thread.activeRunId !== null ||
279 thread.activityRunStatus != null ||
280 liveRunStatuses.has(thread.status) ||
281 thread.pendingRuntimeRequest !== null ||
282 backgroundWorkHoldsCompletion(thread.pendingBackgroundTasks ?? [])
283 )
284 return false;
285 const records =
286 thread.id === threadId
287 ? targetRecords
288 : yield* threads
289 .getThreadRecords(thread.id, ["runs", "runtimeRequests"])
290 .pipe(Effect.mapError((cause) => new DevServerError({ reason: "workspace", cause })));
291 const current = records.thread;
292 const currentDirectory = current.worktreePath ?? projectRoots.get(current.projectId);
293 if (currentDirectory === undefined) return false;
294 const currentRoot = yield* canonicalPath(currentDirectory);
295 if (currentRoot === null) continue;
296 const currentRelative = path.relative(root, currentRoot);
297 if (
298 currentRelative === ".." ||
299 currentRelative.startsWith(`..${path.sep}`) ||
300 path.isAbsolute(currentRelative)
301 )
302 continue;
303 const ownerId =
304 current.lineage.rootThreadId === current.id
305 ? current.lineage.parentThreadId
306 : current.lineage.rootThreadId;
307 const nativeChild = isProviderNativeSubagentThread(current);
308 const owner =
309 nativeChild && ownerId !== null && (ownerId === threadId || owners.has(ownerId))
310 ? yield* threads.getThreadRecords(ownerId, []).pipe(
311 Effect.map((records) => records.thread),
312 Effect.catchTags({ ProjectionStoreThreadNotFoundError: () => Effect.succeed(null) }),
313 Effect.mapError((cause) => new DevServerError({ reason: "workspace", cause })),
314 )
315 : null;
316 if (!(nativeChild ? owner !== null && isParked(owner) : isParked(current))) return false;
317 if (
318 records.runs.some((run) => liveRunStatuses.has(run.status)) ||
319 records.runtimeRequests.some((request) => request.status === "pending")
320 )
321 return false;
322 }
323 return true;
324 });
325
326 const stopUnused = Effect.fn("DevServer.stopUnused")(function* (threadId: ThreadId) {
327 const records = yield* threads.getThreadRecords(threadId, []).pipe(
328 Effect.catchTags({ ProjectionStoreThreadNotFoundError: () => Effect.succeed(null) }),
329 Effect.mapError((cause) => new DevServerError({ reason: "workspace", cause })),
330 );
331 if (records === null) return;
332 const thread = records.thread;
333 const project = yield* projects
334 .getShell(thread.projectId)
335 .pipe(Effect.mapError((cause) => new DevServerError({ reason: "workspace", cause })));
336 if (Option.isNone(project)) return;
337 const root = yield* canonicalPath(thread.worktreePath ?? project.value.workspaceRoot);
338 if (root === null) return;
339 yield* withWorkspaceLease(
340 root,
341 Effect.gen(function* () {
342 const socketPath = path.join(root, ".dev-tui", "ipc.sock");
343 const socketInfo = yield* fs
344 .stat(socketPath)
345 .pipe(
346 Effect.catch((cause) =>
347 cause.reason._tag === "NotFound"
348 ? Effect.succeed(null)
349 : Effect.fail(new DevServerError({ reason: "status", cause })),
350 ),
351 );
352 if (socketInfo === null || !(yield* unused(threadId, root))) return;
353 const runtime = yield* probe(root);
354 if (runtime === null || runtime.stopping) return;
355 if (!(yield* unused(threadId, root))) return;
356 const canonicalSocket = yield* fs
357 .realPath(socketPath)
358 .pipe(Effect.mapError((cause) => new DevServerError({ reason: "status", cause })));
359 if (canonicalSocket !== socketPath) return yield* new DevServerError({ reason: "status" });
360 const currentSocket = yield* fs
361 .stat(socketPath)
362 .pipe(Effect.mapError((cause) => new DevServerError({ reason: "status", cause })));
363 if (
364 currentSocket.dev !== socketInfo.dev ||
365 !Option.isSome(socketInfo.ino) ||
366 !Option.isSome(currentSocket.ino) ||
367 currentSocket.ino.value !== socketInfo.ino.value
368 ) {
369 return yield* new DevServerError({ reason: "status" });
370 }
371 const accepted = yield* request(root, ["shutdown"]);
372 if (accepted !== null)
373 yield* Effect.logDebug("dev server shutdown requested", {
374 threadId,
375 workspaceRoot: root,
376 });
377 }),
378 );
379 });
380
381 return DevServer.of({
382 stopUnused,
383 status: (target) => workspace(target).pipe(Effect.flatMap(read)),
384 start: (target) =>
385 workspace(target).pipe(
386 Effect.flatMap((root) =>
387 withWorkspaceLease(
388 root,
389 Effect.gen(function* () {
390 const current = yield* read(root);
391 if (current.frontendUrl) return current;
392 if (!current.canStart) return yield* new DevServerError({ reason: "unsupported" });
393 const launched = yield* processRunner
394 .run({
395 command: "bash",
396 args: [path.join(root, "dev.sh"), "--headless"],
397 cwd: root,
398 env: { ...environment, DEV_TUI_DIR: devTuiDirectory },
399 timeout: "180 seconds",
400 maxOutputBytes: 128 * 1024,
401 outputMode: "truncate",
402 })
403 .pipe(Effect.mapError((cause) => new DevServerError({ reason: "launch", cause })));
404 if (launched.code !== 0) return yield* new DevServerError({ reason: "launch" });
405 const ready = yield* read(root);
406 if (!ready.frontendUrl) return yield* new DevServerError({ reason: "not-ready" });
407 return ready;
408 }),
409 ),
410 ),
411 ),
412 });
413});
414
415export const layer = Layer.effect(DevServer, make);