refactor: move calldata decoder from timelock/ to utils/calldata/

Author: spalen0

tests/test_ai_explainer.py

@@ -3,7 +3,7 @@
 import unittest
 from unittest.mock import MagicMock, patch
 
-from timelock.calldata_decoder import DecodedCall
+from utils.calldata.decoder import DecodedCall
 from utils.llm.ai_explainer import (
     Explanation,
     _build_prompt,

tests/test_calldata_decoder.py

@@ -3,7 +3,7 @@
 import unittest
 from unittest.mock import patch
 
-from timelock.calldata_decoder import (
+from utils.calldata.decoder import (
     DecodedCall,
     _format_param_value,
     _parse_param_types,
@@ -12,7 +12,7 @@
     format_call_lines,
     resolve_selector,
 )
-from timelock.known_selectors import KNOWN_SELECTORS
+from utils.calldata.known_selectors import KNOWN_SELECTORS
 
 # A selector guaranteed NOT to be in the local table, for testing API fallback.
 _UNKNOWN_SELECTOR = "0x11223344"
@@ -93,7 +93,7 @@ def setUp(self):
 
     def test_known_selector_no_api_call(self):
         """Selectors in the local table should resolve without any API call."""
-        with patch("timelock.calldata_decoder.fetch_json") as mock_fetch:
+        with patch("utils.calldata.decoder.fetch_json") as mock_fetch:
             result = resolve_selector("0xa9059cbb")
             self.assertEqual(result, "transfer(address,uint256)")
             mock_fetch.assert_not_called()
@@ -103,7 +103,7 @@ def test_known_selector_case_insensitive(self):
         result = resolve_selector("0xA9059CBB")
         self.assertEqual(result, "transfer(address,uint256)")
 
-    @patch("timelock.calldata_decoder.fetch_json")
+    @patch("utils.calldata.decoder.fetch_json")
     def test_api_fallback_for_unknown_selector(self, mock_fetch):
         """Selectors not in the local table should fall through to the API."""
         mock_fetch.return_value = {
@@ -117,31 +117,31 @@ def test_api_fallback_for_unknown_selector(self, mock_fetch):
         self.assertEqual(result, "someRareFunction(uint256)")
         mock_fetch.assert_called_once()
 
-    @patch("timelock.calldata_decoder.fetch_json")
+    @patch("utils.calldata.decoder.fetch_json")
     def test_runtime_cache_hit(self, mock_fetch):
         """Previously resolved selectors should be served from runtime cache."""
         _selector_cache[_UNKNOWN_SELECTOR] = "cachedFunc(address)"
         result = resolve_selector(_UNKNOWN_SELECTOR)
         self.assertEqual(result, "cachedFunc(address)")
         mock_fetch.assert_not_called()
 
-    @patch("timelock.calldata_decoder.fetch_json")
+    @patch("utils.calldata.decoder.fetch_json")
     def test_runtime_cache_hit_none(self, mock_fetch):
         """Cached None (previously failed lookup) should return None without API call."""
         _selector_cache[_UNKNOWN_SELECTOR] = None
         result = resolve_selector(_UNKNOWN_SELECTOR)
         self.assertIsNone(result)
         mock_fetch.assert_not_called()
 
-    @patch("timelock.calldata_decoder.fetch_json")
+    @patch("utils.calldata.decoder.fetch_json")
     def test_api_returns_none(self, mock_fetch):
         mock_fetch.return_value = None
         result = resolve_selector(_UNKNOWN_SELECTOR)
         self.assertIsNone(result)
         # Should be cached as None
         self.assertIsNone(_selector_cache[_UNKNOWN_SELECTOR])
 
-    @patch("timelock.calldata_decoder.fetch_json")
+    @patch("utils.calldata.decoder.fetch_json")
     def test_no_match(self, mock_fetch):
         mock_fetch.return_value = {"result": {"function": {_UNKNOWN_SELECTOR: []}}}
         result = resolve_selector(_UNKNOWN_SELECTOR)
@@ -180,7 +180,7 @@ class TestDecodeCalldata(unittest.TestCase):
     def setUp(self):
         _selector_cache.clear()
 
-    @patch("timelock.calldata_decoder.resolve_selector")
+    @patch("utils.calldata.decoder.resolve_selector")
     def test_successful_decode(self, mock_resolve):
         mock_resolve.return_value = "transfer(address,uint256)"
         result = decode_calldata(TRANSFER_CALLDATA)
@@ -194,7 +194,7 @@ def test_successful_decode(self, mock_resolve):
         self.assertEqual(result.params[1][0], "uint256")
         self.assertEqual(result.params[1][1], 1000)
 
-    @patch("timelock.calldata_decoder.resolve_selector")
+    @patch("utils.calldata.decoder.resolve_selector")
     def test_no_params_function(self, mock_resolve):
         mock_resolve.return_value = "pause()"
         # selector only, no param data
@@ -205,7 +205,7 @@ def test_no_params_function(self, mock_resolve):
         self.assertEqual(result.signature, "pause()")
         self.assertEqual(result.params, [])
 
-    @patch("timelock.calldata_decoder.resolve_selector")
+    @patch("utils.calldata.decoder.resolve_selector")
     def test_selector_not_resolved(self, mock_resolve):
         mock_resolve.return_value = None
         result = decode_calldata("0xdeadbeef00112233")
@@ -216,7 +216,7 @@ def test_data_too_short(self):
         self.assertIsNone(decode_calldata(""))
         self.assertIsNone(decode_calldata(None))
 
-    @patch("timelock.calldata_decoder.resolve_selector")
+    @patch("utils.calldata.decoder.resolve_selector")
     def test_malformed_param_data_still_returns(self, mock_resolve):
         """If param decoding fails, should still return DecodedCall with empty params."""
         mock_resolve.return_value = "transfer(address,uint256)"
@@ -234,7 +234,7 @@ class TestFormatCallLines(unittest.TestCase):
     def setUp(self):
         _selector_cache.clear()
 
-    @patch("timelock.calldata_decoder.resolve_selector")
+    @patch("utils.calldata.decoder.resolve_selector")
     def test_successful_format(self, mock_resolve):
         mock_resolve.return_value = "transfer(address,uint256)"
         lines = format_call_lines(TRANSFER_CALLDATA)
@@ -248,7 +248,7 @@ def test_successful_format(self, mock_resolve):
         self.assertIn("uint256", lines[2])
         self.assertIn("1000", lines[2])
 
-    @patch("timelock.calldata_decoder.resolve_selector")
+    @patch("utils.calldata.decoder.resolve_selector")
     def test_fallback_raw_selector(self, mock_resolve):
         mock_resolve.return_value = None
         lines = format_call_lines("0xdeadbeef00112233")
@@ -261,7 +261,7 @@ def test_short_data_returns_empty(self):
         self.assertEqual(format_call_lines(""), [])
         self.assertEqual(format_call_lines(None), [])
 
-    @patch("timelock.calldata_decoder.resolve_selector")
+    @patch("utils.calldata.decoder.resolve_selector")
     def test_no_params_format(self, mock_resolve):
         mock_resolve.return_value = "pause()"
         lines = format_call_lines("0xabcd1234")

compound/test_tally.py tests/test_tally.pycompound/test_tally.py renamed to tests/test_tally.py

@@ -12,8 +12,8 @@
 
 from dotenv import load_dotenv
 
-from timelock.calldata_decoder import format_call_lines
 from utils.cache import cache_filename, get_last_value_for_key_from_file, write_last_value_to_file
+from utils.calldata.decoder import format_call_lines
 from utils.chains import EXPLORER_URLS, Chain
 from utils.llm.ai_explainer import explain_batch_transaction, explain_transaction, format_explanation_line
 from utils.logging import get_logger

timelock/timelock_alerts.py

@@ -0,0 +1,9 @@
+"""Calldata decoding utilities for governance transaction monitoring.
+
+Decodes raw EVM calldata into human-readable function calls using a local
+selector lookup table and the Sourcify 4byte signature database API.
+"""
+
+from utils.calldata.decoder import DecodedCall, decode_calldata, format_call_lines
+
+__all__ = ["DecodedCall", "decode_calldata", "format_call_lines"]

utils/calldata/__init__.py

@@ -11,7 +11,7 @@
 from eth_abi import decode
 from eth_utils import to_checksum_address
 
-from timelock.known_selectors import KNOWN_SELECTORS
+from utils.calldata.known_selectors import KNOWN_SELECTORS
 from utils.http import fetch_json
 from utils.logging import get_logger
 
@@ -173,10 +173,10 @@ def format_call_lines(data_hex: str) -> list[str]:
 
     result = decode_calldata(data_hex)
     if not result:
-        return [f"📝 Function: `{data_hex[:10]}`"]
+        return [f"\U0001f4dd Function: `{data_hex[:10]}`"]
 
-    lines = [f"📝 Function: `{result.signature}`"]
+    lines = [f"\U0001f4dd Function: `{result.signature}`"]
     for type_str, value in result.params:
         formatted = _format_param_value(type_str, value)
-        lines.append(f"    ├ {type_str}: `{formatted}`")
+        lines.append(f"    \u251c {type_str}: `{formatted}`")
     return lines

timelock/calldata_decoder.py utils/calldata/decoder.pytimelock/calldata_decoder.py renamed to utils/calldata/decoder.py

@@ -7,7 +7,7 @@
 
 from dataclasses import dataclass
 
-from timelock.calldata_decoder import DecodedCall, decode_calldata
+from utils.calldata.decoder import DecodedCall, decode_calldata
 from utils.llm import get_llm_provider
 from utils.llm.base import LLMError
 from utils.logging import get_logger

timelock/known_selectors.py utils/calldata/known_selectors.pytimelock/known_selectors.py renamed to utils/calldata/known_selectors.py

@@ -6,7 +6,7 @@
 
 from eth_utils import to_checksum_address
 
-from timelock.calldata_decoder import decode_calldata
+from utils.calldata.decoder import decode_calldata
 from utils.chains import EXPLORER_URLS, Chain
 from utils.logging import get_logger