feat(morpho): live governance decisions with shadow and production delivery

.env.example

@@ -22,6 +22,7 @@ ENVIO_GRAPHQL_URL="https://indexer.hyperindex.xyz/3fec0a4/v1/graphql"
 
 # Event subscriber pilot (protocols/timelock/event_subscriber.py, local output only)
 # EVENT_SUBSCRIBER_STREAMS=timelock               # comma-separated: timelock, morpho (needs MorphoGovernanceEvent in Envio)
+# MORPHO_LIVE_DELIVERY=shadow                     # shadow (errors chat) | observe | production (Morpho channel)
 # 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/runbook.md

@@ -290,6 +290,21 @@ 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.
 
+The stream's live consumer starts in `shadow` mode: for each new event it posts what
+the daily governance jobs would alert to the internal errors chat as
+`[morpho-live-shadow]` (`report` shows the summary), for comparison with the daily
+alerts. Without an errors chat it only records them (`observe`, also selectable with
+`MORPHO_LIVE_DELIVERY=observe`). The daily jobs keep alerting the Morpho group in both
+modes. Events from before the consumer's first run are recorded as history and never
+alerted.
+
+**Cutover to production** ([#450](https://github.com/yearn/monitoring/issues/450)): once a
+shadow period matched the daily Morpho alerts, set `MORPHO_LIVE_DELIVERY=production`
+and restart. Live alerts then go to the Morpho channel within seconds of the event.
+The daily jobs keep running: both paths take the same per-vault lock in
+`/srv/cache` and share the cache that records what was alerted, so each change is
+alerted once, by whichever path sees it first. To roll back, set `shadow` and restart.
+
 Supervision: the unit is `Type=notify` with `WatchdogSec=120`. The heartbeat pings
 the watchdog only while HTTP reconciliation keeps completing, so a wedged worker
 is killed and restarted. `Restart=always` also covers a crash, after which
@@ -317,6 +332,7 @@ Set in `/etc/monitoring/.env`, then restart the unit:
 | Variable | Default | Meaning |
 |---|---|---|
 | `EVENT_SUBSCRIBER_STREAMS` | `timelock` | Streams to follow: `timelock`, `morpho`, comma-separated |
+| `MORPHO_LIVE_DELIVERY` | `shadow` | Live Morpho governance decisions: `shadow` posts them to the internal errors chat (records only without one); `observe` records them only; `production` sends them to the Morpho channel |
 | `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 |

protocols/morpho/README.md

@@ -203,9 +203,34 @@ Both scripts share `MORPHO_FILENAME` (default `cache-id.txt`). New key types add
 
 [`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).
+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 sends no alerts to the Morpho group: the daily `governance.py` / `governance_v2.py` jobs remain the alerting path (see live decisions below). 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. 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.
+#### Live decisions ([#448](https://github.com/yearn/monitoring/issues/448))
+
+[`live_governance.py`](./live_governance.py) runs inside the `morpho` stream's worker and decides, for every newly indexed event, what the daily jobs would alert. The indexer has no V1 revoke/accept events and no V2 `Revoke`, so an event is only a trigger: [`live_rules.py`](./live_rules.py) reads the vault's current state on-chain, at a block at or after the event (an RPC node behind the event is retried), and runs the daily jobs' own checks on it:
+
+| Event | Check (same function as the daily job) |
+|---|---|
+| V1 `SubmitCap`, `SubmitMarketRemoval` | that market's pending cap and removal checks, including the potential-loss alert |
+| V1 `SubmitTimelock` | the timelock (and guardian) check |
+| V2 `Submit` | `executableAt(data)` must still be set, then the pending-operation diff (loss alerts, grouping of new submissions) |
+| V2 `SetOwner`, `SetCurator` | the owner/curator diff |
+| V2 `AddAdapter`, `RemoveAdapter` | the adapter set diff |
+| V2 `Accept` | none: executed/revoked notes stay with the daily job |
+
+So content, severity, per-vault grouping, and dedup match the daily jobs, and an operation revoked or executed before the read is no longer pending and is not alerted. The checks read the daily jobs' cache (an operation they already alerted is not alerted again) but write it only in `production` mode, so in the other modes the daily jobs keep alerting exactly as before. That cache is the daily profile's namespace (`MORPHO_FILENAME` from `automation/jobs.yaml`, applied by [`daily_governance_cache.py`](./daily_governance_cache.py) before `utils.cache` loads); the subscriber refuses to start the Morpho consumer if it resolved another one. An adapter whose `realAssets()` cannot be read fails the evaluation and is retried, never treated as empty.
+
+Decisions and alert requests are stored durably next to the inbox (`morpho_live_decisions`, `morpho_live_alerts` in `morpho_subscriber.db`, see [`live_store.py`](./live_store.py)), one decision per event revision: a restart resumes undecided events, and a reorg that rewrites an event re-evaluates it. Events from blocks before the consumer's first run on the inbox (`live_baseline_at`) are recorded as `historical` and never evaluated. A failing vault retries with capped backoff, and is left out of the batch meanwhile so it cannot crowd out other vaults.
+
+`MORPHO_LIVE_DELIVERY` picks the delivery mode:
+
+- `shadow` (default): each request is posted to the internal errors chat as `[morpho-live-shadow]`, as plain text split to Telegram's limit, in order, retrying a failed or skipped send (e.g. a missing bot token) until Telegram accepts it. A request whose source events a reorg removed or rewrote before delivery is `invalidated` instead. Shadow requires the errors chat (`TELEGRAM_TOPIC_ID_ERRORS` or `TELEGRAM_CHAT_ID_ERRORS`) and falls back to `observe` without it, so its output never reaches the Morpho group.
+- `observe`: requests are recorded as `observed`; nothing is sent.
+- `production`: alerts go to the Morpho channel, like the daily jobs. Per vault, under a `governance_lock` that the daily `governance.py` / `governance_v2.py` jobs also take, the consumer evaluates, sends every message, and only then commits the daily jobs' cache writes. So each change is alerted once, by whichever path sees it first, and the daily jobs keep catching anything the live path missed (indexer gaps, outages) plus resolutions and guardian changes. A failed send, or one Telegram skips for missing credentials, commits nothing and leaves the events undecided; they are re-evaluated and re-sent on retry (a duplicate beats a missed alert, as in the daily jobs).
+
+Switch to `production` only after a shadow period matched the daily alerts ([#450](https://github.com/yearn/monitoring/issues/450)). The lock is a file in `CACHE_DIR` (`morpho_governance_<vault>.lock`), so both processes must share it, as the `monitoring` and `monitoring-timelock-subscriber` units do. `event_subscriber.py --streams morpho report` adds a live-decision summary (outcomes, block-to-decision delay, request statuses); `health` also fails when events stay undecided or shadow alerts stay undelivered for more than 10 minutes.
+
+Enable the stream 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
 

protocols/morpho/_alerts.py

@@ -15,12 +15,14 @@
 
 from __future__ import annotations
 
+from contextlib import AbstractContextManager
 from dataclasses import dataclass, field
 from datetime import datetime
 from functools import partial
 from typing import Any, Callable, List
 
 from utils.alert import Alert, AlertSeverity, send_alert
+from utils.file_lock import file_lock
 from utils.telegram import MAX_MESSAGE_LENGTH
 
 # Ascending severity — a grouped alert is sent at the highest of its sections.
@@ -44,6 +46,16 @@
 REMINDER_WINDOW_SECONDS = 24 * 60 * 60
 
 
+def governance_lock(vault_address: str) -> AbstractContextManager[None]:
+    """Lock one vault's governance check → send → cache-write sequence across processes.
+
+    The daily jobs and the live consumer's production mode read and write the
+    same cache keys; holding this lock around the whole sequence keeps them from
+    both treating one change as new and alerting it twice.
+    """
+    return file_lock(f"morpho_governance_{vault_address.lower()}")
+
+
 @dataclass(frozen=True)
 class LossNotice:
     """A potential-loss alert that is due for one pending operation."""
@@ -155,30 +167,42 @@ def split_into_messages(alerts: List[VaultAlert], budget: int) -> List[List[str]
     return parts
 
 
+def render_vault_alerts(header: str, alerts: List[VaultAlert]) -> List[tuple[AlertSeverity, str]]:
+    """Render buffered sections into the ``(severity, message)`` pairs ``send_vault_alerts`` sends.
+
+    Standalone (potential-loss) sections come first, one message each, under a
+    ``LOSS_PREFIX`` header; the remaining sections are grouped under ``header`` at
+    their highest severity and split into "(i/N)" parts if long.
+    """
+    messages: List[tuple[AlertSeverity, str]] = []
+    for alert in alerts:
+        if alert.standalone:
+            messages += _render_group(f"{LOSS_PREFIX} — {header}", [alert])
+    messages += _render_group(header, [a for a in alerts if not a.standalone])
+    return messages
+
+
 def send_vault_alerts(header: str, alerts: List[VaultAlert], protocol: str) -> None:
     """Send the buffered sections as one Telegram message, or "(i/N)" parts if long.
 
     No-op when ``alerts`` is empty so callers don't have to guard. Every part
     carries the same header and the highest severity of the whole group, so a
-    LOW section bundled with an owner change still pings the channel.
-
-    Standalone (potential-loss) sections go out first, one message each, under a
-    ``LOSS_PREFIX`` header; the remaining sections are grouped as usual.
+    LOW section bundled with an owner change still pings the channel. See
+    ``render_vault_alerts`` for the message layout.
     """
-    for alert in alerts:
-        if alert.standalone:
-            _send_group(f"{LOSS_PREFIX} — {header}", [alert], protocol)
-    _send_group(header, [a for a in alerts if not a.standalone], protocol)
+    for severity, message in render_vault_alerts(header, alerts):
+        send_alert(Alert(severity, message, protocol))
 
 
-def _send_group(header: str, alerts: List[VaultAlert], protocol: str) -> None:
-    """Send sections under one header at their highest severity, split into parts if long."""
+def _render_group(header: str, alerts: List[VaultAlert]) -> List[tuple[AlertSeverity, str]]:
+    """Render sections under one header at their highest severity, split into parts if long."""
     if not alerts:
-        return
+        return []
     severity = max((a.severity for a in alerts), key=_SEVERITY_ORDER.index)
     parts = split_into_messages(alerts, MAX_MESSAGE_LENGTH - _MESSAGE_OVERHEAD - len(header))
     total = len(parts)
+    messages = []
     for index, bodies in enumerate(parts, start=1):
         suffix = f" ({index}/{total})" if total > 1 else ""
-        message = f"{header}{suffix}\n\n" + SECTION_SEPARATOR.join(bodies)
-        send_alert(Alert(severity, message, protocol))
+        messages.append((severity, f"{header}{suffix}\n\n" + SECTION_SEPARATOR.join(bodies)))
+    return messages

protocols/morpho/daily_governance_cache.py

@@ -0,0 +1,52 @@
+"""Point this process's Morpho cache at the daily governance jobs' namespace.
+
+The daily profile in ``automation/jobs.yaml`` runs ``governance.py`` and
+``governance_v2.py`` with ``MORPHO_FILENAME`` set to its own namespace, and
+``utils.cache`` resolves that name once, at import. The continuous subscriber
+runs outside that profile, so without this its live governance checks would read
+(and in production, write) a different namespace: no role/set baselines, no
+record of what the daily jobs already alerted.
+
+Importing this module sets ``MORPHO_FILENAME`` from the daily profile, unless the
+process already sets it explicitly, so it must be imported before anything
+imports ``utils.cache``. ``check_daily_governance_cache`` verifies at startup that
+the order held.
+"""
+
+import os
+
+from automation.config import load_jobs_config
+
+DAILY_PROFILE = "daily"
+
+
+def daily_morpho_filename() -> str:
+    """Return the ``MORPHO_FILENAME`` the daily governance jobs run with."""
+    filename = load_jobs_config().profiles[DAILY_PROFILE].env.get("MORPHO_FILENAME")
+    if not filename:
+        raise RuntimeError(f"automation/jobs.yaml profile {DAILY_PROFILE!r} sets no MORPHO_FILENAME")
+    return filename
+
+
+def use_daily_governance_cache() -> str:
+    """Set ``MORPHO_FILENAME`` to the daily jobs' namespace unless already set, and return it."""
+    return os.environ.setdefault("MORPHO_FILENAME", daily_morpho_filename())
+
+
+def check_daily_governance_cache() -> None:
+    """Fail unless ``utils.cache`` resolved the Morpho cache to the daily jobs' namespace.
+
+    Raises:
+        RuntimeError: ``utils.cache`` was imported before this module, or with another name.
+    """
+    from utils import cache  # noqa: PLC0415 - checks the binding the import above must not have fixed early
+
+    expected = os.environ.get("MORPHO_FILENAME") or daily_morpho_filename()
+    if os.path.basename(cache.morpho_filename) != expected:
+        raise RuntimeError(
+            f"utils.cache uses {cache.morpho_filename}, not the daily governance namespace {expected}: "
+            f"import {__name__} before anything imports utils.cache"
+        )
+
+
+use_daily_governance_cache()

protocols/morpho/governance.py

@@ -4,7 +4,7 @@
 
 from web3 import Web3
 
-from protocols.morpho._alerts import VaultDiff, due_loss_notice, send_vault_alerts
+from protocols.morpho._alerts import VaultDiff, due_loss_notice, governance_lock, send_vault_alerts
 from protocols.morpho._shared import (
     MORPHO_BLUE_ABI_PATH,
     PROTOCOL,
@@ -159,7 +159,7 @@ def _load_vault_market_ids(
     return list(set(queued_markets) | pending_markets)
 
 
-def _load_market_governance_states(
+def load_market_governance_states(
     morpho_contract: Any,
     market_ids: list[bytes],
     client: Any,
@@ -219,7 +219,7 @@ def _try_total_assets(morpho_contract: Any, block_number: int) -> int | None:
         return None
 
 
-def _vault_header(name: str, vault_address: str, chain: Chain) -> str:
+def vault_header(name: str, vault_address: str, chain: Chain) -> str:
     """One-line header for the vault's grouped alert."""
     return f"[{name}]({get_vault_url(vault_address, chain)}) on {chain.name}"
 
@@ -359,7 +359,7 @@ def check_markets_pending_cap(
 ) -> None:
     """Check V1 market cap and removal governance for one vault."""
     market_ids = _load_vault_market_ids(morpho_contract, chain, client, block_number)
-    for state in _load_market_governance_states(morpho_contract, market_ids, client, block_number):
+    for state in load_market_governance_states(morpho_contract, market_ids, client, block_number):
         check_market_governance_state(name, state, chain, diff)
 
 
@@ -422,19 +422,21 @@ def get_data_for_chain(chain: Chain) -> None:
     for vault in vaults:
         morpho_contract = client.eth.contract(address=vault.address, abi=ABI_MORPHO)
         # Buffer every finding for this vault, then send them as one message and
-        # only then record them as alerted.
-        diff = VaultDiff()
-        check_markets_pending_cap(
-            vault.name,
-            morpho_contract,
-            chain,
-            client,
-            diff,
-            block_number=block_number,
-        )
-        check_timelock_and_guardian(morpho_contract, client, diff, block_number)
-        send_vault_alerts(_vault_header(vault.name, vault.address, chain), diff.alerts, PROTOCOL)
-        diff.commit()
+        # only then record them as alerted. The lock keeps the live consumer from
+        # alerting the same change in between.
+        with governance_lock(vault.address):
+            diff = VaultDiff()
+            check_markets_pending_cap(
+                vault.name,
+                morpho_contract,
+                chain,
+                client,
+                diff,
+                block_number=block_number,
+            )
+            check_timelock_and_guardian(morpho_contract, client, diff, block_number)
+            send_vault_alerts(vault_header(vault.name, vault.address, chain), diff.alerts, PROTOCOL)
+            diff.commit()
 
 
 def main() -> None: