diff --git a/protocols/pendle/README.md b/protocols/pendle/README.md index 1cf6410c..37ca1d89 100644 --- a/protocols/pendle/README.md +++ b/protocols/pendle/README.md @@ -10,3 +10,13 @@ Additionally, other contracts like vePENDLE, PENDLE, RewardDistributor, and Voti Arbitrum Safe Multisig: 0x7877AdFaDEd756f3248a0EBfe8Ac2E2eF87b75Ac The owner of SY contracts was changed to [governance proxy contract](https://etherscan.io/address/0x2aD631F72fB16d91c4953A7f4260A97C2fE2f31e) with an additional guardian role that can only pause SY contracts. The governance proxy contract owner is multisig defined above. + +## AI governance context + +PendleSwap upgrade alerts on Ethereum and Arbitrum include the proxy's live owner, +current/proposed verified swap and authorization code, and a pinned reference for +the standard router integration. This distinguishes the optional aggregator leg +from market, PT/YT and SY contracts, and makes swap payload/enum changes visible +even when the external ABI is unchanged. The context is included in the AI prompt +and full report; it does not impose a risk rating. See +[the LLM context documentation](../../utils/llm/README.md#5f-4-pendleswap-upgrade-context-utilsllmpendle_contextpy). diff --git a/tests/test_pendle_context.py b/tests/test_pendle_context.py new file mode 100644 index 00000000..661559b0 --- /dev/null +++ b/tests/test_pendle_context.py @@ -0,0 +1,233 @@ +"""PendleSwap context behavior using small, synthetic source bundles.""" + +from collections.abc import Iterator +from dataclasses import replace +from unittest.mock import MagicMock, patch + +import pytest + +from utils.calldata.decoder import DecodedCall +from utils.llm import pendle_context +from utils.llm.pendle_context import ( + PENDLE_SWAP, + _source_evidence, + format_pendle_prompt, + resolve_pendle_context, +) +from utils.llm.protocol_context import resolve_protocol_context +from utils.verified_contract import VerifiedContract + +OLD = "0xBC17404b7bb500051c75C83E4aA5aE447D967811" +NEW = "0xD14feb6Aaf8650BbfcC8aBEc299B249a80FE7C78" +OWNER = "0x8119EC16F0573B7dAc7C0CB94EB504FB32456ee1" +type Boundaries = tuple[MagicMock, MagicMock, MagicMock] + +_OLD_SOURCE = """ +contract PendleSwap { + function swap() external { _getScaledInputData(SwapType.ODOS); } + function _getScaledInputData(SwapType swapType) internal { + if (swapType == SwapType.ODOS) { _odosScaling(); } else { assert(false); } + } + function _authorizeUpgrade(address) internal onlyOwner {} +} +""" +_NEW_SOURCE = """ +contract PendleSwap { + function swap() external { _zeroExSwap(); } + function _zeroExSwap() internal { _transferOut(tokenOut, msg.sender, netOut); } + function _getScaledInputData(SwapType swapType) internal { assert(false); } + function _authorizeUpgrade(address) internal onlyOwner {} +} +""" +_OWNERSHIP_SOURCE = """ +abstract contract BoringOwnableUpgradeableV2 { + modifier onlyOwner() { require(msg.sender == owner, "not owner"); _; } +} +""" + + +def _record(name: str) -> VerifiedContract: + """Build only the source declarations the adapter consumes, without cache metadata.""" + swap_types = "NONE, KYBERSWAP, RESERVE_1, ZEROX" if name == "new" else "NONE, KYBERSWAP, ODOS, RESERVE_2" + return VerifiedContract( + contract_name="PendleSwap", + compiler_version="", + language="Solidity", + contract_file="PendleSwap.sol", + sources={ + "PendleSwap.sol": _NEW_SOURCE if name == "new" else _OLD_SOURCE, + "IPSwapAggregator.sol": ( + "struct SwapData { SwapType swapType; address extRouter; bytes extCalldata; bool needScale; }\n" + f"enum SwapType {{ {swap_types} }}\ninterface IPSwapAggregator {{}}" + ), + "BoringOwnableUpgradeableV2.sol": _OWNERSHIP_SOURCE, + }, + ) + + +def _upgrade(migrate: bool = False) -> DecodedCall: + """A decoded upgrade with or without an initialization payload.""" + if migrate: + return DecodedCall( + "upgradeToAndCall", "upgradeToAndCall(address,bytes)", [("address", NEW), ("bytes", b"\x01")] + ) + return DecodedCall("upgradeTo", "upgradeTo(address)", [("address", NEW)]) + + +@pytest.fixture +def boundaries() -> Iterator[Boundaries]: + """Replace RPC and source fetches while retaining all context resolution logic.""" + with ( + patch.object(pendle_context, "get_current_implementation", return_value=OLD) as implementation, + patch.object(pendle_context, "fetch_verified_contract") as source, + patch.object(pendle_context.ChainManager, "get_client") as client, + ): + source.side_effect = lambda chain, address: _record("old" if address.lower() == OLD.lower() else "new") + client.return_value.eth.contract.return_value.functions.owner.return_value.call.return_value = OWNER + yield implementation, source, client + + +@pytest.mark.parametrize("chain_id", [1, 42161]) +def test_upgrade_includes_scope_controls_and_route_semantics(boundaries: Boundaries, chain_id: int) -> None: + """Unchanged ABI must not hide enum changes, unsupported scaling or existing authorization.""" + resolved = resolve_protocol_context("PENDLE", chain_id, [(PENDLE_SWAP.lower(), _upgrade())]) + prompt = resolved.prompt + assert "not a live trace" in prompt + assert f"Current proxy owner() (live read): {OWNER}" in prompt + assert "_authorizeUpgrade(address) internal onlyOwner" in prompt + assert 'require(msg.sender == owner, "not owner")' in prompt + before, after = prompt.split("Proposed implementation evidence:") + assert "SwapType.ODOS" in before + assert "RESERVE_2" in before + assert "RESERVE_1" in after + assert "ZEROX" in after + assert "SwapType.ODOS" not in after + assert "assert(false)" in after + assert "_transferOut(tokenOut, msg.sender, netOut)" in after + explorer = "etherscan.io" if chain_id == 1 else "arbiscan.io" + for address in (PENDLE_SWAP, OLD, NEW, OWNER): + assert address in resolved.addresses + assert f"https://{explorer}/address/{address}" in resolved.report + assert resolved.labels[PENDLE_SWAP] == "PendleSwap" + contract_call = boundaries[2].return_value.eth.contract.call_args + assert contract_call.kwargs["address"] == PENDLE_SWAP + + +@pytest.mark.parametrize( + "protocol,chain,target,call", + [ + ("yearn", 1, PENDLE_SWAP, _upgrade()), + ("pendle", 10, PENDLE_SWAP, _upgrade()), + ("pendle", 1, OWNER, _upgrade()), + ("pendle", 1, PENDLE_SWAP, DecodedCall("swap", "swap(address,uint256)", [])), + ("pendle", 1, PENDLE_SWAP, DecodedCall("upgradeTo", "upgradeTo(address)", [])), + ("pendle", 1, PENDLE_SWAP, DecodedCall("upgradeTo", "upgradeTo(address)", [("uint256", 1)])), + ("pendle", 1, PENDLE_SWAP, DecodedCall("upgradeTo", "upgradeTo(address)", [("address", "bad")])), + ], +) +def test_unrelated_or_malformed_calls_make_no_network_requests( + boundaries: Boundaries, + protocol: str, + chain: int, + target: str, + call: DecodedCall, +) -> None: + """Scope checks run before all RPC/source work.""" + assert resolve_pendle_context(protocol, chain, [(target, call)]) == [] + for boundary in boundaries: + boundary.assert_not_called() + + +def test_empty_batch_makes_no_network_requests(boundaries: Boundaries) -> None: + """Empty alerts need no enrichment.""" + assert resolve_pendle_context("pendle", 1, []) == [] + for boundary in boundaries: + boundary.assert_not_called() + + +def test_duplicate_calls_and_migration_are_distinguished(boundaries: Boundaries) -> None: + """A Safe batch retains migration context while repeated identical upgrades are coalesced.""" + contexts = resolve_pendle_context( + "pendle", + 1, + [ + (PENDLE_SWAP, _upgrade()), + (PENDLE_SWAP, _upgrade()), + (PENDLE_SWAP, _upgrade(True)), + ], + ) + assert len(contexts) == 2 + prompt = format_pendle_prompt(contexts) + assert "upgradeTo has no initialization/migration payload" in prompt + assert "upgradeToAndCall includes a bytes payload" in prompt + + +def test_owner_read_failure_preserves_source_evidence(boundaries: Boundaries) -> None: + """An unavailable owner is explicit and cannot remove the code comparison.""" + boundaries[2].return_value.eth.contract.return_value.functions.owner.return_value.call.side_effect = RuntimeError( + "RPC failed" + ) + contexts = resolve_pendle_context("pendle", 1, [(PENDLE_SWAP, _upgrade())]) + prompt = format_pendle_prompt(contexts) + assert "Current proxy owner() (live read): unavailable" in prompt + assert "_zeroExSwap" in prompt + assert contexts[0].owner is None + + +def test_missing_source_and_implementation_are_explicit(boundaries: Boundaries) -> None: + """Missing evidence never turns into a guessed old implementation or safe-storage claim.""" + boundaries[0].return_value = None + boundaries[1].side_effect = None + boundaries[1].return_value = None + contexts = resolve_pendle_context("pendle", 1, [(PENDLE_SWAP, _upgrade())]) + prompt = format_pendle_prompt(contexts) + assert "Current implementation: unavailable" in prompt + assert prompt.count("target source unavailable") == 2 + + +def test_failed_batch_member_does_not_hide_next_upgrade(boundaries: Boundaries) -> None: + """Source/RPC errors remain local to a batch member.""" + boundaries[0].side_effect = [RuntimeError("RPC failed"), OLD] + contexts = resolve_pendle_context("pendle", 1, [(PENDLE_SWAP, _upgrade()), (PENDLE_SWAP, _upgrade(True))]) + assert len(contexts) == 1 + assert contexts[0].is_upgrade_and_call + + +@pytest.mark.parametrize("flattened", [False, True]) +def test_source_excerpts_exclude_unrelated_contracts(flattened: bool) -> None: + """Only the deployed target and relevant declarations belong in context, in either source format.""" + record = _record("new") + sources = dict(record.sources) + sources["PendleSwap.sol"] += '\ncontract Decoy { function swap() external { revert("DECOY"); } }' + if flattened: + record = replace(record, sources={"flat.sol": "\n".join(sources.values())}, contract_file="flat.sol") + else: + record = replace(record, sources=sources) + evidence = _source_evidence(record) + assert "DECOY" not in evidence + assert "enum SwapType" in evidence + assert "struct SwapData" in evidence + assert "_zeroExSwap" in evidence + assert 'require(msg.sender == owner, "not owner")' in evidence + + +def test_other_replacement_and_unresolved_target_are_not_assumed_to_be_pendle() -> None: + """Labels and imported interfaces cannot establish replacement behavior.""" + record = _record("new") + for candidate in (None, replace(record, contract_name="OtherContract"), replace(record, contract_file=None)): + evidence = _source_evidence(candidate) + assert "unavailable" in evidence + assert "_zeroExSwap" not in evidence + + +def test_ambiguous_interfaces_and_owner_bases_are_not_guessed() -> None: + """Duplicate declarations in a bundle leave their facts unresolved.""" + record = _record("new") + sources = dict(record.sources) + for path, source in record.sources.items(): + if path.endswith(("IPSwapAggregator.sol", "BoringOwnableUpgradeableV2.sol")): + sources[f"decoy/{path}"] = source + evidence = _source_evidence(replace(record, sources=sources)) + assert "Verified swap payload/enum" not in evidence + assert "Verified ownership guard" not in evidence + assert "_zeroExSwap" in evidence diff --git a/tests/test_protocol_context.py b/tests/test_protocol_context.py index b50f6a8d..bc0986f6 100644 --- a/tests/test_protocol_context.py +++ b/tests/test_protocol_context.py @@ -81,7 +81,7 @@ def capture(contexts: list, chain_id: int, labels: dict[str, str]) -> str: def test_registered_adapters_cover_the_known_protocols(self) -> None: self.assertEqual( {adapter.name for adapter in protocol_context._ADAPTERS}, - {"infinifi", "infinifi-outland", "3jane", "yearn-v3", "control-transfer"}, + {"infinifi", "infinifi-outland", "3jane", "pendle", "yearn-v3", "control-transfer"}, ) diff --git a/utils/llm/README.md b/utils/llm/README.md index 50736530..564ec000 100644 --- a/utils/llm/README.md +++ b/utils/llm/README.md @@ -281,6 +281,31 @@ For any protocol, calls that hand over control (`set_management`, `transferOwner Involved Safes get `Safe m-of-n` labels. These are applied with `setdefault`, so curated names win. Failures are best-effort and never block the alert. +### 5f-4. PendleSwap Upgrade Context (`utils/llm/pendle_context.py`) + +For Pendle alerts on Ethereum and Arbitrum, direct `upgradeTo` and `upgradeToAndCall` +calls to the published PendleSwap proxy (including calls in Safe multisend batches) +receive adapter-specific context. This identifies the optional aggregator leg separately +from markets, PT/YT and SY contracts, and cites a pinned upstream `ActionBase` integration +reference for input pre-funding, output returned to the calling router, aggregator bypass +branches and caller-supplied output constraints. The architecture reference is explicitly +separate from a live execution trace or a claim about current balances or route usage. + +The adapter reads `owner()` at the proxy and includes complete, contract-scoped verified +members from both current and proposed implementations: swap dispatch, output handling, +scaling, approvals and the upgrade authorization hook. The verified swap payload/enum and +ownership guard are also included when uniquely resolved in the bundle. This exposes +semantic enum changes despite an unchanged ABI, unsupported scaling reverts, and unchanged +authorization/approval code that a diff alone omits. Missing source or owner reads remain +explicitly unavailable; a replacement with a different contract name is not assumed to +retain PendleSwap behavior. Context is rendered in both the prompt and gist and sets no +fixed risk tag or storage-safety verdict. Nested governance wrappers are not resolved by +this adapter. + +The summary critique also checks functional claims against supplied code: removing a +scaling branch does not establish removal of unscaled routes, and an UNKNOWN storage +verdict is not evidence that collisions are more likely. + ### 5g. Adapter Registry (`utils/llm/protocol_context.py`) Adapters register in `_ADAPTERS`; `resolve_protocol_context()` fans one call out to all of them and merges the rendered prompt text, report text, introduced addresses, and address labels. Each adapter guards itself, so registration order carries no meaning and one adapter raising is logged and skipped rather than dropping the alert. Most guard on protocol and chain. The Yearn V3 adapter guards on call shape and `apiVersion()`, and the control-transfer adapter on call shape alone. @@ -548,6 +573,7 @@ utils/llm/ ├── factory.py # Provider factory with env-based config + singleton ├── infinifi_context.py # Infinifi adapter: escrow → farm, custody, setRate APR, whitelist calls ├── openai_compat.py # OpenAI-compatible provider (Venice, OpenAI, etc.) +├── pendle_context.py # PendleSwap upgrades: router integration, owner, complete source members ├── protocol_context.py # Registry fanning one call out to every protocol adapter ├── report.py # Gist report: metadata header + deterministic call flow + analysis ├── threejane_abi.py # 3Jane checked-in ABIs + verified-ABI probes (proxy-aware) diff --git a/utils/llm/ai_explainer.py b/utils/llm/ai_explainer.py index 08444508..8817115f 100644 --- a/utils/llm/ai_explainer.py +++ b/utils/llm/ai_explainer.py @@ -280,6 +280,9 @@ confirmed? 5. Does the risk tag match the magnitude of change shown in the context? (A 10× change to a critical parameter is rarely LOW; a no-op is rarely HIGH.) +6. Are functional claims supported by the full supplied code and context? Do not turn + removal of one scaling branch into removal of all unscaled routes, or treat UNKNOWN + storage compatibility as evidence of a collision or increased collision likelihood. Hard rules for the revision (if you choose to revise): - Do NOT introduce a unit/scale assumption that wasn't supported by the context. @@ -290,7 +293,7 @@ - Do NOT remove an explicit hedge ("unit cannot be confirmed", "without source context", etc.). - Do NOT polish for style alone. Only edit if there's a concrete, specific issue - from items 1-5. + from items 1-6. If every check is satisfied AND no hard rule would be violated by the draft as-is, output exactly: @@ -1567,9 +1570,10 @@ def _build_prompt( if protocol_context: parts.append( - "\n--- Protocol Context (computed from protocol APIs and live on-chain reads) ---\n" - "Every fact below is VERIFIED for this protocol: identities, resolved hashes, decimals, " - "and current values. State them; do not hedge about them or call them unavailable.\n" + protocol_context + "\n--- Protocol Context (verified source, integration references and live on-chain reads) ---\n" + "Resolved identities, hashes, decimals, units and current values are VERIFIED facts. " + "State supplied facts; do not call them unavailable. Distinguish documented integration " + "architecture from live observations, and preserve any explicit validation limits.\n" + protocol_context ) if source_contexts: diff --git a/utils/llm/pendle_context.py b/utils/llm/pendle_context.py new file mode 100644 index 00000000..fa1a4526 --- /dev/null +++ b/utils/llm/pendle_context.py @@ -0,0 +1,247 @@ +"""Enrich PendleSwap upgrades with integration scope and verified implementation code.""" + +import re +from dataclasses import dataclass + +from eth_utils import to_checksum_address + +from utils.calldata.decoder import DecodedCall +from utils.chains import Chain +from utils.llm.report import address_link, checksum_or_none +from utils.logger import get_logger +from utils.proxy import get_current_implementation +from utils.solidity_text import contract_functions, declares_contract, strip_noise, struct_definitions +from utils.source_context import fetch_verified_contract +from utils.verified_contract import VerifiedContract +from utils.web3_wrapper import ChainManager + +logger = get_logger("utils.llm.pendle_context") + +# Pendle's published deployments/1-core.json and deployments/42161-core.json. +PENDLE_SWAP = "0xd4F480965D2347d421F1bEC7F545682E5Ec2151D" +SUPPORTED_CHAINS = {1, 42161} +_REFERENCE_BASE = ( + "https://github.com/pendle-finance/pendle-core-v2-public/blob/87685c89d05087535e9b9647eeda0e3d297d1f06" +) +_ROUTER_REFERENCE = f"{_REFERENCE_BASE}/contracts/router/base/ActionBase.sol" +_INTEGRATION_CONTEXT = ( + "PendleSwap is the external swap-aggregator adapter, separate from Pendle markets, PT/YT " + "contracts and SY implementations. This call upgrades the adapter only.\n" + "Integration reference (standard ActionBase router flow, not a live trace): the router " + "pre-funds PendleSwap with ERC20 input, or sends native input with the swap call. " + "It calls PendleSwap as an optional aggregator leg when minting/redeeming SY. " + "NONE and ETH_WETH branches bypass this adapter. Output returned to msg.sender in this " + "flow goes to the calling router for the next step, not necessarily to the end user. " + "The redeem flow separately checks TokenOutput.minTokenOut, and the mint flow supplies " + "minSyOut to SY.deposit. These are caller-supplied constraints, not proof every caller " + "uses a nonzero minimum.\n" + "Do not infer current balances, route usage, custody guarantees or the impact on all " + "Pendle deposits from this architecture. Assess the affected routes using the old/new " + "code below and the Proxy Upgrade diff; unchanged ABI does not prove unchanged payload " + "semantics. Removing an ODOS scaling branch does not establish removal of all ODOS " + "execution: inspect the needScale=false external-router dispatch separately. Describe " + "a scaler removal as 'removes ODOS calldata scaling' unless the code proves all routes " + "are disabled. Storage UNKNOWN is a validation gap, not a proven collision, and does " + "not by itself establish a higher likelihood of storage collisions. An upgrade-only " + "simulation does not test subsequent swaps. No fixed risk rating follows from this context." +) +_SOURCE_MEMBERS = { + "swap", + "_zeroExSwap", + "_getScaledInputData", + "_approveForExtRouter", + "_safeApproveInfV2", + "_authorizeUpgrade", +} +_OWNER_ABI = [ + { + "type": "function", + "name": "owner", + "stateMutability": "view", + "inputs": [], + "outputs": [{"type": "address", "name": ""}], + } +] + + +@dataclass(frozen=True) +class PendleSwapContext: + """Implementation evidence and current proxy owner for one adapter upgrade.""" + + proxy: str + implementation: str + current_implementation: str | None + owner: str | None + current_source: str + proposed_source: str + is_upgrade_and_call: bool + + @property + def addresses(self) -> list[str]: + """Addresses introduced by this context.""" + return [ + value + for value in ( + self.proxy, + self.implementation, + self.current_implementation, + self.owner, + ) + if value + ] + + @property + def labels(self) -> dict[str, str]: + """Name the known proxy without assuming the replacement's identity.""" + return {self.proxy: "PendleSwap"} + + +def _source_evidence(record: VerifiedContract | None) -> str: + """Extract complete members from the deployed target, plus its enum and ownership guard.""" + if record is None or record.contract_name != "PendleSwap" or not record.compilation_target: + return "PendleSwap target source unavailable or replacement is a different contract." + sections = [f"Verified target: {record.contract_name} in {record.contract_file}"] + members = contract_functions(record.target_source, record.contract_name) or [] + for member in members: + if member.name in _SOURCE_MEMBERS: + sections.append(record.target_source[slice(*member.span)]) + sections.extend(_supporting_evidence(record)) + return "\n\n".join(sections) + + +def _supporting_evidence(record: VerifiedContract) -> list[str]: + """Read uniquely declared payload and ownership definitions from the verified bundle.""" + sections: list[str] = [] + for name in ("IPSwapAggregator", "BoringOwnableUpgradeableV2"): + matches = [(path, source) for path, source in record.sources.items() if declares_contract(source, name)] + if len(matches) != 1: + continue + path, source = matches[0] + if name == "IPSwapAggregator": + sections.append(f"Verified swap payload/enum in {path}:\n{_payload_evidence(source)}") + else: + for member in contract_functions(source, name) or []: + if member.name == "onlyOwner": + sections.append(f"Verified ownership guard in {path}:\n{source[slice(*member.span)]}") + return sections + + +def _payload_evidence(source: str) -> str: + """Extract only swap declarations, even when verification provides a flattened file.""" + sections = [struct_definitions(source, None).get("SwapData", "SwapData declaration unavailable.")] + enums = list(re.finditer(r"\benum\s+SwapType\s*\{[^{}]*\}", strip_noise(source))) + if len(enums) == 1: + sections.append(source[enums[0].start() : enums[0].end()]) + else: + sections.append("SwapType enum unavailable or ambiguous.") + return "\n".join(sections) + + +def _read_owner(chain_id: int, proxy: str) -> str | None: + """Read ownership at the proxy; unavailable state must not hide the source evidence.""" + try: + client = ChainManager.get_client(Chain.from_chain_id(chain_id)) + contract = client.eth.contract(address=to_checksum_address(proxy), abi=_OWNER_ABI) + return to_checksum_address(contract.functions.owner().call()) + except Exception as error: # noqa: BLE001 - optional enrichment must not block the alert + logger.info("PendleSwap owner read failed on chain %s for %s: %s", chain_id, proxy, error) + return None + + +def resolve_pendle_context( + protocol: str, + chain_id: int, + targets_and_calls: list[tuple[str, DecodedCall]], +) -> list[PendleSwapContext]: + """Resolve direct PendleSwap upgrades on Ethereum and Arbitrum, including Safe batches. + + Args: + protocol: Alert protocol, matched case-insensitively. + chain_id: Chain containing the published PendleSwap proxy. + targets_and_calls: Decoded calls and their targets. + + Returns: + Integration context, verified source excerpts and live owner for each + distinct upgrade. Failed enrichments are logged without blocking alerts. + """ + if protocol.lower() != "pendle" or chain_id not in SUPPORTED_CHAINS: + return [] + contexts: list[PendleSwapContext] = [] + seen: set[tuple[str, bool]] = set() + for target, call in targets_and_calls: + implementation = _upgrade_implementation(target, call) + if implementation is None: + continue + is_upgrade_and_call = call.signature == "upgradeToAndCall(address,bytes)" + key = (implementation.lower(), is_upgrade_and_call) + if key in seen: + continue + seen.add(key) + try: + current = get_current_implementation(target, chain_id) + contexts.append( + PendleSwapContext( + proxy=to_checksum_address(target), + implementation=implementation, + current_implementation=current, + owner=_read_owner(chain_id, target), + current_source=_source_evidence(fetch_verified_contract(chain_id, current) if current else None), + proposed_source=_source_evidence(fetch_verified_contract(chain_id, implementation)), + is_upgrade_and_call=is_upgrade_and_call, + ) + ) + except Exception as error: # noqa: BLE001 - retain other batch members on enrichment failure + logger.info("PendleSwap context failed on chain %s for %s: %s", chain_id, implementation, error) + return contexts + + +def _upgrade_implementation(target: str, call: DecodedCall) -> str | None: + """Accept only well-formed upgrade arguments targeting the published swap adapter.""" + if target.lower() != PENDLE_SWAP.lower() or call.signature not in { + "upgradeTo(address)", + "upgradeToAndCall(address,bytes)", + }: + return None + expected_types = ("address", "bytes") if call.signature == "upgradeToAndCall(address,bytes)" else ("address",) + if tuple(param_type for param_type, _ in call.params) != expected_types: + return None + return checksum_or_none(call.params[0][1]) + + +def format_pendle_prompt(contexts: list[PendleSwapContext]) -> str: + """Render integration references separately from live state and verified code.""" + return "\n\n".join(_format_context(context, None, {}) for context in contexts) + + +def format_pendle_report( + contexts: list[PendleSwapContext], + chain_id: int, + labels: dict[str, str], +) -> str: + """Render the same evidence with full explorer links for the gist report.""" + return "\n\n".join(_format_context(context, chain_id, labels) for context in contexts) + + +def _format_context(context: PendleSwapContext, chain_id: int | None, labels: dict[str, str]) -> str: + """Keep prompt and report facts identical, varying only address rendering.""" + + def link(address: str | None) -> str: + """Link known addresses and keep missing state explicit.""" + return address_link(address, chain_id, labels) if address and chain_id else address or "unavailable" + + execution = ( + "upgradeToAndCall includes a bytes payload; inspect its initialization/migration effects separately." + if context.is_upgrade_and_call + else "upgradeTo has no initialization/migration payload." + ) + return ( + f"PendleSwap adapter upgrade on {link(context.proxy)}\n{_INTEGRATION_CONTEXT}\n" + f"Router integration reference: {_ROUTER_REFERENCE}\n" + f"Current implementation: {link(context.current_implementation)}\n" + f"Proposed implementation: {link(context.implementation)}\n" + f"Current proxy owner() (live read): {link(context.owner)}\n{execution}\n" + "Authorization must be assessed from the hook AND ownership guard below; " + "a modifier name alone is not proof.\n" + f"Current implementation evidence:\n```solidity\n{context.current_source}\n```\n" + f"Proposed implementation evidence:\n```solidity\n{context.proposed_source}\n```" + ) diff --git a/utils/llm/protocol_context.py b/utils/llm/protocol_context.py index 00d821d6..90e3619a 100644 --- a/utils/llm/protocol_context.py +++ b/utils/llm/protocol_context.py @@ -36,6 +36,11 @@ format_outland_report, resolve_outland_context, ) +from utils.llm.pendle_context import ( + format_pendle_prompt, + format_pendle_report, + resolve_pendle_context, +) from utils.llm.threejane_context import ( format_threejane_prompt, format_threejane_report, @@ -65,6 +70,7 @@ class _Adapter: _Adapter("infinifi", resolve_infinifi_context, format_infinifi_prompt, format_infinifi_report), _Adapter("infinifi-outland", resolve_outland_context, format_outland_prompt, format_outland_report), _Adapter("3jane", resolve_threejane_context, format_threejane_prompt, format_threejane_report), + _Adapter("pendle", resolve_pendle_context, format_pendle_prompt, format_pendle_report), _Adapter("yearn-v3", resolve_yearn_v3_context, format_yearn_v3_prompt, format_yearn_v3_report), _Adapter( "control-transfer",