Skip to content

Commit 7c0a3ce

Browse files
juliusmarmingeclaude
juliusmarminge
and
claude
committed
fix(server): never probe a port an agent server shares with an unrelated listener
Listeners were collapsed by port before agent ancestry ran, so only the first listener's pid was attributed. Parsers now return every listener, each is attributed, and a port stays owned only while all of its listeners are. Co-Authored-By: Claude Opus 5.5 <[email protected]>
1 parent a327d35 commit 7c0a3ce

2 files changed

Lines changed: 132 additions & 41 deletions

File tree

‎apps/server/src/preview/PortScanner.test.ts‎

Lines changed: 84 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -534,3 +534,87 @@ effectIt.effect("attributes an agent-started background server to its thread", (
534534
expect(yield* owners()).toEqual([]);
535535
}).pipe(Effect.provide(layer));
536536
});
537+
538+
effectIt.effect("never probes a port an agent server shares with an unrelated listener", () => {
539+
const providerPid = 1000;
540+
const agent = `p${providerPid}\ncpython3\nn[::1]:${LSOF_TEST_PORT}\n`;
541+
const unrelated = `p5678\ncrpc\nn127.0.0.1:${LSOF_TEST_PORT}\n`;
542+
let stdout = `${agent}${unrelated}`;
543+
const requests: string[] = [];
544+
const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => {
545+
requests.push(String(input));
546+
return Promise.resolve(new Response("app", { headers: { "content-type": "text/html" } }));
547+
}) as typeof globalThis.fetch;
548+
const layer = layerLsofScanner({
549+
pid: () => providerPid,
550+
fetch: fetchFn,
551+
stdout: (command) => (command === "ps" ? ` ${providerPid} 1\n 5678 1\n` : stdout),
552+
});
553+
554+
return Effect.gen(function* () {
555+
const scanner = yield* PortScanner.PortDiscovery;
556+
yield* scanner.registerThreadProcess({ threadId: "agent-thread", processId: providerPid });
557+
expect(yield* scanner.scan()).toEqual([]);
558+
stdout = `${unrelated}${agent}`;
559+
expect(yield* scanner.scan()).toEqual([]);
560+
expect(requests).toEqual([]);
561+
562+
stdout = agent;
563+
expect((yield* scanner.scan()).map(({ threadId }) => threadId)).toEqual(["agent-thread"]);
564+
expect(requests).toEqual([`http://localhost:${LSOF_TEST_PORT}/`]);
565+
}).pipe(Effect.provide(layer));
566+
});
567+
568+
effectIt.effect("never probes a /proc listener port an agent shares with another process", () => {
569+
const providerPid = 4242;
570+
const header =
571+
" sl local_address rem_address st tx_queue rx_queue tr tm->when retrnsmt uid timeout inode\n";
572+
const ipv4 = (inode: number) =>
573+
` 0: 0100007F:1F90 00000000:0000 0A 00000000:00000000 00:00000000 00000000 1000 0 ${inode} 1 0000000000000000 100 0 0 10 0\n`;
574+
const ipv6 = (inode: number) =>
575+
` 0: 00000000000000000000000001000000:1F90 00000000000000000000000000000000:0000 0A 00000000:00000000 00:00000000 00000000 1000 0 ${inode} 1 0000000000000000 100 0 0 10 0\n`;
576+
// /proc/net/tcp is read before tcp6, so each phase puts the agent's socket first, then second.
577+
let tables = { tcp: header + ipv4(6001), tcp6: header + ipv6(6002) };
578+
const procFds: Record<string, Record<string, string>> = {
579+
[providerPid]: { "3": "socket:[6001]", "5": "socket:[6004]", "7": "socket:[6005]" },
580+
"5678": { "4": "socket:[6002]", "6": "socket:[6003]" },
581+
};
582+
const missing = FileSystem.makeNoop({});
583+
const fileSystem = FileSystem.layerNoop({
584+
readFileString: (path) =>
585+
path === "/proc/net/tcp"
586+
? Effect.succeed(tables.tcp)
587+
: path === "/proc/net/tcp6"
588+
? Effect.succeed(tables.tcp6)
589+
: missing.readFileString(path),
590+
readDirectory: (path) => {
591+
if (path === "/proc") return Effect.succeed(Object.keys(procFds));
592+
const fds = procFds[/^\/proc\/(\d+)\/fd$/.exec(path)?.[1] ?? ""];
593+
return fds === undefined ? missing.readDirectory(path) : Effect.succeed(Object.keys(fds));
594+
},
595+
readLink: (path) => {
596+
const [, pid, fd] = /^\/proc\/(\d+)\/fd\/(\d+)$/.exec(path) ?? [];
597+
const target = pid && fd ? procFds[pid]?.[fd] : undefined;
598+
return target === undefined ? missing.readLink(path) : Effect.succeed(target);
599+
},
600+
});
601+
const requests: string[] = [];
602+
const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => {
603+
requests.push(String(input));
604+
return Promise.resolve(new Response("app", { headers: { "content-type": "text/html" } }));
605+
}) as typeof globalThis.fetch;
606+
const layer = layerProbeFailure(processProbeFailure, fetchFn, fileSystem);
607+
608+
return Effect.gen(function* () {
609+
const scanner = yield* PortScanner.PortDiscovery;
610+
yield* scanner.registerThreadProcess({ threadId: "agent-thread", processId: providerPid });
611+
expect(yield* scanner.scan()).toEqual([]);
612+
tables = { tcp: header + ipv4(6003), tcp6: header + ipv6(6004) };
613+
expect(yield* scanner.scan()).toEqual([]);
614+
expect(requests).toEqual([]);
615+
616+
tables = { tcp: header + ipv4(6005), tcp6: header };
617+
expect((yield* scanner.scan()).map(({ threadId }) => threadId)).toEqual(["agent-thread"]);
618+
expect(requests).toEqual(["http://localhost:8080/"]);
619+
}).pipe(Effect.provide(layer));
620+
});

‎apps/server/src/preview/PortScanner.ts‎

Lines changed: 48 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -178,7 +178,7 @@ const parseLsofOutput = (
178178
raw: string,
179179
terminalByProcessId: ReadonlyMap<number, TerminalProcessOwner> = new Map(),
180180
): ReadonlyArray<DiscoveredLocalServer> => {
181-
const seen = new Map<string, DiscoveredLocalServer>();
181+
const listeners: DiscoveredLocalServer[] = [];
182182
let pid: number | null = null;
183183
let processName: string | null = null;
184184

@@ -199,26 +199,18 @@ const parseLsofOutput = (
199199
if (tag === "n") {
200200
const portMatch = parsePortFromLsofName(value);
201201
if (portMatch == null) continue;
202-
const url = `http://localhost:${portMatch}`;
203-
const key = `localhost:${portMatch}`;
204-
const terminal = pid === null ? null : (terminalByProcessId.get(pid) ?? null);
205-
const existing = seen.get(key);
206-
if (existing) {
207-
if (terminal === null) seen.set(key, { ...existing, terminal: null });
208-
continue;
209-
}
210-
seen.set(key, {
202+
listeners.push({
211203
host: "localhost",
212204
port: portMatch,
213-
url,
205+
url: `http://localhost:${portMatch}`,
214206
processName,
215207
pid,
216-
terminal,
208+
terminal: pid === null ? null : (terminalByProcessId.get(pid) ?? null),
217209
});
218210
}
219211
}
220212

221-
return Array.from(seen.values()).toSorted((a, b) => a.port - b.port);
213+
return listeners;
222214
};
223215

224216
const parsePortFromLsofName = (name: string): number | null => {
@@ -296,28 +288,51 @@ const parseWindowsListenerOutput = (
296288
raw: string,
297289
terminalByProcessId: ReadonlyMap<number, TerminalProcessOwner> = new Map(),
298290
): ReadonlyArray<DiscoveredLocalServer> => {
299-
const seen = new Map<number, DiscoveredLocalServer>();
291+
const listeners: DiscoveredLocalServer[] = [];
300292
for (const line of raw.split(/\r?\n/g)) {
301293
const parsed = parseWindowsListenerLine(line);
302294
if (!parsed) continue;
303-
const { port, pid: normalizedPid, processName } = parsed;
304-
const terminal =
305-
normalizedPid === null ? null : (terminalByProcessId.get(normalizedPid) ?? null);
306-
const existing = seen.get(port);
307-
if (existing) {
308-
if (terminal === null) seen.set(port, { ...existing, terminal: null });
309-
continue;
310-
}
311-
seen.set(port, {
295+
const { port, pid, processName } = parsed;
296+
listeners.push({
312297
host: "localhost",
313298
port,
314299
url: `http://localhost:${port}`,
315300
processName,
316-
pid: normalizedPid,
317-
terminal,
301+
pid,
302+
terminal: pid === null ? null : (terminalByProcessId.get(pid) ?? null),
318303
});
319304
}
320-
return [...seen.values()].toSorted((left, right) => left.port - right.port);
305+
return listeners;
306+
};
307+
308+
// A port can have several listeners (IPv4 and IPv6, SO_REUSEPORT). It is owned only
309+
// while every listener on it is, so an unrelated process sharing it is never probed.
310+
const collapseListenersByPort = (
311+
listeners: ReadonlyArray<DiscoveredLocalServer>,
312+
): ReadonlyArray<DiscoveredLocalServer> => {
313+
const byPort = new Map<number, DiscoveredLocalServer>();
314+
for (const listener of listeners) {
315+
const existing = byPort.get(listener.port);
316+
if (existing === undefined) {
317+
byPort.set(listener.port, listener);
318+
continue;
319+
}
320+
const kept = existing.pid === null && listener.pid !== null ? listener : existing;
321+
byPort.set(
322+
listener.port,
323+
existing.threadId === undefined || listener.threadId === undefined
324+
? {
325+
host: kept.host,
326+
port: kept.port,
327+
url: kept.url,
328+
processName: kept.processName,
329+
pid: kept.pid,
330+
terminal: null,
331+
}
332+
: kept,
333+
);
334+
}
335+
return [...byPort.values()].toSorted((left, right) => left.port - right.port);
321336
};
322337

323338
const PROC_TCP_LISTEN = "0A";
@@ -474,25 +489,17 @@ export const make = Effect.gen(function* PortDiscoveryMake() {
474489
}
475490
yield* Ref.set(socketOwnersRef, owners);
476491
yield* Ref.set(socketOwnerMissesRef, misses);
477-
const seen = new Map<number, DiscoveredLocalServer>();
478-
for (const { port, inode } of listeners) {
492+
return listeners.map(({ port, inode }): DiscoveredLocalServer => {
479493
const owner = owners.get(inode) ?? null;
480-
const terminal = owner === null ? null : (terminalByProcessId.get(owner.pid) ?? null);
481-
const existing = seen.get(port);
482-
if (existing && (existing.pid !== null || owner === null)) {
483-
if (terminal === null) seen.set(port, { ...existing, terminal: null });
484-
continue;
485-
}
486-
seen.set(port, {
494+
return {
487495
host: "localhost",
488496
port,
489497
url: `http://localhost:${port}`,
490498
processName: owner?.processName ?? null,
491499
pid: owner?.pid ?? null,
492-
terminal: existing ? null : terminal,
493-
});
494-
}
495-
return [...seen.values()].toSorted((left, right) => left.port - right.port);
500+
terminal: owner === null ? null : (terminalByProcessId.get(owner.pid) ?? null),
501+
};
502+
});
496503
});
497504

498505
// An agent's background server runs under the provider CLI, not a terminal, so its
@@ -755,8 +762,8 @@ export const make = Effect.gen(function* PortDiscoveryMake() {
755762
terminalByProcessId.set(processId, registration.owner);
756763
}
757764
}
758-
const owned = (servers: ReadonlyArray<DiscoveredLocalServer>) =>
759-
withThreadOwners(servers, state.threadProcesses);
765+
const owned = (listeners: ReadonlyArray<DiscoveredLocalServer>) =>
766+
withThreadOwners(listeners, state.threadProcesses).pipe(Effect.map(collapseListenersByPort));
760767
if (
761768
terminalByProcessId.size === 0 &&
762769
state.threadProcesses.size === 0 &&

0 commit comments

Comments
 (0)