diff --git a/integrationTests/server/sse-finite-generator.test.ts b/integrationTests/server/sse-finite-generator.test.ts
new file mode 100644
index 0000000000..fe6099be74
--- /dev/null
+++ b/integrationTests/server/sse-finite-generator.test.ts
@@ -0,0 +1,230 @@
+/**
+ * QA-537 — regression verify for #1628 "SSE hang on a finite generator streamed to completion",
+ * fixed by PR #1632. `transformIterable` (server/serverHelpers/contentTypes.ts) applied the SSE
+ * `serialize` transform to a generator's terminal `{ value: undefined, done: true }` step, which
+ * threw inside Readable.from's pull loop and left the response hanging, never closed; the fix
+ * passes that step through untransformed.
+ *
+ * The arms vary where the terminal step falls relative to the yielded values — N=0 (it is the
+ * first step produced, the sharpest trigger), N=1, N=5, N=3000 — plus ThrowGen, whose rejection
+ * never reaches the terminal step and so exercises #1789's teardown instead.
+ *
+ * sse-throw-midstream.test.ts (#1789) anchors the throw path; stream-error-contract.test.ts pins
+ * the stream-error contract across SSE, NDJSON and iterable-REST on raw socket bytes. Neither
+ * covers the completion shapes above.
+ *
+ * Reproduction:
+ *   npm run test:integration -- "integrationTests/server/sse-finite-generator.test.ts"
+ */
+import { suite, test, before, after } from 'node:test';
+import { deepStrictEqual, ok, strictEqual } from 'node:assert';
+import { resolve } from 'node:path';
+import { setTimeout as sleep } from 'node:timers/promises';
+import { setupHarperWithFixture, teardownHarper, type ContextWithHarper } from '@harperfast/integration-testing';
+import {
+	awaitFixtureReady,
+	consumeSse,
+	countUncaught,
+	readLogOrThrow,
+	readLogSafe,
+	uncaughtAfterSettle,
+	uncaughtLines,
+	waitForProbe,
+} from '../utils/sseStream.ts';
+// @ts-expect-error utils/client.mjs has no type declarations; runtime resolves fine
+import { createApiClient } from '../apiTests/utils/client.mjs';
+
+const FIXTURE_PATH = resolve(import.meta.dirname, 'sse-finite-generator');
+const skipSuite = process.platform === 'win32';
+
+type Client = ReturnType<typeof createApiClient>;
+
+interface Lifecycle {
+	opened: number;
+	closed: number;
+}
+
+interface ProbeSnap {
+	ok: boolean;
+	finite: Lifecycle;
+	empty: Lifecycle;
+	single: Lifecycle;
+	throwGen: Lifecycle;
+	large: Lifecycle;
+}
+
+suite(
+	'QA-537 SSE finite-generator completion regression verify (#1628 / PR #1632)',
+	{ skip: skipSuite },
+	(ctx: ContextWithHarper) => {
+		let client: Client;
+		let restBase = '';
+		let authHeaders: Record<string, string> = {};
+		let logPath = '';
+		let uncaughtBaseline = 0;
+
+		before(async () => {
+			await setupHarperWithFixture(ctx, FIXTURE_PATH, {
+				config: { threads: { count: 1 }, logging: { console: true, level: 'error' } },
+				env: {},
+			});
+			client = createApiClient(ctx.harper);
+			restBase = client.restURL;
+			authHeaders = { Authorization: client.headers.Authorization as string };
+			({ logPath, uncaughtBaseline } = await awaitFixtureReady(ctx.harper as any, restBase, authHeaders));
+		});
+
+		after(async () => {
+			await teardownHarper(ctx);
+		});
+
+		/** The `n` carried by each delivered event, in arrival order. */
+		function eventNumbers(result: { events: string[] }): number[] {
+			return result.events.map((event) => JSON.parse(event).n);
+		}
+
+		async function assertCompletes(path: string, expectedEvents: number) {
+			const uncaughtBefore = countUncaught(readLogSafe(logPath));
+			const r = await consumeSse(`${restBase}${path}`, authHeaders, 15_000);
+			ok(r.status >= 200 && r.status < 300, `expected 2xx, got ${r.status} (errored=${r.errored?.message ?? null})`);
+			ok(
+				!r.aborted,
+				`${path} must not hit the AbortController timeout — a timeout here means the #1628 hang regressed. raw:\n${r.raw}`
+			);
+			strictEqual(
+				r.events.length,
+				expectedEvents,
+				`expected ${expectedEvents} SSE data events, got ${r.events.length}`
+			);
+			strictEqual(
+				r.terminatedBy,
+				'end',
+				`a generator streamed to completion must close via a clean 'end', got terminatedBy=${r.terminatedBy} errored=${r.errored?.message ?? null}`
+			);
+			strictEqual(
+				(await uncaughtAfterSettle(logPath)) - uncaughtBefore,
+				0,
+				`no NEW uncaughtException should be logged for ${path}`
+			);
+			return r;
+		}
+
+		// ── 1: empty generator (N=0) — the terminal step is the very first step ──────────────
+
+		test('1: EmptyGen (N=0) over SSE — zero events, response still closes cleanly', { timeout: 20_000 }, async () => {
+			await assertCompletes('/EmptyGen/', 0);
+		});
+
+		// ── 2: single-event generator (N=1) ──────────────────────────────────────────────────
+
+		test('2: SingleGen (N=1) over SSE — exactly 1 event, response closes cleanly', { timeout: 20_000 }, async () => {
+			const r = await assertCompletes('/SingleGen/', 1);
+			deepStrictEqual(eventNumbers(r), [0]);
+		});
+
+		// ── 3: canonical finite generator (N=5) ──────────────────────────────────────────────
+
+		test(
+			'3: FiniteGen (N=5) over SSE — all 5 events arrive and the response closes cleanly',
+			{ timeout: 20_000 },
+			async () => {
+				const r = await assertCompletes('/FiniteGen/', 5);
+				deepStrictEqual(eventNumbers(r), [0, 1, 2, 3, 4]);
+			}
+		);
+
+		// ── 4: large finite generator (N=3000) ───────────────────────────────────────────────
+
+		test(
+			'4: LargeGen (N=3000) over SSE — all 3000 events arrive and the response closes cleanly',
+			{ timeout: 20_000 },
+			async () => {
+				const r = await assertCompletes('/LargeGen/', 3000);
+				// Every index, not just the endpoints: a reordering or duplication in the middle of a long
+				// stream preserves both the count and the ends.
+				deepStrictEqual(
+					eventNumbers(r),
+					Array.from({ length: 3000 }, (_, i) => i)
+				);
+			}
+		);
+
+		// ── 5: generator that throws partway (2 of 5) ────────────────────────────────────────
+
+		test(
+			'5: ThrowGen (throws after 2 of 5) over SSE — terminates in bounded time, no uncaughtException',
+			{ timeout: 20_000 },
+			async () => {
+				const uncaughtBefore = countUncaught(readLogSafe(logPath));
+
+				const r = await consumeSse(`${restBase}/ThrowGen/`, authHeaders, 15_000);
+				console.log(
+					`[QA-537][5] ThrowGen: status=${r.status} events=${r.events.length} terminatedBy=${r.terminatedBy} aborted=${r.aborted} errored=${r.errored?.message ?? null} elapsedMs=${r.elapsedMs}`
+				);
+
+				// A rejecting generator exits via #1789's pipeline() teardown, which destroys the response
+				// instead of ending it cleanly. Only boundedness is asserted here; the throw contract
+				// itself is sse-throw-midstream.test.ts's.
+				ok(!r.aborted, `must not hit the AbortController timeout — the response never terminated. raw:\n${r.raw}`);
+				ok(r.terminatedBy !== null, 'response must terminate via end/error/close, not hang indefinitely');
+				// The generator yields 2 before it throws. The upper bound is the contract; the lower bound
+				// is 1 rather than 2 because the abrupt destroy can drop the last chunk before it drains,
+				// and a prefix of zero events would mean the pre-error events were never delivered at all.
+				const delivered = eventNumbers(r);
+				ok(
+					delivered.length >= 1 && delivered.length <= 2,
+					`expected a 1-2 event prefix before the throw, got ${delivered.length}. raw:\n${r.raw}`
+				);
+				deepStrictEqual(delivered, [0, 1].slice(0, delivered.length));
+				strictEqual(
+					(await uncaughtAfterSettle(logPath)) - uncaughtBefore,
+					0,
+					'no NEW uncaughtException should be logged for a mid-stream throw'
+				);
+			}
+		);
+
+		// ── Z: liveness canary + whole-suite uncaughtException sweep ────────────────────────
+
+		test(
+			'Z: liveness canary — worker survived every streaming shape above, no new uncaughtException',
+			{ timeout: 30_000 },
+			async () => {
+				const probe = await waitForProbe<ProbeSnap>(restBase, authHeaders, (snapshot) =>
+					[snapshot.finite, snapshot.empty, snapshot.single, snapshot.throwGen, snapshot.large].every(
+						(counter) => counter.opened >= 1 && counter.opened === counter.closed
+					)
+				);
+				console.log(`[QA-537][Z] liveness probe: ${probe ? JSON.stringify(probe) : 'DEAD'}`);
+				ok(probe !== null, 'Harper must still respond to Probe/ after all streaming cases');
+				ok(
+					probe!.finite && probe!.empty && probe!.single && probe!.throwGen && probe!.large,
+					`Probe/ returned no lifecycle counters: ${JSON.stringify(probe)}`
+				);
+				for (const [name, counter] of Object.entries({
+					FiniteGen: probe!.finite,
+					EmptyGen: probe!.empty,
+					SingleGen: probe!.single,
+					ThrowGen: probe!.throwGen,
+					LargeGen: probe!.large,
+				})) {
+					ok(counter.opened >= 1, `${name} should have been opened`);
+					strictEqual(counter.closed, counter.opened, `${name} generator was not closed`);
+				}
+
+				// Measured against the baseline taken in before(), so a case whose own delta check was
+				// outrun by the log flush is still caught here.
+				await sleep(1_000);
+				const offenders = uncaughtLines(readLogOrThrow(logPath));
+				if (offenders.length > uncaughtBaseline) {
+					console.log(`[QA-537][Z] NEW uncaughtException lines:\n${offenders.slice(uncaughtBaseline).join('\n')}`);
+				}
+				strictEqual(
+					offenders.length - uncaughtBaseline,
+					0,
+					'no uncaughtException should have appeared anywhere across the suite'
+				);
+			}
+		);
+	}
+);
diff --git a/integrationTests/server/sse-finite-generator/config.yaml b/integrationTests/server/sse-finite-generator/config.yaml
new file mode 100644
index 0000000000..efffc0833f
--- /dev/null
+++ b/integrationTests/server/sse-finite-generator/config.yaml
@@ -0,0 +1,5 @@
+graphqlSchema:
+  files: '*.graphql'
+jsResource:
+  files: resources.js
+rest: true
diff --git a/integrationTests/server/sse-finite-generator/resources.js b/integrationTests/server/sse-finite-generator/resources.js
new file mode 100644
index 0000000000..41271ec854
--- /dev/null
+++ b/integrationTests/server/sse-finite-generator/resources.js
@@ -0,0 +1,99 @@
+// Fixture for sse-finite-generator.test.ts (QA-537 / #1628) — finite async generators streamed to
+// completion over SSE, plus a rejecting contrast arm. The test file carries the bug recap.
+//
+// SSE requests reach these via `connect()`: server/REST.ts turns an `Accept: text/event-stream`
+// GET into method CONNECT, which calls `resource.connect(...)`.
+
+const G = (globalThis.__QA537__ ??= {
+	finite: { opened: 0, closed: 0 },
+	empty: { opened: 0, closed: 0 },
+	single: { opened: 0, closed: 0 },
+	throwGen: { opened: 0, closed: 0 },
+	large: { opened: 0, closed: 0 },
+});
+
+function sleep(ms) {
+	return new Promise((r) => setTimeout(r, ms));
+}
+
+export class FiniteGen extends Resource {
+	static loadAsInstance = false;
+	static async *connect(_target, _incomingMessages, _request) {
+		G.finite.opened++;
+		try {
+			for (let i = 0; i < 5; i++) {
+				yield { n: i };
+				await sleep(2);
+			}
+		} finally {
+			G.finite.closed++;
+		}
+	}
+}
+
+// 0 events, so the terminal `done` step is the very first step produced.
+export class EmptyGen extends Resource {
+	static loadAsInstance = false;
+	// eslint-disable-next-line require-yield
+	static async *connect(_target, _incomingMessages, _request) {
+		G.empty.opened++;
+		try {
+			// intentionally yields nothing
+		} finally {
+			G.empty.closed++;
+		}
+	}
+}
+
+export class SingleGen extends Resource {
+	static loadAsInstance = false;
+	static async *connect(_target, _incomingMessages, _request) {
+		G.single.opened++;
+		try {
+			yield { n: 0 };
+		} finally {
+			G.single.closed++;
+		}
+	}
+}
+
+// A rejecting iterator never reaches a `done:true` step, so this is the contrast arm rather than
+// the fixed code path.
+export class ThrowGen extends Resource {
+	static loadAsInstance = false;
+	static async *connect(_target, _incomingMessages, _request) {
+		G.throwGen.opened++;
+		try {
+			for (let i = 0; i < 5; i++) {
+				if (i === 2) throw new Error('QA537-intentional-throw-partway');
+				yield { n: i };
+				await sleep(2);
+			}
+		} finally {
+			G.throwGen.closed++;
+		}
+	}
+}
+
+// The same terminal-step path at volume.
+export class LargeGen extends Resource {
+	static loadAsInstance = false;
+	static async *connect(_target, _incomingMessages, _request) {
+		G.large.opened++;
+		try {
+			for (let i = 0; i < 3000; i++) {
+				yield { n: i };
+			}
+		} finally {
+			G.large.closed++;
+		}
+	}
+}
+
+// Readiness + lifecycle-counter snapshot, served as plain JSON rather than SSE.
+export class Probe extends Resource {
+	static loadAsInstance = false;
+	static async get() {
+		return { ok: true, ...G };
+	}
+}
diff --git a/integrationTests/server/sse-finite-generator/schema.graphql b/integrationTests/server/sse-finite-generator/schema.graphql
new file mode 100644
index 0000000000..151a6c18ef
--- /dev/null
+++ b/integrationTests/server/sse-finite-generator/schema.graphql
@@ -0,0 +1,5 @@
+# A jsResource fixture needs at least one @table/@export type to be a valid component; the
+# generator Resources under test need no table backing.
+type Placeholder @table @export {
+	id: ID @primaryKey
+}
diff --git a/integrationTests/server/sse-throw-midstream.test.ts b/integrationTests/server/sse-throw-midstream.test.ts
index 7a7d9ec246..a1af2b4d9e 100644
--- a/integrationTests/server/sse-throw-midstream.test.ts
+++ b/integrationTests/server/sse-throw-midstream.test.ts
@@ -23,8 +23,8 @@
  * test in contentTypes.test.js) in bounded time, receiving only the events flushed before the
  * throw, with NO uncaughtException logged and the worker still alive afterward.
  *
- * This suite starts from the QA-537 harness/fixture pattern (qa-scratch/qa537-sse-finite-
- * generator/) and its `ThrowGen` case, which *documented* (did not assert) the pre-fix hang.
+ * This suite starts from the QA-537 harness/fixture pattern (sse-finite-generator.test.ts, which
+ * anchors #1628) and its `ThrowGen` case, which *documented* (did not assert) the pre-fix hang.
  * Here the throw cases are promoted to hard assertions of clean termination:
  *   ThrowFirst - throws on the very first step, before any bytes are yielded.
  *   ThrowMid   - yields 3 of an intended 6 events, then throws (genuine mid-stream failure).
@@ -44,14 +44,19 @@
  */
 import { suite, test, before, after } from 'node:test';
 import { ok, strictEqual } from 'node:assert';
-import { resolve, join } from 'node:path';
-import { readFileSync } from 'node:fs';
-import { StringDecoder } from 'node:string_decoder';
+import { resolve } from 'node:path';
 import { setTimeout as sleep } from 'node:timers/promises';
-import http from 'node:http';
-import https from 'node:https';
-import { URL } from 'node:url';
 import { setupHarperWithFixture, teardownHarper, type ContextWithHarper } from '@harperfast/integration-testing';
+import {
+	awaitFixtureReady,
+	consumeSse,
+	countUncaught,
+	readLogOrThrow,
+	readLogSafe,
+	uncaughtAfterSettle,
+	uncaughtLines,
+	waitForProbe,
+} from '../utils/sseStream.ts';
 // @ts-expect-error utils/client.mjs has no type declarations; runtime resolves fine
 import { createApiClient } from '../apiTests/utils/client.mjs';
 
@@ -67,170 +72,6 @@ interface ProbeSnap {
 	clean: { opened: number; closed: number };
 }
 
-async function getProbe(restBase: string, authHeaders: Record<string, string>): Promise<ProbeSnap> {
-	const url = new URL(`${restBase}/Probe/`);
-	const lib = url.protocol === 'https:' ? https : http;
-	return new Promise((resolvePromise, reject) => {
-		const req = lib.request(
-			url,
-			{
-				method: 'GET',
-				headers: { ...authHeaders, Accept: 'application/json' },
-				rejectUnauthorized: false,
-				signal: AbortSignal.timeout(3_000),
-			} as any,
-			(res) => {
-				const chunks: Buffer[] = [];
-				res.on('data', (d: Buffer) => chunks.push(d));
-				res.on('end', () => {
-					try {
-						resolvePromise(JSON.parse(Buffer.concat(chunks).toString('utf8')));
-					} catch (e) {
-						reject(e);
-					}
-				});
-				res.on('error', reject);
-			}
-		);
-		req.on('error', reject);
-		req.end();
-	});
-}
-
-async function waitForProbe(
-	restBase: string,
-	authHeaders: Record<string, string>,
-	predicate: (probe: ProbeSnap) => boolean,
-	timeoutMs = 5_000
-): Promise<ProbeSnap | null> {
-	const deadline = Date.now() + timeoutMs;
-	let probe: ProbeSnap | null = null;
-	while (Date.now() < deadline) {
-		probe = await getProbe(restBase, authHeaders).catch(() => null);
-		if (probe && predicate(probe)) return probe;
-		await sleep(50);
-	}
-	return probe;
-}
-
-// ── AbortController-bounded SSE stream consumer ──────────────────────────────────────────
-// Wraps the request in an AbortController with a generous but bounded timeout, so a genuine
-// regression (hung response) shows up as a caught timeout/abort in the test, never a hung
-// process. Resolves on whichever of 'end' / 'error' / 'close' fires first, since the fixed
-// pipeline()-based teardown closes the response abruptly (not via a clean 'end') on a source
-// error -- see PR #1789.
-
-interface SseResult {
-	status: number;
-	raw: string;
-	events: string[]; // parsed `data: ...` payloads, in order
-	terminatedBy: 'end' | 'error' | 'close' | null; // which event resolved the request
-	aborted: boolean; // did our own timeout abort fire (i.e. it hung)?
-	errored: Error | null;
-	elapsedMs: number;
-}
-
-function consumeSse(urlStr: string, authHeaders: Record<string, string>, timeoutMs = 12_000): Promise<SseResult> {
-	const url = new URL(urlStr);
-	const lib = url.protocol === 'https:' ? https : http;
-	const controller = new AbortController();
-	let timedOut = false;
-	const start = Date.now();
-	const timer = setTimeout(() => {
-		timedOut = true;
-		controller.abort();
-	}, timeoutMs);
-
-	return new Promise((resolvePromise) => {
-		const result: SseResult = {
-			status: 0,
-			raw: '',
-			events: [],
-			terminatedBy: null,
-			aborted: false,
-			errored: null,
-			elapsedMs: 0,
-		};
-		let settled = false;
-		const finish = (terminatedBy: SseResult['terminatedBy'], err?: Error) => {
-			if (settled) return;
-			settled = true;
-			clearTimeout(timer);
-			result.terminatedBy = terminatedBy;
-			result.errored = err ?? null;
-			result.elapsedMs = Date.now() - start;
-			result.events = result.raw
-				.split('\n')
-				.filter((l) => l.startsWith('data: '))
-				.map((l) => l.slice('data: '.length));
-			resolvePromise(result);
-		};
-		const req = lib.request(
-			url,
-			{
-				method: 'GET',
-				headers: { ...authHeaders, Accept: 'text/event-stream' },
-				rejectUnauthorized: false,
-				signal: controller.signal,
-			} as any,
-			(res) => {
-				result.status = res.statusCode ?? 0;
-				// Decode incrementally: a bare d.toString('utf8') per chunk corrupts any multi-byte
-				// character that straddles a TCP chunk boundary into U+FFFD.
-				const decoder = new StringDecoder('utf8');
-				res.on('data', (d: Buffer) => {
-					result.raw += decoder.write(d);
-				});
-				res.on('end', () => {
-					result.raw += decoder.end();
-					finish('end');
-				});
-				res.on('error', (e: Error) => finish('error', e));
-				res.on('close', () => finish('close'));
-			}
-		);
-		req.on('error', (e: any) => {
-			if (timedOut || e?.name === 'AbortError') {
-				if (settled) return;
-				settled = true;
-				clearTimeout(timer);
-				result.aborted = true;
-				result.elapsedMs = Date.now() - start;
-				result.events = result.raw
-					.split('\n')
-					.filter((l) => l.startsWith('data: '))
-					.map((l) => l.slice('data: '.length));
-				resolvePromise(result);
-			} else {
-				finish('error', e);
-			}
-		});
-		req.end();
-	});
-}
-
-function readLogSafe(logPath: string): string {
-	try {
-		return readFileSync(logPath, 'utf8');
-	} catch {
-		return '';
-	}
-}
-
-function countUncaught(log: string): number {
-	return log.split('\n').filter((l) => l.includes('uncaughtException')).length;
-}
-
-/**
- * Asserting a NON-event: the worker logs an uncaughtException asynchronously, so reading hdb.log
- * the instant the response closes can pass vacuously by simply outrunning the flush. Give the
- * writer a bounded settle first — one of the cases AGENTS.md reserves a fixed sleep for.
- */
-async function uncaughtAfterSettle(logPath: string): Promise<number> {
-	await sleep(1_000);
-	return countUncaught(readLogSafe(logPath));
-}
-
 // ── Suite ──────────────────────────────────────────────────────────────────────────────────
 
 suite(
@@ -241,6 +82,7 @@ suite(
 		let restBase = '';
 		let authHeaders: Record<string, string> = {};
 		let logPath = '';
+		let uncaughtBaseline = 0;
 
 		before(async () => {
 			await setupHarperWithFixture(ctx, FIXTURE_PATH, {
@@ -250,26 +92,7 @@ suite(
 			client = createApiClient(ctx.harper);
 			restBase = client.restURL;
 			authHeaders = { Authorization: client.headers.Authorization as string };
-			logPath = (ctx.harper as any).logDir
-				? join((ctx.harper as any).logDir, 'hdb.log')
-				: join((ctx.harper as any).dataRootDir, 'log', 'hdb.log');
-
-			const deadline = Date.now() + 30_000;
-			let ready = false;
-			while (Date.now() < deadline) {
-				try {
-					const p = await getProbe(restBase, authHeaders);
-					if (p?.ok !== undefined) {
-						ready = true;
-						break;
-					}
-				} catch {
-					/* not ready */
-				}
-				await sleep(250);
-			}
-			ok(ready, 'Probe route did not become ready within 30 seconds');
-			readFileSync(logPath, 'utf8');
+			({ logPath, uncaughtBaseline } = await awaitFixtureReady(ctx.harper as any, restBase, authHeaders));
 		});
 
 		after(async () => {
@@ -301,8 +124,15 @@ suite(
 					`expected 0 events (throw before any yield), got ${r.events.length}. raw:\n${r.raw}`
 				);
 
-				const probe = await waitForProbe(restBase, authHeaders, (snapshot) => snapshot.throwFirst.closed >= 1);
-				ok(probe && probe.throwFirst.closed >= 1, 'ThrowFirst generator should have completed cleanup');
+				const probe = await waitForProbe<ProbeSnap>(
+					restBase,
+					authHeaders,
+					(snapshot) => snapshot.throwFirst.closed >= 1
+				);
+				ok(
+					(probe?.throwFirst?.closed ?? 0) >= 1,
+					`ThrowFirst generator should have completed cleanup; probe: ${JSON.stringify(probe)}`
+				);
 				const uncaughtAfter = await uncaughtAfterSettle(logPath);
 				strictEqual(
 					uncaughtAfter - uncaughtBefore,
@@ -339,8 +169,11 @@ suite(
 					ok(r.events[i].includes(`"n":${i}`), `expected event ${i} to contain n=${i}, got: ${r.events[i]}`);
 				}
 
-				const probe = await waitForProbe(restBase, authHeaders, (snapshot) => snapshot.throwMid.closed >= 1);
-				ok(probe && probe.throwMid.closed >= 1, 'ThrowMid generator should have completed cleanup');
+				const probe = await waitForProbe<ProbeSnap>(restBase, authHeaders, (snapshot) => snapshot.throwMid.closed >= 1);
+				ok(
+					(probe?.throwMid?.closed ?? 0) >= 1,
+					`ThrowMid generator should have completed cleanup; probe: ${JSON.stringify(probe)}`
+				);
 				const uncaughtAfter = await uncaughtAfterSettle(logPath);
 				strictEqual(
 					uncaughtAfter - uncaughtBefore,
@@ -377,8 +210,7 @@ suite(
 			'Z: liveness canary -- server survived all throw-mid-stream probes, no new uncaughtException',
 			{ timeout: 30_000 },
 			async () => {
-				const logBefore = readLogSafe(logPath);
-				const p = await waitForProbe(
+				const p = await waitForProbe<ProbeSnap>(
 					restBase,
 					authHeaders,
 					(snapshot) => snapshot.throwFirst.closed >= 1 && snapshot.throwMid.closed >= 1 && snapshot.clean.closed >= 1
@@ -388,6 +220,7 @@ suite(
 					p !== null,
 					'Harper must still respond to Probe/ after all throw-mid-stream cases (worker not crashed/wedged)'
 				);
+				ok(p!.throwFirst && p!.throwMid && p!.clean, `Probe/ returned no lifecycle counters: ${JSON.stringify(p)}`);
 				ok(
 					p!.throwFirst.opened >= 1 && p!.throwFirst.closed >= 1,
 					'ThrowFirst should show a matched open/close pair (generator finally ran)'
@@ -398,19 +231,18 @@ suite(
 				);
 				ok(p!.clean.opened >= 1 && p!.clean.closed >= 1, 'CleanGen should show a matched open/close pair');
 
-				await sleep(1_000); // same non-event settle as the per-case sweeps above
-				const logAfter = readLogSafe(logPath);
-				const newLines = logAfter.slice(logBefore.length);
-				const newUncaught = newLines.split('\n').filter((l) => l.includes('uncaughtException')).length;
-				if (newUncaught > 0) {
-					console.log(
-						`[QA-559][Z] NEW uncaughtException lines found:\n${newLines
-							.split('\n')
-							.filter((l) => l.includes('uncaughtException'))
-							.join('\n')}`
-					);
+				// Measured against the baseline taken in before(), so a case whose own delta check was
+				// outrun by the log flush is still caught here.
+				await sleep(1_000);
+				const offenders = uncaughtLines(readLogOrThrow(logPath));
+				if (offenders.length > uncaughtBaseline) {
+					console.log(`[QA-559][Z] NEW uncaughtException lines:\n${offenders.slice(uncaughtBaseline).join('\n')}`);
 				}
-				strictEqual(newUncaught, 0, 'no uncaughtException should have appeared anywhere across the whole suite');
+				strictEqual(
+					offenders.length - uncaughtBaseline,
+					0,
+					'no uncaughtException should have appeared anywhere across the whole suite'
+				);
 			}
 		);
 	}
diff --git a/integrationTests/utils/sseStream.ts b/integrationTests/utils/sseStream.ts
new file mode 100644
index 0000000000..f291ce0c73
--- /dev/null
+++ b/integrationTests/utils/sseStream.ts
@@ -0,0 +1,244 @@
+/**
+ * Shared helpers for the SSE streaming regression suites in `integrationTests/server/`
+ * (`sse-finite-generator.test.ts`, `sse-throw-midstream.test.ts`): a bounded SSE consumer, a
+ * `/Probe/` reader for the lifecycle counters those fixtures keep, and an `hdb.log`
+ * uncaughtException counter — the bugs these suites anchor surfaced as an uncaught throw inside
+ * the worker rather than as a bad response.
+ *
+ * The raw-socket capture in `stream-error-contract.test.ts` is a deliberately different
+ * technique (it inspects chunk framing and the exact close mechanism) and is not shared.
+ */
+import { readFileSync } from 'node:fs';
+import { join } from 'node:path';
+import { StringDecoder } from 'node:string_decoder';
+import { setTimeout as sleep } from 'node:timers/promises';
+import http from 'node:http';
+import https from 'node:https';
+import { URL } from 'node:url';
+
+export interface SseResult {
+	status: number;
+	raw: string;
+	events: string[];
+	/** which response event resolved the request; null when our own timeout fired instead */
+	terminatedBy: 'end' | 'error' | 'close' | null;
+	/** did our own timeout abort fire — i.e. the response hung? */
+	aborted: boolean;
+	errored: Error | null;
+	elapsedMs: number;
+}
+
+function parseEvents(raw: string): string[] {
+	return raw
+		.split('\n')
+		.filter((line) => line.startsWith('data: '))
+		.map((line) => line.slice('data: '.length));
+}
+
+/**
+ * Consume an SSE response, bounded by an AbortController.
+ *
+ * Resolves on whichever of 'end' / 'error' / 'close' fires first: the pipeline()-based teardown
+ * introduced by #1789 closes the response abruptly rather than via a clean 'end' when the source
+ * generator rejects, so waiting only for 'end' would read a correct abrupt close as a hang.
+ *
+ * `aborted` is decided by our own timer rather than by which event won the race to settle. An
+ * abort destroys the response, so a hung stream emits 'close' too — attributing the settle to
+ * that event would report a genuine hang as a bounded termination.
+ */
+export function consumeSse(
+	urlStr: string,
+	authHeaders: Record<string, string>,
+	timeoutMs = 12_000
+): Promise<SseResult> {
+	const url = new URL(urlStr);
+	const lib = url.protocol === 'https:' ? https : http;
+	const controller = new AbortController();
+	let timedOut = false;
+	const start = Date.now();
+	const timer = setTimeout(() => {
+		timedOut = true;
+		controller.abort();
+	}, timeoutMs);
+
+	return new Promise((resolvePromise) => {
+		const result: SseResult = {
+			status: 0,
+			raw: '',
+			events: [],
+			terminatedBy: null,
+			aborted: false,
+			errored: null,
+			elapsedMs: 0,
+		};
+		let settled = false;
+		let flushDecoder: (() => string) | null = null;
+		const finish = (terminatedBy: SseResult['terminatedBy'], err?: Error) => {
+			if (settled) return;
+			settled = true;
+			clearTimeout(timer);
+			// An abrupt close settles through 'error'/'close', so flushing only on 'end' would drop a
+			// multi-byte character the decoder was still holding.
+			if (flushDecoder) result.raw += flushDecoder();
+			result.aborted = timedOut;
+			result.terminatedBy = timedOut ? null : terminatedBy;
+			result.errored = err ?? null;
+			result.elapsedMs = Date.now() - start;
+			result.events = parseEvents(result.raw);
+			resolvePromise(result);
+		};
+		const req = lib.request(
+			url,
+			{
+				method: 'GET',
+				headers: { ...authHeaders, Accept: 'text/event-stream' },
+				rejectUnauthorized: false,
+				signal: controller.signal,
+			} as any,
+			(res) => {
+				result.status = res.statusCode ?? 0;
+				// A bare d.toString('utf8') per chunk corrupts any multi-byte character that straddles
+				// a TCP chunk boundary into U+FFFD.
+				const decoder = new StringDecoder('utf8');
+				flushDecoder = () => decoder.end();
+				res.on('data', (d: Buffer) => {
+					result.raw += decoder.write(d);
+				});
+				res.on('end', () => finish('end'));
+				res.on('error', (e: Error) => finish('error', e));
+				res.on('close', () => finish('close'));
+			}
+		);
+		req.on('error', (e: any) => {
+			if (timedOut) finish(null);
+			else finish('error', e);
+		});
+		req.end();
+	});
+}
+
+export function getProbe<T>(restBase: string, authHeaders: Record<string, string>, timeoutMs = 3_000): Promise<T> {
+	const url = new URL(`${restBase}/Probe/`);
+	const lib = url.protocol === 'https:' ? https : http;
+	return new Promise((resolvePromise, reject) => {
+		const req = lib.request(
+			url,
+			{
+				method: 'GET',
+				headers: { ...authHeaders, Accept: 'application/json' },
+				rejectUnauthorized: false,
+				signal: AbortSignal.timeout(timeoutMs),
+			} as any,
+			(res) => {
+				const chunks: Buffer[] = [];
+				res.on('data', (d: Buffer) => chunks.push(d));
+				res.on('end', () => {
+					try {
+						resolvePromise(JSON.parse(Buffer.concat(chunks).toString('utf8')));
+					} catch (e) {
+						reject(e);
+					}
+				});
+				res.on('error', reject);
+			}
+		);
+		req.on('error', reject);
+		req.end();
+	});
+}
+
+/**
+ * Poll `/Probe/` until `predicate` holds. A generator's `finally` block runs after the response
+ * has already terminated, so its close counter is not readable the instant the client resolves.
+ * Returns the last snapshot read (which may not satisfy the predicate) or null if none was.
+ */
+export async function waitForProbe<T>(
+	restBase: string,
+	authHeaders: Record<string, string>,
+	predicate: (probe: T) => boolean,
+	timeoutMs = 5_000
+): Promise<T | null> {
+	const deadline = Date.now() + timeoutMs;
+	let probe: T | null = null;
+	while (Date.now() < deadline) {
+		probe = await getProbe<T>(restBase, authHeaders, Math.max(1, deadline - Date.now())).catch(() => null);
+		// A predicate reading counter fields off an error payload throws; that is "not satisfied
+		// yet", and the caller asserts on the snapshot this returns.
+		try {
+			if (probe && predicate(probe)) return probe;
+		} catch {
+			/* not satisfied */
+		}
+		await sleep(50);
+	}
+	return probe;
+}
+
+/**
+ * Wait for a freshly installed SSE fixture's `/Probe/` route, and take the suite's baseline
+ * uncaughtException count.
+ *
+ * The baseline is read with `readFileSync`, not `readLogSafe`: an hdb.log this suite cannot read
+ * (a changed harness layout, a log not yet created) would otherwise turn every uncaughtException
+ * assertion downstream into a vacuous pass, since a missing file counts zero both before and
+ * after. Failing setup is the only way that stays visible.
+ */
+export async function awaitFixtureReady(
+	harper: { logDir?: string; dataRootDir?: string },
+	restBase: string,
+	authHeaders: Record<string, string>,
+	timeoutMs = 30_000
+): Promise<{ logPath: string; uncaughtBaseline: number }> {
+	const logPath = harper.logDir ? join(harper.logDir, 'hdb.log') : join(harper.dataRootDir as string, 'log', 'hdb.log');
+	const deadline = Date.now() + timeoutMs;
+	let ready = false;
+	while (Date.now() < deadline) {
+		const probe = await getProbe<{ ok?: boolean }>(restBase, authHeaders).catch(() => null);
+		if (probe?.ok !== undefined) {
+			ready = true;
+			break;
+		}
+		await sleep(250);
+	}
+	if (!ready) throw new Error(`Probe route did not become ready within ${timeoutMs}ms at ${restBase}/Probe/`);
+	// Polled, not read once: the HTTP port can answer before the log writer has created the file.
+	while (Date.now() < deadline) {
+		try {
+			return { logPath, uncaughtBaseline: countUncaught(readFileSync(logPath, 'utf8')) };
+		} catch {
+			await sleep(100);
+		}
+	}
+	throw new Error(`hdb.log never became readable at ${logPath} — uncaughtException checks would pass vacuously`);
+}
+
+/** For the end-of-suite sweep: a log that vanished mid-run must fail, not silently count zero. */
+export function readLogOrThrow(logPath: string): string {
+	return readFileSync(logPath, 'utf8');
+}
+
+export function readLogSafe(logPath: string): string {
+	try {
+		return readFileSync(logPath, 'utf8');
+	} catch {
+		return '';
+	}
+}
+
+export function uncaughtLines(log: string): string[] {
+	return log.split('\n').filter((line) => line.includes('uncaughtException'));
+}
+
+export function countUncaught(log: string): number {
+	return uncaughtLines(log).length;
+}
+
+/**
+ * Asserting a NON-event: the worker logs an uncaughtException asynchronously, so reading hdb.log
+ * the instant the response closes can pass vacuously by simply outrunning the flush. Give the
+ * writer a bounded settle first — one of the cases AGENTS.md reserves a fixed sleep for.
+ */
+export async function uncaughtAfterSettle(logPath: string): Promise<number> {
+	await sleep(1_000);
+	return countUncaught(readLogSafe(logPath));
+}