#!/usr/bin/env python3 from __future__ import annotations import json import re import time from pathlib import Path ROOT = Path(__file__).resolve().parents[2] PHASE_ORDER = ROOT / "reports" / "extraction" / "promod-uniswap-v2-phase-order-latest.json" DEPLOYMENT_STATUS = ROOT / "cross-chain-pmm-lps" / "config" / "deployment-status.json" REPORT = ROOT / "reports" / "extraction" / "promod-uniswap-v2-phase2-operator-sequence-latest.json" DOC = ROOT / "docs" / "03-deployment" / "PROMOD_UNISWAP_V2_PHASE2_OPERATOR_SEQUENCE.md" RPC_ENV_KEYS = { 1: ["ETHEREUM_MAINNET_RPC"], 10: ["OPTIMISM_RPC_URL", "OPTIMISM_MAINNET_RPC"], 25: ["CRONOS_RPC_URL", "CRONOS_MAINNET_RPC"], 56: ["BSC_RPC_URL", "BSC_MAINNET_RPC"], 100: ["GNOSIS_RPC_URL", "GNOSIS_MAINNET_RPC", "GNOSIS_RPC"], 137: ["POLYGON_MAINNET_RPC", "POLYGON_RPC_URL"], 8453: ["BASE_RPC_URL", "BASE_MAINNET_RPC"], 42161: ["ARBITRUM_RPC_URL", "ARBITRUM_MAINNET_RPC"], 42220: ["CELO_RPC_URL", "CELO_MAINNET_RPC", "CELO_RPC"], 43114: ["AVALANCHE_RPC_URL", "AVALANCHE_MAINNET_RPC"], } def now() -> str: return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) def load(path: Path): return json.loads(path.read_text()) def write_json(path: Path, payload) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(payload, indent=2) + "\n") def write_text(path: Path, text: str) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(text.rstrip() + "\n") def chain_entry(status: dict, chain_id: int) -> dict: return status["chains"].get(str(chain_id)) or status["chains"].get(chain_id) or {} def resolve_token(chain_status: dict, symbol: str) -> str: cw = chain_status.get("cwTokens", {}) value = cw.get(symbol, "") if isinstance(value, dict): return value.get("address") or value.get("token") or "" return value or "" def sanitize(text: str) -> str: return re.sub(r"[^A-Z0-9]+", "_", text.upper()).strip("_") def amount_envs(chain_id: int, pair: str) -> tuple[str, str]: token_a, token_b = pair.split("/") prefix = f"PHASE2_{chain_id}_{sanitize(token_a)}_{sanitize(token_b)}" return f"{prefix}_A_RAW", f"{prefix}_B_RAW" def pair_commands(chain_id: int, pair: str, token_a_addr: str, token_b_addr: str, rpc_key: str) -> dict: factory_var = f"CHAIN_{chain_id}_UNISWAP_V2_FACTORY" router_var = f"CHAIN_{chain_id}_UNISWAP_V2_ROUTER" amount_a_env, amount_b_env = amount_envs(chain_id, pair) token_a, token_b = pair.split("/") prelude = [ "source smom-dbis-138/scripts/load-env.sh >/dev/null", f'export RPC_URL="${{{rpc_key}}}"', f'export FACTORY="${{{factory_var}}}"', f'export ROUTER="${{{router_var}}}"', f'export TOKEN_A="{token_a_addr}"', f'export TOKEN_B="{token_b_addr}"', f'export AMOUNT_A_RAW="${{{amount_a_env}:-}}"', f'export AMOUNT_B_RAW="${{{amount_b_env}:-}}"', 'export SIGNER="$(cast wallet address --private-key "$PRIVATE_KEY")"', 'export DEADLINE="$(( $(date +%s) + 3600 ))"', 'test -n "$AMOUNT_A_RAW" && test -n "$AMOUNT_B_RAW"', ] probe = 'cast call "$FACTORY" \'getPair(address,address)(address)\' "$TOKEN_A" "$TOKEN_B" --rpc-url "$RPC_URL"' create = "\n".join( prelude + [ 'PAIR="$(cast call "$FACTORY" \'getPair(address,address)(address)\' "$TOKEN_A" "$TOKEN_B" --rpc-url "$RPC_URL")"', 'if [[ "$PAIR" == "0x0000000000000000000000000000000000000000" ]]; then', ' cast send "$FACTORY" \'createPair(address,address)(address)\' "$TOKEN_A" "$TOKEN_B" \\', ' --private-key "$PRIVATE_KEY" --rpc-url "$RPC_URL"', "fi", ] ) deploy = "\n".join( prelude + [ 'cast send "$TOKEN_A" \'approve(address,uint256)(bool)\' "$ROUTER" "$AMOUNT_A_RAW" \\', ' --private-key "$PRIVATE_KEY" --rpc-url "$RPC_URL"', "", 'cast send "$TOKEN_B" \'approve(address,uint256)(bool)\' "$ROUTER" "$AMOUNT_B_RAW" \\', ' --private-key "$PRIVATE_KEY" --rpc-url "$RPC_URL"', "", 'cast send "$ROUTER" \\', ' \'addLiquidity(address,address,uint256,uint256,uint256,uint256,address,uint256)\' \\', ' "$TOKEN_A" "$TOKEN_B" "$AMOUNT_A_RAW" "$AMOUNT_B_RAW" "$AMOUNT_A_RAW" "$AMOUNT_B_RAW" "$SIGNER" "$DEADLINE" \\', ' --private-key "$PRIVATE_KEY" --rpc-url "$RPC_URL"', ] ) return { "pair": pair, "token_a": token_a, "token_b": token_b, "token_a_address": token_a_addr, "token_b_address": token_b_addr, "amount_env_a": amount_a_env, "amount_env_b": amount_b_env, "probe_block": probe, "create_if_absent_block": create, "deploy_block": deploy, } def main() -> None: phase = load(PHASE_ORDER) status = load(DEPLOYMENT_STATUS) entries = [] for entry in phase["entries"]: chain_id = entry["chain_id"] chain_status = chain_entry(status, chain_id) rpc_keys = RPC_ENV_KEYS.get(chain_id, []) phase_2_pairs = entry.get("phase_2_full_cw_wrapped_mesh", []) pair_entries = [] for pair in phase_2_pairs: token_a, token_b = pair.split("/") token_a_addr = resolve_token(chain_status, token_a) token_b_addr = resolve_token(chain_status, token_b) if not token_a_addr or not token_b_addr: continue pair_entries.append( pair_commands(chain_id, pair, token_a_addr, token_b_addr, rpc_keys[0] if rpc_keys else "RPC_URL") ) entries.append( { "chain_id": chain_id, "network": entry["network"], "tier": entry["tier"], "phase_1_core_rail": entry["phase_1_core_rail"], "phase_2_pair_count": len(pair_entries), "phase_2_other_gru_v2_cw_tokens": entry.get("phase_2_other_gru_v2_cw_tokens", []), "rpc_env_keys": rpc_keys, "required_uniswap_v2_env_vars": [ f"CHAIN_{chain_id}_UNISWAP_V2_FACTORY", f"CHAIN_{chain_id}_UNISWAP_V2_ROUTER", f"CHAIN_{chain_id}_UNISWAP_V2_START_BLOCK", ], "phase_2_pairs": pair_entries, "post_pair_commands": [ "bash scripts/verify/build-promod-uniswap-v2-live-pair-discovery.sh", "python3 scripts/lib/promod_uniswap_v2_live_pair_discovery.py --write-discovered", "node cross-chain-pmm-lps/scripts/validate-deployment-status.cjs cross-chain-pmm-lps/config/deployment-status.json", "bash scripts/verify/build-promod-uniswap-v2-promotion-gates.sh", ], } ) payload = { "generated_at": now(), "program_name": phase["program_name"], "purpose": "Exact phase-2 live operator sequence for the full cW* wrapped mesh rollout, using per-pair amount envs rather than fixed sizing assumptions.", "mainnet_funding_posture": phase["mainnet_funding_posture"], "entries": entries, "source_artifacts": [ "reports/extraction/promod-uniswap-v2-phase-order-latest.json", "cross-chain-pmm-lps/config/deployment-status.json", ], } write_json(REPORT, payload) lines = [ "# Mr. Promod Uniswap V2 Phase 2 Operator Sequence", "", f"- Generated: `{payload['generated_at']}`", f"- Program: {payload['program_name']}", f"- Mainnet funding posture: `{payload['mainnet_funding_posture']['mode']}` via `{', '.join(payload['mainnet_funding_posture']['required_deployer_assets'])}`", "- Purpose: exact phase-2 live operator sequence for the full `cW*` wrapped mesh rollout.", "- Funding rule: each pair uses explicit raw amount env vars because BTC, gold, FX, and fiat rails do not share a safe default seed size.", "", "| Chain | Network | Phase 2 Pair Count | RPC Keys | Required Env |", "|---|---|---:|---|---|", ] for entry in entries: lines.append( f"| `{entry['chain_id']}` | {entry['network']} | `{entry['phase_2_pair_count']}` | " f"{', '.join(f'`{x}`' for x in entry['rpc_env_keys'])} | " f"{', '.join(f'`{x}`' for x in entry['required_uniswap_v2_env_vars'])} |" ) lines.extend(["", "## Per-Chain Sequence", ""]) for entry in entries: lines.append(f"### Chain `{entry['chain_id']}` — {entry['network']}") lines.append("") lines.append(f"- Phase 1 prerequisite: `{entry['phase_1_core_rail']}`") lines.append(f"- Phase 2 cW* count: `{entry['phase_2_pair_count']}`") lines.append(f"- Other phase-2 cW* tokens: {', '.join(f'`{x}`' for x in entry['phase_2_other_gru_v2_cw_tokens']) or 'none'}") lines.append("") for pair_entry in entry["phase_2_pairs"]: lines.append(f"#### `{pair_entry['pair']}`") lines.append("") lines.append( f"- Amount envs: `{pair_entry['amount_env_a']}`, `{pair_entry['amount_env_b']}`" ) lines.append( f"- Token addresses: `{pair_entry['token_a']}={pair_entry['token_a_address']}`, `{pair_entry['token_b']}={pair_entry['token_b_address']}`" ) lines.append("Probe:") lines.append("```bash") lines.append(pair_entry["probe_block"]) lines.append("```") lines.append("Create if absent:") lines.append("```bash") lines.append(pair_entry["create_if_absent_block"]) lines.append("```") lines.append("Deploy:") lines.append("```bash") lines.append(pair_entry["deploy_block"]) lines.append("```") lines.append("") lines.append("Post-pair refresh commands:") for cmd in entry["post_pair_commands"]: lines.append(f"- `{cmd}`") lines.append("") write_text(DOC, "\n".join(lines)) print(REPORT) if __name__ == "__main__": main()