fix: add a compatibility shim for simulate 3.15 endpoints

pyproject.toml

@@ -8,7 +8,7 @@ readme = "README.md"
 
 [tool.poetry.dependencies]
 python = "^3.10"
-py-algorand-sdk = "2.1.2"
+py-algorand-sdk = "^2.2.0"
 
 [tool.poetry.group.dev.dependencies]
 pytest = "^7.2.0"

src/algokit_utils/_simulate_315_compat.py

@@ -0,0 +1,73 @@
+import base64
+from typing import Any
+
+from algosdk import encoding
+from algosdk.atomic_transaction_composer import (
+    AtomicTransactionComposer,
+    AtomicTransactionComposerStatus,
+    SimulateABIResult,
+    SimulateAtomicTransactionResponse,
+)
+from algosdk.error import AtomicTransactionComposerError
+from algosdk.v2client.algod import AlgodClient
+
+
+def simulate_atc_315(atc: AtomicTransactionComposer, client: AlgodClient) -> SimulateAtomicTransactionResponse:
+    """
+    Ported from algosdk 2.1.2
+
+    Send the transaction group to the `simulate` endpoint and wait for results.
+    An error will be thrown if submission or execution fails.
+    The composer's status must be SUBMITTED or lower before calling this method,
+    since execution is only allowed once.
+
+    Returns:
+        SimulateAtomicTransactionResponse: Object with simulation results for this
+            transaction group, a list of txIDs of the simulated transactions,
+            an array of results for each method call transaction in this group.
+            If a method has no return value (void), then the method results array
+            will contain None for that method's return value.
+    """
+
+    if atc.status > AtomicTransactionComposerStatus.SUBMITTED:
+        raise AtomicTransactionComposerError(  # type: ignore[no-untyped-call]
+            "AtomicTransactionComposerStatus must be submitted or lower to simulate a group"
+        )
+
+    signed_txns = atc.gather_signatures()
+    txn = b"".join(
+        base64.b64decode(encoding.msgpack_encode(txn)) for txn in signed_txns  # type: ignore[no-untyped-call]
+    )
+    simulation_result = client.algod_request(
+        "POST", "/transactions/simulate", data=txn, headers={"Content-Type": "application/x-binary"}
+    )
+    assert isinstance(simulation_result, dict)
+
+    # Only take the first group in the simulate response
+    txn_group: dict[str, Any] = simulation_result["txn-groups"][0]
+    txn_results = txn_group["txn-results"]
+
+    # Parse out abi results
+    results = []
+    for method_index, method in atc.method_dict.items():
+        tx_info = txn_results[method_index]["txn-result"]
+
+        result = atc.parse_result(method, atc.tx_ids[method_index], tx_info)
+        sim_result = SimulateABIResult(
+            tx_id=result.tx_id,
+            raw_value=result.raw_value,
+            return_value=result.return_value,
+            decode_error=result.decode_error,
+            tx_info=result.tx_info,
+            method=result.method,
+        )
+        results.append(sim_result)
+
+    return SimulateAtomicTransactionResponse(
+        version=simulation_result.get("version", 0),
+        failure_message=txn_group.get("failure-message", ""),
+        failed_at=txn_group.get("failed-at"),
+        simulate_response=simulation_result,
+        tx_ids=atc.tx_ids,
+        results=results,
+    )

src/algokit_utils/application_client.py

@@ -4,6 +4,7 @@
 import logging
 import re
 import typing
+from http import HTTPStatus
 from math import ceil
 from pathlib import Path
 from typing import Any, Literal, cast, overload
@@ -20,15 +21,18 @@
     AtomicTransactionResponse,
     LogicSigTransactionSigner,
     MultisigTransactionSigner,
+    SimulateAtomicTransactionResponse,
     TransactionSigner,
     TransactionWithSigner,
 )
 from algosdk.constants import APP_PAGE_MAX_SIZE
+from algosdk.error import AlgodHTTPError
 from algosdk.logic import get_application_address
 from algosdk.source_map import SourceMap
 
 import algokit_utils.application_specification as au_spec
 import algokit_utils.deploy as au_deploy
+from algokit_utils._simulate_315_compat import simulate_atc_315
 from algokit_utils.logic_error import LogicError, parse_logic_error
 from algokit_utils.models import (
     ABIArgsDict,
@@ -165,6 +169,7 @@ def __init__(
         self._approval_program: Program | None = None
         self._approval_source_map: SourceMap | None = None
         self._clear_program: Program | None = None
+        self._use_simulate_315 = False  # flag to determine if old simulate 3.15 encoding should be used
 
         self.template_values: au_deploy.TemplateValueMapping = template_values or {}
         self.existing_deployments = existing_deployments
@@ -865,7 +870,7 @@ def _check_is_compiled(self) -> tuple[Program, Program]:
     def _simulate_readonly_call(
         self, method: Method, atc: AtomicTransactionComposer
     ) -> ABITransactionResponse | TransactionResponse:
-        simulate_response = atc.simulate(self.algod_client)
+        simulate_response = self._simulate_atc(atc)
         if simulate_response.failure_message:
             raise _try_convert_to_logic_error(
                 simulate_response.failure_message,
@@ -877,6 +882,24 @@ def _simulate_readonly_call(
 
         return TransactionResponse.from_atr(simulate_response)
 
+    def _simulate_atc(self, atc: AtomicTransactionComposer) -> SimulateAtomicTransactionResponse:
+        # TODO: remove this once 3.16 is in mainnet
+        # there was a breaking change in algod 3.16 to the simulate endpoint
+        # attempt to transparently handle this by calling the endpoint with the old behaviour if
+        # 3.15 is detected
+        if self._use_simulate_315:
+            return simulate_atc_315(atc, self.algod_client)
+        try:
+            return atc.simulate(self.algod_client)
+        except AlgodHTTPError as ex:
+            if ex.code == HTTPStatus.BAD_REQUEST.value and (
+                "msgpack decode error [pos 12]: no matching struct field found when decoding stream map with key "
+                "txn-groups" in ex.args
+            ):
+                self._use_simulate_315 = True
+                return simulate_atc_315(atc, self.algod_client)
+            raise ex
+
     def _load_reference_and_check_app_id(self) -> None:
         self._load_app_reference()
         self._check_app_id()