feat: authorized event subscriptions and deterministic task coordination

.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/

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.

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.

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,15 +68,15 @@ 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 = (
                 ValkeyOutboxSignal(self.settings.server.valkey_url)
                 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,

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)