feat(subscriber): named event streams with a Morpho governance stream

.env.example

@@ -20,7 +20,8 @@ PROVIDER_URL_KATANA_1=https://rpc.katanarpc.com
 # Yearn large TVL env vars
 ENVIO_GRAPHQL_URL="https://indexer.hyperindex.xyz/3fec0a4/v1/graphql"
 
-# Timelock event subscriber pilot (protocols/timelock/event_subscriber.py, local output only)
+# Event subscriber pilot (protocols/timelock/event_subscriber.py, local output only)
+# EVENT_SUBSCRIBER_STREAMS=timelock               # comma-separated: timelock, morpho (needs MorphoGovernanceEvent in Envio)
 # ENVIO_GRAPHQL_WS_URL=wss://...                  # default: ENVIO_GRAPHQL_URL with https → wss
 # TIMELOCK_SUBSCRIBER_TRANSPORT=both              # both | ws | http
 # TIMELOCK_SUBSCRIBER_NOTIFY=none                 # errors: post pilot notices to the errors chat

deploy/install.sh

@@ -243,8 +243,8 @@ cat <<NEXT
        sudo systemctl enable --now monitoring-api
        systemctl status monitoring-api
 
-     Optional timelock event subscriber pilot (no alerts; TIMELOCK_SUBSCRIBER_NOTIFY=errors
-     posts diagnostics to the errors chat):
+     Optional event subscriber pilot (no alerts; EVENT_SUBSCRIBER_STREAMS picks timelock and/or
+     morpho; TIMELOCK_SUBSCRIBER_NOTIFY=errors posts diagnostics to the errors chat):
        sudo systemctl enable --now monitoring-timelock-subscriber
        systemctl status monitoring-timelock-subscriber
 

deploy/runbook.md

@@ -235,8 +235,9 @@ directly.
 `monitoring-timelock-subscriber` is the continuous timelock event worker from
 [#438](https://github.com/yearn/monitoring/issues/438). It runs alongside the
 hourly `timelock-alerts` task, which still sends every timelock alert. The pilot
-sends no alerts: it records what it observes in `/srv/cache/timelock_subscriber.db`
-and journald. With `TIMELOCK_SUBSCRIBER_NOTIFY=errors` it also posts diagnostic
+sends no alerts: it records what it observes in one SQLite inbox per stream
+(`/srv/cache/timelock_subscriber.db`, and `/srv/cache/morpho_subscriber.db` once
+the `morpho` stream is enabled) and journald. With `TIMELOCK_SUBSCRIBER_NOTIFY=errors` it also posts diagnostic
 notices (new events, WebSocket outages, stalls, failures, a daily summary) to the
 internal errors chat; see the timelock README for the list. Stopping it has no effect on alerting. Design notes are in
 [`protocols/timelock/README.md`](../protocols/timelock/README.md#continuous-subscriber-pilot).
@@ -248,7 +249,8 @@ journalctl -u monitoring-timelock-subscriber -f
 ```
 
 It is a long-running process, so the 10-minute code sync does not reload it.
-Restart it after a change to `protocols/timelock/` or `utils/graphql_ws.py`:
+Restart it after a change to `protocols/timelock/`, `utils/subscriber/`,
+`protocols/morpho/governance_stream.py`, or `utils/graphql_ws.py`:
 `sudo systemctl restart monitoring-timelock-subscriber`.
 
 Check health and read the coverage/latency report:
@@ -260,9 +262,33 @@ CACHE_DIR=/srv/cache uv run python protocols/timelock/event_subscriber.py report
 CACHE_DIR=/srv/cache uv run python protocols/timelock/event_subscriber.py report --hours 168 --json
 ```
 
-`health` fails when the heartbeat is older than three intervals, the last
-reconcile failed, the WebSocket is down, or a chain's indexed head has not
-advanced for `TIMELOCK_SUBSCRIBER_STALL_SECONDS` (default 900).
+`health` and `report` cover every stream in `EVENT_SUBSCRIBER_STREAMS` (or
+`--streams timelock,morpho` before the subcommand). `health` fails when any
+stream's heartbeat is older than three intervals, its last reconcile failed, its
+WebSocket is down, or one of its chains' indexed head has not advanced for
+`TIMELOCK_SUBSCRIBER_STALL_SECONDS` (default 900).
+
+### Enabling the Morpho governance stream
+
+The `morpho` stream ([#447](https://github.com/yearn/monitoring/issues/447)) observes
+`MorphoGovernanceEvent` in the same process. It sends no alerts. Enable it only after
+the Envio deployment serves the entity and has backfilled all five chains:
+
+```sh
+cd /srv/monitoring
+# 1. The entity exists and every chain is indexed (Mainnet, Base, Katana, HyperEVM 999, Robinhood 4663):
+CACHE_DIR=/srv/cache uv run python protocols/timelock/event_subscriber.py --streams morpho reconcile-once
+# 2. Add the stream and restart:
+echo 'EVENT_SUBSCRIBER_STREAMS=timelock,morpho' | sudo tee -a /etc/monitoring/.env
+sudo systemctl restart monitoring-timelock-subscriber
+CACHE_DIR=/srv/cache uv run python protocols/timelock/event_subscriber.py health
+```
+
+`reconcile-once` exits 1 while the entity is missing (`field 'MorphoGovernanceEvent'
+not found`) or a chain has no `chain_metadata` row. A first run only sets checkpoints
+at each chain's indexed head; rerun it to confirm the range queries succeed. To
+disable the stream, drop it from `EVENT_SUBSCRIBER_STREAMS` and restart; its inbox is
+kept.
 
 Supervision: the unit is `Type=notify` with `WatchdogSec=120`. The heartbeat pings
 the watchdog only while HTTP reconciliation keeps completing, so a wedged worker
@@ -290,15 +316,17 @@ Set in `/etc/monitoring/.env`, then restart the unit:
 
 | Variable | Default | Meaning |
 |---|---|---|
-| `TIMELOCK_SUBSCRIBER_TRANSPORT` | `both` | `both`, `ws`, or `http` |
+| `EVENT_SUBSCRIBER_STREAMS` | `timelock` | Streams to follow: `timelock`, `morpho`, comma-separated |
+| `TIMELOCK_SUBSCRIBER_TRANSPORT` | `both` | `both`, `ws`, or `http` (every stream) |
 | `TIMELOCK_SUBSCRIBER_NOTIFY` | `none` | `errors` posts pilot notices to the internal errors chat (needs `TELEGRAM_TOPIC_ID_ERRORS` or `TELEGRAM_CHAT_ID_ERRORS`) |
 | `TIMELOCK_SUBSCRIBER_RECONCILE_SECONDS` | `60` | HTTP reconcile interval |
 | `TIMELOCK_SUBSCRIBER_STALL_SECONDS` | `900` | Indexed head age that counts as stalled |
 | `TIMELOCK_SUBSCRIBER_START_LOOKBACK_SECONDS` | `0` | History to backfill on a chain's first run |
 | `ENVIO_GRAPHQL_WS_URL` | derived | WebSocket URL; defaults to `ENVIO_GRAPHQL_URL` with `wss://` |
 
 To start the pilot over, stop the unit and delete
-`/srv/cache/timelock_subscriber.db*`. Checkpoints start again at the indexed head.
+`/srv/cache/timelock_subscriber.db*` (or `/srv/cache/morpho_subscriber.db*` for
+the Morpho stream alone). Checkpoints start again at the indexed head.
 
 ---
 

deploy/systemd/monitoring-timelock-subscriber.service

@@ -4,12 +4,13 @@
 #   sudo systemctl enable --now monitoring-timelock-subscriber
 #   sudo systemctl status monitoring-timelock-subscriber
 #
-# Continuous timelock event subscriber (observation pilot, issue #438). It runs
-# alongside the hourly timelock-alerts task and never replaces it: it records
-# what it sees in __CACHE_DIR__/timelock_subscriber.db and journald, and sends no
-# alerts. TIMELOCK_SUBSCRIBER_NOTIFY=errors in the env file adds diagnostic
-# notices to the internal errors chat. See protocols/timelock/README.md and
-# deploy/runbook.md.
+# Continuous event subscriber (observation pilot, issues #438 and #447). It runs
+# alongside the scheduled monitors and never replaces them: for each stream in
+# EVENT_SUBSCRIBER_STREAMS (default timelock; add morpho for Morpho governance)
+# it records what it sees in __CACHE_DIR__/<stream>_subscriber.db and journald,
+# and sends no alerts. TIMELOCK_SUBSCRIBER_NOTIFY=errors in the env file adds
+# diagnostic notices to the internal errors chat. See
+# protocols/timelock/README.md and deploy/runbook.md.
 #
 # Type=notify + WatchdogSec: the worker sends READY=1 once its loops start and
 # WATCHDOG=1 from its heartbeat only while HTTP reconciliation keeps completing,
@@ -43,7 +44,7 @@ ExecStartPre=+/bin/bash -c '\
     exit 1; \
   fi'
 
-# ENVIO_GRAPHQL_URL comes from here; TIMELOCK_SUBSCRIBER_* override the defaults.
+# ENVIO_GRAPHQL_URL comes from here; EVENT_SUBSCRIBER_STREAMS and TIMELOCK_SUBSCRIBER_* override the defaults.
 EnvironmentFile=__ETC_DIR__/.env
 ExecStart=__REPO_DIR__/.venv/bin/python protocols/timelock/event_subscriber.py run
 

protocols/morpho/README.md

@@ -199,6 +199,14 @@ Both scripts share `MORPHO_FILENAME` (default `cache-id.txt`). New key types add
 | `v2_role` | `owner` / `curator` | lowercase address | Last-known instant-role address |
 | `v2_set` | `sentinels` / `allocators` / `adapters` | comma-joined lowercase addresses | Last-known role set |
 
+### Continuous governance event stream (pilot)
+
+[`governance_stream.py`](./governance_stream.py) defines the `morpho` stream of the continuous event subscriber ([#447](https://github.com/yearn/monitoring/issues/447), part of [#437](https://github.com/yearn/monitoring/issues/437)). It follows `MorphoGovernanceEvent`, which [yearn-envio#71](https://github.com/yearn/yearn-envio/pull/71) indexes for every V1 and V2 vault in [`config.py`](./config.py): V1 `SubmitCap`, `SubmitMarketRemoval`, `SubmitTimelock`; V2 `Submit`, `Accept`, `SetOwner`, `SetCurator`, `AddAdapter`, `RemoveAdapter`. Adapter-internal timelocks (`burnShares`) are not indexed.
+
+It runs in the timelock subscriber's process with `EVENT_SUBSCRIBER_STREAMS=timelock,morpho` and uses the same WebSocket + HTTP reconciliation, per-chain checkpoints, reorg tracking, health, and report (see the [timelock README](../timelock/README.md#continuous-subscriber-pilot)); its inbox is `$CACHE_DIR/morpho_subscriber.db`. It is observation only: **it sends no alerts**, and the daily `governance.py` / `governance_v2.py` jobs remain the alerting path. With notices enabled, each new event is posted to the internal errors chat as `[morpho-subscriber]` with the vault, its generation, the sender, and the ABI-named arguments. Lifecycle rules and grouped delivery are [#448](https://github.com/yearn/monitoring/issues/448); comparing with the daily snapshots is [#449](https://github.com/yearn/monitoring/issues/449).
+
+Enable it only once the Envio deployment serves `MorphoGovernanceEvent` and has backfilled all five chains; until then its reconcile cycles and WebSocket fail with `field 'MorphoGovernanceEvent' not found` (without affecting the timelock stream). Its first reconcile starts at each chain's indexed head, so history is never announced.
+
 ### Limitations
 
 `MorphoMarketV1AdapterV2` has its own internal timelock system (`setSkimRecipient`, `burnShares`, `increaseTimelock`, etc.). The Morpho GraphQL API does **not** expose adapter-internal pending operations, and replaying `Submit` events on each adapter would reintroduce the RPC cost we're explicitly avoiding. Phase-2 candidate.

protocols/morpho/governance_stream.py

@@ -0,0 +1,74 @@
+"""The ``MorphoGovernanceEvent`` stream followed by the continuous event subscriber.
+
+yearn-envio indexes governance events emitted by every vault in ``config.py``
+(V1 ``SubmitCap`` / ``SubmitMarketRemoval`` / ``SubmitTimelock``; V2 ``Submit`` /
+``Accept`` / ``SetOwner`` / ``SetCurator`` / ``AddAdapter`` / ``RemoveAdapter``) as
+one entity with ABI-named arguments in ``argsJson``. This stream observes them
+with the same WebSocket + HTTP reconciliation as the timelock pilot (#447). It
+sends no alerts: lifecycle rules and delivery are #448, and the daily
+``governance.py`` / ``governance_v2.py`` jobs stay the alerting path.
+"""
+
+import json
+from typing import Any
+
+from protocols.morpho._shared import PROTOCOL
+from protocols.morpho.config import VAULTS_V1_BY_CHAIN, VAULTS_V2_BY_CHAIN, iter_vaults
+from utils.subscriber.stream import EventStream, StreamLabel, StreamScope
+
+MORPHO_GOVERNANCE_FIELDS = """
+        id
+        vaultAddress
+        vaultVersion
+        eventName
+        chainId
+        blockNumber
+        blockHash
+        blockTimestamp
+        logIndex
+        transactionHash
+        transactionIndex
+        transactionFrom
+        argsJson
+"""
+
+# Longest ``argsJson`` shown in a notice; V2 ``Submit`` carries full calldata.
+_MAX_ARGS_CHARS = 600
+
+
+def describe_governance_event(payload: dict[str, Any]) -> list[str]:
+    """Return notice lines for one event: vault generation, sender, and its ABI-named arguments."""
+    lines = [f"Vault V{payload.get('vaultVersion')}, sent by {payload.get('transactionFrom') or 'n/a'}"]
+    raw = payload.get("argsJson")
+    if raw:
+        try:
+            args = json.dumps(json.loads(raw), separators=(",", ":"), sort_keys=True)
+        except TypeError, ValueError:
+            args = str(raw)
+        lines.append(f"Args: {args[:_MAX_ARGS_CHARS]}" + ("…" if len(args) > _MAX_ARGS_CHARS else ""))
+    return lines
+
+
+MORPHO_GOVERNANCE_STREAM = EventStream(
+    name="morpho",
+    noun="Morpho governance",
+    entity="MorphoGovernanceEvent",
+    address_field="vaultAddress",
+    address_label="Vault",
+    fields=MORPHO_GOVERNANCE_FIELDS,
+    db_filename="morpho_subscriber.db",
+    table_prefix="morpho_sub",
+    describe=describe_governance_event,
+)
+
+
+def morpho_governance_scope() -> StreamScope:
+    """Return every configured V1 and V2 vault, keyed by chain, with its notice label."""
+    addresses: dict[int, set[str]] = {}
+    labels: dict[tuple[str, int], StreamLabel] = {}
+    for version, vaults_by_chain in (("V1", VAULTS_V1_BY_CHAIN), ("V2", VAULTS_V2_BY_CHAIN)):
+        for chain, vault in iter_vaults(vaults_by_chain):
+            address = vault.address.lower()
+            addresses.setdefault(chain.chain_id, set()).add(address)
+            labels[(address, chain.chain_id)] = StreamLabel(f"{vault.name} {version}", PROTOCOL)
+    return StreamScope(addresses=addresses, labels=labels)

protocols/timelock/README.md

@@ -15,12 +15,21 @@ The script runs hourly via the [monitoring runner](../../automation/jobs.yaml).
 
 `event_subscriber.py` is a continuously running worker that observes the same `TimelockEvent` rows as `timelock_alerts.py` with lower latency ([#438](https://github.com/yearn/monitoring/issues/438), the first step of [#437](https://github.com/yearn/monitoring/issues/437)). It is an observation pilot: **it sends no alerts and changes nothing about current delivery**. The hourly task remains the only sender to protocol groups. Optionally it posts diagnostic notices to the internal errors chat (see [Pilot notices](#pilot-notices)). It runs as the `monitoring-timelock-subscriber` systemd unit; operations are in the [runbook](../../deploy/runbook.md#timelock-event-subscriber-pilot).
 
-It runs two transports side by side and records which one saw each event first:
+The worker follows **named event streams** ([#447](https://github.com/yearn/monitoring/issues/447)). The shared code lives in [`utils/subscriber/`](../../utils/subscriber/); each stream is an `EventStream` that names its Envio entity, the field holding the monitored contract, its field selection, and its own SQLite file and table prefix:
+
+| Stream | Entity | Defined in | Inbox |
+|---|---|---|---|
+| `timelock` (default) | `TimelockEvent` | [`event_stream.py`](./event_stream.py) | `timelock_subscriber.db`, tables `timelock_sub_*` |
+| `morpho` | `MorphoGovernanceEvent` | [`protocols/morpho/governance_stream.py`](../morpho/governance_stream.py) | `morpho_subscriber.db`, tables `morpho_sub_*` |
+
+`EVENT_SUBSCRIBER_STREAMS` (or `--streams`, before the subcommand) picks the streams, e.g. `timelock,morpho`; the default is `timelock` alone, so an existing deployment keeps its behavior. All selected streams run in one process (`run_workers`): one HTTP poster with a persistent session per thread, one `chain_metadata` read per cycle shared by every stream (`SharedChainHeads`), one WebSocket connection and one reconcile loop per stream, and a systemd watchdog ping only while every stream is alive. A stream's failure (for example an entity the Envio deployment does not serve yet) shows in its own log, checkpoints, and health, and never blocks another stream. The `run`, `reconcile-once`, `report`, and `health` subcommands cover every selected stream.
+
+Each stream runs two transports side by side and records which one saw each event first:
 
 - **WebSocket**: a `graphql-transport-ws` live query (`utils/graphql_ws.py`) over the newest 50 monitored events, filtered by monitored chains and addresses. Hasura resends the whole snapshot on every change, so most rows are repeats; the inbox deduplicates them. On disconnect the worker reconnects with capped exponential backoff (1s to 60s).
 - **HTTP reconciliation**, every 60s and also while the WebSocket is connected: for each chain, read `chain_metadata.latest_processed_block`, then fetch `(checkpoint - overlap, head]` in chunks of at most `MAX_BLOCK_SPAN` blocks, paginated in pages of 200 ordered by `(blockNumber, logIndex, id)`. This is the completeness guarantee.
 
-State lives in `$CACHE_DIR/timelock_subscriber.db` (SQLite, WAL), separate from the hourly monitor's state. The schema is versioned (`PRAGMA user_version`); a database from an earlier revision is migrated in place on open, keeping observations, changes, and checkpoints. A row stored before the event selection grew (e.g. before `blockHash`) is upgraded silently on its next observation, not recorded as a change.
+State lives in the stream's file under `$CACHE_DIR` (SQLite, WAL), separate from the scheduled monitors' state. The tables below are named for the timelock stream; other streams use their own prefix. The schema is versioned (`PRAGMA user_version`); a database from an earlier revision is migrated in place on open, keeping observations, changes, and checkpoints (v1 → v2 renamed `timelock_address` to `contract_address`). A row stored before the event selection grew (e.g. before `blockHash`) is upgraded silently on its next observation, not recorded as a change.
 
 - **Per-chain checkpoints** (`timelock_sub_checkpoints`): the highest block fully fetched over HTTP, never moved backwards. They are independent per chain and do not use `TIMELOCK_LAST_TS`, so a failing or lagging chain never holds back the others. A chain's first run starts at the indexed head unless `TIMELOCK_SUBSCRIBER_START_LOOKBACK_SECONDS` is set.
 - **Event provenance** (`timelock_sub_events`): one row per Envio event id (`<chainId>_<blockNumber>_<logIndex>`) with its block hash, a content fingerprint, the first transport and time it was seen, first-seen times and observation counts per transport.
@@ -36,7 +45,7 @@ Edge cases:
 
 Supervision: a heartbeat every 30s stores worker status, flags any chain whose indexed head has not advanced for 15 minutes, and pings the systemd watchdog only while reconciliation keeps completing. `event_subscriber.py health` exits 1 on a stale heartbeat, failed reconcile, WebSocket outage, or stalled chain.
 
-`event_subscriber.py report --hours N` prints the coverage/latency report used to choose the production transport:
+`event_subscriber.py report --hours N` prints each selected stream's coverage/latency report, used to choose the production transport:
 
 - WebSocket coverage of HTTP-confirmed events, plus WebSocket-only and HTTP-only counts.
 - Delay from block timestamp to first sighting per transport (p50/p95/max), and how far the WebSocket leads HTTP. Envio exposes no per-row index time, so the faster transport's delay is an upper bound on indexing delay.
@@ -46,7 +55,7 @@ Supervision: a heartbeat every 30s stores worker status, flags any chain whose i
 
 ### Pilot notices
 
-With `TIMELOCK_SUBSCRIBER_NOTIFY=errors` (or `run --notify errors`), the worker posts plain-text diagnostics to the internal errors chat (`TELEGRAM_TOPIC_ID_ERRORS` or `TELEGRAM_CHAT_ID_ERRORS`), labelled `[timelock-subscriber]` and stored in alert history under `yearn-internal`. They are for comparing the pilot with the hourly alerts, not alerts themselves. Notices stay off unless an errors chat is configured: `send_error_message` would otherwise fall back to a protocol group.
+With `TIMELOCK_SUBSCRIBER_NOTIFY=errors` (or `run --notify errors`), the worker posts plain-text diagnostics to the internal errors chat (`TELEGRAM_TOPIC_ID_ERRORS` or `TELEGRAM_CHAT_ID_ERRORS`), labelled `[timelock-subscriber]` and stored in alert history under `yearn-internal`. Each stream posts under its own label (`[timelock-subscriber]`, `[morpho-subscriber]`), and its texts name the stream ("New Morpho governance event"). They are for comparing the pilot with the scheduled alerts, not alerts themselves. Notices stay off unless an errors chat is configured: `send_error_message` would otherwise fall back to a protocol group.
 
 | Notice | When |
 |---|---|