johnandersen777.bsky.social/asyncai

Switch repository

Loading…

# 0005 — MCP watches: push-driven change notifications

Status: accepted

Date: 2026-11-03

## Context

Users want to tell the agent "watch X and tell me when it changes". X may be
Kubernetes objects, a file, a queue, or any other thing an MCP server
exposes.

Constraints:

- **No polling, anywhere.** Not in asyncai, not in the MCP server, not in
  model-written code. Every hop is pushed.
- **Generic.** It must work with any MCP server that follows a documented
  pattern, not only the kcp servers.
- **Code mode.** The model sets up watches from `run_code` (ADR 0004) and
  may filter events with Python.

MCP already has the push primitive. A server marks content as a *resource*,
the client subscribes to the resource URI, and the server sends
`notifications/resources/updated` when it changes. MCP 2026-07-28 carries
subscriptions on a long-lived `subscriptions/listen` request; rmcp 3.8
implements it.

## Decision

```mermaid
sequenceDiagram
    participant S as change source (k8s watch, inotify, ...)
    participant M as MCP server
    participant A as asyncai WatchManager
    participant P as Python filter (Monty)
    participant G as agent turn

    A->>M: subscriptions/listen {resourceSubscriptions:[uri]}
    A->>M: resources/read uri (baseline)
    S-->>M: change (pushed)
    M-->>A: notifications/resources/updated {uri}
    A->>M: resources/read uri
    A->>A: extract events (event-log seq or state diff)
    A->>P: on_event(event) per event
    P-->>A: notice text or None
    A->>A: dedupe per object, debounce 3 s (max 15 s)
    A->>G: <watch-event> turn
```

### Server contract: what an MCP server must expose

A server is watchable when it does all of the following.

1. **Capability.** `capabilities.resources.subscribe = true`.
2. **Discoverable URIs.** Each watchable thing is listed by `resources/list`
   (fixed URIs) or `resources/templates/list` (RFC 6570 templates). Each
   gets a description that ends with a code-mode example, e.g.
   `Example: await watch_start(name="cm", uri="kcp-events://watch/root:app/configmaps?namespace=*")`.
3. **Subscriptions.** MCP **2026-07-28** (`server/discover` lifecycle) with
   `subscriptions/listen` accepting `resourceSubscriptions`. Legacy
   `resources/subscribe` is not supported: it only exists on pre-2026
   protocol versions, and asyncai does not use legacy paths for watches.
4. **Push.** The server sends `notifications/resources/updated {uri}` after
   each change. Notifications carry no payload, may be coalesced, and must
   be idempotent.
5. **Native change source.** The server learns about changes by push:
   Kubernetes watch, inotify, database `LISTEN/NOTIFY`, webhooks, or its
   own write path. A server that polls internally does not meet this
   contract.
6. **Read shape.** `resources/read uri` returns JSON text in one of two
   shapes.

| Shape | `mimeType` | Body | asyncai behavior |
|---|---|---|---|
| **state** | anything else (`application/json` recommended) | current state | diffs against the previous read |
| **event-log** | `application/vnd.asyncai.events+json` | `{"events": [{"seq": int, "type": str, ...}]}` | delivers events with `seq` greater than the last seen |

Event-log events SHOULD carry `object` and, for `MODIFIED`, `old` (the
previous object). asyncai then computes `changed` and drops no-op updates.

- **State resources** suit small state, such as config, a document, or a
  set of names. asyncai diffs:
  - JSON lists by item key, the first present of `metadata.uid`, `uid`,
    `id`, `metadata.name`, `name`; otherwise the whole item;
  - JSON objects by top-level key;
  - anything else as whole text.
- **Event-log resources** suit large or fast-changing collections. The
  server keeps a bounded ring of recent events with a monotonically
  increasing `seq`. Reads are stateless, so many readers can share one log.
  When the oldest `seq` in a read is beyond the last seen + 1, asyncai
  emits a `GAP` event, so filters know events were missed.
- **Failures.** An event-log server reports source failures in-band as
  `{"seq": n, "type": "ERROR", "message": "..."}`. Any server may end the
  subscription; asyncai then resubscribes (A3).
- **Writes stay tools.** Resources are read-only.
- **No caching.** Do not set `ttlMs` on watched resources. MCP clients
  may serve a cached read, which would hide the change just announced.

### Client (asyncai)

1. **Routing.** A URI goes to the server that lists it exactly, otherwise
   the server whose template matches, otherwise the first server that
   advertises `resources.subscribe`.
2. **Subscribe.** asyncai connects with `server/discover` (MCP 2026-07-28)
   and opens one `subscriptions/listen` per watch. A server that is not on
   2026-07-28 can still serve tools: asyncai falls back to `initialize`
   for the connection. Watching it fails with an error naming the
   required version.
3. **Resubscribe.** When a subscription ends, asyncai resubscribes with
   exponential backoff (1 s, 2 s, 4 s, 8 s, 16 s). After 5 failures the
   watch fails and the agent is told. This is reconnect-on-failure, not
   polling: nothing runs while a subscription is healthy.

### Code-mode built-ins

These are present when at least one MCP server is connected.

| Function | Returns |
|---|---|
| `list_resources()` | `[{"uri", "template", "name", "description", "example", "mime_type", "server", "subscribable"}]`; `uri` is the template when `template` is true; `example` is the description's `Example:` line |
| `read_resource(uri=...)` | parsed JSON, else `str` |
| `watch_start(name=..., uri=..., on_event=None)` | `{"name", "uri", "shape"}` |
| `watch_stop(name=...)` | `bool` |
| `watch_list()` | `[{"name", "uri", "shape", "events", "notices"}]` |

```python
await watch_start(
    name="cm",
    uri="kcp-events://watch/root:app/configmaps?namespace=*",
    on_event='''
m = event["object"]["metadata"]
f'{event["type"].lower()} {m["namespace"]}/{m["name"]}' if event["type"] in ("ADDED", "DELETED") else None
''',
)
```

**Event dict.** `on_event` sees `event`, a dict with:

| Key | Value |
|---|---|
| `watch`, `uri` | watch name and URI |
| `type` | event-log `type`; state diffs: `ADDED` / `DELETED` / `MODIFIED` / `CHANGED`; `GAP` |
| `old` | previous object for `MODIFIED` (event-log: from the server; state: from the diff); `None` otherwise |
| `changed` | `MODIFIED` with `old`: sorted dotted paths that differ, depth ≤ 3, ignoring `metadata.resourceVersion` / `generation` / `managedFields` |
| other keys | event-log: the server's event fields. State: `key`, plus `new` (or `item`) |

A `MODIFIED` event whose `changed` is empty is dropped before the filter.

**Filter.**

- `on_event` is Python source. It runs once per event, in the watch's own
  REPL scope, so variables persist across events.
- That scope's catalog is read-only: MCP tools without mutating
  annotations, plus `read_resource` and `list_resources`. A filter cannot
  write and cannot start watches.
- If `on_event` leaves `None` (and prints nothing), the event is dropped.
  Otherwise the value becomes one notice line; a `str` is used as-is, other
  values as compact JSON.
- Without `on_event`, the default line is `TYPE namespace/name` for
  Kubernetes-like objects, `TYPE key` for state diffs, and compact JSON
  otherwise. `MODIFIED` adds up to three changes:
  `MODIFIED bid-1 (status.phase: Deciding→Published)`.
- **Dedupe.** A line equal to the last line for the same object (item key)
  is dropped. Identical lines in one notice collapse to `line ×N`.

**Validation.** `on_event` must be source text: a function object (it
crosses the sandbox as its `repr`) is rejected. `watch_start` dry-runs
`on_event` on a sample event. A `SyntaxError` or `NameError` fails the call
in the model's turn, and no subscription is made; missing keys on the
sample are fine. Live, the model passed `on_event=fmt`; before this check
every event would have failed.

**Fail fast.** The baseline read happens inside `watch_start`, so a server
that rejects the URI fails the call. Servers should make the first read of
a watched URI wait for their source to start (a signal, not a poll) and
return its error, as the kcp server does. The model can then call
`watch_start` first and learn from the error, instead of exploring.

**Coalescing.** A state resource diffs against the previous read. Changes
that land before one read become one diff, e.g. two quick writes produce
the net change. Event-log resources deliver every event, by `seq`.

**Debounce.** Trailing: each new line moves the flush to 3 s later, capped
at 15 s after the burst's first line. One notice holds at most 20 lines
(`+N more`). A controller's burst (`ADDED`, several `MODIFIED`) becomes one
agent turn.

**Persistence.** With `--watch-state PATH`, watch specs (`name`, `uri`,
`on_event`) are written on every start and stop (temp file + rename) and
restarted at startup. One `restore` notice lists them; changes made while
the agent was down may be missing (baselines are fresh). A spec that can
never start (bad `on_event`, over a limit) is dropped from the file. A
transient failure (server down, subscribe failed) keeps the spec saved and
listed by `watch_list()` with `restore_failed: true` until the agent starts
it again or stops it.

### Delivery to the agent: core integration

Notices enter the core the same way user input does. They do not form a
side channel.

```mermaid
flowchart TD
    W["WatchManager (tokio tasks)"] -->|WatchNotice| C["CLI loop: session MonitorEvent + route"]
    C -->|Notice{source,text}| N["Orchestrator notice stream (WorkSet: MainWork::Notice)"]
    I["stdin / atirc input"] --> In["Orchestrator input stream"]
    N --> R["WaitingEngine: AddMessage → RunThread"]
    In --> R
    R --> RT["AgentRuntime"]
    RT -->|idle| Run["new run"]
    RT -->|run active| Q["append message + queued_run"]
    Q -->|run completes| Run
```

- **Orchestrator.**
  - The reactor multiplexes a fifth `WorkSet` stream, `MainWork::Notice`
    (`main.system.notice_stream` in NDJSON output), beside input, waiting
    rules, agent events, and the runtime.
  - `OrchestratorHandle::notice_tx` accepts `Notice { source, text }`. Any
    background source can use it; watches are the first.
  - A notice goes through the same waiting rules as input:
    `AddMessage(role=user)` then `RunThread`.
- **Coalescing comes from the runtime.** A notice that arrives while a run
  is active is appended to the thread and sets `queued_run`. When the run
  completes, one follow-up run sees every appended message. N notices
  during a run produce one extra run, not N.
- **Reply routing.** The CLI tracks inputs that no run has claimed yet,
  with the reply channel for each. On `new_thread_run_created`, the new
  run claims all of them and replies go to the most recent claimed
  channel. Notices use the channel of the last addressed atirc message,
  otherwise the first `--channel`. This replaces per-input FIFO
  bookkeeping, which drifted whenever inputs coalesced.
- **Shutdown.** A loop is idle when no input is unclaimed and no run is
  active.
- **Filters run concurrently with turns.** Each watch has its own REPL
  scope (its own pool worker or engine thread), so filtering never blocks
  a model turn, and a slow turn never blocks event intake.

A notice becomes this message:

```text
<watch-event name="cm" uri="kcp-events://watch/root:app/configmaps?namespace=*">
added default/final-check
</watch-event>
This is a watch notification, not a user message. Tell the channel only if the change matters to what was asked; otherwise reply exactly NO_REPLY.
```

- **`NO_REPLY`.** A reply that is empty, or has a line that is only
  `NO_REPLY`, publishes nothing (`reply.suppressed`). Models rarely emit an
  empty reply; a token works. Prompts use it for chatter from other
  agents too.

- **atirc mode.** The reply goes to the channel of the last addressed
  message, otherwise the first `--channel`.
- **Interactive mode.** The output is printed as usual.
- **Session.** The notice is a `MonitorEvent` (queue-operation +
  attachment, ADR 0002) with task id `watch-<name>` and description
  `watch <name> <uri>`, never a user record. Its note is the instruction
  above, not Claude Code's PushNotification sentence.

### Limits

| Limit | Default |
|---|---|
| watches per agent | 5 |
| filter execution per event | 2 s |
| filter tool calls per event | 10 |
| events processed per update | 100 |
| lines per notice | 20 |
| debounce (quiet) / max wait | 3 s / 15 s |
| resubscribe attempts | 5 (1 s → 16 s backoff) |

### Logging (NDJSON)

Each event below is emitted once for the case it names:

| Event | Emitted when |
|---|---|
| `watch.started` | a watch is created |
| `watch.subscribed` | a `subscriptions/listen` subscription is established |
| `watch.event` | each event is processed; the log records whether the filter dropped or kept it |
| `watch.notice` | a notice is sent to the agent |
| `watch.resubscribing` | a subscription ended and asyncai is reconnecting |
| `watch.failed` | the watch gave up after its resubscribe attempts |
| `watch.stopped` | a watch is removed |
| `watch.restored` | `--watch-state` watches restarted at startup |
| `reply.suppressed` | a `NO_REPLY` reply was not published |

`watch.notice` is logged always. The rest are logged with `--verbose`.

## Reference servers

### kcp: event-log resources

```text
kcp-events://watch/{workspace}/{resource}{?namespace,labelSelector,fieldSelector,kcp}
```

- The server keeps one watcher per URI, shared by all subscribers and
  stopped when the last one leaves. Each watcher works like this:
  1. **Seed:** `kubectl get <resource> -o json` records known objects
     (`uid` → `resourceVersion`). The first read of the URI awaits this
     (up to 30 s); a failed seed fails the read with kubectl's error plus
     `did you mean: ...`.
  2. **Watch:** stream
     `kubectl get <resource> -w --output-watch-events -o json --request-timeout=0`.
     The output is parsed incrementally as concatenated JSON. `ADDED`
     events for already-known `uid` + `resourceVersion` pairs are
     suppressed. `metadata.managedFields` is stripped.
  3. **Ring:** 256 events with `seq`; `MODIFIED` carries `old` (previous
     known object). Each new event triggers
     `notifications/resources/updated`.
  4. **Reconnect:** when the stream ends (server timeout, token expiry,
     network), the watcher waits with exponential backoff, re-lists,
     emits synthetic `ADDED` / `MODIFIED` / `DELETED` events for anything
     that changed in the gap, then watches again.
- Credentials:
  - the read-only server re-mints its workspace token on every reconnect;
  - the read-write server uses the admin kubeconfig.
- The read-only ClusterRole already grants `watch`.
- `kcp-api://{workspace}/resources{?kcp}` is a plain readable resource
  (same JSON as `get_api_resources`), not watchable.

### asyncai-mcp-simple: state resource

- `memo://notes` is a JSON object.
- The `memo_set(key, value)` tool changes it and notifies listeners.

## Invariants

1. No component polls. Waits are subscriptions, stream reads, debounce
   timers after an event, or backoff after a failure.
2. A watch filter can neither mutate state nor start watches.
3. Watch notices reach the model as turns and the session as monitor
   events, never as user records.
4. A watch never outlives the agent process. Dropping the executor cancels
   every subscription. `--watch-state` restarts specs, not subscriptions.
5. Event-log delivery is at-most-once per `seq`; gaps are reported as
   `GAP`.

## Test plan

- **Unit:**
  - state diff for lists, objects, and text;
  - event-log `seq` and `GAP`;
  - URI template matching;
  - kcp URI parsing;
  - incremental JSON stream parsing with split chunks;
  - kcp dedupe and reconcile;
  - default notice text, `old` / `changed`, no-op `MODIFIED` dropped;
  - debounce batching.
- **Watch manager:** per-object dedupe, trailing debounce and `max_wait`,
  `--watch-state` round trip, `NO_REPLY` publishes nothing.
- **Fake `ResourceSource`:**
  - start → push → notice;
  - filter drop and transform;
  - the filter is denied mutating tools;
  - `watch_stop`;
  - the 5-watch limit;
  - `list_resources` / `read_resource`.
- **Simple server over stdio:** `memo_set` → state-diff notice through
  `subscriptions/listen`.
- **kcp server over stdio:** a fake `kubectl -w` tails a control file. The
  test appends an event → notification → `read_resource` returns it with
  `seq`; a duplicate `ADDED` is suppressed.

No test sleeps to wait for a condition. Tests await the notification or
notice channels with a timeout.

## Live verification

```bash
asyncai atirc -c '#alice' --verbose \
  --mcp-stdio asyncai-mcp-simple \
  --mcp-stdio socialweb-computer-kcp-readonly 2>&1 | jq -c 'select(.event|startswith("watch."))'
```

```bash
ATIRC_TOKEN=$(atirc token) atirc pub -c '#alice' \
  'alice: watch configmaps in root:app (all namespaces) and tell me when one appears or disappears'
```

```bash
K=/var/lib/socialweb-computer/swc-bidder/handles/<hash>/kcp/admin.kubeconfig
sudo -u swc-bidder kubectl --kubeconfig $K --server https://127.0.0.1:<port>/clusters/root:app \
  -n default create configmap watch-probe --from-literal=a=b
```

Expected: `watch.notice`, a new turn, and a channel post naming
`default/watch-probe`.

## Live results (2026-11-03, local 27B model, real kcp)

| Step | Result |
|---|---|
| "watch configmaps in root:app" | `watch_start` with a model-written `on_event`; `event-log` shape |
| external `kubectl create configmap watch-probe` | `watch.notice` ~2 s later (debounce), one turn, channel post |
| broken `on_event` (SyntaxError) | notice reported the failure; the model replaced the watch in the next turn (now caught at `watch_start`) |
| external `kubectl delete configmap watch-probe` | `DELETED default/watch-probe` → 1-round turn, no tool calls, post |
| generic `memo://notes` (simple server), two writes during one turn | one notice `ADDED owner`, `MODIFIED todo` → one follow-up turn |

## Live results after session review (2026-11-03)

| Step | Result |
|---|---|
| model passes `on_event=fmt` (function) | stored as a repr, accepted (bug) → now rejected at `watch_start`; restore of such specs fails, is reported, and drops them from `--watch-state` |
| restart with 3 good saved watches | `watch.restored` 3/3; restore notice → 1 round → `NO_REPLY` (`reply.suppressed`) |
| vague "can you set up monitors for bidder compute contract events?" with `--prompt atirc-socialweb-computer` | 3 rounds, 4 watches (bidder, bids, computevms, compute statesyncs via `labelSelector`) from the prompt's recipe; before: 9 rounds, no watch |
| `On Start` "set up monitors ..." with no user message, empty state | 4 rounds: checked the bidder exists, started 3 watches, one summary post |
| restart: restore notice + `On Start` | before: two turns, the same summary posted twice → restore notice folded into the `On Start` turn: one turn, 2 rounds, no duplicate watches |
| real RFP from a requester (ComputeContract → our bidder), 2 cycles | phase lines: Bid ADDED → Deciding → Published → Won; StateSync Delivering → Acked; ComputeVM Pending → Preparing → Booting → Running; release: Running → Releasing, DELETED. Posts: bid won (requester), VM running (cpu, memory, disk, guest IP), VM released |
| same, findings fixed | DELETED rendered `None -> Releasing` and was suppressed (recipe prints `gone`; prompt says post it); truncated arrays read the same on both sides (path only now); the model ran `NO_REPLY` as code (now ends the turn); an uncapped local model decoded 71 486 tokens (now `max_tokens` 8192, 600 s timeout) |
| "watch configmaps in root default; only create/delete" | `watch_start` in the first `run_code`, 3 rounds |
| create + 3 patches + delete | one `ADDED` notice (patches dropped by the filter), one `DELETED` notice; each 1 round, one-line post |
| follow-up question on data from the previous turn | 0 MCP calls: the model reused a variable from the scope note |
| `atproprecords` typo | `did you mean: atprotorecords, ...` → fixed in the next round; reply led with the correction, so `prompts/atirc.md` now says answer first, correction after → `- association` then the note |

## Consequences

- Any MCP server becomes watchable by exposing resources and pushing
  `updated` notifications. asyncai needs no server-specific code.
- State resources cost one read per change. Large collections should use
  event-log resources.
- A notice costs one model turn. Filters and debounce keep that rate down.
- Watches are in memory unless `--watch-state` is set; restored watches
  start from a fresh baseline.

## Alternatives considered

- **Polling with diffs.** Rejected: the user requirement is no polling,
  and it wastes requests between changes.
- **Streaming events in tool results** (MCP progress notifications on a
  long-running tool call). This ties a watch to an open tool call, and
  tools are request/response. Resources plus subscriptions are the MCP
  primitive for change feeds.
- **Notification payloads** (custom notifications carrying events). These
  are not standard. `updated` + `read` works with every MCP client.