From 24d6e657339055610792a749c9ec7a9c90494001 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Tue, 27 Oct 2026 14:58:59 +0800
Subject: [PATCH] test(authority): bound Python read input independently of
 pipe EOF

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../control_plane_ts/python_read_consumer.ts  | 26 +++++++++++++++++++
 .../succession_read_conformance.ts            |  8 ++----
 .../todo_consumer_scope_conformance.ts        |  8 ++----
 3 files changed, 30 insertions(+), 12 deletions(-)
 create mode 100644 tests/control_plane_ts/python_read_consumer.ts

diff --git a/tests/control_plane_ts/python_read_consumer.ts b/tests/control_plane_ts/python_read_consumer.ts
new file mode 100644
index 0000000000..36f4000c28
--- /dev/null
+++ b/tests/control_plane_ts/python_read_consumer.ts
@@ -0,0 +1,26 @@
+import {spawnSync} from "node:child_process";
+import {closeSync, mkdtempSync, openSync, rmSync, writeFileSync} from "node:fs";
+import {tmpdir} from "node:os";
+import {join} from "node:path";
+import {resolveTestPython} from "../../scripts/test-python.mjs";
+
+// Keep the complete JSON on stdin, with a finite source independent of pipe EOF.
+// Large synchronous pipe writes can strand the consumer in stdin.read() under
+// the concurrent conformance suite before any product operation is reached.
+export function runPythonReadConsumer(consumer: string, source: unknown) {
+  const directory = mkdtempSync(join(tmpdir(), "loopx-python-read-"));
+  try {
+    const path = join(directory, "source.json");
+    writeFileSync(path, JSON.stringify(source), {encoding: "utf8", mode: 0o600});
+    const input = openSync(path, "r");
+    try {
+      return spawnSync(resolveTestPython(), ["-c", consumer], {
+        encoding: "utf8", timeout: 90_000, stdio: [input, "pipe", "pipe"],
+      });
+    } finally {
+      closeSync(input);
+    }
+  } finally {
+    rmSync(directory, {recursive: true, force: true});
+  }
+}
diff --git a/tests/control_plane_ts/succession_read_conformance.ts b/tests/control_plane_ts/succession_read_conformance.ts
index 0a3ba8c09d..836c690501 100644
--- a/tests/control_plane_ts/succession_read_conformance.ts
+++ b/tests/control_plane_ts/succession_read_conformance.ts
@@ -1,12 +1,9 @@
 import assert from "node:assert/strict";
-import {spawnSync} from "node:child_process";
 import test from "node:test";
 import type {AuthorityStoreConformanceFactory} from "./authority_store_conformance.ts";
 import {productionScaleSuccessionFixture} from "./production_scale_coordination_fixture.ts";
 import {validateCoordinationTodoReadModel} from "../../loopx/control_plane/coordination/coordination_projection.ts";
-import {resolveTestPython} from "../../scripts/test-python.mjs";
-
-const PYTHON = resolveTestPython();
+import {runPythonReadConsumer} from "./python_read_consumer.ts";
 
 // Exercise the shipped Python consumer → typed policy, not a second test reducer.
 const CONSUMER = `
@@ -32,8 +29,7 @@ export function registerSuccessionReadConformance(name: string, factory: Authori
       next_projection: projection, events: [], receipts: []})).status, "applied");
     const before = await store.loadAuthority();
     assert.equal(before.status, "loaded");
-    const child = spawnSync(PYTHON, ["-c", CONSUMER], {encoding: "utf8", timeout: 90_000,
-      input: JSON.stringify({todos: before.head.todos, cases})});
+    const child = runPythonReadConsumer(CONSUMER, {todos: before.head.todos, cases});
     assert.equal(child.status, 0, child.stderr);
     const actual = JSON.parse(child.stdout);
     assert.deepEqual(actual.inferred_source, {gap: 0, gates: [], closure: false});
diff --git a/tests/control_plane_ts/todo_consumer_scope_conformance.ts b/tests/control_plane_ts/todo_consumer_scope_conformance.ts
index 58f4f6b7e6..fdb690e8ae 100644
--- a/tests/control_plane_ts/todo_consumer_scope_conformance.ts
+++ b/tests/control_plane_ts/todo_consumer_scope_conformance.ts
@@ -1,12 +1,9 @@
 import assert from "node:assert/strict";
-import {spawnSync} from "node:child_process";
 import test from "node:test";
 import type {JsonObject} from "../../loopx/control_plane/effect_program.ts";
 import type {AuthorityStoreConformanceFactory} from "./authority_store_conformance.ts";
 import {productionScaleConsumerScopeFixture} from "./production_scale_coordination_fixture.ts";
-import {resolveTestPython} from "../../scripts/test-python.mjs";
-
-const PYTHON = resolveTestPython();
+import {runPythonReadConsumer} from "./python_read_consumer.ts";
 
 // The actual Python read consumer still hosts rendering; policy runs in TS.
 const CONSUMER = `
@@ -45,8 +42,7 @@ export function registerTodoConsumerScopeConformance(name: string, factory: Auth
       next_projection: projection, events: [], receipts: []})).status, "applied");
     const before = await store.loadAuthority(); assert.equal(before.status, "loaded");
     if (before.status !== "loaded") return;
-    const child = spawnSync(PYTHON, ["-c", CONSUMER], {encoding: "utf8", timeout: 90_000,
-      input: JSON.stringify({todos: before.head.todos, cases})});
+    const child = runPythonReadConsumer(CONSUMER, {todos: before.head.todos, cases});
     assert.equal(child.status, 0, child.stderr);
     const result = JSON.parse(child.stdout);
     const expected = ["todo_scope_explicit_action", "todo_scope_explicit_gate", "todo_scope_global"];