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

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"}]

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.

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:

<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

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

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

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

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.