Add contract reference table to LLM reports

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,118 @@ 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_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))
+
+    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_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()), "")
+
+
 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")

utils/llm/README.md

@@ -361,6 +361,18 @@ Registers a new type-2 farm in FarmRegistry. …
 <the LLM detail>
 ```
 
+The report also includes a code-generated `## Reference` table before Analysis:
+
+```markdown
+| Address | Label | Role | Description |
+|---|---|---|---|
+| [`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, 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
 value, nested `bytes` payloads unwrapped up to `MAX_BYTES_RECURSION_DEPTH`), so it

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:

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,14 +68,26 @@ 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 usages."""
+
+    address: str
+    label: str
+    usages: list[tuple[str, str]] = field(default_factory=list)
 
 
 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)
+        return str(to_checksum_address(addr))
     except ValueError:
         return None
 
@@ -305,9 +318,116 @@ 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)))
+    usage = (role, description)
+    if usage not in entry.usages:
+        entry.usages.append(usage)
+
+
+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_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,
+    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 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", "Execution authority for 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(dict.fromkeys(role for role, _description in reference.usages)))
+        descriptions = "<br>".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)
+
+
 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 +479,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 +504,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)