test: fix split-UTF-8 decoding in test tools, trim test memory, and remove an RSS flake (batch 2)

TESTING.md

@@ -1,6 +1,6 @@
 # Testing data-connectors
 
-Every local test command runs through `scripts/test-sandbox/test-sandbox`, a vendored copy of the host-wide test governor (see `scripts/test-sandbox/README.md`). It gives each run a systemd scope under the shared `agent-tests.slice` with 400% CPU, 8 GB and 4096 tasks, one of the host-wide slots, a loopback-only network unless the lane opts out, and a private `/tmp`. Stopping the scope kills every descendant. Each run prints a JSON receipt (wall, CPU, peak memory) on stderr. Where systemd user sessions, bwrap or flock are missing (macOS, GitHub runners), it runs the command directly and says so in the receipt.
+Every local test command runs through `scripts/test-sandbox/test-sandbox`, a vendored copy of the host-wide test governor (see `scripts/test-sandbox/README.md`). It gives each run a systemd scope under the shared `agent-tests.slice` with 400% CPU, 8 GB and 4096 tasks, one of the host-wide slots, a loopback-only network unless the lane opts out, and a private `/tmp`. Stopping the scope kills every descendant. Each run prints a JSON receipt (wall, CPU, peak memory) on stderr. Each layer it cannot use on a host is skipped and named in the receipt; see `scripts/test-sandbox/README.md` for what that means on macOS and on GitHub runners.
 
 ## Tiers
 
@@ -9,7 +9,7 @@ Every local test command runs through `scripts/test-sandbox/test-sandbox`, a ven
 | While editing | `npm run test:fast` (root or packages/polyfill-connectors) | Tests related to files changed since `git merge-base HEAD origin/main`, plus uncommitted and untracked files: the static import graph, plus every test in the same `connectors/<name>/` directory. Namespace tests run in their own lane. When the selector cannot trust the graph (for example a change under `scripts/related-tests/` or a deleted file), it runs the whole sandboxed lane. | one-connector edit: 42 tests, 4-10 s |
 | Same, against any base | `npm run test:related -- <base>` (in packages/polyfill-connectors) | Same selection; on FULL_SUITE it runs both polyfill lanes. | |
 | Before a PR | `npm test` (root) | Every lane in turn (below). | about 28 min, 16 of them PageShim |
-| PR into main | GitHub Actions | `verify + test`, PageShim shards, connector shards, Contract Guardrails, cross-repo integrity. PRs into other branches run no CI. On GitHub runners the sandbox also enforces (cgroup+netns). | `verify + test` Test step 699 s (was 15-18 min at concurrency 2); PageShim shards 588 s and 835 s |
+| PR into main | GitHub Actions | `verify + test`, PageShim shards, connector shards, Contract Guardrails, cross-repo integrity. PRs into other branches run no CI. `verify + test` runs its lanes in the sandbox with cgroup limits and network isolation, and Contract Guardrails with cgroup limits only; the PageShim and connector shards and cross-repo integrity run without it (each CI job is its own VM). | `verify + test` Test step 699 s (was 15-18 min at concurrency 2); PageShim shards 588 s and 835 s |
 
 The root `npm test` runs these lanes, each in its own sandbox:
 
@@ -25,10 +25,10 @@ The root `npm test` runs these lanes, each in its own sandbox:
 
 ## What is shared, per-worker, per-test, locked
 
-- Shared: the checkout, `node_modules`, the Playwright browser cache, the built PageShim bundles of one run. Tests treat them as read-only, except that the namespace tests plant and remove sockets under the repository root (see below).
+- Shared: the checkout, `node_modules`, the Playwright browser cache, the built PageShim bundles of one run. Tests write into the checkout in a few places: the namespace tests plant and remove sockets under the repository root (see below); PageShim tests build bundles under `.tmp/pageshim/` and `.scratch/`; `bin/connector-dev.test.ts` creates and removes `packages/polyfill-connectors/runs/<id>` and `connectors/<id>/`; `bin/connector-init.test.ts` scaffolds and removes `connectors/zz_init_smoke_<pid>/` and `connectors/zz_init_browser_smoke_<pid>/` (a killed run leaves them, and the manifest tests then fail); `bin/scenario-cli.test.ts` writes fixture connectors under `packages/polyfill-connectors/tmp/`; and one consent-time-field test makes a temporary directory inside `node_modules`. Do not run two suites in one checkout at once.
 - Per test file: one `node --test` child process. Module state (PageShim fixture modules with `setLoggedIn`) is per file; tests in one file run one after another.
 - Per test: temp directories (`mkdtemp` under the sandbox's private `/tmp`), local HTTP servers on port 0, iMessage `chat.db` files, PageShim browsers (one Chromium per `runHarness`, about 140 ms to start).
-- Locked (serial): the namespaces lane. Its files share the repository root as global state (`isolation-mechanism.test.ts` plants sockets under it that `scenario-verify-strict.test.ts`'s isolation scan refuses), so one run executes them one at a time. Nothing locks them across two runs in the same checkout, so do not run two suites in one checkout at once.
+- Locked (serial): the namespaces lane. Its files share the repository root as global state (`isolation-mechanism.test.ts` plants sockets under it that `scenario-verify-strict.test.ts`'s isolation scan refuses), so one run executes them one at a time. Nothing locks them across two runs in the same checkout.
 - Never shared: the network. Unit lanes have loopback only; PageShim routes deny every request that no fixture serves.
 
 ## Rules

connectors/apple_health/index.test.ts

@@ -109,6 +109,7 @@ interface RunOpts {
 	peakRssPollIntervalMs?: number;
 	/** Restrict every health-area stream to specific record ids, as a host may. */
 	resources?: string[];
+	retainRawStdout?: boolean;
 	state?: Record<string, unknown>;
 	/** The streams to request. Defaults to every data stream and the receipts. */
 	streams?: readonly string[];
@@ -155,6 +156,9 @@ async function run(dir: string, opts: RunOpts = {}) {
 			? {}
 			: { peakRssPollIntervalMs: opts.peakRssPollIntervalMs }),
 		...(opts.timeoutMs === undefined ? {} : { timeoutMs: opts.timeoutMs }),
+		...(opts.retainRawStdout === undefined
+			? {}
+			: { retainRawStdout: opts.retainRawStdout }),
 		start: {
 			...(opts.fullRefresh ? { collection_mode: "full_refresh" } : {}),
 			scope: { streams: streams.map((name) => scopeFor(name, opts)) },
@@ -3286,8 +3290,10 @@ test("with both identity budgets spent, memory stays within the streaming bound
 		`<?xml version="1.0" encoding="UTF-8"?>\n<HealthData locale="en_US">\n <ExportDate value="2026-09-01 12:00:00 -0500"/>\n` +
 		`${parts.join("\n")}\n</HealthData>\n`;
 	await withImportDir({ "export.xml": xml }, async (dir) => {
+		// Only parsed messages are read here, and there are ~500k of them.
 		const result = await run(dir, {
 			peakRssPollIntervalMs: 200,
+			retainRawStdout: false,
 			timeoutMs: 540_000,
 		});
 		assert.ok(result.peakRssBytes !== null, "expected a sampled peak RSS");

connectors/google_maps/validation-from-file.test.ts

@@ -561,7 +561,9 @@ test("validateGoogleMapsTimelineArtifactFromFile: a large sparse export never ca
 	}
 });
 
-test("outcome proof: a 110 MiB sparse Timeline export separates streaming from whole-buffer parsing at one heap limit", async () => {
+test("outcome proof: a 110 MiB sparse Timeline export separates streaming from whole-buffer parsing at one heap limit", {
+	timeout: 240_000,
+}, async () => {
 	// Deterministic worker-thread oracle, not an in-process memory-delta
 	// measurement (the prior in-process external-memory-delta test here was
 	// proven decorative by mutation testing -- it passed even when reverted
@@ -620,7 +622,9 @@ test("outcome proof: a 110 MiB sparse Timeline export separates streaming from w
 		// A worker resource limit reports OOM without aborting the process or
 		// generating a core dump for this expected negative-control outcome.
 		const heapLimitMiB = 128;
-		const timeoutMs = 30_000;
+		// A hang guard, not a speed bound: on a 4-vCPU CI runner with four test
+		// files at once, the streaming pass ran past 30 s (the whole test took 38.6 s).
+		const timeoutMs = 90_000;
 		type OracleOutcome = {
 			error: (Error & { code?: string }) | undefined;
 			exitCode: number;

connectors/strava/no-whole-file-read.test.ts

@@ -96,6 +96,8 @@ test("large synthetic activities.csv emits with bounded RSS", async () => {
 				TZ: "UTC",
 			},
 			peakRssPollIntervalMs: 25,
+			// Only parsed messages are asserted on, so skip the raw stdout copy.
+			retainRawStdout: false,
 			start: {
 				scope: {
 					streams: [{ name: "activities" }, ],

connectors/whatsapp/no-whole-file-read.test.ts

@@ -139,6 +139,8 @@ test("a large synthetic WhatsApp .txt export (well past the old 1 GiB cap) parse
 				WHATSAPP_EXPORT_DIR: importRoot,
 			},
 			peakRssPollIntervalMs: 25,
+			// Only parsed messages are asserted on, so skip the raw stdout copy.
+			retainRawStdout: false,
 			start: {
 				scope: { streams: [{ name: "chats" }, { name: "messages" }] },
 				type: "START",

packages/polyfill-connectors/bin/connector-dev.test.ts

@@ -23,6 +23,7 @@ import {
 	readFileSync,
 	rmdirSync,
 	rmSync,
+	writeFileSync,
 } from "node:fs";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
@@ -1068,3 +1069,62 @@ test("connector-dev CLI: a connector cannot launder its auth failure into runtim
 		rmSync(tmpDir, { recursive: true, force: true });
 	}
 });
+
+test("connector-dev CLI: preserves multi-byte STATE cursors across pipe chunks", () => {
+	const tmpDir = mkdtempSync(join(tmpdir(), "connector-dev-utf8-"));
+	const connector = `connector-dev-utf8-${String(process.pid)}`;
+	const entrypoint = join(tmpDir, "connector.mjs");
+	const expected = "€".repeat(100_001);
+	try {
+		writeFileSync(
+			entrypoint,
+			`
+import { once } from "node:events";
+import { setTimeout as delay } from "node:timers/promises";
+await once(process.stdin, "data");
+const emit = (message) => new Promise((resolve, reject) => {
+  process.stdout.write(JSON.stringify(message) + "\\n", (error) => error ? reject(error) : resolve());
+});
+await emit({ type: "RECORD", stream: "items", key: "utf8", emitted_at: "2026-08-01T00:00:00.000Z", data: { id: "utf8", value: "fixture" } });
+const bytes = Buffer.from(JSON.stringify({
+  type: "STATE", stream: "items", cursor: { value: "€".repeat(100_001) }
+}) + "\\n");
+// The 64 KiB split lands inside a three-byte character. Waiting after the
+// first write makes the split deterministic even when pipe reads coalesce.
+const split = 65_536;
+await new Promise((resolve, reject) => {
+  process.stdout.write(bytes.subarray(0, split), (error) => error ? reject(error) : resolve());
+});
+await delay(40);
+await new Promise((resolve, reject) => {
+  process.stdout.write(bytes.subarray(split), (error) => error ? reject(error) : resolve());
+});
+await emit({ type: "DONE", status: "succeeded", records_emitted: 1 });
+process.stdin.resume();
+`,
+		);
+		const result = runCli([
+			connector,
+			"--entrypoint",
+			entrypoint,
+			"--summary-out",
+			join(tmpDir, "summary.json"),
+		]);
+		assert.equal(result.code, 0, result.stderr);
+		const saved = JSON.parse(
+			readFileSync(lastStatePathFor(connector), "utf8"),
+		) as LastState;
+		assert.deepEqual(Object.keys(saved.state), ["items"]);
+		assert.equal(
+			(saved.state.items as { value: string }).value === expected,
+			true,
+			"saved cursor must retain every multi-byte character exactly",
+		);
+	} finally {
+		rmSync(tmpDir, { recursive: true, force: true });
+		rmSync(join(PACKAGE_ROOT, "runs", connector), {
+			recursive: true,
+			force: true,
+		});
+	}
+});

packages/polyfill-connectors/bin/connector-dev.ts

@@ -987,8 +987,9 @@ export function runAndStream(
 			}
 		};
 
-		child.stdout?.on("data", (chunk: Buffer) => {
-			stdoutBuffer += chunk.toString();
+		child.stdout?.setEncoding("utf8");
+		child.stdout?.on("data", (chunk: string) => {
+			stdoutBuffer += chunk;
 			let newlineIndex = stdoutBuffer.indexOf("\n");
 			while (newlineIndex !== -1) {
 				const line = stdoutBuffer.slice(0, newlineIndex);

packages/polyfill-connectors/bin/scenario-cli.test.ts

@@ -4059,3 +4059,129 @@ test("scenario-verify: replay subprocess forces PLAYWRIGHT_BROWSERS_PATH and PDP
 		rmSync(connectorPath, { force: true });
 	}
 });
+
+function writeMultiByteConnector(directory: string): string {
+	const entrypoint = join(directory, "connector.mjs");
+	writeFileSync(
+		entrypoint,
+		`
+import { once } from "node:events";
+import { setTimeout as delay } from "node:timers/promises";
+await once(process.stdin, "data");
+const emit = (message) => new Promise((resolve, reject) => {
+  process.stdout.write(JSON.stringify(message) + "\\n", (error) => error ? reject(error) : resolve());
+});
+await emit({ type: "RECORD", stream: "items", key: "utf8", emitted_at: "2026-08-01T00:00:00.000Z", data: { id: "utf8", value: "fixture" } });
+const bytes = Buffer.from(JSON.stringify({
+  type: "STATE", stream: "items", cursor: { value: "€".repeat(100_001) }
+}) + "\\n");
+// The 64 KiB split lands inside a three-byte character. Waiting after the
+// first write makes the split deterministic even when pipe reads coalesce.
+const split = 65_536;
+await new Promise((resolve, reject) => {
+  process.stdout.write(bytes.subarray(0, split), (error) => error ? reject(error) : resolve());
+});
+await delay(40);
+await new Promise((resolve, reject) => {
+  process.stdout.write(bytes.subarray(split), (error) => error ? reject(error) : resolve());
+});
+await emit({ type: "DONE", status: "succeeded", records_emitted: 1 });
+process.stdin.resume();
+`,
+	);
+	return entrypoint;
+}
+
+test("scenario-record: preserves multi-byte STATE cursors across pipe chunks", () => {
+	const tmpDir = mkdtempSync(
+		join(packageScratchDir(), "scenario-record-utf8-"),
+	);
+	const scenarioPath = join(tmpDir, "utf8.scenario.json");
+	try {
+		const entrypoint = writeMultiByteConnector(tmpDir);
+		const result = runRecordCli(
+			[
+				"scenario-utf8-fixture",
+				"--entrypoint",
+				entrypoint,
+				"--runs",
+				"1",
+				"--out",
+				scenarioPath,
+			],
+			{},
+		);
+		assert.equal(result.code, 0, result.stderr);
+		const scenario = JSON.parse(
+			readFileSync(scenarioPath, "utf8"),
+		) as ConnectorScenario;
+		assert.equal(scenario.capture.complete, true);
+		const finalState = scenario.runs[0]?.expected.final_state as
+			| { items?: { value?: string } }
+			| undefined;
+		assert.equal(
+			finalState?.items?.value === "€".repeat(100_001),
+			true,
+			"recorded cursor must retain every multi-byte character exactly",
+		);
+	} finally {
+		rmSync(tmpDir, { recursive: true, force: true });
+	}
+});
+
+test("scenario-verify: preserves multi-byte STATE cursors across pipe chunks", () => {
+	const tmpDir = mkdtempSync(
+		join(packageScratchDir(), "scenario-verify-utf8-"),
+	);
+	const scenarioPath = join(tmpDir, "utf8.scenario.json");
+	try {
+		const entrypoint = writeMultiByteConnector(tmpDir);
+		// Construct the oracle independently: a broken recorder must not hide
+		// identical corruption in the verifier by supplying a corrupted cursor.
+		const scenario: ConnectorScenario = {
+			format: "pdpp.connector-scenario/1",
+			connector: { id: "scenario-utf8-fixture" },
+			capture: {
+				captured_at: "2026-08-01T00:00:00.000Z",
+				evidence_class: "synthetic-spike",
+				privacy_class: "local-only",
+				recorder_version: "test",
+				complete: true,
+			},
+			runs: [
+				{
+					start: { scope: { streams: [{ name: "items" }] }, state: null },
+					interactions: [],
+					expected: {
+						records: {
+							items: {
+								count: 1,
+								ids: ["utf8"],
+								ops: ["upsert"],
+								record_sha256s: [
+									hashCanonicalJson({ id: "utf8", value: "fixture" }),
+								],
+							},
+						},
+						final_state: { items: { value: "€".repeat(100_001) } },
+					},
+				},
+			],
+		};
+		writeFileSync(scenarioPath, JSON.stringify(scenario));
+		const result = runVerifyCli([
+			"scenario-utf8-fixture",
+			"--entrypoint",
+			entrypoint,
+			scenarioPath,
+		]);
+		assert.equal(
+			result.code,
+			0,
+			`stdout=${result.stdout.slice(0, 2000)} stderr=${result.stderr}`,
+		);
+		assert.match(result.stdout, /run 0: PASS/);
+	} finally {
+		rmSync(tmpDir, { recursive: true, force: true });
+	}
+});

packages/polyfill-connectors/bin/scenario-record.ts

@@ -1684,12 +1684,13 @@ function runRecordSubprocess(args: {
 			}
 		};
 
-		child.stdout.on("data", (chunk: Buffer) => {
+		child.stdout.setEncoding("utf8");
+		child.stdout.on("data", (chunk: string) => {
 			// Activity — resets the inactivity window (a no-op while suspended
 			// for a pending INTERACTION; see `answerInteraction`/`handleParsedLine`
 			// above).
 			watchdog.touch();
-			stdoutBuffer += chunk.toString();
+			stdoutBuffer += chunk;
 			let newlineIndex = stdoutBuffer.indexOf("\n");
 			while (newlineIndex !== -1) {
 				const line = stdoutBuffer.slice(0, newlineIndex);
@@ -1700,9 +1701,10 @@ function runRecordSubprocess(args: {
 				newlineIndex = stdoutBuffer.indexOf("\n");
 			}
 		});
-		child.stderr.on("data", (chunk: Buffer) => {
+		child.stderr.setEncoding("utf8");
+		child.stderr.on("data", (chunk: string) => {
 			watchdog.touch();
-			stderr += chunk.toString();
+			stderr += chunk;
 		});
 		child.on("error", (err) => {
 			watchdog.dispose();

packages/polyfill-connectors/bin/scenario-verify.ts

@@ -1504,15 +1504,16 @@ function runReplaySubprocess(args: {
 			}
 		};
 
-		childStdout.on("data", (chunk: Buffer) => {
+		childStdout.setEncoding("utf8");
+		childStdout.on("data", (chunk: string) => {
 			// Activity — resets the inactivity window (replay's scripted
 			// interaction answering never waits on a human, so this watchdog is
 			// never suspended — see this file's "Inactivity watchdog" section).
 			watchdog.touch();
 			if (hardFailure) {
 				return;
 			}
-			stdoutBuffer += chunk.toString();
+			stdoutBuffer += chunk;
 			let newlineIndex = stdoutBuffer.indexOf("\n");
 			while (newlineIndex !== -1) {
 				const line = stdoutBuffer.slice(0, newlineIndex);
@@ -1529,9 +1530,10 @@ function runReplaySubprocess(args: {
 				newlineIndex = stdoutBuffer.indexOf("\n");
 			}
 		});
-		childStderr.on("data", (chunk: Buffer) => {
+		childStderr.setEncoding("utf8");
+		childStderr.on("data", (chunk: string) => {
 			watchdog.touch();
-			stderr += chunk.toString();
+			stderr += chunk;
 		});
 		child.on("error", (err) => {
 			watchdog.dispose();

packages/polyfill-connectors/scripts/no-await-in-loops-allowlist.ts

@@ -686,7 +686,7 @@ export const NO_AWAIT_IN_LOOPS_ALLOWLIST: readonly NoAwaitInLoopsAllowlistEntry[
 		},
 		{
 			path: "connectors/apple_health/index.test.ts",
-			line: 3498,
+			line: 3504,
 			column: 3,
 			category: "test_assertion_sequencing",
 			note: "small-XML-entry archive cases: each imported in turn by its own subprocess, whose peak RSS is measured alone",

packages/polyfill-connectors/src/connector-runtime-consent-time-field.test.ts

@@ -121,8 +121,12 @@ test("the generated consent-time-field modules match the manifests on disk", ()
 		);
 		// One slice per manifest, no extra, each current.
 		const expected = readdirSync(join(scratch, "consent-time-fields")).sort();
+		// `npm run build` (and so `npm pack`) emits .js and .d.ts next to each
+		// slice; only the generated sources are the generator's output.
 		assert.deepEqual(
-			readdirSync(join(tracked, "consent-time-fields")).sort(),
+			readdirSync(join(tracked, "consent-time-fields"))
+				.filter((name) => name.endsWith(".ts") && !name.endsWith(".d.ts"))
+				.sort(),
 			expected,
 			regenerate,
 		);