feat: cuda arch detection for shorter builds, build failure detection

3 files changed, 324 insertions, 2 deletions

diff --git a/app/tests/test_backends_audiocpp.py b/app/tests/test_backends_audiocpp.py
index 6f97438..4a8ee5f 100644
--- a/app/tests/test_backends_audiocpp.py
+++ b/app/tests/test_backends_audiocpp.py
@@ -1407,6 +1407,189 @@ class BuildAudiocppTests(unittest.TestCase):
         self.assertEqual(len(notices), 1)
         self.assertIn("No build script found", notices[0])
 
+    def test_cuda_build_appends_detected_arch_flag(self):
+        with patch.object(common, "run_console_subprocess",
+                          return_value=0) as run, \
+                patch.object(make_server.build, "detect_cuda_arch",
+                             return_value="86"):
+            make_server.build.build_audiocpp(self.checkout, "cuda")
+        argv = run.call_args[0][0]
+        self.assertIn("--cuda-arch", argv)
+        self.assertEqual(argv[argv.index("--cuda-arch") + 1], "86")
+
+    def test_tui_stall_notice_reports_the_hang(self):
+        emitted, emit = self._emit()
+        with patch.object(common, "LOG_DIR", self.log_dir), \
+                patch.object(common, "run_console_subprocess",
+                             return_value=124):
+            rc = make_server.build.build_audiocpp(self.checkout, "cuda",
+                                            emit=emit)
+        self.assertEqual(rc, 124)
+        log_text = self._log_files()[0].read_text(encoding="utf-8")
+        self.assertIn("stalled", log_text)
+        notices = common.drain_post_tui_notices()
+        self.assertEqual(len(notices), 1)
+        self.assertIn("stalled", notices[0])
+        self.assertIn("stopped", notices[0])
+
+    def test_tui_ptxas_failure_appends_detected_arch_guidance(self):
+        emitted, emit = self._emit()
+
+        def fake_run(argv, cwd=None, emit=None, cancel=None, **kwargs):
+            emit("ptxas fatal   : (C7907) Internal compiler error.")
+            return 1
+
+        with patch.object(common, "LOG_DIR", self.log_dir), \
+                patch.object(common, "run_console_subprocess",
+                             side_effect=fake_run), \
+                patch.object(make_server.build, "detect_cuda_arch",
+                             return_value="86"):
+            rc = make_server.build.build_audiocpp(self.checkout, "cuda",
+                                            emit=emit)
+        self.assertEqual(rc, 1)
+        notice = common.drain_post_tui_notices()[0]
+        self.assertIn("CUDA toolkit bug", notice)
+        self.assertIn("AUDIOCPP_CUDA_ARCH", notice)
+        self.assertIn("Detected arch for this machine: 86", notice)
+        # The undetected-GPU table is not needed when detection worked.
+        self.assertNotIn("Arch per GPU", notice)
+
+    def test_tui_ptxas_failure_lists_gpu_table_when_undetected(self):
+        emitted, emit = self._emit()
+
+        def fake_run(argv, cwd=None, emit=None, cancel=None, **kwargs):
+            emit("ptxas fatal   : (C7907) Internal compiler error.")
+            return 1
+
+        with patch.object(common, "LOG_DIR", self.log_dir), \
+                patch.object(common, "run_console_subprocess",
+                             side_effect=fake_run), \
+                patch.object(make_server.build, "detect_cuda_arch",
+                             return_value=None):
+            rc = make_server.build.build_audiocpp(self.checkout, "cuda",
+                                            emit=emit)
+        self.assertEqual(rc, 1)
+        notice = common.drain_post_tui_notices()[0]
+        self.assertIn("Arch per GPU", notice)
+        self.assertIn("3090", notice)
+        self.assertIn("4090", notice)
+
+    def test_tui_plain_failure_has_no_ptxas_guidance(self):
+        emitted, emit = self._emit()
+        with patch.object(common, "LOG_DIR", self.log_dir), \
+                patch.object(common, "run_console_subprocess",
+                             return_value=3):
+            rc = make_server.build.build_audiocpp(self.checkout, "cuda",
+                                            emit=emit)
+        notice = common.drain_post_tui_notices()[0]
+        self.assertIn("failed (exit code 3)", notice)
+        self.assertNotIn("AUDIOCPP_CUDA_ARCH", notice)
+
+
+class DetectCudaArchTests(unittest.TestCase):
+    """detect_cuda_arch: env override, nvidia-smi probe, None fallbacks."""
+
+    def setUp(self):
+        # Run every case without an AUDIOCPP_CUDA_ARCH leak from the
+        # developer's own shell.
+        clean = {k: v for k, v in os.environ.items()
+                 if k != make_server.build.CUDA_ARCH_ENV}
+        patcher = patch.dict(os.environ, clean, clear=True)
+        patcher.start()
+        self.addCleanup(patcher.stop)
+
+    def test_env_override_wins_without_probing(self):
+        with patch.dict(os.environ,
+                        {make_server.build.CUDA_ARCH_ENV: "86"}), \
+                patch.object(common, "run_console_subprocess_quiet") as run:
+            self.assertEqual(make_server.build.detect_cuda_arch(), "86")
+        run.assert_not_called()
+
+    def test_env_override_multi_gpu_and_commas(self):
+        with patch.dict(os.environ,
+                        {make_server.build.CUDA_ARCH_ENV: "86, 89;75"}):
+            self.assertEqual(make_server.build.detect_cuda_arch(),
+                             "86;89;75")
+
+    def test_env_override_real_virtual_suffixes_allowed(self):
+        with patch.dict(os.environ,
+                        {make_server.build.CUDA_ARCH_ENV: "86-real"}):
+            self.assertEqual(make_server.build.detect_cuda_arch(), "86-real")
+
+    def test_invalid_env_override_ignored_and_probe_runs(self):
+        probe = MagicMock(returncode=0, stdout=b"8.6\n")
+        with patch.dict(os.environ,
+                        {make_server.build.CUDA_ARCH_ENV: "rtx"}), \
+                patch.object(common, "run_console_subprocess_quiet",
+                             return_value=probe) as run:
+            self.assertEqual(make_server.build.detect_cuda_arch(), "86")
+        self.assertEqual(run.call_args[0][0][0], "nvidia-smi")
+
+    def test_compute_caps_parsed_and_deduped(self):
+        probe = MagicMock(returncode=0, stdout=b"8.6\n8.6\n12.0\n")
+        with patch.object(common, "run_console_subprocess_quiet",
+                          return_value=probe):
+            self.assertEqual(make_server.build.detect_cuda_arch(), "86;120")
+
+    def test_nvidia_smi_failure_yields_none(self):
+        probe = MagicMock(returncode=1, stdout=b"")
+        with patch.object(common, "run_console_subprocess_quiet",
+                          return_value=probe):
+            self.assertIsNone(make_server.build.detect_cuda_arch())
+
+    def test_unparsable_output_yields_none(self):
+        probe = MagicMock(returncode=0,
+                          stdout=b"NVIDIA-SMI has failed because...\n")
+        with patch.object(common, "run_console_subprocess_quiet",
+                          return_value=probe):
+            self.assertIsNone(make_server.build.detect_cuda_arch())
+
+    def test_unstartable_probe_yields_none(self):
+        with patch.object(common, "run_console_subprocess_quiet",
+                          return_value=None):
+            self.assertIsNone(make_server.build.detect_cuda_arch())
+
+    def test_probe_is_bounded_by_a_timeout(self):
+        probe = MagicMock(returncode=0, stdout=b"8.6\n")
+        with patch.object(common, "run_console_subprocess_quiet",
+                          return_value=probe) as run:
+            make_server.build.detect_cuda_arch()
+        self.assertIsNotNone(run.call_args[1].get("timeout"))
+
+
+class CudaArchArgvTests(unittest.TestCase):
+    """_cuda_arch_argv: the --cuda-arch flags and their status line."""
+
+    def _argv(self, backend, arch, emit=None):
+        with patch.object(make_server.build, "detect_cuda_arch",
+                          return_value=arch):
+            return make_server.build._cuda_arch_argv(backend, emit=emit)
+
+    def test_cuda_build_gets_the_detected_arch(self):
+        self.assertEqual(self._argv("cuda", "86"),
+                         ["--cuda-arch", "86"])
+
+    def test_multi_gpu_arch_passed_verbatim(self):
+        self.assertEqual(self._argv("cuda", "86;89"),
+                         ["--cuda-arch", "86;89"])
+
+    def test_detection_failure_means_no_flag(self):
+        self.assertEqual(self._argv("cuda", None), [])
+
+    def test_non_cuda_backends_never_get_the_flag(self):
+        for backend in ("cpu", "vulkan", "hip"):
+            with self.subTest(backend=backend):
+                self.assertEqual(self._argv(backend, "86"), [])
+
+    def test_status_lines_report_the_outcome(self):
+        detected, emit = [], lambda line: detected.append(line)
+        self._argv("cuda", "86", emit=emit)
+        self.assertIn("CUDA architecture: 86", detected[0])
+        undetected, emit = [], lambda line: undetected.append(line)
+        self._argv("cuda", None, emit=emit)
+        self.assertIn("portable default list", undetected[0])
+        self.assertIn(make_server.build.CUDA_ARCH_ENV, undetected[0])
+
 
 class AudiocppUpdateTests(unittest.TestCase):
     """update: stop the server, refresh the checkout, rebuild when stale.
diff --git a/app/tests/test_backends_common.py b/app/tests/test_backends_common.py
index a94994b..08ffc2c 100644
--- a/app/tests/test_backends_common.py
+++ b/app/tests/test_backends_common.py
@@ -2,8 +2,9 @@
 
 The streaming mode of ``run_console_subprocess`` (used by the in-TUI task
 view) is exercised with a real child process: output lines are captured and
-forwarded, cancellation kills the child and returns 130, and an on_cancel
-hook runs first.
+forwarded, cancellation kills the child and returns 130, an on_cancel hook
+runs first, and the no-output stall watchdog kills a wedged child and
+returns 124.
 """
 
 import sys
@@ -51,6 +52,58 @@ class RunConsoleSubprocessStreamingTests(unittest.TestCase):
         self.assertEqual(touched, [True])
 
 
+class RunConsoleSubprocessStallTests(unittest.TestCase):
+    """The no-output watchdog: a silent child is killed and reported 124."""
+
+    def test_stall_kills_a_silent_child_and_returns_124(self):
+        lines = []
+        rc = common.run_console_subprocess(
+            [sys.executable, "-c",
+             "import sys, time; print('start', flush=True); "
+             "time.sleep(60)"],
+            emit=lines.append, stall_timeout=0.5)
+        self.assertEqual(rc, 124)
+        self.assertEqual(lines[0], "start")
+        # The stall is announced to the view before the kill.
+        self.assertTrue(any("[ERROR]" in line and "No output" in line
+                            for line in lines), lines)
+
+    def test_no_stall_while_output_keeps_flowing(self):
+        lines = []
+        rc = common.run_console_subprocess(
+            [sys.executable, "-c",
+             "import sys, time\n"
+             "for _ in range(6):\n"
+             "    print('tick', flush=True)\n"
+             "    time.sleep(0.2)\n"],
+            emit=lines.append, stall_timeout=1.0)
+        self.assertEqual(rc, 0)
+        self.assertEqual(lines, ["tick"] * 6)
+
+    def test_console_path_has_no_watchdog(self):
+        # Without emit the child inherits the terminal; stall_timeout is
+        # a no-op there (the caller sees raw output and can Ctrl-C).
+        rc = common.run_console_subprocess(
+            [sys.executable, "-c", "print('hi')"], stall_timeout=0.001)
+        self.assertEqual(rc, 0)
+
+
+class RunConsoleSubprocessQuietTimeoutTests(unittest.TestCase):
+    """A timed-out quiet probe returns a failed result, never raises."""
+
+    def test_timeout_returns_failed_result(self):
+        proc = common.run_console_subprocess_quiet(
+            [sys.executable, "-c", "import time; time.sleep(30)"],
+            timeout=0.5)
+        self.assertIsNotNone(proc)
+        self.assertEqual(proc.returncode, -1)
+
+    def test_untimed_probe_still_reports_the_exit_code(self):
+        proc = common.run_console_subprocess_quiet(
+            [sys.executable, "-c", "import sys; sys.exit(5)"])
+        self.assertEqual(proc.returncode, 5)
+
+
 class GitCloneTests(unittest.TestCase):
     def test_git_clone_console_passes_through(self):
         with mock.patch.object(common, "run_console_subprocess",
diff --git a/app/tests/test_taskview.py b/app/tests/test_taskview.py
index b7f4389..646fccb 100644
--- a/app/tests/test_taskview.py
+++ b/app/tests/test_taskview.py
@@ -592,5 +592,91 @@ class LanesViewTests(_FakeTui, unittest.TestCase):
         self.assertLessEqual(x + len(label) - 1, 1 + pane_w - 2)
 
 
+class SilenceCueTests(_FakeTui, unittest.TestCase):
+    """The "(no output 6m)" cue for a running step that stopped emitting."""
+
+    def test_silence_cue_text_format(self):
+        self.assertEqual(taskview._silence_cue_text(90), "(no output 1m)")
+        self.assertEqual(taskview._silence_cue_text(3700), "(no output 61m)")
+
+    def test_no_cue_while_output_is_recent(self):
+        view, _ = self.make_view(steps=[_step("one")])
+        view.handle_event({"kind": "step_start", "index": 0, "title": "one"})
+        view._ingest_line("working")
+        self.assertIsNone(taskview._silent_secs(view.last_line_at,
+                                                view.last_line_at + 59))
+
+    def test_cue_after_the_threshold(self):
+        view, _ = self.make_view(steps=[_step("one")])
+        view.handle_event({"kind": "step_start", "index": 0, "title": "one"})
+        view._ingest_line("working")
+        self.assertEqual(taskview._silent_secs(view.last_line_at,
+                                               view.last_line_at + 60), 60)
+
+    def test_step_start_resets_and_done_clears_the_tracking(self):
+        view, _ = self.make_view(steps=[_step("one"), _step("two")])
+        view.handle_event({"kind": "step_start", "index": 0, "title": "one"})
+        start = view.last_line_at
+        view.handle_event({"kind": "step_done", "index": 0, "rc": 0})
+        self.assertIsNone(view.last_line_at)
+        view.handle_event({"kind": "step_start", "index": 1, "title": "two"})
+        self.assertGreaterEqual(view.last_line_at, start)
+
+    def test_render_shows_the_cue_for_a_silent_step(self):
+        now = [1000.0]
+
+        def clock():
+            return now[0]
+
+        screen = FakeScreen(width=80, height=24)
+        with patch.object(taskview.TaskView, "_worker_main",
+                          lambda self: None):
+            view = taskview.TaskView(screen, "Setup", [_step("one")],
+                                     clock=clock)
+        view.handle_event({"kind": "step_start", "index": 0, "title": "one"})
+        view._ingest_line("working")
+        now[0] = view.last_line_at + 300
+        view.render()
+        text = " ".join(t for _, _, t, _ in screen.strings)
+        self.assertIn("(no output 5m)", text)
+
+    def test_render_hides_the_cue_when_output_is_fresh(self):
+        now = [1000.0]
+
+        def clock():
+            return now[0]
+
+        screen = FakeScreen(width=80, height=24)
+        with patch.object(taskview.TaskView, "_worker_main",
+                          lambda self: None):
+            view = taskview.TaskView(screen, "Setup", [_step("one")],
+                                     clock=clock)
+        view.handle_event({"kind": "step_start", "index": 0, "title": "one"})
+        view._ingest_line("working")
+        now[0] = view.last_line_at + 5
+        view.render()
+        text = " ".join(t for _, _, t, _ in screen.strings)
+        self.assertNotIn("(no output", text)
+
+    def test_lane_pane_shows_the_cue_for_a_silent_lane(self):
+        now = [1000.0]
+
+        def clock():
+            return now[0]
+
+        screen = FakeScreen(width=80, height=24)
+        view = taskview.LanesView(
+            screen, "Setup",
+            [taskview.TaskLane("Build", [_step("one")])], clock=clock)
+        lane = view._lanes[0]
+        view._handle_lane_event(
+            lane, {"kind": "step_start", "index": 0, "title": "one"})
+        view._ingest_lane_line(lane, "working")
+        now[0] = lane.last_line_at + 120
+        view.render()
+        text = " ".join(t for _, _, t, _ in screen.strings)
+        self.assertIn("(no output 2m)", text)
+
+
 if __name__ == "__main__":
     unittest.main()