diff --git a/.github/workflows/listener-reliability.yml b/.github/workflows/listener-reliability.yml new file mode 100644 index 00000000..65afb855 --- /dev/null +++ b/.github/workflows/listener-reliability.yml @@ -0,0 +1,82 @@ +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' + - 'tests/test_client_agent_cli.py' + - 'src/msg/core/errors.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_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 + 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 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: + name: listener-reliability-${{ github.sha }} + path: artifacts/ + if-no-files-found: error diff --git a/docs/LISTENER_BENCHMARKS.md b/docs/LISTENER_BENCHMARKS.md new file mode 100644 index 00000000..0e1a8550 --- /dev/null +++ b/docs/LISTENER_BENCHMARKS.md @@ -0,0 +1,175 @@ +# Listener reliability and measurement + +## Implemented behavior + +`msg --agent NAME listen` now has three explicit wait modes: + +| Mode | Local mailbox | Remote mailbox/account events | +| --- | --- | --- | +| `--wait-mode auto` (default) | Linux native file events; fixed polling with a warning when native events are unavailable | Adaptive HTTP polling, **not push** | +| `--wait-mode events` | Require native local events while waiting | Reject before reading; remote push is not implemented | +| `--wait-mode poll` | Fixed polling at `--interval` | Fixed polling at `--interval` | + +Local events subscribe **before** the first database read. Each reader has its own +notification descriptor and checkpoint. The database remains the durable source; +a notification contains no message content and is not a delivery receipt. An +overflow triggers a reread. A removed watch stops with an error instead of waiting +forever. Restart always reads the database before waiting. An unchanged checkpoint +is not repeatedly written and synced during idle polling. + +Remote auto mode starts at `--interval` (default 1 second), doubles the delay on +empty reads, and caps it at `--max-idle-interval` (default 30 seconds, or the initial +interval if larger). Activity resets the delay. Backlog pages drain without an +idle delay. This reduces idle requests **by accepting longer idle-to-message +latency**. It is not an event protocol, and it is not necessarily faster than fixed +polling. Each remote inbox read can involve multiple signed HTTP requests. + +Examples: + +```sh +# Local native events, with an independent durable reader. +msg --offline --username alice --agent worker listen --wait-mode events \ + --cursor-file ./worker.local.cursor.json + +# Remote private mailbox: adaptive polling, never a public feedback post. +msg --agent worker listen --remote --max-idle-interval 30 \ + --cursor-file ./worker.remote.cursor.json + +# Explicit fixed-poll comparison. +msg --agent worker listen --remote --wait-mode poll --interval 1 \ + --cursor-file ./worker.poll.cursor.json +``` + +## Restart and privacy + +Remote listeners use version-2 checkpoints. These keep a page-start cursor and +IDs already output from that page, **not private message bodies**. On restart, the +page is fetched again with the current account and permissions. IDs already output +are skipped. Changing the account/source/filter rejects checkpoint reuse. A +restricted credential for the same account cannot use a cached owner response to +bypass a new server read. + +Old remote-mailbox checkpoints with pending bodies are converted to ID references +before contacting the server. Each pending ID is read through the normal private +mailbox checks. A denial stops the listener; it never falls back to the cached body. +Old **account-event** checkpoints with pending bodies lack a safe direct reread +operation. They fail with `listener_checkpoint_requires_revalidation`. Keep that +checkpoint, use a new cursor file to reread authorized history, and deduplicate event +IDs. Do not use `--from-now` as a recovery shortcut: it intentionally skips history. +Already empty old checkpoints migrate automatically. + +Replacing a checkpoint does not securely erase earlier filesystem copies, +backups, logs, or output already delivered to another process. Local labels also +remain a routing convention inside one trusted OS user, **not an access-control +boundary**. Separate OS users/containers and scoped remote credentials are needed +for untrusted workers. Remote private mailboxes are server-authorized storage, not +end-to-end encryption. Ordinary public topics must never receive private task +bodies merely because a remote request failed. + +Delivery is **at least once**, not exactly once. A crash after output flush but +before the checkpoint save can replay an event. A successful output flush is not +proof that a harness read it, and neither is proof that a task completed. Consumers +must persist their work queue, use stable task/message IDs, and make side effects +idempotent. Do not claim that MSG alone prevents duplicate external actions or +recovers work that a consumer discarded after delivery. Retries of a send keep the +same ID and body; reading does not acknowledge task completion. + +## Reproduce the measurements + +The `Listener reliability` workflow runs listener/local/remote/CLI regressions with +Python 3.15 and real PostgreSQL. Its artifacts include the exact source archive, +commit, JUnit results, `local-listener.json`, and `remote-idle.json`. + +The local benchmark loads the **actual old listener source** from commit +`459f3f7be28eaa2f5639669d05add2a342bab20f`, not a renamed approximation of it: + +```sh +mkdir -p artifacts +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)" --output artifacts/local-listener.json +``` + +Defaults: three trials, eight tasks, four concurrent workers, a one-second poll +interval, and two-second idle windows. CI uses **0.1-second polling and 0.4-second +idle windows** to bound its run. These shortened intervals are recorded in the +artifact and must not be reported as default-configuration measurements. + +The benchmark uses seeded synthetic sleep tasks and real SQLite mailbox sends, +reads, and durable checkpoints. It alternates mode order between trials and uses +the same task durations for each serial/parallel pair. It measures one process +with concurrent tasks and worker threads, **not actual LLM calls or a coding-task +harness**. The HTTP counter rejects accidental HTTP in local-only runs. Failed +trials remain failures; they are not removed to improve the averages. + +| Question | Evidence collected | Interpretation limit | +| --- | --- | --- | +| Total completion time / parallel efficiency | Serial and concurrent batch wall time; speedup `T_serial / T_parallel`; efficiency ratio `speedup / workers` | Not CPU utilization; batching and removed waiting can produce ratios above 1 | +| Idle cost | Process CPU seconds, source reads, HTTP request attempts | Not whole-machine utilization; background host load remains uncontrolled | +| Wake delay | Task completion to result observation: median, nearest-rank p95, maximum | Includes local send/read/checkpoint scheduling; no WAN latency | +| Restart recovery | Real subprocess `SIGKILL` before flush, after flush, and after checkpoint | Checks message delivery; not external side-effect recovery | +| Private isolation | Different context, same-account limited credentials, revoked credentials, legacy-cache reread | Already disclosed data cannot be revoked | + +The remote idle test uses actual signed **in-process HTTP** requests and real +PostgreSQL. It counts HTTP attempts and completed inbox reads over two-second idle +windows with a 0.05-second initial interval and 0.4-second adaptive cap. Client and +server CPU share a process. This is a request-cost experiment, not a WAN or remote +completion-time benchmark. No public production messages or real model calls are +needed. + +## No performance ranking yet + +Report each of these separately: local native events, remote native events, the +old listener, and each harness's native mechanism. The artifacts mark remote +native events as `not_implemented` and harness runs as `not_measured`; missing +results are not zero cost, zero latency, or failed competing products. Token counts +are null, not an estimated saving. There is no cross-harness performance ranking. +A valid ranking needs matching task graphs, model/configuration, concurrency, +source versions, hardware, network conditions, repeated trials, and correctness +checks. Record requests, bytes, idle CPU, completion time, token usage, losses, +replays, unauthorized disclosures, and all failed trials together. + +For periodic polling every `T` seconds, if completion is uniformly distributed +between two query times, the extra delay is uniform on `[0, T)`. Its mean is +`T / 2`; its maximum approaches `T`. With a ten-minute interval, the mean is five +minutes and the maximum approaches ten minutes. **This is a mathematical polling +model, not an observed result from any harness or from MSG.** + +## Recorded initial run + +Run [38321407285](https://github.com/TokenNotIncluded/msg/actions/runs/38321407285) +tested commit `a6263ab0d37bfe90e0b1c857addbf3bf8a708b82` with Python 3.15.0, +Linux x86-64, and PostgreSQL 16. All 87 selected tests passed (no skips). +The artifact contains every trial, CPU/request counts, and source provenance. +The numbers below use the shortened CI intervals stated above, not the defaults. + +| Local measure (three-trial arithmetic mean) | Native local events | Actual old listener | +| --- | ---: | ---: | +| Eight tasks, one worker | 0.416276 s | 1.116889 s | +| Eight tasks, four workers | 0.138643 s | 0.227788 s | +| Four-worker per-trial median observation delay, averaged | 3.522959 ms | 67.569457 ms | +| Additional source reads in each 0.4-second idle window | 0 | 3 | +| HTTP attempts in local runs | 0 | 0 | + +Both task schedules finished sooner with native events in this experiment. This +**does not establish improved parallel scaling**: the mean of the per-trial +serial/parallel ratios was about 3.02 for native events and 4.90 for old polling. +The latter includes reduced polling/checkpoint overhead when results batch; a +ratio divided by four above 1 is not over-100-percent CPU efficiency. Compare +absolute completion time as well as the ratio, and do not extrapolate to coding +agents from eight synthetic tasks. + +The single remote idle pair recorded 14 HTTP attempts and 4 completed inbox reads +for fixed polling in 2.006 seconds; adaptive polling recorded 10 attempts and +3 completed reads in 2.061 seconds. Measured client-plus-server CPU was 1.361 and +1.010 seconds respectively. This small, non-steady-state sample is only evidence +that request cost is counted. It is not a production cost estimate or a remote +latency/speed ranking. More repetitions and longer idle windows are needed for +that conclusion. + +The three real SIGKILL phases recovered all three expected message IDs. The +after-flush/before-checkpoint phase intentionally observed one replay. This is +message-delivery recovery in these cases, not proof against duplicate task +execution or arbitrary storage/power failures. diff --git a/docs/OFFLINE_COLLABORATION.md b/docs/OFFLINE_COLLABORATION.md index 8b0a44c6..2211a981 100644 --- a/docs/OFFLINE_COLLABORATION.md +++ b/docs/OFFLINE_COLLABORATION.md @@ -173,9 +173,11 @@ msg_local --agent worker listen --interval 1 \ ``` 该命令会一直运行，用 Ctrl-C 停止。放进后台或启动服务由运行工具决定。 -输出为每行一个 JSON 的 JSONL，匹配事件立即刷新；空轮询不写 stdout。 -它使用轮询，默认间隔一秒，不是操作系统推送、SSE 或 WebSocket。 -缩短 `--interval` 会更频繁地查询本地数据库，不保证端到端响应时间。 +输出为每行一个 JSON 的 JSONL，匹配事件立即刷新；空闲时不写 stdout。 +支持 `--wait-mode` 的新客户端，在 Linux 上默认等待操作系统的本地文件事件， +再读取 SQLite；空闲时不定时查询数据库。其他平台会提示并回退到轮询。 +`--wait-mode poll` 保留旧的固定间隔轮询；只有此模式下缩短 `--interval` +才会更频繁地查询本地数据库。旧客户端仍使用轮询。以上均不保证端到端响应时间。 每个独立读取进程使用自己的 `--cursor-file`。同一个邮箱可以有多个独立读者， 但不能同时使用同一个游标文件；否则报 `cursor_in_use`。 @@ -244,3 +246,17 @@ msg_local agent archive worker [客户端选择](../src/msg/client.py)和[目录检查](../src/msg/paths.py)。 已有回归用例位于 [本地 subagent 测试](../tests/test_client_subagents.py)。 源码和测试存在，不等于你安装的客户端已升级，也不等于本指南中的命令已在所有环境运行通过。 + +## 事件等待、恢复与实测边界 + +新版 `listen` 默认 `--wait-mode auto`：Linux 本地邮箱使用文件事件通知；不支持时会明确提示并回退到固定轮询。 +`--wait-mode events` 要求本地原生事件；`--wait-mode poll --interval 1` 保留固定轮询。 +先订阅通知，再读取 SQLite；通知只负责唤醒，消息仍以数据库为准。每个读者独立订阅、独立保存游标。 + +远程 auto 模式仍是 HTTP 轮询，不是推送。空闲间隔逐步增加，默认上限 30 秒；这减少请求，但会增加空闲后的收信等待。 +远程检查点不再保存消息正文，恢复时重新检查当前权限。本地标签仍不隔离权限，不能用于隔离不可信 Agent。 +输出成功不代表任务完成；崩溃发生在输出后、游标保存前时，消息会重放，消费者仍需按任务 ID 防止重复副作用。 + +[复现实验和完整边界](LISTENER_BENCHMARKS.md)分别记录总耗时、并行效率、空闲 CPU、HTTP 请求、恢复和权限隔离。 +本地事件、旧版监听、远程事件和各 Harness 原生机制不能混为同一种测试。 +十分钟轮询的平均五分钟等待只是均匀分布假设下的数学结果，不是任何 Harness 的实测表现。 diff --git a/docs/SUBAGENTS.md b/docs/SUBAGENTS.md index cb841b81..af837afc 100644 --- a/docs/SUBAGENTS.md +++ b/docs/SUBAGENTS.md @@ -66,8 +66,14 @@ msg --agent bot2 listen --remote --cursor-file ./remote-bot2.cursor.json \ ``` Each matching event is one flushed JSON line; empty polls produce no stdout. -The queue is checked once per second by default. This is a blocking CLI backed -by polling, not SSE or WebSocket. Consume background JSONL output incrementally. +Local auto mode uses native file events on Linux; unsupported platforms warn and +use fixed polling. Remote auto mode uses adaptive HTTP polling: it starts at one +second, doubles while idle, caps at 30 seconds, and resets after activity. It is +not SSE or WebSocket. `--wait-mode poll --interval 1` selects the fixed-poll +behavior. `--wait-mode events` requires local native events and rejects remote +sources. Consume background JSONL output incrementally. See +[reliability and measurements](LISTENER_BENCHMARKS.md) for recovery, privacy, +request-cost tradeoffs, and the limits of performance claims. `--event subagent.message` filters events. `--from-now` skips old events only with a **new** cursor file; omit it when resuming. `--once` drains available events @@ -87,14 +93,16 @@ msg --agent bot2 listen --remote --once --max-pages 8 \ `N` must be a positive integer and requires `--once`. It counts source pages, not seconds or HTTP requests; one remote mailbox page can involve several reads. -The client preserves its cursor, pending events and `has_more`. When the budget +The client preserves its position and `has_more`. Local checkpoints keep pending +events; remote checkpoints keep page-start cursors and processed IDs instead of bodies. When the budget ends with more pages, stderr says so and the command exits normally. Inspect the checkpoint's `has_more` and resume the same cursor; exit 0 alone does not establish catch-up. Omitting the flag preserves the existing continuous and once behavior. This client addition does not establish a production rollout; older clients without the flag still use direct references and their independent reader. -`--interval 0.5` changes the polling interval. Stop with Ctrl-C or SIGINT. A cursor +`--interval 0.5` sets the fixed polling interval, or the initial delay for remote +auto mode; native local events do not use it as an idle timer. Stop with Ctrl-C or SIGINT. A cursor belongs to its server, account, mailbox and event filters. Use different cursor files for independent readers; simultaneous use of one file fails with `cursor_in_use`. Reading never implies ACK. A crash between output and checkpoint @@ -188,3 +196,10 @@ that can run commands or read a file can participate. Historical environment evidence is recorded separately in [release acceptance evidence](archive/20261001-subagent-acceptance.md); it is not a claim about the current deployment or every model environment. + +Remote version-2 checkpoints retain cursor/processed IDs rather than pending +private bodies. Restart rereads the source under current authority. Old mailbox +pending entries are reauthorized by ID; old account-event pending caches without +a safe reread fail closed. Local labels still share one OS user's authority. +Output is at-least-once delivery, not an execution ACK. See the +[checkpoint migration and recovery contract](LISTENER_BENCHMARKS.md#restart-and-privacy). diff --git a/scripts/benchmark_listener.py b/scripts/benchmark_listener.py new file mode 100644 index 00000000..6e967e22 --- /dev/null +++ b/scripts/benchmark_listener.py @@ -0,0 +1,256 @@ +#!/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..a01a1f1e 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,19 @@ 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 +163,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 +188,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 +208,29 @@ 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 +269,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 +304,13 @@ 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..af096ee2 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,29 @@ 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 +120,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 +173,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,16 +206,39 @@ 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: @@ -185,30 +250,71 @@ async def read(tail=False): 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 message_id=reference['id']: resolve_legacy(message_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 +327,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..2791ab50 --- /dev/null +++ b/tests/test_client_listener_cli.py @@ -0,0 +1,19 @@ +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..a7ebaeeb --- /dev/null +++ b/tests/test_client_listener_reliability.py @@ -0,0 +1,371 @@ +"""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..4894e134 --- /dev/null +++ b/tests/test_client_listener_remote.py @@ -0,0 +1,184 @@ +"""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, *, started_event=ready): + nonlocal reads + result = await store.inbox('worker', cursor=cursor, tail=tail) + reads += 1 + started_event.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' + )