diff --git a/monitoring.yaml b/monitoring.yaml index da89818..c71705c 100644 --- a/monitoring.yaml +++ b/monitoring.yaml @@ -242,6 +242,7 @@ protocols: display_name: "Maple Finance" description: "Maple syrupUSDC pool monitoring" cadence: "Hourly" + disabled: true tasks: - protocols/maple/main.py monitors: diff --git a/protocols/safe/main.py b/protocols/safe/main.py index cc78204..847cf1a 100644 --- a/protocols/safe/main.py +++ b/protocols/safe/main.py @@ -231,6 +231,9 @@ def _explain_safe_tx( skip_simulation=True, context_note=context_note, refine=True, + # A utility label names the multisend contract, not the Safe, so the + # report's Contract link must point at the target in that case. + label_address=target if utility_label else safe_address, ) # Non-multisend DELEGATECALLs (rare): skip sim but still try to explain. diff --git a/tests/test_ai_explainer.py b/tests/test_ai_explainer.py index 6040449..bee2387 100644 --- a/tests/test_ai_explainer.py +++ b/tests/test_ai_explainer.py @@ -14,6 +14,7 @@ _explanation_from_json, _format_decimal, _parse_explanation, + collect_unique_addresses, explain_transaction, format_explanation_line, ) @@ -247,6 +248,14 @@ def test_schema_tag_overrides_inlined_tag(self) -> None: class TestParseExplanation(unittest.TestCase): """Tests for _parse_explanation.""" + def test_heading_containing_keyword_is_not_a_marker(self) -> None: + """'## Detailed Analysis' must not match the DETAIL marker and get sliced.""" + raw = "## Detailed Analysis\n\nThe call registers a farm." + result = _parse_explanation(raw) + self.assertEqual(result.detail, "") + self.assertIn("Detailed Analysis", result.summary) + self.assertIn("registers a farm", result.summary) + def test_both_sections(self) -> None: raw = "TLDR: Short summary here.\n\nDETAIL:\nDetailed analysis here." result = _parse_explanation(raw) @@ -279,9 +288,78 @@ def test_multiline_detail(self) -> None: self.assertIn("Risk: HIGH", result.detail) +class TestAddressLinksSection(unittest.TestCase): + """The prompt hands the LLM ready-made explorer links to copy.""" + + def test_links_block_included(self) -> None: + calls = [DecodedCall(function_name="pause", signature="pause()")] + links = "- [`0xAbc`](https://etherscan.io/address/0xAbc)" + result = _build_prompt(target="0xTarget", value=0, decoded_calls=calls, simulation=None, address_links=links) + self.assertIn("--- Address Links", result) + self.assertIn(links, result) + + def test_section_omitted_when_no_links(self) -> None: + calls = [DecodedCall(function_name="pause", signature="pause()")] + result = _build_prompt(target="0xTarget", value=0, decoded_calls=calls, simulation=None) + self.assertNotIn("--- Address Links", result) + + def test_hyperlink_rule_in_system_prompt(self) -> None: + self.assertIn("markdown link to the block explorer", SYSTEM_INSTRUCTIONS) + + +class TestCollectUniqueAddresses(unittest.TestCase): + """Targets and address args are gathered once, deduped, checksummed.""" + + def test_target_and_args_deduped(self) -> None: + farm = "0x79e1b8e45932a7c802ea3dab3844e5dea68d971f" + registry = "0xF5f2718708f471e43968271956CC01aaA8c46119" + call = DecodedCall( + function_name="addFarms", + signature="addFarms(uint256,address[])", + params=[("uint256", 2), ("address[]", (farm, farm))], + ) + result = collect_unique_addresses([(registry, call)]) + self.assertEqual(result, [registry, "0x79e1B8e45932A7C802eA3dAb3844e5DEa68d971f"]) + + def test_addresses_inside_tuple_args_collected(self) -> None: + """Struct args (e.g. MarketParams) must contribute their addresses.""" + farm = "0x79e1b8e45932a7c802ea3dab3844e5dea68d971f" + registry = "0xF5f2718708f471e43968271956CC01aaA8c46119" + call = DecodedCall( + function_name="createMarket", + signature="createMarket((address,uint256))", + params=[("(address,uint256)", (farm, 5))], + ) + self.assertEqual( + collect_unique_addresses([(registry, call)]), + [registry, "0x79e1B8e45932A7C802eA3dAb3844e5DEa68d971f"], + ) + + def test_zero_and_malformed_addresses_dropped(self) -> None: + call = DecodedCall( + function_name="transfer", + signature="transfer(address,uint256)", + params=[("address", "0x" + "00" * 20), ("uint256", 1)], + ) + self.assertEqual(collect_unique_addresses([("0xnothex", call)]), []) + + class TestFormatExplanationLine(unittest.TestCase): """Tests for format_explanation_line.""" + @patch("utils.llm.ai_explainer.upload_to_gist", return_value="https://gist.wavey.info/abc123") + def test_report_published_when_present(self, mock_gist: MagicMock) -> None: + """The full report (metadata + call flow + analysis) is what gets uploaded.""" + explanation = Explanation( + summary="Pauses the vault. HIGH", + detail="Full detail here.", + report="## Call Flow\n\n1. pause()", + title="Yearn Timelock - 11/08/2026 10:00 - HIGH", + ) + result = format_explanation_line(explanation) + mock_gist.assert_called_once_with(explanation.report, title=explanation.title) + self.assertIn("https://gist.wavey.info/abc123", result) + @patch("utils.llm.ai_explainer.upload_to_gist", return_value="https://gist.wavey.info/abc123") def test_format_with_detail(self, mock_gist: MagicMock) -> None: explanation = Explanation(summary="This pauses the protocol.", detail="Full detail here.") @@ -648,7 +726,7 @@ def test_undecodable_bytes_falls_back_to_raw(self) -> None: signature="initialize(bytes)", params=[("bytes", garbage)], ) - with patch("utils.llm.ai_explainer.decode_calldata", return_value=None): + with patch("utils.calldata.decoder.decode_calldata", return_value=None): result = _format_decoded_calls([outer]) self.assertIn(f"bytes: {garbage}", result) @@ -666,7 +744,7 @@ def test_unknown_selector_skipped_no_network(self) -> None: signature="exec(bytes)", params=[("bytes", unknown)], ) - with patch("utils.llm.ai_explainer.decode_calldata") as mock_decode: + with patch("utils.calldata.decoder.decode_calldata") as mock_decode: result = _format_decoded_calls([outer]) mock_decode.assert_not_called() self.assertIn(f"bytes: {unknown}", result) @@ -681,15 +759,16 @@ def test_unaligned_bytes_skipped(self) -> None: signature="execTx(bytes)", params=[("bytes", sigs_blob)], ) - with patch("utils.llm.ai_explainer.decode_calldata") as mock_decode: + with patch("utils.calldata.decoder.decode_calldata") as mock_decode: result = _format_decoded_calls([outer]) mock_decode.assert_not_called() self.assertIn(sigs_blob, result) def test_recursion_depth_capped(self) -> None: - from utils.llm.ai_explainer import _MAX_BYTES_RECURSION_DEPTH, _format_decoded_calls + from utils.calldata.decoder import MAX_BYTES_RECURSION_DEPTH + from utils.llm.ai_explainer import _format_decoded_calls - # Mock _try_decode_inner_bytes so it always returns a self-referential + # Mock try_decode_inner_calldata so it always returns a self-referential # call, bypassing the selector/alignment guard. Without the depth cap # this would recurse forever. self_referential = DecodedCall( @@ -697,9 +776,9 @@ def test_recursion_depth_capped(self) -> None: signature="wrap(bytes)", params=[("bytes", "0xfeedfacefeedfacefeedfacefeedfacefeedface")], ) - with patch("utils.llm.ai_explainer._try_decode_inner_bytes", return_value=self_referential): + with patch("utils.llm.ai_explainer.try_decode_inner_calldata", return_value=self_referential): result = _format_decoded_calls([self_referential]) - self.assertEqual(result.count("↳"), _MAX_BYTES_RECURSION_DEPTH) + self.assertEqual(result.count("↳"), MAX_BYTES_RECURSION_DEPTH) class TestAddressLabels(unittest.TestCase): diff --git a/tests/test_llm_report.py b/tests/test_llm_report.py new file mode 100644 index 0000000..a4e9b49 --- /dev/null +++ b/tests/test_llm_report.py @@ -0,0 +1,263 @@ +"""Tests for the gist report renderer (utils/llm/report.py).""" + +import unittest +from datetime import datetime, timezone +from unittest.mock import patch + +from utils.calldata.decoder import MAX_BYTES_RECURSION_DEPTH, DecodedCall +from utils.llm.report import ( + CallEntry, + ReportContext, + address_link, + array_element_type, + build_report, + build_title, + explorer_address_url, + format_address_links_block, + format_call_flow, + iter_address_values, + tuple_component_types, +) + +REGISTRY = "0xF5f2718708f471e43968271956CC01aaA8c46119" +FARM = "0x79e1b8e45932a7c802ea3dab3844e5dea68d971f" +FARM_CKS = "0x79e1B8e45932A7C802eA3dAb3844e5DEa68d971f" +TIMELOCK = "0x4B174afbeD7b98BA01F50E36109EEE5e6d327c32" + + +def _add_farms_ctx(**overrides) -> ReportContext: + call = DecodedCall( + function_name="addFarms", + signature="addFarms(uint256,address[])", + params=[("uint256", 2), ("address[]", (FARM,))], + ) + defaults = { + "entries": [CallEntry(target=REGISTRY, call=call, param_names=["_type", "_farms"])], + "chain_id": 1, + "labels": {REGISTRY: "FarmRegistry"}, + "protocol": "INFINIFI", + "label": "Infinifi Shorttimelock", + "from_address": TIMELOCK, + } + defaults.update(overrides) + return ReportContext(**defaults) # type: ignore[arg-type] + + +class TestAddressLink(unittest.TestCase): + def test_full_checksummed_address_is_linked(self) -> None: + result = address_link(FARM, 1) + self.assertEqual(result, f"[`{FARM_CKS}`](https://etherscan.io/address/{FARM_CKS})") + + def test_label_is_appended(self) -> None: + self.assertTrue(address_link(REGISTRY, 1, {REGISTRY: "FarmRegistry"}).endswith("(FarmRegistry)")) + + def test_chain_specific_explorer(self) -> None: + self.assertIn("arbiscan.io", address_link(FARM, 42161)) + self.assertIn("basescan.org", address_link(FARM, 8453)) + + def test_unknown_chain_falls_back_to_plain_code(self) -> None: + self.assertEqual(address_link(FARM, 999999), f"`{FARM_CKS}`") + self.assertEqual(explorer_address_url(999999, FARM), "") + + def test_non_address_passes_through(self) -> None: + self.assertEqual(address_link("not-an-address", 1), "not-an-address") + + +class TestAddressLinksBlock(unittest.TestCase): + def test_lists_one_markdown_link_per_address(self) -> None: + block = format_address_links_block([REGISTRY, FARM], 1, {REGISTRY: "FarmRegistry"}) + self.assertIn(f"- [`{REGISTRY}`](https://etherscan.io/address/{REGISTRY}) (FarmRegistry)", block) + self.assertIn(f"- [`{FARM_CKS}`](https://etherscan.io/address/{FARM_CKS})", block) + + def test_empty_when_chain_has_no_explorer(self) -> None: + self.assertEqual(format_address_links_block([REGISTRY], 999999), "") + + +class TestFormatCallFlow(unittest.TestCase): + def test_renders_sender_target_and_params(self) -> None: + flow = format_call_flow(_add_farms_ctx()) + self.assertIn(f"**From:** [`{TIMELOCK}`](https://etherscan.io/address/{TIMELOCK})", flow) + self.assertIn("1. **`addFarms(uint256,address[])`**", flow) + self.assertIn(f"on [`{REGISTRY}`](https://etherscan.io/address/{REGISTRY}) (FarmRegistry)", flow) + self.assertIn("- `uint256 _type`: `2`", flow) + self.assertIn(f" - [`{FARM_CKS}`](https://etherscan.io/address/{FARM_CKS})", flow) + + def test_bare_types_when_param_names_unknown(self) -> None: + ctx = _add_farms_ctx( + entries=[ + CallEntry( + target=REGISTRY, + call=DecodedCall(function_name="setFee", signature="setFee(uint256)", params=[("uint256", 2500)]), + ) + ] + ) + self.assertIn("- `uint256`: `2,500`", format_call_flow(ctx)) + + def test_no_inputs_marker(self) -> None: + ctx = _add_farms_ctx( + entries=[CallEntry(target=REGISTRY, call=DecodedCall(function_name="pause", signature="pause()"))] + ) + self.assertIn("_no inputs_", format_call_flow(ctx)) + + def test_eth_value_shown(self) -> None: + ctx = _add_farms_ctx( + entries=[ + CallEntry( + target=REGISTRY, + call=DecodedCall(function_name="deposit", signature="deposit()"), + value=10**18, + ) + ] + ) + self.assertIn("**ETH value:** `1.000000` ETH", format_call_flow(ctx)) + + def test_zero_sender_omitted(self) -> None: + ctx = _add_farms_ctx(from_address="0x" + "00" * 20) + self.assertNotIn("**From:**", format_call_flow(ctx)) + + def test_numbered_across_batch(self) -> None: + call = DecodedCall(function_name="pause", signature="pause()") + ctx = _add_farms_ctx( + entries=[CallEntry(target=REGISTRY, call=call), CallEntry(target=FARM, call=call)], + ) + flow = format_call_flow(ctx) + self.assertIn("1. **`pause()`**", flow) + self.assertIn("2. **`pause()`**", flow) + + def test_nested_bytes_decoded(self) -> None: + # `0x8456cb59` is pause() — a known selector, resolvable offline. + outer = DecodedCall( + function_name="upgradeToAndCall", + signature="upgradeToAndCall(address,bytes)", + params=[("address", FARM), ("bytes", "0x8456cb59")], + ) + flow = format_call_flow(_add_farms_ctx(entries=[CallEntry(target=REGISTRY, call=outer)])) + self.assertIn("↳ `pause()`", flow) + + def test_nested_recursion_capped(self) -> None: + self_referential = DecodedCall( + function_name="wrap", + signature="wrap(bytes)", + params=[("bytes", "0xfeedfacefeedfacefeedfacefeedfacefeedface")], + ) + ctx = _add_farms_ctx(entries=[CallEntry(target=REGISTRY, call=self_referential)]) + with patch("utils.llm.report.try_decode_inner_calldata", return_value=self_referential): + flow = format_call_flow(ctx) + self.assertEqual(flow.count("↳"), MAX_BYTES_RECURSION_DEPTH) + + def test_empty_without_entries(self) -> None: + self.assertEqual(format_call_flow(_add_farms_ctx(entries=[])), "") + + +class TestCompositeParams(unittest.TestCase): + """Addresses nested in tuple/struct args must still render as explorer links.""" + + def _flow(self, signature: str, params: list) -> str: + call = DecodedCall(function_name=signature.split("(")[0], signature=signature, params=params) + return format_call_flow(_add_farms_ctx(entries=[CallEntry(target=REGISTRY, call=call)])) + + def test_type_decomposition(self) -> None: + self.assertEqual(array_element_type("uint256[3]"), "uint256") + self.assertEqual(array_element_type("(address,uint256)[]"), "(address,uint256)") + self.assertIsNone(array_element_type("address")) + self.assertEqual(tuple_component_types("(address,uint256)"), ["address", "uint256"]) + self.assertEqual(tuple_component_types("(address,(address,uint256))"), ["address", "(address,uint256)"]) + self.assertIsNone(tuple_component_types("address[]")) + + def test_tuple_addresses_linked(self) -> None: + flow = self._flow("configure((address,uint256))", [("(address,uint256)", (FARM, 5))]) + self.assertIn(f" - `address`: [`{FARM_CKS}`](https://etherscan.io/address/{FARM_CKS})", flow) + self.assertIn(" - `uint256`: `5`", flow) + self.assertNotIn(FARM, flow) # no raw lowercase tuple dump + + def test_array_of_tuples_indexed(self) -> None: + flow = self._flow("setCaps((address,uint256)[])", [("(address,uint256)[]", ((FARM, 5), (REGISTRY, 9)))]) + self.assertIn(" - `[0]`:", flow) + self.assertIn(" - `[1]`:", flow) + self.assertIn(f"https://etherscan.io/address/{FARM_CKS}", flow) + self.assertIn(f"https://etherscan.io/address/{REGISTRY}", flow) + + def test_nested_tuple_addresses_linked(self) -> None: + flow = self._flow("init((address,(address,uint256)))", [("(address,(address,uint256))", (REGISTRY, (FARM, 1)))]) + self.assertIn(f"https://etherscan.io/address/{FARM_CKS}", flow) + + def test_scalar_array_rendering_unchanged(self) -> None: + """Plain address[] keeps its compact, index-free bullets.""" + flow = self._flow("addFarms(address[])", [("address[]", (FARM,))]) + self.assertIn(f" - [`{FARM_CKS}`](https://etherscan.io/address/{FARM_CKS})", flow) + self.assertNotIn("`[0]`", flow) + + def test_arity_mismatch_falls_back_to_scalar(self) -> None: + """A value that doesn't match its tuple type is printed, not crashed on.""" + flow = self._flow("configure((address,uint256))", [("(address,uint256)", (FARM,))]) + self.assertIn("`(address,uint256)`:", flow) + + def test_iter_address_values_walks_composites(self) -> None: + self.assertEqual(list(iter_address_values("(address,uint256)", (FARM, 5))), [FARM]) + self.assertEqual(list(iter_address_values("(address,uint256)[]", ((FARM, 5), (REGISTRY, 9)))), [FARM, REGISTRY]) + self.assertEqual(list(iter_address_values("uint256", 5)), []) + self.assertEqual(list(iter_address_values("address", FARM)), [FARM]) + + +class TestBuildTitle(unittest.TestCase): + NOW = datetime(2026, 8, 11, 10, 0, tzinfo=timezone.utc) + + def test_contract_timestamp_and_risk(self) -> None: + title = build_title(_add_farms_ctx(), "LOW", now=self.NOW) + self.assertEqual(title, "Infinifi Shorttimelock - 11/08/2026 10:00 - LOW") + + def test_risk_omitted_when_unknown(self) -> None: + self.assertEqual(build_title(_add_farms_ctx(), now=self.NOW), "Infinifi Shorttimelock - 11/08/2026 10:00") + + def test_falls_back_to_protocol_then_fallback(self) -> None: + self.assertEqual( + build_title(_add_farms_ctx(label=""), "LOW", now=self.NOW), "INFINIFI - 11/08/2026 10:00 - LOW" + ) + self.assertEqual( + build_title(_add_farms_ctx(label="", protocol=""), "LOW", now=self.NOW, fallback="AI Report"), + "AI Report - 11/08/2026 10:00 - LOW", + ) + + +class TestBuildReport(unittest.TestCase): + def test_sections_in_order(self) -> None: + report = build_report("Registers a type-2 farm.", "Long analysis.", _add_farms_ctx(), "MEDIUM") + self.assertLess(report.index("## Summary"), report.index("## Call Flow")) + self.assertLess(report.index("## Call Flow"), report.index("## Analysis")) + + def test_metadata_header(self) -> None: + report = build_report("Summary.", "Analysis.", _add_farms_ctx(label_address=TIMELOCK), "HIGH") + self.assertIn("- **Protocol:** INFINIFI", report) + self.assertIn( + f"- **Contract:** Infinifi Shorttimelock — [`{TIMELOCK}`](https://etherscan.io/address/{TIMELOCK})", + report, + ) + self.assertIn("- **Chain:** Mainnet (chain id 1)", report) + self.assertIn("- **Risk:** HIGH", report) + + def test_contract_unlinked_without_label_address(self) -> None: + report = build_report("Summary.", "Analysis.", _add_farms_ctx()) + self.assertIn("- **Contract:** Infinifi Shorttimelock\n", report) + + def test_contract_unlinked_on_chain_without_explorer(self) -> None: + ctx = _add_farms_ctx(chain_id=999999, label_address=TIMELOCK) + self.assertIn("- **Contract:** Infinifi Shorttimelock\n", build_report("S.", "A.", ctx)) + + def test_redundant_analysis_heading_stripped(self) -> None: + report = build_report("Summary.", "## Detailed Analysis\n\nThe call registers a farm.", _add_farms_ctx()) + self.assertEqual(report.count("Analysis"), 1) + self.assertIn("The call registers a farm.", report) + + def test_own_subheadings_kept(self) -> None: + report = build_report("Summary.", "### Call Breakdown\n\nDetails.", _add_farms_ctx()) + self.assertIn("### Call Breakdown", report) + + def test_risk_omitted_when_unknown(self) -> None: + self.assertNotIn("**Risk:**", build_report("Summary.", "Analysis.", _add_farms_ctx())) + + def test_empty_without_content(self) -> None: + self.assertEqual(build_report("", "", _add_farms_ctx()), "") + + +if __name__ == "__main__": + unittest.main() diff --git a/utils/calldata/decoder.py b/utils/calldata/decoder.py index 0d7c228..084381b 100644 --- a/utils/calldata/decoder.py +++ b/utils/calldata/decoder.py @@ -177,14 +177,16 @@ def _parse_param_types(signature: str) -> list[str]: return [] inner = signature[start + 1 : end].strip() - return _split_top_level(inner) if inner else [] + return split_top_level_types(inner) if inner else [] -def _split_top_level(types: str) -> list[str]: +def split_top_level_types(types: str) -> list[str]: """Split a comma-separated type list on top-level commas only. Commas inside ``(...)`` tuples or ``[...]`` array sizes are preserved so a - type like ``(address,uint256)[]`` stays intact. + type like ``(address,uint256)[]`` stays intact. Public so consumers that + walk decoded values (e.g. the gist report renderer) can decompose a tuple + type into its component types. """ parts: list[str] = [] depth = 0 @@ -292,6 +294,58 @@ def decode_calldata(data_hex: str, chain_id: int | None = None, target: str | No return DecodedCall(function_name=func_name, signature=signature, params=params) +# How many levels of `bytes` parameters holding inner calldata we unwrap. Two is +# enough for the shapes we see (governor execute → upgradeToAndCall → initializer) +# and bounds the work on adversarial or self-referential payloads. +MAX_BYTES_RECURSION_DEPTH = 2 + + +def looks_like_calldata(byte_len: int) -> bool: + """True if a `bytes` blob's length matches the calldata shape (selector + ABI words). + + Real calldata is either a bare 4-byte selector (e.g. `pause()`) or + selector + N 32-byte words. Anything else — packed Safe `signatures` + blobs, EIP-712 hashes, Universal-Router-style packed paths — fails + this check and is left as opaque hex. + """ + return byte_len == 4 or (byte_len >= 36 and (byte_len - 4) % 32 == 0) + + +def try_decode_inner_calldata(value: object) -> DecodedCall | None: + """If ``value`` looks like calldata for an offline-known function, decode it. + + Gated on (1) length matching the calldata shape and (2) the selector + being resolvable without a network call. Without these guards we'd + spam the Sourcify 4byte API on every `signatures`/hash/packed-bytes + parameter, paying a 30s timeout each miss to maybe get a false positive. + + Args: + value: A decoded ``bytes`` parameter, either as ``bytes`` or hex string. + + Returns: + The decoded inner call, or None when the blob isn't recognizable calldata. + """ + if isinstance(value, bytes): + if not looks_like_calldata(len(value)): + return None + hex_str = "0x" + value.hex() + elif isinstance(value, str): + hex_str = value if value.startswith("0x") else "0x" + value + # Each hex char is 4 bits, so byte_len = (len(hex_str) - 2) // 2. + if len(hex_str) < 10 or not looks_like_calldata((len(hex_str) - 2) // 2): + return None + else: + return None + + if not is_selector_resolvable_offline(hex_str[:10]): + return None + + try: + return decode_calldata(hex_str) + except (ValueError, TypeError): + return None + + def format_call_lines(data_hex: str) -> list[str]: """Decode calldata and return formatted lines for an alert message. diff --git a/utils/llm/README.md b/utils/llm/README.md index ca47830..a66f42a 100644 --- a/utils/llm/README.md +++ b/utils/llm/README.md @@ -39,7 +39,9 @@ Generates human-readable explanations for queued governance transactions (timelo ┌─────────────────────┐ │ Telegram Alert │ │ 🤖 AI Summary: ... │ - └─────────────────────┘ + │ [Full details] ────┼──▶ Wavey Gist report + └─────────────────────┘ (metadata + summary + + call flow + analysis) ``` ## Pipeline Steps @@ -158,6 +160,7 @@ The prompt is split into a **system** prompt (static instructions) and a **user* - Starts with a verb, no "This transaction…" preamble - Trailing risk tag in caps (LOW / MEDIUM / HIGH / CRITICAL) +- Summary is plain text (it goes to Telegram); the detail is markdown and must render **every address as a block-explorer hyperlink**, copied verbatim from the prompt's `--- Address Links ---` section so the model never assembles an explorer URL or picks the wrong chain's explorer - Refuses to assume parameter units from function name alone - Trusts source-context natspec over prior assumptions - Quotes concrete before→after deltas when state reads are available @@ -186,6 +189,10 @@ Upgrade the pool implementation to add an emergency pause. Call 1: upgradeTo(address) address: 0xNewImpl +--- Address Links (use these exact markdown links in the detailed report) --- +- [`0xProxy`](https://etherscan.io/address/0xProxy) (PoolAddressesProvider) +- [`0xNewImpl`](https://etherscan.io/address/0xNewImpl) + --- Shared Across Batch --- (optional, for batch txs with uniform args) arg[0] (address) is identical across all 4 calls: '0x...' @@ -227,7 +234,7 @@ The full prompt is logged at INFO level for debugging. ### 7. Two-Stage Generation: Summary, then Detail Derived From It -`_generate_explanation()` produces the `Explanation` dataclass (`summary` → Telegram, `detail` → Wavey Gist) in two stages so the two artifacts the team sees can never disagree on the headline number or risk verdict: +`_generate_explanation()` produces the `Explanation` dataclass (`summary` → Telegram, `detail` → wrapped into `report` → Wavey Gist) in two stages so the two artifacts the team sees can never disagree on the headline number or risk verdict: 1. **Summary (authoritative).** `_generate_summary()` produces just the `summary` + `risk_tag`. 2. **Detail (derived).** `_expand_detail()` then writes the full report *from* the confirmed summary (`DETAIL_EXPANSION_TASK`), required to stay consistent with its magnitudes and risk level. @@ -272,7 +279,59 @@ Upgrades AAVE pool impl 0xOld → 0xNew. Verify audited. MEDIUM. [Full details](https://gist.wavey.info/abc123) ``` -The "Full details" link points to a Wavey Gist upload with the detailed analysis. +The "Full details" link points to a Wavey Gist upload of the **full report** +(`Explanation.report`, built by `utils/llm/report.py`). The gist is titled +` -
- ` (UTC, e.g. +`Infinifi Shorttimelock - 11/08/2026 10:00 - LOW`) so a list of reports is +scannable; it falls back to the protocol name, then `AI Transaction Analysis`. +When no report was built (e.g. an explanation generated without report context), +the bare detail is published under the fallback title instead. + +### 10. Gist Report (`utils/llm/report.py`) + +The gist is the artifact a reviewer actually opens, so it carries more than the LLM's prose: + +```markdown +- **Protocol:** INFINIFI +- **Contract:** Infinifi Shorttimelock — [`0x4B17…7c32`](https://etherscan.io/address/0x4B17…) +- **Chain:** Mainnet (chain id 1) +- **Risk:** MEDIUM + +## Summary +Registers a new type-2 farm in FarmRegistry. … + +## Call Flow +**From:** [`0x4B17…7c32`](https://etherscan.io/address/0x4B17…) + +1. **`addFarms(uint256,address[])`** on [`0xF5f2…6119`](https://etherscan.io/address/0xF5f2…) (FarmRegistry) + - `uint256 _type`: `2` + - `address[] _farms`: + - [`0x79e1…971f`](https://etherscan.io/address/0x79e1…) + +## Analysis + +``` + +**Call Flow is built in Python, not asked of the LLM** — it comes straight from the +decoded calldata (`CallEntry` per call: target, signature, ABI parameter names, ETH +value, nested `bytes` payloads unwrapped up to `MAX_BYTES_RECURSION_DEPTH`), so it +can't be hallucinated, re-ordered, or summarized away. Arrays and tuple/struct +arguments are decomposed recursively (`array_element_type` / `tuple_component_types`), +so an address inside a `MarketParams`-style struct is still rendered as a link and +still reaches label lookup and the Address Links section — `iter_address_values()` +walks the same type structure for collection. Every address is rendered +full-length (never truncated) as a link to the chain's explorer from +`EXPLORER_URLS`, annotated with its contract label / token symbol when known; +chains with no configured explorer degrade to plain code spans. + +`format_address_links_block()` reuses the same renderer to give the LLM the exact +markdown link for each address in the transaction — that's what makes the +"always hyperlink addresses" rule reliable in the generated analysis. + +The header's **Contract** line links to `ReportContext.label_address`, which +defaults to the executing timelock/Safe (`from_address`). Safe multisend batches +label the *utility* contract instead, so `_explain_safe_tx()` passes the outer +target as `label_address` in that path. ## Configuration @@ -317,6 +376,7 @@ utils/llm/ ├── base.py # Abstract LLMProvider base class + LLMError ├── factory.py # Provider factory with env-based config + singleton ├── openai_compat.py # OpenAI-compatible provider (Venice, OpenAI, etc.) +├── report.py # Gist report: metadata header + deterministic call flow + analysis └── README.md # This file utils/source_context.py # Etherscan v2 source fetch + natspec extractor + proxy follow diff --git a/utils/llm/ai_explainer.py b/utils/llm/ai_explainer.py index c76eac1..71c6b23 100644 --- a/utils/llm/ai_explainer.py +++ b/utils/llm/ai_explainer.py @@ -12,11 +12,19 @@ from eth_utils import function_signature_to_4byte_selector, to_checksum_address -from utils.calldata.decoder import DecodedCall, decode_calldata, is_selector_resolvable_offline +from utils.calldata.decoder import MAX_BYTES_RECURSION_DEPTH, DecodedCall, decode_calldata, try_decode_inner_calldata from utils.erc20_metadata import fetch_erc20_metadata from utils.impl_diff import diff_implementations, format_impl_diff from utils.llm import get_llm_provider from utils.llm.base import LLMError, LLMProvider +from utils.llm.report import ( + CallEntry, + ReportContext, + build_report, + build_title, + format_address_links_block, + iter_address_values, +) from utils.logger import get_logger from utils.on_chain_state import StateRead, format_state_reads, read_before_state from utils.proxy import build_diff_url, detect_proxy_upgrade, get_current_implementation @@ -44,12 +52,14 @@ Start with a verb describing the effect. Do NOT open with "This transaction", "The proposal", or similar — the reader already knows what kind of tx this is. End with a risk tag in caps: LOW / MEDIUM / HIGH / CRITICAL. +Plain text only — the TLDR goes to a chat client, so no markdown links, no URLs, and +prefer contract names over raw addresses. Good example: "Lowers swap fee 30→25 bps on USDC/USDT pool. Marginal LP revenue cut. LOW." Bad (too terse, drops impact): "Adds farm. LOW." Bad (preamble + run-on): "This governance transaction adjusts the swap fee parameter on the USDC/USDT pool from 30 basis points to 25 basis points, which slightly reduces revenue for liquidity providers. Risk is LOW." -DETAIL: thorough analysis covering: +DETAIL: thorough analysis rendered as markdown (it is published as a web page), covering: - What each call does and why - Parameter values and their significance (use Current State section if present to compute deltas) - Asset/token flow changes @@ -57,6 +67,17 @@ - Risk assessment with explicit reasoning - Any concerns or notable observations +Address hyperlink rule (applies to DETAIL only): +- EVERY address you mention must be a markdown link to the block explorer, never a + bare or truncated address. Write [`0xFullChecksumAddress`](explorer-url) — full + address as the link text. +- An Address Links section is provided with the exact markdown for each address in + this transaction. Copy those lines verbatim; never assemble an explorer URL yourself + and never guess which explorer a chain uses. +- If an address is not in that section, write the full address in backticks unlinked. +- The report already contains a code-generated Call Flow listing every call and + argument, so do not re-list the raw calldata — explain what it means. + Critical rules for parameter interpretation: - Do NOT assume the semantic meaning of a parameter from its function name. DeFi protocols use inverted or non-standard conventions (a "maxSlippage" may be a min-output ratio; @@ -121,6 +142,8 @@ # (not a period) may precede the tag, so the preceding sentence's period is # preserved: "…vault. LOW." → "…vault." _TRAILING_RISK_TAG_RE = re.compile(r"\s*\b(?:" + "|".join(_RISK_TAGS) + r")\b[\s.]*$", re.IGNORECASE) +# Same match, but capturing, so the report header and gist title can name the risk. +_TRAILING_RISK_TAG_CAPTURE_RE = re.compile(r"\b(" + "|".join(_RISK_TAGS) + r")\b[\s.]*$", re.IGNORECASE) DETAIL_REPORT_TITLE = "AI Transaction Analysis" # JSON Schema for stage 1 (summary + risk_tag only). risk_tag is enum-constrained so @@ -143,12 +166,17 @@ {summary} -Write ONLY the thorough DETAIL analysis now. Cover what each call does and why, -parameter values and significance, asset/token flow, state changes, and an explicit +Write ONLY the thorough DETAIL analysis now, as markdown. Cover what each call does and +why, parameter values and significance, asset/token flow, state changes, and an explicit risk rationale. It MUST stay fully consistent with the TLDR above — same magnitudes, same risk level. Do not contradict its numbers or verdict and do not restate it -verbatim; expand on the reasoning. Output the detail text directly, with no "TLDR:" -or "DETAIL:" header and no trailing risk tag.""" +verbatim; expand on the reasoning. + +Render every address as a markdown explorer link, copying the exact line from the +Address Links section (full address as the link text) — never a bare or shortened +address. Use `###` for any sub-headings; `#` and `##` are reserved for the report's +own structure. Output the detail text directly, with no "TLDR:" or "DETAIL:" header +and no trailing risk tag.""" # Self-critique runs on the summary alone (stage 1), before the detail is expanded — # the summary is authoritative, so it's the artifact worth refining. Detail-specific @@ -195,10 +223,20 @@ @dataclass(frozen=True) class Explanation: - """AI-generated transaction explanation with short and detailed versions.""" + """AI-generated transaction explanation with short and detailed versions. + + ``report`` is the full markdown page published to Wavey Gist — the detail + wrapped in metadata, the summary, and the code-built call flow. It's empty + when the explanation was produced without report context (or generation + failed), in which case the bare ``detail`` is published instead. + ``title`` names that page (contract, timestamp, risk); it falls back to + ``DETAIL_REPORT_TITLE`` when unset. + """ summary: str detail: str + report: str = "" + title: str = "" def _collect_state_reads( @@ -456,57 +494,76 @@ def _annotate_address(addr: str, labels: dict[str, str]) -> str: def _extract_address_args(decoded: DecodedCall, _depth: int = 0) -> list[str]: - """All address-typed argument values (scalars and arrays) for one decoded call. + """All address-typed argument values for one decoded call. + + Covers scalars, arrays, and addresses nested inside tuple/struct arguments + (``iter_address_values`` walks the type), so a ``MarketParams``-style struct + still yields its addresses for label lookup and the Address Links section. Recurses into ``bytes`` parameters that hold nested calldata, capped at - ``_MAX_BYTES_RECURSION_DEPTH``, so labels are also collected for inner + ``MAX_BYTES_RECURSION_DEPTH``, so labels are also collected for inner calls (e.g. addresses passed to an ``upgradeToAndCall`` initializer). """ out: list[str] = [] for type_str, value in decoded.params: - if type_str == "address" and isinstance(value, str): - out.append(value) - elif type_str.startswith("address[") and isinstance(value, (list, tuple)): - out.extend(v for v in value if isinstance(v, str)) - elif type_str == "bytes" and _depth < _MAX_BYTES_RECURSION_DEPTH: - inner = _try_decode_inner_bytes(value) + out.extend(iter_address_values(type_str, value)) + if type_str == "bytes" and _depth < MAX_BYTES_RECURSION_DEPTH: + inner = try_decode_inner_calldata(value) if inner: out.extend(_extract_address_args(inner, _depth + 1)) return out -def _collect_address_labels( - targets_and_calls: list[tuple[str, DecodedCall]], - chain_id: int, -) -> dict[str, str]: - """Look up `{checksum_address: contract_name}` for every relevant address. +def collect_unique_addresses(targets_and_calls: list[tuple[str, DecodedCall]]) -> list[str]: + """Every distinct non-zero address in the transaction, checksummed, in first-seen order. - Includes each call's own target (so the prompt can annotate the - ``Target: …`` line — especially useful when the target is an ERC20 - and the decimals matter) plus every address-typed argument. Lookups - run concurrently so a batch alert with N distinct addresses doesn't - pay N × ~3s serially. Best-effort: any lookup failure is silently dropped. + Covers each call's own target plus every address-typed argument (including + those nested inside `bytes` payloads). Shared by the label lookup and the + prompt's Address Links section so both cover exactly the same set. """ seen: set[str] = set() - candidates: list[str] = [] # checksum addresses to look up + out: list[str] = [] def _consider(raw: str) -> None: + if not isinstance(raw, str): + return addr_lower = raw.lower() if addr_lower in seen: return seen.add(addr_lower) - if len(addr_lower) != 42 or int(addr_lower, 16) == 0: + if len(addr_lower) != 42: + return + try: + if int(addr_lower, 16) == 0: + return + except ValueError: return checksum = _checksum_or_none(raw) if checksum is None: return - candidates.append(checksum) + out.append(checksum) for target, decoded in targets_and_calls: if target: _consider(target) for raw in _extract_address_args(decoded): _consider(raw) + return out + + +def _collect_address_labels( + targets_and_calls: list[tuple[str, DecodedCall]], + chain_id: int, +) -> dict[str, str]: + """Look up `{checksum_address: contract_name}` for every relevant address. + + Includes each call's own target (so the prompt can annotate the + ``Target: …`` line — especially useful when the target is an ERC20 + and the decimals matter) plus every address-typed argument. Lookups + run concurrently so a batch alert with N distinct addresses doesn't + pay N × ~3s serially. Best-effort: any lookup failure is silently dropped. + """ + candidates = collect_unique_addresses(targets_and_calls) def fetch(checksum: str) -> tuple[str, str] | None: try: @@ -532,50 +589,6 @@ def fetch(checksum: str) -> tuple[str, str] | None: return {checksum: label for entry in results if entry for checksum, label in [entry]} -_MAX_BYTES_RECURSION_DEPTH = 2 - - -def _looks_like_calldata(byte_len: int) -> bool: - """True if a `bytes` blob's length matches the calldata shape (selector + ABI words). - - Real calldata is either a bare 4-byte selector (e.g. `pause()`) or - selector + N 32-byte words. Anything else — packed Safe `signatures` - blobs, EIP-712 hashes, Universal-Router-style packed paths — fails - this check and is left as opaque hex. - """ - return byte_len == 4 or (byte_len >= 36 and (byte_len - 4) % 32 == 0) - - -def _try_decode_inner_bytes(value: object) -> DecodedCall | None: - """If ``value`` looks like calldata for an offline-known function, decode it. - - Gated on (1) length matching the calldata shape and (2) the selector - being resolvable without a network call. Without these guards we'd - spam the Sourcify 4byte API on every `signatures`/hash/packed-bytes - parameter, paying a 30s timeout each miss to maybe get a false positive. - """ - if isinstance(value, bytes): - raw_len = len(value) - if not _looks_like_calldata(raw_len): - return None - hex_str = "0x" + value.hex() - elif isinstance(value, str): - hex_str = value if value.startswith("0x") else "0x" + value - # Each hex char is 4 bits, so byte_len = (len(hex_str) - 2) // 2. - if len(hex_str) < 10 or not _looks_like_calldata((len(hex_str) - 2) // 2): - return None - else: - return None - - if not is_selector_resolvable_offline(hex_str[:10]): - return None - - try: - return decode_calldata(hex_str) - except (ValueError, TypeError): - return None - - def _collect_risk_anchors(decoded_calls: list[DecodedCall]) -> str: """Build the Risk Anchors prompt section for calls with known anchors. @@ -675,8 +688,8 @@ def _format_decoded_calls( else: lines.append(f"{_indent} {label}:") lines.extend(f"{_indent} - {_annotate_address(v, labels)}" for v in value) - elif type_str == "bytes" and _depth < _MAX_BYTES_RECURSION_DEPTH: - inner = _try_decode_inner_bytes(value) + elif type_str == "bytes" and _depth < MAX_BYTES_RECURSION_DEPTH: + inner = try_decode_inner_calldata(value) if inner: lines.append(f"{_indent} {label}: ↳") lines.append(_format_decoded_calls([inner], labels, _depth=_depth + 1, _indent=nested_indent)) @@ -833,6 +846,7 @@ def _build_prompt( param_names_per_call: list[list[str] | None] | None = None, safety_notes: list[str] | None = None, description: str = "", + address_links: str = "", ) -> str: """Build the user prompt for the LLM (per-transaction context only). @@ -863,6 +877,13 @@ def _build_prompt( f"\n--- Decoded Calldata ---\n{_format_decoded_calls(decoded_calls, address_labels, param_names_per_call)}" ) + if address_links: + parts.append( + "\n--- Address Links (use these exact markdown links in the detailed report) ---\n" + "Copy a line verbatim whenever you mention the address; never write a bare, " + "shortened, or self-assembled address link.\n" + address_links + ) + constants_note = _format_batch_param_constants(decoded_calls) if constants_note: parts.append(f"\n--- Shared Across Batch ---\n{constants_note}") @@ -906,8 +927,15 @@ def _marker_pattern(keyword: str) -> "re.Pattern[str]": """Compile (and cache) the section-marker regex for ``keyword``. Handles variations: 'KEYWORD:', '## KEYWORD', '**KEYWORD**', '**KEYWORD:**', etc. + + The keyword must not be followed by another word character: a detail that + opens with a heading like ``## Detailed Analysis`` used to match on + "Detail" and get sliced down to "ed Analysis". """ - return re.compile(rf"(?:^|\n)\s*(?:#{{1,4}}\s+)?(?:\*{{2}})?{keyword}(?:\*{{2}})?[:\s]*", re.IGNORECASE) + return re.compile( + rf"(?:^|\n)\s*(?:#{{1,4}}\s+)?(?:\*{{2}})?{keyword}(?![A-Za-z0-9_])(?:\*{{2}})?[:\s]*", + re.IGNORECASE, + ) def _find_marker(text: str, keyword: str) -> tuple[int, int]: @@ -959,6 +987,18 @@ def _strip_trailing_risk_tag(text: str) -> str: return _TRAILING_RISK_TAG_RE.sub("", text).rstrip() +def _split_risk_tag(summary: str) -> tuple[str, str]: + """Split a summary into (prose without the trailing tag, uppercase tag). + + The tag is returned separately so the report can show it as a header field + instead of a word dangling off the last sentence. Tag is "" when absent. + """ + match = _TRAILING_RISK_TAG_CAPTURE_RE.search(summary) + if not match: + return summary.strip(), "" + return _strip_trailing_risk_tag(summary), match.group(1).upper() + + def _explanation_from_json(data: dict) -> Explanation: """Build an Explanation from a structured-output object. @@ -1066,7 +1106,12 @@ def _expand_detail(provider: LLMProvider, prompt: str, summary: str) -> str: return parsed.detail or raw.strip() -def _generate_explanation(provider: LLMProvider, prompt: str, refine: bool = False) -> Explanation: +def _generate_explanation( + provider: LLMProvider, + prompt: str, + refine: bool = False, + report_ctx: ReportContext | None = None, +) -> Explanation: """Two-stage generation: authoritative summary first, then a detail expanded from it. The Telegram-visible summary is the single source of truth; the linked full report @@ -1074,6 +1119,9 @@ def _generate_explanation(provider: LLMProvider, prompt: str, refine: bool = Fal artifacts can never diverge — the failure mode that showed ~50.8k in the summary while the report had the correct figure. ``refine`` adds a summary self-critique pass before expansion (~1 extra call). + + When ``report_ctx`` is given, the detail is wrapped into the full gist page + (metadata header, summary, deterministic call flow, analysis). """ summary_draft = _generate_summary(provider, prompt) if not summary_draft.summary: @@ -1089,7 +1137,14 @@ def _generate_explanation(provider: LLMProvider, prompt: str, refine: bool = Fal detail = summary_draft.detail if not detail and provider.supports_structured_output: detail = _expand_detail(provider, prompt, summary_draft.summary) - return Explanation(summary=summary_draft.summary, detail=detail) + + report = "" + title = "" + if detail and report_ctx is not None: + prose, risk_tag = _split_risk_tag(summary_draft.summary) + report = build_report(prose, detail, report_ctx, risk_tag) + title = build_title(report_ctx, risk_tag, fallback=DETAIL_REPORT_TITLE) + return Explanation(summary=summary_draft.summary, detail=detail, report=report, title=title) def explain_transaction( @@ -1104,6 +1159,7 @@ def explain_transaction( context_note: str = "", refine: bool = True, description: str = "", + label_address: str = "", ) -> Explanation | None: """Generate an AI explanation for a governance transaction. @@ -1130,6 +1186,10 @@ def explain_transaction( description: Optional proposer-supplied description of intent. When set, the LLM compares stated intent against the decoded actions and flags any divergence. + label_address: Address that ``label`` names, linked from the report's + Contract header. Defaults to ``from_address`` — pass it explicitly + when the label describes something else (e.g. a Safe multisend + utility rather than the Safe itself). Returns: Explanation with summary and detail, or None on failure. @@ -1172,6 +1232,8 @@ def explain_transaction( else: logger.info("Simulation unavailable, proceeding with decoded calldata only") + address_links = format_address_links_block(collect_unique_addresses([(target, decoded)]), chain_id, address_labels) + prompt = _build_prompt( target=target, value=value, @@ -1188,12 +1250,23 @@ def explain_transaction( param_names_per_call=param_names, safety_notes=safety_notes, description=description, + address_links=address_links, ) logger.info("Full AI context for %s:\n%s", target, prompt) + report_ctx = ReportContext( + entries=[CallEntry(target=target, call=decoded, value=value, param_names=param_names[0])], + chain_id=chain_id, + labels=address_labels, + protocol=protocol, + label=label, + from_address=from_address, + label_address=label_address or from_address, + ) + try: provider = get_llm_provider() - explanation = _generate_explanation(provider, prompt, refine=refine) + explanation = _generate_explanation(provider, prompt, refine=refine, report_ctx=report_ctx) logger.info("AI summary using %s:\n%s", provider.model_name, explanation.summary) if explanation.detail: logger.info("AI detail:\n%s", explanation.detail) @@ -1213,6 +1286,7 @@ def explain_batch_transaction( context_note: str = "", refine: bool = True, description: str = "", + label_address: str = "", ) -> Explanation | None: """Generate an AI explanation for a batch/multicall governance transaction. @@ -1233,6 +1307,10 @@ def explain_batch_transaction( description: Optional proposer-supplied description of intent. When set, the LLM compares stated intent against the decoded actions and flags any divergence. + label_address: Address that ``label`` names, linked from the report's + Contract header. Defaults to ``from_address`` — pass it explicitly + when the label describes something else (e.g. a Safe multisend + utility rather than the Safe itself). Returns: Explanation with summary and detail, or None on failure. @@ -1292,6 +1370,7 @@ def explain_batch_transaction( targets = ", ".join(c.get("target", "?") for c in calls) total_value = sum(int(c.get("value", "0")) for c in calls) + address_links = format_address_links_block(collect_unique_addresses(decoded_with_target), chain_id, address_labels) prompt = _build_prompt( target=targets, @@ -1309,12 +1388,26 @@ def explain_batch_transaction( param_names_per_call=param_names, safety_notes=safety_notes, description=description, + address_links=address_links, ) logger.info("Full AI context for batch (%s calls):\n%s", len(calls), prompt) + report_ctx = ReportContext( + entries=[ + CallEntry(target=tgt, call=call, value=val, param_names=names) + for (tgt, call, val), names in zip(targets_calls_values, param_names) + ], + chain_id=chain_id, + labels=address_labels, + protocol=protocol, + label=label, + from_address=from_address, + label_address=label_address or from_address, + ) + try: provider = get_llm_provider() - explanation = _generate_explanation(provider, prompt, refine=refine) + explanation = _generate_explanation(provider, prompt, refine=refine, report_ctx=report_ctx) logger.info("Batch AI summary using %s:\n%s", provider.model_name, explanation.summary) if explanation.detail: logger.info("Batch AI detail:\n%s", explanation.detail) @@ -1327,12 +1420,16 @@ def explain_batch_transaction( def format_explanation_line(explanation: Explanation) -> str: """Format the AI explanation for inclusion in a Telegram alert message. - Uses the short summary for the Telegram message. The detailed analysis - is uploaded to Wavey Gist for easy access. + Uses the short summary for the Telegram message. The full report — metadata, + summary, call flow, and analysis — is uploaded to Wavey Gist and linked. The + bare detail is published instead when no report was built (explanations + generated without report context). """ line = f"\n🤖 *AI Summary:*\n{escape_markdown(explanation.summary)}" if explanation.detail: - detail_url = upload_to_gist(explanation.detail, title=DETAIL_REPORT_TITLE) + detail_url = upload_to_gist( + explanation.report or explanation.detail, title=explanation.title or DETAIL_REPORT_TITLE + ) if detail_url: line += f"\n[Full details]({detail_url})" else: diff --git a/utils/llm/report.py b/utils/llm/report.py new file mode 100644 index 0000000..3fa0e05 --- /dev/null +++ b/utils/llm/report.py @@ -0,0 +1,340 @@ +"""Render the full transaction report published to Wavey Gist. + +The Telegram alert only carries the short AI summary; the linked gist is the +full artifact a reviewer opens. It pairs the LLM's analysis with a +deterministic, code-built **call flow** — the exact function each call hits, +its arguments, and every address rendered as a block-explorer hyperlink. + +The call flow is built here rather than asked of the LLM on purpose: it is +ground truth straight from the decoded calldata, so it can't be hallucinated, +mis-ordered, or summarized away. +""" + +import re +from collections.abc import Iterator +from dataclasses import dataclass, field +from datetime import datetime, timezone + +from eth_utils import to_checksum_address + +from utils.calldata.decoder import ( + MAX_BYTES_RECURSION_DEPTH, + DecodedCall, + split_top_level_types, + try_decode_inner_calldata, +) +from utils.chains import EXPLORER_URLS, Chain + +ZERO_ADDRESS = "0x0000000000000000000000000000000000000000" + +# The report already opens the section with "## Analysis", so a detail that +# starts with its own "Detailed Analysis" heading would double up. +_REDUNDANT_ANALYSIS_HEADING_RE = re.compile( + r"^\s*(?:#{1,6}\s*)?\**\s*(?:detailed|full|in-depth)?\s*analysis\s*\**\s*:?\s*\n+", + re.IGNORECASE, +) + + +@dataclass(frozen=True) +class CallEntry: + """One decoded call in the transaction, with the context needed to render it.""" + + target: str + call: DecodedCall + value: int = 0 + param_names: list[str] | None = None + + +@dataclass(frozen=True) +class ReportContext: + """Everything the gist report needs beyond the LLM's summary and detail.""" + + entries: list[CallEntry] = field(default_factory=list) + chain_id: int = 0 + labels: dict[str, str] = field(default_factory=dict) + protocol: str = "" + label: str = "" + from_address: str = "" + # Address the ``label`` names — usually the executing timelock/Safe, but a + # Safe multisend batch labels the utility contract instead. Linked from the + # report's Contract header line. + label_address: str = "" + + +def checksum_or_none(addr: object) -> str | None: + """Return the checksummed address, or None if ``addr`` isn't a hex address.""" + if not isinstance(addr, str) or not addr.startswith("0x"): + return None + try: + return to_checksum_address(addr) + except ValueError: + return None + + +def explorer_address_url(chain_id: int, address: str) -> str: + """Block-explorer address URL, or "" when the chain has no configured explorer.""" + explorer = EXPLORER_URLS.get(chain_id) + checksum = checksum_or_none(address) + if not explorer or checksum is None: + return "" + return f"{explorer}/address/{checksum}" + + +def address_link(address: str, chain_id: int, labels: dict[str, str] | None = None) -> str: + """Render an address as a markdown explorer link, suffixed with its label. + + Full addresses are always shown (never truncated) so the reader can copy + and verify them. Falls back to plain text on chains with no explorer, and + returns the input unchanged when it isn't a parseable address. + """ + checksum = checksum_or_none(address) + if checksum is None: + return str(address) + label = (labels or {}).get(checksum) + url = explorer_address_url(chain_id, checksum) + rendered = f"[`{checksum}`]({url})" if url else f"`{checksum}`" + return f"{rendered} ({label})" if label else rendered + + +def format_address_links_block(addresses: list[str], chain_id: int, labels: dict[str, str] | None = None) -> str: + """Prompt section listing the exact markdown link to use for each address. + + Handing the LLM ready-made links is what makes the "always hyperlink + addresses" rule reliable — it copies a line instead of assembling an + explorer URL from memory (and picking the wrong chain's explorer). + Returns "" when there is nothing to link. + """ + lines: list[str] = [] + for addr in addresses: + rendered = address_link(addr, chain_id, labels) + if rendered.startswith("["): # only useful when an explorer link was produced + lines.append(f"- {rendered}") + return "\n".join(lines) + + +def _format_param_value(type_str: str, value: object, chain_id: int, labels: dict[str, str]) -> str: + """Render a single scalar parameter value for the markdown call flow.""" + if type_str == "address" and isinstance(value, str): + return address_link(value, chain_id, labels) + if isinstance(value, bytes): + return f"`0x{value.hex()}`" + if isinstance(value, int) and not isinstance(value, bool): + return f"`{value:,}`" + return f"`{value}`" + + +def _param_label(type_str: str, name: str | None) -> str: + """Render a Solidity-style ``type name`` declaration, falling back to bare type.""" + return f"`{type_str} {name}`" if name else f"`{type_str}`" + + +def array_element_type(type_str: str) -> str | None: + """``T[]`` / ``T[3]`` → ``T``; None when the type isn't an array.""" + if not type_str.endswith("]"): + return None + open_idx = type_str.rfind("[") + return type_str[:open_idx] if open_idx > 0 else None + + +def tuple_component_types(type_str: str) -> list[str] | None: + """``(address,uint256)`` → ``["address", "uint256"]``; None when not a tuple.""" + if not (type_str.startswith("(") and type_str.endswith(")")): + return None + inner = type_str[1:-1].strip() + return split_top_level_types(inner) if inner else [] + + +def _is_composite(type_str: str) -> bool: + """True for array and tuple types, which render as nested bullets.""" + return array_element_type(type_str) is not None or tuple_component_types(type_str) is not None + + +def iter_address_values(type_str: str, value: object) -> Iterator[str]: + """Yield every ``address`` leaf inside a decoded parameter value. + + Walks arrays and tuples (and their nesting) so a struct argument like + ``(address,address,uint256)`` contributes its addresses to label lookup + and to the prompt's Address Links section — without this they'd only ever + be stringified into the report as raw, unlinked text. + """ + element = array_element_type(type_str) + if element is not None and isinstance(value, (list, tuple)): + for item in value: + yield from iter_address_values(element, item) + return + components = tuple_component_types(type_str) + if components is not None and isinstance(value, (list, tuple)) and len(components) == len(value): + for component, item in zip(components, value): + yield from iter_address_values(component, item) + return + if type_str == "address" and isinstance(value, str): + yield value + + +def _render_param( + label: str, + type_str: str, + value: object, + chain_id: int, + labels: dict[str, str], + indent: str, + depth: int = 0, +) -> list[str]: + """Render one parameter, expanding arrays and tuples into nested bullets. + + Composites recurse so addresses nested in a struct or an array of structs + still come out as explorer links rather than a stringified Python tuple. + Recursion terminates on the type string, which is finite. + """ + element = array_element_type(type_str) + if element is not None and isinstance(value, (list, tuple)): + if not value: + return [f"{indent}- {label}: _(empty)_"] + lines = [f"{indent}- {label}:"] + for i, item in enumerate(value): + if _is_composite(element): + lines.extend(_render_param(f"`[{i}]`", element, item, chain_id, labels, indent + " ", depth)) + else: + lines.append(f"{indent} - {_format_param_value(element, item, chain_id, labels)}") + return lines + + components = tuple_component_types(type_str) + if components is not None and isinstance(value, (list, tuple)) and len(components) == len(value): + if not components: + return [f"{indent}- {label}: _(empty)_"] + lines = [f"{indent}- {label}:"] + for component, item in zip(components, value): + lines.extend(_render_param(f"`{component}`", component, item, chain_id, labels, indent + " ", depth)) + return lines + + if type_str == "bytes" and depth < MAX_BYTES_RECURSION_DEPTH: + inner = try_decode_inner_calldata(value) + if inner is not None: + lines = [f"{indent}- {label}: ↳ `{inner.signature}`"] + lines.extend(_format_params(inner, chain_id, labels, None, indent + " ", depth + 1)) + return lines + + return [f"{indent}- {label}: {_format_param_value(type_str, value, chain_id, labels)}"] + + +def _format_params( + call: DecodedCall, + chain_id: int, + labels: dict[str, str], + param_names: list[str] | None, + indent: str, + depth: int = 0, +) -> list[str]: + """Render a call's parameters as an indented markdown bullet list.""" + lines: list[str] = [] + for i, (type_str, value) in enumerate(call.params): + name = param_names[i] if param_names is not None and i < len(param_names) else None + lines.extend(_render_param(_param_label(type_str, name), type_str, value, chain_id, labels, indent, depth)) + return lines + + +def format_call_flow(ctx: ReportContext) -> str: + """Render the decoded calls as a numbered markdown flow with explorer links. + + Returns "" when there is nothing to render. + """ + if not ctx.entries: + return "" + + lines: list[str] = [] + sender = checksum_or_none(ctx.from_address) + if sender and sender != ZERO_ADDRESS: + lines.append(f"**From:** {address_link(sender, ctx.chain_id, ctx.labels)}") + lines.append("") + + for i, entry in enumerate(ctx.entries, start=1): + target = address_link(entry.target, ctx.chain_id, ctx.labels) if entry.target else "_unknown target_" + lines.append(f"{i}. **`{entry.call.signature}`** on {target}") + if entry.value > 0: + lines.append(f" - **ETH value:** `{entry.value / 1e18:.6f}` ETH") + param_lines = _format_params(entry.call, ctx.chain_id, ctx.labels, entry.param_names, indent=" ") + lines.extend(param_lines or [" - _no inputs_"]) + lines.append("") + + return "\n".join(lines).rstrip() + + +def _chain_name(chain_id: int) -> str: + try: + return Chain.from_chain_id(chain_id).network_name.capitalize() + except ValueError: + return f"Chain {chain_id}" + + +def _format_metadata(ctx: ReportContext, risk_tag: str) -> str: + """Header bullet list: protocol, contract label, chain, risk.""" + lines: list[str] = [] + if ctx.protocol: + lines.append(f"- **Protocol:** {ctx.protocol}") + if ctx.label: + # Link the label to the contract it names, so the header itself is + # clickable rather than a bare name the reader has to go look up. + linked = address_link(ctx.label_address, ctx.chain_id) if ctx.label_address else "" + lines.append( + f"- **Contract:** {ctx.label} — {linked}" if linked.startswith("[") else f"- **Contract:** {ctx.label}" + ) + if ctx.chain_id: + lines.append(f"- **Chain:** {_chain_name(ctx.chain_id)} (chain id {ctx.chain_id})") + if risk_tag: + lines.append(f"- **Risk:** {risk_tag}") + return "\n".join(lines) + + +def build_title(ctx: ReportContext, risk_tag: str = "", now: datetime | None = None, fallback: str = "") -> str: + """Gist title: `` -
- ``. + + Naming the contract and the time makes a list of gists scannable — the + previous constant title left every report looking identical. The timestamp + is UTC (the runners' clock); ``now`` is injectable for tests. + + Args: + ctx: Report context; its ``label`` (else ``protocol``) names the report. + risk_tag: LOW / MEDIUM / HIGH / CRITICAL, appended when known. + now: Timestamp to render. Defaults to the current UTC time. + fallback: Name to use when the context has neither label nor protocol. + + Returns: + The title string; never empty as long as ``fallback`` is set. + """ + stamp = (now or datetime.now(timezone.utc)).strftime("%d/%m/%Y %H:%M") + parts = [ctx.label or ctx.protocol or fallback, stamp] + if risk_tag: + parts.append(risk_tag) + return " - ".join(part for part in parts if part) + + +def build_report(summary: str, detail: str, ctx: ReportContext, risk_tag: str = "") -> str: + """Assemble the full markdown gist body. + + Sections: metadata header, the Telegram-visible summary (so the gist is + self-contained), the deterministic call flow, and the LLM's analysis. + + Args: + summary: The authoritative TLDR, risk tag already stripped by the caller. + detail: The LLM's detailed analysis. + ctx: Decoded calls, labels, and alert metadata. + risk_tag: LOW / MEDIUM / HIGH / CRITICAL, when known. + + Returns: + Markdown body, or "" when there is nothing worth publishing. + """ + if not detail and not summary: + return "" + + sections: list[str] = [] + metadata = _format_metadata(ctx, risk_tag) + if metadata: + sections.append(metadata) + if summary: + sections.append(f"## Summary\n\n{summary}") + call_flow = format_call_flow(ctx) + if call_flow: + sections.append(f"## Call Flow\n\n{call_flow}") + if detail: + sections.append(f"## Analysis\n\n{_REDUNDANT_ANALYSIS_HEADING_RE.sub('', detail)}") + return "\n\n".join(sections)