test(#838): reject retry masking and dropped replies

packages/mosaic/package.json

@@ -25,7 +25,7 @@
     "lint": "eslint src",
     "typecheck": "tsc --noEmit",
     "test": "vitest run --passWithNoTests && pnpm run test:framework-shell",
-    "test:framework-shell": "python3 src/mutator-gate/runtime_tools_unittest.py && python3 src/mutator-gate/runtime_launch_guard_unittest.py && python3 framework/tools/lease-broker/check-runtime-launches.py --root ../.. && bash framework/tools/codex/test-pr-diff-context.sh"
+    "test:framework-shell": "python3 src/lease-broker/daemon_deadline_unittest.py && python3 src/mutator-gate/runtime_tools_unittest.py && python3 src/mutator-gate/runtime_launch_guard_unittest.py && python3 framework/tools/lease-broker/check-runtime-launches.py --root ../.. && bash framework/tools/codex/test-pr-diff-context.sh"
   },
   "dependencies": {
     "@mosaicstack/brain": "workspace:*",

packages/mosaic/src/lease-broker/broker-test-client.spec.ts

@@ -53,29 +53,39 @@ afterEach(async () => {
 });
 
 describe('newline-framed broker test client', () => {
-  test('retries an empty early close and parses the complete newline-terminated retry', async () => {
-    const broker = await scriptedBroker([[], [Buffer.from('{"ok":true'), Buffer.from('}\n')]]);
+  test('rejects an empty early close without retrying into a false green', async () => {
+    const broker = await scriptedBroker([[], [Buffer.from('{"ok":true}\n')]]);
 
-    await expect(requestBrokerReply(broker.socketPath, { action: 'probe' })).resolves.toEqual({
-      ok: true,
-    });
-    expect(broker.connections()).toBe(2);
-  });
-
-  test('rejects repeated truncated early closes with response bytes and length', async () => {
-    const truncated = Buffer.from('{"ok":');
-    const broker = await scriptedBroker([[truncated], [truncated]]);
-
-    const failure = await requestBrokerReply(
-      broker.socketPath,
-      { action: 'probe' },
-      { earlyCloseRetries: 1 },
-    ).catch((error: unknown) => error);
+    const failure = await requestBrokerReply(broker.socketPath, { action: 'probe' }).catch(
+      (error: unknown) => error,
+    );
 
     expect(failure).toBeInstanceOf(BrokerTransportError);
     expect(failure).toMatchObject({
       kind: 'early-close',
-      attempts: 2,
+      attempts: 1,
+      responseLength: 0,
+      responseHex: '',
+    });
+    expect(failure).toHaveProperty(
+      'message',
+      expect.stringMatching(/closed before newline.*length=0.*hex=<empty>/i),
+    );
+    expect(broker.connections()).toBe(1);
+  });
+
+  test('rejects repeated truncated early closes with response bytes and length', async () => {
+    const truncated = Buffer.from('{"ok":');
+    const broker = await scriptedBroker([[truncated], [Buffer.from('{"ok":true}\n')]]);
+
+    const failure = await requestBrokerReply(broker.socketPath, { action: 'probe' }).catch(
+      (error: unknown) => error,
+    );
+
+    expect(failure).toBeInstanceOf(BrokerTransportError);
+    expect(failure).toMatchObject({
+      kind: 'early-close',
+      attempts: 1,
       responseLength: 6,
       responseHex: '7b226f6b223a',
     });
@@ -83,7 +93,7 @@ describe('newline-framed broker test client', () => {
       'message',
       expect.stringMatching(/closed before newline.*length=6.*bytes=.*ok/i),
     );
-    expect(broker.connections()).toBe(2);
+    expect(broker.connections()).toBe(1);
   });
 
   test('rejects a newline-terminated malformed broker reply without retrying', async () => {

packages/mosaic/src/lease-broker/daemon_deadline_unittest.py

@@ -0,0 +1,72 @@
+#!/usr/bin/env python3
+"""Regression tests for bounded lease-broker read/handle/send deadlines."""
+
+from __future__ import annotations
+
+import importlib.util
+import json
+import socket
+import threading
+import time
+import unittest
+from pathlib import Path
+
+
+DAEMON_PATH = Path(__file__).parents[2] / "framework/tools/lease-broker/daemon.py"
+SPEC = importlib.util.spec_from_file_location("lease_broker_deadline_daemon", DAEMON_PATH)
+if SPEC is None or SPEC.loader is None:
+    raise RuntimeError("unable to load lease broker daemon")
+DAEMON = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(DAEMON)
+
+
+def original_connection_budget() -> float:
+    read_budget = getattr(DAEMON, "READ_DEADLINE_SECONDS", None)
+    if isinstance(read_budget, (int, float)):
+        return float(read_budget)
+    return float(DAEMON.CONNECTION_DEADLINE_SECONDS)
+
+
+class SlowBroker:
+    def __init__(self, delay: float) -> None:
+        self.delay = delay
+        self.calls = 0
+
+    def handle(self, _peer: tuple[int, int, int], request: dict[str, object]) -> dict[str, object]:
+        self.calls += 1
+        time.sleep(self.delay)
+        return {"ok": True, "echo": request.get("action")}
+
+
+class BrokerDeadlineTest(unittest.TestCase):
+    def test_completed_slow_handle_gets_a_complete_framed_reply(self) -> None:
+        broker = SlowBroker(original_connection_budget() + 0.1)
+        broker_lock = threading.Lock()
+        server, client = socket.socketpair(socket.AF_UNIX, socket.SOCK_STREAM)
+        client.settimeout(original_connection_budget() + 2.0)
+        worker = threading.Thread(
+            target=DAEMON.handle_connection,
+            args=(server, broker, broker_lock),
+            daemon=True,
+        )
+        worker.start()
+        client.sendall(b'{"action":"probe"}\n')
+        client.shutdown(socket.SHUT_WR)
+
+        reply = bytearray()
+        while True:
+            chunk = client.recv(4096)
+            if not chunk:
+                break
+            reply.extend(chunk)
+        worker.join(timeout=2.0)
+        client.close()
+
+        self.assertFalse(worker.is_alive())
+        self.assertEqual(broker.calls, 1)
+        self.assertTrue(reply.endswith(b"\n"), f"unframed reply: {bytes(reply)!r}")
+        self.assertEqual(json.loads(reply), {"ok": True, "echo": "probe"})
+
+
+if __name__ == "__main__":
+    unittest.main()