From 7db23de4586eea84ea1d3e096a01489d1ffb0a06 Mon Sep 17 00:00:00 2001 From: spalen0 Date: Sat, 29 Aug 2026 13:25:40 +0200 Subject: [PATCH 1/2] Add contract reference table to LLM reports --- tests/test_llm_report.py | 55 +++++++++++++++++- utils/llm/README.md | 12 ++++ utils/llm/ai_explainer.py | 2 + utils/llm/report.py | 116 ++++++++++++++++++++++++++++++++++++-- 4 files changed, 178 insertions(+), 7 deletions(-) diff --git a/tests/test_llm_report.py b/tests/test_llm_report.py index 31f6c185..ace427e7 100644 --- a/tests/test_llm_report.py +++ b/tests/test_llm_report.py @@ -15,6 +15,7 @@ explorer_address_url, format_address_links_block, format_call_flow, + format_reference_table, iter_address_values, tuple_component_types, ) @@ -24,6 +25,7 @@ FARM = "0x79e1b8e45932a7c802ea3dab3844e5dea68d971f" FARM_CKS = "0x79e1B8e45932A7C802eA3dAb3844e5DEa68d971f" TIMELOCK = "0x4B174afbeD7b98BA01F50E36109EEE5e6d327c32" +DROP = "0xE4C72b4dE5b0F9ACcEA880Ad0b1F944F85A9dAA0" def _add_farms_ctx(**overrides) -> ReportContext: @@ -259,18 +261,67 @@ def test_falls_back_to_protocol_then_fallback(self) -> None: ) +class TestReferenceTable(unittest.TestCase): + def test_renders_executor_target_argument_and_protocol_context(self) -> None: + ctx = _add_farms_ctx( + labels={REGISTRY: "FarmRegistry", FARM_CKS: "New Silver 2 Senior", DROP: "New Silver Series 2 DROP"}, + label_address=TIMELOCK, + related_addresses=[FARM, DROP], + ) + + table = format_reference_table(ctx) + + self.assertIn("| Address | Label | Role | Description |", table) + self.assertIn(f"| [`{TIMELOCK}`](https://etherscan.io/address/{TIMELOCK}) | Infinifi Shorttimelock |", table) + self.assertIn( + f"| [`{REGISTRY}`](https://etherscan.io/address/{REGISTRY}) | FarmRegistry | Call target |", table + ) + self.assertIn("Receives `addFarms(uint256,address[])`", table) + self.assertIn("New Silver 2 Senior | Calldata argument; Protocol context", table) + self.assertIn("Passed as `_farms` to `addFarms(uint256,address[])`", table) + self.assertIn("New Silver Series 2 DROP | Protocol context", table) + + def test_deduplicates_repeated_addresses_and_descriptions(self) -> None: + call = _add_farms_ctx().entries[0] + ctx = _add_farms_ctx(entries=[call, call], related_addresses=[REGISTRY]) + + table = format_reference_table(ctx) + + self.assertEqual(table.count(f"| [`{REGISTRY}`](https://etherscan.io/address/{REGISTRY})"), 1) + self.assertEqual(table.count("Receives `addFarms(uint256,address[])`"), 1) + + def test_escapes_dynamic_table_text(self) -> None: + ctx = _add_farms_ctx(labels={REGISTRY: "Farm | Registry\nMain"}) + self.assertIn("Farm \\| Registry Main", format_reference_table(ctx)) + + def test_includes_addresses_from_nested_calldata(self) -> None: + transfer = "0xa9059cbb" + FARM[2:].zfill(64) + f"{1:064x}" + outer = DecodedCall(function_name="execute", signature="execute(bytes)", params=[("bytes", transfer)]) + ctx = _add_farms_ctx(entries=[CallEntry(target=REGISTRY, call=outer)]) + + table = format_reference_table(ctx) + + self.assertIn(f"[`{FARM_CKS}`](https://etherscan.io/address/{FARM_CKS})", table) + self.assertIn("to `transfer(address,uint256)`", table) + + def test_empty_without_addresses(self) -> None: + self.assertEqual(format_reference_table(ReportContext()), "") + + 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")) + self.assertLess(report.index("## Call Flow"), report.index("## Reference")) + self.assertLess(report.index("## Reference"), report.index("## Analysis")) def test_protocol_context_is_deterministic_section_before_analysis(self) -> None: ctx = _add_farms_ctx(protocol_context="- **Farm:** New Silver 2 Senior") report = build_report("Updates the rate.", "Long analysis.", ctx, "LOW") self.assertIn("## Protocol Context\n\n- **Farm:** New Silver 2 Senior", report) self.assertLess(report.index("## Call Flow"), report.index("## Protocol Context")) - self.assertLess(report.index("## Protocol Context"), report.index("## Analysis")) + self.assertLess(report.index("## Protocol Context"), report.index("## Reference")) + self.assertLess(report.index("## Reference"), report.index("## Analysis")) def test_metadata_header(self) -> None: report = build_report("Summary.", "Analysis.", _add_farms_ctx(label_address=TIMELOCK), "HIGH") diff --git a/utils/llm/README.md b/utils/llm/README.md index d251788e..a2e7e030 100644 --- a/utils/llm/README.md +++ b/utils/llm/README.md @@ -361,6 +361,18 @@ Registers a new type-2 farm in FarmRegistry. … ``` +The report also includes a code-generated `## Reference` table before Analysis: + +```markdown +| Address | Label | Role | Description | +|---|---|---|---| +| [`0x4B17…7c32`](https://etherscan.io/address/0x4B17…) | Infinifi Shorttimelock | Executor | Executes the governance transaction | +| [`0x11F6…4189`](https://etherscan.io/address/0x11F6…) | RWAEscrowRateManager | Call target | Receives `setRate(address,uint256)` | +| [`0xE4C7…dAA0`](https://etherscan.io/address/0xE4C7…) | New Silver Series 2 DROP | Protocol context | Resolved by the INFINIFI protocol adapter | +``` + +The table deduplicates the executor, report contract, call targets, address-valued calldata arguments, and addresses introduced by protocol adapters. Roles and descriptions come from those deterministic relationships; the LLM does not generate them. + **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 diff --git a/utils/llm/ai_explainer.py b/utils/llm/ai_explainer.py index 28056a27..8802700d 100644 --- a/utils/llm/ai_explainer.py +++ b/utils/llm/ai_explainer.py @@ -1333,6 +1333,7 @@ def explain_transaction( from_address=from_address, label_address=label_address or from_address, protocol_context=protocol_ctx.report, + related_addresses=protocol_ctx.addresses, ) try: @@ -1489,6 +1490,7 @@ def explain_batch_transaction( from_address=from_address, label_address=label_address or from_address, protocol_context=protocol_ctx.report, + related_addresses=protocol_ctx.addresses, ) try: diff --git a/utils/llm/report.py b/utils/llm/report.py index ad196ef4..7c00402f 100644 --- a/utils/llm/report.py +++ b/utils/llm/report.py @@ -2,8 +2,9 @@ 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. +deterministic, code-built **call flow** and **reference table** — 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, @@ -67,6 +68,19 @@ class ReportContext: # are not part of the raw calldata flow (for example an Infinifi farm and # the non-accounting ERC20 targets configured in its escrow). protocol_context: str = "" + # Addresses introduced by protocol-specific context. These may not occur in + # calldata but still belong in the report's deterministic reference table. + related_addresses: list[str] = field(default_factory=list) + + +@dataclass +class _ReferenceEntry: + """One address and its accumulated deterministic report roles.""" + + address: str + label: str + roles: list[str] = field(default_factory=list) + descriptions: list[str] = field(default_factory=list) def checksum_or_none(addr: object) -> str | None: @@ -74,7 +88,7 @@ def checksum_or_none(addr: object) -> str | None: if not isinstance(addr, str) or not addr.startswith("0x"): return None try: - return to_checksum_address(addr) + return str(to_checksum_address(addr)) except ValueError: return None @@ -305,9 +319,98 @@ def format_call_flow(ctx: ReportContext) -> str: return "\n".join(lines).rstrip() +def _reference_label(ctx: ReportContext, address: str) -> str: + """Return the best deterministic label available for a reference row.""" + if checksum_or_none(ctx.label_address) == address and ctx.label: + return ctx.label + return ctx.labels.get(address, "") + + +def _add_reference( + entries: dict[str, _ReferenceEntry], + ctx: ReportContext, + raw_address: str, + role: str, + description: str, +) -> None: + """Add or enrich one checksummed reference entry.""" + address = checksum_or_none(raw_address) + if address is None or address == ZERO_ADDRESS: + return + key = address.lower() + entry = entries.setdefault(key, _ReferenceEntry(address, _reference_label(ctx, address))) + if role not in entry.roles: + entry.roles.append(role) + if description not in entry.descriptions: + entry.descriptions.append(description) + + +def _table_cell(value: str) -> str: + """Escape dynamic text for one GitHub-flavored Markdown table cell.""" + return value.replace("|", "\\|").replace("\r", " ").replace("\n", " ").strip() + + +def _iter_reference_arguments( + call: DecodedCall, + param_names: list[str] | None, + depth: int = 0, +) -> Iterator[tuple[str, str]]: + """Yield address arguments and factual descriptions, including nested calldata.""" + for index, (type_str, value) in enumerate(call.params): + name = param_names[index] if param_names is not None and index < len(param_names) else "" + parameter = f"`{name}`" if name else f"argument {index + 1}" + description = f"Passed as {parameter} to `{call.signature}`" + for address in iter_address_values(type_str, value): + yield address, description + if type_str == "bytes" and depth < MAX_BYTES_RECURSION_DEPTH: + inner = try_decode_inner_calldata(value) + if inner is not None: + yield from _iter_reference_arguments(inner, None, depth + 1) + + +def format_reference_table(ctx: ReportContext) -> str: + """Render addresses used by the transaction as a deterministic table.""" + references: dict[str, _ReferenceEntry] = {} + _add_reference(references, ctx, ctx.from_address, "Executor", "Executes the governance transaction") + + label_address = checksum_or_none(ctx.label_address) + sender = checksum_or_none(ctx.from_address) + if label_address is not None and label_address != sender: + _add_reference(references, ctx, label_address, "Alert contract", "Contract named in the report header") + + for entry in ctx.entries: + _add_reference( + references, + ctx, + entry.target, + "Call target", + f"Receives `{entry.call.signature}`", + ) + for address, description in _iter_reference_arguments(entry.call, entry.param_names): + _add_reference(references, ctx, address, "Calldata argument", description) + + context_description = ( + f"Resolved by the {ctx.protocol} protocol adapter" if ctx.protocol else "Resolved by protocol context" + ) + for address in ctx.related_addresses: + _add_reference(references, ctx, address, "Protocol context", context_description) + + if not references: + return "" + + lines = ["| Address | Label | Role | Description |", "|---|---|---|---|"] + for reference in references.values(): + address = address_link(reference.address, ctx.chain_id) + label = _table_cell(reference.label) or "—" + roles = _table_cell("; ".join(reference.roles)) + descriptions = _table_cell("; ".join(reference.descriptions)) + lines.append(f"| {address} | {label} | {roles} | {descriptions} |") + return "\n".join(lines) + + def _chain_name(chain_id: int) -> str: try: - return Chain.from_chain_id(chain_id).network_name.capitalize() + return str(Chain.from_chain_id(chain_id).network_name).capitalize() except ValueError: return f"Chain {chain_id}" @@ -359,7 +462,7 @@ def build_report(summary: str, detail: str, ctx: ReportContext, risk_tag: str = Sections: metadata header, the Telegram-visible summary (so the gist is self-contained), the deterministic call flow, optional protocol context, - and the LLM's analysis. + a deterministic address reference, and the LLM's analysis. Args: summary: The authoritative TLDR, risk tag already stripped by the caller. @@ -384,6 +487,9 @@ def build_report(summary: str, detail: str, ctx: ReportContext, risk_tag: str = sections.append(f"## Call Flow\n\n{call_flow}") if ctx.protocol_context: sections.append(f"## Protocol Context\n\n{ctx.protocol_context}") + reference = format_reference_table(ctx) + if reference: + sections.append(f"## Reference\n\n{reference}") if detail: sections.append(f"## Analysis\n\n{_REDUNDANT_ANALYSIS_HEADING_RE.sub('', detail)}") return "\n\n".join(sections) From 39bde19c4e826f8ca481ff623d5174e23fdbbdf2 Mon Sep 17 00:00:00 2001 From: spalen0 Date: Sat, 29 Aug 2026 14:19:09 +0200 Subject: [PATCH 2/2] fix: resolve code review findings for PR #346 Preserve role and description pairs in the reference table, decode calldata nested in bytes arrays and tuples, and use schedule-safe executor wording. Align the documentation with emitted roles and full explorer links. --- tests/test_llm_report.py | 51 ++++++++++++++++++++++++++++++++++++++++ utils/llm/README.md | 8 +++---- utils/llm/report.py | 43 +++++++++++++++++++++++---------- 3 files changed, 85 insertions(+), 17 deletions(-) diff --git a/tests/test_llm_report.py b/tests/test_llm_report.py index ace427e7..9d82a14d 100644 --- a/tests/test_llm_report.py +++ b/tests/test_llm_report.py @@ -290,6 +290,29 @@ def test_deduplicates_repeated_addresses_and_descriptions(self) -> None: self.assertEqual(table.count(f"| [`{REGISTRY}`](https://etherscan.io/address/{REGISTRY})"), 1) self.assertEqual(table.count("Receives `addFarms(uint256,address[])`"), 1) + def test_keeps_each_description_paired_with_its_role(self) -> None: + set_rate = DecodedCall(function_name="setRate", signature="setRate(uint256)", params=[("uint256", 1)]) + pause = DecodedCall(function_name="pause", signature="pause()") + set_owner = DecodedCall( + function_name="setOwner", + signature="setOwner(address)", + params=[("address", REGISTRY)], + ) + ctx = _add_farms_ctx( + entries=[ + CallEntry(target=REGISTRY, call=set_rate), + CallEntry(target=REGISTRY, call=pause), + CallEntry(target=DROP, call=set_owner, param_names=["owner"]), + ], + ) + + row = next(line for line in format_reference_table(ctx).splitlines() if line.startswith(f"| [`{REGISTRY}`]")) + + self.assertIn("| Call target; Calldata argument |", row) + self.assertIn("**Call target:** Receives `setRate(uint256)`", row) + self.assertIn("**Call target:** Receives `pause()`", row) + self.assertIn("**Calldata argument:** Passed as `owner` to `setOwner(address)`", row) + def test_escapes_dynamic_table_text(self) -> None: ctx = _add_farms_ctx(labels={REGISTRY: "Farm | Registry\nMain"}) self.assertIn("Farm \\| Registry Main", format_reference_table(ctx)) @@ -304,6 +327,34 @@ def test_includes_addresses_from_nested_calldata(self) -> None: self.assertIn(f"[`{FARM_CKS}`](https://etherscan.io/address/{FARM_CKS})", table) self.assertIn("to `transfer(address,uint256)`", table) + def test_includes_addresses_from_bytes_array_payloads(self) -> None: + transfer = "0xa9059cbb" + FARM[2:].zfill(64) + f"{1:064x}" + batch = DecodedCall( + function_name="executeBatch", + signature="executeBatch(bytes[])", + params=[("bytes[]", (transfer,))], + ) + ctx = _add_farms_ctx(entries=[CallEntry(target=REGISTRY, call=batch)]) + + table = format_reference_table(ctx) + + self.assertIn(f"[`{FARM_CKS}`](https://etherscan.io/address/{FARM_CKS})", table) + self.assertIn("to `transfer(address,uint256)`", table) + + def test_includes_addresses_from_tuple_payloads(self) -> None: + transfer = "0xa9059cbb" + FARM[2:].zfill(64) + f"{1:064x}" + wrapper = DecodedCall( + function_name="execute", + signature="execute((address,bytes))", + params=[("(address,bytes)", (REGISTRY, transfer))], + ) + ctx = _add_farms_ctx(entries=[CallEntry(target=DROP, call=wrapper)]) + + table = format_reference_table(ctx) + + self.assertIn(f"[`{FARM_CKS}`](https://etherscan.io/address/{FARM_CKS})", table) + self.assertIn("to `transfer(address,uint256)`", table) + def test_empty_without_addresses(self) -> None: self.assertEqual(format_reference_table(ReportContext()), "") diff --git a/utils/llm/README.md b/utils/llm/README.md index a2e7e030..f27adcf7 100644 --- a/utils/llm/README.md +++ b/utils/llm/README.md @@ -366,12 +366,12 @@ The report also includes a code-generated `## Reference` table before Analysis: ```markdown | Address | Label | Role | Description | |---|---|---|---| -| [`0x4B17…7c32`](https://etherscan.io/address/0x4B17…) | Infinifi Shorttimelock | Executor | Executes the governance transaction | -| [`0x11F6…4189`](https://etherscan.io/address/0x11F6…) | RWAEscrowRateManager | Call target | Receives `setRate(address,uint256)` | -| [`0xE4C7…dAA0`](https://etherscan.io/address/0xE4C7…) | New Silver Series 2 DROP | Protocol context | Resolved by the INFINIFI protocol adapter | +| [`0x4B174afbeD7b98BA01F50E36109EEE5e6d327c32`](https://etherscan.io/address/0x4B174afbeD7b98BA01F50E36109EEE5e6d327c32) | Infinifi Shorttimelock | Executor | **Executor:** Execution authority for the governance transaction | +| [`0x11F6FAb3f4D8635880C3e80cbae8AEF8136D4189`](https://etherscan.io/address/0x11F6FAb3f4D8635880C3e80cbae8AEF8136D4189) | RWAEscrowRateManager | Call target | **Call target:** Receives `setRate(address,uint256)` | +| [`0xE4C72b4dE5b0F9ACcEA880Ad0b1F944F85A9dAA0`](https://etherscan.io/address/0xE4C72b4dE5b0F9ACcEA880Ad0b1F944F85A9dAA0) | New Silver Series 2 DROP | Protocol context | **Protocol context:** Resolved by the INFINIFI protocol adapter | ``` -The table deduplicates the executor, report contract, call targets, address-valued calldata arguments, and addresses introduced by protocol adapters. Roles and descriptions come from those deterministic relationships; the LLM does not generate them. +The table deduplicates the executor, alert contract, call targets, address-valued calldata arguments, and addresses introduced by protocol adapters. Every description is prefixed with its role so multi-use addresses remain unambiguous. Roles and descriptions come from those deterministic relationships; the LLM does not generate them. **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 diff --git a/utils/llm/report.py b/utils/llm/report.py index 7c00402f..6773b7fa 100644 --- a/utils/llm/report.py +++ b/utils/llm/report.py @@ -75,12 +75,11 @@ class ReportContext: @dataclass class _ReferenceEntry: - """One address and its accumulated deterministic report roles.""" + """One address and its accumulated deterministic report usages.""" address: str label: str - roles: list[str] = field(default_factory=list) - descriptions: list[str] = field(default_factory=list) + usages: list[tuple[str, str]] = field(default_factory=list) def checksum_or_none(addr: object) -> str | None: @@ -339,10 +338,9 @@ def _add_reference( return key = address.lower() entry = entries.setdefault(key, _ReferenceEntry(address, _reference_label(ctx, address))) - if role not in entry.roles: - entry.roles.append(role) - if description not in entry.descriptions: - entry.descriptions.append(description) + usage = (role, description) + if usage not in entry.usages: + entry.usages.append(usage) def _table_cell(value: str) -> str: @@ -350,6 +348,24 @@ def _table_cell(value: str) -> str: return value.replace("|", "\\|").replace("\r", " ").replace("\n", " ").strip() +def _iter_inner_calldata(type_str: str, value: object) -> Iterator[DecodedCall]: + """Yield decodable calldata from bytes leaves inside arrays and tuples.""" + element = array_element_type(type_str) + if element is not None and isinstance(value, (list, tuple)): + for item in value: + yield from _iter_inner_calldata(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_inner_calldata(component, item) + return + if type_str == "bytes": + inner = try_decode_inner_calldata(value) + if inner is not None: + yield inner + + def _iter_reference_arguments( call: DecodedCall, param_names: list[str] | None, @@ -362,16 +378,15 @@ def _iter_reference_arguments( description = f"Passed as {parameter} to `{call.signature}`" for address in iter_address_values(type_str, value): yield address, description - if type_str == "bytes" and depth < MAX_BYTES_RECURSION_DEPTH: - inner = try_decode_inner_calldata(value) - if inner is not None: + if depth < MAX_BYTES_RECURSION_DEPTH: + for inner in _iter_inner_calldata(type_str, value): yield from _iter_reference_arguments(inner, None, depth + 1) def format_reference_table(ctx: ReportContext) -> str: """Render addresses used by the transaction as a deterministic table.""" references: dict[str, _ReferenceEntry] = {} - _add_reference(references, ctx, ctx.from_address, "Executor", "Executes the governance transaction") + _add_reference(references, ctx, ctx.from_address, "Executor", "Execution authority for the governance transaction") label_address = checksum_or_none(ctx.label_address) sender = checksum_or_none(ctx.from_address) @@ -402,8 +417,10 @@ def format_reference_table(ctx: ReportContext) -> str: for reference in references.values(): address = address_link(reference.address, ctx.chain_id) label = _table_cell(reference.label) or "—" - roles = _table_cell("; ".join(reference.roles)) - descriptions = _table_cell("; ".join(reference.descriptions)) + roles = _table_cell("; ".join(dict.fromkeys(role for role, _description in reference.usages))) + descriptions = "
".join( + f"**{_table_cell(role)}:** {_table_cell(description)}" for role, description in reference.usages + ) lines.append(f"| {address} | {label} | {roles} | {descriptions} |") return "\n".join(lines)