diff --git a/.github/workflows/event-coordination.yml b/.github/workflows/event-coordination.yml new file mode 100644 index 00000000..f15a50ec --- /dev/null +++ b/.github/workflows/event-coordination.yml @@ -0,0 +1,76 @@ +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_client_listener_cli.py tests/test_client_listener_remote.py \ + tests/test_client_listener_reliability.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..27af6879 --- /dev/null +++ b/docs/EVENT_COORDINATION.md @@ -0,0 +1,145 @@ +# 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. + +Remote version-2 listener checkpoints save cursors and processed event IDs, +never private event bodies. After restart the source is read again under current +authority; revoked events cannot be emitted from a saved body. Old version-1 +mailbox and account-event caches are reduced to IDs before remote reads. The +reader re-fetches those IDs under current access, omitting items that are now +invisible. Old subscription checkpoints with bodies and no safe ID-based recovery +fail closed. A permanent authorization failure does not advance the checkpoint. +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. 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. + +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 af837afc..01e004d6 100644 --- a/docs/SUBAGENTS.md +++ b/docs/SUBAGENTS.md @@ -203,3 +203,9 @@ pending entries are reauthorized by ID; old account-event pending caches without a safe reread fail closed. Local labels still share one OS user's authority. Output is at-least-once delivery, not an execution ACK. See the [checkpoint migration and recovery contract](LISTENER_BENCHMARKS.md#restart-and-privacy). + +## 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 a01a1f1e..96079914 100644 --- a/src/msg/client_agent_cli.py +++ b/src/msg/client_agent_cli.py @@ -67,6 +67,12 @@ 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) @@ -98,6 +104,13 @@ def add_commands(commands): def local_command(args): + 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 @@ -163,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): +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:') @@ -192,6 +207,7 @@ async def stream(state, args, fetch, source, *, wait_for_change=None, resolve_le max_idle_interval=idle_limit, cursor_only=remote, resolve_legacy=resolve_legacy, + server_wait=server_wait, ) return 0 @@ -270,9 +286,30 @@ async def fetch(cursor, tail=False): async def run_remote(client, args): require(not args.offline, 'offline_remote_conflict') if args.command == 'listen': - require(getattr(args, 'wait_mode', 'auto') != 'events', 'listener_events_unavailable') + require( + getattr(args, 'wait_mode', 'auto') != 'events' or getattr(args, 'subscribe', False), + 'listener_events_unavailable', + ) 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 ''), + 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): @@ -294,7 +331,25 @@ async def fetch(cursor, tail=False): 'has_more': more, } - return await stream(client.state, args, fetch, 'account-events') + async def resolve_legacy(message_id): + current = await client.call('communication.event', {'id': str(message_id)}) + if current.status != 'ok' and current.error.code == 'not_found': + return None + event = wire(client.checked(current).data) + 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', + resolve_legacy=resolve_legacy, + ) from msg.client_subagents_remote import RemoteAgents store = RemoteAgents(client, username=args.username) diff --git a/src/msg/client_listener.py b/src/msg/client_listener.py index af096ee2..5b31b747 100644 --- a/src/msg/client_listener.py +++ b/src/msg/client_listener.py @@ -177,6 +177,8 @@ async def listen( max_idle_interval: float | None = None, cursor_only: bool = False, resolve_legacy=None, + revalidate=None, + server_wait: bool = False, ) -> int: """Drain a source, then wait for a native hint or a bounded polling delay. @@ -204,6 +206,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') if state['version'] == 2 and not cursor_only: @@ -268,6 +271,12 @@ def emit(item): while state['legacy_pending']: reference = state['legacy_pending'][0] item = await retry(lambda message_id=reference['id']: resolve_legacy(message_id)) + if item is None: + # A formerly authorized item is no longer visible. Do not emit a + # persisted body or discard unrelated pending references. + state['legacy_pending'].pop(0) + persist() + continue if not isinstance(item, dict) or _event_id(item) != reference['id']: raise Failure('invalid_listener_page') emit(item) @@ -301,6 +310,16 @@ def emit(item): processed = set(state.get('processed', ())) while state['pending']: item = state['pending'][0] + if restored and revalidate is not None: + # A legacy local checkpoint can contain a cached private body. + # Re-fetch with today's authority before sending it to stdout. + item = await retry(lambda: revalidate(state['pending'][0])) + if item is None: + state['pending'] = state['pending'][1:] + persist() + continue + if not isinstance(item, dict): + raise Failure('invalid_listener_page') if not cursor_only or _event_id(item) not in processed: emit(item) if cursor_only: @@ -312,6 +331,7 @@ def emit(item): persist() if max_events is not None and emitted >= max_events: return emitted + restored = False if cursor_only: state.update(cursor=page_end, processed=[]) persist() @@ -331,7 +351,7 @@ def emit(item): idle_delay = interval if wait_for_change is not None: await wait_for_change() - else: + elif not server_wait: await asyncio.sleep(idle_delay) if not activity: idle_delay = min(idle_delay * 2, idle_limit) diff --git a/src/msg/client_subscription.py b/src/msg/client_subscription.py new file mode 100644 index 00000000..eea6afc7 --- /dev/null +++ b/src/msg/client_subscription.py @@ -0,0 +1,68 @@ +"""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..ebc2831c --- /dev/null +++ b/src/msg/core/task_graph.py @@ -0,0 +1,124 @@ +"""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 = dict.fromkeys(('pending', 'running', 'succeeded', 'failed', 'blocked'), 0) + 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..af913393 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,27 @@ 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..91cc0a11 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,77 @@ 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 +204,41 @@ 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..1234220e --- /dev/null +++ b/src/msg/plugins/task_plans.py @@ -0,0 +1,264 @@ +"""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(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, + ) + 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, + ) + 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} + + @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..d550eab9 --- /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..f780fd86 --- /dev/null +++ b/tests/test_client_subscription.py @@ -0,0 +1,68 @@ +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: + 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 + ) + + +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) diff --git a/tests/test_event_subscription.py b/tests/test_event_subscription.py new file mode 100644 index 00000000..1de19b2a --- /dev/null +++ b/tests/test_event_subscription.py @@ -0,0 +1,198 @@ +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) + + +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 new file mode 100644 index 00000000..8aba3c5e --- /dev/null +++ b/tests/test_event_wait.py @@ -0,0 +1,125 @@ +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..b1f32cbe --- /dev/null +++ b/tests/test_listener_revalidation.py @@ -0,0 +1,85 @@ +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..3f345ca0 --- /dev/null +++ b/tests/test_task_graph.py @@ -0,0 +1,127 @@ +"""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 = { + '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..2439c52a --- /dev/null +++ b/tests/test_task_plans.py @@ -0,0 +1,277 @@ +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.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, + expected=((previous.data['id'], previous.data['generation']),), + ) + assert result.status == 'ok', wire(result.error) + 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' + 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', + claimed, + 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' + + +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))