From a5472bab8fcf21f343c7015968c4cb4b5de7826c Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Tue, 13 Oct 2026 07:33:27 +0800 Subject: [PATCH 1/4] feat: add authorized resumable event waits and deterministic task plans --- .github/workflows/event-coordination.yml | 74 ++++++++ docs/EVENT_COORDINATION.md | 140 ++++++++++++++ docs/SUBAGENTS.md | 6 + src/msg/application.py | 8 +- src/msg/client_agent_cli.py | 49 ++++- src/msg/client_listener.py | 22 ++- src/msg/client_subscription.py | 60 ++++++ src/msg/core/event_wait.py | 64 +++++++ src/msg/core/task_graph.py | 119 ++++++++++++ src/msg/plugins/collaboration_resources.py | 23 ++- src/msg/plugins/common.py | 17 +- src/msg/plugins/event_stream.py | 81 ++++++-- src/msg/plugins/task_plans.py | 203 +++++++++++++++++++++ src/msg/storage/event_signal.py | 94 ++++++++++ tests/test_client_subscription.py | 47 +++++ tests/test_event_subscription.py | 130 +++++++++++++ tests/test_event_wait.py | 123 +++++++++++++ tests/test_listener_revalidation.py | 76 ++++++++ tests/test_task_graph.py | 90 +++++++++ tests/test_task_plans.py | 108 +++++++++++ 20 files changed, 1509 insertions(+), 25 deletions(-) create mode 100644 .github/workflows/event-coordination.yml create mode 100644 docs/EVENT_COORDINATION.md create mode 100644 src/msg/client_subscription.py create mode 100644 src/msg/core/event_wait.py create mode 100644 src/msg/core/task_graph.py create mode 100644 src/msg/plugins/task_plans.py create mode 100644 src/msg/storage/event_signal.py create mode 100644 tests/test_client_subscription.py create mode 100644 tests/test_event_subscription.py create mode 100644 tests/test_event_wait.py create mode 100644 tests/test_listener_revalidation.py create mode 100644 tests/test_task_graph.py create mode 100644 tests/test_task_plans.py diff --git a/.github/workflows/event-coordination.yml b/.github/workflows/event-coordination.yml new file mode 100644 index 00000000..c4387451 --- /dev/null +++ b/.github/workflows/event-coordination.yml @@ -0,0 +1,74 @@ +name: Event and task coordination +on: + pull_request: + paths: + - 'src/msg/**' + - 'tests/**' + - '.github/workflows/event-coordination.yml' + push: + branches: [main] + paths: + - 'src/msg/**' + - 'tests/**' + - '.github/workflows/event-coordination.yml' + workflow_dispatch: +permissions: + contents: read +concurrency: + group: event-coordination-${{ github.event.pull_request.head.repo.full_name || github.repository }}-${{ github.head_ref || github.ref_name }} + cancel-in-progress: true +jobs: + verify: + runs-on: ubuntu-latest + timeout-minutes: 12 + 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" + --health-interval 10s --health-timeout 5s --health-retries 5 + env: + MSG_TEST_POSTGRES_URL_TEMPLATE: postgresql://msgtest:msgtest@127.0.0.1:5432/{database} + steps: + - uses: actions/checkout@v7 + - uses: actions/setup-python@v7 + with: + python-version: '3.15' + allow-prereleases: true + - name: Install focused verification dependencies + run: | + python -m pip install uv==0.12.20 + uv sync --locked --extra server + uv pip install pytest pytest-asyncio ruff==0.16.9 + echo "$GITHUB_WORKSPACE/.venv/bin" >> "$GITHUB_PATH" + - name: Check formatting and syntax + run: | + mkdir -p artifacts + # Save a repair diff even when the check fails; never rewrite or push source from CI. + ruff format --diff src tests > artifacts/format.diff || true + ruff check src tests --output-format concise > artifacts/lint.txt || true + python -W error -m compileall -q src + - name: Run focused real-database and client regressions + run: | + python -m pytest \ + tests/test_task_graph.py tests/test_task_plans.py \ + tests/test_event_wait.py tests/test_event_subscription.py \ + tests/test_client_subscription.py tests/test_listener_revalidation.py \ + tests/test_client_listener.py tests/test_collaboration_resources.py \ + tests/test_resource_event_unification.py \ + --junitxml=artifacts/event-coordination.xml + - name: Require clean formatting + if: always() + run: | + ruff check src tests + ruff format --check src tests + - uses: actions/upload-artifact@v7.0.1 + if: always() + with: + name: event-coordination + path: artifacts/ diff --git a/docs/EVENT_COORDINATION.md b/docs/EVENT_COORDINATION.md new file mode 100644 index 00000000..fe838686 --- /dev/null +++ b/docs/EVENT_COORDINATION.md @@ -0,0 +1,140 @@ +# Authorized event subscriptions and deterministic task plans + +## Subscribe without model polling + +```sh +msg listen --subscribe --resource RESOURCE_ID --from-now --once \ + --cursor-file ./task-events.json +msg listen --subscribe --resource RESOURCE_ID --cursor-file ./task-events.json +``` + +Use the same account, service, resource and event filters on restart. Omit +`--from-now` when resuming. Each independent reader needs its own checkpoint. +`--once` drains available pages; `--max-pages` bounds that drain. The existing +`--event` filter and `--max-events` limit continue to work. Omit `--resource` for +the account's authorized watched events. Subscription mode requires a server; +it never switches to an offline queue, mailbox or public channel. + +`communication.events@2` adds `wait_ms` (0–20000), `tail`, `has_more`, and an +`authority_changed` flag. Each event has protected before/after resume cursors. +One PostgreSQL LISTEN connection per server process receives empty commit hints. +It does not carry identities, private references or bodies. Each reader wakes +and repeats the normal signed read, in a new transaction. Requests do not hold a +transaction while waiting. Rollback does not publish a commit hint. Missing hints +are repaired by the next bounded read from the durable event log. + +A wait has a 20-second maximum, not a one-second HTTP polling loop. Idle requests +perform an initial authorized read, a race-closing read after LISTEN is ready, +and a final read at the heartbeat. Busy channels have a 32-wakeup reread budget; +after that, the final read occurs at the heartbeat. Each process admits at most +128 waiting requests. If waiting is unavailable or full, the response explicitly +sets `wait_supported=false`; the CLI uses its configured interval. Neither this +bound nor the implementation alone establishes an end-to-end speed ranking. + +The original signed request is reauthenticated after each wake. An expired or +revoked credential, or a scope that is no longer readable, stops delivery without +advancing the caller's durable checkpoint. After explicit authorization recovery, +resume the same checkpoint. Network failures use the existing capped retry delay. +The server may restart its notification connection without losing the log position. + +Version 2 accepts existing signed `events-v1` positions for the **same actor, +account and resource**. A cursor is a position, not permission. Version 2 rechecks +current access even when the older cursor's short expiry has passed. Version 1 +retains its original expiry and authority-change errors. Other cursor families +(`communication.changes`, sync checkpoints and mailbox cursors) are not +interchangeable. Use a separate subscription checkpoint instead of renaming an +old mailbox checkpoint or changing its binding. + +When authority changes, version 2 retains the sequence and reports +`authority_changed=true`. The CLI reports this on stderr. A caller that maintains +a resource cache must rebuild that cache under current permissions: newly granted +access can include history **before** its event position. Continuing a cursor is +not a complete state synchronization. If log retention has removed the position, +`resync_required` remains a hard error; no automatic reset or silent skip occurs. + +On restart, cached pending subscription events are reread from their protected +before-cursor before output. Revoked events are discarded, not emitted from disk. +Existing remote mailbox readers also reread their pending messages; existing +account-event readers use the authorized event projection rather than cached audit +bodies. A permanent authorization failure leaves the checkpoint unchanged. Reading +is not ACK. Output and checkpoint persistence remain at-least-once: deduplicate by +event/message ID. Local readers keep their existing offline behavior. + +## Task IDs, dependency release and concurrency + +A task plan is a private `collab_request` Resource, not an extra identity, new +public topic, model conversation or executable script. The existing transaction, +Resource/Revision, sharing, generation and receipt mechanisms apply. Each task has +an ID unique within the plan. Use `(plan_id, task_id)` as the complete task key. +A plan accepts at most 256 tasks and 64 simultaneous running tasks. + +Create with `communication.task_plan_create@1`: + +```json +{ + "title": "Review two components and combine results", + "concurrency": 2, + "tasks": [ + {"id": "a", "instruction": "Review component A"}, + {"id": "b", "instruction": "Review component B"}, + {"id": "combine", "instruction": "Combine the reviews", "dependencies": ["a", "b"]} + ] +} +``` + +The default worker account is the creator. An explicit `workers` list permits +additional account IDs. Those accounts also need current read access to the plan +and any required dependency result. Read sharing alone does not grant task +mutation. Worker labels within one account are trusted routing labels, not an +isolation boundary; use distinct authorized identities for untrusted workers. + +Call these operations through the existing signed client/API: + +| Operation | Action | +| --- | --- | +| `communication.task_claim` | Claim the first ready task, within the plan's concurrency limit. Return its task ID and unique attempt ID. | +| `communication.task_complete` | Associate a fixed `result: {id, revision}` or an explicit error with that exact claim. | +| `communication.task_renew` | Extend the active claim by 1–3600 seconds; default 300. | +| `communication.task_status` | Read the plan, current visible results, counts, ready tasks and completion aggregate. | + +Mutations provide the plan's expected generation in the ordinary request envelope. +The database serializes writes across processes. A competing generation loses +rather than claiming the same slot. Retry a lost response with the **same request +ID and unchanged packet**. After a confirmed generation conflict, read the current +plan and make a new decision/request. Do not change a packet under an existing ID. + +Completion requires `id` (the plan), `task_id`, `worker`, and `attempt`, plus exactly +one `result` or `error`. The server binds a claim to its account and label. A +successful result must pin an existing revision. Results are stored as immutable +relations; task JSON contains only an internal slot, not a copied private result +ID or body. Status reads reveal each result only after a current access check. +Unrelated hidden result links can remain in the plan without becoming readable to +another worker. A newly added result still requires normal read permission. + +The code, not a model, performs cycle detection, result correlation, completion +counts, failure propagation, readiness checks and concurrency control. Completing +both `a` and `b` makes `combine` ready in the same transaction. A failure blocks its +dependent tasks. Repeated identical results are safe; conflicting results or a +wrong/expired claim are rejected. The ordinary request claim/fulfill/cancel +operations cannot bypass a task plan's transitions. + +The normal committed event points to the plan's fixed revision. A runtime can +subscribe to that plan, read the changed state once, and call `task_claim` for +available workers. It does not need a model to repeatedly ask whether work is done. +MSG does not start a model, run a tool or stop a remote process. Those actions still +belong to the worker runtime. Leases are evaluated by status/transition calls; +there is no new automatic timer service in this change. Expired work becomes failed +and blocks dependents; it is **not** silently reexecuted. This avoids pretending +that a cursor or lease can ensure exactly-once external side effects. + +Old credentials with a fixed operation ceiling may need explicit authorization +for the new operation versions. The server does not widen their permissions or +change public/private visibility during upgrade. + +## Verification + +`Event and task coordination` runs focused client tests plus real PostgreSQL, +Ed25519 and Resource/Revision tests on Python 3.15. It checks resume boundaries, +permission loss, commit versus rollback wakeups, current result visibility, +racing claims, dependency release and pinned results. CI evidence is not a +production deployment or a benchmark against other agent runtimes. diff --git a/docs/SUBAGENTS.md b/docs/SUBAGENTS.md index cb841b81..cd609093 100644 --- a/docs/SUBAGENTS.md +++ b/docs/SUBAGENTS.md @@ -188,3 +188,9 @@ 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. + +## Authorized subscriptions and task plans + +See [event coordination](EVENT_COORDINATION.md) for `listen --subscribe`, +permission-aware cursor recovery, and deterministic task IDs, dependency release, +result aggregation and concurrency limits. Legacy mailbox modes remain separate. diff --git a/src/msg/application.py b/src/msg/application.py index 581539fd..78c8656f 100644 --- a/src/msg/application.py +++ b/src/msg/application.py @@ -9,7 +9,7 @@ from msg.core.codec import decode, loads, unb64 from msg.core.cursors import CursorCodec from msg.core.errors import require -from msg.core.executor import OperationExecutor +from msg.core.event_wait import EventWaitingExecutor from msg.core.models import Certificate, Scope from msg.core.registry import Registry from msg.plugins import install_registry @@ -68,7 +68,7 @@ def temporary_ceiling(self): async def open_storage(self): if self.metadata is None: - from msg.storage.postgres import PostgresMetadataStore + from msg.storage.event_signal import EventMetadataStore from msg.storage.valkey_bus import ValkeyOutboxSignal signal = ( @@ -76,7 +76,7 @@ async def open_storage(self): if self.settings.server.valkey_url else None ) - self.metadata = PostgresMetadataStore(self.settings.server.postgres_dsn, signal=signal) + self.metadata = EventMetadataStore(self.settings.server.postgres_dsn, signal=signal) if self.contents is None: self.contents = GitContentStore( self.settings.server.content_dir, @@ -172,7 +172,7 @@ async def load(self): def new_executor(self, authenticator=None): """Compose identical limits/ports for HTTP and the restricted SSH entry.""" - executor = OperationExecutor( + executor = EventWaitingExecutor( self.registry, self.metadata, self.contents, diff --git a/src/msg/client_agent_cli.py b/src/msg/client_agent_cli.py index 5cdb3550..c3b2d2fc 100644 --- a/src/msg/client_agent_cli.py +++ b/src/msg/client_agent_cli.py @@ -65,6 +65,11 @@ def add_commands(commands): default=[], help='Exact event type; repeat to include more types.', ) + listener.add_argument( + '--subscribe', action='store_true', + help='Use authorized server event waits (events@2); no mailbox or offline fallback.', + ) + listener.add_argument('--resource', help='With --subscribe, limit events to this resource.') 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) @@ -83,6 +88,8 @@ def add_commands(commands): def local_command(args): + if getattr(args, 'subscribe', False): + return False return (args.command == 'agent' or args.command == 'listen' and args.agent) and not args.remote @@ -148,7 +155,7 @@ 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, *, revalidate=None, server_wait=False): from msg.client_listener import listen await listen( @@ -167,6 +174,8 @@ async def stream(state, args, fetch, source): max_events=args.max_events, max_pages=getattr(args, 'max_pages', None), from_now=args.from_now, + revalidate=revalidate, + server_wait=server_wait, ) return 0 @@ -223,6 +232,18 @@ async def run_remote(client, args): require(not args.offline, 'offline_remote_conflict') if args.user: await client.require_username(args.user) + if args.command == 'listen' and getattr(args, 'subscribe', False): + from msg.client_subscription import Subscription + + require(not args.agent, 'subscription_requires_account_reader') + source = Subscription( + client, resource=args.resource, once=args.once, interval=args.interval, + ) + return await stream( + client.state, args, source.fetch, 'subscription:' + (args.resource or ''), + revalidate=source.revalidate, server_wait=True, + ) + require(not getattr(args, 'resource', None), 'subscription_required') if args.command == 'listen' and not args.agent: async def fetch(cursor, tail=False): @@ -244,7 +265,21 @@ async def fetch(cursor, tail=False): 'has_more': more, } - return await stream(client.state, args, fetch, 'account-events') + async def revalidate(item): + current = await client.call('communication.event', {'id': item['id']}) + if current.status != 'ok' and current.error.code == 'not_found': + return None + event = wire(client.checked(current).data) + # Old audit bodies must not survive a current-authorization refresh. + return { + **event, 'type': event['operation'], + 'data': {'operation': event['operation']}, + 'resources': [address['ref'] for address in event['resources']], + } + + return await stream( + client.state, args, fetch, 'account-events', revalidate=revalidate, + ) from msg.client_subagents_remote import RemoteAgents store = RemoteAgents(client, username=args.username) @@ -254,7 +289,15 @@ 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') + async def revalidate(item): + try: + return await store.read(name, item['id']) + except Failure as exc: + if exc.code != 'not_found': + raise + return None + + return await stream(client.state, args, fetch, 'remote', revalidate=revalidate) if args.action == 'create': result = await store.create(args.name) elif args.action == 'show': diff --git a/src/msg/client_listener.py b/src/msg/client_listener.py index 3bc9ab5b..46e74a49 100644 --- a/src/msg/client_listener.py +++ b/src/msg/client_listener.py @@ -140,6 +140,8 @@ async def listen( max_pages: int | None = None, from_now: bool = False, output=None, + revalidate=None, + server_wait=False, ) -> int: """Poll until cancelled, emitting and flushing one JSON object per matching event. @@ -162,6 +164,7 @@ async def listen( binding = _binding(context, event_types) types = binding['event_types'] state = _load(cursor_file, binding, cursor) + restored = bool(state['pending']) if from_now and (cursor is not None or (cursor_file is not None and cursor_file.exists())): raise Failure('listener_start_position_conflict') stream = sys.stdout if output is None else output @@ -169,11 +172,11 @@ async def listen( pages = 0 retry_delay = interval - async def read(tail=False): + async def retry(action): nonlocal retry_delay while True: try: - value = _page(await fetch(state['cursor'], tail=tail)) + value = await action() retry_delay = interval return value except Failure as exc: @@ -185,6 +188,9 @@ 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))) + if from_now: page = await read(tail=True) state.update(cursor=page['cursor'], pending=[], has_more=False) @@ -199,6 +205,14 @@ async def read(tail=False): state.update(cursor=page['cursor'], pending=page['items'], has_more=page['has_more']) while state['pending']: item = state['pending'][0] + if restored and revalidate is not None: + item = await retry(lambda: revalidate(state['pending'][0])) + if item is None: + state['pending'] = state['pending'][1:] + _save(cursor_file, state) + continue + if not isinstance(item, dict): + raise Failure('invalid_listener_page') 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') @@ -208,6 +222,7 @@ async def read(tail=False): _save(cursor_file, state) if max_events is not None and emitted >= max_events: return emitted + restored = False _save(cursor_file, state) if max_pages is not None and pages >= max_pages: if state['has_more']: @@ -221,4 +236,5 @@ async def read(tail=False): continue if once: return emitted - await asyncio.sleep(interval) + if not server_wait: + await asyncio.sleep(interval) diff --git a/src/msg/client_subscription.py b/src/msg/client_subscription.py new file mode 100644 index 00000000..7bd8923c --- /dev/null +++ b/src/msg/client_subscription.py @@ -0,0 +1,60 @@ +"""Authorized long-poll subscriptions using the normal signed client and cursors.""" + +import asyncio +import sys + +from msg.core.codec import wire +from msg.core.errors import require + + +class Subscription: + def __init__(self, client, *, resource=None, once=False, interval=1): + self.client = client + self.resource = resource + self.once = once + self.interval = interval + + async def _read(self, cursor, *, tail=False, wait=False, limit=50): + args = {'limit': limit, 'wait_ms': 20000 if wait else 0} + if self.resource is not None: + args['resource'] = self.resource + if cursor is not None: + args['cursor'] = cursor + if tail: + args['tail'] = True + result = self.client.checked(await self.client.call( + 'communication.events', args, contract_version=2, + )) + return wire(result.data) + + async def fetch(self, cursor, tail=False): + page = await self._read(cursor, tail=tail, wait=not (self.once or tail)) + if page.get('authority_changed'): + print( + 'msg listen: authority changed; position retained. Refresh derived resource state.', + file=sys.stderr, flush=True, + ) + if ( + not self.once and not tail and not page['items'] and not page['has_more'] + and not page.get('wait_supported') + ): + # Explicitly advertised fallback, not a silent switch of source or account. + await asyncio.sleep(self.interval) + return page + + async def revalidate(self, item): + # Never output a durable pending body under yesterday's authority. The + # protected before-cursor lets the server redo the same selection today. + cursor = item.get('before_cursor') + require(isinstance(cursor, str), 'invalid_listener_checkpoint') + while True: + page = await self._read(cursor, limit=1) + for current in page['items']: + if current['id'] == item['id']: + return current + if current['seq'] > item['seq']: + return None + if not page['has_more']: + return None + require(page['cursor'] != cursor, 'invalid_listener_page') + cursor = page['cursor'] diff --git a/src/msg/core/event_wait.py b/src/msg/core/event_wait.py new file mode 100644 index 00000000..0ab7d137 --- /dev/null +++ b/src/msg/core/event_wait.py @@ -0,0 +1,64 @@ +"""Bounded event waits wrap, never replace, the canonical authorization pipeline.""" + +import asyncio +import time +from dataclasses import replace + +from msg.core.executor import OperationExecutor + + +class EventWaitingExecutor(OperationExecutor): + async def execute(self, request, *, entry='network'): + result = await super().execute(request, entry=entry) + if ( + request.operation != 'communication.events' + or request.contract_version != 2 + or result.status != 'ok' + ): + return result + timeout = request.arguments.get('wait_ms', 0) / 1000 + signal = getattr(self.metadata, 'event_signal', None) + current = getattr(self.metadata, '_current', None) + if not timeout or request.arguments.get('tail') or signal is None: + return result + # A nested batch must not keep an outer transaction or writer lock open. + if (current is not None and current.get() is not None) or signal.waiters >= 128: + return replace(result, data={**result.data, 'wait_supported': False}) + + def available(value): + return ( + value.status != 'ok' + or value.data.get('items') + or value.data.get('has_more') + or value.data.get('authority_changed') + ) + + if available(result): + return result + deadline = time.monotonic() + timeout + signal.waiters += 1 + try: + if not await signal.start(min(2, max(0, deadline - time.monotonic()))): + return replace(result, data={**result.data, 'wait_supported': False}) + generation = signal.generation + # Register before rereading: commits in the read/LISTEN gap are + # included in this second read, not lost until a later heartbeat. + result = await super().execute(request, entry=entry) + checks = 0 + while not available(result) and time.monotonic() < deadline: + remaining = deadline - time.monotonic() + if checks >= 32: + # Bound read amplification on a globally busy event channel. + # Do one final authorized read at the heartbeat deadline. + await asyncio.sleep(max(0, remaining)) + else: + await signal.wait(generation, max(0, remaining)) + await asyncio.sleep(min(0.05, max(0, deadline - time.monotonic()))) + generation = signal.generation + # The original signed packet is unchanged. Authentication, + # revocation, credential expiry and per-resource ACL run again. + result = await super().execute(request, entry=entry) + checks += 1 + return result + finally: + signal.waiters -= 1 diff --git a/src/msg/core/task_graph.py b/src/msg/core/task_graph.py new file mode 100644 index 00000000..dbb81160 --- /dev/null +++ b/src/msg/core/task_graph.py @@ -0,0 +1,119 @@ +"""Bounded task state transitions. This module never starts an agent or runs a tool.""" + +from copy import deepcopy +from graphlib import CycleError, TopologicalSorter + +from msg.core.errors import require + +TERMINAL = frozenset({'succeeded', 'failed', 'blocked'}) +MAX_TASKS = 256 + + +def create(tasks, concurrency): + require(type(concurrency) is int and 1 <= concurrency <= 64, 'invalid_task_concurrency') + require(isinstance(tasks, list) and 1 <= len(tasks) <= MAX_TASKS, 'invalid_task_graph') + records = {} + for task in tasks: + name = task['id'] + require(name not in records, 'duplicate_task_id') + dependencies = task.get('dependencies', []) + require(len(set(dependencies)) == len(dependencies), 'duplicate_task_dependency') + records[name] = { + 'id': name, + 'instruction': task['instruction'], + 'dependencies': list(dependencies), + 'status': 'pending', + } + require( + all(dep in records for task in records.values() for dep in task['dependencies']), + 'unknown_task_dependency', + ) + try: + order = list(TopologicalSorter({k: t['dependencies'] for k, t in records.items()}).static_order()) + except CycleError: + require(False, 'task_dependency_cycle') + return {'concurrency': concurrency, 'tasks': [records[name] for name in order]} + + +def settled(graph, now): + """Expire leases and propagate failure without making unfinished work runnable again.""" + graph = deepcopy(graph) + tasks = {task['id']: task for task in graph['tasks']} + for task in tasks.values(): + if task['status'] == 'running' and task['lease_until'] <= now: + task.update(status='failed', error='lease_expired', finished_at=now) + if task['status'] == 'pending' and any( + tasks[dep]['status'] in {'failed', 'blocked'} for dep in task['dependencies'] + ): + task.update(status='blocked', error='dependency_failed', finished_at=now) + return graph + + +def summary(graph): + tasks = {task['id']: task for task in graph['tasks']} + counts = {status: 0 for status in ('pending', 'running', 'succeeded', 'failed', 'blocked')} + ready = [] + for task in tasks.values(): + counts[task['status']] += 1 + if task['status'] == 'pending' and all( + tasks[dep]['status'] == 'succeeded' for dep in task['dependencies'] + ): + ready.append(task['id']) + completed = sum(counts[status] for status in TERMINAL) + return { + **counts, + 'total': len(tasks), + 'completed': completed, + 'done': completed == len(tasks), + 'ready': ready, + 'available_slots': max(0, graph['concurrency'] - counts['running']), + } + + +def claim(graph, *, worker, attempt, now, lease_seconds, permitted=None): + graph = settled(graph, now) + state = summary(graph) + if not state['available_slots']: + return graph, None + for task in graph['tasks']: + if task['id'] in state['ready'] and (permitted is None or task['id'] in permitted): + task.update( + status='running', worker=worker, attempt=attempt, + lease_until=now + lease_seconds, started_at=now, + ) + return graph, deepcopy(task) + return graph, None + + +def finish(graph, *, task_id, worker, attempt, now, result_slot=None, error=None): + graph = settled(graph, now) + task = next((task for task in graph['tasks'] if task['id'] == task_id), None) + require(task is not None, 'task_not_found') + require(task.get('worker') == worker and task.get('attempt') == attempt, 'task_claim_mismatch') + require((result_slot is not None) != (error is not None), 'task_result_required') + target = 'succeeded' if result_slot is not None else 'failed' + if task['status'] in TERMINAL: + require( + task['status'] == target + and task.get('result_slot') == result_slot + and task.get('error') == error, + 'task_result_conflict', + ) + return graph + require(task['status'] == 'running', 'task_not_running') + task.update(status=target, finished_at=now) + if result_slot is not None: + task['result_slot'] = result_slot + else: + task['error'] = error + return settled(graph, now) + + +def renew(graph, *, task_id, worker, attempt, now, lease_seconds): + graph = settled(graph, now) + task = next((task for task in graph['tasks'] if task['id'] == task_id), None) + require(task is not None, 'task_not_found') + require(task.get('worker') == worker and task.get('attempt') == attempt, 'task_claim_mismatch') + require(task['status'] == 'running', 'task_not_running') + task['lease_until'] = now + lease_seconds + return graph diff --git a/src/msg/plugins/collaboration_resources.py b/src/msg/plugins/collaboration_resources.py index b332105d..39789fc8 100644 --- a/src/msg/plugins/collaboration_resources.py +++ b/src/msg/plugins/collaboration_resources.py @@ -27,9 +27,14 @@ def resource_types(app): + from msg.plugins.task_plans import GRAPH, WORKERS + types = [] fields = { 'request': { + 'scheduler_version': {'const': 1}, + 'workers': WORKERS, + 'graph': GRAPH, 'title': TEXT, 'description': TEXT, 'requirements': TEXT, @@ -123,7 +128,8 @@ async def _project(app, ctx, request, tx, resource, revision, record): 'generation': resource.generation, 'resource_refs': [], } - for relation in revision.relations: + result_refs = {} + for slot, relation in enumerate(revision.relations): try: await _safe_ref(app, ctx, request, tx, relation.target.id) except Failure as exc: @@ -138,12 +144,23 @@ async def _project(app, ctx, request, tx, resource, revision, record): }: continue raise + result_refs[slot] = wire(relation.target) if relation.type in {'target', 'content'}: result[relation.type + '_ref'] = wire(relation.target) elif relation.type == 'state': result['state_ref'] = relation.target.id else: result['resource_refs'].append(relation.target.id) + if record.get('scheduler_version') == 1: + from msg.core import task_graph + + graph = task_graph.settled(record['graph'], ctx.now.timestamp()) + result['summary'] = task_graph.summary(graph) + for task in graph['tasks']: + slot = task.pop('result_slot', None) + if slot in result_refs: + task['result'] = result_refs[slot] + result['graph'] = graph if ( record.get('expires_at') and parse_time(record['expires_at']) <= ctx.now @@ -335,6 +352,9 @@ async def list_records(ctx, request, tx): from msg.plugins.proposals import install as install_proposals install_proposals(app, op) + from msg.plugins.task_plans import install as install_task_plans + + install_task_plans(app, op) def install_transition(kind, action): @op( @@ -343,6 +363,7 @@ def install_transition(kind, action): async def transition(ctx, request, tx): subject = await _self(app, ctx, request, tx) resource, revision, record = await _load(app, ctx, request, tx, kind) + require('scheduler_version' not in record, 'task_plan_required') await assert_generation(request, resource) require( not record.get('expires_at') or parse_time(record['expires_at']) > ctx.now, diff --git a/src/msg/plugins/common.py b/src/msg/plugins/common.py index c065df75..56476989 100644 --- a/src/msg/plugins/common.py +++ b/src/msg/plugins/common.py @@ -270,6 +270,7 @@ async def revise_resource( source_digest=None, content_created_at=None, summary=SUMMARY_UNSET, + preserve_existing_relations=False, ): from msg.plugins.board_presentation import protect_edit @@ -327,10 +328,24 @@ async def revise_resource( blob.media_type in {'text/plain', 'text/markdown'} and blob.size <= 4096, 'invalid_board_description', ) + preserved = () + if preserve_existing_relations: + require( + resource.type == 'collab_request' + and request.operation in { + 'communication.task_claim', 'communication.task_complete', 'communication.task_renew', + }, + 'controlled_resource', + ) + # Server-managed task results remain pinned even when an independent + # worker cannot read another task's result. Only an exact current link + # can survive; every new link still needs ordinary read permission. + preserved = (await tx.revision(ResourceRef(id=resource.id))).relations for relation in relations: spec = app.registry.resource_type(resource.type, resource.type_version) require(relation.type in spec.relations, 'invalid_relation_type') - await check_access(app, ctx, request, tx, relation.target.id, 'read') + if relation not in preserved: + await check_access(app, ctx, request, tx, relation.target.id, 'read') if relation.target.revision is not None: await tx.revision(relation.target) rid = revision_id or new_id('v') diff --git a/src/msg/plugins/event_stream.py b/src/msg/plugins/event_stream.py index aa35f5ef..6deeb5c1 100644 --- a/src/msg/plugins/event_stream.py +++ b/src/msg/plugins/event_stream.py @@ -1,5 +1,6 @@ """One durable, authorized event stream over the committed event log.""" +import asyncio import time from datetime import timedelta @@ -85,6 +86,7 @@ async def events(ctx, request, tx): (subject,), ) topics = {row[0] for row in memberships} + subscription = request.contract_version == 2 binding = { 'actor': ctx.principal.actor, 'subject': subject, @@ -92,35 +94,71 @@ async def events(ctx, request, tx): 'certificates': sorted(ctx.principal.certificates), 'resource': scope, } + previous_binding = dict(binding) + if subscription: + binding.pop('credential_id') + binding.pop('certificates') authority = digest({ 'epoch': tx.setting('authorization_epoch', 0), 'watches': sorted(watches), 'topics': [tuple(row) for row in memberships], + **({'credential_id': ctx.principal.credential_id, + 'certificates': sorted(ctx.principal.certificates)} if subscription else {}), }) floor = tx.setting('sync_floor', 0) position = floor expires = ctx.now + timedelta(minutes=15) cursor = request.arguments.get('cursor') + authority_changed = False if cursor: - saved = app.cursors.decode(cursor, 'events-v1', binding) - require(isinstance(saved, dict), 'invalid_cursor') - require(ctx.now < parse_time(saved['expires_at']), 'cursor_expired') - require(saved.get('authority') == authority, 'resync_required') + if subscription: + inspected = app.cursors.inspect(cursor) + require(inspected.get('kind') in {'events-v1', 'events-v2'}, 'cursor_kind_mismatch') + query = inspected.get('query') + require(isinstance(query, dict), 'invalid_cursor') + stable = {key: query.get(key) for key in binding} + require(stable == binding, 'cursor_query_mismatch') + saved = inspected.get('position') + require(isinstance(saved, dict), 'invalid_cursor') + # A signed position is not authority. Recheck today's credential, + # scope and each reference, including after a credential rotation. + authority_changed = saved.get('authority') != authority + else: + saved = app.cursors.decode(cursor, 'events-v1', previous_binding) + require(isinstance(saved, dict), 'invalid_cursor') + require(ctx.now < parse_time(saved['expires_at']), 'cursor_expired') + require(saved.get('authority') == authority, 'resync_required') + expires = parse_time(saved['expires_at']) position = saved.get('seq') require(type(position) is int and position >= floor, 'resync_required') - expires = parse_time(saved['expires_at']) + + tail = max(position, tx.one('SELECT COALESCE(MAX(seq),0) FROM events')[0]) + if subscription and request.arguments.get('tail'): + require(cursor is None, 'listener_start_position_conflict') + position = tail + + def resume(sequence): + state = {'seq': sequence, 'authority': authority} + if not subscription: + state['expires_at'] = wire(expires) + return app.cursors.encode('events-v2' if subscription else 'events-v1', binding, state) items = [] limit = request.arguments.get('limit', 50) + budget = 64 if subscription else 500 rows = tx.rows( - 'SELECT seq,body FROM events WHERE seq>? ORDER BY seq LIMIT 500', (position,) + 'SELECT seq,body FROM events WHERE seq>? AND seq<=? ORDER BY seq LIMIT ?', + (position, tail, budget), ) for seq, raw in rows: require(time.monotonic() < ctx.deadline_monotonic, 'query_cost_exceeded') + before = position position = seq event = decode(Event, loads(raw)) permitted = [] for ref in event.resources: + if subscription: + await asyncio.sleep(0) if not await readable(app, ctx, request, tx, ref.id): continue ancestors = {r.id for r in await tx.ancestors(ref.id)} @@ -160,20 +198,37 @@ async def events(ctx, request, tx): if await readable(app, ctx, request, tx, event.subject) else None, ) - items.append({'seq': seq, **envelope}) + item = {'seq': seq, **envelope} + if subscription: + item.update(before_cursor=resume(before), resume_cursor=resume(seq)) + items.append(item) if len(items) >= limit: break - token = app.cursors.encode( - 'events-v1', - binding, - {'seq': position, 'authority': authority, 'expires_at': wire(expires)}, - ) + token = resume(position) args = {**request.arguments, 'cursor': token} return HandlerOutput( data={ 'items': items, 'cursor': token, - 'next': next_link(app, request.operation, args), + **({'next': next_link(app, request.operation, args)} if not subscription else {}), 'next_requires_auth': True, + **({ + 'has_more': position < tail, + 'authority_changed': authority_changed, + 'wait_supported': getattr(app.metadata, 'event_signal', None) is not None, + } if subscription else {}), } ) + + op( + 'communication.events', + obj({ + 'resource': {'type': 'string', 'minLength': 1, 'maxLength': 2048}, + 'cursor': STRING, + 'limit': {'type': 'integer', 'minimum': 1, 'maximum': 200}, + 'tail': {'type': 'boolean'}, + 'wait_ms': {'type': 'integer', 'minimum': 0, 'maximum': 20000}, + }), + effect='read', + version=2, + )(events) diff --git a/src/msg/plugins/task_plans.py b/src/msg/plugins/task_plans.py new file mode 100644 index 00000000..0cb8dc7b --- /dev/null +++ b/src/msg/plugins/task_plans.py @@ -0,0 +1,203 @@ +"""Deterministic scheduling inside private collaboration request resources.""" + +from msg.core import task_graph +from msg.core.codec import canonical, decode, wire +from msg.core.errors import Failure, require +from msg.core.models import HandlerOutput, Relation, ResourceRef +from msg.plugins.collaboration import _safe_ref, _self +from msg.plugins.common import assert_generation, create_resource, new_id, revise_resource +from msg.plugins.schemas import IDENTIFIER, REF, obj + +NAME = {'type': 'string', 'pattern': '^[A-Za-z0-9_.:-]{1,80}$'} +TEXT = {'type': 'string', 'minLength': 1, 'maxLength': 4096} +DEPENDENCIES = {'type': 'array', 'items': NAME, 'uniqueItems': True, 'maxItems': 256} +TASK_FIELDS = {'id': NAME, 'instruction': TEXT, 'dependencies': DEPENDENCIES} +TASK_INPUT = obj(TASK_FIELDS, ('id', 'instruction')) +TASK_RECORD = obj( + { + **TASK_FIELDS, + 'status': {'enum': ['pending', 'running', 'succeeded', 'failed', 'blocked']}, + 'worker': {'type': 'string', 'maxLength': 256}, + 'attempt': NAME, + 'lease_until': {'type': 'number'}, + 'started_at': {'type': 'number'}, + 'finished_at': {'type': 'number'}, + 'result_slot': {'type': 'integer', 'minimum': 0, 'maximum': 255}, + 'error': TEXT, + }, + ('id', 'instruction', 'dependencies', 'status'), +) +CONCURRENCY = {'type': 'integer', 'minimum': 1, 'maximum': 64} +GRAPH = obj( + { + 'concurrency': CONCURRENCY, + 'tasks': {'type': 'array', 'items': TASK_RECORD, 'minItems': 1, 'maxItems': 256}, + }, + ('concurrency', 'tasks'), +) +WORKERS = {'type': 'array', 'items': IDENTIFIER, 'minItems': 1, 'maxItems': 64, 'uniqueItems': True} +LEASE = {'type': 'integer', 'minimum': 1, 'maximum': 3600} +# Loss of resource access is not a transient failure or a reason to publish a result. +HIDDEN = { + 'permission_denied', 'credential_ceiling', 'certificate_gate', 'delegation_scope', + 'ancestor_inactive', 'not_found', 'collaboration_ref_forbidden', +} + + +def install(app, op): + from msg.plugins.collaboration_resources import KINDS, _folder, _load, _project, _result + + @op( + 'communication.task_plan_create', + obj( + { + 'title': TEXT, + 'tasks': {'type': 'array', 'items': TASK_INPUT, 'minItems': 1, 'maxItems': 256}, + 'concurrency': CONCURRENCY, + 'workers': WORKERS, + }, + ('title', 'tasks', 'concurrency'), + ), + signature=True, + ) + async def create(ctx, request, tx): + subject = await _self(app, ctx, request, tx) + args = request.arguments + workers = args.get('workers', [subject]) + for worker in workers: + await tx.subject(worker) + record = { + 'title': args['title'], + 'description': 'Explicit task plan. The runtime starts and stops workers.', + 'requirements': 'Only successful dependencies release a task.', + 'requester': subject, + 'assignee': None, + 'subject': subject, + 'created_at': wire(ctx.now), + 'status': 'open', + 'scheduler_version': 1, + 'workers': workers, + 'graph': task_graph.create(args['tasks'], args['concurrency']), + } + parent = await _folder(app, ctx, request, tx, subject, 'request') + resource = await create_resource( + app, ctx, request, tx, parent=parent, type=KINDS['request'], name=new_id('plan'), + body=canonical(record), media_type='application/json', mode=0o600, + ) + return _result(resource, record) + + async def load(ctx, request, tx, *, mutate=False): + subject = await _self(app, ctx, request, tx) + resource, revision, record = await _load(app, ctx, request, tx, 'request') + require(record.get('scheduler_version') == 1, 'task_plan_required') + if mutate: + require(subject in record['workers'], 'task_worker_forbidden') + await assert_generation(request, resource) + return subject, resource, revision, record + + @op('communication.task_status', obj({'id': IDENTIFIER}, ('id',)), effect='read') + async def status(ctx, request, tx): + _, resource, revision, record = await load(ctx, request, tx) + return HandlerOutput(data={ + 'plan': await _project(app, ctx, request, tx, resource, revision, record), + }) + + async def save(ctx, request, tx, resource, revision, record, graph, *, task=None): + state = task_graph.summary(graph) + record = {**record, 'graph': graph, 'updated_at': wire(ctx.now)} + record['status'] = ( + 'fulfilled' if state['succeeded'] == state['total'] + else 'cancelled' if state['done'] else 'claimed' if state['running'] else 'open' + ) + resource = await revise_resource( + app, ctx, request, tx, resource, canonical(record), 'application/json', + relations=revision.relations, preserve_existing_relations=True, + ) + output = _result(resource, record) + # No cached instructions, dependency results, or private reference bodies. + return HandlerOutput(resources=output.resources, data={ + **output.data, + 'task_id': task['id'] if task else None, + 'attempt': task['attempt'] if task else None, + 'summary': state, + }) + + @op( + 'communication.task_claim', + obj({'id': IDENTIFIER, 'worker': NAME, 'lease_seconds': LEASE}, ('id', 'worker')), + signature=True, + ) + async def claim(ctx, request, tx): + subject, resource, revision, record = await load(ctx, request, tx, mutate=True) + graph = task_graph.settled(record['graph'], ctx.now.timestamp()) + tasks = {task['id']: task for task in graph['tasks']} + permitted = set() + for name in task_graph.summary(graph)['ready']: + try: + for dependency in tasks[name]['dependencies']: + ref = revision.relations[tasks[dependency]['result_slot']].target + await _safe_ref(app, ctx, request, tx, ref.id) + except Failure as exc: + if exc.code not in HIDDEN: + raise + else: + permitted.add(name) + graph, task = task_graph.claim( + graph, worker=subject + '#' + request.arguments['worker'], attempt=new_id('attempt'), + now=ctx.now.timestamp(), lease_seconds=request.arguments.get('lease_seconds', 300), + permitted=permitted, + ) + return await save(ctx, request, tx, resource, revision, record, graph, task=task) + + common = {'id': IDENTIFIER, 'task_id': NAME, 'worker': NAME, 'attempt': NAME} + + @op( + 'communication.task_complete', + obj({**common, 'result': REF, 'error': TEXT}, tuple(common)), + signature=True, + ) + async def complete(ctx, request, tx): + from dataclasses import replace + + subject, resource, revision, record = await load(ctx, request, tx, mutate=True) + args = request.arguments + require(('result' in args) != ('error' in args), 'task_result_required') + relations = list(revision.relations) + slot = None + if 'result' in args: + ref = decode(ResourceRef, args['result']) + require(ref.revision is not None, 'task_result_revision_required') + rid = await _safe_ref(app, ctx, request, tx, ref.id) + ref = ResourceRef(id=rid, revision=ref.revision) + await tx.revision(ref) + relation = Relation(type='reference', target=ref) + if relation not in relations: + relations.append(relation) + slot = relations.index(relation) + graph = task_graph.finish( + record['graph'], task_id=args['task_id'], worker=subject + '#' + args['worker'], + attempt=args['attempt'], now=ctx.now.timestamp(), result_slot=slot, + error=args.get('error'), + ) + return await save( + ctx, request, tx, resource, replace(revision, relations=tuple(relations)), record, graph, + task=next(task for task in graph['tasks'] if task['id'] == args['task_id']), + ) + + @op( + 'communication.task_renew', + obj({**common, 'lease_seconds': LEASE}, tuple(common)), + signature=True, + ) + async def renew(ctx, request, tx): + subject, resource, revision, record = await load(ctx, request, tx, mutate=True) + args = request.arguments + graph = task_graph.renew( + record['graph'], task_id=args['task_id'], worker=subject + '#' + args['worker'], + attempt=args['attempt'], now=ctx.now.timestamp(), + lease_seconds=args.get('lease_seconds', 300), + ) + return await save( + ctx, request, tx, resource, revision, record, graph, + task=next(task for task in graph['tasks'] if task['id'] == args['task_id']), + ) diff --git a/src/msg/storage/event_signal.py b/src/msg/storage/event_signal.py new file mode 100644 index 00000000..9c61e9a7 --- /dev/null +++ b/src/msg/storage/event_signal.py @@ -0,0 +1,94 @@ +"""One LISTEN connection per process. Notifications carry no event or identity data.""" + +import asyncio +import contextvars +from contextlib import aclosing, asynccontextmanager, suppress + +import psycopg + +from msg.storage.postgres import PostgresMetadataStore + + +class EventSignal: + def __init__(self, dsn): + self.dsn = dsn + self.generation = 0 + self.waiters = 0 + self.ready = asyncio.Event() + self.changed = asyncio.Event() + self.task = None + self.closed = False + + def _bump(self): + self.generation += 1 + old, self.changed = self.changed, asyncio.Event() + old.set() + + async def start(self, timeout=2): + if self.closed: + return False + if self.task is None: + self.task = asyncio.create_task(self._receive(), context=contextvars.Context()) + try: + await asyncio.wait_for(self.ready.wait(), timeout) + except TimeoutError: + return False + return not self.closed + + async def _receive(self): + delay = 1 + while True: + try: + async with await psycopg.AsyncConnection.connect( + self.dsn, autocommit=True, connect_timeout=3 + ) as connection: + await connection.execute('LISTEN msg_committed_events') + self._bump() + self.ready.set() + delay = 1 + async with aclosing(connection.notifies()) as notices: + async for _ in notices: + self._bump() + except (psycopg.Error, OSError): + # The durable log remains authoritative. Reconnection wakes + # every reader, which repeats its independently authorized read. + pass + finally: + self.ready.clear() + self._bump() + await asyncio.sleep(delay) + delay = min(delay * 2, 30) + + async def wait(self, generation, timeout): + if self.closed or generation != self.generation: + return + with suppress(TimeoutError): + await asyncio.wait_for(self.changed.wait(), timeout) + + async def close(self): + self.closed = True + self._bump() + if self.task is not None: + self.task.cancel() + with suppress(asyncio.CancelledError): + await self.task + self.task = None + + +class EventMetadataStore(PostgresMetadataStore): + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.event_signal = EventSignal(self.dsn) + + @asynccontextmanager + async def transaction(self, *, write): + async with super().transaction(write=write) as tx: + yield tx + if write: + # PostgreSQL delivers this only after commit. A rollback, + # including a nested savepoint rollback, cannot wake readers. + tx.execute("SELECT pg_notify('msg_committed_events', '')", write=True) + + async def close(self): + await self.event_signal.close() + await super().close() diff --git a/tests/test_client_subscription.py b/tests/test_client_subscription.py new file mode 100644 index 00000000..bea5f0be --- /dev/null +++ b/tests/test_client_subscription.py @@ -0,0 +1,47 @@ +from types import SimpleNamespace + +from msg.client_subscription import Subscription + + +class Client: + def __init__(self, pages): + self.pages = iter(pages) + self.calls = [] + + async def call(self, operation, args, **kwargs): + self.calls.append((operation, args, kwargs)) + return SimpleNamespace(data=next(self.pages)) + + @staticmethod + def checked(result): + return result + + +async def test_subscription_uses_explicit_version_and_persists_server_cursor(): + client = Client([{'items': [], 'cursor': 'next', 'has_more': False, 'wait_supported': True}]) + source = Subscription(client, resource='r_private') + page = await source.fetch('saved') + assert page['cursor'] == 'next' + operation, args, options = client.calls[0] + assert operation == 'communication.events' and options['contract_version'] == 2 + assert args == {'resource': 'r_private', 'cursor': 'saved', 'wait_ms': 20000, 'limit': 50} + + +async def test_pending_revalidation_returns_new_projection_not_cached_body(): + current = {'id': 'e_done', 'seq': 10, 'resources': []} + client = Client([ + {'items': [], 'cursor': 'scan', 'has_more': True}, + {'items': [current], 'cursor': 'end', 'has_more': False}, + ]) + source = Subscription(client) + refreshed = await source.revalidate({'id': 'e_done', 'seq': 10, 'before_cursor': 'before'}) + assert refreshed == current + assert [args['cursor'] for _, args, _ in client.calls] == ['before', 'scan'] + assert all(args['wait_ms'] == 0 for _, args, _ in client.calls) + + +async def test_pending_revocation_drops_cached_event(): + client = Client([{'items': [{'id': 'e_later', 'seq': 11}], 'cursor': 'end', 'has_more': True}]) + assert await Subscription(client).revalidate({ + 'id': 'e_done', 'seq': 10, 'before_cursor': 'before', 'secret': 'stale', + }) is None diff --git a/tests/test_event_subscription.py b/tests/test_event_subscription.py new file mode 100644 index 00000000..141e3b69 --- /dev/null +++ b/tests/test_event_subscription.py @@ -0,0 +1,130 @@ +import asyncio +from datetime import timedelta + +import pytest +from test_resource_event_unification import new_post +from test_service import NOW, call, register + +from msg.core.codec import wire + + +async def events(app, key, subject, **args): + return await call(app, 'communication.events', args, key=key, subject=subject, contract_version=2) + + +async def test_authority_change_reuses_position_and_never_crosses_account(installed): + app, _ = installed + key, owner, _ = await register(app, 'subscription-owner') + other_key, other, _ = await register(app, 'subscription-other') + first = await events(app, key, owner, tail=True) + assert first.status == 'ok', wire(first) + cursor = first.data['cursor'] + async with app.metadata.transaction(write=True) as tx: + tx.set_setting('authorization_epoch', tx.setting('authorization_epoch', 0) + 1) + resumed = await events(app, key, owner, cursor=cursor) + assert resumed.status == 'ok', wire(resumed) + assert resumed.data['authority_changed'] + before = app.cursors.inspect(cursor)['position']['seq'] + assert app.cursors.inspect(resumed.data['cursor'])['position']['seq'] == before + denied = await events(app, other_key, other, cursor=cursor) + assert denied.error.code == 'cursor_query_mismatch' + async with app.metadata.transaction(write=True) as tx: + tx.set_setting('sync_floor', before + 1) + lost = await events(app, key, owner, cursor=cursor) + assert lost.error.code == 'resync_required' + + +async def test_legacy_event_cursor_promotes_without_replaying_consumed_events(installed): + app, _ = installed + key, owner, _ = await register(app, 'subscription-legacy') + old = await call(app, 'communication.events', {}, key=key, subject=owner) + assert old.status == 'ok', wire(old) + while len(old.data['items']) == 50: + old = await call(app, 'communication.events', {'cursor': old.data['cursor']}, + key=key, subject=owner) + cursor = old.data['cursor'] + saved = app.cursors.inspect(cursor) + saved['position']['expires_at'] = wire(NOW - timedelta(seconds=1)) + cursor = app.cursors.encode(saved['kind'], saved['query'], saved['position']) + resumed = await events(app, key, owner, cursor=cursor) + assert resumed.status == 'ok', wire(resumed) + assert app.cursors.inspect(resumed.data['cursor'])['kind'] == 'events-v2' + assert all(item['seq'] > saved['position']['seq'] for item in resumed.data['items']) + assert (await call(app, 'communication.events', {'cursor': cursor}, + key=key, subject=owner)).error.code == 'cursor_expired' + + +async def test_committed_write_wakes_waiter_and_contains_only_authorized_references(installed): + app, _ = installed + key, owner, _ = await register(app, 'subscription-wakeup') + baseline = await events(app, key, owner, tail=True) + signal = app.metadata.event_signal + pending = asyncio.create_task(events(app, key, owner, cursor=baseline.data['cursor'], wait_ms=2000)) + try: + await asyncio.wait_for(signal.ready.wait(), 1) + note = await call(app, 'identity.note_put', {'name': 'done', 'body': 'never in notification'}, + key=key, subject=owner) + assert note.status == 'ok', wire(note) + result = await asyncio.wait_for(pending, 1) + assert result.status == 'ok', wire(result) + assert result.data['items'] + assert 'never in notification' not in str(wire(result.data)) + assert all('before_cursor' in item and 'resume_cursor' in item for item in result.data['items']) + assert signal.waiters == 0 + finally: + if not pending.done(): + pending.cancel() + await asyncio.gather(pending, return_exceptions=True) + + +async def test_rollback_does_not_notify_and_second_process_commit_does(installed): + from msg.storage.event_signal import EventMetadataStore + + app, _ = installed + signal = app.metadata.event_signal + assert await signal.start() + generation = signal.generation + with pytest.raises(RuntimeError): + async with app.metadata.transaction(write=True) as tx: + tx.set_setting('not-committed', True) + raise RuntimeError('rollback') + await asyncio.sleep(0.05) + assert signal.generation == generation + second = EventMetadataStore(app.settings.server.postgres_dsn, initialize=False) + try: + async with second.transaction(write=True) as tx: + tx.set_setting('committed-on-second-store', True) + await signal.wait(generation, 1) + assert signal.generation > generation + finally: + await second.close() + + +async def test_scope_revoked_during_wait_fails_closed_and_regrant_resumes(installed): + app, _ = installed + key, owner, cert = await register(app, 'subscription-scope-owner') + other_key, other, _ = await register(app, 'subscription-scope-reader') + post = await new_post(app, key, owner, cert) + rid = post.resources[0].id + baseline = await events(app, other_key, other, resource=rid, tail=True) + assert baseline.status == 'ok', wire(baseline) + pending = asyncio.create_task(events( + app, other_key, other, resource=rid, cursor=baseline.data['cursor'], wait_ms=2000, + )) + try: + await asyncio.wait_for(app.metadata.event_signal.ready.wait(), 1) + changed = await call(app, 'content.chmod', {'id': rid, 'mode': '0600'}, key=key, + subject=owner, certs=(cert,), expected=((rid, post.data['generation']),)) + assert changed.status == 'ok', wire(changed) + denied = await asyncio.wait_for(pending, 1) + assert denied.status == 'error' and not denied.data + restored = await call(app, 'content.chmod', {'id': rid, 'mode': '0644'}, key=key, + subject=owner, certs=(cert,), expected=((rid, changed.data['generation']),)) + assert restored.status == 'ok', wire(restored) + resumed = await events(app, other_key, other, resource=rid, cursor=baseline.data['cursor']) + assert resumed.status == 'ok', wire(resumed) + assert resumed.data['authority_changed'] + finally: + if not pending.done(): + pending.cancel() + await asyncio.gather(pending, return_exceptions=True) diff --git a/tests/test_event_wait.py b/tests/test_event_wait.py new file mode 100644 index 00000000..3c0e353d --- /dev/null +++ b/tests/test_event_wait.py @@ -0,0 +1,123 @@ +import asyncio +from dataclasses import dataclass, field +from types import SimpleNamespace + +import pytest + +from msg.core.event_wait import EventWaitingExecutor +from msg.core.executor import OperationExecutor + + +@dataclass +class Result: + status: str = 'ok' + data: dict = field(default_factory=lambda: {'items': [], 'has_more': False}) + + +class Signal: + def __init__(self): + self.generation = 0 + self.waiters = 0 + self.starts = 0 + self.waiting = asyncio.Event() + self.changed = asyncio.Event() + + async def start(self, timeout=2): + self.starts += 1 + return True + + async def wait(self, generation, timeout): + self.waiting.set() + try: + await asyncio.wait_for(self.changed.wait(), timeout) + except TimeoutError: + pass + + +def executor(monkeypatch, read): + monkeypatch.setattr(OperationExecutor, 'execute', read) + service = EventWaitingExecutor.__new__(EventWaitingExecutor) + signal = Signal() + service.metadata = SimpleNamespace(event_signal=signal) + packet = SimpleNamespace( + operation='communication.events', contract_version=2, arguments={'wait_ms': 200}, + ) + return service, signal, packet + + +async def test_registration_race_rechecks_before_sleep(monkeypatch): + calls = [] + + async def read(self, packet, *, entry): + calls.append(packet) + return Result(data={'items': ['committed'] if len(calls) == 2 else [], 'has_more': False}) + + service, signal, packet = executor(monkeypatch, read) + result = await service.execute(packet) + assert result.data['items'] == ['committed'] + assert calls == [packet, packet] + assert signal.starts == 1 and signal.waiters == 0 + assert not signal.waiting.is_set() + + +async def test_permission_revoked_while_waiting_returns_error(monkeypatch): + calls = [] + + async def read(self, packet, *, entry): + calls.append(packet) + return Result(status='error') if signal.changed.is_set() else Result() + + service, signal, packet = executor(monkeypatch, read) + pending = asyncio.create_task(service.execute(packet)) + await signal.waiting.wait() + signal.changed.set() + result = await pending + assert result.status == 'error' + assert len(calls) == 3 and all(value is packet for value in calls) + assert signal.waiters == 0 + + +async def test_cancellation_releases_waiter(monkeypatch): + async def read(self, packet, *, entry): + return Result() + + service, signal, packet = executor(monkeypatch, read) + pending = asyncio.create_task(service.execute(packet)) + await signal.waiting.wait() + pending.cancel() + with pytest.raises(asyncio.CancelledError): + await pending + assert signal.waiters == 0 + + +async def test_unauthorized_requests_do_not_open_listener(monkeypatch): + async def read(self, packet, *, entry): + return Result(status='error') + + service, signal, packet = executor(monkeypatch, read) + assert (await service.execute(packet)).status == 'error' + assert signal.starts == 0 + + +async def test_idle_wait_has_bounded_reads_and_no_write(monkeypatch): + calls = [] + + async def read(self, packet, *, entry): + calls.append(packet) + return Result() + + service, signal, packet = executor(monkeypatch, read) + packet.arguments['wait_ms'] = 20 + assert (await service.execute(packet)).status == 'ok' + assert len(calls) == 3 + assert signal.waiters == 0 + + +async def test_saturation_does_not_allocate_more_waiters(monkeypatch): + async def read(self, packet, *, entry): + return Result() + + service, signal, packet = executor(monkeypatch, read) + signal.waiters = 128 + assert (await service.execute(packet)).data['wait_supported'] is False + assert signal.starts == 0 and signal.waiters == 128 diff --git a/tests/test_listener_revalidation.py b/tests/test_listener_revalidation.py new file mode 100644 index 00000000..0983c876 --- /dev/null +++ b/tests/test_listener_revalidation.py @@ -0,0 +1,76 @@ +import io +import json + +import pytest + +from msg.client_listener import listen +from msg.core.errors import Failure + + +async def test_resume_refreshes_pending_body_and_drops_revoked_item(tmp_path): + path = tmp_path / 'cursor.json' + output = io.StringIO() + + async def fetch(cursor, tail=False): + return { + 'items': [{'id': 1}, {'id': 2, 'body': 'old secret'}, {'id': 3}], + 'cursor': 'end', 'has_more': False, + } + + await listen(fetch, context={}, cursor_file=path, max_events=1, output=output) + + async def refresh(item): + return {'id': 2, 'body': 'current allowed'} if item['id'] == 2 else None + + await listen( + fetch, context={}, cursor_file=path, once=True, output=output, revalidate=refresh, + ) + assert [json.loads(line) for line in output.getvalue().splitlines()] == [ + {'id': 1}, {'id': 2, 'body': 'current allowed'}, + ] + assert json.loads(path.read_text())['pending'] == [] + + +async def test_resume_permission_failure_leaves_checkpoint_and_stdout_unchanged(tmp_path): + path = tmp_path / 'cursor.json' + output = io.StringIO() + + async def fetch(cursor, tail=False): + return {'items': [{'id': 1}, {'id': 2}], 'cursor': 'end', 'has_more': False} + + await listen(fetch, context={}, cursor_file=path, max_events=1, output=output) + original = path.read_bytes() + stdout = output.getvalue() + + async def refresh(item): + raise Failure('permission_denied') + + with pytest.raises(Failure, match='permission_denied'): + await listen(fetch, context={}, cursor_file=path, once=True, output=output, revalidate=refresh) + assert path.read_bytes() == original + assert output.getvalue() == stdout + + +async def test_resume_network_retry_reuses_pending_item(tmp_path, monkeypatch): + path = tmp_path / 'cursor.json' + output = io.StringIO() + + async def fetch(cursor, tail=False): + return {'items': [{'id': 1}, {'id': 2}], 'cursor': 'end', 'has_more': False} + + await listen(fetch, context={}, cursor_file=path, max_events=1, output=output) + calls = [] + + async def refresh(item): + calls.append(item['id']) + if len(calls) == 1: + raise OSError('offline') + return item + + async def sleep(delay): + pass + + monkeypatch.setattr('msg.client_listener.asyncio.sleep', sleep) + await listen(fetch, context={}, cursor_file=path, once=True, output=output, revalidate=refresh) + assert calls == [2, 2] + assert len(output.getvalue().splitlines()) == 2 diff --git a/tests/test_task_graph.py b/tests/test_task_graph.py new file mode 100644 index 00000000..344d72df --- /dev/null +++ b/tests/test_task_graph.py @@ -0,0 +1,90 @@ +"""Deterministic task scheduling does not call a model or use a database.""" + +import pytest + +from msg.core.errors import Failure +from msg.core.task_graph import claim, create, finish, renew, settled, summary + + +def graph(concurrency=2): + return create([ + {'id': 'a', 'instruction': 'First'}, + {'id': 'b', 'instruction': 'Second'}, + {'id': 'join', 'instruction': 'Combine', 'dependencies': ['a', 'b']}, + ], concurrency) + + +def take(value, worker='one', attempt='attempt-one', now=1): + return claim(value, worker=worker, attempt=attempt, now=now, lease_seconds=10) + + +def test_fan_in_concurrency_results_and_idempotent_completion(): + original = graph() + current, a = take(original) + current, b = take(current, 'two', 'attempt-two') + capped, nothing = take(current, 'three', 'attempt-three') + assert nothing is None and capped == current + assert summary(original)['running'] == 0 + args = dict(task_id=a['id'], worker='one', attempt='attempt-one', now=2, result_slot=0) + current = finish(current, **args) + assert finish(current, **args) == current + assert summary(current)['ready'] == [] + current = finish(current, task_id=b['id'], worker='two', attempt='attempt-two', now=2, result_slot=1) + assert summary(current)['ready'] == ['join'] + current, joined = take(current, 'three', 'attempt-three', now=3) + assert joined['id'] == 'join' + current = finish(current, task_id='join', worker='three', attempt='attempt-three', now=4, result_slot=2) + assert summary(current)['completed'] == 3 + assert summary(current)['done'] + + +def test_expired_work_is_not_automatically_executed_twice(): + current, _ = take(graph(1)) + expired = settled(current, 11) + assert expired['tasks'][0]['status'] == 'failed' + assert expired['tasks'][2]['status'] == 'blocked' + next_graph, next_task = take(expired, 'two', 'attempt-two', now=12) + assert next_task['id'] == 'b' + with pytest.raises(Failure, match='task_result_conflict'): + finish(next_graph, task_id='a', worker='one', attempt='attempt-one', now=12, result_slot=0) + + +def test_wrong_claim_or_conflicting_result_cannot_unlock_dependencies(): + current, task = take(graph()) + with pytest.raises(Failure, match='task_claim_mismatch'): + finish(current, task_id=task['id'], worker='other', attempt=task['attempt'], now=2, result_slot=0) + current = finish(current, task_id='a', worker='one', attempt='attempt-one', now=2, result_slot=0) + with pytest.raises(Failure, match='task_result_conflict'): + finish(current, task_id='a', worker='one', attempt='attempt-one', now=2, result_slot=1) + + +def test_renew_and_failed_dependency_propagation(): + current, _ = take(graph()) + current = renew(current, task_id='a', worker='one', attempt='attempt-one', now=5, lease_seconds=20) + assert settled(current, 11)['tasks'][0]['status'] == 'running' + failed = finish(current, task_id='a', worker='one', attempt='attempt-one', now=12, error='test failed') + assert summary(failed)['blocked'] == 1 + assert summary(failed)['ready'] == ['b'] + + +def test_dependency_results_must_be_available_to_the_claimant(): + current, task = claim(graph(), worker='one', attempt='a', now=1, lease_seconds=10, permitted={'b'}) + assert task['id'] == 'b' + assert summary(current)['ready'] == ['a'] + + +@pytest.mark.parametrize('tasks,code', [ + ([{'id': 'a', 'instruction': 'A', 'dependencies': ['absent']}], 'unknown_task_dependency'), + ([{'id': 'a', 'instruction': 'A', 'dependencies': ['a']}], 'task_dependency_cycle'), + ([{'id': 'a', 'instruction': 'A'}, {'id': 'a', 'instruction': 'B'}], 'duplicate_task_id'), + ([{'id': 'a', 'instruction': 'A', 'dependencies': ['b', 'b']}, {'id': 'b', 'instruction': 'B'}], 'duplicate_task_dependency'), +]) +def test_invalid_graphs(tasks, code): + with pytest.raises(Failure, match=code): + create(tasks, 1) + + +@pytest.mark.parametrize('concurrency', [True, 0, -1, 65, 1.5]) +def test_invalid_concurrency(concurrency): + with pytest.raises(Failure, match='invalid_task_concurrency'): + graph(concurrency) diff --git a/tests/test_task_plans.py b/tests/test_task_plans.py new file mode 100644 index 00000000..0f01a074 --- /dev/null +++ b/tests/test_task_plans.py @@ -0,0 +1,108 @@ +from datetime import timedelta + +from test_service import NOW, call, register + +from msg.core.codec import wire + + +async def plan(app, key, owner, **extra): + result = await call(app, 'communication.task_plan_create', { + 'title': 'Two tasks', 'concurrency': 1, + 'tasks': [ + {'id': 'a', 'instruction': 'Produce result'}, + {'id': 'b', 'instruction': 'Use result', 'dependencies': ['a']}, + ], **extra, + }, key=key, subject=owner) + assert result.status == 'ok', wire(result) + return result + + +async def change(app, operation, previous, key, owner, **arguments): + result = await call( + app, 'communication.task_' + operation, {'id': previous.data['id'], **arguments}, + key=key, subject=owner, + expected=((previous.data['id'], previous.data['generation']),), + ) + assert result.status == 'ok', wire(result) + return result + + +async def test_plan_releases_dependencies_correlates_results_and_limits_concurrency(installed): + app, _ = installed + key, owner, _ = await register(app, 'task-owner') + made = await plan(app, key, owner) + claimed = await change(app, 'claim', made, key, owner, worker='one') + assert claimed.data['task_id'] == 'a' + busy = await change(app, 'claim', claimed, key, owner, worker='two') + assert busy.data['task_id'] is None and busy.data['summary']['running'] == 1 + note = await call(app, 'identity.note_put', {'name': 'answer', 'body': 'fixed result'}, + key=key, subject=owner) + assert note.status == 'ok', wire(note) + completed = await change( + app, 'complete', busy, key, owner, task_id='a', worker='one', + attempt=claimed.data['attempt'], result=wire(note.resources[0]), + ) + assert completed.data['summary']['ready'] == ['b'] + second = await change(app, 'claim', completed, key, owner, worker='two') + assert second.data['task_id'] == 'b' + final = await change( + app, 'complete', second, key, owner, task_id='b', worker='two', + attempt=second.data['attempt'], result=wire(note.resources[0]), + ) + assert final.data['summary']['done'] and final.data['summary']['succeeded'] == 2 + state = await call(app, 'communication.task_status', {'id': made.data['id']}, key=key, subject=owner) + assert state.data['plan']['graph']['tasks'][0]['result'] == wire(note.resources[0]) + assert state.data['plan']['status'] == 'fulfilled' + old = await call(app, 'communication.request_claim', {'id': made.data['id']}, key=key, + subject=owner, expected=((made.data['id'], final.data['generation']),)) + assert old.error.code == 'task_plan_required' + + +async def test_read_sharing_does_not_grant_task_mutation_or_dependency_result(installed): + app, _ = installed + key, owner, _ = await register(app, 'task-private') + other_key, other, _ = await register(app, 'task-reader') + made = await plan(app, key, owner) + shared = await call(app, 'sharing.grant', { + 'resource': made.data['id'], 'grantee': other, + 'expires_at': wire(NOW + timedelta(days=1)), + }, key=key, subject=owner) + assert shared.status == 'ok', wire(shared) + denied = await call(app, 'communication.task_claim', {'id': made.data['id'], 'worker': 'reader'}, + key=other_key, subject=other, + expected=((made.data['id'], made.data['generation']),)) + assert denied.error.code == 'task_worker_forbidden' + claimed = await change(app, 'claim', made, key, owner, worker='one') + note = await call(app, 'identity.note_put', {'name': 'secret', 'body': 'private result'}, + key=key, subject=owner) + completed = await change(app, 'complete', claimed, key, owner, task_id='a', worker='one', + attempt=claimed.data['attempt'], result=wire(note.resources[0])) + state = await call(app, 'communication.task_status', {'id': made.data['id']}, + key=other_key, subject=other) + assert state.status == 'ok', wire(state) + assert note.resources[0].id not in str(wire(state.data)) + assert 'result' not in state.data['plan']['graph']['tasks'][0] + assert completed.data['summary']['ready'] == ['b'] + + +async def test_generation_fences_racing_claims_and_unpinned_results(installed): + import asyncio + + app, _ = installed + key, owner, _ = await register(app, 'task-race') + made = await plan(app, key, owner) + args = {'id': made.data['id'], 'worker': 'one'} + results = await asyncio.gather(*[ + call(app, 'communication.task_claim', args, key=key, subject=owner, + expected=((made.data['id'], made.data['generation']),)) + for _ in range(2) + ]) + assert sorted(result.status for result in results) == ['error', 'ok'] + winner = next(result for result in results if result.status == 'ok') + loser = next(result for result in results if result.status == 'error') + assert loser.error.code == 'generation_conflict' + bad = await call(app, 'communication.task_complete', { + 'id': made.data['id'], 'task_id': 'a', 'worker': 'one', + 'attempt': winner.data['attempt'], 'result': {'id': made.data['id']}, + }, key=key, subject=owner, expected=((made.data['id'], winner.data['generation']),)) + assert bad.error.code == 'task_result_revision_required' From b1681ffd0d14b041f2deab05d5245b69c28cb3a6 Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Tue, 13 Oct 2026 07:52:16 +0800 Subject: [PATCH 2/4] fix: accept immutable task input and prevent idle claim event loops --- docs/EVENT_COORDINATION.md | 4 +- src/msg/client_agent_cli.py | 29 +++- src/msg/client_subscription.py | 18 +- src/msg/core/task_graph.py | 13 +- src/msg/plugins/event_stream.py | 24 ++- src/msg/plugins/task_plans.py | 113 +++++++++--- src/msg/storage/event_signal.py | 2 +- tests/test_client_subscription.py | 12 +- tests/test_event_subscription.py | 100 +++++++++-- tests/test_event_wait.py | 4 +- tests/test_listener_revalidation.py | 17 +- tests/test_task_graph.py | 75 +++++--- tests/test_task_plans.py | 255 +++++++++++++++++++++++----- 13 files changed, 530 insertions(+), 136 deletions(-) diff --git a/docs/EVENT_COORDINATION.md b/docs/EVENT_COORDINATION.md index fe838686..5215ba36 100644 --- a/docs/EVENT_COORDINATION.md +++ b/docs/EVENT_COORDINATION.md @@ -99,7 +99,9 @@ Call these operations through the existing signed client/API: Mutations provide the plan's expected generation in the ordinary request envelope. The database serializes writes across processes. A competing generation loses -rather than claiming the same slot. Retry a lost response with the **same request +rather than claiming the same slot. If no task is ready and no lease expired, +`task_claim` returns `task_not_ready` without writing a revision or completion +event. Wait for a new event instead of retrying immediately. Retry a lost response with the **same request ID and unchanged packet**. After a confirmed generation conflict, read the current plan and make a new decision/request. Do not change a packet under an existing ID. diff --git a/src/msg/client_agent_cli.py b/src/msg/client_agent_cli.py index c3b2d2fc..1429e6fb 100644 --- a/src/msg/client_agent_cli.py +++ b/src/msg/client_agent_cli.py @@ -66,7 +66,8 @@ def add_commands(commands): help='Exact event type; repeat to include more types.', ) listener.add_argument( - '--subscribe', action='store_true', + '--subscribe', + action='store_true', help='Use authorized server event waits (events@2); no mailbox or offline fallback.', ) listener.add_argument('--resource', help='With --subscribe, limit events to this resource.') @@ -88,6 +89,10 @@ def add_commands(commands): def local_command(args): + require( + not getattr(args, 'resource', None) or getattr(args, 'subscribe', False), + 'subscription_required', + ) if getattr(args, 'subscribe', False): return False return (args.command == 'agent' or args.command == 'listen' and args.agent) and not args.remote @@ -237,11 +242,18 @@ async def run_remote(client, args): require(not args.agent, 'subscription_requires_account_reader') source = Subscription( - client, resource=args.resource, once=args.once, interval=args.interval, + client, + resource=args.resource, + once=args.once, + interval=args.interval, ) return await stream( - client.state, args, source.fetch, 'subscription:' + (args.resource or ''), - revalidate=source.revalidate, server_wait=True, + client.state, + args, + source.fetch, + 'subscription:' + (args.resource or ''), + revalidate=source.revalidate, + server_wait=True, ) require(not getattr(args, 'resource', None), 'subscription_required') if args.command == 'listen' and not args.agent: @@ -272,13 +284,18 @@ async def revalidate(item): event = wire(client.checked(current).data) # Old audit bodies must not survive a current-authorization refresh. return { - **event, 'type': event['operation'], + **event, + 'type': event['operation'], 'data': {'operation': event['operation']}, 'resources': [address['ref'] for address in event['resources']], } return await stream( - client.state, args, fetch, 'account-events', revalidate=revalidate, + client.state, + args, + fetch, + 'account-events', + revalidate=revalidate, ) from msg.client_subagents_remote import RemoteAgents diff --git a/src/msg/client_subscription.py b/src/msg/client_subscription.py index 7bd8923c..eea6afc7 100644 --- a/src/msg/client_subscription.py +++ b/src/msg/client_subscription.py @@ -22,9 +22,13 @@ async def _read(self, cursor, *, tail=False, wait=False, limit=50): args['cursor'] = cursor if tail: args['tail'] = True - result = self.client.checked(await self.client.call( - 'communication.events', args, contract_version=2, - )) + result = self.client.checked( + await self.client.call( + 'communication.events', + args, + contract_version=2, + ) + ) return wire(result.data) async def fetch(self, cursor, tail=False): @@ -32,10 +36,14 @@ async def fetch(self, cursor, tail=False): if page.get('authority_changed'): print( 'msg listen: authority changed; position retained. Refresh derived resource state.', - file=sys.stderr, flush=True, + file=sys.stderr, + flush=True, ) if ( - not self.once and not tail and not page['items'] and not page['has_more'] + not self.once + and not tail + and not page['items'] + and not page['has_more'] and not page.get('wait_supported') ): # Explicitly advertised fallback, not a silent switch of source or account. diff --git a/src/msg/core/task_graph.py b/src/msg/core/task_graph.py index dbb81160..ebc2831c 100644 --- a/src/msg/core/task_graph.py +++ b/src/msg/core/task_graph.py @@ -29,7 +29,9 @@ def create(tasks, concurrency): 'unknown_task_dependency', ) try: - order = list(TopologicalSorter({k: t['dependencies'] for k, t in records.items()}).static_order()) + order = list( + TopologicalSorter({k: t['dependencies'] for k, t in records.items()}).static_order() + ) except CycleError: require(False, 'task_dependency_cycle') return {'concurrency': concurrency, 'tasks': [records[name] for name in order]} @@ -51,7 +53,7 @@ def settled(graph, now): def summary(graph): tasks = {task['id']: task for task in graph['tasks']} - counts = {status: 0 for status in ('pending', 'running', 'succeeded', 'failed', 'blocked')} + counts = dict.fromkeys(('pending', 'running', 'succeeded', 'failed', 'blocked'), 0) ready = [] for task in tasks.values(): counts[task['status']] += 1 @@ -78,8 +80,11 @@ def claim(graph, *, worker, attempt, now, lease_seconds, permitted=None): for task in graph['tasks']: if task['id'] in state['ready'] and (permitted is None or task['id'] in permitted): task.update( - status='running', worker=worker, attempt=attempt, - lease_until=now + lease_seconds, started_at=now, + status='running', + worker=worker, + attempt=attempt, + lease_until=now + lease_seconds, + started_at=now, ) return graph, deepcopy(task) return graph, None diff --git a/src/msg/plugins/event_stream.py b/src/msg/plugins/event_stream.py index 6deeb5c1..91cc0a11 100644 --- a/src/msg/plugins/event_stream.py +++ b/src/msg/plugins/event_stream.py @@ -102,8 +102,14 @@ async def events(ctx, request, tx): 'epoch': tx.setting('authorization_epoch', 0), 'watches': sorted(watches), 'topics': [tuple(row) for row in memberships], - **({'credential_id': ctx.principal.credential_id, - 'certificates': sorted(ctx.principal.certificates)} if subscription else {}), + **( + { + 'credential_id': ctx.principal.credential_id, + 'certificates': sorted(ctx.principal.certificates), + } + if subscription + else {} + ), }) floor = tx.setting('sync_floor', 0) position = floor @@ -212,11 +218,15 @@ def resume(sequence): 'cursor': token, **({'next': next_link(app, request.operation, args)} if not subscription else {}), 'next_requires_auth': True, - **({ - 'has_more': position < tail, - 'authority_changed': authority_changed, - 'wait_supported': getattr(app.metadata, 'event_signal', None) is not None, - } if subscription else {}), + **( + { + 'has_more': position < tail, + 'authority_changed': authority_changed, + 'wait_supported': getattr(app.metadata, 'event_signal', None) is not None, + } + if subscription + else {} + ), } ) diff --git a/src/msg/plugins/task_plans.py b/src/msg/plugins/task_plans.py index 0cb8dc7b..1234220e 100644 --- a/src/msg/plugins/task_plans.py +++ b/src/msg/plugins/task_plans.py @@ -39,8 +39,13 @@ LEASE = {'type': 'integer', 'minimum': 1, 'maximum': 3600} # Loss of resource access is not a transient failure or a reason to publish a result. HIDDEN = { - 'permission_denied', 'credential_ceiling', 'certificate_gate', 'delegation_scope', - 'ancestor_inactive', 'not_found', 'collaboration_ref_forbidden', + 'permission_denied', + 'credential_ceiling', + 'certificate_gate', + 'delegation_scope', + 'ancestor_inactive', + 'not_found', + 'collaboration_ref_forbidden', } @@ -77,12 +82,20 @@ async def create(ctx, request, tx): 'status': 'open', 'scheduler_version': 1, 'workers': workers, - 'graph': task_graph.create(args['tasks'], args['concurrency']), + 'graph': task_graph.create(wire(args['tasks']), args['concurrency']), } parent = await _folder(app, ctx, request, tx, subject, 'request') resource = await create_resource( - app, ctx, request, tx, parent=parent, type=KINDS['request'], name=new_id('plan'), - body=canonical(record), media_type='application/json', mode=0o600, + app, + ctx, + request, + tx, + parent=parent, + type=KINDS['request'], + name=new_id('plan'), + body=canonical(record), + media_type='application/json', + mode=0o600, ) return _result(resource, record) @@ -98,29 +111,46 @@ async def load(ctx, request, tx, *, mutate=False): @op('communication.task_status', obj({'id': IDENTIFIER}, ('id',)), effect='read') async def status(ctx, request, tx): _, resource, revision, record = await load(ctx, request, tx) - return HandlerOutput(data={ - 'plan': await _project(app, ctx, request, tx, resource, revision, record), - }) + return HandlerOutput( + data={ + 'plan': await _project(app, ctx, request, tx, resource, revision, record), + } + ) async def save(ctx, request, tx, resource, revision, record, graph, *, task=None): state = task_graph.summary(graph) record = {**record, 'graph': graph, 'updated_at': wire(ctx.now)} record['status'] = ( - 'fulfilled' if state['succeeded'] == state['total'] - else 'cancelled' if state['done'] else 'claimed' if state['running'] else 'open' + 'fulfilled' + if state['succeeded'] == state['total'] + else 'cancelled' + if state['done'] + else 'claimed' + if state['running'] + else 'open' ) resource = await revise_resource( - app, ctx, request, tx, resource, canonical(record), 'application/json', - relations=revision.relations, preserve_existing_relations=True, + app, + ctx, + request, + tx, + resource, + canonical(record), + 'application/json', + relations=revision.relations, + preserve_existing_relations=True, ) output = _result(resource, record) # No cached instructions, dependency results, or private reference bodies. - return HandlerOutput(resources=output.resources, data={ - **output.data, - 'task_id': task['id'] if task else None, - 'attempt': task['attempt'] if task else None, - 'summary': state, - }) + return HandlerOutput( + resources=output.resources, + data={ + **output.data, + 'task_id': task['id'] if task else None, + 'attempt': task['attempt'] if task else None, + 'summary': state, + }, + ) @op( 'communication.task_claim', @@ -143,10 +173,22 @@ async def claim(ctx, request, tx): else: permitted.add(name) graph, task = task_graph.claim( - graph, worker=subject + '#' + request.arguments['worker'], attempt=new_id('attempt'), - now=ctx.now.timestamp(), lease_seconds=request.arguments.get('lease_seconds', 300), + graph, + worker=subject + '#' + request.arguments['worker'], + attempt=new_id('attempt'), + now=ctx.now.timestamp(), + lease_seconds=request.arguments.get('lease_seconds', 300), permitted=permitted, ) + require( + task is not None or graph != record['graph'], + 'task_not_ready', + details={ + 'id': resource.id, + 'generation': resource.generation, + 'summary': task_graph.summary(graph), + }, + ) return await save(ctx, request, tx, resource, revision, record, graph, task=task) common = {'id': IDENTIFIER, 'task_id': NAME, 'worker': NAME, 'attempt': NAME} @@ -175,12 +217,22 @@ async def complete(ctx, request, tx): relations.append(relation) slot = relations.index(relation) graph = task_graph.finish( - record['graph'], task_id=args['task_id'], worker=subject + '#' + args['worker'], - attempt=args['attempt'], now=ctx.now.timestamp(), result_slot=slot, + record['graph'], + task_id=args['task_id'], + worker=subject + '#' + args['worker'], + attempt=args['attempt'], + now=ctx.now.timestamp(), + result_slot=slot, error=args.get('error'), ) return await save( - ctx, request, tx, resource, replace(revision, relations=tuple(relations)), record, graph, + ctx, + request, + tx, + resource, + replace(revision, relations=tuple(relations)), + record, + graph, task=next(task for task in graph['tasks'] if task['id'] == args['task_id']), ) @@ -193,11 +245,20 @@ async def renew(ctx, request, tx): subject, resource, revision, record = await load(ctx, request, tx, mutate=True) args = request.arguments graph = task_graph.renew( - record['graph'], task_id=args['task_id'], worker=subject + '#' + args['worker'], - attempt=args['attempt'], now=ctx.now.timestamp(), + record['graph'], + task_id=args['task_id'], + worker=subject + '#' + args['worker'], + attempt=args['attempt'], + now=ctx.now.timestamp(), lease_seconds=args.get('lease_seconds', 300), ) return await save( - ctx, request, tx, resource, revision, record, graph, + ctx, + request, + tx, + resource, + revision, + record, + graph, task=next(task for task in graph['tasks'] if task['id'] == args['task_id']), ) diff --git a/src/msg/storage/event_signal.py b/src/msg/storage/event_signal.py index 9c61e9a7..d550eab9 100644 --- a/src/msg/storage/event_signal.py +++ b/src/msg/storage/event_signal.py @@ -49,7 +49,7 @@ async def _receive(self): async with aclosing(connection.notifies()) as notices: async for _ in notices: self._bump() - except (psycopg.Error, OSError): + except psycopg.Error, OSError: # The durable log remains authoritative. Reconnection wakes # every reader, which repeats its independently authorized read. pass diff --git a/tests/test_client_subscription.py b/tests/test_client_subscription.py index bea5f0be..f0215499 100644 --- a/tests/test_client_subscription.py +++ b/tests/test_client_subscription.py @@ -42,6 +42,12 @@ async def test_pending_revalidation_returns_new_projection_not_cached_body(): async def test_pending_revocation_drops_cached_event(): client = Client([{'items': [{'id': 'e_later', 'seq': 11}], 'cursor': 'end', 'has_more': True}]) - assert await Subscription(client).revalidate({ - 'id': 'e_done', 'seq': 10, 'before_cursor': 'before', 'secret': 'stale', - }) is None + assert ( + await Subscription(client).revalidate({ + 'id': 'e_done', + 'seq': 10, + 'before_cursor': 'before', + 'secret': 'stale', + }) + is None + ) diff --git a/tests/test_event_subscription.py b/tests/test_event_subscription.py index 141e3b69..1de19b2a 100644 --- a/tests/test_event_subscription.py +++ b/tests/test_event_subscription.py @@ -9,7 +9,9 @@ async def events(app, key, subject, **args): - return await call(app, 'communication.events', args, key=key, subject=subject, contract_version=2) + return await call( + app, 'communication.events', args, key=key, subject=subject, contract_version=2 + ) async def test_authority_change_reuses_position_and_never_crosses_account(installed): @@ -40,8 +42,9 @@ async def test_legacy_event_cursor_promotes_without_replaying_consumed_events(in old = await call(app, 'communication.events', {}, key=key, subject=owner) assert old.status == 'ok', wire(old) while len(old.data['items']) == 50: - old = await call(app, 'communication.events', {'cursor': old.data['cursor']}, - key=key, subject=owner) + old = await call( + app, 'communication.events', {'cursor': old.data['cursor']}, key=key, subject=owner + ) cursor = old.data['cursor'] saved = app.cursors.inspect(cursor) saved['position']['expires_at'] = wire(NOW - timedelta(seconds=1)) @@ -50,8 +53,9 @@ async def test_legacy_event_cursor_promotes_without_replaying_consumed_events(in assert resumed.status == 'ok', wire(resumed) assert app.cursors.inspect(resumed.data['cursor'])['kind'] == 'events-v2' assert all(item['seq'] > saved['position']['seq'] for item in resumed.data['items']) - assert (await call(app, 'communication.events', {'cursor': cursor}, - key=key, subject=owner)).error.code == 'cursor_expired' + assert ( + await call(app, 'communication.events', {'cursor': cursor}, key=key, subject=owner) + ).error.code == 'cursor_expired' async def test_committed_write_wakes_waiter_and_contains_only_authorized_references(installed): @@ -59,17 +63,26 @@ async def test_committed_write_wakes_waiter_and_contains_only_authorized_referen key, owner, _ = await register(app, 'subscription-wakeup') baseline = await events(app, key, owner, tail=True) signal = app.metadata.event_signal - pending = asyncio.create_task(events(app, key, owner, cursor=baseline.data['cursor'], wait_ms=2000)) + pending = asyncio.create_task( + events(app, key, owner, cursor=baseline.data['cursor'], wait_ms=2000) + ) try: await asyncio.wait_for(signal.ready.wait(), 1) - note = await call(app, 'identity.note_put', {'name': 'done', 'body': 'never in notification'}, - key=key, subject=owner) + note = await call( + app, + 'identity.note_put', + {'name': 'done', 'body': 'never in notification'}, + key=key, + subject=owner, + ) assert note.status == 'ok', wire(note) result = await asyncio.wait_for(pending, 1) assert result.status == 'ok', wire(result) assert result.data['items'] assert 'never in notification' not in str(wire(result.data)) - assert all('before_cursor' in item and 'resume_cursor' in item for item in result.data['items']) + assert all( + 'before_cursor' in item and 'resume_cursor' in item for item in result.data['items'] + ) assert signal.waiters == 0 finally: if not pending.done(): @@ -108,18 +121,39 @@ async def test_scope_revoked_during_wait_fails_closed_and_regrant_resumes(instal rid = post.resources[0].id baseline = await events(app, other_key, other, resource=rid, tail=True) assert baseline.status == 'ok', wire(baseline) - pending = asyncio.create_task(events( - app, other_key, other, resource=rid, cursor=baseline.data['cursor'], wait_ms=2000, - )) + pending = asyncio.create_task( + events( + app, + other_key, + other, + resource=rid, + cursor=baseline.data['cursor'], + wait_ms=2000, + ) + ) try: await asyncio.wait_for(app.metadata.event_signal.ready.wait(), 1) - changed = await call(app, 'content.chmod', {'id': rid, 'mode': '0600'}, key=key, - subject=owner, certs=(cert,), expected=((rid, post.data['generation']),)) + changed = await call( + app, + 'content.chmod', + {'id': rid, 'mode': '0600'}, + key=key, + subject=owner, + certs=(cert,), + expected=((rid, post.data['generation']),), + ) assert changed.status == 'ok', wire(changed) denied = await asyncio.wait_for(pending, 1) assert denied.status == 'error' and not denied.data - restored = await call(app, 'content.chmod', {'id': rid, 'mode': '0644'}, key=key, - subject=owner, certs=(cert,), expected=((rid, changed.data['generation']),)) + restored = await call( + app, + 'content.chmod', + {'id': rid, 'mode': '0644'}, + key=key, + subject=owner, + certs=(cert,), + expected=((rid, changed.data['generation']),), + ) assert restored.status == 'ok', wire(restored) resumed = await events(app, other_key, other, resource=rid, cursor=baseline.data['cursor']) assert resumed.status == 'ok', wire(resumed) @@ -128,3 +162,37 @@ async def test_scope_revoked_during_wait_fails_closed_and_regrant_resumes(instal if not pending.done(): pending.cancel() await asyncio.gather(pending, return_exceptions=True) + + +async def test_notification_connection_loss_recovers_same_durable_cursor(installed): + app, _ = installed + key, owner, _ = await register(app, 'subscription-reconnect') + initial = await events(app, key, owner, tail=True) + cursor = initial.data['cursor'] + signal = app.metadata.event_signal + assert await signal.start() + generation = signal.generation + async with app.metadata.transaction(write=False) as tx: + killed = tx.rows( + 'SELECT pg_terminate_backend(pid) FROM pg_stat_activity ' + "WHERE datname=current_database() AND query='LISTEN msg_committed_events' " + 'AND pid<>pg_backend_pid()' + ) + assert killed and all(row[0] for row in killed) + await signal.wait(generation, 2) + note = await call( + app, + 'identity.note_put', + {'name': 'during-reconnect', 'body': 'Durable result'}, + key=key, + subject=owner, + ) + assert note.status == 'ok', wire(note.error) + assert await signal.start(timeout=5) + resumed = await events(app, key, owner, cursor=cursor) + assert resumed.status == 'ok', wire(resumed.error) + assert any( + address['ref']['id'] == note.resources[0].id + for item in resumed.data['items'] + for address in item['resources'] + ) diff --git a/tests/test_event_wait.py b/tests/test_event_wait.py index 3c0e353d..8aba3c5e 100644 --- a/tests/test_event_wait.py +++ b/tests/test_event_wait.py @@ -40,7 +40,9 @@ def executor(monkeypatch, read): signal = Signal() service.metadata = SimpleNamespace(event_signal=signal) packet = SimpleNamespace( - operation='communication.events', contract_version=2, arguments={'wait_ms': 200}, + operation='communication.events', + contract_version=2, + arguments={'wait_ms': 200}, ) return service, signal, packet diff --git a/tests/test_listener_revalidation.py b/tests/test_listener_revalidation.py index 0983c876..b1f32cbe 100644 --- a/tests/test_listener_revalidation.py +++ b/tests/test_listener_revalidation.py @@ -14,7 +14,8 @@ async def test_resume_refreshes_pending_body_and_drops_revoked_item(tmp_path): async def fetch(cursor, tail=False): return { 'items': [{'id': 1}, {'id': 2, 'body': 'old secret'}, {'id': 3}], - 'cursor': 'end', 'has_more': False, + 'cursor': 'end', + 'has_more': False, } await listen(fetch, context={}, cursor_file=path, max_events=1, output=output) @@ -23,10 +24,16 @@ async def refresh(item): return {'id': 2, 'body': 'current allowed'} if item['id'] == 2 else None await listen( - fetch, context={}, cursor_file=path, once=True, output=output, revalidate=refresh, + fetch, + context={}, + cursor_file=path, + once=True, + output=output, + revalidate=refresh, ) assert [json.loads(line) for line in output.getvalue().splitlines()] == [ - {'id': 1}, {'id': 2, 'body': 'current allowed'}, + {'id': 1}, + {'id': 2, 'body': 'current allowed'}, ] assert json.loads(path.read_text())['pending'] == [] @@ -46,7 +53,9 @@ async def refresh(item): raise Failure('permission_denied') with pytest.raises(Failure, match='permission_denied'): - await listen(fetch, context={}, cursor_file=path, once=True, output=output, revalidate=refresh) + await listen( + fetch, context={}, cursor_file=path, once=True, output=output, revalidate=refresh + ) assert path.read_bytes() == original assert output.getvalue() == stdout diff --git a/tests/test_task_graph.py b/tests/test_task_graph.py index 344d72df..3f345ca0 100644 --- a/tests/test_task_graph.py +++ b/tests/test_task_graph.py @@ -7,11 +7,14 @@ def graph(concurrency=2): - return create([ - {'id': 'a', 'instruction': 'First'}, - {'id': 'b', 'instruction': 'Second'}, - {'id': 'join', 'instruction': 'Combine', 'dependencies': ['a', 'b']}, - ], concurrency) + return create( + [ + {'id': 'a', 'instruction': 'First'}, + {'id': 'b', 'instruction': 'Second'}, + {'id': 'join', 'instruction': 'Combine', 'dependencies': ['a', 'b']}, + ], + concurrency, + ) def take(value, worker='one', attempt='attempt-one', now=1): @@ -25,15 +28,25 @@ def test_fan_in_concurrency_results_and_idempotent_completion(): capped, nothing = take(current, 'three', 'attempt-three') assert nothing is None and capped == current assert summary(original)['running'] == 0 - args = dict(task_id=a['id'], worker='one', attempt='attempt-one', now=2, result_slot=0) + args = { + 'task_id': a['id'], + 'worker': 'one', + 'attempt': 'attempt-one', + 'now': 2, + 'result_slot': 0, + } current = finish(current, **args) assert finish(current, **args) == current assert summary(current)['ready'] == [] - current = finish(current, task_id=b['id'], worker='two', attempt='attempt-two', now=2, result_slot=1) + current = finish( + current, task_id=b['id'], worker='two', attempt='attempt-two', now=2, result_slot=1 + ) assert summary(current)['ready'] == ['join'] current, joined = take(current, 'three', 'attempt-three', now=3) assert joined['id'] == 'join' - current = finish(current, task_id='join', worker='three', attempt='attempt-three', now=4, result_slot=2) + current = finish( + current, task_id='join', worker='three', attempt='attempt-three', now=4, result_slot=2 + ) assert summary(current)['completed'] == 3 assert summary(current)['done'] @@ -52,33 +65,57 @@ def test_expired_work_is_not_automatically_executed_twice(): def test_wrong_claim_or_conflicting_result_cannot_unlock_dependencies(): current, task = take(graph()) with pytest.raises(Failure, match='task_claim_mismatch'): - finish(current, task_id=task['id'], worker='other', attempt=task['attempt'], now=2, result_slot=0) - current = finish(current, task_id='a', worker='one', attempt='attempt-one', now=2, result_slot=0) + finish( + current, + task_id=task['id'], + worker='other', + attempt=task['attempt'], + now=2, + result_slot=0, + ) + current = finish( + current, task_id='a', worker='one', attempt='attempt-one', now=2, result_slot=0 + ) with pytest.raises(Failure, match='task_result_conflict'): finish(current, task_id='a', worker='one', attempt='attempt-one', now=2, result_slot=1) def test_renew_and_failed_dependency_propagation(): current, _ = take(graph()) - current = renew(current, task_id='a', worker='one', attempt='attempt-one', now=5, lease_seconds=20) + current = renew( + current, task_id='a', worker='one', attempt='attempt-one', now=5, lease_seconds=20 + ) assert settled(current, 11)['tasks'][0]['status'] == 'running' - failed = finish(current, task_id='a', worker='one', attempt='attempt-one', now=12, error='test failed') + failed = finish( + current, task_id='a', worker='one', attempt='attempt-one', now=12, error='test failed' + ) assert summary(failed)['blocked'] == 1 assert summary(failed)['ready'] == ['b'] def test_dependency_results_must_be_available_to_the_claimant(): - current, task = claim(graph(), worker='one', attempt='a', now=1, lease_seconds=10, permitted={'b'}) + current, task = claim( + graph(), worker='one', attempt='a', now=1, lease_seconds=10, permitted={'b'} + ) assert task['id'] == 'b' assert summary(current)['ready'] == ['a'] -@pytest.mark.parametrize('tasks,code', [ - ([{'id': 'a', 'instruction': 'A', 'dependencies': ['absent']}], 'unknown_task_dependency'), - ([{'id': 'a', 'instruction': 'A', 'dependencies': ['a']}], 'task_dependency_cycle'), - ([{'id': 'a', 'instruction': 'A'}, {'id': 'a', 'instruction': 'B'}], 'duplicate_task_id'), - ([{'id': 'a', 'instruction': 'A', 'dependencies': ['b', 'b']}, {'id': 'b', 'instruction': 'B'}], 'duplicate_task_dependency'), -]) +@pytest.mark.parametrize( + 'tasks,code', + [ + ([{'id': 'a', 'instruction': 'A', 'dependencies': ['absent']}], 'unknown_task_dependency'), + ([{'id': 'a', 'instruction': 'A', 'dependencies': ['a']}], 'task_dependency_cycle'), + ([{'id': 'a', 'instruction': 'A'}, {'id': 'a', 'instruction': 'B'}], 'duplicate_task_id'), + ( + [ + {'id': 'a', 'instruction': 'A', 'dependencies': ['b', 'b']}, + {'id': 'b', 'instruction': 'B'}, + ], + 'duplicate_task_dependency', + ), + ], +) def test_invalid_graphs(tasks, code): with pytest.raises(Failure, match=code): create(tasks, 1) diff --git a/tests/test_task_plans.py b/tests/test_task_plans.py index 0f01a074..2439c52a 100644 --- a/tests/test_task_plans.py +++ b/tests/test_task_plans.py @@ -6,24 +6,35 @@ async def plan(app, key, owner, **extra): - result = await call(app, 'communication.task_plan_create', { - 'title': 'Two tasks', 'concurrency': 1, - 'tasks': [ - {'id': 'a', 'instruction': 'Produce result'}, - {'id': 'b', 'instruction': 'Use result', 'dependencies': ['a']}, - ], **extra, - }, key=key, subject=owner) - assert result.status == 'ok', wire(result) + result = await call( + app, + 'communication.task_plan_create', + { + 'title': 'Two tasks', + 'concurrency': 1, + 'tasks': [ + {'id': 'a', 'instruction': 'Produce result'}, + {'id': 'b', 'instruction': 'Use result', 'dependencies': ['a']}, + ], + **extra, + }, + key=key, + subject=owner, + ) + assert result.status == 'ok', wire(result.error) return result async def change(app, operation, previous, key, owner, **arguments): result = await call( - app, 'communication.task_' + operation, {'id': previous.data['id'], **arguments}, - key=key, subject=owner, + app, + 'communication.task_' + operation, + {'id': previous.data['id'], **arguments}, + key=key, + subject=owner, expected=((previous.data['id'], previous.data['generation']),), ) - assert result.status == 'ok', wire(result) + assert result.status == 'ok', wire(result.error) return result @@ -33,28 +44,63 @@ async def test_plan_releases_dependencies_correlates_results_and_limits_concurre made = await plan(app, key, owner) claimed = await change(app, 'claim', made, key, owner, worker='one') assert claimed.data['task_id'] == 'a' - busy = await change(app, 'claim', claimed, key, owner, worker='two') - assert busy.data['task_id'] is None and busy.data['summary']['running'] == 1 - note = await call(app, 'identity.note_put', {'name': 'answer', 'body': 'fixed result'}, - key=key, subject=owner) + async with app.metadata.transaction(write=False) as tx: + events_before = tx.one('SELECT COUNT(*) FROM events')[0] + busy = await call( + app, + 'communication.task_claim', + {'id': made.data['id'], 'worker': 'two'}, + key=key, + subject=owner, + expected=((made.data['id'], claimed.data['generation']),), + ) + assert busy.error.code == 'task_not_ready' + assert busy.data['summary']['running'] == 1 + async with app.metadata.transaction(write=False) as tx: + assert tx.one('SELECT COUNT(*) FROM events')[0] == events_before + note = await call( + app, 'identity.note_put', {'name': 'answer', 'body': 'fixed result'}, key=key, subject=owner + ) assert note.status == 'ok', wire(note) completed = await change( - app, 'complete', busy, key, owner, task_id='a', worker='one', - attempt=claimed.data['attempt'], result=wire(note.resources[0]), + app, + 'complete', + claimed, + key, + owner, + task_id='a', + worker='one', + attempt=claimed.data['attempt'], + result=wire(note.resources[0]), ) - assert completed.data['summary']['ready'] == ['b'] + assert completed.data['summary']['ready'] == ('b',) second = await change(app, 'claim', completed, key, owner, worker='two') assert second.data['task_id'] == 'b' final = await change( - app, 'complete', second, key, owner, task_id='b', worker='two', - attempt=second.data['attempt'], result=wire(note.resources[0]), + app, + 'complete', + second, + key, + owner, + task_id='b', + worker='two', + attempt=second.data['attempt'], + result=wire(note.resources[0]), ) assert final.data['summary']['done'] and final.data['summary']['succeeded'] == 2 - state = await call(app, 'communication.task_status', {'id': made.data['id']}, key=key, subject=owner) + state = await call( + app, 'communication.task_status', {'id': made.data['id']}, key=key, subject=owner + ) assert state.data['plan']['graph']['tasks'][0]['result'] == wire(note.resources[0]) assert state.data['plan']['status'] == 'fulfilled' - old = await call(app, 'communication.request_claim', {'id': made.data['id']}, key=key, - subject=owner, expected=((made.data['id'], final.data['generation']),)) + old = await call( + app, + 'communication.request_claim', + {'id': made.data['id']}, + key=key, + subject=owner, + expected=((made.data['id'], final.data['generation']),), + ) assert old.error.code == 'task_plan_required' @@ -63,26 +109,53 @@ async def test_read_sharing_does_not_grant_task_mutation_or_dependency_result(in key, owner, _ = await register(app, 'task-private') other_key, other, _ = await register(app, 'task-reader') made = await plan(app, key, owner) - shared = await call(app, 'sharing.grant', { - 'resource': made.data['id'], 'grantee': other, - 'expires_at': wire(NOW + timedelta(days=1)), - }, key=key, subject=owner) + shared = await call( + app, + 'sharing.grant', + { + 'resource': made.data['id'], + 'grantee': other, + 'expires_at': wire(NOW + timedelta(days=1)), + }, + key=key, + subject=owner, + ) assert shared.status == 'ok', wire(shared) - denied = await call(app, 'communication.task_claim', {'id': made.data['id'], 'worker': 'reader'}, - key=other_key, subject=other, - expected=((made.data['id'], made.data['generation']),)) + denied = await call( + app, + 'communication.task_claim', + {'id': made.data['id'], 'worker': 'reader'}, + key=other_key, + subject=other, + expected=((made.data['id'], made.data['generation']),), + ) assert denied.error.code == 'task_worker_forbidden' claimed = await change(app, 'claim', made, key, owner, worker='one') - note = await call(app, 'identity.note_put', {'name': 'secret', 'body': 'private result'}, - key=key, subject=owner) - completed = await change(app, 'complete', claimed, key, owner, task_id='a', worker='one', - attempt=claimed.data['attempt'], result=wire(note.resources[0])) - state = await call(app, 'communication.task_status', {'id': made.data['id']}, - key=other_key, subject=other) + note = await call( + app, + 'identity.note_put', + {'name': 'secret', 'body': 'private result'}, + key=key, + subject=owner, + ) + completed = await change( + app, + 'complete', + claimed, + key, + owner, + task_id='a', + worker='one', + attempt=claimed.data['attempt'], + result=wire(note.resources[0]), + ) + state = await call( + app, 'communication.task_status', {'id': made.data['id']}, key=other_key, subject=other + ) assert state.status == 'ok', wire(state) assert note.resources[0].id not in str(wire(state.data)) assert 'result' not in state.data['plan']['graph']['tasks'][0] - assert completed.data['summary']['ready'] == ['b'] + assert completed.data['summary']['ready'] == ('b',) async def test_generation_fences_racing_claims_and_unpinned_results(installed): @@ -93,16 +166,112 @@ async def test_generation_fences_racing_claims_and_unpinned_results(installed): made = await plan(app, key, owner) args = {'id': made.data['id'], 'worker': 'one'} results = await asyncio.gather(*[ - call(app, 'communication.task_claim', args, key=key, subject=owner, - expected=((made.data['id'], made.data['generation']),)) + call( + app, + 'communication.task_claim', + args, + key=key, + subject=owner, + expected=((made.data['id'], made.data['generation']),), + ) for _ in range(2) ]) assert sorted(result.status for result in results) == ['error', 'ok'] winner = next(result for result in results if result.status == 'ok') loser = next(result for result in results if result.status == 'error') assert loser.error.code == 'generation_conflict' - bad = await call(app, 'communication.task_complete', { - 'id': made.data['id'], 'task_id': 'a', 'worker': 'one', - 'attempt': winner.data['attempt'], 'result': {'id': made.data['id']}, - }, key=key, subject=owner, expected=((made.data['id'], winner.data['generation']),)) + bad = await call( + app, + 'communication.task_complete', + { + 'id': made.data['id'], + 'task_id': 'a', + 'worker': 'one', + 'attempt': winner.data['attempt'], + 'result': {'id': made.data['id']}, + }, + key=key, + subject=owner, + expected=((made.data['id'], winner.data['generation']),), + ) assert bad.error.code == 'task_result_revision_required' + + +async def test_worker_hidden_results_survive_independent_completion(installed): + app, _ = installed + key, owner, _ = await register(app, 'task-result-owner') + other_key, other, _ = await register(app, 'task-result-worker') + made = await plan( + app, + key, + owner, + workers=[owner, other], + tasks=[ + {'id': 'a', 'instruction': 'Private first task'}, + {'id': 'independent', 'instruction': 'Independent work'}, + {'id': 'dependent', 'instruction': 'Needs first result', 'dependencies': ['a']}, + ], + ) + grant = await call( + app, + 'sharing.grant', + { + 'resource': made.data['id'], + 'grantee': other, + 'expires_at': wire(NOW + timedelta(days=1)), + }, + key=key, + subject=owner, + ) + assert grant.status == 'ok', wire(grant.error) + claimed = await change(app, 'claim', made, key, owner, worker='owner') + note = await call( + app, + 'identity.note_put', + {'name': 'private-result', 'body': 'Owner only'}, + key=key, + subject=owner, + ) + assert note.status == 'ok', wire(note.error) + completed = await change( + app, + 'complete', + claimed, + key, + owner, + task_id='a', + worker='owner', + attempt=claimed.data['attempt'], + result=wire(note.resources[0]), + ) + independent = await change(app, 'claim', completed, other_key, other, worker='other') + assert independent.data['task_id'] == 'independent' + finished = await change( + app, + 'complete', + independent, + other_key, + other, + task_id='independent', + worker='other', + attempt=independent.data['attempt'], + error='No result', + ) + blocked = await call( + app, + 'communication.task_claim', + {'id': made.data['id'], 'worker': 'other'}, + key=other_key, + subject=other, + expected=((made.data['id'], finished.data['generation']),), + ) + assert blocked.error.code == 'task_not_ready' + mine = await call( + app, 'communication.task_status', {'id': made.data['id']}, key=key, subject=owner + ) + theirs = await call( + app, 'communication.task_status', {'id': made.data['id']}, key=other_key, subject=other + ) + assert mine.status == theirs.status == 'ok' + assert mine.data['plan']['graph']['tasks'][0]['result'] == wire(note.resources[0]) + assert note.resources[0].id not in str(wire(theirs.data)) From fa6e69afbd99d20c69bf1803e0c0514d8c20c2fa Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Tue, 13 Oct 2026 07:58:19 +0800 Subject: [PATCH 3/4] fix: scope subscription arguments to listener commands --- src/msg/client_agent_cli.py | 9 +++++---- tests/test_client_subscription.py | 15 +++++++++++++++ 2 files changed, 20 insertions(+), 4 deletions(-) diff --git a/src/msg/client_agent_cli.py b/src/msg/client_agent_cli.py index 1429e6fb..6b2a8191 100644 --- a/src/msg/client_agent_cli.py +++ b/src/msg/client_agent_cli.py @@ -89,10 +89,11 @@ def add_commands(commands): def local_command(args): - require( - not getattr(args, 'resource', None) or getattr(args, 'subscribe', False), - 'subscription_required', - ) + if args.command == 'listen': + require( + not getattr(args, 'resource', None) or getattr(args, 'subscribe', False), + 'subscription_required', + ) if getattr(args, 'subscribe', False): return False return (args.command == 'agent' or args.command == 'listen' and args.agent) and not args.remote diff --git a/tests/test_client_subscription.py b/tests/test_client_subscription.py index f0215499..f780fd86 100644 --- a/tests/test_client_subscription.py +++ b/tests/test_client_subscription.py @@ -1,6 +1,10 @@ from types import SimpleNamespace +import pytest + +from msg.client_agent_cli import local_command from msg.client_subscription import Subscription +from msg.core.errors import Failure class Client: @@ -51,3 +55,14 @@ async def test_pending_revocation_drops_cached_event(): }) is None ) + + +def test_regular_read_resource_does_not_request_subscription(): + args = SimpleNamespace(command='read', resource='/main', agent=None, remote=False) + assert local_command(args) is False + + +def test_listen_resource_cannot_silently_route_to_local(): + args = SimpleNamespace(command='listen', resource='/main', agent='worker', remote=False) + with pytest.raises(Failure, match='subscription_required'): + local_command(args) From 2bd54ab3143e0fe30b5a45a09f3475a0464ad536 Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Tue, 13 Oct 2026 08:21:34 +0800 Subject: [PATCH 4/4] style: apply exact Ruff formatting for merged subscription CLI --- src/msg/client_agent_cli.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/src/msg/client_agent_cli.py b/src/msg/client_agent_cli.py index 0bdd5fef..96079914 100644 --- a/src/msg/client_agent_cli.py +++ b/src/msg/client_agent_cli.py @@ -176,7 +176,9 @@ def cursor_path(state, args, source): return private_directory(state.paths.state / 'agent-listeners') / (key + '.json') -async def stream(state, args, fetch, source, *, wait_for_change=None, resolve_legacy=None, server_wait=False): +async def stream( + state, args, fetch, source, *, wait_for_change=None, resolve_legacy=None, server_wait=False +): from msg.client_listener import listen remote = not source.startswith('local:') @@ -285,8 +287,7 @@ async def run_remote(client, args): require(not args.offline, 'offline_remote_conflict') if args.command == 'listen': require( - getattr(args, 'wait_mode', 'auto') != 'events' - or getattr(args, 'subscribe', False), + getattr(args, 'wait_mode', 'auto') != 'events' or getattr(args, 'subscribe', False), 'listener_events_unavailable', ) if args.user: