feat: improve logging (#418)

src/halmos/__main__.py

@@ -3,6 +3,7 @@
 import gc
 import io
 import json
+import logging
 import os
 import re
 import signal
@@ -43,6 +44,20 @@
 from .config import Config as HalmosConfig
 from .config import arg_parser, default_config, resolve_config_files, toml_parser
 from .exceptions import HalmosException
+from .logs import (
+    COUNTEREXAMPLE_INVALID,
+    COUNTEREXAMPLE_UNKNOWN,
+    INTERNAL_ERROR,
+    LOOP_BOUND,
+    PARSING_ERROR,
+    REVERT_ALL,
+    debug,
+    error,
+    logger,
+    logger_unique,
+    warn,
+    warn_code,
+)
 from .mapper import BuildOut, DeployAddressMapper, Mapper
 from .sevm import (
     EMPTY_BALANCE,
@@ -77,26 +92,14 @@
     con,
     create_solver,
     cyan,
-    debug,
-    error,
     green,
     hexify,
     indent_text,
     red,
     stringify,
     unbox_int,
-    warn,
     yellow,
 )
-from .warnings import (
-    COUNTEREXAMPLE_INVALID,
-    COUNTEREXAMPLE_UNKNOWN,
-    INTERNAL_ERROR,
-    LOOP_BOUND,
-    PARSING_ERROR,
-    REVERT_ALL,
-    warn_code,
-)
 
 StrModel = dict[str, str]
 AnyModel = StrModel | str
@@ -486,8 +489,7 @@ def setup(
             raise HalmosException(f"No successful path found in {setup_sig}")
 
         if len(setup_exs) > 1:
-            if args.debug:
-                print("\n".join(map(str, setup_exs)))
+            debug("\n".join(map(str, setup_exs)))
 
             raise HalmosException(f"Multiple paths were found in {setup_sig}")
 
@@ -501,8 +503,7 @@ def setup(
                 LOOP_BOUND,
                 f"{setup_sig}: paths have not been fully explored due to the loop unrolling bound: {args.loop}",
             )
-            if args.debug:
-                print("\n".join(jumpid_str(x) for x in sevm.logs.bounded_loops))
+            debug("\n".join(jumpid_str(x) for x in sevm.logs.bounded_loops))
 
     if args.statistics:
         print(setup_timer.report())
@@ -794,8 +795,7 @@ def future_callback(future_model):
             LOOP_BOUND,
             f"{funsig}: paths have not been fully explored due to the loop unrolling bound: {args.loop}",
         )
-        if args.debug:
-            print("\n".join(jumpid_str(x) for x in logs.bounded_loops))
+        debug("\n".join(jumpid_str(x) for x in logs.bounded_loops))
 
     # print post-states
     if args.print_states:
@@ -893,8 +893,7 @@ def run_sequential(run_args: RunArgs) -> list[TestResult]:
         try:
             test_config = with_devdoc(args, funsig, run_args.contract_json)
             solver = mk_solver(test_config)
-            if test_config.debug:
-                debug(f"{test_config.formatted_layers()}")
+            debug(f"{test_config.formatted_layers()}")
             test_result = run(setup_ex, run_args.abi, fun_info, test_config, solver)
         except Exception as err:
             print(f"{color_error('[ERROR]')} {funsig}")
@@ -960,8 +959,8 @@ def solve(
             )
 
         with open(dump_filename, "w") as f:
-            if args.debug:
-                print(f"Writing SMT query to {dump_filename}")
+            if args.verbose >= 1:
+                debug(f"Writing SMT query to {dump_filename}")
             if args.cache_solver:
                 f.write("(set-option :produce-unsat-cores true)\n")
             f.write("(set-logic QF_AUFBV)\n")
@@ -974,9 +973,9 @@ def solve(
                 f.write("(get-unsat-core)\n")
 
     if args.solver_command:
-        if args.debug:
-            print("  Checking with external solver process")
-            print(f"    {args.solver_command} {dump_filename} >{dump_filename}.out")
+        if args.verbose >= 1:
+            debug("  Checking with external solver process")
+            debug(f"    {args.solver_command} {dump_filename} >{dump_filename}.out")
 
         # solver_timeout_assertion == 0 means no timeout,
         # which translates to timeout_seconds=None for subprocess.run
@@ -994,8 +993,8 @@ def solve(
             with open(f"{dump_filename}.out", "w") as f:
                 f.write(res_str)
 
-            if args.debug:
-                print(f"    {res_str_head}")
+            if args.verbose >= 1:
+                debug(f"    {res_str_head}")
 
             if res_str_head == "unsat":
                 unsat_core = parse_unsat_core(res_str) if args.cache_solver else None
@@ -1189,8 +1188,7 @@ def parse_build_out(args: HalmosConfig) -> dict:
                 # - defines only free functions
                 # - ...
                 if contract_type is None:
-                    if args.debug:
-                        debug(f"Skipped {json_filename}, no contract definition found")
+                    debug(f"Skipped {json_filename}, no contract definition found")
                     continue
 
                 compiler_version = json_out["metadata"]["compiler"]["version"]
@@ -1230,12 +1228,11 @@ def parse_symbols(args: HalmosConfig, contract_map: dict, contract_name: str) ->
         Mapper().parse_ast(json_out["ast"])
 
     except Exception:
-        if args.debug:
-            debug(f"error parsing symbols for contract {contract_name}")
-            debug(traceback.format_exc())
-        else:
-            # we parse symbols as best effort, don't propagate exceptions
-            pass
+        debug(f"error parsing symbols for contract {contract_name}")
+        debug(traceback.format_exc())
+
+        # we parse symbols as best effort, don't propagate exceptions
+        pass
 
 
 def parse_devdoc(funsig: str, contract_json: dict) -> str | None:
@@ -1351,6 +1348,10 @@ def _main(_args=None) -> MainResult:
     if args.disable_gc:
         gc.disable()
 
+    if args.debug:
+        logger.setLevel(logging.DEBUG)
+        logger_unique.setLevel(logging.DEBUG)
+
     #
     # compile
     #
@@ -1367,8 +1368,7 @@ def _main(_args=None) -> MainResult:
     ]
 
     # run forge without capturing stdout/stderr
-    if args.debug:
-        debug(f"Running {' '.join(build_cmd)}")
+    debug(f"Running {' '.join(build_cmd)}")
 
     build_exitcode = subprocess.run(build_cmd).returncode
 
@@ -1400,16 +1400,14 @@ def on_exit(exitcode: int) -> MainResult:
         result = MainResult(exitcode, test_results_map)
 
         if args.json_output:
-            if args.debug:
-                debug(f"Writing output to {args.json_output}")
+            debug(f"Writing output to {args.json_output}")
             with open(args.json_output, "w") as json_file:
                 json.dump(asdict(result), json_file, indent=4)
 
         return result
 
     def on_signal(signum, frame):
-        if args.debug:
-            debug(f"Signal {signum} received")
+        debug(f"Signal {signum} received")
         exitcode = 128 + signum
         on_exit(exitcode)
         sys.exit(exitcode)

src/halmos/assertions.py

@@ -13,8 +13,8 @@
     is_bv,
 )
 
-from halmos.exceptions import HalmosException
-from halmos.utils import (
+from .exceptions import HalmosException
+from .utils import (
     bytes_to_bv_value,
     extract_bytes,
     extract_bytes32_array_argument,

src/halmos/bytevec.py

@@ -6,6 +6,7 @@
 from sortedcontainers import SortedDict
 from z3 import BitVecRef, If, eq, is_bool, is_bv, is_bv_value, simplify, substitute
 
+from .logs import warn
 from .utils import (
     Byte,
     Word,
@@ -16,7 +17,6 @@
     extract_bytes,
     try_bv_value_to_bytes,
     unbox_int,
-    warn,
 )
 
 UnwrappedBytes = bytes | Byte

src/halmos/calldata.py

@@ -12,6 +12,7 @@
 
 from .bytevec import ByteVec
 from .config import Config as HalmosConfig
+from .logs import debug_once
 from .utils import con, uid
 
 
@@ -139,10 +140,9 @@ def get_dyn_sizes(self, name: str, typ: Type) -> tuple[list[int], BitVecRef]:
                 if isinstance(typ, DynamicArrayType)
                 else self.args.default_bytes_lengths  # bytes or string
             )
-            if self.args.debug:
-                print(
-                    f"Warning: no size provided for {name}; default value {sizes} will be used."
-                )
+            debug_once(
+                f"no size provided for {name}; default value {sizes} will be used."
+            )
 
         size_var = BitVec(f"p_{name}_length_{uid()}_{self.new_symbol_id():>02}", 256)
 

src/halmos/cheatcodes.py

@@ -31,6 +31,7 @@
     mk_calldata,
 )
 from .exceptions import FailCheatcode, HalmosException, InfeasiblePath, NotConcreteError
+from .logs import debug
 from .mapper import BuildOut
 from .utils import (
     Address,
@@ -740,9 +741,7 @@ def handle(sevm, ex, arg: ByteVec, stack, step_id) -> ByteVec | None:
 
             cmd = extract_string_array_argument(arg, 0)
 
-            debug = sevm.options.debug
-            verbose = sevm.options.verbose
-            if debug or verbose:
+            if sevm.options.debug or sevm.options.verbose:
                 print(f"[vm.ffi] {cmd}")
 
             process = Popen(cmd, stdout=PIPE, stderr=PIPE)
@@ -754,8 +753,7 @@ def handle(sevm, ex, arg: ByteVec, stack, step_id) -> ByteVec | None:
 
             out_str = stdout.decode("utf-8").strip()
 
-            if debug:
-                print(f"[vm.ffi] {cmd}, stdout: {green(out_str)}")
+            debug(f"[vm.ffi] {cmd}, stdout: {green(out_str)}")
 
             if decode_hex(out_str) is not None:
                 # encode hex strings as is for compatibility with foundry's ffi

src/halmos/config.py

@@ -10,7 +10,7 @@
 
 import toml
 
-from .utils import warn
+from .logs import warn
 
 # common strings
 internal = "internal"

src/halmos/console.py

@@ -1,21 +1,23 @@
 from z3 import BitVec, BitVecRef
 
+from .logs import (
+    info,
+    warn,
+)
 from .utils import (
     con_addr,
     extract_bytes,
     extract_bytes_argument,
     extract_funsig,
     extract_string_argument,
     hexify,
-    info,
     int_of,
     magenta,
     render_address,
     render_bool,
     render_bytes,
     render_int,
     render_uint,
-    warn,
 )
 
 

src/halmos/logs.py

@@ -0,0 +1,84 @@
+import logging
+from dataclasses import dataclass
+
+from rich.logging import RichHandler
+
+#
+# Basic logging
+#
+
+logging.basicConfig(
+    format="%(message)s",
+    handlers=[RichHandler(level=logging.NOTSET, show_time=False)],
+)
+
+logger = logging.getLogger("halmos")
+
+
+def debug(text: str) -> None:
+    logger.debug(text)
+
+
+def info(text: str) -> None:
+    logger.info(text)
+
+
+def warn(text: str) -> None:
+    logger.warning(text)
+
+
+def error(text: str) -> None:
+    logger.error(text)
+
+
+#
+# Logging with filtering out duplicate log messages
+#
+
+
+class UniqueLoggingFilter(logging.Filter):
+    def __init__(self):
+        self.records = set()
+
+    def filter(self, record):
+        if record.msg in self.records:
+            return False
+        self.records.add(record.msg)
+        return True
+
+
+logger_unique = logging.getLogger("halmos.unique")
+logger_unique.addFilter(UniqueLoggingFilter())
+
+
+def debug_once(text: str) -> None:
+    logger_unique.debug(text)
+
+
+#
+# Warnings with error code
+#
+
+WARNINGS_BASE_URL = "https://github.com/a16z/halmos/wiki/warnings"
+
+
+@dataclass
+class ErrorCode:
+    code: str
+
+    def url(self) -> str:
+        return f"{WARNINGS_BASE_URL}#{self.code}"
+
+
+PARSING_ERROR = ErrorCode("parsing-error")
+INTERNAL_ERROR = ErrorCode("internal-error")
+LIBRARY_PLACEHOLDER = ErrorCode("library-placeholder")
+COUNTEREXAMPLE_INVALID = ErrorCode("counterexample-invalid")
+COUNTEREXAMPLE_UNKNOWN = ErrorCode("counterexample-unknown")
+UNSUPPORTED_OPCODE = ErrorCode("unsupported-opcode")
+REVERT_ALL = ErrorCode("revert-all")
+LOOP_BOUND = ErrorCode("loop-bound")
+
+
+def warn_code(error_code: ErrorCode, msg: str):
+    logger.warning(f"{msg}\n(see {error_code.url()})")