feat(timelock): add AI summary to stale ready-operation alerts

monitoring.yaml

@@ -421,7 +421,7 @@ protocols:
       - name: "Timelock Events"
         description: "CallScheduled/ProposalQueued/QueueTransaction/StartVote/ProposalScheduled events across monitored timelocks via Envio; TimelockController/Compound alerts show the full operation ID"
       - name: "Stale Ready Operations"
-        description: "Daily: TimelockController operations ready >7 days but neither executed nor cancelled; one silent alert per operation showing the timelock, operation ID, readiness date, call details, and schedule transaction link, with an omitted-call count when details exceed the message budget"
+        description: "Daily: TimelockController operations ready >7 days but neither executed nor cancelled; one silent alert per operation showing the timelock, operation ID, readiness date, call details, schedule transaction link, and AI summary with full report link, with an omitted-call count when details exceed the message budget"
 
   unibtc:
     display_name: "Bedrock uniBTC"

protocols/timelock/README.md

@@ -74,6 +74,8 @@ Only `TimelockController` timelocks are checked. Compound-style queues expire af
 
 Alerts show the timelock, full operation ID in inline code for copying into `getTimestamp(id)` or other state queries, schedule transaction link, readiness date, and call details.
 
+Each alert also includes the AI summary and a "Full details" report link, generated the same way as in the scheduled alert. The analysis is run again, so it reflects the chain state at the time of the reminder. It is skipped for the same protocols (`SKIP_AI_SUMMARY_PROTOCOLS`). If the explainer fails, the alert goes out without the summary. The summary is never truncated; when space runs short, call lines are dropped first. If the summary is too long to fit even after every call line is dropped, the alert goes out without it and a warning is logged.
+
 ## GraphQL Schema
 
 The script queries the unified `TimelockEvent` type from the Envio indexer. The query fetches all timelock types (TimelockController, Aave, Compound, Lido, Maple) for monitored addresses.

protocols/timelock/stale_operations.py

@@ -13,7 +13,8 @@
 anything else is the time it became ready. Envio indexes only CallScheduled, so
 the state comes from the chain. Operations ready for longer than
 ``TIMELOCK_STALE_DAYS`` (default 7) are reported once each, silently, to the
-protocol's channel.
+protocol's channel, with the same AI summary and full report link the scheduled
+alert carries.
 """
 
 import os
@@ -24,15 +25,18 @@
 from eth_utils import to_checksum_address
 
 from protocols.timelock.timelock_alerts import (
+    SKIP_AI_SUMMARY_PROTOCOLS,
     TIMELOCKS,
     YEARN_TIMELOCK_INTERNAL_PROTOCOL,
     TimelockConfig,
     alert_history_protocol,
+    get_ai_explanation,
     load_events,
 )
 from utils.cache import cache_filename, get_last_value_for_key_from_file, write_last_value_to_file
 from utils.calldata.decoder import decode_calldata
 from utils.chains import EXPLORER_URLS, Chain
+from utils.llm.ai_explainer import format_explanation_line
 from utils.logger import get_logger
 from utils.telegram import MAX_MESSAGE_LENGTH, escape_markdown, send_envio_error_message, send_telegram_message
 from utils.web3_wrapper import ChainManager
@@ -71,6 +75,7 @@ class Operation:
     scheduled_at: int
     transaction_hash: str
     calls: tuple[tuple[str, str], ...]  # (target, calldata)
+    values: tuple[str, ...] = ()  # wei sent with each call, aligned with ``calls``
 
 
 def load_scheduled_operations(since_ts: int) -> list[Operation] | None:
@@ -109,6 +114,7 @@ def load_scheduled_operations(since_ts: int) -> list[Operation] | None:
                 scheduled_at=int(events[0]["blockTimestamp"]),
                 transaction_hash=events[0]["transactionHash"],
                 calls=tuple((event.get("target") or "", event.get("data") or "0x") for event in events),
+                values=tuple(str(event.get("value") or "0") for event in events),
             )
         )
     return operations
@@ -192,8 +198,26 @@ def _format_calls(operation: Operation, budget: int) -> str:
     return "\n".join(lines)
 
 
-def format_operation(operation: Operation, ready_at: int, now: int) -> str:
-    """Alert text for one stale operation."""
+def explain_operation(operation: Operation) -> str:
+    """AI summary line (with the full report link) for an operation, or "" when skipped or failed."""
+    if operation.timelock.protocol.upper() in SKIP_AI_SUMMARY_PROTOCOLS:
+        return ""
+    values = operation.values or ("0",) * len(operation.calls)
+    events = [
+        {"target": target, "data": data, "value": value}
+        for (target, data), value in zip(operation.calls, values, strict=True)
+    ]
+    explanation = get_ai_explanation(events, operation.timelock, operation.timelock.chain_id)
+    return format_explanation_line(explanation) if explanation else ""
+
+
+def format_operation(operation: Operation, ready_at: int, now: int, ai_line: str = "") -> str:
+    """Alert text for one stale operation.
+
+    ``ai_line`` follows the calls and is never truncated: call lines are dropped
+    to make room. A summary too long to fit even with every call dropped is left
+    out, so the operation still gets a plain alert instead of failing the run.
+    """
     chain_id = operation.timelock.chain_id
     explorer = EXPLORER_URLS.get(chain_id)
     address = to_checksum_address(operation.timelock.address)
@@ -210,7 +234,16 @@ def format_operation(operation: Operation, ready_at: int, now: int) -> str:
         f"Ready since {_date(ready_at)} ({(now - ready_at) // DAY} days)\n"
         "Calls:\n"
     )
-    budget = MAX_MESSAGE_LENGTH - len(_HEADER) - len(_footer()) - len(prefix)
+    budget = _entry_budget() - len(prefix)
+    if ai_line:
+        text = prefix + _format_calls(operation, budget - len(ai_line)) + ai_line
+        if len(text) <= _entry_budget():
+            return text
+        logger.warning(
+            "AI summary for %s is too long for the alert (%s chars); sending without it",
+            operation.operation_id,
+            len(ai_line),
+        )
     return prefix + _format_calls(operation, budget)
 
 
@@ -225,13 +258,18 @@ def _footer() -> str:
     )
 
 
+def _entry_budget() -> int:
+    """Characters one operation's text may use in a message, after the header and footer."""
+    return MAX_MESSAGE_LENGTH - len(_HEADER) - len(_footer())
+
+
 def chunk_entries(entries: list[tuple[str, str]]) -> list[list[tuple[str, str]]]:
     """Group (cache_key, text) entries into messages that fit Telegram's limit.
 
     send_telegram_message truncates an oversized message and still succeeds, so
     an operation in a cut-off tail would be cached as alerted without being sent.
     """
-    budget = MAX_MESSAGE_LENGTH - len(_HEADER) - len(_footer())
+    budget = _entry_budget()
     chunks: list[list[tuple[str, str]]] = []
     size = 0
     for entry in entries:
@@ -280,7 +318,7 @@ def main() -> None:
         if str(get_last_value_for_key_from_file(cache_filename, key)) == "1":
             continue
         by_protocol.setdefault(operation.timelock.protocol, []).append(
-            (key, format_operation(operation, ready_at, now))
+            (key, format_operation(operation, ready_at, now, explain_operation(operation)))
         )
 
     for protocol, entries in by_protocol.items():

protocols/timelock/timelock_alerts.py

@@ -344,7 +344,7 @@ def _explainer_calls_from_events(events: list[dict]) -> list[dict[str, str]]:
     return calls
 
 
-def _get_ai_explanation(events: list[dict], timelock_info: TimelockConfig, chain_id: int) -> Explanation | None:
+def get_ai_explanation(events: list[dict], timelock_info: TimelockConfig, chain_id: int) -> Explanation | None:
     """Generate AI explanation for timelock events. Returns None on any failure."""
     try:
         # Maple's ProposalScheduled event only carries an opaque proposalId — the
@@ -524,7 +524,7 @@ def build_alert_message(events: list[dict], timelock_info: TimelockConfig) -> st
     # governance proposals are already monitored by a dedicated script.
     ai_line = ""
     if timelock_info.protocol.upper() not in SKIP_AI_SUMMARY_PROTOCOLS:
-        explanation = _get_ai_explanation(events, timelock_info, chain_id)
+        explanation = get_ai_explanation(events, timelock_info, chain_id)
         if explanation:
             ai_line = format_explanation_line(explanation)
 

tests/test_timelock_alerts.py

@@ -45,7 +45,7 @@ def _make_event(
 class TestBuildAlertMessageTruncation(unittest.TestCase):
     """Test that build_alert_message respects MAX_MESSAGE_LENGTH and priority."""
 
-    @patch("protocols.timelock.timelock_alerts._get_ai_explanation", return_value=None)
+    @patch("protocols.timelock.timelock_alerts.get_ai_explanation", return_value=None)
     def test_short_message_no_truncation(self, _mock_ai: object) -> None:
         """A simple message should not be truncated."""
         events = [_make_event()]
@@ -54,7 +54,7 @@ def test_short_message_no_truncation(self, _mock_ai: object) -> None:
         self.assertIn("TIMELOCK: New Operation Scheduled", msg)
         self.assertIn("Test Timelock", msg)
 
-    @patch("protocols.timelock.timelock_alerts._get_ai_explanation", return_value=None)
+    @patch("protocols.timelock.timelock_alerts.get_ai_explanation", return_value=None)
     def test_batch_calls_are_numbered_from_one(self, _mock_ai: object) -> None:
         """Telegram numbering must match the AI report's 1-based call flow."""
         events = [_make_event(index=i, target=f"0x{i + 1:040x}", data="0x8456cb59") for i in range(2)]
@@ -63,7 +63,7 @@ def test_batch_calls_are_numbered_from_one(self, _mock_ai: object) -> None:
         self.assertIn("--- Call 2 ---", msg)
         self.assertNotIn("--- Call 0 ---", msg)
 
-    @patch("protocols.timelock.timelock_alerts._get_ai_explanation", return_value=None)
+    @patch("protocols.timelock.timelock_alerts.get_ai_explanation", return_value=None)
     def test_long_call_details_truncated(self, _mock_ai: object) -> None:
         """When call details are very long, they should be truncated to fit."""
         events = [
@@ -78,7 +78,7 @@ def test_long_call_details_truncated(self, _mock_ai: object) -> None:
         self.assertLessEqual(len(msg), MAX_MESSAGE_LENGTH)
         self.assertIn("truncated", msg)
 
-    @patch("protocols.timelock.timelock_alerts._get_ai_explanation", return_value=None)
+    @patch("protocols.timelock.timelock_alerts.get_ai_explanation", return_value=None)
     def test_truncation_keeps_markdown_entities_balanced(self, _mock_ai: object) -> None:
         """Truncating call details must not sever a `signature` mid-entity.
 
@@ -103,7 +103,7 @@ def test_truncation_keeps_markdown_entities_balanced(self, _mock_ai: object) ->
         self.assertEqual(msg.count("["), msg.count("]"), "unbalanced link brackets")
 
     @patch("protocols.timelock.timelock_alerts.format_explanation_line")
-    @patch("protocols.timelock.timelock_alerts._get_ai_explanation")
+    @patch("protocols.timelock.timelock_alerts.get_ai_explanation")
     def test_ai_summary_preserved_over_call_details(self, mock_ai: MagicMock, mock_format: MagicMock) -> None:
         """AI summary must be preserved even when call details are long."""
         from utils.llm.ai_explainer import Explanation
@@ -132,7 +132,7 @@ def test_ai_summary_preserved_over_call_details(self, mock_ai: MagicMock, mock_f
         self.assertIn("truncated", msg)
 
     @patch("protocols.timelock.timelock_alerts.format_explanation_line")
-    @patch("protocols.timelock.timelock_alerts._get_ai_explanation")
+    @patch("protocols.timelock.timelock_alerts.get_ai_explanation")
     def test_message_under_limit_with_ai(self, mock_ai: MagicMock, mock_format: MagicMock) -> None:
         """When everything fits, nothing should be truncated."""
         from utils.llm.ai_explainer import Explanation
@@ -148,7 +148,7 @@ def test_message_under_limit_with_ai(self, mock_ai: MagicMock, mock_format: Magi
         self.assertIn("Short summary.", msg)
         self.assertNotIn("...", msg)
 
-    @patch("protocols.timelock.timelock_alerts._get_ai_explanation")
+    @patch("protocols.timelock.timelock_alerts.get_ai_explanation")
     def test_ai_skipped_for_governance_protocol(self, mock_ai: MagicMock) -> None:
         """Protocols with dedicated governance monitoring skip the AI summary entirely."""
         aave_info = TimelockConfig(
@@ -164,7 +164,7 @@ def test_ai_skipped_for_governance_protocol(self, mock_ai: MagicMock) -> None:
         self.assertIn("TIMELOCK: New Operation Scheduled", msg)
 
 
-@patch("protocols.timelock.timelock_alerts._get_ai_explanation", return_value=None)
+@patch("protocols.timelock.timelock_alerts.get_ai_explanation", return_value=None)
 class TestOperationIdLine(unittest.TestCase):
     """The scheduled alert shows the same copyable Operation ID as the stale-operation alert."""
 
@@ -364,47 +364,47 @@ class TestAiExplanationCallGates(unittest.TestCase):
 
     @patch("protocols.timelock.timelock_alerts.explain_transaction")
     def test_empty_calldata_single_call_reaches_explainer(self, mock_explain: unittest.mock.MagicMock) -> None:
-        from protocols.timelock.timelock_alerts import _get_ai_explanation
+        from protocols.timelock.timelock_alerts import get_ai_explanation
 
         mock_explain.return_value = None
         event = _make_event(data="0x", value="1000")
-        _get_ai_explanation([event], TIMELOCK_INFO, 1)
+        get_ai_explanation([event], TIMELOCK_INFO, 1)
         mock_explain.assert_called_once()
         self.assertEqual(mock_explain.call_args.kwargs["calldata"], "0x")
         self.assertEqual(mock_explain.call_args.kwargs["value"], 1000)
 
     @patch("protocols.timelock.timelock_alerts.explain_transaction")
     def test_null_value_does_not_crash_explainer(self, mock_explain: unittest.mock.MagicMock) -> None:
-        from protocols.timelock.timelock_alerts import _get_ai_explanation
+        from protocols.timelock.timelock_alerts import get_ai_explanation
 
         mock_explain.return_value = None
         event = _make_event(data="0x8456cb59")
         event["value"] = None
-        _get_ai_explanation([event], TIMELOCK_INFO, 1)
+        get_ai_explanation([event], TIMELOCK_INFO, 1)
         mock_explain.assert_called_once()
         self.assertEqual(mock_explain.call_args.kwargs["value"], 0)
 
     @patch("protocols.timelock.timelock_alerts.explain_transaction")
     def test_short_calldata_single_call_reaches_explainer(self, mock_explain: unittest.mock.MagicMock) -> None:
-        from protocols.timelock.timelock_alerts import _get_ai_explanation
+        from protocols.timelock.timelock_alerts import get_ai_explanation
 
         mock_explain.return_value = None
         event = _make_event(data="0x12")
-        _get_ai_explanation([event], TIMELOCK_INFO, 1)
+        get_ai_explanation([event], TIMELOCK_INFO, 1)
         mock_explain.assert_called_once()
         self.assertEqual(mock_explain.call_args.kwargs["calldata"], "0x12")
 
     @patch("protocols.timelock.timelock_alerts.explain_batch_transaction")
     def test_mixed_batch_keeps_empty_and_short_entries(self, mock_batch: unittest.mock.MagicMock) -> None:
-        from protocols.timelock.timelock_alerts import _get_ai_explanation
+        from protocols.timelock.timelock_alerts import get_ai_explanation
 
         mock_batch.return_value = None
         events = [
             _make_event(data="0x8456cb59", target="0x" + "11" * 20),
             _make_event(data="0x", target="0x" + "22" * 20, value="1"),
             _make_event(data="0x12", target="0x" + "33" * 20),
         ]
-        _get_ai_explanation(events, TIMELOCK_INFO, 1)
+        get_ai_explanation(events, TIMELOCK_INFO, 1)
         mock_batch.assert_called_once()
         calls = mock_batch.call_args.kwargs["calls"]
         self.assertEqual([c["data"] for c in calls], ["0x8456cb59", "0x", "0x12"])
@@ -415,9 +415,9 @@ def test_mixed_batch_keeps_empty_and_short_entries(self, mock_batch: unittest.mo
     def test_events_without_target_are_skipped(
         self, mock_batch: unittest.mock.MagicMock, mock_single: unittest.mock.MagicMock
     ) -> None:
-        from protocols.timelock.timelock_alerts import _get_ai_explanation
+        from protocols.timelock.timelock_alerts import get_ai_explanation
 
-        result = _get_ai_explanation([_make_event(target="", data="0x8456cb59")], TIMELOCK_INFO, 1)
+        result = get_ai_explanation([_make_event(target="", data="0x8456cb59")], TIMELOCK_INFO, 1)
         self.assertIsNone(result)
         mock_batch.assert_not_called()
         mock_single.assert_not_called()