diff --git a/docs/scratchpads/838-broker-acceptance-flake.md b/docs/scratchpads/838-broker-acceptance-flake.md new file mode 100644 index 0000000..1e52a57 --- /dev/null +++ b/docs/scratchpads/838-broker-acceptance-flake.md @@ -0,0 +1,44 @@ +# Issue #838 — Broker acceptance socket flake + +## Objective + +Eliminate high-contention uncaught JSON parse failures in lease-broker and mutator-gate acceptance helpers without laundering malformed/empty replies into passing assertions. + +## Constraints + +- Branch: `fix/838-broker-acceptance-flake` in `/home/hermes/agent-work/stack-838-flakefix`. +- Red-first TDD with a forced empty/truncated reply path and deterministic rejection or documented retry. +- Read newline-framed replies completely; reject malformed broker replies with byte length/content context. +- Determine RIDER2 branch before finalizing: test-harness-only `(a)` or daemon write truncation `(b)`. +- If product-side, prove the real Claude/Pi adapter read path fails closed; fix any allow-risk. +- Coverage >=85% per changed executable through real tests. +- Full suite and repository gates green; independent exact-head review required. +- Push and open a PR containing `closes #838`; do not merge. + +## 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. +- Independent Codex exact-head review requested one framing fix: a valid frame followed by trailing bytes in a later socket chunk could resolve before the garbage arrived. Security review also flagged complete token-bearing reply bodies in diagnostic properties/logs. +- Review RED evidence: delayed cross-chunk garbage resolved `{ ok: true }` instead of rejecting, and a truncated `promotion_token` remained in the typed error. The tests use separate timed writes to prevent kernel/event-loop coalescing. +- Review remediation: the shared client now accumulates through EOF, requires exactly one terminal newline, then parses inside a rejecting error boundary. Diagnostic bodies are capped at 256 bytes, sensitive broker fields are fully redacted, and a SHA-256 digest preserves correlation without credential disclosure. +- Post-review coverage: `broker-test-client.ts` 99.35% statements/lines, 86.66% branches, 100% functions. Final full suite is 73 files / 1,394 tests plus all Python/shell gates; recursive build, typecheck, lint, and format are green. +- Fresh exact-head Codex security review: risk `none`, no findings. Fresh code review found only that the real-adapter subprocess proof lacked a timeout; it now has a five-second bound and fails with runtime/wire context while closing the fake server. The focused Python suite remains 15/15 green. +- Mandatory Opus SECREV remains coordinator-owned and pending before any push/PR decision; this build is intentionally local-only. 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/package.json b/packages/mosaic/package.json index d1b8b07..3ffda77 100644 --- a/packages/mosaic/package.json +++ b/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:*", diff --git a/packages/mosaic/src/lease-broker/broker-test-client.spec.ts b/packages/mosaic/src/lease-broker/broker-test-client.spec.ts new file mode 100644 index 0000000..f1c6d68 --- /dev/null +++ b/packages/mosaic/src/lease-broker/broker-test-client.spec.ts @@ -0,0 +1,181 @@ +import { mkdtemp, rm } from 'node:fs/promises'; +import { createServer, type Server, type Socket } from 'node:net'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; + +import { afterEach, describe, expect, test } from 'vitest'; + +import { BrokerTransportError, readBrokerReply, requestBrokerReply } from './broker-test-client.js'; + +const roots: string[] = []; +const servers: Server[] = []; +const sockets: Socket[] = []; + +async function scriptedBroker( + replies: ReadonlyArray | 'hang'>, +): Promise<{ socketPath: string; connections: () => number }> { + const root = await mkdtemp(join(tmpdir(), 'mosaic-broker-client-')); + roots.push(root); + const socketPath = join(root, 'broker.sock'); + let connections = 0; + const server = createServer({ allowHalfOpen: true }, (socket) => { + sockets.push(socket); + const chunks = replies[connections] ?? replies.at(-1) ?? []; + connections += 1; + socket.once('end', () => { + if (chunks === 'hang') return; + void (async () => { + for (const chunk of chunks) { + socket.write(chunk); + await new Promise((resolve) => setTimeout(resolve, 5)); + } + socket.end(); + })(); + }); + socket.resume(); + }); + servers.push(server); + await new Promise((resolve, reject) => { + server.once('error', reject); + server.listen(socketPath, resolve); + }); + return { socketPath, connections: () => connections }; +} + +afterEach(async () => { + for (const socket of sockets.splice(0)) socket.destroy(); + await Promise.all( + servers + .splice(0) + .map( + (server) => + new Promise((resolve, reject) => + server.close((error) => (error ? reject(error) : resolve())), + ), + ), + ); + await Promise.all(roots.splice(0).map((root) => rm(root, { recursive: true, force: true }))); +}); + +describe('newline-framed broker test client', () => { + test('rejects an empty early close without retrying into a false green', async () => { + const broker = await scriptedBroker([[], [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: 0, + responseHex: '', + }); + expect(failure).toHaveProperty( + 'message', + expect.stringMatching(/closed before newline.*length=0.*hex=/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', + }); + expect(failure).toHaveProperty( + 'message', + expect.stringMatching(/closed before newline.*length=6.*bytes=.*ok/i), + ); + expect(broker.connections()).toBe(1); + }); + + test('rejects a newline-terminated malformed broker reply without retrying', async () => { + const broker = await scriptedBroker([[Buffer.from('{bad}\n')]]); + + await expect(requestBrokerReply(broker.socketPath, { action: 'probe' })).rejects.toThrow( + /malformed broker reply.*length=6.*bytes="\{bad\}\\n"/i, + ); + expect(broker.connections()).toBe(1); + }); + + test('rejects trailing bytes delivered after a complete frame in a later data event', async () => { + const broker = await scriptedBroker([[Buffer.from('{"ok":true}\n'), Buffer.from('extra')]]); + + const failure = await requestBrokerReply(broker.socketPath, { action: 'probe' }).catch( + (error: unknown) => error, + ); + + expect(failure).toBeInstanceOf(BrokerTransportError); + expect(failure).toMatchObject({ kind: 'malformed-reply', responseLength: 17 }); + expect(failure).toHaveProperty( + 'message', + expect.stringMatching(/bytes after the newline terminator/i), + ); + }); + + test('redacts security tokens from typed transport diagnostics', async () => { + const token = 'a'.repeat(64); + const reply = Buffer.from(`{"ok":true,"promotion_token":"${token}`); + const broker = await scriptedBroker([[reply]]); + + const failure = await requestBrokerReply(broker.socketPath, { action: 'probe' }).catch( + (error: unknown) => error, + ); + + expect(failure).toBeInstanceOf(BrokerTransportError); + expect(failure).toMatchObject({ + kind: 'early-close', + responseLength: reply.length, + responsePreview: '', + }); + expect(String((failure as Error).message)).not.toContain(token); + expect(JSON.stringify(failure)).not.toContain(token); + }); + + test.each([ + ['trailing bytes', Buffer.from('{"ok":true}\nextra'), /bytes after the newline/i], + ['non-object JSON', Buffer.from('[]\n'), /JSON value is not an object/i], + ['oversized frame', Buffer.alloc(64 * 1024 + 1, 0x78), /exceeds 65536 bytes/i], + ])('rejects %s with deterministic framing context', async (_label, reply, message) => { + const broker = await scriptedBroker([[reply as Buffer]]); + + await expect(requestBrokerReply(broker.socketPath, { action: 'probe' })).rejects.toThrow( + message as RegExp, + ); + expect(broker.connections()).toBe(1); + }); + + test('reports timeout, connection, and writer failures as promise rejections', async () => { + const hanging = await scriptedBroker(['hang']); + await expect( + requestBrokerReply(hanging.socketPath, { action: 'probe' }, { timeoutMs: 10 }), + ).rejects.toThrow(/timed out before newline.*length=0.*hex=/i); + + await expect( + requestBrokerReply(join(rootForMissingSocket(), 'missing.sock'), {}), + ).rejects.toThrow(/socket error before newline.*length=0/i); + + const writerFailure = await scriptedBroker(['hang']); + await expect( + readBrokerReply(writerFailure.socketPath, () => { + throw new Error('writer failed'); + }), + ).rejects.toThrow('writer failed'); + }); +}); + +function rootForMissingSocket(): string { + return join(tmpdir(), `mosaic-missing-broker-${process.pid}-${Date.now()}`); +} 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..b3aeddb --- /dev/null +++ b/packages/mosaic/src/lease-broker/broker-test-client.ts @@ -0,0 +1,187 @@ +import { createHash } from 'node:crypto'; +import { createConnection, type Socket } from 'node:net'; + +const DEFAULT_TIMEOUT_MS = 3_000; +const MAX_REPLY_BYTES = 64 * 1024; +const MAX_DIAGNOSTIC_BYTES = 256; +const SENSITIVE_REPLY_FIELD = /"(?:promotion_token|session_id|token)"\s*:/; + +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 readonly responsePreview: string; + public readonly responseSha256: string; + + public constructor( + public readonly kind: BrokerTransportFailureKind, + description: string, + response: Buffer, + public readonly attempts = 1, + ) { + const diagnostics = responseDiagnostics(response); + super(`${description}; attempts=${attempts}; ${responseContext(response, diagnostics)}`); + this.name = 'BrokerTransportError'; + this.responseLength = response.length; + this.responseBytes = diagnostics.preview; + this.responseHex = diagnostics.hex; + this.responsePreview = diagnostics.preview; + this.responseSha256 = diagnostics.sha256; + } +} + +interface ResponseDiagnostics { + preview: string; + hex: string; + sha256: string; +} + +function responseDiagnostics(response: Buffer): ResponseDiagnostics { + const fullText = response.toString('utf8'); + const sensitive = SENSITIVE_REPLY_FIELD.test(fullText); + const bounded = response.subarray(0, MAX_DIAGNOSTIC_BYTES); + const suffix = response.length > MAX_DIAGNOSTIC_BYTES ? '…' : ''; + return { + preview: + response.length === 0 + ? '' + : sensitive + ? '' + : `${bounded.toString('utf8')}${suffix}`, + hex: sensitive ? '' : `${bounded.toString('hex')}${suffix}`, + sha256: createHash('sha256').update(response).digest('hex'), + }; +} + +function responseContext(response: Buffer, diagnostics = responseDiagnostics(response)): string { + const hex = diagnostics.hex.length === 0 ? '' : diagnostics.hex; + return `length=${response.length}; bytes=${JSON.stringify(diagnostics.preview)}; hex=${hex}; sha256=${diagnostics.sha256}`; +} + +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`)); + } + }); + socket.once('end', () => { + if (settled) return; + const newline = response.indexOf(0x0a); + if (response.length === 0 || newline < 0) { + fail( + new BrokerTransportError( + 'early-close', + 'Broker reply socket closed before newline', + response, + ), + ); + 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('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 new file mode 100644 index 0000000..015db26 --- /dev/null +++ b/packages/mosaic/src/lease-broker/daemon_deadline_unittest.py @@ -0,0 +1,103 @@ +#!/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_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() + 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() 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..0d8b99b 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,53 @@ 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", + } + try: + 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, + timeout=5, + ) + except subprocess.TimeoutExpired as exc: + server.close() + thread.join(timeout=2) + self.fail( + f"{runtime} gate hung on wire reply {wire_reply!r}: {exc}" + ) + 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,