1/**
2 * In-process PortScanner implementation.
3 *
4 * macOS/Linux: parses `lsof -iTCP -sTCP:LISTEN -P -n -F pcn` (-F output is a
5 * stable line-prefixed field format; this is the only `lsof` flag set we rely
6 * on).
7 *
8 * Linux without lsof: reads listening sockets from `/proc/net/tcp{,6}` and
9 * finds their processes through `/proc/<pid>/fd`.
10 *
11 * Only listeners owned by T3 terminal processes and explicitly configured
12 * URLs receive probes. Without listener ownership, only configured URLs do.
13 *
14 * Listening ports are published only after a bounded HTTP(S) probe finds a
15 * successful HTML document or a redirect to one.
16 * Positive and negative results are cached briefly by candidate URL and listener identity,
17 * limiting repeated requests without leaving stale classifications around.
18 *
19 * Polling is reference-counted via scoped `retain`. A single layer-scoped fiber
20 * polls forever, but each tick is a no-op when the retain count is zero.
21 */
22import {
23 CONFIGURED_LOCAL_SERVER_URLS_MAX_ITEMS,
24 PREVIEW_URL_MAX_LENGTH,
25 ThreadId,
26 type DiscoveredLocalServer,
27} from "@t3tools/contracts";
28import * as HostProcess from "@t3tools/shared/HostProcess";
29import { isLoopbackHost, LSOF_LOCAL_HOST_TOKENS } from "@t3tools/shared/preview";
30import * as Cause from "effect/Cause";
31import * as Clock from "effect/Clock";
32import * as Context from "effect/Context";
33import * as Duration from "effect/Duration";
34import * as Effect from "effect/Effect";
35import * as FileSystem from "effect/FileSystem";
36import * as Layer from "effect/Layer";
37import * as Option from "effect/Option";
38import * as PlatformError from "effect/PlatformError";
39import * as Ref from "effect/Ref";
40import * as Schedule from "effect/Schedule";
41import * as Scope from "effect/Scope";
42import * as Semaphore from "effect/Semaphore";
43import { FetchHttpClient, HttpClient } from "effect/http";
44
45import * as ProcessRunner from "../processRunner.ts";
46
47export class PortDiscovery extends Context.Service<
48 PortDiscovery,
49 {
50 readonly scan: (
51 configuredUrls?: ReadonlyArray<string>,
52 ) => Effect.Effect<ReadonlyArray<DiscoveredLocalServer>>;
53 readonly subscribe: (
54 input: {
55 readonly configuredUrls: ReadonlyArray<string>;
56 readonly initialSnapshot: ReadonlyArray<DiscoveredLocalServer>;
57 },
58 listener: (servers: ReadonlyArray<DiscoveredLocalServer>) => Effect.Effect<void>,
59 ) => Effect.Effect<void, never, Scope.Scope>;
60 readonly retain: Effect.Effect<void, never, Scope.Scope>;
61 readonly registerTerminalProcesses: (input: {
62 readonly threadId: string;
63 readonly terminalId: string;
64 readonly processIds: ReadonlyArray<number>;
65 }) => Effect.Effect<void>;
66 readonly unregisterTerminal: (input: {
67 readonly threadId: string;
68 readonly terminalId: string;
69 }) => Effect.Effect<void>;
70 }
71>()("t3/preview/PortScanner/PortDiscovery") {}
72
73const POLL_INTERVAL = Duration.seconds(3);
74const LSOF_TIMEOUT_MS = 5_000;
75/** File descriptors one `/proc` walk may read before it stops looking for socket owners. */
76const PROC_FD_WALK_LIMIT = 50_000;
77/** Scans that retry the fd walk for a listening socket whose owner was not found. */
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));
82const WEB_PROBE_CONCURRENCY = 16;
83const NAVIGATION_REDIRECT_STATUSES = new Set([301, 302, 303, 307, 308]);
84
85type Listener = (servers: ReadonlyArray<DiscoveredLocalServer>) => Effect.Effect<void>;
86
87interface ListenerSubscription {
88 readonly configuredUrls: ReadonlyArray<string>;
89 readonly lastSnapshot: ReadonlyArray<DiscoveredLocalServer>;
90}
91
92interface ScannerState {
93 readonly listeners: ReadonlyMap<Listener, ListenerSubscription>;
94 readonly terminalProcesses: ReadonlyMap<
95 string,
96 {
97 readonly owner: TerminalProcessOwner;
98 readonly processIds: ReadonlySet<number>;
99 }
100 >;
101 readonly retainCount: number;
102}
103
104interface TerminalProcessOwner {
105 readonly threadId: ThreadId;
106 readonly terminalId: string;
107}
108
109interface WebProbeCacheEntry {
110 readonly pid: number | null;
111 readonly isWeb: boolean;
112 readonly expiresAtMillis: number;
113}
114
115interface WebProbeGroup {
116 readonly server: DiscoveredLocalServer;
117 readonly urls: ReadonlyArray<string>;
118 readonly configuredKey: string | null;
119}
120
121interface WebProbeSnapshot {
122 readonly discovered: ReadonlyArray<DiscoveredLocalServer>;
123 readonly configured: ReadonlyMap<string, DiscoveredLocalServer>;
124}
125
126const terminalOwnerKey = (owner: {
127 readonly threadId: string;
128 readonly terminalId: string;
129}): string => `${owner.threadId}\u0000${owner.terminalId}`;
130
131const parseConfiguredUrl = (raw: string): URL | null => {
132 try {
133 const url = new URL(raw);
134 if (url.protocol !== "http:" && url.protocol !== "https:") return null;
135 if (!isLoopbackHost(url.hostname)) return null;
136 return url;
137 } catch {
138 return null;
139 }
140};
141
142const localServerKey = (host: string, port: number): string =>
143 `${isLoopbackHost(host) ? "loopback" : host.toLowerCase()}:${port}`;
144
145const urlPort = (url: URL): number =>
146 url.port.length > 0 ? Number.parseInt(url.port, 10) : url.protocol === "http:" ? 80 : 443;
147
148const webProbeCacheKey = (raw: string): string => {
149 const url = new URL(raw);
150 url.hash = "";
151 return url.href;
152};
153
154const normalizeConfiguredUrls = (urls: ReadonlyArray<string>): ReadonlyArray<string> => [
155 ...new Set(
156 urls
157 .slice(0, CONFIGURED_LOCAL_SERVER_URLS_MAX_ITEMS)
158 .filter((raw) => raw.length <= PREVIEW_URL_MAX_LENGTH)
159 .map(parseConfiguredUrl)
160 .filter((url): url is URL => url !== null && url.href.length <= PREVIEW_URL_MAX_LENGTH)
161 .map((url) => {
162 if (url.hostname === "0.0.0.0") url.hostname = "localhost";
163 return url.href;
164 })
165 .filter((url) => url.length <= PREVIEW_URL_MAX_LENGTH),
166 ),
167];
168
169const projectWebProbeSnapshot = (
170 snapshot: WebProbeSnapshot,
171 configuredUrls: ReadonlyArray<string>,
172): ReadonlyArray<DiscoveredLocalServer> => {
173 const visibleByServer = new Map<string, DiscoveredLocalServer>();
174 for (const raw of normalizeConfiguredUrls(configuredUrls)) {
175 const url = new URL(raw);
176 const port = urlPort(url);
177 const serverKey = localServerKey(url.hostname, port);
178 if (visibleByServer.has(serverKey)) continue;
179 const configured = snapshot.configured.get(webProbeCacheKey(raw));
180 if (configured) visibleByServer.set(serverKey, { ...configured, url: raw });
181 }
182 for (const server of snapshot.discovered) {
183 const key = localServerKey(server.host, server.port);
184 if (!visibleByServer.has(key)) visibleByServer.set(key, server);
185 }
186 return [...visibleByServer.values()].toSorted((left, right) => left.port - right.port);
187};
188
189const parseLsofOutput = (
190 raw: string,
191 terminalByProcessId: ReadonlyMap<number, TerminalProcessOwner> = new Map(),
192): ReadonlyArray<DiscoveredLocalServer> => {
193 const seen = new Map<string, DiscoveredLocalServer>();
194 let pid: number | null = null;
195 let processName: string | null = null;
196
197 for (const line of raw.split("\n")) {
198 if (line.length === 0) continue;
199 const tag = line.charAt(0);
200 const value = line.slice(1);
201 if (tag === "p") {
202 const parsed = Number.parseInt(value, 10);
203 pid = Number.isFinite(parsed) && parsed > 0 ? parsed : null;
204 processName = null;
205 continue;
206 }
207 if (tag === "c") {
208 processName = value.trim() || null;
209 continue;
210 }
211 if (tag === "n") {
212 const portMatch = parsePortFromLsofName(value);
213 if (portMatch == null) continue;
214 const url = `http://localhost:${portMatch}`;
215 const key = `localhost:${portMatch}`;
216 const terminal = pid === null ? null : (terminalByProcessId.get(pid) ?? null);
217 const existing = seen.get(key);
218 // A port receives probes only when T3 terminals own all of its listeners.
219 if (existing) {
220 if (terminal === null) seen.set(key, { ...existing, terminal: null });
221 continue;
222 }
223 seen.set(key, {
224 host: "localhost",
225 port: portMatch,
226 url,
227 processName,
228 pid,
229 terminal,
230 });
231 }
232 }
233
234 return Array.from(seen.values()).toSorted((a, b) => a.port - b.port);
235};
236
237const parsePortFromLsofName = (name: string): number | null => {
238 // Examples: "*:5173", "127.0.0.1:5173", "[::1]:5173", "localhost:5173",
239 // "192.168.1.10:5173 (LISTEN)" — we only care if the host part is local.
240 const trimmed = name.split(" ", 1)[0]?.trim() ?? "";
241 if (trimmed.length === 0) return null;
242 const lastColon = trimmed.lastIndexOf(":");
243 if (lastColon < 0) return null;
244 const hostPart = trimmed.slice(0, lastColon);
245 const portPart = trimmed.slice(lastColon + 1);
246 if (!LSOF_LOCAL_HOST_TOKENS.has(hostPart)) return null;
247 const port = Number.parseInt(portPart, 10);
248 if (!Number.isFinite(port) || port <= 0 || port >= 65536) return null;
249 return port;
250};
251
252const parseWindowsListenerOutput = (
253 raw: string,
254 terminalByProcessId: ReadonlyMap<number, TerminalProcessOwner> = new Map(),
255): ReadonlyArray<DiscoveredLocalServer> => {
256 const seen = new Map<number, DiscoveredLocalServer>();
257 for (const line of raw.split(/\r?\n/g)) {
258 const [hostRaw, portRaw, pidRaw, processNameRaw] = line.trim().split("|", 4);
259 const host = hostRaw?.trim() ?? "";
260 if (!LSOF_LOCAL_HOST_TOKENS.has(host) && host !== "::") continue;
261 const port = Number(portRaw);
262 const pid = Number(pidRaw);
263 if (!Number.isInteger(port) || port <= 0 || port >= 65536) continue;
264 const normalizedPid = Number.isInteger(pid) && pid > 0 ? pid : null;
265 const terminal =
266 normalizedPid === null ? null : (terminalByProcessId.get(normalizedPid) ?? null);
267 const existing = seen.get(port);
268 if (existing) {
269 if (terminal === null) seen.set(port, { ...existing, terminal: null });
270 continue;
271 }
272 seen.set(port, {
273 host: "localhost",
274 port,
275 url: `http://localhost:${port}`,
276 processName: processNameRaw?.trim() || null,
277 pid: normalizedPid,
278 terminal,
279 });
280 }
281 return [...seen.values()].toSorted((left, right) => left.port - right.port);
282};
283
284/** `/proc/net/tcp` state for a listening socket. */
285const PROC_TCP_LISTEN = "0A";
286
287/**
288 * Whether a `/proc/net/tcp{,6}` address is one the preview can reach on
289 * loopback: the wildcard or a loopback address, as `lsof` parsing accepts.
290 * Each 32-bit word of the hex address is little-endian.
291 */
292const procLocalAddress = (hex: string): boolean => {
293 if (hex.length === 8) {
294 return hex === "00000000" || hex.endsWith("7F");
295 }
296 if (hex.length !== 32) return false;
297 if (hex === "0".repeat(32) || hex === `${"0".repeat(24)}01000000`) return true;
298 // IPv4-mapped (::ffff:a.b.c.d): wildcard or 127.0.0.0/8.
299 return hex.startsWith(`${"0".repeat(16)}FFFF0000`) && procLocalAddress(hex.slice(24));
300};
301
302/** Listening loopback ports in `/proc/net/tcp{,6}` content, with their socket inodes. */
303const parseProcNetTcp = (
304 raw: string,
305): ReadonlyArray<{ readonly port: number; readonly inode: string }> => {
306 const listeners: Array<{ port: number; inode: string }> = [];
307 for (const line of raw.split("\n").slice(1)) {
308 const fields = line.trim().split(/\s+/);
309 const [address, portHex] = fields[1]?.split(":") ?? [];
310 if (fields[3] !== PROC_TCP_LISTEN || !address || !portHex) continue;
311 if (!procLocalAddress(address.toUpperCase())) continue;
312 const port = Number.parseInt(portHex, 16);
313 const inode = fields[9];
314 if (!Number.isInteger(port) || port <= 0 || port >= 65536 || !inode) continue;
315 listeners.push({ port, inode });
316 }
317 return listeners;
318};
319
320const serversEqual = (
321 left: ReadonlyArray<DiscoveredLocalServer>,
322 right: ReadonlyArray<DiscoveredLocalServer>,
323): boolean => {
324 if (left.length !== right.length) return false;
325 for (let i = 0; i < left.length; i += 1) {
326 const a = left[i];
327 const b = right[i];
328 if (!a || !b) return false;
329 if (
330 a.host !== b.host ||
331 a.port !== b.port ||
332 a.url !== b.url ||
333 a.processName !== b.processName ||
334 a.pid !== b.pid ||
335 a.terminal?.threadId !== b.terminal?.threadId ||
336 a.terminal?.terminalId !== b.terminal?.terminalId
337 ) {
338 return false;
339 }
340 }
341 return true;
342};
343
344const isCommandNotFound = (error: ProcessRunner.ProcessSpawnError): boolean =>
345 PlatformError.isPlatformError(error.cause) && error.cause.reason._tag === "NotFound";
346
347/** @public Service construction is part of the canonical Effect module API. */
348export const make = Effect.gen(function* PortDiscoveryMake() {
349 const processRunner = yield* ProcessRunner.ProcessRunner;
350 const fileSystem = yield* FileSystem.FileSystem;
351 const hostPlatform = yield* HostProcess.Platform;
352 const httpClient = (yield* HttpClient.HttpClient).pipe(HttpClient.withScope);
353 const stateRef = yield* Ref.make<ScannerState>({
354 listeners: new Map(),
355 terminalProcesses: new Map(),
356 retainCount: 0,
357 });
358 const webProbeCacheRef = yield* Ref.make<ReadonlyMap<string, WebProbeCacheEntry>>(new Map());
359 const scanSemaphore = yield* Semaphore.make(1);
360 const lsofMissingRef = yield* Ref.make(false);
361 /** Confirmed owners of listening socket inodes. */
362 const socketOwnersRef = yield* Ref.make<
363 ReadonlyMap<string, { readonly pid: number; readonly processName: string | null }>
364 >(new Map());
365 /** Fd walks that found no owner, per socket inode; retried up to `SOCKET_OWNER_ATTEMPTS`. */
366 const socketOwnerMissesRef = yield* Ref.make<ReadonlyMap<string, number>>(new Map());
367
368 /** Maps socket inodes to the processes holding them, by `/proc/<pid>/fd`. */
369 const findSocketOwners = Effect.fn("PortDiscovery.findSocketOwners")(function* (
370 inodes: ReadonlySet<string>,
371 ) {
372 const owners = new Map<string, { pid: number; processName: string | null }>();
373 const pids = (yield* fileSystem
374 .readDirectory("/proc")
375 .pipe(Effect.orElseSucceed((): ReadonlyArray<string> => []))).filter((entry) =>
376 /^\d+$/.test(entry),
377 );
378 let budget = PROC_FD_WALK_LIMIT;
379 for (const pidText of pids) {
380 if (owners.size === inodes.size || budget <= 0) break;
381 // Other users' fds are unreadable; their ports still list, without a pid.
382 const fds = yield* fileSystem
383 .readDirectory(`/proc/${pidText}/fd`)
384 .pipe(Effect.orElseSucceed((): ReadonlyArray<string> => []));
385 const walked = fds.slice(0, budget);
386 budget -= walked.length;
387 const links = yield* Effect.forEach(
388 walked,
389 (fd) =>
390 fileSystem.readLink(`/proc/${pidText}/fd/${fd}`).pipe(Effect.orElseSucceed(() => "")),
391 { concurrency: 16 },
392 );
393 const held = links.flatMap((link) => /^socket:\[(\d+)\]$/.exec(link)?.[1] ?? []);
394 const matched = held.filter((inode) => inodes.has(inode) && !owners.has(inode));
395 if (matched.length === 0) continue;
396 const processName = yield* fileSystem.readFileString(`/proc/${pidText}/comm`).pipe(
397 Effect.map((name) => name.trim() || null),
398 Effect.orElseSucceed(() => null),
399 );
400 for (const inode of matched) owners.set(inode, { pid: Number(pidText), processName });
401 }
402 return owners;
403 });
404
405 /**
406 * Linux listeners from `/proc/net/tcp{,6}`, for hosts without `lsof`. Null
407 * when neither file is readable, so only configured URLs are probed.
408 */
409 const scanProcListeners = Effect.fn("PortDiscovery.scanProcListeners")(function* (
410 terminalByProcessId: ReadonlyMap<number, TerminalProcessOwner>,
411 ) {
412 if (hostPlatform !== "linux") return null;
413 const tables = yield* Effect.forEach(["/proc/net/tcp", "/proc/net/tcp6"], (path) =>
414 fileSystem.readFileString(path).pipe(Effect.option),
415 );
416 if (tables.every(Option.isNone)) return null;
417 const listeners = tables.flatMap((table) =>
418 Option.isSome(table) ? parseProcNetTcp(table.value) : [],
419 );
420 const known = yield* Ref.get(socketOwnersRef);
421 const knownMisses = yield* Ref.get(socketOwnerMissesRef);
422 const inodes = new Set(listeners.map((listener) => listener.inode));
423 // Listeners rarely change, so the fd walk runs only for sockets without a
424 // confirmed owner, and gives up on one after a few walks miss it (another
425 // user's process, or one that exited mid-walk).
426 const unseen = new Set(
427 [...inodes].filter(
428 (inode) => !known.has(inode) && (knownMisses.get(inode) ?? 0) < SOCKET_OWNER_ATTEMPTS,
429 ),
430 );
431 const found = unseen.size === 0 ? new Map() : yield* findSocketOwners(unseen);
432 const owners = new Map([...known, ...found].filter(([inode]) => inodes.has(inode)));
433 const misses = new Map([...knownMisses].filter(([inode]) => inodes.has(inode)));
434 for (const inode of unseen) {
435 if (!found.has(inode)) misses.set(inode, (misses.get(inode) ?? 0) + 1);
436 }
437 yield* Ref.set(socketOwnersRef, owners);
438 yield* Ref.set(socketOwnerMissesRef, misses);
439 const seen = new Map<number, DiscoveredLocalServer>();
440 for (const { port, inode } of listeners) {
441 const owner = owners.get(inode) ?? null;
442 const terminal = owner === null ? null : (terminalByProcessId.get(owner.pid) ?? null);
443 const existing = seen.get(port);
444 if (existing && (existing.pid !== null || owner === null)) {
445 if (terminal === null) seen.set(port, { ...existing, terminal: null });
446 continue;
447 }
448 seen.set(port, {
449 host: "localhost",
450 port,
451 url: `http://localhost:${port}`,
452 processName: owner?.processName ?? null,
453 pid: owner?.pid ?? null,
454 // A replaced listener had no known owner, so the port stays unowned.
455 terminal: existing ? null : terminal,
456 });
457 }
458 return [...seen.values()].toSorted((left, right) => left.port - right.port);
459 });
460
461 const probeWebUrl = Effect.fn("PortDiscovery.probeWebUrl")((url: string) =>
462 httpClient.get(url).pipe(
463 Effect.map((response) => {
464 const location = response.headers.location?.trim();
465 if (NAVIGATION_REDIRECT_STATUSES.has(response.status) && location) return url;
466 if (response.status < 200 || response.status >= 300) return null;
467 if (response.status === 204 || response.status === 205) return null;
468 const contentType = response.headers["content-type"]
469 ?.split(";", 1)[0]
470 ?.trim()
471 .toLowerCase();
472 return contentType === "text/html" || contentType === "application/xhtml+xml" ? url : null;
473 }),
474 Effect.scoped,
475 Effect.timeoutOption(WEB_PROBE_TIMEOUT),
476 Effect.map(Option.getOrNull),
477 Effect.orElseSucceed(() => null),
478 Effect.provideService(FetchHttpClient.RequestInit, { redirect: "manual" }),
479 ),
480 );
481
482 const makeWebProbeGroups = (
483 servers: ReadonlyArray<DiscoveredLocalServer>,
484 configuredUrls: ReadonlyArray<string>,
485 ): ReadonlyArray<WebProbeGroup> => {
486 const serversByKey = new Map(
487 servers.map((server) => [localServerKey(server.host, server.port), server] as const),
488 );
489 const groups: WebProbeGroup[] = [];
490 const configuredResources = new Set<string>();
491
492 for (const raw of configuredUrls) {
493 const url = new URL(raw);
494 const port = urlPort(url);
495 const key = localServerKey(url.hostname, port);
496 const resourceKey = webProbeCacheKey(raw);
497 if (configuredResources.has(resourceKey)) continue;
498 configuredResources.add(resourceKey);
499 groups.push({
500 server: serversByKey.get(key) ?? {
501 host: url.hostname,
502 port,
503 url: raw,
504 processName: null,
505 pid: null,
506 terminal: null,
507 },
508 urls: [raw],
509 configuredKey: resourceKey,
510 });
511 }
512
513 for (const server of servers) {
514 // A listener alone does not identify its protocol. Sending HTTP or TLS to
515 // another app's binary RPC listener can crash it before classification.
516 if (server.terminal === null) continue;
517 groups.push({
518 server,
519 urls: [`http://${server.host}:${server.port}`, `https://${server.host}:${server.port}`],
520 configuredKey: null,
521 });
522 }
523
524 return groups;
525 };
526
527 const probeWebServers = Effect.fn("PortDiscovery.probeWebServers")(function* (
528 servers: ReadonlyArray<DiscoveredLocalServer>,
529 configuredUrls: ReadonlyArray<string>,
530 ) {
531 const nowMillis = yield* Clock.currentTimeMillis;
532 const cached = yield* Ref.get(webProbeCacheRef);
533 const groups = makeWebProbeGroups(servers, configuredUrls);
534 const batchProbes = new Map<
535 string,
536 Effect.Effect<{ readonly probe: WebProbeCacheEntry; readonly fresh: boolean }>
537 >();
538 const batchProbeSemaphore = yield* Semaphore.make(1);
539 const getProbe = (url: string, pid: number | null) => {
540 const key = webProbeCacheKey(url);
541 const identity = `${key}\u0000${pid ?? ""}`;
542 return batchProbeSemaphore
543 .withPermits(1)(
544 Effect.gen(function* () {
545 const existing = batchProbes.get(identity);
546 if (existing) return [existing] as const;
547 const cachedProbe = cached.get(key);
548 const cachedIsCurrent =
549 cachedProbe?.pid === pid && cachedProbe.expiresAtMillis > nowMillis;
550 const memoized = yield* Effect.cached(
551 cachedIsCurrent
552 ? Effect.succeed({ probe: cachedProbe, fresh: false })
553 : probeWebUrl(url).pipe(
554 Effect.map((result) => ({
555 probe: { pid, isWeb: result !== null, expiresAtMillis: 0 },
556 fresh: true,
557 })),
558 ),
559 );
560 batchProbes.set(identity, memoized);
561 return [memoized] as const;
562 }),
563 )
564 .pipe(Effect.flatMap(([probe]) => probe));
565 };
566 const probed = yield* Effect.forEach(
567 groups,
568 (group) =>
569 Effect.gen(function* () {
570 const probes: Array<readonly [string, WebProbeCacheEntry, boolean]> = [];
571 let visibleUrl: string | null = null;
572 for (const url of group.urls) {
573 const key = webProbeCacheKey(url);
574 const { probe, fresh } = yield* getProbe(url, group.server.pid);
575 probes.push([key, probe, fresh]);
576 if (probe.isWeb) {
577 visibleUrl = url;
578 break;
579 }
580 }
581 return { group, probes, visibleUrl };
582 }),
583 { concurrency: WEB_PROBE_CONCURRENCY },
584 );
585 const completedAtMillis = yield* Clock.currentTimeMillis;
586 const nextCache = new Map(
587 [...cached].filter(([, probe]) => probe.expiresAtMillis > completedAtMillis),
588 );
589 const discovered: DiscoveredLocalServer[] = [];
590 const configured = new Map<string, DiscoveredLocalServer>();
591 for (const { group, probes, visibleUrl } of probed) {
592 for (const [key, probe, fresh] of probes) {
593 nextCache.set(
594 key,
595 fresh ? { ...probe, expiresAtMillis: completedAtMillis + WEB_PROBE_CACHE_TTL_MS } : probe,
596 );
597 }
598 if (visibleUrl === null) continue;
599 const server = { ...group.server, url: visibleUrl };
600 if (group.configuredKey === null) discovered.push(server);
601 else configured.set(group.configuredKey, server);
602 }
603 yield* Ref.set(webProbeCacheRef, nextCache);
604 return { discovered, configured } satisfies WebProbeSnapshot;
605 });
606
607 const recoverProcessProbeFailure =
608 (probe: "lsof" | "windows-listeners") => (error: ProcessRunner.ProcessRunError) =>
609 Effect.logDebug("preview port process probe failed; probing configured URLs only", {
610 cause: error,
611 probe,
612 platform: hostPlatform,
613 }).pipe(Effect.as(null));
614
615 const scanUnlocked = Effect.fn("PortDiscovery.scanUnlocked")(function* (
616 configuredUrls: ReadonlyArray<string>,
617 ) {
618 const state = yield* Ref.get(stateRef);
619 const terminalByProcessId = new Map<number, TerminalProcessOwner>();
620 for (const registration of state.terminalProcesses.values()) {
621 for (const processId of registration.processIds) {
622 terminalByProcessId.set(processId, registration.owner);
623 }
624 }
625 if (hostPlatform === "win32") {
626 const recoverWindowsProbeFailure = recoverProcessProbeFailure("windows-listeners");
627 const command =
628 'Get-NetTCPConnection -State Listen -ErrorAction Stop | ForEach-Object { $processName = (Get-Process -Id $_.OwningProcess -ErrorAction SilentlyContinue).ProcessName; Write-Output "$($_.LocalAddress)|$($_.LocalPort)|$($_.OwningProcess)|$processName" }';
629 const listeners = yield* processRunner
630 .run({
631 command: "powershell.exe",
632 args: ["-NoProfile", "-NonInteractive", "-Command", command],
633 timeout: Duration.millis(WINDOWS_LISTENER_TIMEOUT_MS),
634 maxOutputBytes: 1024 * 1024,
635 })
636 .pipe(
637 Effect.map((result) => parseWindowsListenerOutput(result.stdout, terminalByProcessId)),
638 Effect.catchTags({
639 ProcessSpawnError: recoverWindowsProbeFailure,
640 ProcessStdinError: recoverWindowsProbeFailure,
641 ProcessOutputLimitError: recoverWindowsProbeFailure,
642 ProcessReadError: recoverWindowsProbeFailure,
643 ProcessTimeoutError: recoverWindowsProbeFailure,
644 }),
645 );
646 if (listeners !== null) return yield* probeWebServers(listeners, configuredUrls);
647 return yield* probeWebServers([], configuredUrls);
648 }
649 const recoverLsofProbeFailure = recoverProcessProbeFailure("lsof");
650 if (yield* Ref.get(lsofMissingRef)) {
651 const fromProc = yield* scanProcListeners(terminalByProcessId);
652 return yield* probeWebServers(fromProc ?? [], configuredUrls);
653 }
654 const lsofResult = yield* processRunner
655 .run({
656 command: "lsof",
657 args: ["-iTCP", "-sTCP:LISTEN", "-P", "-n", "-F", "pcn"],
658 timeout: Duration.millis(LSOF_TIMEOUT_MS),
659 // Partial output could hide another app's listener on an owned port.
660 maxOutputBytes: 1024 * 1024,
661 })
662 .pipe(
663 Effect.map((result) => parseLsofOutput(result.stdout, terminalByProcessId)),
664 Effect.catchTags({
665 // A missing lsof stays missing; later scans skip straight to the fallback.
666 ProcessSpawnError: (error) =>
667 (isCommandNotFound(error) ? Ref.set(lsofMissingRef, true) : Effect.void).pipe(
668 Effect.andThen(recoverLsofProbeFailure(error)),
669 ),
670 ProcessStdinError: recoverLsofProbeFailure,
671 ProcessOutputLimitError: recoverLsofProbeFailure,
672 ProcessReadError: recoverLsofProbeFailure,
673 ProcessTimeoutError: recoverLsofProbeFailure,
674 }),
675 );
676 if (lsofResult !== null) return yield* probeWebServers(lsofResult, configuredUrls);
677 const fromProc = yield* scanProcListeners(terminalByProcessId);
678 return yield* probeWebServers(fromProc ?? [], configuredUrls);
679 });
680
681 const scanSnapshot = Effect.fn("PortDiscovery.scanSnapshot")(
682 (configuredUrls: ReadonlyArray<string>) =>
683 scanSemaphore.withPermits(1)(scanUnlocked(configuredUrls)),
684 );
685
686 const scanOnce: PortDiscovery["Service"]["scan"] = (configuredUrls = []) => {
687 const normalized = normalizeConfiguredUrls(configuredUrls);
688 return scanSnapshot(normalized).pipe(
689 Effect.map((snapshot) => projectWebProbeSnapshot(snapshot, normalized)),
690 );
691 };
692
693 const pollTick = Effect.fn("PortDiscovery.pollTick")(
694 function* () {
695 const configuredUrls = [
696 ...new Set(
697 [...(yield* Ref.get(stateRef)).listeners.values()].flatMap(
698 (subscription) => subscription.configuredUrls,
699 ),
700 ),
701 ];
702 const snapshot = yield* scanSnapshot(configuredUrls);
703 const notifications = yield* Ref.modify(stateRef, (state) => {
704 const listeners = new Map(state.listeners);
705 const changed: Array<readonly [Listener, ReadonlyArray<DiscoveredLocalServer>]> = [];
706 for (const [listener, subscription] of listeners) {
707 const next = projectWebProbeSnapshot(snapshot, subscription.configuredUrls);
708 if (serversEqual(subscription.lastSnapshot, next)) continue;
709 listeners.set(listener, { ...subscription, lastSnapshot: next });
710 changed.push([listener, next]);
711 }
712 return [changed, { ...state, listeners }];
713 });
714 yield* Effect.forEach(notifications, ([listener, servers]) => listener(servers), {
715 discard: true,
716 });
717 },
718 Effect.catchCause((cause: Cause.Cause<never>) =>
719 Effect.logWarning("preview port scan failed", Cause.pretty(cause)),
720 ),
721 );
722
723 // Single layer-scoped polling fiber. Ticks skip the scan and its span when no
724 // client is currently retained, so the cost is one Ref.get every POLL_INTERVAL.
725 const pollIfRetained = Ref.get(stateRef).pipe(
726 Effect.flatMap((state) => (state.retainCount > 0 ? pollTick() : Effect.void)),
727 );
728 yield* Effect.forkScoped(pollIfRetained.pipe(Effect.repeat(Schedule.spaced(POLL_INTERVAL))));
729
730 const acquireRetention = Effect.fn("PortDiscovery.retain")(function* () {
731 const wasIdle = yield* Ref.modify(stateRef, (state) => [
732 state.retainCount === 0,
733 { ...state, retainCount: state.retainCount + 1 },
734 ]);
735 if (wasIdle) {
736 // Run an immediate scan + broadcast so the new retainer doesn't have
737 // to wait up to POLL_INTERVAL for the first emission.
738 yield* pollTick();
739 }
740 });
741
742 const retain: PortDiscovery["Service"]["retain"] = Effect.acquireRelease(acquireRetention(), () =>
743 Ref.update(stateRef, (state) => ({
744 ...state,
745 retainCount: Math.max(0, state.retainCount - 1),
746 })),
747 );
748
749 const subscribe: PortDiscovery["Service"]["subscribe"] = Effect.fn("PortDiscovery.subscribe")(
750 (input, listener) =>
751 Effect.acquireRelease(
752 Ref.update(stateRef, (state) => {
753 const listeners = new Map(state.listeners);
754 listeners.set(listener, {
755 configuredUrls: normalizeConfiguredUrls(input.configuredUrls),
756 lastSnapshot: input.initialSnapshot,
757 });
758 return { ...state, listeners };
759 }),
760 () =>
761 Ref.update(stateRef, (state) => {
762 const listeners = new Map(state.listeners);
763 listeners.delete(listener);
764 return { ...state, listeners };
765 }),
766 ),
767 );
768
769 const registerTerminalProcesses: PortDiscovery["Service"]["registerTerminalProcesses"] =
770 Effect.fn("PortDiscovery.registerTerminalProcesses")(function* (input) {
771 const owner = {
772 threadId: ThreadId.make(input.threadId),
773 terminalId: input.terminalId,
774 };
775 const processIds = new Set(
776 input.processIds.filter((processId) => Number.isInteger(processId) && processId > 0),
777 );
778 yield* Ref.update(stateRef, (state) => {
779 const terminalProcesses = new Map(state.terminalProcesses);
780 const key = terminalOwnerKey(owner);
781 if (processIds.size === 0) {
782 terminalProcesses.delete(key);
783 } else {
784 terminalProcesses.set(key, { owner, processIds });
785 }
786 return { ...state, terminalProcesses };
787 });
788 });
789
790 const unregisterTerminal: PortDiscovery["Service"]["unregisterTerminal"] = Effect.fn(
791 "PortDiscovery.unregisterTerminal",
792 )(function* (input) {
793 yield* Ref.update(stateRef, (state) => {
794 const terminalProcesses = new Map(state.terminalProcesses);
795 terminalProcesses.delete(terminalOwnerKey(input));
796 return { ...state, terminalProcesses };
797 });
798 });
799
800 return PortDiscovery.of({
801 scan: scanOnce,
802 subscribe,
803 retain,
804 registerTerminalProcesses,
805 unregisterTerminal,
806 });
807}).pipe(Effect.withSpan("PortDiscovery.make"));
808
809export const layer = Layer.effect(PortDiscovery, make);