Introduce textual as a UI layer for the solve command, allowing better

statistics and understanding of what is going on

Author: damageboy

vxsort/smallsort/codegen/pyproject.toml

@@ -10,6 +10,7 @@ dependencies = [
     "pytest-cov>=7.0.0",
     "tabulate>=0.9.0",
     "rich>=14.3.1",
+    "textual>=0.87.1",
     "zstandard>=0.25.0",
     "z3-solver>=4.15.8.0",
     "psutil>=5.9.0",

vxsort/smallsort/codegen/scripts/ui_audit.sh

@@ -0,0 +1,53 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+# Reproducible audit for UI transition cleanup.
+# 1) Collect coverage from targeted UI tests.
+# 2) Append coverage from a real textual solve run.
+# 3) Report module-level coverage for UI-related modules.
+# 4) Run vulture over UI modules for dead-code hints.
+
+ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+cd "${ROOT_DIR}"
+
+echo "[ui-audit] Erasing old coverage data"
+uv run coverage erase
+
+echo "[ui-audit] Running targeted UI tests with coverage"
+uv run pytest \
+  tests/test_textual_progress_ui.py \
+  tests/test_wave_engine.py \
+  tests/test_instruction_stats.py \
+  --cov=textual_progress_ui \
+  --cov=wave_engine \
+  --cov=instruction_stats \
+  --cov=success_progress \
+  --cov=bitonic_compiler \
+  --cov-report=term-missing \
+  -q
+
+echo "[ui-audit] Running minimal textual solve and appending coverage"
+uv run coverage run --append \
+  --source=bitonic_compiler,success_progress,wave_engine,textual_progress_ui,instruction_stats \
+  src/bitonic_compiler.py \
+  --num-vecs 2 \
+  --vector-machine AVX2 \
+  --datatype i64 \
+  --gadget-depth 1 \
+  --depth-limit 1 \
+  --max-waves 1 \
+  --wave-attempts 100 \
+  --wave-outputs 20 \
+  --runtime-ui textual
+
+echo "[ui-audit] Combined coverage report"
+uv run coverage report -m
+
+echo "[ui-audit] Dead-code hints (vulture)"
+set +e
+uv run vulture src/textual_progress_ui.py src/wave_engine.py src/success_progress.py src/bitonic_compiler.py src/instruction_stats.py
+VULTURE_EXIT=$?
+set -e
+if [[ ${VULTURE_EXIT} -ne 0 ]]; then
+  echo "[ui-audit] vulture returned ${VULTURE_EXIT} (informational findings above)"
+fi

vxsort/smallsort/codegen/src/bitonic_compiler.py

@@ -843,6 +843,13 @@ def main():
         metavar="N",
         help="Budget per stage per wave in distinct outputs (default: 100).",
     )
+    parser.add_argument(
+        "--runtime-ui",
+        type=str,
+        default="auto",
+        choices=["auto", "textual", "none"],
+        help="Wave runtime UI mode. 'auto' uses Textual only on an interactive TTY (default: auto).",
+    )
 
     args = parser.parse_args()
 
@@ -952,6 +959,7 @@ def main():
         llvm_mca_path=args.llvm_mca_path,
         max_waves=args.max_waves,
         top_k=args.top_k or 10,
+        runtime_ui=args.runtime_ui,
     )
 
     engine = WaveEngine(wave_config)

vxsort/smallsort/codegen/src/bitonic_verifier.py

@@ -335,6 +335,7 @@ def _simulate_steps(
         """
         N = 2 * self.elements_per_vector
         num_stages = len(steps)
+        solver = Solver()
 
         elems = [BitVec(f"e_{i}", self.lane_width) for i in range(N)]
 
@@ -347,10 +348,18 @@ def _simulate_steps(
             is_last_stage = step_idx == num_stages - 1
 
             new_top = self._apply_concrete_instructions(
-                top_vec, bottom_vec, gadget.top_instructions, is_top=True
+                top_vec,
+                bottom_vec,
+                gadget.top_instructions,
+                is_top=True,
+                solver=solver,
             )
             new_bottom = self._apply_concrete_instructions(
-                top_vec, bottom_vec, gadget.bottom_instructions, is_top=False
+                top_vec,
+                bottom_vec,
+                gadget.bottom_instructions,
+                is_top=False,
+                solver=solver,
             )
 
             if natural_order and is_last_stage:
@@ -563,7 +572,12 @@ def _extract_output_elements(
         return sorted_elems
 
     def _apply_concrete_instructions(
-        self, top_reg, bottom_reg, instructions: list[InstructionSpec], is_top: bool
+        self,
+        top_reg,
+        bottom_reg,
+        instructions: list[InstructionSpec],
+        is_top: bool,
+        solver: Solver | None = None,
     ):
         """Apply concrete (non-symbolic) instructions to compute an output register.
 
@@ -578,6 +592,8 @@ def _apply_concrete_instructions(
             3+ instruction chains to reference any earlier intermediate.
         """
         current_reg = top_reg if is_top else bottom_reg
+        if solver is None:
+            solver = Solver()
         prev_output = None
         results: list = []
 
@@ -598,7 +614,7 @@ def _apply_concrete_instructions(
                     results,
                 )
 
-            current_reg = self._dispatch_intrinsic(intrinsic, args)
+            current_reg = self._dispatch_intrinsic(intrinsic, args, solver=solver)
             prev_output = current_reg
             results.append(current_reg)
 
@@ -645,6 +661,6 @@ def _resolve_arg(
 
         return value
 
-    def _dispatch_intrinsic(self, intrinsic, args: dict):
+    def _dispatch_intrinsic(self, intrinsic, args: dict, solver: Solver | None = None):
         """Dispatch an intrinsic call using shared signature patterns."""
-        return _dispatch_intrinsic_by_signature(intrinsic, args)
+        return _dispatch_intrinsic_by_signature(intrinsic, args, solver=solver)

vxsort/smallsort/codegen/src/gadget_synthesizer.py

@@ -248,23 +248,25 @@ def _accepts_solver(intrinsic) -> bool:
         return False
 
 
-def _dispatch_intrinsic_fallback(intrinsic, args: dict, solver: Solver):
+def _dispatch_intrinsic_fallback(intrinsic, args: dict, solver: Solver | None = None):
     """Fallback positional dispatch that preserves incoming argument order."""
     arg_order = tuple(args.keys())
-    if _accepts_solver(intrinsic):
+    if solver is not None and _accepts_solver(intrinsic):
         return intrinsic(*(args[key] for key in arg_order), solver=solver)
     return intrinsic(*(args[key] for key in arg_order))
 
 
-def _dispatch_intrinsic_by_signature(intrinsic, args: dict, solver: Solver):
+def _dispatch_intrinsic_by_signature(
+    intrinsic, args: dict, solver: Solver | None = None
+):
     """Call an intrinsic using the first matching dispatch rule.
 
     Falls back to legacy positional call order when no rule matches.
     """
     rule = _match_dispatch_rule(frozenset(args.keys()))
     if rule is None:
         return _dispatch_intrinsic_fallback(intrinsic, args, solver=solver)
-    if _accepts_solver(intrinsic):
+    if solver is not None and _accepts_solver(intrinsic):
         return intrinsic(*(args[key] for key in rule.call_order), solver=solver)
     return intrinsic(*(args[key] for key in rule.call_order))
 

vxsort/smallsort/codegen/src/instruction_stats.py

@@ -32,19 +32,19 @@ class InstructionAttemptStats:
 def normalize_instruction_stats_key(spec: InstructionSpec) -> str:
     """Return a stable key for an instruction family.
 
-    The key is the intrinsic name, optionally suffixed with ``_imm`` for
-    immediate-controlled instructions or ``_v`` for actual control-vector
+    The key is the intrinsic name, optionally suffixed with ``/imm`` for
+    immediate-controlled instructions or ``/v`` for actual control-vector
     instruction families.
     """
     args = spec.args or {}
     arg_names = {str(name).lower() for name in args}
 
     if any(any(marker in name for marker in _IMM_MARKERS) for name in arg_names):
-        return f"{spec.intrinsic_name}_imm"
+        return f"{spec.intrinsic_name}/imm"
 
     intrinsic_name = spec.intrinsic_name.lower()
     if any(marker in intrinsic_name for marker in _CONTROL_VECTOR_MARKERS):
-        return f"{spec.intrinsic_name}_v"
+        return f"{spec.intrinsic_name}/v"
 
     return spec.intrinsic_name