From 0ae8c4f1634f07dd230150b0bdd2a620abd23bf4 Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Tue, 13 Oct 2026 06:53:59 +0800 Subject: [PATCH 1/2] test(listener): add reproducible baseline and recovery validation --- .github/workflows/listener-reliability.yml | 69 ++++++++++++++++++++++ 1 file changed, 69 insertions(+) create mode 100644 .github/workflows/listener-reliability.yml diff --git a/.github/workflows/listener-reliability.yml b/.github/workflows/listener-reliability.yml new file mode 100644 index 00000000..79012941 --- /dev/null +++ b/.github/workflows/listener-reliability.yml @@ -0,0 +1,69 @@ +name: Listener reliability +on: + push: + branches: ['fix/listener-reliability-*'] + pull_request: + paths: + - 'src/msg/client_listener.py' + - 'src/msg/client_agent_cli.py' + - 'src/msg/client_subagents*.py' + - 'src/msg/client_events.py' + - 'tests/test_client_listener*.py' + - 'tests/test_client_subagents*.py' + - 'scripts/benchmark_listener.py' + - '.github/workflows/listener-reliability.yml' + workflow_dispatch: +permissions: + contents: read +concurrency: + group: listener-${{ github.head_ref || github.ref_name }} + cancel-in-progress: false +jobs: + validation: + runs-on: ubuntu-latest + timeout-minutes: 15 + services: + postgres: + image: postgres:16 + env: + POSTGRES_USER: msgtest + POSTGRES_PASSWORD: msgtest + POSTGRES_DB: postgres + ports: + - 5432:5432 + options: >- + --health-cmd "pg_isready -U msgtest -d postgres" + --health-interval 5s --health-timeout 5s --health-retries 10 + env: + MSG_TEST_POSTGRES_URL_TEMPLATE: postgresql://msgtest:msgtest@127.0.0.1:5432/{database} + steps: + - uses: actions/checkout@v7 + with: + persist-credentials: false + - name: Record source for reproduction + run: | + mkdir -p artifacts + git rev-parse HEAD > artifacts/commit.txt + git archive --format=tar.gz -o artifacts/source.tar.gz HEAD + - uses: actions/setup-python@v7 + with: + python-version: '3.15' + allow-prereleases: true + - name: Install smartcard build headers + run: sudo apt-get update && sudo apt-get install -y libpcsclite-dev pkg-config + - name: Install locked dependencies + run: | + python -m pip install uv==0.12.20 + uv sync --locked --extra dev --python "$(command -v python)" + - name: Listener, local mailbox and private remote mailbox regressions + run: >- + uv run --locked --extra dev pytest + tests/test_client_listener.py tests/test_client_subagents.py + tests/test_client_subagents_remote.py + --junitxml=artifacts/listener.xml + - uses: actions/upload-artifact@v7.0.1 + if: always() + with: + name: listener-reliability-${{ github.sha }} + path: artifacts/ + if-no-files-found: error From a6263ab0d37bfe90e0b1c857addbf3bf8a708b82 Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Tue, 13 Oct 2026 07:17:57 +0800 Subject: [PATCH 2/2] fix(listener): add native local waits and reauthorize remote recovery Keep remote checkpoints free of message bodies, bound idle polling, and measure actual mailbox/HTTP costs without claiming a cross-harness ranking. Add real SIGKILL recovery and scoped-credential regression tests. --- .github/workflows/listener-reliability.yml | 15 +- scripts/benchmark_listener.py | 219 +++++++++++++++ src/msg/client_agent_cli.py | 47 +++- src/msg/client_events.py | 99 +++++++ src/msg/client_listener.py | 156 +++++++++-- src/msg/core/errors.py | 2 + tests/test_client_listener_cli.py | 17 ++ tests/test_client_listener_reliability.py | 306 +++++++++++++++++++++ tests/test_client_listener_remote.py | 135 +++++++++ 9 files changed, 969 insertions(+), 27 deletions(-) create mode 100644 scripts/benchmark_listener.py create mode 100644 src/msg/client_events.py create mode 100644 tests/test_client_listener_cli.py create mode 100644 tests/test_client_listener_reliability.py create mode 100644 tests/test_client_listener_remote.py diff --git a/.github/workflows/listener-reliability.yml b/.github/workflows/listener-reliability.yml index 79012941..65afb855 100644 --- a/.github/workflows/listener-reliability.yml +++ b/.github/workflows/listener-reliability.yml @@ -10,6 +10,8 @@ on: - 'src/msg/client_events.py' - 'tests/test_client_listener*.py' - 'tests/test_client_subagents*.py' + - 'tests/test_client_agent_cli.py' + - 'src/msg/core/errors.py' - 'scripts/benchmark_listener.py' - '.github/workflows/listener-reliability.yml' workflow_dispatch: @@ -35,11 +37,13 @@ jobs: --health-cmd "pg_isready -U msgtest -d postgres" --health-interval 5s --health-timeout 5s --health-retries 10 env: + MSG_BENCHMARK_OUTPUT: artifacts MSG_TEST_POSTGRES_URL_TEMPLATE: postgresql://msgtest:msgtest@127.0.0.1:5432/{database} steps: - uses: actions/checkout@v7 with: persist-credentials: false + fetch-depth: 0 - name: Record source for reproduction run: | mkdir -p artifacts @@ -59,8 +63,17 @@ jobs: run: >- uv run --locked --extra dev pytest tests/test_client_listener.py tests/test_client_subagents.py - tests/test_client_subagents_remote.py + tests/test_client_subagents_remote.py tests/test_client_agent_cli.py + tests/test_client_listener_reliability.py tests/test_client_listener_remote.py + tests/test_client_listener_cli.py --junitxml=artifacts/listener.xml + - name: Record bounded local measurements, not a harness ranking + run: | + git show 459f3f7be28eaa2f5639669d05add2a342bab20f:src/msg/client_listener.py > artifacts/baseline_listener.py + uv run --locked --extra dev python scripts/benchmark_listener.py \ + --baseline-listener artifacts/baseline_listener.py \ + --revision "$(git rev-parse HEAD)" --trials 3 --tasks 8 --workers 4 \ + --poll-interval 0.1 --idle-seconds 0.4 --output artifacts/local-listener.json - uses: actions/upload-artifact@v7.0.1 if: always() with: diff --git a/scripts/benchmark_listener.py b/scripts/benchmark_listener.py new file mode 100644 index 00000000..95680aed --- /dev/null +++ b/scripts/benchmark_listener.py @@ -0,0 +1,219 @@ +#!/usr/bin/env python3 +"""Real local MSG mailboxes with synthetic tasks; not a model/harness ranking.""" + +from __future__ import annotations + +import argparse +import asyncio +import hashlib +import importlib.util +import json +import math +import platform +import random +import statistics +import tempfile +import time +import traceback +from contextlib import ExitStack, contextmanager +from pathlib import Path + +import httpx + +from msg.client import ClientState +from msg.client_events import LocalEvents +from msg.client_listener import listen +from msg.client_subagents import LocalAgents + + +@contextmanager +def count_http(): + """Count and reject any accidental HTTP from a local-only measurement.""" + counter = {'attempts': 0} + original_sync, original_async = httpx.Client.send, httpx.AsyncClient.send + + def blocked(*args, **kwargs): + counter['attempts'] += 1 + raise RuntimeError('Unexpected HTTP in local mailbox benchmark') + + async def blocked_async(*args, **kwargs): + return blocked(*args, **kwargs) + + httpx.Client.send, httpx.AsyncClient.send = blocked, blocked_async + try: + yield counter + finally: + httpx.Client.send, httpx.AsyncClient.send = original_sync, original_async + + +def load_baseline(path): + spec = importlib.util.spec_from_file_location('msg_listener_baseline', path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module.listen + + +async def trial(function, native, durations, workers, interval, idle_seconds): + with tempfile.TemporaryDirectory(prefix='msg-listener-bench-') as root, ExitStack() as stack: + directory = Path(root) + state = ClientState(directory / 'account', server='https://benchmark.invalid') + store = LocalAgents(state, username='benchmark') + store.create('lead') + store.create('worker') + events = stack.enter_context(LocalEvents(store.path)) if native else None + requests = stack.enter_context(count_http()) + ready = asyncio.Event() + acknowledged = [asyncio.Event() for _ in durations] + seen, latencies = set(), [] + reads = 0 + + async def fetch(cursor, tail=False): + nonlocal reads + result = await asyncio.to_thread(store.inbox, 'lead', cursor=cursor, tail=tail) + reads += 1 + ready.set() + return result + + class Output: + def write(self, line): + event = json.loads(line) + if event['id'] in seen: + raise RuntimeError('Duplicate message delivery in uninterrupted trial') + seen.add(event['id']) + body = json.loads(event['message']) + latencies.append((time.monotonic_ns() - body['completed_ns']) / 1_000_000) + acknowledged[body['task']].set() + + def flush(self): + pass + + options = {'wait_for_change': events.wait} if events is not None else {} + reader = asyncio.create_task(function( + fetch, context={'source': 'local', 'agent': 'lead'}, + cursor_file=directory / 'cursor.json', interval=interval, + max_events=len(durations), output=Output(), **options, + )) + tasks = [] + try: + await asyncio.wait_for(ready.wait(), 10) + idle_start, idle_cpu = time.perf_counter(), time.process_time() + idle_reads, idle_requests = reads, requests['attempts'] + await asyncio.sleep(idle_seconds) + idle = { + 'seconds': time.perf_counter() - idle_start, + 'cpu_seconds': time.process_time() - idle_cpu, + 'source_reads': reads - idle_reads, + 'http_requests': requests['attempts'] - idle_requests, + } + + async def worker(index): + # Release a worker slot only after the coordinator observes its + # result. The same independent tasks run with 1 and N workers. + for number in range(index, len(durations), workers): + await asyncio.sleep(durations[number]) + body = json.dumps({'task': number, 'completed_ns': time.monotonic_ns()}) + await asyncio.to_thread( + store.send, 'worker', 'lead', body, message_id=f'task-{number}', + ) + await acknowledged[number].wait() + + started, cpu, active_reads = time.perf_counter(), time.process_time(), reads + tasks = [asyncio.create_task(worker(index)) for index in range(workers)] + await asyncio.wait_for(asyncio.gather(*tasks, reader), max(30, len(durations) * interval * 3)) + elapsed, cpu_seconds = time.perf_counter() - started, time.process_time() - cpu + expected = {f'task-{number}' for number in range(len(durations))} + if seen != expected: + raise RuntimeError('Incomplete task results') + ordered = sorted(latencies) + return { + 'status': 'measured', 'workers': workers, 'tasks': len(durations), + 'task_successes': len(seen), 'duplicate_deliveries': 0, + 'total_task_seconds': elapsed, 'active_cpu_seconds': cpu_seconds, + 'active_source_reads': reads - active_reads, + 'http_requests': requests['attempts'], 'idle': idle, + 'completion_to_observation_ms': { + 'median': statistics.median(latencies), + 'p95_nearest_rank': ordered[math.ceil(len(ordered) * 0.95) - 1], + 'max': max(latencies), + }, + } + finally: + for task in [reader, *tasks]: + task.cancel() + await asyncio.gather(reader, *tasks, return_exceptions=True) + + +async def measure(args): + modes = [('local_events', listen, True)] + missing = { + 'remote_events': {'status': 'not_implemented'}, + 'harness_native': {'status': 'not_measured', 'reason': 'No native harness adapters or runs.'}, + 'remote_polling': {'status': 'separate_test', 'artifact': 'remote-idle.json'}, + } + if args.baseline_listener: + modes.append(('legacy_listen', load_baseline(args.baseline_listener), False)) + else: + missing['legacy_listen'] = {'status': 'not_measured', 'reason': 'Provide --baseline-listener.'} + if platform.system() != 'Linux': + modes = [entry for entry in modes if not entry[2]] + missing['local_events'] = {'status': 'not_supported', 'reason': 'Native test requires Linux.'} + results, failed = [], False + for number in range(args.trials): + rng = random.Random(args.seed + number) + durations = [rng.uniform(0.02, 0.08) for _ in range(args.tasks)] + # Alternate order to reduce systematic warm-up/order bias. + for name, function, native in (modes if number % 2 == 0 else list(reversed(modes))): + pair = {} + for workers in (1, args.workers): + try: + row = await trial(function, native, durations, workers, + args.poll_interval, args.idle_seconds) + except Exception as exc: + traceback.print_exc() + row = {'status': 'failed', 'error_type': type(exc).__name__, 'workers': workers} + failed = True + pair[str(workers)] = row + result = {'mode': name, 'trial': number, 'runs': pair} + if all(row['status'] == 'measured' for row in pair.values()): + speedup = pair['1']['total_task_seconds'] / pair[str(args.workers)]['total_task_seconds'] + result.update(observed_speedup=speedup, parallel_efficiency=speedup / args.workers) + results.append(result) + return { + 'schema_version': 1, 'status': 'failed' if failed else 'measured', + 'revision': args.revision, 'python': platform.python_version(), + 'platform': platform.platform(), 'poll_interval_seconds': args.poll_interval, + 'idle_seconds_requested': args.idle_seconds, 'seed': args.seed, + 'baseline_sha256': hashlib.sha256(args.baseline_listener.read_bytes()).hexdigest() + if args.baseline_listener else None, + 'workload': 'Seeded synthetic sleep tasks, real local MSG SQLite send/read/checkpoint operations.', + 'scope': 'Single process with concurrent tasks and SQLite worker threads; not LLM or harness work.', + 'model_calls': 0, 'model_tokens': None, 'ranking': None, + 'missing_comparisons': missing, 'results': results, + } + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--output', type=Path, required=True) + parser.add_argument('--baseline-listener', type=Path) + parser.add_argument('--revision', default='unrecorded') + parser.add_argument('--trials', type=int, default=3) + parser.add_argument('--tasks', type=int, default=8) + parser.add_argument('--workers', type=int, default=4) + parser.add_argument('--poll-interval', type=float, default=1.0) + parser.add_argument('--idle-seconds', type=float, default=2.0) + parser.add_argument('--seed', type=int, default=1729) + args = parser.parse_args() + if args.trials < 1 or args.tasks < 2 or not 2 <= args.workers <= args.tasks: + parser.error('Use trials >= 1 and 2 <= workers <= tasks.') + if any(not math.isfinite(v) or v <= 0 for v in (args.poll_interval, args.idle_seconds)): + parser.error('Intervals must be finite and positive.') + result = asyncio.run(measure(args)) + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text(json.dumps(result, indent=2) + '\n') + print(json.dumps({'status': result['status'], 'output': str(args.output)})) + return int(result['status'] == 'failed') + + +if __name__ == '__main__': + raise SystemExit(main()) diff --git a/src/msg/client_agent_cli.py b/src/msg/client_agent_cli.py index 5cdb3550..3cec1545 100644 --- a/src/msg/client_agent_cli.py +++ b/src/msg/client_agent_cli.py @@ -3,6 +3,8 @@ from __future__ import annotations import asyncio +import math +import sys from hashlib import sha256 from pathlib import Path from types import SimpleNamespace @@ -68,6 +70,15 @@ def add_commands(commands): listener.add_argument('--cursor') listener.add_argument('--cursor-file', type=Path, help='Independent durable listener position.') listener.add_argument('--interval', type=float, default=1.0) + listener.add_argument( + '--wait-mode', choices=('auto', 'events', 'poll'), default='auto', + help='Auto: local native events where supported, remote adaptive polling. ' + 'Events: require local native events. Poll: fixed-interval legacy behavior.', + ) + listener.add_argument( + '--max-idle-interval', type=float, default=30.0, + help='Remote auto mode: maximum idle polling delay in seconds (default: 30).', + ) listener.add_argument( '--from-now', action='store_true', help='Skip existing events on a new listener.' ) @@ -148,9 +159,15 @@ def cursor_path(state, args, source): return private_directory(state.paths.state / 'agent-listeners') / (key + '.json') -async def stream(state, args, fetch, source): +async def stream(state, args, fetch, source, *, wait_for_change=None, resolve_legacy=None): from msg.client_listener import listen + remote = not source.startswith('local:') + idle_limit = args.interval + if remote and getattr(args, 'wait_mode', 'auto') == 'auto': + maximum = getattr(args, 'max_idle_interval', 30.0) + require(math.isfinite(maximum) and maximum > 0, 'invalid_listener_interval') + idle_limit = max(args.interval, maximum) await listen( fetch, context={ @@ -167,6 +184,10 @@ async def stream(state, args, fetch, source): max_events=args.max_events, max_pages=getattr(args, 'max_pages', None), from_now=args.from_now, + wait_for_change=wait_for_change, + max_idle_interval=idle_limit, + cursor_only=remote, + resolve_legacy=resolve_legacy, ) return 0 @@ -183,6 +204,23 @@ async def run_local(state, args): async def fetch(cursor, tail=False): return await asyncio.to_thread(store.inbox, name, cursor=cursor, tail=tail) + mode = getattr(args, 'wait_mode', 'auto') + if mode != 'poll' and not args.once: + from msg.client_events import EventsUnavailable, LocalEvents + + try: + events = LocalEvents(store.path) + except (EventsUnavailable, NotImplementedError): + if mode == 'events': + raise Failure('listener_events_unavailable') from None + print('msg listen: native events unavailable; using polling.', + file=sys.stderr, flush=True) + else: + with events: + return await stream( + state, args, fetch, 'local:' + store.username, + wait_for_change=events.wait, + ) return await stream(state, args, fetch, 'local:' + store.username) if args.action == 'create': result = store.create(args.name) @@ -221,6 +259,8 @@ async def fetch(cursor, tail=False): async def run_remote(client, args): require(not args.offline, 'offline_remote_conflict') + if args.command == 'listen': + require(getattr(args, 'wait_mode', 'auto') != 'events', 'listener_events_unavailable') if args.user: await client.require_username(args.user) if args.command == 'listen' and not args.agent: @@ -254,7 +294,10 @@ async def fetch(cursor, tail=False): async def fetch(cursor, tail=False): return await store.inbox(name, cursor=cursor, tail=tail) - return await stream(client.state, args, fetch, 'remote') + return await stream( + client.state, args, fetch, 'remote', + resolve_legacy=lambda message_id: store.read(name, message_id), + ) if args.action == 'create': result = await store.create(args.name) elif args.action == 'show': diff --git a/src/msg/client_events.py b/src/msg/client_events.py new file mode 100644 index 00000000..635cc537 --- /dev/null +++ b/src/msg/client_events.py @@ -0,0 +1,99 @@ +"""Local database change hints. The mailbox, not the hint, is the durable queue.""" + +from __future__ import annotations + +import asyncio +import ctypes +import os +import struct +import sys +from pathlib import Path + + +class EventsUnavailable(OSError): + """The platform cannot provide native local file notifications.""" + + +class LocalEvents: + """Subscribe before reading, then wait without a timer or database polling. + + Linux inotify watches the private mailbox directory. Notifications carry no + message bodies. A lost/overflowed hint causes a fresh mailbox read; a removed + watch fails closed. Every reader owns its descriptor, so readers cannot steal + notifications from each other. Other platforms use explicit polling fallback. + """ + + def __init__(self, database: Path): + if sys.platform != 'linux': + raise EventsUnavailable('Native local events require Linux.') + library = ctypes.CDLL(None, use_errno=True) + try: + initialize = library.inotify_init1 + add_watch = library.inotify_add_watch + except AttributeError: + raise EventsUnavailable('inotify is not available.') from None + initialize.argtypes, initialize.restype = [ctypes.c_int], ctypes.c_int + add_watch.argtypes = [ctypes.c_int, ctypes.c_char_p, ctypes.c_uint32] + add_watch.restype = ctypes.c_int + self._loop = asyncio.get_running_loop() + self._changed = asyncio.Event() + self._name = os.fsencode(database.name) + self._lost = False + self._descriptor = initialize(os.O_NONBLOCK | os.O_CLOEXEC) + if self._descriptor < 0: + raise EventsUnavailable(ctypes.get_errno(), 'Cannot initialize local events.') + # MODIFY, ATTRIB, MOVED_FROM/TO, CREATE, DELETE, DELETE_SELF, MOVE_SELF. + # Do not watch OPEN/CLOSE: a read-only mailbox scan also opens SQLite. + mask = 0x2 | 0x4 | 0x40 | 0x80 | 0x100 | 0x200 | 0x400 | 0x800 + try: + if add_watch(self._descriptor, os.fsencode(database.parent), mask) < 0: + raise OSError(ctypes.get_errno(), 'Cannot watch the local mailbox.') + self._loop.add_reader(self._descriptor, self._ready) + except BaseException: + os.close(self._descriptor) + self._descriptor = -1 + raise + + def _ready(self): + try: + while True: + data = os.read(self._descriptor, 65536) + if not data: + self._lost = True + self._changed.set() + return + offset = 0 + while offset < len(data): + _, mask, _, length = struct.unpack_from('=iIII', data, offset) + offset += 16 + name = data[offset : offset + length].rstrip(b'\0') + offset += length + # UNMOUNT/IGNORED or a removed/moved directory invalidates + # the subscription. Never silently sleep on a dead watch. + if mask & (0x2000 | 0x8000 | 0x400 | 0x800): + self._lost = True + if self._lost or mask & 0x4000 or name == self._name: + self._changed.set() + except BlockingIOError: + pass + except OSError: + self._lost = True + self._changed.set() + + async def wait(self): + await self._changed.wait() + self._changed.clear() + if self._lost: + raise OSError('Local event subscription was lost; restart the listener.') + + def close(self): + if self._descriptor >= 0: + self._loop.remove_reader(self._descriptor) + os.close(self._descriptor) + self._descriptor = -1 + + def __enter__(self): + return self + + def __exit__(self, *_): + self.close() diff --git a/src/msg/client_listener.py b/src/msg/client_listener.py index 3bc9ab5b..a8c1dbc5 100644 --- a/src/msg/client_listener.py +++ b/src/msg/client_listener.py @@ -27,7 +27,7 @@ def _safe_path(path: Path) -> None: info = path.stat() if not stat.S_ISREG(info.st_mode) or info.st_uid != os.getuid(): raise Failure('unsafe_listener_checkpoint') - if info.st_mode & 0o077: + if info.st_mode & 0o077 or info.st_nlink != 1: raise Failure('unsafe_listener_checkpoint') @@ -47,7 +47,12 @@ def _checkpoint_lock(path: Path | None): raise Failure('unsafe_listener_checkpoint') from None try: info = os.fstat(descriptor) - if not stat.S_ISREG(info.st_mode) or info.st_uid != os.getuid() or info.st_mode & 0o077: + if ( + not stat.S_ISREG(info.st_mode) + or info.st_uid != os.getuid() + or info.st_mode & 0o077 + or info.st_nlink != 1 + ): raise Failure('unsafe_listener_checkpoint') try: fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB) @@ -67,6 +72,13 @@ async def exclusive(fetch, *, cursor_file=None, **options): return exclusive +def _event_id(item): + value = item.get('id') + if type(value) not in (str, int) or value == '': + raise Failure('invalid_listener_page') + return value + + def _load(path: Path | None, binding: dict, cursor: str | None) -> dict: state = {'version': 1, 'context': binding, 'cursor': cursor, 'pending': [], 'has_more': False} if path is None: @@ -77,14 +89,27 @@ def _load(path: Path | None, binding: dict, cursor: str | None) -> dict: try: saved = json.loads(path.read_text()) valid = ( - saved.get('version') == 1 + saved.get('version') in (1, 2) and saved.get('context') == binding and isinstance(saved.get('pending'), list) and isinstance(saved.get('has_more'), bool) and (saved.get('cursor') is None or isinstance(saved.get('cursor'), str)) and all(isinstance(item, dict) for item in saved['pending']) ) - except OSError, ValueError, AttributeError, KeyError: + if saved.get('version') == 2: + valid = ( + valid + and not saved['pending'] + and isinstance(saved.get('processed'), list) + and all(type(value) in (str, int) and value != '' for value in saved['processed']) + and isinstance(saved.get('legacy_pending'), list) + and all( + isinstance(item, dict) and set(item) == {'id'} + and type(item['id']) in (str, int) and item['id'] != '' + for item in saved['legacy_pending'] + ) + ) + except (OSError, ValueError, AttributeError, KeyError, TypeError): raise Failure('invalid_listener_checkpoint') from None if not valid: raise Failure('listener_checkpoint_context_mismatch') @@ -93,10 +118,16 @@ def _load(path: Path | None, binding: dict, cursor: str | None) -> dict: return saved +def _snapshot(state): + # Remote checkpoints contain positions and delivered IDs, never private + # bodies. The uncommitted page is fetched again under current authority. + return {**state, 'pending': []} if state['version'] == 2 else state + + def _save(path: Path | None, state: dict) -> None: if path is not None: _safe_path(path) - durable_write(path, json.dumps(state, ensure_ascii=False).encode()) + durable_write(path, json.dumps(_snapshot(state), ensure_ascii=False).encode()) def _binding(context: dict, event_types) -> dict: @@ -140,17 +171,26 @@ async def listen( max_pages: int | None = None, from_now: bool = False, output=None, + wait_for_change=None, + max_idle_interval: float | None = None, + cursor_only: bool = False, + resolve_legacy=None, ) -> int: - """Poll until cancelled, emitting and flushing one JSON object per matching event. + """Drain a source, then wait for a native hint or a bounded polling delay. - ``fetch(cursor, tail=False)`` must return items, cursor and has_more. Cursor - checkpoints contain the unprinted remainder of a page, so an exact event - limit or restart never skips messages. Delivery is at least once if a process - dies between printing and saving. Use a different checkpoint for every reader. + Subscribe to hints *before* calling this function. Hints carry no messages: + every wake reads the durable source. With ``cursor_only``, stable event IDs + and a page-start cursor replace cached bodies; restart rechecks source ACLs. + An old body checkpoint requires ``resolve_legacy(id)`` to reread pending mail. + Delivery is at least once across a crash between flush and checkpoint save. + Output/flush is not a task ACK and cannot guarantee exactly-once execution. """ if not math.isfinite(interval) or interval <= 0: raise Failure('invalid_listener_interval') - if max_events is not None and (isinstance(max_events, bool) or max_events < 1): + idle_limit = interval if max_idle_interval is None else max_idle_interval + if not math.isfinite(idle_limit) or idle_limit < interval: + raise Failure('invalid_listener_interval') + if max_events is not None and (type(max_events) is not int or max_events < 1): raise Failure('invalid_listener_event_limit') if max_pages is not None: if type(max_pages) is not int or max_pages < 1: @@ -164,51 +204,112 @@ async def listen( state = _load(cursor_file, binding, cursor) if from_now and (cursor is not None or (cursor_file is not None and cursor_file.exists())): raise Failure('listener_start_position_conflict') + if state['version'] == 2 and not cursor_only: + raise Failure('listener_checkpoint_context_mismatch') + if cursor_only and state['version'] == 1: + if state['pending'] and resolve_legacy is None: + raise Failure('listener_checkpoint_requires_revalidation') + state = { + **state, 'version': 2, 'pending': [], 'processed': [], + 'legacy_pending': [{'id': _event_id(item)} for item in state['pending']], + } + if state.get('legacy_pending') and resolve_legacy is None: + raise Failure('listener_checkpoint_requires_revalidation') stream = sys.stdout if output is None else output emitted = 0 pages = 0 retry_delay = interval + idle_delay = interval + saved_snapshot = None - async def read(tail=False): + def persist(): + nonlocal saved_snapshot + snapshot = json.dumps(_snapshot(state), ensure_ascii=False) + if snapshot != saved_snapshot: + _save(cursor_file, state) + saved_snapshot = snapshot + + async def retry(call): nonlocal retry_delay while True: try: - value = _page(await fetch(state['cursor'], tail=tail)) + value = await call() retry_delay = interval return value except Failure as exc: if not exc.retryable: raise - except OSError, TimeoutError, httpx.TransportError: + except (OSError, TimeoutError, httpx.TransportError): pass print('msg listen: connection interrupted; retrying.', file=sys.stderr, flush=True) await asyncio.sleep(min(retry_delay, 30.0)) retry_delay = min(retry_delay * 2, 30.0) + async def read(tail=False): + return _page(await retry(lambda: fetch(state['cursor'], tail=tail))) + + def emit(item): + nonlocal emitted + event_type = item.get('event_type', item.get('type')) + if not types or event_type in types: + stream.write(json.dumps(item, ensure_ascii=False, separators=(',', ':')) + '\n') + stream.flush() + emitted += 1 + + if cursor_only: + # Remove legacy plaintext before contacting the server. Failure to + # authorize a reread must never fall back to the cached private body. + persist() + while state['legacy_pending']: + reference = state['legacy_pending'][0] + item = await retry(lambda: resolve_legacy(reference['id'])) + if not isinstance(item, dict) or _event_id(item) != reference['id']: + raise Failure('invalid_listener_page') + emit(item) + state['legacy_pending'].pop(0) + persist() + if max_events is not None and emitted >= max_events: + return emitted + if from_now: page = await read(tail=True) state.update(cursor=page['cursor'], pending=[], has_more=False) - _save(cursor_file, state) + persist() while True: + page_end = state['cursor'] + activity = bool(state['pending']) if not state['pending']: page = await read() pages += 1 if page['has_more'] and page['cursor'] == state['cursor']: raise Failure('invalid_listener_page') - state.update(cursor=page['cursor'], pending=page['items'], has_more=page['has_more']) + if cursor_only: + ids = [_event_id(item) for item in page['items']] + if len(ids) != len(set(ids)): + raise Failure('invalid_listener_page') + page_end = page['cursor'] + state.update(pending=page['items'], has_more=page['has_more']) + if not cursor_only: + state['cursor'] = page_end + activity = bool(page['items']) + processed = set(state.get('processed', ())) while state['pending']: item = state['pending'][0] - event_type = item.get('event_type', item.get('type')) - if not types or event_type in types: - stream.write(json.dumps(item, ensure_ascii=False, separators=(',', ':')) + '\n') - stream.flush() - emitted += 1 + if not cursor_only or _event_id(item) not in processed: + emit(item) + if cursor_only: + state['processed'].append(_event_id(item)) + processed.add(_event_id(item)) state['pending'] = state['pending'][1:] - _save(cursor_file, state) + if cursor_only and not state['pending']: + state.update(cursor=page_end, processed=[]) + persist() if max_events is not None and emitted >= max_events: return emitted - _save(cursor_file, state) + if cursor_only: + state.update(cursor=page_end, processed=[]) + persist() if max_pages is not None and pages >= max_pages: if state['has_more']: print( @@ -221,4 +322,11 @@ async def read(tail=False): continue if once: return emitted - await asyncio.sleep(interval) + if activity: + idle_delay = interval + if wait_for_change is not None: + await wait_for_change() + else: + await asyncio.sleep(idle_delay) + if not activity: + idle_delay = min(idle_delay * 2, idle_limit) diff --git a/src/msg/core/errors.py b/src/msg/core/errors.py index decc0fb0..03e2e12a 100644 --- a/src/msg/core/errors.py +++ b/src/msg/core/errors.py @@ -27,6 +27,8 @@ 'subagent_account_mismatch': 'Subagents can exchange internal messages only within the same account.', 'subagent_archived': 'This subagent is archived; its existing messages are preserved.', 'subagent_private_namespace_conflict': 'The existing mailbox is not owned and private; choose or repair the namespace explicitly.', + 'listener_events_unavailable': 'Native event waiting is not available for this source. Use auto or poll; remote event push is not implemented.', + 'listener_checkpoint_requires_revalidation': 'This old checkpoint has cached events without a safe reread. Keep it and use a new cursor file to reread authorized history; deduplicate event IDs.', 'cursor_in_use': 'Another listener is using this cursor file. Give each listener its own file.', 'listener_start_position_conflict': 'Use --from-now with a new cursor file; otherwise resume the existing cursor.', 'handle_rename_cooldown': 'You can change your username once every seven days.', diff --git a/tests/test_client_listener_cli.py b/tests/test_client_listener_cli.py new file mode 100644 index 00000000..ca6e01db --- /dev/null +++ b/tests/test_client_listener_cli.py @@ -0,0 +1,17 @@ +import pytest + +from msg.cli import parser + +def test_wait_modes_are_explicit(): + assert parser().parse_args(['listen']).wait_mode == 'auto' + assert parser().parse_args(['listen', '--wait-mode', 'poll']).wait_mode == 'poll' + assert parser().parse_args(['listen', '--wait-mode', 'events']).wait_mode == 'events' + + +async def test_remote_events_fail_before_any_network_call(): + from types import SimpleNamespace + from msg.client_agent_cli import run_remote + from msg.core.errors import Failure + + with pytest.raises(Failure, match='listener_events_unavailable'): + await run_remote(None, SimpleNamespace(command='listen', offline=False, wait_mode='events')) diff --git a/tests/test_client_listener_reliability.py b/tests/test_client_listener_reliability.py new file mode 100644 index 00000000..5a8c9a65 --- /dev/null +++ b/tests/test_client_listener_reliability.py @@ -0,0 +1,306 @@ +"""Measure concrete listener boundaries; never infer a harness speed ranking.""" + +import asyncio +import io +import json +import os +import signal +import sys + +import pytest + +from msg import client_listener +from msg.client_events import LocalEvents +from msg.client_listener import listen +from msg.core.errors import Failure + +CONTEXT = {'server': 'https://example.test', 'subject': 'alice', 'agent': 'worker'} + + +def page(items=(), cursor='end', has_more=False): + return {'items': list(items), 'cursor': cursor, 'has_more': has_more} + + +async def test_remote_checkpoint_contains_no_bodies_and_rereads_pending(tmp_path): + checkpoint = tmp_path / 'cursor.json' + calls = [] + body = 'private original' + + async def fetch(cursor, tail=False): + calls.append(cursor) + return page([{'id': str(i), 'message': body} for i in range(3)]) + + output = io.StringIO() + await listen(fetch, context=CONTEXT, cursor_only=True, cursor_file=checkpoint, + max_events=1, output=output) + saved = json.loads(checkpoint.read_text()) + assert saved['version'] == 2 and saved['cursor'] is None + assert saved['processed'] == ['0'] and saved['pending'] == [] + assert body not in checkpoint.read_text() + body = 'fresh authorized result' + await listen(fetch, context=CONTEXT, cursor_only=True, cursor_file=checkpoint, + max_events=2, output=output) + results = [json.loads(line) for line in output.getvalue().splitlines()] + assert [item['id'] for item in results] == ['0', '1', '2'] + assert [item['message'] for item in results[1:]] == [body, body] + assert calls == [None, None] + assert json.loads(checkpoint.read_text())['cursor'] == 'end' + assert body not in checkpoint.read_text() + + +async def test_remote_resume_denied_has_no_cached_output(tmp_path): + checkpoint = tmp_path / 'cursor.json' + + async def fetch(cursor, tail=False): + return page([{'id': '1', 'message': 'secret'}, {'id': '2', 'message': 'secret'}]) + + await listen(fetch, context=CONTEXT, cursor_only=True, cursor_file=checkpoint, + max_events=1, output=io.StringIO()) + saved = checkpoint.read_bytes() + + async def denied(cursor, tail=False): + raise Failure('permission_denied') + + output = io.StringIO() + with pytest.raises(Failure, match='permission_denied'): + await listen(denied, context=CONTEXT, cursor_only=True, + cursor_file=checkpoint, once=True, output=output) + assert output.getvalue() == '' + assert checkpoint.read_bytes() == saved + assert b'secret' not in saved + with pytest.raises(Failure, match='context_mismatch'): + await listen(denied, context={**CONTEXT, 'subject': 'bob'}, cursor_only=True, + cursor_file=checkpoint, once=True, output=output) + + +async def test_legacy_remote_cache_is_scrubbed_before_authorization(tmp_path): + checkpoint = tmp_path / 'cursor.json' + + async def fetch(cursor, tail=False): + return page([{'id': '1', 'message': 'secret'}, {'id': '2', 'message': 'secret'}]) + + await listen(fetch, context=CONTEXT, cursor_file=checkpoint, + max_events=1, output=io.StringIO()) + assert 'secret' in checkpoint.read_text() + output = io.StringIO() + + async def denied(message_id): + assert message_id == '2' and 'secret' not in checkpoint.read_text() + raise Failure('credential_ceiling') + + with pytest.raises(Failure, match='credential_ceiling'): + await listen(fetch, context=CONTEXT, cursor_only=True, resolve_legacy=denied, + cursor_file=checkpoint, once=True, output=output) + assert output.getvalue() == '' + assert json.loads(checkpoint.read_text())['legacy_pending'] == [{'id': '2'}] + + async def authorized(message_id): + return {'id': message_id, 'message': 'fresh'} + + await listen(fetch, context=CONTEXT, cursor_only=True, resolve_legacy=authorized, + cursor_file=checkpoint, max_events=1, output=output) + assert json.loads(output.getvalue())['message'] == 'fresh' + assert 'fresh' not in checkpoint.read_text() + + +async def test_legacy_account_events_require_safe_migration(tmp_path): + checkpoint = tmp_path / 'cursor.json' + + async def fetch(cursor, tail=False): + return page([{'id': '1'}, {'id': '2', 'message': 'private'}]) + + await listen(fetch, context=CONTEXT, cursor_file=checkpoint, + max_events=1, output=io.StringIO()) + output = io.StringIO() + with pytest.raises(Failure, match='requires_revalidation'): + await listen(fetch, context=CONTEXT, cursor_only=True, + cursor_file=checkpoint, once=True, output=output) + assert output.getvalue() == '' + + +async def test_idle_backoff_is_bounded_without_repeated_checkpoint_writes(tmp_path, monkeypatch): + delays, writes = [], [] + original = client_listener._save + + def save(path, state): + writes.append(state['cursor']) + original(path, state) + + async def fetch(cursor, tail=False): + return page(cursor='unchanged') + + async def sleep(delay): + delays.append(delay) + if len(delays) == 6: + raise asyncio.CancelledError + + monkeypatch.setattr(client_listener, '_save', save) + monkeypatch.setattr(asyncio, 'sleep', sleep) + with pytest.raises(asyncio.CancelledError): + await listen(fetch, context=CONTEXT, cursor_file=tmp_path / 'cursor.json', + max_idle_interval=8, output=io.StringIO()) + assert delays == [1, 2, 4, 8, 8, 8] + assert writes == ['unchanged'] + + +async def test_idle_backoff_resets_on_activity(monkeypatch): + calls, delays = 0, [] + + async def fetch(cursor, tail=False): + nonlocal calls + calls += 1 + return page([{'id': 'result'}] if calls == 3 else [], cursor=str(calls)) + + async def sleep(delay): + delays.append(delay) + if len(delays) == 5: + raise asyncio.CancelledError + + monkeypatch.setattr(asyncio, 'sleep', sleep) + with pytest.raises(asyncio.CancelledError): + await listen(fetch, context=CONTEXT, max_idle_interval=30, output=io.StringIO()) + assert delays == [1, 2, 1, 1, 2] + + +@pytest.mark.parametrize('limit', [0, -1, True, 1.5, '2']) +async def test_invalid_event_limit_fails_before_fetch(limit): + async def fetch(cursor, tail=False): + raise AssertionError('Invalid input must not contact a source.') + + with pytest.raises(Failure, match='invalid_listener_event_limit'): + await listen(fetch, context=CONTEXT, max_events=limit) + + +@pytest.mark.parametrize('bad', [None, True, '', [], {}]) +async def test_cursor_only_requires_stable_event_ids(bad): + async def fetch(cursor, tail=False): + return page([{'id': bad}]) + + output = io.StringIO() + with pytest.raises(Failure, match='invalid_listener_page'): + await listen(fetch, context=CONTEXT, cursor_only=True, once=True, output=output) + assert output.getvalue() == '' + + +@pytest.mark.skipif(sys.platform != 'linux', reason='Native inotify test') +async def test_native_events_block_idle_and_wake_all_readers(tmp_path): + database = tmp_path / 'mailbox.sqlite3' + database.write_bytes(b'') + calls = [0, 0] + ready = [asyncio.Event(), asyncio.Event()] + outputs = [io.StringIO(), io.StringIO()] + with LocalEvents(database) as first, LocalEvents(database) as second: + async def reader(index, events): + async def fetch(cursor, tail=False): + calls[index] += 1 + ready[index].set() + return page([{'id': '1'}] if database.read_bytes() else []) + return await listen(fetch, context=CONTEXT, max_events=1, + wait_for_change=events.wait, output=outputs[index]) + + tasks = [asyncio.create_task(reader(0, first)), asyncio.create_task(reader(1, second))] + try: + await asyncio.wait_for(asyncio.gather(*(event.wait() for event in ready)), 2) + (tmp_path / 'unrelated.txt').write_text('not a mailbox change') + await asyncio.sleep(0.04) + assert calls == [1, 1] and all(not task.done() for task in tasks) + database.write_bytes(b'committed message') + assert await asyncio.wait_for(asyncio.gather(*tasks), 2) == [1, 1] + assert calls == [2, 2] + finally: + for task in tasks: + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + assert first._descriptor == second._descriptor == -1 + assert [json.loads(output.getvalue())['id'] for output in outputs] == ['1', '1'] + + +@pytest.mark.skipif(sys.platform != 'linux', reason='Native inotify test') +async def test_write_between_subscription_and_wait_is_not_lost(tmp_path): + database = tmp_path / 'mailbox.sqlite3' + database.touch() + with LocalEvents(database) as events: + database.write_text('already committed') + await asyncio.wait_for(events.wait(), 2) + + +@pytest.mark.skipif(sys.platform != 'linux', reason='Native inotify test') +async def test_removed_native_watch_fails_instead_of_sleeping_forever(tmp_path): + directory = tmp_path / 'mail' + directory.mkdir() + with LocalEvents(directory / 'mailbox.sqlite3') as events: + directory.rmdir() + with pytest.raises(OSError, match='subscription was lost'): + await asyncio.wait_for(events.wait(), 2) + + +async def test_hardlinked_checkpoint_is_rejected(tmp_path): + target = tmp_path / 'target' + target.write_text('{}') + target.chmod(0o600) + checkpoint = tmp_path / 'cursor.json' + os.link(target, checkpoint) + + async def fetch(cursor, tail=False): + raise AssertionError('Unsafe checkpoint must not contact a source.') + + with pytest.raises(Failure, match='unsafe_listener_checkpoint'): + await listen(fetch, context=CONTEXT, cursor_file=checkpoint, once=True) + + +@pytest.mark.skipif(sys.platform != 'linux', reason='SIGKILL fault injection') +@pytest.mark.parametrize('phase', ['before_flush', 'after_flush', 'after_checkpoint']) +async def test_sigkill_recovery_no_message_loss_and_explicit_replay(tmp_path, phase): + checkpoint = tmp_path / 'cursor.json' + script = r''' +import asyncio, os, signal, sys +from pathlib import Path +from msg import client_listener as module +checkpoint, phase = Path(sys.argv[1]), sys.argv[2] +original = module._save + +def save(path, state): + original(path, state) + if phase == 'after_checkpoint' and state.get('processed') == ['1']: + os.kill(os.getpid(), signal.SIGKILL) +module._save = save + +class Output: + def write(self, value): + if phase == 'before_flush': + os.kill(os.getpid(), signal.SIGKILL) + sys.stdout.write(value) + def flush(self): + sys.stdout.flush() + if phase == 'after_flush': + os.kill(os.getpid(), signal.SIGKILL) + +async def fetch(cursor, tail=False): + return {'items': [] if cursor == 'end' else [{'id': str(i)} for i in range(1,4)], + 'cursor': 'end', 'has_more': False} +asyncio.run(module.listen(fetch, context={'agent': 'worker'}, cursor_only=True, + cursor_file=checkpoint, once=True, output=Output())) +''' + + async def run(mode): + process = await asyncio.create_subprocess_exec( + sys.executable, '-c', script, str(checkpoint), mode, + stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, + ) + try: + stdout, stderr = await asyncio.wait_for(process.communicate(), 5) + assert not stderr, stderr.decode() + return process.returncode, [json.loads(line)['id'] for line in stdout.splitlines()] + finally: + if process.returncode is None: + process.kill() + await process.wait() + + code, first = await run(phase) + assert code == -signal.SIGKILL + code, second = await run('resume') + assert code == 0 + delivered = first + second + assert set(delivered) == {'1', '2', '3'} + assert len(delivered) == (4 if phase == 'after_flush' else 3) + assert json.loads(checkpoint.read_text())['cursor'] == 'end' diff --git a/tests/test_client_listener_remote.py b/tests/test_client_listener_remote.py new file mode 100644 index 00000000..71614c14 --- /dev/null +++ b/tests/test_client_listener_remote.py @@ -0,0 +1,135 @@ +"""Listener restart must not turn a cached page into an authority bypass.""" + +import asyncio +import io +import json +import os +import time +from pathlib import Path + +import httpx +import pytest +from test_client_link import args, linked, run_command +from test_client_subagents_remote import client_for + +from msg.client_listener import listen +from msg.client_subagents_remote import RemoteAgents +from msg.core.errors import Failure +from msg.transports.http import create_app + + +def context(client, label): + return {'server': client.state.server, 'subject': client.state.subject, + 'source': 'remote', 'agent': label} + + +async def test_same_account_lesser_authority_and_revocation_cannot_replay_cache(installed, tmp_path): + app, _ = installed + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=create_app(app)), base_url=app.settings.service_url, + ) as http: + owner, helper, _, _ = await linked(app, tmp_path, http) + full, limited = RemoteAgents(owner), RemoteAgents(helper) + assert owner.state.subject == helper.state.subject + await full.create('private') + start = await full.inbox('private', tail=True) + for number in range(2): + await full.send('private', 'private', 'owner-only-result', message_id=f'private-{number}') + checkpoint = tmp_path / 'owner.json' + + async def owner_fetch(cursor, tail=False): + return await full.inbox('private', cursor=cursor, tail=tail) + + await listen(owner_fetch, context=context(owner, 'private'), cursor_only=True, + cursor=start['cursor'], cursor_file=checkpoint, max_events=1, + output=io.StringIO()) + assert 'owner-only-result' not in checkpoint.read_text() + + async def limited_fetch(cursor, tail=False): + return await limited.inbox('private', cursor=cursor, tail=tail) + + output = io.StringIO() + with pytest.raises(Failure, match='credential_ceiling'): + await listen(limited_fetch, context=context(helper, 'private'), cursor_only=True, + cursor_file=checkpoint, once=True, output=output) + assert output.getvalue() == '' + + start = await limited.inbox('reviewer', tail=True) + for number in range(2): + await full.send('reviewer-lead', 'reviewer', 'task-result', message_id=f'task-{number}') + checkpoint = tmp_path / 'helper.json' + + async def helper_fetch(cursor, tail=False): + return await limited.inbox('reviewer', cursor=cursor, tail=tail) + + await listen(helper_fetch, context=context(helper, 'reviewer'), cursor_only=True, + cursor=start['cursor'], cursor_file=checkpoint, max_events=1, + output=io.StringIO()) + assert 'task-result' not in checkpoint.read_text() + await run_command(owner, args('revoke', name='reviewer')) + with pytest.raises(Failure, match='authority_source_inactive'): + await listen(helper_fetch, context=context(helper, 'reviewer'), cursor_only=True, + cursor_file=checkpoint, once=True, output=output) + assert output.getvalue() == '' + + +async def test_remote_idle_cost_records_actual_http_requests(installed, tmp_path): + """ASGI HTTP + real PostgreSQL, not a wide-area-network latency benchmark.""" + app, _ = installed + client, http = await client_for(app, tmp_path / 'idle', 'idle-bench') + store = RemoteAgents(client) + rows = [] + try: + await store.create('worker') + initial = await store.inbox('worker', tail=True) + counts = {'requests': 0} + + async def count_request(request): + counts['requests'] += 1 + + http.event_hooks['request'].append(count_request) + for mode, maximum in [('fixed_poll', 0.05), ('adaptive_poll', 0.4)]: + ready = asyncio.Event() + reads = 0 + + async def fetch(cursor, tail=False): + nonlocal reads + result = await store.inbox('worker', cursor=cursor, tail=tail) + reads += 1 + ready.set() + return result + + output = io.StringIO() + task = asyncio.create_task(listen( + fetch, context=context(client, 'worker'), cursor=initial['cursor'], + cursor_only=True, interval=0.05, max_idle_interval=maximum, output=output, + )) + try: + await asyncio.wait_for(ready.wait(), 10) + requests_before, reads_before = counts['requests'], reads + started, cpu = time.perf_counter(), time.process_time() + await asyncio.sleep(2) + finally: + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + row = { + 'mode': mode, 'status': 'measured', 'transport': 'in_process_http', + 'database': 'real_postgresql', 'poll_interval_seconds': 0.05, + 'max_idle_interval_seconds': maximum, + 'idle_seconds': time.perf_counter() - started, + 'idle_cpu_seconds_client_and_server': time.process_time() - cpu, + 'idle_http_requests_started': counts['requests'] - requests_before, + 'idle_source_reads_completed': reads - reads_before, + } + assert row['idle_http_requests_started'] > 0 and output.getvalue() == '' + rows.append(row) + finally: + await http.aclose() + target = Path(os.environ.get('MSG_BENCHMARK_OUTPUT', tmp_path)) + target.mkdir(parents=True, exist_ok=True) + (target / 'remote-idle.json').write_text(json.dumps({ + 'results': rows, + 'warning': 'Two-second idle windows with shortened intervals. No WAN or harness ranking.', + 'remote_events': {'status': 'not_implemented'}, + }, indent=2) + '\n')