fix(llm): decode Yearn Executor execute(bytes) scripts

tests/test_executor_script.py

@@ -0,0 +1,176 @@
+"""Tests for decoding Yearn governance Executor ``execute(bytes)`` scripts.
+
+The script packs each call as a 32-byte ``target | calldata size`` header plus
+the calldata, so it is not ABI-encoded and the generic nested-bytes decoding
+can't see inside it. Fixture: yChad nonce 2299, which had the Executor accept
+management of five reward distributors and the Funding Distributor.
+"""
+
+import unittest
+from unittest.mock import patch
+
+from eth_abi import encode
+from eth_utils import to_checksum_address as _cs
+
+from utils.calldata.decoder import DecodedCall, ScriptCall, decode_calldata
+from utils.calldata.wrappers import InnerCall, split_packed_script, unwrap_calls, unwrap_executed_calls
+from utils.llm.ai_explainer import _extract_address_args, _format_decoded_calls
+from utils.llm.report import _format_params, _iter_reference_arguments
+
+ACCEPT_MANAGEMENT = "0x759be10c"
+FUNDING_DISTRIBUTOR = _cs("0xbcc932e4750c3e465a7e54a06a34f9edf8f6116b")
+TARGETS = [
+    _cs("0x95547ede56cf74b73dd78a37f547127dffda6113"),
+    _cs("0x952b31960c97e76362ac340d07d183ada15e3d6e"),
+    _cs("0x2548bf65916fdabb5a5673fc4225011ff29ee884"),
+    _cs("0x7efc3953bed2fc20b9f825ebffab1cc8b072a000"),
+    _cs("0xa139250bcb3880a97c75fd4287d5bf2a6caf98a3"),
+    FUNDING_DISTRIBUTOR,
+]
+# The `_script` argument exactly as queued in yChad nonce 2299.
+NONCE_2299_SCRIPT = bytes.fromhex(
+    "95547ede56cf74b73dd78a37f547127dffda6113000000000000000000000004759be10c"
+    "952b31960c97e76362ac340d07d183ada15e3d6e000000000000000000000004759be10c"
+    "2548bf65916fdabb5a5673fc4225011ff29ee884000000000000000000000004759be10c"
+    "7efc3953bed2fc20b9f825ebffab1cc8b072a000000000000000000000000004759be10c"
+    "a139250bcb3880a97c75fd4287d5bf2a6caf98a3000000000000000000000004759be10c"
+    "bcc932e4750c3e465a7e54a06a34f9edf8f6116b000000000000000000000004759be10c"
+)
+
+
+def script_entry(target: str, data: bytes) -> bytes:
+    """One packed call: ``target (160 bits) | size (96 bits)`` header, then calldata."""
+    return ((int(target, 16) << 96) | len(data)).to_bytes(32, "big") + data
+
+
+def execute_call(script: bytes) -> str:
+    return "0x09c5eabe" + encode(["bytes"], [script]).hex()
+
+
+class TestSplitPackedScript(unittest.TestCase):
+    def test_nonce_2299_script_is_six_accept_management_calls(self) -> None:
+        pairs = split_packed_script(NONCE_2299_SCRIPT)
+        self.assertEqual([_cs(t) for t, _ in pairs], TARGETS)
+        self.assertEqual({"0x" + d.hex() for _, d in pairs}, {ACCEPT_MANAGEMENT})
+
+    def test_variable_length_calldata(self) -> None:
+        set_management = bytes.fromhex("fd066ecc") + encode(["address"], [FUNDING_DISTRIBUTOR])
+        script = script_entry(TARGETS[0], set_management) + script_entry(TARGETS[1], b"")
+        self.assertEqual(split_packed_script(script), [(TARGETS[0].lower(), set_management), (TARGETS[1].lower(), b"")])
+
+    def test_malformed_scripts_raise(self) -> None:
+        cases = {
+            "empty": b"",
+            "truncated header": NONCE_2299_SCRIPT[:-5][:31],
+            "calldata overruns": NONCE_2299_SCRIPT[:-1],
+            "trailing bytes": NONCE_2299_SCRIPT + b"\x00",
+            "zero target": script_entry("0x0", b"\x75\x9b\xe1\x0c"),
+            "too many calls": script_entry(TARGETS[0], b"") * 65,
+        }
+        for name, script in cases.items():
+            with self.subTest(name), self.assertRaises(ValueError):
+                split_packed_script(script)
+
+    def test_abi_encoded_calldata_is_not_a_script(self) -> None:
+        abi_payload = bytes.fromhex("fd066ecc") + encode(["address"], [FUNDING_DISTRIBUTOR])
+        with self.assertRaises(ValueError):
+            split_packed_script(abi_payload)
+
+
+class TestUnwrapExecutorScript(unittest.TestCase):
+    def test_execute_runs_its_script_calls_now(self) -> None:
+        expected = [InnerCall(t, ACCEPT_MANAGEMENT, f"execute call {i}") for i, t in enumerate(TARGETS, start=1)]
+        self.assertEqual(unwrap_calls(execute_call(NONCE_2299_SCRIPT)), expected)
+        self.assertEqual(unwrap_executed_calls(execute_call(NONCE_2299_SCRIPT)), expected)
+
+    def test_execute_bytes_that_is_not_a_script_unwraps_to_nothing(self) -> None:
+        self.assertEqual(unwrap_calls(execute_call(b"\x01\x02\x03")), [])
+
+
+class TestDecodeScriptCalls(unittest.TestCase):
+    def test_decoder_attaches_decoded_script_calls_offline(self) -> None:
+        with patch("utils.calldata.decoder.fetch_json", side_effect=AssertionError("network")):
+            decoded = decode_calldata(execute_call(NONCE_2299_SCRIPT))
+        assert decoded is not None
+        self.assertEqual(decoded.signature, "execute(bytes)")
+        self.assertEqual([sc.target for sc in decoded.script_calls], TARGETS)
+        self.assertEqual({sc.decoded.signature for sc in decoded.script_calls if sc.decoded}, {"accept_management()"})
+
+    def test_inner_calls_use_the_inner_target_abi_when_chain_is_known(self) -> None:
+        with patch("utils.calldata.decoder._resolve_signature_via_abi", return_value="accept_management()") as resolve:
+            decode_calldata(execute_call(script_entry(TARGETS[0], bytes.fromhex("759be10c"))), chain_id=1)
+        self.assertIn((1, TARGETS[0], ACCEPT_MANAGEMENT), [c.args for c in resolve.call_args_list])
+
+    def test_unknown_inner_selector_stays_undecoded_without_network(self) -> None:
+        script = script_entry(TARGETS[0], bytes.fromhex("12345678"))
+        with (
+            patch("utils.calldata.decoder.is_selector_resolvable_offline", side_effect=lambda s: s == "0x09c5eabe"),
+            patch("utils.calldata.decoder.fetch_json", side_effect=AssertionError("network")),
+        ):
+            decoded = decode_calldata(execute_call(script))
+        assert decoded is not None
+        self.assertEqual(decoded.script_calls, (ScriptCall(TARGETS[0], "0x12345678", None),))
+
+    def test_other_calls_have_no_script_calls(self) -> None:
+        decoded = decode_calldata("0x759be10c")
+        assert decoded is not None
+        self.assertEqual(decoded.script_calls, ())
+
+
+class TestScriptRendering(unittest.TestCase):
+    def setUp(self) -> None:
+        decoded = decode_calldata(execute_call(NONCE_2299_SCRIPT))
+        assert decoded is not None
+        self.decoded = decoded
+
+    def test_prompt_lists_each_script_call_with_its_target(self) -> None:
+        rendered = _format_decoded_calls([self.decoded], {FUNDING_DISTRIBUTOR: "Funding Distributor"}, [["_script"]])
+        self.assertIn("Executor script, 6 calls; each is made by the Executor", rendered)
+        self.assertIn(f"Script call 6 → {FUNDING_DISTRIBUTOR} (Funding Distributor): accept_management()", rendered)
+        self.assertNotIn(NONCE_2299_SCRIPT.hex()[:40], rendered)
+
+    def test_prompt_marks_undecoded_script_calls(self) -> None:
+        call = DecodedCall(
+            self.decoded.function_name,
+            self.decoded.signature,
+            self.decoded.params,
+            (
+                ScriptCall(TARGETS[0], "0x12345678", None),
+                ScriptCall(TARGETS[1], "0x", None),
+                ScriptCall(TARGETS[2], "0x010203", None),
+            ),
+        )
+        rendered = _format_decoded_calls([call])
+        self.assertIn(f"Script call 1 → {TARGETS[0]}: UNDECODED (selector 0x12345678", rendered)
+        self.assertIn(f"Script call 2 → {TARGETS[1]}: empty calldata", rendered)
+        self.assertIn(f"Script call 3 → {TARGETS[2]}: UNDECODED (short payload 0x010203, no full selector)", rendered)
+
+    def test_report_keeps_short_payload_apart_from_empty_calldata(self) -> None:
+        call = DecodedCall(
+            self.decoded.function_name,
+            self.decoded.signature,
+            self.decoded.params,
+            (ScriptCall(TARGETS[0], "0x", None), ScriptCall(TARGETS[1], "0x010203", None)),
+        )
+        lines = _format_params(call, 1, {}, ["_script"], "")
+        self.assertTrue(lines[1].startswith("  - Script call 1: empty calldata (no function selector) on "))
+        self.assertTrue(
+            lines[2].startswith("  - Script call 2: undecoded short payload `0x010203` (no full selector) on ")
+        )
+
+    def test_report_links_each_script_target(self) -> None:
+        lines = _format_params(self.decoded, 1, {}, ["_script"], "")
+        self.assertEqual(lines[0], "- `bytes _script`: ↳ Executor script, 6 calls made by the Executor")
+        self.assertIn(
+            f"  - Script call 6: `accept_management()` on "
+            f"[`{FUNDING_DISTRIBUTOR}`](https://etherscan.io/address/{FUNDING_DISTRIBUTOR})",
+            lines,
+        )
+
+    def test_script_targets_get_labels_and_references(self) -> None:
+        self.assertEqual(_extract_address_args(self.decoded), TARGETS)
+        self.assertEqual(list(_iter_reference_arguments(self.decoded)), TARGETS)
+
+
+if __name__ == "__main__":
+    unittest.main()

utils/calldata/decoder.py

@@ -15,6 +15,7 @@
 
 from utils.cache import cache_path
 from utils.calldata.known_selectors import KNOWN_SELECTORS
+from utils.calldata.wrappers import PACKED_SCRIPT_SIGNATURE, split_packed_script
 from utils.http_client import fetch_json
 from utils.logger import get_logger
 
@@ -63,13 +64,29 @@ def _persist_selector(selector: str, signature: str | None) -> None:
 _selector_cache: dict[str, str | None] = _load_selector_cache()
 
 
+@dataclass(frozen=True)
+class ScriptCall:
+    """One call packed inside a Yearn Executor ``execute(bytes)`` script."""
+
+    target: str  # checksummed
+    data: str  # 0x-prefixed calldata
+    decoded: "DecodedCall | None"
+
+
 @dataclass(frozen=True)
 class DecodedCall:
-    """Result of decoding a calldata hex string."""
+    """Result of decoding a calldata hex string.
+
+    ``script_calls`` holds the calls packed in a Yearn Executor
+    ``execute(bytes)`` script, in execution order; empty for every other call.
+    The script is not ABI-encoded calldata, so the generic nested-``bytes``
+    decoding can't see inside it.
+    """
 
     function_name: str
     signature: str
     params: list[tuple[str, Any]] = field(default_factory=list)
+    script_calls: tuple[ScriptCall, ...] = ()
 
 
 def is_selector_resolvable_offline(selector_hex: str) -> bool:
@@ -336,7 +353,34 @@ def decode_calldata(data_hex: str, chain_id: int | None = None, target: str | No
             except Exception:
                 logger.debug("Failed to decode params for %s with types %s", selector, param_types)
 
-    return DecodedCall(function_name=func_name, signature=signature, params=params)
+    script_calls = _decode_script_calls(signature, params, chain_id)
+    return DecodedCall(function_name=func_name, signature=signature, params=params, script_calls=script_calls)
+
+
+def _decode_script_calls(signature: str, params: list[tuple[str, Any]], chain_id: int | None) -> tuple[ScriptCall, ...]:
+    """Split and decode a Yearn Executor script; () for anything else.
+
+    Each inner call is decoded against its own target's verified ABI when
+    ``chain_id`` is known. Without it, only offline-resolvable selectors are
+    decoded so an unknown selector never costs a 4byte round trip.
+    """
+    if signature != PACKED_SCRIPT_SIGNATURE or len(params) != 1 or not isinstance(params[0][1], bytes):
+        return ()
+    try:
+        pairs = split_packed_script(params[0][1])
+    except ValueError as e:
+        logger.debug("execute(bytes) argument is not an Executor script: %s", e)
+        return ()
+
+    calls: list[ScriptCall] = []
+    for target, data in pairs:
+        target = to_checksum_address(target)
+        data_hex = "0x" + data.hex()
+        decoded: DecodedCall | None = None
+        if len(data) >= 4 and (chain_id is not None or is_selector_resolvable_offline(data_hex[:10])):
+            decoded = decode_calldata(data_hex, chain_id=chain_id, target=target)
+        calls.append(ScriptCall(target=target, data=data_hex, decoded=decoded))
+    return tuple(calls)
 
 
 # How many levels of `bytes` parameters holding inner calldata we unwrap. Two is

utils/calldata/known_selectors.py

@@ -41,6 +41,11 @@
     # Maple GovernorTimelock
     "0xd9ab9270": "scheduleProposals(address[],bytes[])",
     "0x2d6e853c": "proposeRoleUpdates(bytes32[],address[],bool[])",
+    # Yearn governance Executor (Executor.vy) and the two-step management roles it takes over
+    "0x09c5eabe": "execute(bytes)",
+    "0x0fe42159": "set_operator(address,bool)",
+    "0xfd066ecc": "set_management(address)",
+    "0x759be10c": "accept_management()",
     # Safe (Gnosis)
     "0x6a761202": "execTransaction(address,uint256,bytes,uint8,uint256,uint256,uint256,address,address,bytes)",
     # Initializers — frequently appear as the inner payload of upgradeToAndCall / upgradeAndCall

utils/calldata/wrappers.py

@@ -14,6 +14,9 @@
       (the inner calldata is ``selector(signature) ++ data`` when ``signature``
       is non-empty, else ``data`` as-is)
     - Maple GovernorTimelock: ``scheduleProposals``
+    - Yearn governance Executor: ``execute(bytes)``, whose script packs each call
+      as a 32-byte header ``target (160 bits) | calldata size (96 bits)`` followed
+      by the calldata
 """
 
 from collections.abc import Callable
@@ -62,6 +65,58 @@ def _maple(values: tuple) -> list[tuple[str, bytes]]:
     return list(zip(values[0], values[1]))
 
 
+# Yearn governance Executor (Executor.vy) script limits.
+_SCRIPT_HEADER_BYTES = 32
+_SCRIPT_SIZE_BITS = 96
+_SCRIPT_MAX_CALLS = 64
+
+# Signature of the Executor entrypoint whose single ``bytes`` argument is a packed script.
+PACKED_SCRIPT_SIGNATURE = "execute(bytes)"
+
+
+def split_packed_script(script: bytes) -> list[tuple[str, bytes]]:
+    """Split a Yearn Executor script into its ``(target, calldata)`` pairs.
+
+    Mirrors ``Executor.execute``: each call is a 32-byte header holding the
+    target in the high 160 bits and the calldata size in the low 96 bits,
+    followed by that many calldata bytes. The script must be consumed exactly.
+
+    Args:
+        script: The raw ``_script`` bytes.
+
+    Returns:
+        The calls in execution order.
+
+    Raises:
+        ValueError: If the bytes are not a well-formed, non-empty script.
+    """
+    pairs: list[tuple[str, bytes]] = []
+    offset = 0
+    while offset < len(script):
+        if len(pairs) == _SCRIPT_MAX_CALLS:
+            raise ValueError(f"script has more than {_SCRIPT_MAX_CALLS} calls")
+        if offset + _SCRIPT_HEADER_BYTES > len(script):
+            raise ValueError(f"truncated script header at byte {offset}")
+        header = int.from_bytes(script[offset : offset + _SCRIPT_HEADER_BYTES], "big")
+        offset += _SCRIPT_HEADER_BYTES
+        size = header & ((1 << _SCRIPT_SIZE_BITS) - 1)
+        target = header >> _SCRIPT_SIZE_BITS
+        if target == 0:
+            raise ValueError("script call targets the zero address")
+        if offset + size > len(script):
+            raise ValueError(f"script call at byte {offset} overruns the script")
+        pairs.append(("0x" + target.to_bytes(20, "big").hex(), script[offset : offset + size]))
+        offset += size
+    if not pairs:
+        raise ValueError("empty script")
+    return pairs
+
+
+def _packed_script(values: tuple) -> list[tuple[str, bytes]]:
+    """(script,) — Yearn governance Executor.execute."""
+    return split_packed_script(values[0])
+
+
 # signature → how to pull (target, calldata) pairs out of the decoded values
 _WRAPPER_SIGNATURES: dict[str, _Extractor] = {
     "schedule(address,uint256,bytes,bytes32,bytes32,uint256)": _single,
@@ -71,6 +126,7 @@ def _maple(values: tuple) -> list[tuple[str, bytes]]:
     "queueTransaction(address,uint256,string,bytes,uint256)": _compound,
     "executeTransaction(address,uint256,string,bytes,uint256)": _compound,
     "scheduleProposals(address[],bytes[])": _maple,
+    PACKED_SCRIPT_SIGNATURE: _packed_script,
 }
 
 

utils/llm/README.md

@@ -59,6 +59,8 @@ Converts raw hex calldata into a structured `DecodedCall`:
 
 Result: `DecodedCall(function_name="upgradeTo", signature="upgradeTo(address)", params=[("address", "0x...")])`
 
+A Yearn governance Executor `execute(bytes)` call also gets `script_calls`: its `_script` is packed (a 32-byte `target | calldata size` header per call, then the calldata), not ABI-encoded, so the generic nested-`bytes` decoding can't read it. `split_packed_script` (`utils/calldata/wrappers.py`) splits it the same way `Executor.execute` does and fails on any malformed script. Each inner call is then decoded against its own target's ABI. The prompt and the report's Call Flow list each script call with its target instead of the hex. Before this, the model sliced yChad nonce 2299's script one hex digit off, reported a nonexistent selector `0x4759be10` where the script actually made six `accept_management()` calls, and rated the batch HIGH because of "unresolved downstream calls". The system prompt now forbids reading selectors or targets off raw hex.
+
 ### 2. Verified Source Context (`utils/source_context.py`)
 
 For each `(target, function)` pair, fetches the verified source via the Etherscan v2 multichain API and extracts:
@@ -150,6 +152,7 @@ For the ProxyAdmin pattern, the tx target is the ProxyAdmin and the actual proxy
 | OpenZeppelin `TimelockController` | `schedule`, `scheduleBatch`, `execute`, `executeBatch` |
 | Compound-style `Timelock` | `queueTransaction`, `executeTransaction` (inner calldata = `selector(signature) ++ data` when `signature` is set) |
 | Maple `GovernorTimelock` | `scheduleProposals` |
+| Yearn governance `Executor` | `execute(bytes)` (packed script, see Calldata Decoding) |
 
 Without this, a Safe tx that schedules or executes a timelock batch of upgrades (the CAP multisig calling `executeBatch` on its `TimelockController`) was explained with no implementation diff at all, since only the outer `executeBatch` was checked. Safe alerts (`protocols/safe/main.py`) also print the same `🅿️ Proxy` / `🔄 Upgrade` / `📊 Diff` lines as timelock alerts, for direct, multisend and wrapped upgrades, all rendered by `format_upgrade_lines`.