fix(#838): bound broker reply deadlines

docs/scratchpads/838-broker-acceptance-flake.md

@@ -18,3 +18,21 @@ Eliminate high-contention uncaught JSON parse failures in lease-broker and mutat
 ## Progress
 
 - 2026-07-17: Reclaimed from #824. Confirmed clean worktree on `fix/838-broker-acceptance-flake` at main base `abd2791f59b3f06f46dd08e55298ced72f6aa7c2`.
+- RED evidence: after dependency setup, `broker-test-client.spec.ts` failed to load the intentionally absent shared client module. Its contract forces empty-close retry, repeated truncated-close rejection with byte context, and newline-terminated malformed-reply rejection.
+- RIDER2 verdict: branch **(b)**. `daemon.py` starts one connection deadline before request reading, then may spend that budget waiting for `broker_lock`; after handling, `remaining <= 0` returns without writing. A timed `sendall` failure can likewise close after a partial write. A deterministic socketpair probe held the lock past `CONNECTION_DEADLINE_SECONDS` and observed `deadline_probe_reply_length=0`.
+- Adapter tripwire: real subprocess executions of `mutator-gate.py` for both `--runtime claude` and `--runtime pi` against actual Unix servers returning empty and truncated replies all exited 2 with `GATE_UNAVAILABLE`. Adapter fail-close is proven; there is no ALLOW risk.
+- Residual: production broker reply loss remains an availability-denial path under extreme contention, but cannot grant mutator authority. The acceptance-only client retries early closes and otherwise rejects with response byte length, escaped bytes, and hex; malformed newline-framed replies are never retried or converted to reply objects.
+- Shared client now owns newline framing and parsing for both acceptance suites. Focused suites pass 60 tests, including the real adapter tripwire.
+- Coverage gate: changed helper is 100% statements/lines/functions and 90% branches; existing `skill.ts` remains above 85% per-file thresholds.
+
+## Scope corrections and bounded product repair
+
+- Coordinator correction superseded the initial push/PR and retry language: this lane is BUILD-ONLY, and early-close retries are forbidden because they can mask a committed broker transaction. Nothing may be pushed or opened until explicitly cleared.
+- Revised RED evidence: the no-retry empty/truncated tests failed because the first failed exchange was retried into `{ ok: true }`; the daemon regression failed because a slow completed `broker.handle()` produced `b''` instead of a newline-framed reply.
+- Client repair: both former duplicated helpers use one shared reader. Empty, truncated, malformed, oversized, timed-out, and socket-error replies reject `BrokerTransportError` with a typed `kind`, attempt count fixed at one, response length, escaped bytes, and hex. No retry and no catch-to-reply conversion exists.
+- Product repair: `daemon.py` now has independent bounded read, broker-lock queue, and send budgets. Lock queue exhaustion returns explicit `BROKER_BUSY` before `broker.handle()` can mutate state. Once handling starts it finishes atomically, and its reply always receives a fresh send timeout instead of being skipped because read/lock/fsync consumed a shared deadline.
+- Product GREEN evidence: deterministic socketpair tests prove lock saturation returns framed `BROKER_BUSY` without invoking `handle()`, and a handle that completes after the former one-second shared deadline still returns its complete framed reply.
+- RIDER2b remains **fail-closed**: real Claude and Pi `mutator-gate.py` subprocesses against empty and truncated Unix-socket replies exit 2 with `GATE_UNAVAILABLE`; no malformed/default ALLOW was observed.
+- Residual/tripwire: an unavoidable peer disconnect or send failure can still lose acknowledgement after a valid transaction commits. The affected adapter call fails closed. A valid `promote_lease` may nevertheless remain VERIFIED after its acknowledgement is lost; that is authority-observability divergence requiring WI-3/Opus security review rather than expansion of #838. #838 does not add retries or attempt a protocol redesign.
+- Final package evidence: recursive Mosaic dependency build passed; 73 Vitest files / 1,392 tests passed; deadline unit tests 2/2, real runtime tool tests 15/15, launch guard tests 12/12, inventory 14/14, and shell regressions passed.
+- Final coverage: `broker-test-client.ts` 99.24% statements/lines, 86.11% branches, 100% functions under per-file >=85% thresholds. Repository typecheck 42/42, lint 23/23, and format check passed.

packages/mosaic/framework/tools/lease-broker/daemon.py

@@ -26,7 +26,9 @@ MAX_PENDING_TOKENS: Final = 256
 MAX_IN_FLIGHT_CONNECTIONS: Final = 16
 MAX_LEASE_TTL_SECONDS: Final = 300
 STATE_VERSION: Final = 1
-CONNECTION_DEADLINE_SECONDS: Final = 1.0
+READ_DEADLINE_SECONDS: Final = 1.0
+HANDLE_QUEUE_TIMEOUT_SECONDS: Final = 1.0
+SEND_TIMEOUT_SECONDS: Final = 1.0
 HEX_256_LENGTH: Final = 64
 LEASE_UNVERIFIED: Final = "UNVERIFIED"
 LEASE_PENDING: Final = "PENDING_VERIFICATION"
@@ -570,26 +572,33 @@ def handle_connection(
     broker_lock: threading.Lock,
 ) -> None:
     with connection:
-        deadline = time.monotonic() + CONNECTION_DEADLINE_SECONDS
+        read_deadline = time.monotonic() + READ_DEADLINE_SECONDS
         try:
             raw = connection.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12)
             peer = struct.unpack("3i", raw)
-            request = read_frame(connection, deadline)
+            request = read_frame(connection, read_deadline)
         except BrokerFailure as exc:
             reply = {"ok": False, "code": exc.code}
         except OSError:
             return
         else:
-            try:
-                with broker_lock:
-                    reply = broker.handle(peer, request)
-            except BrokerFailure as exc:
-                reply = {"ok": False, "code": exc.code}
+            # Queueing is bounded and fails closed before broker.handle mutates
+            # state. Once handling starts it must finish atomically; its reply
+            # then receives an independent send budget so a slow fsync cannot
+            # consume the write opportunity and create client/broker ambiguity.
+            acquired = broker_lock.acquire(timeout=HANDLE_QUEUE_TIMEOUT_SECONDS)
+            if not acquired:
+                reply = {"ok": False, "code": "BROKER_BUSY"}
+            else:
+                try:
+                    try:
+                        reply = broker.handle(peer, request)
+                    except BrokerFailure as exc:
+                        reply = {"ok": False, "code": exc.code}
+                finally:
+                    broker_lock.release()
         try:
-            remaining = deadline - time.monotonic()
-            if remaining <= 0:
-                return
-            connection.settimeout(remaining)
+            connection.settimeout(SEND_TIMEOUT_SECONDS)
             connection.sendall((json.dumps(reply, separators=(",", ":")) + "\n").encode())
         except OSError:
             return

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

@@ -0,0 +1,157 @@
+import { createConnection, type Socket } from 'node:net';
+
+const DEFAULT_TIMEOUT_MS = 3_000;
+const MAX_REPLY_BYTES = 64 * 1024;
+
+export interface BrokerTestClientOptions {
+  timeoutMs?: number;
+}
+
+export type BrokerTransportFailureKind =
+  | 'early-close'
+  | 'malformed-reply'
+  | 'socket-error'
+  | 'timeout';
+
+export class BrokerTransportError extends Error {
+  public readonly responseLength: number;
+  public readonly responseBytes: string;
+  public readonly responseHex: string;
+
+  public constructor(
+    public readonly kind: BrokerTransportFailureKind,
+    description: string,
+    response: Buffer,
+    public readonly attempts = 1,
+  ) {
+    super(`${description}; attempts=${attempts}; ${responseContext(response)}`);
+    this.name = 'BrokerTransportError';
+    this.responseLength = response.length;
+    this.responseBytes = response.toString('utf8');
+    this.responseHex = response.toString('hex');
+  }
+}
+
+function responseContext(response: Buffer): string {
+  const text = JSON.stringify(response.toString('utf8'));
+  const hex = response.length === 0 ? '<empty>' : response.toString('hex');
+  return `length=${response.length}; bytes=${text}; hex=${hex}`;
+}
+
+function malformedReply(response: Buffer, reason: string): BrokerTransportError {
+  return new BrokerTransportError('malformed-reply', `Malformed broker reply: ${reason}`, response);
+}
+
+function readBrokerReplyAttempt<T extends object>(
+  socketPath: string,
+  write: (socket: Socket) => void,
+  timeoutMs: number,
+): Promise<T> {
+  return new Promise<T>((resolve, reject) => {
+    const socket = createConnection(socketPath);
+    let response = Buffer.alloc(0);
+    let settled = false;
+
+    const settle = (callback: () => void): void => {
+      if (settled) return;
+      settled = true;
+      clearTimeout(timer);
+      callback();
+      socket.destroy();
+    };
+    const fail = (error: Error): void => settle(() => reject(error));
+    const timer = setTimeout(
+      () =>
+        fail(
+          new BrokerTransportError(
+            'timeout',
+            `Broker reply timed out before newline after ${timeoutMs}ms`,
+            response,
+          ),
+        ),
+      timeoutMs,
+    );
+
+    socket.once('error', (error) =>
+      fail(
+        new BrokerTransportError(
+          'socket-error',
+          `Broker reply socket error before newline: ${error.message}`,
+          response,
+        ),
+      ),
+    );
+    socket.on('data', (chunk: Buffer) => {
+      if (settled) return;
+      response = Buffer.concat([response, chunk]);
+      if (response.length > MAX_REPLY_BYTES) {
+        fail(malformedReply(response, `exceeds ${MAX_REPLY_BYTES} bytes`));
+        return;
+      }
+
+      const newline = response.indexOf(0x0a);
+      if (newline < 0) return;
+      if (newline !== response.length - 1) {
+        fail(malformedReply(response, 'contains bytes after the newline terminator'));
+        return;
+      }
+
+      try {
+        const parsed: unknown = JSON.parse(response.subarray(0, newline).toString('utf8'));
+        if (typeof parsed !== 'object' || parsed === null || Array.isArray(parsed)) {
+          fail(malformedReply(response, 'JSON value is not an object'));
+          return;
+        }
+        settle(() => resolve(parsed as T));
+      } catch (error: unknown) {
+        fail(
+          malformedReply(response, error instanceof Error ? error.message : 'JSON parsing failed'),
+        );
+      }
+    });
+    socket.once('end', () => {
+      if (!settled) {
+        fail(
+          new BrokerTransportError(
+            'early-close',
+            'Broker reply socket closed before newline',
+            response,
+          ),
+        );
+      }
+    });
+    socket.once('connect', () => {
+      try {
+        write(socket);
+      } catch (error: unknown) {
+        fail(error instanceof Error ? error : new Error(String(error)));
+      }
+    });
+  });
+}
+
+/** Read exactly one complete newline-framed JSON object; transport failures reject. */
+export async function readBrokerReply<T extends object>(
+  socketPath: string,
+  write: (socket: Socket) => void,
+  options: BrokerTestClientOptions = {},
+): Promise<T> {
+  return await readBrokerReplyAttempt<T>(
+    socketPath,
+    write,
+    options.timeoutMs ?? DEFAULT_TIMEOUT_MS,
+  );
+}
+
+/** Send one newline-framed request and read its complete broker reply. */
+export async function requestBrokerReply<T extends object>(
+  socketPath: string,
+  requestValue: object,
+  options?: BrokerTestClientOptions,
+): Promise<T> {
+  return await readBrokerReply<T>(
+    socketPath,
+    (socket) => socket.end(`${JSON.stringify(requestValue)}\n`),
+    options,
+  );
+}

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

@@ -39,6 +39,37 @@ class SlowBroker:
 
 
 class BrokerDeadlineTest(unittest.TestCase):
+    def test_lock_queue_timeout_returns_explicit_fail_closed_reply_without_handling(self) -> None:
+        broker = SlowBroker(0)
+        broker_lock = threading.Lock()
+        broker_lock.acquire()
+        server, client = socket.socketpair(socket.AF_UNIX, socket.SOCK_STREAM)
+        client.settimeout(DAEMON.HANDLE_QUEUE_TIMEOUT_SECONDS + 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()
+        try:
+            while True:
+                chunk = client.recv(4096)
+                if not chunk:
+                    break
+                reply.extend(chunk)
+        finally:
+            broker_lock.release()
+            worker.join(timeout=2.0)
+            client.close()
+
+        self.assertFalse(worker.is_alive())
+        self.assertEqual(broker.calls, 0)
+        self.assertEqual(json.loads(reply), {"ok": False, "code": "BROKER_BUSY"})
+
     def test_completed_slow_handle_gets_a_complete_framed_reply(self) -> None:
         broker = SlowBroker(original_connection_budget() + 0.1)
         broker_lock = threading.Lock()

packages/mosaic/src/lease-broker/lease-broker.acceptance.spec.ts

@@ -7,6 +7,8 @@ import { spawn, spawnSync, type ChildProcess } from 'node:child_process';
 
 import { afterEach, describe, expect, test, vi } from 'vitest';
 
+import { readBrokerReply, requestBrokerReply } from './broker-test-client.js';
+
 interface BrokerReply {
   ok: boolean;
   code?: string;
@@ -41,34 +43,11 @@ async function rawRequest(
   socketPath: string,
   write: (socket: Socket) => void,
 ): Promise<BrokerReply> {
-  return await withTimeout(
-    new Promise<BrokerReply>((resolve, reject) => {
-      const socket = createConnection(socketPath);
-      let response = '';
-      socket.setEncoding('utf8');
-      socket.once('error', reject);
-      socket.on('data', (chunk: string) => {
-        response += chunk;
-      });
-      socket.once('end', () => resolve(JSON.parse(response) as BrokerReply));
-      socket.once('connect', () => write(socket));
-    }),
-    'raw broker request',
-  );
+  return await readBrokerReply<BrokerReply>(socketPath, write);
 }
 
 async function request(socketPath: string, requestValue: object): Promise<BrokerReply> {
-  return await new Promise<BrokerReply>((resolve, reject) => {
-    const socket = createConnection(socketPath);
-    let response = '';
-    socket.setEncoding('utf8');
-    socket.once('error', reject);
-    socket.on('data', (chunk: string) => {
-      response += chunk;
-    });
-    socket.once('end', () => resolve(JSON.parse(response) as BrokerReply));
-    socket.once('connect', () => socket.end(`${JSON.stringify(requestValue)}\n`));
-  });
+  return await requestBrokerReply<BrokerReply>(socketPath, requestValue);
 }
 
 async function startBroker(

packages/mosaic/src/mutator-gate/mutator-gate.acceptance.spec.ts

@@ -1,5 +1,4 @@
 import { chmod, mkdir, mkdtemp, readFile, rm, writeFile } from 'node:fs/promises';
-import { createConnection } from 'node:net';
 import { tmpdir } from 'node:os';
 import { join } from 'node:path';
 import { spawn, spawnSync, type ChildProcess } from 'node:child_process';
@@ -7,6 +6,7 @@ import { spawn, spawnSync, type ChildProcess } from 'node:child_process';
 import { afterEach, describe, expect, test } from 'vitest';
 
 import { launchClaudex, type ClaudexHarnessAdapter } from '../commands/claudex.js';
+import { requestBrokerReply } from '../lease-broker/broker-test-client.js';
 
 interface BrokerReply {
   ok: boolean;
@@ -43,17 +43,7 @@ const binding = (compaction_epoch = 1) => ({
 });
 
 async function request(socketPath: string, requestValue: object): Promise<BrokerReply> {
-  return await new Promise<BrokerReply>((resolve, reject) => {
-    const socket = createConnection(socketPath);
-    let response = '';
-    socket.setEncoding('utf8');
-    socket.once('error', reject);
-    socket.on('data', (chunk: string) => {
-      response += chunk;
-    });
-    socket.once('end', () => resolve(JSON.parse(response) as BrokerReply));
-    socket.once('connect', () => socket.end(`${JSON.stringify(requestValue)}\n`));
-  });
+  return await requestBrokerReply<BrokerReply>(socketPath, requestValue);
 }
 
 async function startBroker(): Promise<BrokerPaths> {

packages/mosaic/src/mutator-gate/runtime_tools_unittest.py

@@ -9,7 +9,10 @@ import json
 import os
 import runpy
 import socket
+import subprocess
 import sys
+import tempfile
+import threading
 import unittest
 from contextlib import redirect_stderr
 from pathlib import Path
@@ -225,6 +228,45 @@ class LaunchRuntimeTest(unittest.TestCase):
 
 
 class ExecutableEntrypointTest(unittest.TestCase):
+    def test_real_claude_and_pi_gates_fail_closed_on_empty_or_truncated_reply(self) -> None:
+        for runtime in ("claude", "pi"):
+            for wire_reply in (b"", b'{"ok":true'):
+                with self.subTest(runtime=runtime, wire_reply=wire_reply), tempfile.TemporaryDirectory() as root:
+                    socket_path = Path(root) / "broker.sock"
+                    server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+                    server.bind(str(socket_path))
+                    server.listen(1)
+
+                    def serve_reply() -> None:
+                        with server:
+                            connection, _ = server.accept()
+                            with connection:
+                                while connection.recv(4096):
+                                    pass
+                                if wire_reply:
+                                    connection.sendall(wire_reply)
+
+                    thread = threading.Thread(target=serve_reply, daemon=True)
+                    thread.start()
+                    environment = {
+                        **os.environ,
+                        "MOSAIC_LEASE_BROKER_SOCKET": str(socket_path),
+                        "MOSAIC_LEASE_SESSION_ID": "d" * 64,
+                        "MOSAIC_RUNTIME_GENERATION": "1",
+                    }
+                    result = subprocess.run(
+                        [sys.executable, str(TOOLS_DIR / "mutator-gate.py"), "--runtime", runtime],
+                        input=b'{"tool_name":"Read"}\n',
+                        capture_output=True,
+                        env=environment,
+                        check=False,
+                    )
+                    thread.join(timeout=2)
+
+                    self.assertFalse(thread.is_alive())
+                    self.assertEqual(result.returncode, 2)
+                    self.assertIn(b"GATE_UNAVAILABLE", result.stderr)
+
     def test_launcher_entrypoint_returns_usage_without_a_command(self) -> None:
         with patch.object(
             sys,

packages/mosaic/vitest.config.ts

@@ -7,9 +7,10 @@ export default defineConfig({
     testTimeout: 30_000,
     coverage: {
       provider: 'v8',
-      include: ['src/commands/skill.ts'],
+      include: ['src/commands/skill.ts', 'src/lease-broker/broker-test-client.ts'],
       reporter: ['text', 'json-summary'],
       thresholds: {
+        perFile: true,
         statements: 85,
         branches: 85,
         functions: 85,