Open History to see the last commit
1import {
2 CONFIGURED_LOCAL_SERVER_URLS_MAX_ITEMS,
3 PREVIEW_URL_MAX_LENGTH,
4 PreviewStopLocalServerError,
5 ThreadId,
6 type DiscoveredLocalServer,
7 type PreviewStopLocalServerInput,
8} from "@t3tools/contracts";
9import { signalProcessGroup } from "@t3tools/provider-core/server/processGroup";
10import * as HostProcess from "@t3tools/shared/HostProcess";
11import { isLoopbackHost, LSOF_LOCAL_HOST_TOKENS } from "@t3tools/shared/preview";
12import * as Cause from "effect/Cause";
13import * as Clock from "effect/Clock";
14import * as Context from "effect/Context";
15import * as Duration from "effect/Duration";
16import * as Effect from "effect/Effect";
17import * as FileSystem from "effect/FileSystem";
18import * as Layer from "effect/Layer";
19import * as Option from "effect/Option";
20import * as PlatformError from "effect/PlatformError";
21import * as Ref from "effect/Ref";
22import * as Schedule from "effect/Schedule";
23import * as Scope from "effect/Scope";
24import * as Semaphore from "effect/Semaphore";
25import { FetchHttpClient, HttpClient } from "effect/http";
26
27import * as WindowsTcpListeners from "../process/windowsTcpListeners.ts";
28import * as ProcessRunner from "../processRunner.ts";
29
30export class PortDiscovery extends Context.Service<
31 PortDiscovery,
32 {
33 readonly scan: (
34 configuredUrls?: ReadonlyArray<string>,
35 ) => Effect.Effect<ReadonlyArray<DiscoveredLocalServer>>;
36 readonly subscribe: (
37 input: {
38 readonly configuredUrls: ReadonlyArray<string>;
39 readonly initialSnapshot: ReadonlyArray<DiscoveredLocalServer>;
40 },
41 listener: (servers: ReadonlyArray<DiscoveredLocalServer>) => Effect.Effect<void>,
42 ) => Effect.Effect<void, never, Scope.Scope>;
43 readonly retain: Effect.Effect<void, never, Scope.Scope>;
44 readonly registerTerminalProcesses: (input: {
45 readonly threadId: string;
46 readonly terminalId: string;
47 readonly processIds: ReadonlyArray<number>;
48 }) => Effect.Effect<void>;
49 readonly unregisterTerminal: (input: {
50 readonly threadId: string;
51 readonly terminalId: string;
52 }) => Effect.Effect<void>;
53 /**
54 * Counts a running agent until the returned function is called. While one runs, any
55 * listener may be a server it started. Synchronous because provider spawn callbacks run
56 * outside Effect.
57 */
58 readonly trackAgent: () => () => void;
59 /** Stops a listed server's process group, never this server, an agent CLI, or a shell. */
60 readonly stopServer: (
61 input: PreviewStopLocalServerInput,
62 ) => Effect.Effect<void, PreviewStopLocalServerError>;
63 }
64>()("t3/preview/PortScanner/PortDiscovery") {}
65
66/**
67 * Agents launch with this set to their thread, and every job they start inherits it, so a
68 * server stays attributed after its launching shell exits or the agent settles.
69 */
70export const AGENT_THREAD_ENV = "T3CODE_THREAD_ID";
71
72/** `0` turns off listener discovery; configured preview URLs are still probed. */
73const PORT_DISCOVERY_ENV = "T3CODE_PORT_DISCOVERY";
74
75const POLL_INTERVAL = Duration.seconds(3);
76const LSOF_TIMEOUT_MS = 5_000;
77const PROC_FD_WALK_LIMIT = 50_000;
78const SOCKET_OWNER_ATTEMPTS = 3;
79const WINDOWS_LISTENER_TIMEOUT_MS = 5_000;
80const WEB_PROBE_TIMEOUT = Duration.seconds(1);
81const WEB_PROBE_CACHE_TTL_MS = Duration.toMillis(Duration.seconds(15));
82/** Bounds how long a cached Windows image name can outlive its process if Windows reuses the PID. */
83const WINDOWS_PROCESS_NAME_TTL_MS = Duration.toMillis(Duration.seconds(60));
84const WINDOWS_PROCESS_NAME_CONCURRENCY = 4;
85const WEB_PROBE_CONCURRENCY = 16;
86const NAVIGATION_REDIRECT_STATUSES = new Set([301, 302, 303, 307, 308]);
87const PS_TIMEOUT_MS = 5_000;
88const PS_MAX_OUTPUT_BYTES = 8 * 1024 * 1024;
89const STOP_GRACE_MS = 3_000;
90const STOP_POLL_INTERVAL = Duration.millis(250);
91
92type Listener = (servers: ReadonlyArray<DiscoveredLocalServer>) => Effect.Effect<void>;
93
94interface ListenerSubscription {
95 readonly configuredUrls: ReadonlyArray<string>;
96 readonly lastSnapshot: ReadonlyArray<DiscoveredLocalServer>;
97}
98
99interface ScannerState {
100 readonly listeners: ReadonlyMap<Listener, ListenerSubscription>;
101 readonly terminalProcesses: ReadonlyMap<
102 string,
103 {
104 readonly owner: TerminalProcessOwner;
105 readonly processIds: ReadonlySet<number>;
106 }
107 >;
108 readonly retainCount: number;
109}
110
111interface TerminalProcessOwner {
112 readonly threadId: ThreadId;
113 readonly terminalId: string;
114}
115
116type ServerOwner = Pick<DiscoveredLocalServer, "terminal" | "agent">;
117
118const UNOWNED: ServerOwner = { terminal: null, agent: null };
119
120const isOwned = (server: DiscoveredLocalServer): boolean =>
121 server.terminal !== null || (server.agent ?? null) !== null;
122
123interface ProcessTableRow {
124 readonly pid: number;
125 readonly ppid: number;
126 readonly pgid: number;
127}
128
129interface WebProbeCacheEntry {
130 readonly pid: number | null;
131 readonly isWeb: boolean;
132 readonly expiresAtMillis: number;
133}
134
135interface WebProbeGroup {
136 readonly server: DiscoveredLocalServer;
137 readonly url: string;
138 readonly configuredKey: string;
139}
140
141interface WebProbeSnapshot {
142 readonly discovered: ReadonlyArray<DiscoveredLocalServer>;
143 readonly configured: ReadonlyMap<string, DiscoveredLocalServer>;
144}
145
146const terminalOwnerKey = (owner: {
147 readonly threadId: string;
148 readonly terminalId: string;
149}): string => `${owner.threadId}\u0000${owner.terminalId}`;
150
151const parseConfiguredUrl = (raw: string): URL | null => {
152 try {
153 const url = new URL(raw);
154 if (url.protocol !== "http:" && url.protocol !== "https:") return null;
155 if (!isLoopbackHost(url.hostname)) return null;
156 return url;
157 } catch {
158 return null;
159 }
160};
161
162const localServerKey = (host: string, port: number): string =>
163 `${isLoopbackHost(host) ? "loopback" : host.toLowerCase()}:${port}`;
164
165const urlPort = (url: URL): number =>
166 url.port.length > 0 ? Number.parseInt(url.port, 10) : url.protocol === "http:" ? 80 : 443;
167
168const webProbeCacheKey = (raw: string): string => {
169 const url = new URL(raw);
170 url.hash = "";
171 return url.href;
172};
173
174const normalizeConfiguredUrls = (urls: ReadonlyArray<string>): ReadonlyArray<string> => [
175 ...new Set(
176 urls
177 .slice(0, CONFIGURED_LOCAL_SERVER_URLS_MAX_ITEMS)
178 .filter((raw) => raw.length <= PREVIEW_URL_MAX_LENGTH)
179 .map(parseConfiguredUrl)
180 .filter((url): url is URL => url !== null && url.href.length <= PREVIEW_URL_MAX_LENGTH)
181 .map((url) => {
182 if (url.hostname === "0.0.0.0") url.hostname = "localhost";
183 return url.href;
184 })
185 .filter((url) => url.length <= PREVIEW_URL_MAX_LENGTH),
186 ),
187];
188
189const projectWebProbeSnapshot = (
190 snapshot: WebProbeSnapshot,
191 configuredUrls: ReadonlyArray<string>,
192): ReadonlyArray<DiscoveredLocalServer> => {
193 const visibleByServer = new Map<string, DiscoveredLocalServer>();
194 const firstConfiguredByServer = new Map<string, string>();
195 for (const raw of normalizeConfiguredUrls(configuredUrls)) {
196 const url = new URL(raw);
197 const port = urlPort(url);
198 const serverKey = localServerKey(url.hostname, port);
199 if (!firstConfiguredByServer.has(serverKey)) firstConfiguredByServer.set(serverKey, raw);
200 if (visibleByServer.has(serverKey)) continue;
201 const configured = snapshot.configured.get(webProbeCacheKey(raw));
202 if (configured) visibleByServer.set(serverKey, { ...configured, url: raw });
203 }
204 // An owned listener whose configured check failed still opens at this caller's configured URL.
205 for (const server of snapshot.discovered) {
206 const key = localServerKey(server.host, server.port);
207 if (visibleByServer.has(key)) continue;
208 const configuredUrl = firstConfiguredByServer.get(key);
209 visibleByServer.set(key, configuredUrl ? { ...server, url: configuredUrl } : server);
210 }
211 return [...visibleByServer.values()].toSorted((left, right) => left.port - right.port);
212};
213
214const parseLsofOutput = (
215 raw: string,
216 owners: ReadonlyMap<number, ServerOwner> = new Map(),
217): ReadonlyArray<DiscoveredLocalServer> => {
218 const seen = new Map<string, DiscoveredLocalServer>();
219 let pid: number | null = null;
220 let processName: string | null = null;
221
222 for (const line of raw.split("\n")) {
223 if (line.length === 0) continue;
224 const tag = line.charAt(0);
225 const value = line.slice(1);
226 if (tag === "p") {
227 const parsed = Number.parseInt(value, 10);
228 pid = Number.isFinite(parsed) && parsed > 0 ? parsed : null;
229 processName = null;
230 continue;
231 }
232 if (tag === "c") {
233 processName = value.trim() || null;
234 continue;
235 }
236 if (tag === "n") {
237 const portMatch = parsePortFromLsofName(value);
238 if (portMatch == null) continue;
239 const url = `http://localhost:${portMatch}`;
240 const key = `localhost:${portMatch}`;
241 const owner = pid === null ? undefined : owners.get(pid);
242 const existing = seen.get(key);
243 if (existing) {
244 if (owner === undefined) seen.set(key, { ...existing, ...UNOWNED });
245 continue;
246 }
247 seen.set(key, {
248 host: "localhost",
249 port: portMatch,
250 url,
251 processName,
252 pid,
253 ...(owner ?? UNOWNED),
254 });
255 }
256 }
257
258 return Array.from(seen.values()).toSorted((a, b) => a.port - b.port);
259};
260
261const parsePortFromLsofName = (name: string): number | null => {
262 const trimmed = name.split(" ", 1)[0]?.trim() ?? "";
263 if (trimmed.length === 0) return null;
264 const lastColon = trimmed.lastIndexOf(":");
265 if (lastColon < 0) return null;
266 const hostPart = trimmed.slice(0, lastColon);
267 const portPart = trimmed.slice(lastColon + 1);
268 if (!LSOF_LOCAL_HOST_TOKENS.has(hostPart)) return null;
269 const port = Number.parseInt(portPart, 10);
270 if (!Number.isFinite(port) || port <= 0 || port >= 65536) return null;
271 return port;
272};
273
274const exitedWithError = (result: ProcessRunner.ProcessRunOutput): boolean =>
275 result.code !== null && result.code !== 0;
276
277const parseTasklistProcessName = (raw: string, pid: number): string | null => {
278 for (const line of raw.split(/\r?\n/g)) {
279 const match = /^"([^"]*)","(\d+)"/.exec(line.trim());
280 if (match && Number(match[2]) === pid) return match[1]!.replace(/\.exe$/i, "") || null;
281 }
282 return null;
283};
284
285const discoveredWindowsListeners = (
286 listeners: ReadonlyArray<WindowsTcpListeners.WindowsTcpListener>,
287 owners: ReadonlyMap<number, ServerOwner>,
288): ReadonlyArray<DiscoveredLocalServer> => {
289 const seen = new Map<number, DiscoveredLocalServer>();
290 for (const { port, pid } of listeners) {
291 const owner = pid === null ? undefined : owners.get(pid);
292 const existing = seen.get(port);
293 if (existing) {
294 if (owner === undefined) seen.set(port, { ...existing, ...UNOWNED });
295 continue;
296 }
297 seen.set(port, {
298 host: "localhost",
299 port,
300 url: `http://localhost:${port}`,
301 processName: null,
302 pid,
303 ...(owner ?? UNOWNED),
304 });
305 }
306 return [...seen.values()].toSorted((left, right) => left.port - right.port);
307};
308
309const PROC_TCP_LISTEN = "0A";
310
311/** Each 32-bit word of the hex address is little-endian. */
312const procLocalAddress = (hex: string): boolean => {
313 if (hex.length === 8) {
314 return hex === "00000000" || hex === "0100007F";
315 }
316 if (hex.length !== 32) return false;
317 if (hex === "0".repeat(32) || hex === `${"0".repeat(24)}01000000`) return true;
318 return hex.startsWith(`${"0".repeat(16)}FFFF0000`) && procLocalAddress(hex.slice(24));
319};
320
321const parseProcNetTcp = (
322 raw: string,
323): ReadonlyArray<{ readonly port: number; readonly inode: string }> => {
324 const listeners: Array<{ port: number; inode: string }> = [];
325 for (const line of raw.split("\n").slice(1)) {
326 const fields = line.trim().split(/\s+/);
327 const [address, portHex] = fields[1]?.split(":") ?? [];
328 if (fields[3] !== PROC_TCP_LISTEN || !address || !portHex) continue;
329 if (!procLocalAddress(address.toUpperCase())) continue;
330 const port = Number.parseInt(portHex, 16);
331 const inode = fields[9];
332 if (!Number.isInteger(port) || port <= 0 || port >= 65536 || !inode) continue;
333 listeners.push({ port, inode });
334 }
335 return listeners;
336};
337
338/** Processes holding a loopback or wildcard listener in `lsof -F pcn` output. */
339const lsofProcessIds = (raw: string): ReadonlyArray<number> => {
340 const pids = new Set<number>();
341 let pid: number | null = null;
342 for (const line of raw.split("\n")) {
343 if (line.startsWith("p")) {
344 const parsed = Number.parseInt(line.slice(1), 10);
345 pid = Number.isFinite(parsed) && parsed > 0 ? parsed : null;
346 } else if (
347 line.startsWith("n") &&
348 pid !== null &&
349 parsePortFromLsofName(line.slice(1)) !== null
350 ) {
351 pids.add(pid);
352 }
353 }
354 return [...pids];
355};
356
357const AGENT_THREAD_ASSIGNMENT = new RegExp(`(?:^|\\s)${AGENT_THREAD_ENV}=(\\S+)`);
358
359/** Reads `/proc/<pid>/environ`, whose entries are NUL-separated. */
360const parseProcEnvironThread = (raw: string): ThreadId | null => {
361 const prefix = `${AGENT_THREAD_ENV}=`;
362 const value = raw
363 .split("\0")
364 .find((entry) => entry.startsWith(prefix))
365 ?.slice(prefix.length)
366 .trim();
367 return value ? ThreadId.make(value) : null;
368};
369
370/**
371 * Reads macOS `ps -E -o pid=,command=`, which appends each process's launch environment to its
372 * argv with spaces, so an argument that spells the assignment out also matches.
373 */
374const parsePsEnvironmentThreads = (raw: string): ReadonlyMap<number, ThreadId | null> => {
375 const threads = new Map<number, ThreadId | null>();
376 for (const line of raw.split("\n")) {
377 const match = /^\s*(\d+)\s(.*)$/.exec(line);
378 if (!match) continue;
379 const value = AGENT_THREAD_ASSIGNMENT.exec(match[2]!)?.[1];
380 threads.set(Number(match[1]), value ? ThreadId.make(value) : null);
381 }
382 return threads;
383};
384
385const parseProcessTable = (raw: string): ReadonlyMap<number, ProcessTableRow> => {
386 const rows = new Map<number, ProcessTableRow>();
387 for (const line of raw.split("\n")) {
388 const [pid, ppid, pgid] = line.trim().split(/\s+/).map(Number);
389 if (Number.isInteger(pid) && Number.isInteger(ppid) && Number.isInteger(pgid)) {
390 rows.set(pid!, { pid: pid!, ppid: ppid!, pgid: pgid! });
391 }
392 }
393 return rows;
394};
395
396/**
397 * The process group stopping `pid` signals. Shells with job control and agent tool shells give
398 * every job its own group, so the group is the job. Null when the group also holds this server,
399 * a direct child of it (agent CLIs and terminal shells), or one of its ancestors.
400 */
401export const stoppableProcessGroup = (
402 table: ReadonlyMap<number, ProcessTableRow>,
403 pid: number,
404 serverPid: number,
405): number | null => {
406 const group = table.get(pid)?.pgid;
407 if (group === undefined || group <= 1) return null;
408 const guarded = new Set([serverPid]);
409 for (
410 let parent = table.get(serverPid)?.ppid;
411 parent !== undefined && parent > 1 && !guarded.has(parent);
412 parent = table.get(parent)?.ppid
413 ) {
414 guarded.add(parent);
415 }
416 for (const row of table.values()) {
417 if (row.pgid === group && (guarded.has(row.pid) || row.ppid === serverPid)) return null;
418 }
419 return group;
420};
421
422const serversEqual = (
423 left: ReadonlyArray<DiscoveredLocalServer>,
424 right: ReadonlyArray<DiscoveredLocalServer>,
425): boolean => {
426 if (left.length !== right.length) return false;
427 for (let i = 0; i < left.length; i += 1) {
428 const a = left[i];
429 const b = right[i];
430 if (!a || !b) return false;
431 if (
432 a.host !== b.host ||
433 a.port !== b.port ||
434 a.url !== b.url ||
435 a.processName !== b.processName ||
436 a.pid !== b.pid ||
437 a.terminal?.threadId !== b.terminal?.threadId ||
438 a.terminal?.terminalId !== b.terminal?.terminalId ||
439 a.agent?.threadId !== b.agent?.threadId
440 ) {
441 return false;
442 }
443 }
444 return true;
445};
446
447const isNoSuchProcess = (cause: unknown): boolean =>
448 typeof cause === "object" && cause !== null && "code" in cause && cause.code === "ESRCH";
449
450const isCommandNotFound = (error: ProcessRunner.ProcessSpawnError): boolean =>
451 PlatformError.isPlatformError(error.cause) && error.cause.reason._tag === "NotFound";
452
453/** @public */
454export const make = Effect.gen(function* PortDiscoveryMake() {
455 const processRunner = yield* ProcessRunner.ProcessRunner;
456 const fileSystem = yield* FileSystem.FileSystem;
457 const hostPlatform = yield* HostProcess.Platform;
458 const listenerDiscovery = (yield* HostProcess.Environment)[PORT_DISCOVERY_ENV] !== "0";
459 const httpClient = (yield* HttpClient.HttpClient).pipe(HttpClient.withScope);
460 const stateRef = yield* Ref.make<ScannerState>({
461 listeners: new Map(),
462 terminalProcesses: new Map(),
463 retainCount: 0,
464 });
465 const webProbeCacheRef = yield* Ref.make<ReadonlyMap<string, WebProbeCacheEntry>>(new Map());
466 const scanSemaphore = yield* Semaphore.make(1);
467 const lsofMissingRef = yield* Ref.make(false);
468 const windowsProcessNamesRef = yield* Ref.make<
469 ReadonlyMap<number, { readonly name: string | null; readonly expiresAtMillis: number }>
470 >(new Map());
471 const socketOwnersRef = yield* Ref.make<
472 ReadonlyMap<string, { readonly pid: number; readonly processName: string | null }>
473 >(new Map());
474 const socketOwnerMissesRef = yield* Ref.make<ReadonlyMap<string, number>>(new Map());
475 // Each current listener pid's agent thread, or null without one. A process keeps its launch
476 // environment, so a pid is read once while it listens. Cleared while no client retains the
477 // scanner, since unscanned pids can be reused.
478 const agentThreadsRef = yield* Ref.make<ReadonlyMap<number, ThreadId | null>>(new Map());
479 let runningAgents = 0;
480 // An agent that exited, or a cache cleared with a marked listener, owes one listener read: the
481 // servers it left may still run. So does startup, for agents that ran before a restart.
482 let agentScanPending = true;
483
484 const readAgentThreads = Effect.fn("PortDiscovery.readAgentThreads")(function* (
485 pids: ReadonlyArray<number>,
486 ) {
487 const cached = yield* Ref.get(agentThreadsRef);
488 const unread = pids.filter((pid) => !cached.has(pid));
489 const read: ReadonlyMap<number, ThreadId | null> =
490 unread.length === 0
491 ? new Map()
492 : hostPlatform === "linux"
493 ? new Map(
494 yield* Effect.forEach(
495 unread,
496 (pid) =>
497 fileSystem.readFileString(`/proc/${pid}/environ`).pipe(
498 Effect.map(parseProcEnvironThread),
499 Effect.orElseSucceed(() => null),
500 Effect.map((threadId) => [pid, threadId] as const),
501 ),
502 { concurrency: 16 },
503 ),
504 )
505 : yield* processRunner
506 .run({
507 command: "ps",
508 args: ["-E", "-ww", "-o", "pid=,command=", "-p", unread.join(",")],
509 timeout: Duration.millis(PS_TIMEOUT_MS),
510 maxOutputBytes: PS_MAX_OUTPUT_BYTES,
511 })
512 .pipe(
513 Effect.map((result) => parsePsEnvironmentThreads(result.stdout)),
514 Effect.orElseSucceed(() => {
515 agentScanPending = true;
516 return new Map();
517 }),
518 );
519 const next = new Map<number, ThreadId | null>();
520 for (const pid of pids) {
521 const threadId = cached.has(pid) ? cached.get(pid) : read.get(pid);
522 if (threadId !== undefined) next.set(pid, threadId);
523 }
524 yield* Ref.set(agentThreadsRef, next);
525 return next;
526 });
527
528 const withAgentOwners = Effect.fn("PortDiscovery.withAgentOwners")(function* (
529 terminalOwners: ReadonlyMap<number, ServerOwner>,
530 pids: ReadonlyArray<number>,
531 ) {
532 // Runs only once a listener snapshot exists, so a failed read leaves the scan pending.
533 agentScanPending = false;
534 const threads = yield* readAgentThreads(pids.filter((pid) => !terminalOwners.has(pid)));
535 const owners = new Map(terminalOwners);
536 for (const [pid, threadId] of threads) {
537 if (threadId !== null) owners.set(pid, { terminal: null, agent: { threadId } });
538 }
539 return owners;
540 });
541
542 const findSocketOwners = Effect.fn("PortDiscovery.findSocketOwners")(function* (
543 inodes: ReadonlySet<string>,
544 ) {
545 const owners = new Map<string, { pid: number; processName: string | null }>();
546 const entries = yield* fileSystem.readDirectory("/proc").pipe(Effect.option);
547 if (Option.isNone(entries)) return null;
548 const pids = entries.value.filter((entry) => /^\d+$/.test(entry));
549 let budget = PROC_FD_WALK_LIMIT;
550 for (const pidText of pids) {
551 if (owners.size === inodes.size || budget <= 0) break;
552 const fds = yield* fileSystem
553 .readDirectory(`/proc/${pidText}/fd`)
554 .pipe(Effect.orElseSucceed((): ReadonlyArray<string> => []));
555 const walked = fds.slice(0, budget);
556 budget -= walked.length;
557 const links = yield* Effect.forEach(
558 walked,
559 (fd) =>
560 fileSystem.readLink(`/proc/${pidText}/fd/${fd}`).pipe(Effect.orElseSucceed(() => "")),
561 { concurrency: 16 },
562 );
563 const held = links.flatMap((link) => /^socket:\[(\d+)\]$/.exec(link)?.[1] ?? []);
564 const matched = held.filter((inode) => inodes.has(inode) && !owners.has(inode));
565 if (matched.length === 0) continue;
566 const processName = yield* fileSystem.readFileString(`/proc/${pidText}/comm`).pipe(
567 Effect.map((name) => name.trim() || null),
568 Effect.orElseSucceed(() => null),
569 );
570 for (const inode of matched) owners.set(inode, { pid: Number(pidText), processName });
571 }
572 const lsofMissing = yield* Ref.get(lsofMissingRef);
573 return budget <= 0 && owners.size < inodes.size && !lsofMissing ? null : owners;
574 });
575
576 const scanProcListeners = Effect.fn("PortDiscovery.scanProcListeners")(function* (
577 resolveOwners: (pids: ReadonlyArray<number>) => Effect.Effect<ReadonlyMap<number, ServerOwner>>,
578 ) {
579 if (hostPlatform !== "linux") return null;
580 const tables = yield* Effect.forEach(["/proc/net/tcp", "/proc/net/tcp6"], (path) =>
581 fileSystem.readFileString(path).pipe(
582 // Kernels without IPv6 omit tcp6; a denied table can hide listeners on a shared port.
583 Effect.catchIf(
584 (error) => error.reason._tag === "NotFound",
585 () => Effect.succeed(null),
586 ),
587 Effect.option,
588 ),
589 );
590 if (tables.some(Option.isNone)) return null;
591 const readable = tables.flatMap((table) =>
592 Option.isSome(table) && table.value !== null ? [table.value] : [],
593 );
594 if (
595 readable.length === 0 ||
596 readable.some(
597 (table) => !/^\s*sl\s+local_address\s+(?:rem|remote)_address\s+st\s/.test(table),
598 )
599 )
600 return null;
601 const listeners = readable.flatMap(parseProcNetTcp);
602 const known = yield* Ref.get(socketOwnersRef);
603 const knownMisses = yield* Ref.get(socketOwnerMissesRef);
604 const inodes = new Set(listeners.map((listener) => listener.inode));
605 // Walk fds only for unowned sockets; stop after repeated misses for inaccessible or exited processes.
606 const unseen = new Set(
607 [...inodes].filter(
608 (inode) => !known.has(inode) && (knownMisses.get(inode) ?? 0) < SOCKET_OWNER_ATTEMPTS,
609 ),
610 );
611 const found = unseen.size === 0 ? new Map() : yield* findSocketOwners(unseen);
612 if (found === null) return null;
613 const owners = new Map([...known, ...found].filter(([inode]) => inodes.has(inode)));
614 const misses = new Map([...knownMisses].filter(([inode]) => inodes.has(inode)));
615 for (const inode of unseen) {
616 if (!found.has(inode)) misses.set(inode, (misses.get(inode) ?? 0) + 1);
617 }
618 yield* Ref.set(socketOwnersRef, owners);
619 yield* Ref.set(socketOwnerMissesRef, misses);
620 const serverOwners = yield* resolveOwners([
621 ...new Set([...owners.values()].map(({ pid }) => pid)),
622 ]);
623 const seen = new Map<number, DiscoveredLocalServer>();
624 for (const { port, inode } of listeners) {
625 const socketOwner = owners.get(inode) ?? null;
626 const owner = socketOwner === null ? undefined : serverOwners.get(socketOwner.pid);
627 const existing = seen.get(port);
628 if (existing && (existing.pid !== null || socketOwner === null)) {
629 if (owner === undefined) seen.set(port, { ...existing, ...UNOWNED });
630 continue;
631 }
632 seen.set(port, {
633 host: "localhost",
634 port,
635 url: `http://localhost:${port}`,
636 processName: socketOwner?.processName ?? null,
637 pid: socketOwner?.pid ?? null,
638 ...(existing ? UNOWNED : (owner ?? UNOWNED)),
639 });
640 }
641 return [...seen.values()].toSorted((left, right) => left.port - right.port);
642 });
643
644 const probeWebUrl = Effect.fn("PortDiscovery.probeWebUrl")((url: string) =>
645 httpClient.get(url).pipe(
646 Effect.map((response) => {
647 const location = response.headers.location?.trim();
648 if (NAVIGATION_REDIRECT_STATUSES.has(response.status) && location) return url;
649 if (response.status < 200 || response.status >= 300) return null;
650 if (response.status === 204 || response.status === 205) return null;
651 const contentType = response.headers["content-type"]
652 ?.split(";", 1)[0]
653 ?.trim()
654 .toLowerCase();
655 return contentType === "text/html" || contentType === "application/xhtml+xml" ? url : null;
656 }),
657 Effect.scoped,
658 Effect.timeoutOption(WEB_PROBE_TIMEOUT),
659 Effect.map(Option.getOrNull),
660 Effect.orElseSucceed(() => null),
661 Effect.provideService(FetchHttpClient.RequestInit, { redirect: "manual" }),
662 ),
663 );
664
665 const makeWebProbeGroups = (
666 servers: ReadonlyArray<DiscoveredLocalServer>,
667 configuredUrls: ReadonlyArray<string>,
668 ): ReadonlyArray<WebProbeGroup> => {
669 const serversByKey = new Map(
670 servers.map((server) => [localServerKey(server.host, server.port), server] as const),
671 );
672 const groups: WebProbeGroup[] = [];
673 const configuredResources = new Set<string>();
674
675 for (const raw of configuredUrls) {
676 const url = new URL(raw);
677 const port = urlPort(url);
678 const key = localServerKey(url.hostname, port);
679 const resourceKey = webProbeCacheKey(raw);
680 if (configuredResources.has(resourceKey)) continue;
681 configuredResources.add(resourceKey);
682 groups.push({
683 server: serversByKey.get(key) ?? {
684 host: url.hostname,
685 port,
686 url: raw,
687 processName: null,
688 pid: null,
689 terminal: null,
690 },
691 url: raw,
692 configuredKey: resourceKey,
693 });
694 }
695
696 return groups;
697 };
698
699 // Only configured URLs are requested. Owned listeners are listed as plain HTTP without a
700 // request: owning a process says nothing about its protocol, and HTTP or TLS bytes break
701 // binary ones such as Unity's test runner or Steam's IPC.
702 // Idle polls omit this span; requested scans and uncached probes retain their own spans.
703 const probeWebServers = Effect.fnUntraced(function* (
704 servers: ReadonlyArray<DiscoveredLocalServer>,
705 configuredUrls: ReadonlyArray<string>,
706 ) {
707 const nowMillis = yield* Clock.currentTimeMillis;
708 const cached = yield* Ref.get(webProbeCacheRef);
709 const groups = makeWebProbeGroups(servers, configuredUrls);
710 const probed = yield* Effect.forEach(
711 groups,
712 (group) => {
713 const { pid } = group.server;
714 const cachedProbe = cached.get(group.configuredKey);
715 return cachedProbe?.pid === pid && cachedProbe.expiresAtMillis > nowMillis
716 ? Effect.succeed({ group, probe: cachedProbe, fresh: false })
717 : probeWebUrl(group.url).pipe(
718 Effect.map((result) => ({
719 group,
720 probe: { pid, isWeb: result !== null, expiresAtMillis: 0 },
721 fresh: true,
722 })),
723 );
724 },
725 { concurrency: WEB_PROBE_CONCURRENCY },
726 );
727 const completedAtMillis = yield* Clock.currentTimeMillis;
728 const nextCache = new Map(
729 [...cached].filter(([, probe]) => probe.expiresAtMillis > completedAtMillis),
730 );
731 const configured = new Map<string, DiscoveredLocalServer>();
732 for (const { group, probe, fresh } of probed) {
733 nextCache.set(
734 group.configuredKey,
735 fresh ? { ...probe, expiresAtMillis: completedAtMillis + WEB_PROBE_CACHE_TTL_MS } : probe,
736 );
737 if (probe.isWeb) configured.set(group.configuredKey, { ...group.server, url: group.url });
738 }
739 yield* Ref.set(webProbeCacheRef, nextCache);
740 return { discovered: servers.filter(isOwned), configured } satisfies WebProbeSnapshot;
741 });
742
743 // tasklist costs 200-300ms per PID; query visible servers with bounded concurrency and cache names.
744 const withWindowsProcessNames = Effect.fn("PortDiscovery.withWindowsProcessNames")(function* (
745 snapshot: WebProbeSnapshot,
746 ) {
747 const servers = [...snapshot.discovered, ...snapshot.configured.values()];
748 const pids = new Set(servers.flatMap((server) => (server.pid === null ? [] : [server.pid])));
749 const nowMillis = yield* Clock.currentTimeMillis;
750 const cached = [...(yield* Ref.get(windowsProcessNamesRef))].filter(
751 ([pid, entry]) => pids.has(pid) && entry.expiresAtMillis > nowMillis,
752 );
753 const cachedPids = new Set(cached.map(([pid]) => pid));
754 const expiresAtMillis = nowMillis + WINDOWS_PROCESS_NAME_TTL_MS;
755 const fetched = yield* Effect.forEach(
756 [...pids].filter((pid) => !cachedPids.has(pid)),
757 (pid) =>
758 processRunner
759 .run({
760 command: "tasklist.exe",
761 args: ["/FO", "CSV", "/NH", "/FI", `PID eq ${pid}`],
762 timeout: Duration.millis(WINDOWS_LISTENER_TIMEOUT_MS),
763 maxOutputBytes: 64 * 1024,
764 })
765 .pipe(
766 Effect.map((result) =>
767 exitedWithError(result)
768 ? []
769 : [
770 [
771 pid,
772 { name: parseTasklistProcessName(result.stdout, pid), expiresAtMillis },
773 ] as const,
774 ],
775 ),
776 Effect.orElseSucceed(() => []),
777 ),
778 { concurrency: WINDOWS_PROCESS_NAME_CONCURRENCY },
779 );
780 const names = new Map([...cached, ...fetched.flat()]);
781 yield* Ref.set(windowsProcessNamesRef, names);
782 const withName = (server: DiscoveredLocalServer): DiscoveredLocalServer =>
783 server.pid === null || server.processName !== null
784 ? server
785 : { ...server, processName: names.get(server.pid)?.name ?? null };
786 return {
787 discovered: snapshot.discovered.map(withName),
788 configured: new Map([...snapshot.configured].map(([key, server]) => [key, withName(server)])),
789 } satisfies WebProbeSnapshot;
790 });
791
792 const recoverProcessProbeFailure =
793 (probe: "lsof" | "windows-listeners") => (error: ProcessRunner.ProcessRunError) =>
794 Effect.logDebug("preview port process probe failed; probing configured URLs only", {
795 cause: error,
796 probe,
797 platform: hostPlatform,
798 }).pipe(Effect.as(null));
799
800 // Routine polling helpers share operation/probe spans instead of tracing every tick.
801 const scanUnlocked = Effect.fnUntraced(function* (configuredUrls: ReadonlyArray<string>) {
802 if (!listenerDiscovery) return yield* probeWebServers([], configuredUrls);
803 const state = yield* Ref.get(stateRef);
804 const terminalOwners = new Map<number, ServerOwner>();
805 for (const registration of state.terminalProcesses.values()) {
806 for (const processId of registration.processIds) {
807 terminalOwners.set(processId, { terminal: registration.owner, agent: null });
808 }
809 }
810 // A server an agent started can outlive the agent, so listeners are read while the last scan
811 // still found one.
812 const agentsPossible =
813 hostPlatform !== "win32" &&
814 (runningAgents > 0 ||
815 agentScanPending ||
816 [...(yield* Ref.get(agentThreadsRef)).values()].some((threadId) => threadId !== null));
817 if (terminalOwners.size === 0 && configuredUrls.length === 0 && !agentsPossible) {
818 return { discovered: [], configured: new Map() } satisfies WebProbeSnapshot;
819 }
820 const resolveOwners = (pids: ReadonlyArray<number>) =>
821 agentsPossible ? withAgentOwners(terminalOwners, pids) : Effect.succeed(terminalOwners);
822 if (hostPlatform === "win32") {
823 const recoverWindowsProbeFailure = recoverProcessProbeFailure("windows-listeners");
824 const listeners = yield* WindowsTcpListeners.queryWindowsTcpListeners(
825 processRunner,
826 "netstat.exe",
827 ).pipe(
828 Effect.map((snapshot) =>
829 snapshot === null ? null : discoveredWindowsListeners(snapshot, terminalOwners),
830 ),
831 Effect.catchTags({
832 ProcessSpawnError: recoverWindowsProbeFailure,
833 ProcessStdinError: recoverWindowsProbeFailure,
834 ProcessOutputLimitError: recoverWindowsProbeFailure,
835 ProcessReadError: recoverWindowsProbeFailure,
836 ProcessTimeoutError: recoverWindowsProbeFailure,
837 }),
838 );
839 return yield* withWindowsProcessNames(
840 yield* probeWebServers(listeners ?? [], configuredUrls),
841 );
842 }
843 const fromProc = yield* scanProcListeners(resolveOwners);
844 if (fromProc !== null) return yield* probeWebServers(fromProc, configuredUrls);
845 const recoverLsofProbeFailure = recoverProcessProbeFailure("lsof");
846 if (yield* Ref.get(lsofMissingRef)) return yield* probeWebServers([], configuredUrls);
847 const lsofResult = yield* processRunner
848 .run({
849 command: "lsof",
850 args: ["-iTCP", "-sTCP:LISTEN", "-P", "-n", "-F", "pcn"],
851 timeout: Duration.millis(LSOF_TIMEOUT_MS),
852 // Partial output could hide another app's listener on an owned port.
853 maxOutputBytes: 1024 * 1024,
854 })
855 .pipe(
856 Effect.flatMap((result) =>
857 resolveOwners(lsofProcessIds(result.stdout)).pipe(
858 Effect.map((owners) => parseLsofOutput(result.stdout, owners)),
859 ),
860 ),
861 Effect.catchTags({
862 ProcessSpawnError: (error) =>
863 (isCommandNotFound(error) ? Ref.set(lsofMissingRef, true) : Effect.void).pipe(
864 Effect.andThen(recoverLsofProbeFailure(error)),
865 ),
866 ProcessStdinError: recoverLsofProbeFailure,
867 ProcessOutputLimitError: recoverLsofProbeFailure,
868 ProcessReadError: recoverLsofProbeFailure,
869 ProcessTimeoutError: recoverLsofProbeFailure,
870 }),
871 );
872 if (lsofResult !== null) return yield* probeWebServers(lsofResult, configuredUrls);
873 // A missing lsof makes a bounded, partial /proc owner walk the only available discovery.
874 const withoutLsof = (yield* Ref.get(lsofMissingRef))
875 ? yield* scanProcListeners(resolveOwners)
876 : null;
877 return yield* probeWebServers(withoutLsof ?? [], configuredUrls);
878 });
879
880 const scanSnapshot = (configuredUrls: ReadonlyArray<string>) =>
881 scanSemaphore.withPermits(1)(scanUnlocked(configuredUrls));
882
883 const scanOnce: PortDiscovery["Service"]["scan"] = (configuredUrls = []) => {
884 const normalized = normalizeConfiguredUrls(configuredUrls);
885 return scanSnapshot(normalized).pipe(
886 Effect.map((snapshot) => projectWebProbeSnapshot(snapshot, normalized)),
887 Effect.withSpan("PortDiscovery.scan"),
888 );
889 };
890
891 const pollTick = Effect.fnUntraced(
892 function* () {
893 const configuredUrls = [
894 ...new Set(
895 [...(yield* Ref.get(stateRef)).listeners.values()].flatMap(
896 (subscription) => subscription.configuredUrls,
897 ),
898 ),
899 ];
900 const snapshot = yield* scanSnapshot(configuredUrls);
901 const notifications = yield* Ref.modify(stateRef, (state) => {
902 const listeners = new Map(state.listeners);
903 const changed: Array<readonly [Listener, ReadonlyArray<DiscoveredLocalServer>]> = [];
904 for (const [listener, subscription] of listeners) {
905 const next = projectWebProbeSnapshot(snapshot, subscription.configuredUrls);
906 if (serversEqual(subscription.lastSnapshot, next)) continue;
907 listeners.set(listener, { ...subscription, lastSnapshot: next });
908 changed.push([listener, next]);
909 }
910 return [changed, { ...state, listeners }];
911 });
912 yield* Effect.forEach(notifications, ([listener, servers]) => listener(servers), {
913 discard: true,
914 });
915 },
916 Effect.catchCause((cause: Cause.Cause<never>) =>
917 Effect.logWarning("preview port scan failed", Cause.pretty(cause)),
918 ),
919 );
920
921 const pollIfRetained = Ref.get(stateRef).pipe(
922 Effect.flatMap((state) => (state.retainCount > 0 ? pollTick() : Effect.void)),
923 );
924 yield* Effect.forkScoped(pollIfRetained.pipe(Effect.repeat(Schedule.spaced(POLL_INTERVAL))));
925
926 const acquireRetention = Effect.fn("PortDiscovery.retain")(function* () {
927 const wasIdle = yield* Ref.modify(stateRef, (state) => [
928 state.retainCount === 0,
929 { ...state, retainCount: state.retainCount + 1 },
930 ]);
931 if (wasIdle) {
932 yield* pollTick();
933 }
934 });
935
936 const releaseRetention = Effect.gen(function* () {
937 const idle = yield* Ref.modify(stateRef, (state) => {
938 const retainCount = Math.max(0, state.retainCount - 1);
939 return [retainCount === 0, { ...state, retainCount }];
940 });
941 if (!idle) return;
942 yield* scanSemaphore.withPermits(1)(
943 Effect.gen(function* () {
944 yield* Ref.set(socketOwnersRef, new Map());
945 yield* Ref.set(socketOwnerMissesRef, new Map());
946 const threads = yield* Ref.getAndSet(agentThreadsRef, new Map());
947 if ([...threads.values()].some((threadId) => threadId !== null)) agentScanPending = true;
948 }),
949 );
950 });
951
952 const retain: PortDiscovery["Service"]["retain"] = Effect.acquireRelease(
953 acquireRetention(),
954 () => releaseRetention,
955 );
956
957 const subscribe: PortDiscovery["Service"]["subscribe"] = Effect.fn("PortDiscovery.subscribe")(
958 (input, listener) =>
959 Effect.acquireRelease(
960 Ref.update(stateRef, (state) => {
961 const listeners = new Map(state.listeners);
962 listeners.set(listener, {
963 configuredUrls: normalizeConfiguredUrls(input.configuredUrls),
964 lastSnapshot: input.initialSnapshot,
965 });
966 return { ...state, listeners };
967 }),
968 () =>
969 Ref.update(stateRef, (state) => {
970 const listeners = new Map(state.listeners);
971 listeners.delete(listener);
972 return { ...state, listeners };
973 }),
974 ),
975 );
976
977 const registerTerminalProcesses: PortDiscovery["Service"]["registerTerminalProcesses"] =
978 Effect.fn("PortDiscovery.registerTerminalProcesses")(function* (input) {
979 const owner = {
980 threadId: ThreadId.make(input.threadId),
981 terminalId: input.terminalId,
982 };
983 const processIds = new Set(
984 input.processIds.filter((processId) => Number.isInteger(processId) && processId > 0),
985 );
986 yield* Ref.update(stateRef, (state) => {
987 const terminalProcesses = new Map(state.terminalProcesses);
988 const key = terminalOwnerKey(owner);
989 if (processIds.size === 0) {
990 terminalProcesses.delete(key);
991 } else {
992 terminalProcesses.set(key, { owner, processIds });
993 }
994 return { ...state, terminalProcesses };
995 });
996 });
997
998 const unregisterTerminal: PortDiscovery["Service"]["unregisterTerminal"] = Effect.fn(
999 "PortDiscovery.unregisterTerminal",
1000 )(function* (input) {
1001 yield* Ref.update(stateRef, (state) => {
1002 const terminalProcesses = new Map(state.terminalProcesses);
1003 terminalProcesses.delete(terminalOwnerKey(input));
1004 return { ...state, terminalProcesses };
1005 });
1006 });
1007
1008 const trackAgent: PortDiscovery["Service"]["trackAgent"] = () => {
1009 runningAgents += 1;
1010 let released = false;
1011 return () => {
1012 if (released) return;
1013 released = true;
1014 runningAgents -= 1;
1015 agentScanPending = true;
1016 };
1017 };
1018
1019 // Fails when `ps` does, so an empty table never reads as every process having exited.
1020 const readProcessTable = Effect.suspend(() =>
1021 processRunner.run({
1022 command: "ps",
1023 args: ["-A", "-o", "pid=,ppid=,pgid="],
1024 timeout: Duration.millis(PS_TIMEOUT_MS),
1025 maxOutputBytes: PS_MAX_OUTPUT_BYTES,
1026 }),
1027 ).pipe(
1028 Effect.mapError((cause) => new PreviewStopLocalServerError({ reason: "signal-failed", cause })),
1029 Effect.filterOrFail(
1030 (result) => !exitedWithError(result),
1031 () => new PreviewStopLocalServerError({ reason: "signal-failed" }),
1032 ),
1033 Effect.map((result) => parseProcessTable(result.stdout)),
1034 );
1035
1036 const signalUntilStopped = Effect.fn("PortDiscovery.signalUntilStopped")(function* (
1037 input: PreviewStopLocalServerInput,
1038 ) {
1039 if (hostPlatform === "win32") {
1040 return yield* new PreviewStopLocalServerError({ reason: "unsupported" });
1041 }
1042 // A socket can outlive its cached owner. Wait for any in-flight scan before invalidating
1043 // that owner, so the /proc fallback confirms who holds it before a signal is sent.
1044 yield* scanSemaphore.withPermits(1)(
1045 Ref.update(
1046 socketOwnersRef,
1047 (owners) => new Map([...owners].filter(([, owner]) => owner.pid !== input.pid)),
1048 ),
1049 );
1050 // The pid comes from a client, so only a server this scan attributes to a thread is signaled,
1051 // from its environment as read now.
1052 yield* Ref.update(agentThreadsRef, (threads) => {
1053 const next = new Map(threads);
1054 next.delete(input.pid);
1055 return next;
1056 });
1057 agentScanPending = true;
1058 const listed = (yield* scanOnce()).some(
1059 (server) => server.pid === input.pid && server.port === input.port && isOwned(server),
1060 );
1061 if (!listed) return yield* new PreviewStopLocalServerError({ reason: "not-found" });
1062 const table = yield* readProcessTable;
1063 // It exited after the scan, so there is nothing left to stop.
1064 if (!table.has(input.pid)) return;
1065 const group = stoppableProcessGroup(table, input.pid, process.pid);
1066 if (group === null) return yield* new PreviewStopLocalServerError({ reason: "shared-group" });
1067 // False when the whole group already exited.
1068 const signalGroup = (signal: "SIGTERM" | "SIGKILL") =>
1069 Effect.suspend(() => {
1070 try {
1071 signalProcessGroup(group, signal);
1072 return Effect.succeed(true);
1073 } catch (cause) {
1074 return isNoSuchProcess(cause)
1075 ? Effect.succeed(false)
1076 : Effect.fail(new PreviewStopLocalServerError({ reason: "signal-failed", cause }));
1077 }
1078 });
1079 if (!(yield* signalGroup("SIGTERM"))) return;
1080 // While one of the group's original members is still in it, the group has not been reused,
1081 // so SIGKILL goes only to the job that ignored SIGTERM.
1082 const members = [...table.values()].filter((row) => row.pgid === group).map((row) => row.pid);
1083 const jobRunning = readProcessTable.pipe(
1084 Effect.map((current) => members.some((pid) => current.get(pid)?.pgid === group)),
1085 Effect.orElseSucceed(() => false),
1086 );
1087 const deadline = (yield* Clock.currentTimeMillis) + STOP_GRACE_MS;
1088 while ((yield* Clock.currentTimeMillis) < deadline) {
1089 yield* Effect.sleep(STOP_POLL_INTERVAL);
1090 if (!(yield* jobRunning)) return;
1091 }
1092 yield* signalGroup("SIGKILL");
1093 });
1094
1095 const stopServer: PortDiscovery["Service"]["stopServer"] = (input) =>
1096 // Subscribers see the server leave now rather than on the next poll.
1097 signalUntilStopped(input).pipe(Effect.andThen(pollIfRetained));
1098
1099 return PortDiscovery.of({
1100 scan: scanOnce,
1101 subscribe,
1102 retain,
1103 registerTerminalProcesses,
1104 unregisterTerminal,
1105 trackAgent,
1106 stopServer,
1107 });
1108}).pipe(Effect.withSpan("PortDiscovery.make"));
1109
1110export const layer = Layer.effect(PortDiscovery, make);
1111
1112/** A scanner that tracks nothing, for tests that build provider runners without a host. */
1113export const layerTest = Layer.mock(PortDiscovery)({ trackAgent: () => () => undefined });