feat: initial draft implementation for debugger support

src/algokit_utils/_debug_utils.py

@@ -2,14 +2,20 @@
 import json
 import logging
 import typing
-from dataclasses import asdict, dataclass, field
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
 from pathlib import Path
 
+from algosdk.atomic_transaction_composer import (
+    AtomicTransactionComposer,
+    EmptySigner,
+    SimulateAtomicTransactionResponse,
+)
 from algosdk.encoding import checksum
 from algosdk.source_map import SourceMap
+from algosdk.v2client.models import SimulateRequest, SimulateRequestTransactionGroup, SimulateTraceConfig
 
 from algokit_utils import deploy
-from algokit_utils.config import config
 
 if typing.TYPE_CHECKING:
     from algosdk.v2client.algod import AlgodClient
@@ -19,6 +25,8 @@
 ALGOKIT_DIR = ".algokit"
 DEBUGGER_DIR = "sourcemaps"
 SOURCES_FILE = "avm.sources"
+TRACES_FILE_EXT = ".avm.trace"
+DEBUG_TRACES_DIR = "debug_traces"
 
 
 @dataclass
@@ -41,10 +49,15 @@ class AVMDebuggerSourceMap:
 
     @classmethod
     def from_dict(cls, data: dict) -> "AVMDebuggerSourceMap":
-        return cls(txn_group_sources=[AVMDebuggerSourceMapEntry(**item) for item in data.get("txn_group_sources", [])])
+        return cls(
+            txn_group_sources=[
+                AVMDebuggerSourceMapEntry(location=item["sourcemap-location"], program_hash=item["hash"])
+                for item in data.get("txn-group-sources", [])
+            ]
+        )
 
-    def __str__(self) -> str:
-        return json.dumps(asdict(self))
+    def to_dict(self) -> dict:
+        return {"txn-group-sources": [json.loads(str(item)) for item in self.txn_group_sources]}
 
 
 @dataclass
@@ -63,20 +76,16 @@ def _load_or_create_sources(project_root: Path) -> AVMDebuggerSourceMap:
         return AVMDebuggerSourceMap.from_dict(json.load(f))
 
 
-def _upsert_debug_sourcemaps(sourcemaps: list[AVMDebuggerSourceMapEntry]) -> None:
-    if not config.project_root:
-        logger.warning("Project root not specified; skipping sourcemap persistence")
-        return
-
-    sources_path = Path(str(config.project_root)) / ALGOKIT_DIR / DEBUGGER_DIR / SOURCES_FILE
+def _upsert_debug_sourcemaps(sourcemaps: list[AVMDebuggerSourceMapEntry], project_root: Path) -> None:
+    sources_path = project_root / ALGOKIT_DIR / DEBUGGER_DIR / SOURCES_FILE
     sources = _load_or_create_sources(sources_path)
 
     for sourcemap in sourcemaps:
         if sourcemap not in sources.txn_group_sources:
             sources.txn_group_sources.append(sourcemap)
 
     with sources_path.open("w") as f:
-        json.dump(sources, f)
+        json.dump(sources.to_dict(), f)
 
 
 def _build_avm_sourcemap(
@@ -99,14 +108,53 @@ def _build_avm_sourcemap(
     return AVMDebuggerSourceMapEntry(str(source_map_output), program_hash)
 
 
-def persist_sourcemaps(sources: list[PersistSourceMapInput], client: "AlgodClient") -> None:
-    if not config.project_root:
-        logger.warning("Project root not specified; skipping persisting sourcemaps")
-        return
-
+def persist_sourcemaps(
+    sources: list[PersistSourceMapInput],
+    project_root: Path,
+    client: "AlgodClient",
+) -> None:
     sourcemaps = [
-        _build_avm_sourcemap(source.teal, source.app_name, source.file_name, config.project_root, client)
-        for source in sources
+        _build_avm_sourcemap(source.teal, source.app_name, source.file_name, project_root, client) for source in sources
     ]
 
-    _upsert_debug_sourcemaps(sourcemaps)
+    _upsert_debug_sourcemaps(sourcemaps, project_root)
+
+
+def simulate_response(atc: AtomicTransactionComposer, algod_client: "AlgodClient") -> SimulateAtomicTransactionResponse:
+    unsigned_txn_groups = atc.build_group()
+    empty_signer = EmptySigner()
+    txn_list = [txn_group.txn for txn_group in unsigned_txn_groups]
+    fake_signed_transactions = empty_signer.sign_transactions(txn_list, [])
+    txn_group = [SimulateRequestTransactionGroup(txns=fake_signed_transactions)]
+    trace_config = SimulateTraceConfig(enable=True, stack_change=True, scratch_change=True)
+
+    simulate_request = SimulateRequest(
+        txn_groups=txn_group, allow_more_logs=True, allow_empty_signatures=True, exec_trace_config=trace_config
+    )
+    return atc.simulate(algod_client, simulate_request)
+
+
+def simulate_and_persist_response(
+    atc: AtomicTransactionComposer, project_root: Path, algod_client: "AlgodClient"
+) -> None:
+    atc_to_simulate = atc.clone()
+    for txn_with_sign in atc_to_simulate.txn_list:
+        sp = algod_client.suggested_params()
+        txn_with_sign.txn.first_valid_round = sp.first
+        txn_with_sign.txn.last_valid_round = sp.last
+        txn_with_sign.txn.genesis_hash = sp.gh
+    response = simulate_response(atc_to_simulate, algod_client)
+    txn_types = [
+        txn_result["txn-results"][0]["txn-result"]["txn"]["txn"]["type"]
+        for txn_result in response.simulate_response["txn-groups"]
+    ]
+    txn_types_count = {txn_type: txn_types.count(txn_type) for txn_type in set(txn_types)}
+    txn_types_str = "_".join([f"{count}#{txn_type}" for txn_type, count in txn_types_count.items()])
+    last_round = response.simulate_response["last-round"]
+    output_file = (
+        project_root
+        / DEBUG_TRACES_DIR
+        / f'{datetime.now(tz=timezone.utc).strftime("%Y%m%d_%H%M%S")}_lr{last_round}_{txn_types_str}{TRACES_FILE_EXT}'
+    )
+    output_file.parent.mkdir(parents=True, exist_ok=True)
+    output_file.write_text(json.dumps(response.simulate_response, indent=2))

src/algokit_utils/application_client.py

@@ -4,7 +4,6 @@
 import logging
 import re
 import typing
-from datetime import datetime, timezone
 from math import ceil
 from pathlib import Path
 from typing import Any, Literal, cast, overload
@@ -19,7 +18,6 @@
     AccountTransactionSigner,
     AtomicTransactionComposer,
     AtomicTransactionResponse,
-    EmptySigner,
     LogicSigTransactionSigner,
     MultisigTransactionSigner,
     SimulateAtomicTransactionResponse,
@@ -29,11 +27,15 @@
 from algosdk.constants import APP_PAGE_MAX_SIZE
 from algosdk.logic import get_application_address
 from algosdk.source_map import SourceMap
-from algosdk.v2client.models import SimulateRequest, SimulateRequestTransactionGroup, SimulateTraceConfig
 
 import algokit_utils.application_specification as au_spec
 import algokit_utils.deploy as au_deploy
-from algokit_utils._debug_utils import PersistSourceMapInput, persist_sourcemaps
+from algokit_utils._debug_utils import (
+    PersistSourceMapInput,
+    persist_sourcemaps,
+    simulate_and_persist_response,
+    simulate_response,
+)
 from algokit_utils.config import config
 from algokit_utils.logic_error import LogicError, parse_logic_error
 from algokit_utils.models import (
@@ -349,7 +351,7 @@ def deploy(
             self.algod_client, self.app_spec, template_values
         )
 
-        if config.debug:
+        if config.debug and config.project_root:
             persist_sourcemaps(
                 sources=[
                     PersistSourceMapInput(
@@ -359,6 +361,7 @@ def deploy(
                         teal=self._clear_program.teal, app_name=self.app_name, file_name="clear.teal"
                     ),
                 ],
+                project_root=config.project_root,
                 client=self.algod_client,
             )
 
@@ -882,7 +885,7 @@ def _check_is_compiled(self) -> tuple[Program, Program]:
                 self.algod_client, self.app_spec, self.template_values
             )
 
-        if config.debug:
+        if config.debug and config.project_root:
             persist_sourcemaps(
                 sources=[
                     PersistSourceMapInput(
@@ -892,6 +895,7 @@ def _check_is_compiled(self) -> tuple[Program, Program]:
                         teal=self._clear_program.teal, app_name=self.app_name, file_name="clear.teal"
                     ),
                 ],
+                project_root=config.project_root,
                 client=self.algod_client,
             )
 
@@ -900,21 +904,19 @@ def _check_is_compiled(self) -> tuple[Program, Program]:
     def _simulate_readonly_call(
         self, method: Method, atc: AtomicTransactionComposer
     ) -> ABITransactionResponse | TransactionResponse:
-        simulate_response = _simulate_response(atc, self.algod_client)
+        response = simulate_response(atc, self.algod_client)
         traces = None
         if config.debug:
-            traces = _create_simulate_traces(simulate_response)
-        if simulate_response.failure_message:
+            traces = _create_simulate_traces(response)
+        if response.failure_message:
             raise _try_convert_to_logic_error(
-                simulate_response.failure_message,
+                response.failure_message,
                 self.app_spec.approval_program,
                 self._get_approval_source_map,
                 traces,
-            ) or Exception(
-                f"Simulate failed for readonly method {method.get_signature()}: {simulate_response.failure_message}"
-            )
+            ) or Exception(f"Simulate failed for readonly method {method.get_signature()}: {response.failure_message}")
 
-        return TransactionResponse.from_atr(simulate_response)
+        return TransactionResponse.from_atr(response)
 
     def _load_reference_and_check_app_id(self) -> None:
         self._load_app_reference()
@@ -1295,10 +1297,13 @@ def execute_atc_with_logic_error(
     ```
     """
     try:
+        if config.debug and config.project_root and config.trace_all:
+            simulate_and_persist_response(atc, config.project_root, algod_client)
+
         return atc.execute(algod_client, wait_rounds=wait_rounds)
     except Exception as ex:
         if config.debug:
-            simulate = _simulate_response(atc, algod_client)
+            simulate = simulate_response(atc, algod_client)
             traces = _create_simulate_traces(simulate)
         else:
             traces = None
@@ -1309,29 +1314,6 @@ def execute_atc_with_logic_error(
         if logic_error:
             raise logic_error from ex
         raise ex
-    finally:
-        if config.debug:
-            if not config.project_root:
-                logger.warning(
-                    "No project root specified, unable to save simulated traces. "
-                    "To save simulated traces, set config.project_root to a directory or execute "
-                    "from an algokit project."
-                )
-            elif config.trace_all:
-                atc_to_simulate = atc.clone()
-                for txn_with_sign in atc_to_simulate.txn_list:
-                    sp = algod_client.suggested_params()
-                    txn_with_sign.txn.first_valid_round = sp.first
-                    txn_with_sign.txn.last_valid_round = sp.last
-                    txn_with_sign.txn.genesis_hash = sp.gh
-                simulate = _simulate_response(atc_to_simulate, algod_client)
-                output_file = (
-                    config.project_root
-                    / "debug_traces"
-                    / f'{datetime.now(tz=timezone.utc).strftime("%Y%m%d_%H%M%S")}.avm.trace'
-                )
-                output_file.parent.mkdir(parents=True, exist_ok=True)
-                output_file.write_text(json.dumps(simulate.simulate_response, indent=2))
 
 
 def _create_simulate_traces(simulate: SimulateAtomicTransactionResponse) -> list[dict[str, Any]]:
@@ -1354,22 +1336,6 @@ def _create_simulate_traces(simulate: SimulateAtomicTransactionResponse) -> list
     return traces
 
 
-def _simulate_response(
-    atc: AtomicTransactionComposer, algod_client: "AlgodClient"
-) -> SimulateAtomicTransactionResponse:
-    unsigned_txn_groups = atc.build_group()
-    empty_signer = EmptySigner()
-    txn_list = [txn_group.txn for txn_group in unsigned_txn_groups]
-    fake_signed_transactions = empty_signer.sign_transactions(txn_list, [])
-    txn_group = [SimulateRequestTransactionGroup(txns=fake_signed_transactions)]
-    trace_config = SimulateTraceConfig(enable=True, stack_change=True, scratch_change=True)
-
-    simulate_request = SimulateRequest(
-        txn_groups=txn_group, allow_more_logs=True, allow_empty_signatures=True, exec_trace_config=trace_config
-    )
-    return atc.simulate(algod_client, simulate_request)
-
-
 def _convert_transaction_parameters(
     args: TransactionParameters | TransactionParametersDict | None,
 ) -> CreateCallParameters:

tests/conftest.py

@@ -2,10 +2,8 @@
 import math
 import random
 import subprocess
-from collections.abc import Generator
 from pathlib import Path
 from typing import TYPE_CHECKING
-from unittest.mock import Mock, patch
 from uuid import uuid4
 
 import algosdk.transaction
@@ -163,15 +161,6 @@ def app_spec() -> ApplicationSpecification:
     return read_spec("app_client_test.json", deletable=True, updatable=True, template_values={"VERSION": 1})
 
 
-# This fixture is automatically applied to all application call tests.
-# If you need to run a test without debug mode, you can reference this mock within the test and disable it explicitly.
-@pytest.fixture(autouse=True)
-def mock_config() -> Generator[Mock, None, None]:
-    with patch("algokit_utils.application_client.config", new_callable=Mock) as mock_config:
-        mock_config.debug = True
-        yield mock_config
-
-
 def generate_test_asset(algod_client: "AlgodClient", sender: Account, total: int | None) -> int:
     if total is None:
         total = math.floor(random.random() * 100) + 20

tests/test_app_client_call.py

@@ -1,6 +1,7 @@
+from collections.abc import Generator
 from pathlib import Path
 from typing import TYPE_CHECKING
-from unittest.mock import Mock
+from unittest.mock import Mock, patch
 
 import algokit_utils
 import pytest
@@ -35,6 +36,16 @@ def client_fixture(algod_client: "AlgodClient", app_spec: ApplicationSpecificati
     return client
 
 
+# This fixture is automatically applied to all application call tests.
+# If you need to run a test without debug mode, you can reference this mock within the test and disable it explicitly.
+@pytest.fixture(autouse=True)
+def mock_config() -> Generator[Mock, None, None]:
+    with patch("algokit_utils.application_client.config", new_callable=Mock) as mock_config:
+        mock_config.debug = True
+        mock_config.project_root = None
+        yield mock_config
+
+
 def test_app_client_from_app_spec_path(algod_client: "AlgodClient") -> None:
     client = ApplicationClient(algod_client, Path(__file__).parent / "app_client_test.json")
 

tests/test_debug_utils.approvals/test_build_teal_sourcemaps.approved.txt

@@ -0,0 +1 @@
+{"txn-group-sources": [{"sourcemap-location": "dummy", "hash": "p0zHRDBy7V/S2TDcAaXYEd2OF8KSxV/1y0ceYEL6qxU="}, {"sourcemap-location": "dummy", "hash": "p0zHRDBy7V/S2TDcAaXYEd2OF8KSxV/1y0ceYEL6qxU="}]}

tests/test_debug_utils.py

@@ -1,16 +1,17 @@
+import json
 from typing import TYPE_CHECKING
 
 import pytest
-from algokit_utils._debug_utils import PersistSourceMapInput, persist_sourcemaps
-from algokit_utils.config import config
+from algokit_utils._debug_utils import AVMDebuggerSourceMap, PersistSourceMapInput, persist_sourcemaps
+
+from tests.conftest import check_output_stability
 
 if TYPE_CHECKING:
     from algosdk.v2client.algod import AlgodClient
 
 
 def test_build_teal_sourcemaps(algod_client: "AlgodClient", tmp_path_factory: pytest.TempPathFactory) -> None:
     cwd = tmp_path_factory.mktemp("cwd")
-    config.configure(debug=True, project_root=cwd)
 
     approval = """
 #pragma version 6
@@ -26,4 +27,20 @@ def test_build_teal_sourcemaps(algod_client: "AlgodClient", tmp_path_factory: py
         PersistSourceMapInput(teal=clear, app_name="cool_app", file_name="clear"),
     ]
 
-    persist_sourcemaps(sources=sources, client=algod_client)
+    persist_sourcemaps(sources=sources, project_root=cwd, client=algod_client)
+
+    root_path = cwd / ".algokit" / "sourcemaps"
+    sourcemap_file_path = root_path / "avm.sources"
+    app_output_path = root_path / "cool_app"
+
+    assert (sourcemap_file_path).exists()
+    assert (app_output_path / "approval.teal").exists()
+    assert (app_output_path / "approval.tok.map").exists()
+    assert (app_output_path / "clear.teal").exists()
+    assert (app_output_path / "clear.tok.map").exists()
+
+    result = AVMDebuggerSourceMap.from_dict(json.loads(sourcemap_file_path.read_text()))
+    for item in result.txn_group_sources:
+        item.location = "dummy"
+
+    check_output_stability(json.dumps(result.to_dict()))