Commit ea646c0834a3394ecb0be4a30c5d367e5a9002bd | the-knife

apps/server/src/diagnostics/ProcessDiagnostics.test.ts+1-1

...	...	@@ -19,7 +19,7 @@ function makeNativeSnapshot(
19	19	processes: ResourceMonitorSnapshotEvent["processes"],
20	20	): ResourceMonitorSnapshotEvent {
21	21	return {
22		version: 2,
	22	version: 3,
23	23	type: "snapshot",
24	24	sequence: 1,
25	25	sampledAtUnixMs: DateTime.toEpochMillis(DateTime.makeUnsafe("2026-05-05T10:00:00.000Z")),

apps/server/src/resourceTelemetry/Model.test.ts+1-1

...	...	@@ -39,7 +39,7 @@ function nativeSnapshot(
39	39	sequence = 1,
40	40	): ResourceMonitorSnapshotEvent {
41	41	return {
42		version: 2,
	42	version: 3,
43	43	type: "snapshot",
44	44	sequence,
45	45	sampledAtUnixMs,

apps/server/src/resourceTelemetry/NativeTelemetryClient.ts+92-1

...	...	@@ -5,6 +5,7 @@ import type {
5	5	ResourceMonitorEvent,
6	6	ResourceMonitorExternalProcess,
7	7	ResourceMonitorHelloEvent,
	8	ResourceMonitorProcessTableEntry,
8	9	ResourceMonitorSnapshotEvent,
9	10	ResourceTelemetrySourceStatus,
10	11	} from "@t3tools/contracts";
...	...	@@ -44,6 +45,7 @@ const BATTERY_SAMPLE_INTERVAL_MS = 5_000;
44	45	const CONSTRAINED_SAMPLE_INTERVAL_MS = 15_000;
45	46	const HANDSHAKE_TIMEOUT = Duration.seconds(5);
46	47	const SAMPLE_REQUEST_TIMEOUT = Duration.seconds(5);
	48	const PROCESS_TABLE_REQUEST_TIMEOUT = Duration.seconds(5);
47	49	const HISTORY_REQUEST_TIMEOUT = Duration.seconds(15);
48	50	const INITIAL_RESTART_DELAY = Duration.millis(500);
49	51	const MAX_RESTART_DELAY = Duration.seconds(10);
...	...	@@ -76,7 +78,7 @@ export class NativeTelemetryHandshakeTimedOut extends Schema.TaggedErrorClass<Na
76	78	class NativeTelemetryRequestTimedOut extends Schema.TaggedErrorClass<NativeTelemetryRequestTimedOut>()(
77	79	"NativeTelemetryRequestTimedOut",
78	80	{
79		operation: Schema.Literals(["readHistory", "sampleNow"]),
	81	operation: Schema.Literals(["processTable", "readHistory", "sampleNow"]),
80	82	timeoutMs: Schema.Number,
81	83	},
82	84	) {
...	...	@@ -192,6 +194,10 @@ export class NativeTelemetryClient extends Context.Service<
192	194	snapshot: HostPowerSnapshot,
193	195	) => Effect.Effect<void, NativeTelemetryClientError>;
194	196	readonly sampleNow: Effect.Effect<NativeTelemetrySnapshot, NativeTelemetryClientError>;
	197	readonly processTable: Effect.Effect<
	198	ReadonlyArray<ResourceMonitorProcessTableEntry>,
	199	NativeTelemetryClientError
	200	>;
195	201	readonly retry: Effect.Effect<boolean>;
196	202	readonly health: Effect.Effect<NativeTelemetryClientHealth>;
197	203	readonly subscribeHealth: Effect.Effect<
...	...	@@ -382,6 +388,12 @@ export const make = Effect.fn("resourceTelemetry.nativeTelemetryClient.make")(fu
382	388	const pendingSamples = yield* Ref.make(
383	389	new Map<string, Deferred.Deferred<NativeTelemetrySnapshot, NativeTelemetryClientError>>(),
384	390	);
	391	const pendingProcessTables = yield* Ref.make(
	392	new Map<
	393	string,
	394	Deferred.Deferred<ReadonlyArray<ResourceMonitorProcessTableEntry>, NativeTelemetryClientError>
	395	>(),
	396	);
385	397	const pendingHistories = yield* Ref.make(new Map<string, PendingHistoryRequest>());
386	398	const snapshots = yield* PubSub.sliding<NativeTelemetrySnapshot>(8);
387	399	const healthChanges = yield* PubSub.sliding<NativeTelemetryClientHealth>(4);
...	...	@@ -399,10 +411,14 @@ export const make = Effect.fn("resourceTelemetry.nativeTelemetryClient.make")(fu
399	411	const failPending = (error: NativeTelemetryClientError) =>
400	412	Effect.gen(function* () {
401	413	const samples = yield* Ref.getAndSet(pendingSamples, new Map());
	414	const processTables = yield* Ref.getAndSet(pendingProcessTables, new Map());
402	415	const histories = yield* Ref.getAndSet(pendingHistories, new Map());
403	416	yield* Effect.forEach(samples.values(), (deferred) => Deferred.fail(deferred, error), {
404	417	discard: true,
405	418	});
	419	yield* Effect.forEach(processTables.values(), (deferred) => Deferred.fail(deferred, error), {
	420	discard: true,
	421	});
406	422	yield* Effect.forEach(
407	423	histories.values(),
408	424	(request) => Deferred.fail(request.deferred, error),
...	...	@@ -481,6 +497,21 @@ export const make = Effect.fn("resourceTelemetry.nativeTelemetryClient.make")(fu
481	497	}
482	498	}
483	499	});
	500	case "processTable":
	501	return Ref.modify(pendingProcessTables, (pending) => {
	502	const next = new Map(pending);
	503	const deferred = next.get(event.requestId);
	504	next.delete(event.requestId);
	505	return [Option.fromUndefinedOr(deferred), next] as const;
	506	}).pipe(
	507	Effect.flatMap(
	508	Option.match({
	509	onNone: () => Effect.void,
	510	onSome: (deferred) => Deferred.succeed(deferred, event.processes),
	511	}),
	512	),
	513	Effect.asVoid,
	514	);
484	515	case "historyChunk":
485	516	return Effect.gen(function* () {
486	517	const latestSnapshot = event.snapshots.at(-1);
...	...	@@ -936,6 +967,60 @@ export const make = Effect.fn("resourceTelemetry.nativeTelemetryClient.make")(fu
936	967	);
937	968	});
938	969
	970	const processTable: NativeTelemetryClient["Service"]["processTable"] = Effect.gen(function* () {
	971	const current = yield* Ref.get(state);
	972	if (!canCommandNativeTelemetrySidecar(current.status, Option.isSome(current.handle))) {
	973	return yield* new NativeTelemetryUnavailable({
	974	reason: Option.getOrElse(current.lastError, () => "sidecar is not running"),
	975	});
	976	}
	977
	978	const requestId = yield* crypto.randomUUIDv4.pipe(
	979	Effect.mapError(
	980	(cause) => new NativeTelemetryCommandFailed({ operation: "createRequestId", cause }),
	981	),
	982	);
	983	const deferred = yield* Deferred.make<
	984	ReadonlyArray<ResourceMonitorProcessTableEntry>,
	985	NativeTelemetryClientError
	986	>();
	987	yield* Ref.update(pendingProcessTables, (pending) => {
	988	const next = new Map(pending);
	989	next.set(requestId, deferred);
	990	return next;
	991	});
	992	return yield* writeCommand(Option.getOrThrow(current.handle), {
	993	version: RESOURCE_MONITOR_PROTOCOL_VERSION,
	994	type: "processTable",
	995	requestId,
	996	}).pipe(
	997	Effect.andThen(
	998	Deferred.await(deferred).pipe(
	999	Effect.timeoutOption(PROCESS_TABLE_REQUEST_TIMEOUT),
	1000	Effect.flatMap(
	1001	Option.match({
	1002	onNone: () =>
	1003	Effect.fail(
	1004	new NativeTelemetryRequestTimedOut({
	1005	operation: "processTable",
	1006	timeoutMs: Duration.toMillis(PROCESS_TABLE_REQUEST_TIMEOUT),
	1007	}),
	1008	),
	1009	onSome: Effect.succeed,
	1010	}),
	1011	),
	1012	),
	1013	),
	1014	Effect.ensuring(
	1015	Ref.update(pendingProcessTables, (pending) => {
	1016	const next = new Map(pending);
	1017	next.delete(requestId);
	1018	return next;
	1019	}),
	1020	),
	1021	);
	1022	});
	1023
939	1024	const health = currentHealth;
940	1025
941	1026	return NativeTelemetryClient.of({
...	...	@@ -957,6 +1042,7 @@ export const make = Effect.fn("resourceTelemetry.nativeTelemetryClient.make")(fu
957	1042	setExternalProcesses,
958	1043	setHostPowerState,
959	1044	sampleNow,
	1045	processTable,
960	1046	retry: Ref.get(state).pipe(
961	1047	Effect.flatMap((current) =>
962	1048	!canRequestNativeTelemetryRetry(current.status, Option.isSome(current.handle))
...	...	@@ -1010,6 +1096,11 @@ export const layerTest = (
1010	1096	reason: "No resource monitor sample was configured for this test.",
1011	1097	}),
1012	1098	),
	1099	processTable: Effect.fail(
	1100	new NativeTelemetryUnavailable({
	1101	reason: "No resource monitor process table was configured for this test.",
	1102	}),
	1103	),
1013	1104	retry: Effect.succeed(false),
1014	1105	health,
1015	1106	subscribeHealth:

apps/server/src/resourceTelemetry/ResourceTelemetry.test.ts+2-2

...	...	@@ -82,7 +82,7 @@ function nativeSnapshot(input: {
82	82	}),
83	83	];
84	84	return {
85		version: 2,
	85	version: 3,
86	86	type: "snapshot",
87	87	sequence: input.sequence,
88	88	sampledAtUnixMs: input.sampledAtUnixMs,
...	...	@@ -497,7 +497,7 @@ describe("ResourceTelemetry", () => {
497	497	const nativeHealth = yield* Ref.make<NativeTelemetryClient.NativeTelemetryClientHealth>({
498	498	status: "healthy",
499	499	hello: Option.some({
500		version: 2,
	500	version: 3,
501	501	type: "hello",
502	502	sidecarVersion: "0.1.0",
503	503	sidecarPid: 9_000,

apps/server/src/resourceTelemetry/ResourceTelemetryHistory.test.ts+1-1

...	...	@@ -64,7 +64,7 @@ function snapshot(
64	64	}),
65	65	];
66	66	return {
67		version: 2,
	67	version: 3,
68	68	type: "snapshot",
69	69	sequence,
70	70	sampledAtUnixMs,

apps/server/src/server.ts+1

...	...	@@ -374,6 +374,7 @@ const PortScannerLayerLive = PortScanner.layer.pipe(Layer.provide(ProcessRunner.
374	374	const TerminalLayerLive = TerminalManager.layer.pipe(
375	375	Layer.provide(PtyAdapterLive),
376	376	Layer.provide(PortScannerLayerLive),
	377	Layer.provide(NativeTelemetryLayerLive),
377	378	);
378	379
379	380	const PreviewLayerLive = Layer.empty.pipe(

apps/server/src/terminal/Manager.test.ts+96

...	...	@@ -14,6 +14,7 @@ import {
14	14	} from "@t3tools/contracts";
15	15	import { HostProcessPlatform } from "@t3tools/shared/hostProcess";
16	16	import * as Data from "effect/Data";
	17	import * as Clock from "effect/Clock";
17	18	import * as Deferred from "effect/Deferred";
18	19	import * as Duration from "effect/Duration";
19	20	import * as Effect from "effect/Effect";
...	...	@@ -220,6 +221,10 @@ interface CreateManagerOptions {
220	221	readonly childCommand: string \| null;
221	222	readonly processIds: ReadonlyArray<number>;
222	223	}>;
	224	processTable?: Effect.Effect<
	225	ReadonlyArray<{ readonly pid: number; readonly ppid: number; readonly name: string }>,
	226	never
	227	>;
223	228	subprocessPollIntervalMs?: number;
224	229	processKillGraceMs?: number;
225	230	maxRetainedInactiveSessions?: number;
...	...	@@ -265,6 +270,7 @@ const createManager = (
265	270	...(options.subprocessInspector !== undefined
266	271	? { subprocessInspector: options.subprocessInspector }
267	272	: {}),
	273	...(options.processTable !== undefined ? { processTable: options.processTable } : {}),
268	274	...(options.subprocessPollIntervalMs !== undefined
269	275	? { subprocessPollIntervalMs: options.subprocessPollIntervalMs }
270	276	: {}),
...	...	@@ -1192,6 +1198,96 @@ it.layer(
1192	1198	}),
1193	1199	);
1194	1200
	1201	it("calculates snapshot failure backoff and success reset delays", () => {
	1202	assert.equal(TerminalManager.subprocessSnapshotPollDelayMs(1_000, 0), 1_000);
	1203	assert.equal(TerminalManager.subprocessSnapshotPollDelayMs(1_000, 1), 2_000);
	1204	assert.equal(TerminalManager.subprocessSnapshotPollDelayMs(1_000, 2), 4_000);
	1205	assert.equal(TerminalManager.subprocessSnapshotPollDelayMs(1_000, 30), 60_000);
	1206	});
	1207
	1208	it.effect("uses process snapshots from the resource monitor", () =>
	1209	Effect.gen(function* () {
	1210	let snapshotCalls = 0;
	1211	const { manager, getEvents } = yield* createManager(5, {
	1212	subprocessPollIntervalMs: 20,
	1213	processTable: Effect.sync(() => {
	1214	snapshotCalls += 1;
	1215	return [{ pid: 100, ppid: 9000, name: "ping.exe" }];
	1216	}),
	1217	}).pipe(Effect.provide(withHostPlatform("win32")));
	1218
	1219	yield* manager.open(openInput());
	1220	yield* waitFor(
	1221	Effect.map(getEvents, (events) =>
	1222	events.some(
	1223	(event) =>
	1224	event.type === "activity" && event.hasRunningSubprocess && event.label === "ping",
	1225	),
	1226	),
	1227	"1200 millis",
	1228	);
	1229	expect(snapshotCalls).toBeGreaterThan(0);
	1230	}),
	1231	);
	1232
	1233	it.effect("backs off the spawned fallback when the resource monitor snapshot fails", () =>
	1234	Effect.gen(function* () {
	1235	const fallbackCalls: Array<number> = [];
	1236	const processRunner: ProcessRunner.ProcessRunner["Service"] = {
	1237	run: () =>
	1238	Clock.currentTimeMillis.pipe(
	1239	Effect.map((now) => {
	1240	fallbackCalls.push(now);
	1241	return {
	1242	stdout: " 100 9000 vim",
	1243	stderr: "",
	1244	code: ChildProcessSpawner.ExitCode(0),
	1245	timedOut: false,
	1246	stdoutTruncated: false,
	1247	stderrInvalidUtf8: false,
	1248	stdoutInvalidUtf8: false,
	1249	stderrTruncated: false,
	1250	};
	1251	}),
	1252	),
	1253	};
	1254
	1255	const { manager, getEvents } = yield* createManager(5, {
	1256	subprocessPollIntervalMs: 20,
	1257	processTable: Effect.fail("sidecar unavailable").pipe(
	1258	Effect.mapError((cause) => cause as never),
	1259	),
	1260	}).pipe(
	1261	Effect.provideService(ProcessRunner.ProcessRunner, processRunner),
	1262	Effect.provide(withHostPlatform("linux")),
	1263	);
	1264
	1265	yield* manager.open(openInput());
	1266	// The fallback data is still applied while the sidecar is down.
	1267	yield* waitFor(
	1268	Effect.map(getEvents, (events) =>
	1269	events.some(
	1270	(event) =>
	1271	event.type === "activity" &&
	1272	event.hasRunningSubprocess === true &&
	1273	event.label === "vim",
	1274	),
	1275	),
	1276	"1200 millis",
	1277	);
	1278
	1279	yield* waitFor(
	1280	Effect.sync(() => fallbackCalls.length >= 4),
	1281	"2000 millis",
	1282	);
	1283	// Four snapshots at the 20 ms base cadence would span ~60 ms. Backoff
	1284	// (40 + 80 + 160 ms) stretches the same four snapshots past 150 ms, so
	1285	// a stalled sidecar no longer hot-loops the spawned fallback.
	1286	const spanMs = fallbackCalls[3]! - fallbackCalls[0]!;
	1287	expect(spanMs).toBeGreaterThan(150);
	1288	}),
	1289	);
	1290
1195	1291	it.effect("caps persisted history to configured line limit", () =>
1196	1292	Effect.gen(function* () {
1197	1293	const { manager, ptyAdapter } = yield* createManager(3);

apps/server/src/terminal/Manager.ts+106-30

...	...	@@ -28,6 +28,7 @@ import {
28	28	type TerminalMetadataStreamEvent,
29	29	type TerminalOpenInput,
30	30	type TerminalResizeInput,
	31	type ResourceMonitorProcessTableEntry,
31	32	type TerminalRestartInput,
32	33	type TerminalSessionSnapshot,
33	34	type TerminalSessionStatus,
...	...	@@ -70,6 +71,7 @@ import {
70	71	import { expandHomePath } from "../pathExpansion.ts";
71	72	import * as ProcessRunner from "../processRunner.ts";
72	73	import * as PortScanner from "../preview/PortScanner.ts";
	74	import * as NativeTelemetryClient from "../resourceTelemetry/NativeTelemetryClient.ts";
73	75	import * as PtyAdapter from "./PtyAdapter.ts";
74	76
75	77	export {
...	...	@@ -92,6 +94,7 @@ const DEFAULT_HISTORY_BYTE_LIMIT = 8 * 1024 * 1024;
92	94	const MAX_HISTORY_CHUNK_LENGTH = 16 * 1024;
93	95	const DEFAULT_PERSIST_DEBOUNCE_MS = 40;
94	96	const DEFAULT_SUBPROCESS_POLL_INTERVAL_MS = 1_000;
	97	const MAX_SUBPROCESS_POLL_INTERVAL_MS = 60_000;
95	98	const DEFAULT_PROCESS_KILL_GRACE_MS = 1_000;
96	99	const DEFAULT_MAX_RETAINED_INACTIVE_SESSIONS = 128;
97	100	const DEFAULT_OPEN_COLS = 120;
...	...	@@ -106,7 +109,7 @@ class TerminalSubprocessCheckError extends Schema.TaggedErrorClass<TerminalSubpr
106	109	"TerminalSubprocessCheckError",
107	110	{
108	111	cause: Schema.optional(Schema.Defect()),
109		command: Schema.Literals(["powershell", "ps"]),
	112	command: Schema.Literals(["powershell", "ps", "resource-monitor"]),
110	113	exitCode: Schema.optional(Schema.NullOr(Schema.Number)),
111	114	timedOut: Schema.optional(Schema.Boolean),
112	115	stdoutTruncated: Schema.optional(Schema.Boolean),
...	...	@@ -641,6 +644,13 @@ interface TerminalProcessTableSnapshot {
641	644	readonly commandById: ReadonlyMap<number, string>;
642	645	}
643	646
	647	export function subprocessSnapshotPollDelayMs(
	648	pollIntervalMs: number,
	649	failureCount: number,
	650	): number {
	651	return Math.min(pollIntervalMs * 2 ** failureCount, MAX_SUBPROCESS_POLL_INTERVAL_MS);
	652	}
	653
644	654	function parsePosixProcessTable(stdout: string): TerminalProcessTableSnapshot {
645	655	const childrenByParent = new Map<number, number[]>();
646	656	const commandById = new Map<number, string>();
...	...	@@ -660,15 +670,15 @@ function parsePosixProcessTable(stdout: string): TerminalProcessTableSnapshot {
660	670	return { childrenByParent, commandById };
661	671	}
662	672
663		function parseWindowsProcessTable(stdout: string): TerminalProcessTableSnapshot {
	673	function processTableSnapshotFromProcesses(
	674	processes: ReadonlyArray<ResourceMonitorProcessTableEntry>,
	675	): TerminalProcessTableSnapshot {
664	676	const childrenByParent = new Map<number, number[]>();
665	677	const commandById = new Map<number, string>();
666		for (const line of stdout.split(/\r?\n/g)) {
667		const [pidRaw, parentPidRaw, nameRaw] = line.trim().split("\|", 3);
668		const pid = Number(pidRaw);
669		const parentPid = Number(parentPidRaw);
	678	for (const process of processes) {
	679	const { pid, ppid: parentPid, name } = process;
670	680	if (!Number.isInteger(pid) \|\| !Number.isInteger(parentPid)) continue;
671		commandById.set(pid, nameRaw?.trim() ?? "");
	681	commandById.set(pid, name.trim());
672	682	const children = childrenByParent.get(parentPid) ?? [];
673	683	children.push(pid);
674	684	childrenByParent.set(parentPid, children);
...	...	@@ -763,14 +773,11 @@ const windowsProcessTableSnapshot = Effect.fn("terminal.windowsProcessTableSnaps
763	773	TerminalSubprocessCheckError,
764	774	ProcessRunner.ProcessRunner
765	775	> {
	776	const processRunner = yield* ProcessRunner.ProcessRunner;
766	777	const command =
767	778	'Get-CimInstance Win32_Process -ErrorAction Stop \| ForEach-Object { Write-Output "$($_.ProcessId)\|$($_.ParentProcessId)\|$($_.Name)" }';
768		const processRunner = yield* ProcessRunner.ProcessRunner;
769	779	const result = yield* processRunner
770	780	.run({
771		// powershell.exe is a real executable — never spawn it through cmd.exe
772		// shell mode, which would re-tokenize the `-Command` payload (pipes,
773		// semicolons) before PowerShell ever sees it.
774	781	command: "powershell.exe",
775	782	args: ["-NoProfile", "-NonInteractive", "-Command", command],
776	783	timeout: "1500 millis",
...	...	@@ -780,16 +787,10 @@ const windowsProcessTableSnapshot = Effect.fn("terminal.windowsProcessTableSnaps
780	787	})
781	788	.pipe(
782	789	Effect.mapError(
783		(cause) =>
784		new TerminalSubprocessCheckError({
785		cause,
786		command: "powershell",
787		}),
	790	(cause) => new TerminalSubprocessCheckError({ cause, command: "powershell" }),
788	791	),
789	792	);
790	793	if (result.code !== 0 \|\| result.timedOut \|\| result.stdoutTruncated) {
791		// Not authoritative: an empty or partial table would mark every terminal
792		// idle and clear its registered process ids. Failing skips the tick.
793	794	return yield* new TerminalSubprocessCheckError({
794	795	command: "powershell",
795	796	exitCode: result.code,
...	...	@@ -797,7 +798,15 @@ const windowsProcessTableSnapshot = Effect.fn("terminal.windowsProcessTableSnaps
797	798	stdoutTruncated: result.stdoutTruncated,
798	799	});
799	800	}
800		return parseWindowsProcessTable(result.stdout);
	801	const processes = result.stdout.split(/\r?\n/g).flatMap((line) => {
	802	const [pidRaw, ppidRaw, name = ""] = line.trim().split("\|", 3);
	803	const pid = Number(pidRaw);
	804	const ppid = Number(ppidRaw);
	805	return Number.isInteger(pid) && pid > 0 && Number.isInteger(ppid)
	806	? [{ pid, ppid, name }]
	807	: [];
	808	});
	809	return processTableSnapshotFromProcesses(processes);
801	810	},
802	811	);
803	812
...	...	@@ -1310,6 +1319,10 @@ interface TerminalManagerOptions {
1310	1319	shellResolver?: () => string;
1311	1320	env?: NodeJS.ProcessEnv;
1312	1321	subprocessInspector?: TerminalSubprocessInspector;
	1322	processTable?: Effect.Effect<
	1323	ReadonlyArray<ResourceMonitorProcessTableEntry>,
	1324	TerminalSubprocessCheckError
	1325	>;
1313	1326	subprocessPollIntervalMs?: number;
1314	1327	processKillGraceMs?: number;
1315	1328	maxRetainedInactiveSessions?: number;
...	...	@@ -1376,6 +1389,7 @@ export const make = Effect.fn("TerminalManager.make")(function* () {
1376	1389	const { terminalLogsDir } = yield* ServerConfig.ServerConfig;
1377	1390	const ptyAdapter = yield* PtyAdapter.PtyAdapter;
1378	1391	const portDiscovery = yield* PortScanner.PortDiscovery;
	1392	const nativeTelemetry = yield* NativeTelemetryClient.NativeTelemetryClient;
1379	1393	const serverSettings = yield* ServerSettings.ServerSettingsService;
1380	1394	const path = yield* Path.Path;
1381	1395	const resolveProviderInstanceEnvironment = Effect.fn(
...	...	@@ -1391,6 +1405,11 @@ export const make = Effect.fn("TerminalManager.make")(function* () {
1391	1405	return yield* makeWithOptions({
1392	1406	logsDir: terminalLogsDir,
1393	1407	ptyAdapter,
	1408	processTable: nativeTelemetry.processTable.pipe(
	1409	Effect.mapError(
	1410	(cause) => new TerminalSubprocessCheckError({ cause, command: "resource-monitor" }),
	1411	),
	1412	),
1394	1413	registerTerminalProcesses: portDiscovery.registerTerminalProcesses,
1395	1414	unregisterTerminal: portDiscovery.unregisterTerminal,
1396	1415	resolveProviderInstanceEnvironment,
...	...	@@ -1437,23 +1456,60 @@ export const makeWithOptions = Effect.fn("TerminalManager.makeWithOptions")(func
1437	1456	// One process-table snapshot per poll tick, shared across every terminal.
1438	1457	// Per-terminal `pgrep`/`ps` calls multiply spawn load by terminal count and
1439	1458	// can exhaust the PID space on hosts with many sessions (#6332).
1440		const fetchProcessTableSnapshot = (
	1459	const fallbackProcessTableSnapshot = (
1441	1460	platform === "win32"
1442	1461	? windowsProcessTableSnapshot()
1443	1462	: posixProcessTableSnapshot(yield* resolvePosixPsCommand())
1444	1463	).pipe(Effect.provideService(ProcessRunner.ProcessRunner, processRunner));
	1464	const fetchProcessTableSnapshot: Effect.Effect<
	1465	{
	1466	readonly snapshot: TerminalProcessTableSnapshot;
	1467	/**
	1468	* False when the sidecar snapshot failed and this table came from the
	1469	* spawned fallback. The data is still applied, but the tick counts as
	1470	* a failure so polling backs off instead of hot-looping the fallback.
	1471	*/
	1472	readonly snapshotSucceeded: boolean;
	1473	},
	1474	TerminalSubprocessCheckError
	1475	> = options.processTable
	1476	? options.processTable.pipe(
	1477	Effect.map((entries) => ({
	1478	snapshot: processTableSnapshotFromProcesses(entries),
	1479	snapshotSucceeded: true,
	1480	})),
	1481	Effect.catch(() =>
	1482	fallbackProcessTableSnapshot.pipe(
	1483	Effect.map((snapshot) => ({ snapshot, snapshotSucceeded: false })),
	1484	),
	1485	),
	1486	)
	1487	: fallbackProcessTableSnapshot.pipe(
	1488	Effect.map((snapshot) => ({ snapshot, snapshotSucceeded: true })),
	1489	);
1445	1490	const customSubprocessInspector = options.subprocessInspector;
1446	1491	const acquireSubprocessInspector: Effect.Effect<
1447		TerminalSubprocessInspector,
	1492	{
	1493	readonly inspector: TerminalSubprocessInspector;
	1494	readonly snapshotSucceeded: boolean;
	1495	},
1448	1496	TerminalSubprocessCheckError
1449	1497	> =
1450	1498	customSubprocessInspector !== undefined
1451		? Effect.succeed(customSubprocessInspector)
	1499	? Effect.succeed({ inspector: customSubprocessInspector, snapshotSucceeded: true })
1452	1500	: Effect.map(
1453	1501	fetchProcessTableSnapshot,
1454		(snapshot): TerminalSubprocessInspector =>
1455		(terminalPid) =>
	1502	({
	1503	snapshot,
	1504	snapshotSucceeded,
	1505	}): {
	1506	readonly inspector: TerminalSubprocessInspector;
	1507	readonly snapshotSucceeded: boolean;
	1508	} => ({
	1509	inspector: (terminalPid) =>
1456	1510	Effect.succeed(deriveSubprocessInspectResult(snapshot, terminalPid, platform)),
	1511	snapshotSucceeded,
	1512	}),
1457	1513	);
1458	1514	const subprocessPollIntervalMs =
1459	1515	options.subprocessPollIntervalMs ?? DEFAULT_SUBPROCESS_POLL_INTERVAL_MS;
...	...	@@ -2305,7 +2361,7 @@ export const makeWithOptions = Effect.fn("TerminalManager.makeWithOptions")(func
2305	2361	);
2306	2362
2307	2363	if (runningSessions.length === 0) {
2308		return;
	2364	return true;
2309	2365	}
2310	2366
2311	2367	const inspectorOption = yield* acquireSubprocessInspector.pipe(
...	...	@@ -2313,15 +2369,22 @@ export const makeWithOptions = Effect.fn("TerminalManager.makeWithOptions")(func
2313	2369	Effect.catch((reason) =>
2314	2370	Effect.logWarning("failed to snapshot processes for terminal subprocess polling", {
2315	2371	reason,
2316		}).pipe(Effect.as(Option.none<TerminalSubprocessInspector>())),
	2372	}).pipe(
	2373	Effect.as(
	2374	Option.none<{
	2375	readonly inspector: TerminalSubprocessInspector;
	2376	readonly snapshotSucceeded: boolean;
	2377	}>(),
	2378	),
	2379	),
2317	2380	),
2318	2381	);
2319	2382
2320	2383	if (Option.isNone(inspectorOption)) {
2321		return;
	2384	return false;
2322	2385	}
2323	2386
2324		const subprocessInspector = inspectorOption.value;
	2387	const { inspector: subprocessInspector, snapshotSucceeded } = inspectorOption.value;
2325	2388
2326	2389	const checkSubprocessActivity = Effect.fn("terminal.checkSubprocessActivity")(function* (
2327	2390	session: TerminalSessionState & { pid: number },
...	...	@@ -2390,6 +2453,7 @@ export const makeWithOptions = Effect.fn("TerminalManager.makeWithOptions")(func
2390	2453	concurrency: "unbounded",
2391	2454	discard: true,
2392	2455	});
	2456	return snapshotSucceeded;
2393	2457	});
2394	2458
2395	2459	const hasRunningSessions = readManagerState.pipe(
...	...	@@ -2398,14 +2462,26 @@ export const makeWithOptions = Effect.fn("TerminalManager.makeWithOptions")(func
2398	2462	),
2399	2463	);
2400	2464
	2465	let subprocessSnapshotFailureCount = 0;
2401	2466	yield* Effect.forever(
2402	2467	hasRunningSessions.pipe(
2403	2468	Effect.flatMap((active) =>
2404	2469	active
2405	2470	? pollSubprocessActivity().pipe(
2406		Effect.flatMap(() => Effect.sleep(subprocessPollIntervalMs)),
	2471	Effect.flatMap((snapshotSucceeded) => {
	2472	subprocessSnapshotFailureCount = snapshotSucceeded
	2473	? 0
	2474	: Math.min(subprocessSnapshotFailureCount + 1, 30);
	2475	const delayMs = subprocessSnapshotPollDelayMs(
	2476	subprocessPollIntervalMs,
	2477	subprocessSnapshotFailureCount,
	2478	);
	2479	return Effect.sleep(delayMs);
	2480	}),
2407	2481	)
2408		: Effect.sleep(subprocessPollIntervalMs),
	2482	: Effect.sync(() => {
	2483	subprocessSnapshotFailureCount = 0;
	2484	}).pipe(Effect.flatMap(() => Effect.sleep(subprocessPollIntervalMs))),
2409	2485	),
2410	2486	),
2411	2487	).pipe(Effect.forkIn(workerScope));

native/resource-monitor/src/main.rs+79-3

...	...	@@ -8,7 +8,7 @@ use sysinfo::{
8	8	MINIMUM_CPU_UPDATE_INTERVAL, Pid, ProcessRefreshKind, ProcessesToUpdate, System, UpdateKind,
9	9	};
10	10
11		const PROTOCOL_VERSION: u32 = 2;
	11	const PROTOCOL_VERSION: u32 = 3;
12	12	const MIN_SAMPLE_INTERVAL_MS: u64 = 250;
13	13	const MAX_SAMPLE_INTERVAL_MS: u64 = 60_000;
14	14	const PROCESS_START_TIME_PRECISION_MS: u64 = 1_000;
...	...	@@ -66,6 +66,10 @@ enum Command {
66	66	version: u32,
67	67	request_id: String,
68	68	},
	69	ProcessTable {
	70	version: u32,
	71	request_id: String,
	72	},
69	73	ReadHistory {
70	74	version: u32,
71	75	request_id: String,
...	...	@@ -84,6 +88,7 @@ impl Command {
84	88	\| Self::SetSampleInterval { version, .. }
85	89	\| Self::SetStreaming { version, .. }
86	90	\| Self::SampleNow { version, .. }
	91	\| Self::ProcessTable { version, .. }
87	92	\| Self::ReadHistory { version, .. }
88	93	\| Self::Shutdown { version } => *version,
89	94	}
...	...	@@ -146,6 +151,24 @@ struct ProcessSample {
146	151	io_semantics: IoSemantics,
147	152	}
148	153
	154	#[derive(Debug, Serialize)]
	155	#[serde(rename_all = "camelCase")]
	156	struct ProcessTableEntry {
	157	pid: u32,
	158	ppid: u32,
	159	name: String,
	160	}
	161
	162	#[derive(Debug, Serialize)]
	163	#[serde(rename_all = "camelCase")]
	164	struct ProcessTableEvent<'a> {
	165	version: u32,
	166	#[serde(rename = "type")]
	167	event_type: &'static str,
	168	request_id: &'a str,
	169	processes: Vec<ProcessTableEntry>,
	170	}
	171
149	172	impl ProcessSample {
150	173	fn estimated_history_bytes(&self) -> usize {
151	174	std::mem::size_of::<Self>()
...	...	@@ -351,6 +374,41 @@ impl Collector {
351	374	self.cpu_baseline_refreshed_at = Some(Instant::now());
352	375	}
353	376
	377	fn process_table(&self) -> Vec<ProcessTableEntry> {
	378	// Use a dedicated System so this refresh cannot reset the CPU
	379	// baseline tracked by self.system for snapshots.
	380	let mut process_table_system = System::new();
	381	process_table_system.refresh_processes_specifics(
	382	ProcessesToUpdate::All,
	383	true,
	384	ProcessRefreshKind::nothing().without_tasks(),
	385	);
	386	let mut processes = process_table_system
	387	.processes()
	388	.iter()
	389	.filter_map(\|(pid, process)\| {
	390	let pid = pid.as_u32();
	391	// Pid 0 is the kernel idle process on some platforms. The
	392	// processTable contract requires positive pids, and one zero
	393	// would fail the whole event decode on the server, so drop it
	394	// here. It can never be a terminal descendant.
	395	if pid == 0 {
	396	return None;
	397	}
	398	Some(ProcessTableEntry {
	399	pid,
	400	ppid: process.parent().map(Pid::as_u32).unwrap_or(0),
	401	name: truncate_utf8(
	402	process.name().to_string_lossy().into_owned(),
	403	MAX_PROCESS_NAME_BYTES,
	404	),
	405	})
	406	})
	407	.collect::<Vec<_>>();
	408	processes.sort_by_key(\|process\| process.pid);
	409	processes
	410	}
	411
354	412	fn sample(&mut self, config: &CollectorConfig, request_id: Option<String>) -> SnapshotEvent {
355	413	if let Some(delay) =
356	414	remaining_cpu_measurement_delay(self.cpu_baseline_refreshed_at.take(), Instant::now())
...	...	@@ -871,6 +929,15 @@ fn main() -> io::Result<()> {
871	929	)?;
872	930	}
873	931	}
	932	Command::ProcessTable { request_id, .. } => {
	933	let event = ProcessTableEvent {
	934	version: PROTOCOL_VERSION,
	935	event_type: "processTable",
	936	request_id: &request_id,
	937	processes: collector.process_table(),
	938	};
	939	write_event(&mut writer, &event)?;
	940	}
874	941	Command::ReadHistory {
875	942	request_id,
876	943	window_ms,
...	...	@@ -981,7 +1048,7 @@ mod tests {
981	1048	#[test]
982	1049	fn decodes_protocol_commands() {
983	1050	let configure = serde_json::from_str::<Command>(
984		r#"{"version":2,"type":"configure","rootPid":42,"sampleIntervalMs":1000,"externalProcesses":[{"pid":7}]}"#,
	1051	r#"{"version":3,"type":"configure","rootPid":42,"sampleIntervalMs":1000,"externalProcesses":[{"pid":7}]}"#,
985	1052	)
986	1053	.expect("configure command");
987	1054
...	...	@@ -1001,7 +1068,7 @@ mod tests {
1001	1068	}
1002	1069
1003	1070	let read_history = serde_json::from_str::<Command>(
1004		r#"{"version":2,"type":"readHistory","requestId":"history-1","windowMs":60000}"#,
	1071	r#"{"version":3,"type":"readHistory","requestId":"history-1","windowMs":60000}"#,
1005	1072	)
1006	1073	.expect("read history command");
1007	1074	assert!(matches!(
...	...	@@ -1012,6 +1079,15 @@ mod tests {
1012	1079	..
1013	1080	} if request_id == "history-1"
1014	1081	));
	1082
	1083	let process_table = serde_json::from_str::<Command>(
	1084	r#"{"version":3,"type":"processTable","requestId":"processes-1"}"#,
	1085	)
	1086	.expect("process table command");
	1087	assert!(matches!(
	1088	process_table,
	1089	Command::ProcessTable { request_id, .. } if request_id == "processes-1"
	1090	));
1015	1091	}
1016	1092
1017	1093	#[test]

packages/contracts/src/resourceTelemetry.ts+25-1

...	...	@@ -4,7 +4,7 @@ import { NonNegativeInt, PositiveInt, TrimmedNonEmptyString } from "./baseSchema
4	4	import { HostPowerSnapshot } from "./background.ts";
5	5	import { DesktopUpdateStateSchema } from "./ipc.ts";
6	6
7		export const RESOURCE_MONITOR_PROTOCOL_VERSION = 2 as const;
	7	export const RESOURCE_MONITOR_PROTOCOL_VERSION = 3 as const;
8	8
9	9	/** Whole-host capacity, independent of T3's process diagnostics. */
10	10	export const HostResourcesSnapshot = Schema.Struct({
...	...	@@ -112,6 +112,13 @@ export const ResourceMonitorSampleNowCommand = Schema.Struct({
112	112	});
113	113	export type ResourceMonitorSampleNowCommand = typeof ResourceMonitorSampleNowCommand.Type;
114	114
	115	export const ResourceMonitorProcessTableCommand = Schema.Struct({
	116	version: Schema.Literal(RESOURCE_MONITOR_PROTOCOL_VERSION),
	117	type: Schema.Literal("processTable"),
	118	requestId: TrimmedNonEmptyString,
	119	});
	120	export type ResourceMonitorProcessTableCommand = typeof ResourceMonitorProcessTableCommand.Type;
	121
115	122	export const ResourceMonitorSetSampleIntervalCommand = Schema.Struct({
116	123	version: Schema.Literal(RESOURCE_MONITOR_PROTOCOL_VERSION),
117	124	type: Schema.Literal("setSampleInterval"),
...	...	@@ -147,6 +154,7 @@ export const ResourceMonitorCommand = Schema.Union([
147	154	ResourceMonitorSetSampleIntervalCommand,
148	155	ResourceMonitorSetStreamingCommand,
149	156	ResourceMonitorSampleNowCommand,
	157	ResourceMonitorProcessTableCommand,
150	158	ResourceMonitorReadHistoryCommand,
151	159	ResourceMonitorShutdownCommand,
152	160	]);
...	...	@@ -178,6 +186,21 @@ export const ResourceMonitorSnapshotEvent = Schema.Struct({
178	186	});
179	187	export type ResourceMonitorSnapshotEvent = typeof ResourceMonitorSnapshotEvent.Type;
180	188
	189	export const ResourceMonitorProcessTableEntry = Schema.Struct({
	190	pid: PositiveInt,
	191	ppid: NonNegativeInt,
	192	name: Schema.String,
	193	});
	194	export type ResourceMonitorProcessTableEntry = typeof ResourceMonitorProcessTableEntry.Type;
	195
	196	export const ResourceMonitorProcessTableEvent = Schema.Struct({
	197	version: Schema.Literal(RESOURCE_MONITOR_PROTOCOL_VERSION),
	198	type: Schema.Literal("processTable"),
	199	requestId: TrimmedNonEmptyString,
	200	processes: Schema.Array(ResourceMonitorProcessTableEntry),
	201	});
	202	export type ResourceMonitorProcessTableEvent = typeof ResourceMonitorProcessTableEvent.Type;
	203
181	204	export const ResourceMonitorHistoryChunkEvent = Schema.Struct({
182	205	version: Schema.Literal(RESOURCE_MONITOR_PROTOCOL_VERSION),
183	206	type: Schema.Literal("historyChunk"),
...	...	@@ -199,6 +222,7 @@ export type ResourceMonitorErrorEvent = typeof ResourceMonitorErrorEvent.Type;
199	222	export const ResourceMonitorEvent = Schema.Union([
200	223	ResourceMonitorHelloEvent,
201	224	ResourceMonitorSnapshotEvent,
	225	ResourceMonitorProcessTableEvent,
202	226	ResourceMonitorHistoryChunkEvent,
203	227	ResourceMonitorErrorEvent,
204	228	]);

author	Utkarsh Patil <73941998+UtkarshUsername@users.noreply.github.com> 2026-09-07 01:17:38+05:30
committer	GitHub <noreply@github.com> 2026-09-06 12:47:38-07:00
log	ea646c0834a3394ecb0be4a30c5d367e5a9002bd
tree	d1f38e9b651690ebd49829ea82cda7bedd8cb8d5
parent	95d99373b74edfb3bfd092c61a15999a95727423
signature	Signed by PGP key B5690EEEBB952194

fix(server): stop Windows terminal polling from spiking CPU (#9476)