diff --git a/docs/scratchpads/838-broker-acceptance-flake.md b/docs/scratchpads/838-broker-acceptance-flake.md index 6b9e051..1a9fa41 100644 --- a/docs/scratchpads/838-broker-acceptance-flake.md +++ b/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. diff --git a/packages/mosaic/framework/tools/lease-broker/daemon.py b/packages/mosaic/framework/tools/lease-broker/daemon.py index d33e453..1ac8e22 100644 --- a/packages/mosaic/framework/tools/lease-broker/daemon.py +++ b/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 diff --git a/packages/mosaic/src/lease-broker/broker-test-client.ts b/packages/mosaic/src/lease-broker/broker-test-client.ts new file mode 100644 index 0000000..f26ed88 --- /dev/null +++ b/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 ? '' : 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( + socketPath: string, + write: (socket: Socket) => void, + timeoutMs: number, +): Promise { + return new Promise((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( + socketPath: string, + write: (socket: Socket) => void, + options: BrokerTestClientOptions = {}, +): Promise { + return await readBrokerReplyAttempt( + socketPath, + write, + options.timeoutMs ?? DEFAULT_TIMEOUT_MS, + ); +} + +/** Send one newline-framed request and read its complete broker reply. */ +export async function requestBrokerReply( + socketPath: string, + requestValue: object, + options?: BrokerTestClientOptions, +): Promise { + return await readBrokerReply( + socketPath, + (socket) => socket.end(`${JSON.stringify(requestValue)}\n`), + options, + ); +} diff --git a/packages/mosaic/src/lease-broker/daemon_deadline_unittest.py b/packages/mosaic/src/lease-broker/daemon_deadline_unittest.py index a63f912..015db26 100644 --- a/packages/mosaic/src/lease-broker/daemon_deadline_unittest.py +++ b/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() diff --git a/packages/mosaic/src/lease-broker/lease-broker.acceptance.spec.ts b/packages/mosaic/src/lease-broker/lease-broker.acceptance.spec.ts index d2fbc29..04d1d5e 100644 --- a/packages/mosaic/src/lease-broker/lease-broker.acceptance.spec.ts +++ b/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 { - return await withTimeout( - new Promise((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(socketPath, write); } async function request(socketPath: string, requestValue: object): Promise { - return await new Promise((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(socketPath, requestValue); } async function startBroker( diff --git a/packages/mosaic/src/mutator-gate/mutator-gate.acceptance.spec.ts b/packages/mosaic/src/mutator-gate/mutator-gate.acceptance.spec.ts index 932eab1..b75f3db 100644 --- a/packages/mosaic/src/mutator-gate/mutator-gate.acceptance.spec.ts +++ b/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 { - return await new Promise((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(socketPath, requestValue); } async function startBroker(): Promise { diff --git a/packages/mosaic/src/mutator-gate/runtime_tools_unittest.py b/packages/mosaic/src/mutator-gate/runtime_tools_unittest.py index e809802..ac49b69 100644 --- a/packages/mosaic/src/mutator-gate/runtime_tools_unittest.py +++ b/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, diff --git a/packages/mosaic/vitest.config.ts b/packages/mosaic/vitest.config.ts index 9476528..f542764 100644 --- a/packages/mosaic/vitest.config.ts +++ b/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,