From fc26e965bbecb540965300b097ae2688f375b8ec Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Tue, 28 Jul 2026 09:29:57 -0700 Subject: [PATCH] fix(tools): apply idle timeout to command STT runner (class fix for #50081) Port the progress-based idle-timeout pattern from _run_command_tts (PR #50087, @CleanDev-Fix) to _run_command_stt: the timeout resets on any stdout/stderr output, so a slow-but-alive STT provider survives while a silently stalled one is killed. Stuck detection stays progress-based, never wall-clock. --- tests/tools/test_transcription_tools.py | 60 ++++++++++++++ tools/transcription_tools.py | 105 ++++++++++++++++++++---- 2 files changed, 149 insertions(+), 16 deletions(-) diff --git a/tests/tools/test_transcription_tools.py b/tests/tools/test_transcription_tools.py index 6fdf9dd6cff4..c5e39728c403 100644 --- a/tests/tools/test_transcription_tools.py +++ b/tests/tools/test_transcription_tools.py @@ -2482,3 +2482,63 @@ class TestTranscribeCredentialReadGuard: # The error is the shared read-guard message, not an audio-validation # or provider error — proving the guard fired before dispatch. assert result["error"] == expected + + +class TestRunCommandSttIdleTimeout: + """_run_command_stt uses a progress-based idle timeout (mirrors TTS runner).""" + + @staticmethod + def _shell_command(*args): + import shlex + if os.name == "nt": + return subprocess.list2cmdline(list(args)) + return " ".join(shlex.quote(str(arg)) for arg in args) + + def test_stderr_progress_extends_beyond_timeout(self, tmp_path): + """A slow-but-alive command that keeps emitting output survives an + idle timeout shorter than its total runtime.""" + from tools.transcription_tools import _run_command_stt + + script = tmp_path / "progress_then_exit.py" + script.write_text( + "\n".join([ + "import sys, time", + "for idx in range(4):", + " print(f'tick {idx}', file=sys.stderr, flush=True)", + " time.sleep(0.15)", + "print('done', flush=True)", + ]), + encoding="utf-8", + ) + + result = _run_command_stt( + self._shell_command(sys.executable, "-u", str(script)), + timeout=0.25, + ) + + assert result.returncode == 0 + assert "tick 3" in result.stderr + assert "done" in result.stdout + + def test_silent_stall_still_times_out(self, tmp_path): + """A silently stalled command is killed once the idle window elapses, + and pre-stall output is preserved on the TimeoutExpired.""" + from tools.transcription_tools import _run_command_stt + + script = tmp_path / "progress_then_hang.py" + script.write_text( + "\n".join([ + "import sys, time", + "print('starting pass 1', file=sys.stderr, flush=True)", + "time.sleep(1.0)", + ]), + encoding="utf-8", + ) + + with pytest.raises(subprocess.TimeoutExpired) as excinfo: + _run_command_stt( + self._shell_command(sys.executable, "-u", str(script)), + timeout=0.2, + ) + + assert "starting pass 1" in (excinfo.value.stderr or "") diff --git a/tools/transcription_tools.py b/tools/transcription_tools.py index 877f799621c0..042f05aa646c 100644 --- a/tools/transcription_tools.py +++ b/tools/transcription_tools.py @@ -30,12 +30,14 @@ Usage:: import logging import os import platform +import queue import re import shlex import shutil import subprocess import tempfile import threading +import time from pathlib import Path from typing import Optional, Dict, Any from urllib.parse import urljoin @@ -653,9 +655,12 @@ def _terminate_command_stt_process_tree(proc: subprocess.Popen) -> None: def _run_command_stt(command: str, timeout: float) -> subprocess.CompletedProcess: - """Run a command-provider shell command with process-tree timeout cleanup. + """Run a command-provider shell command with process-tree idle cleanup. - Mirrors ``tools.tts_tool._run_command_tts``. + Mirrors ``tools.tts_tool._run_command_tts``: ``timeout`` is an IDLE + timeout, reset whenever the command emits output on stdout/stderr — + a slow-but-alive provider survives, a silently stalled one is killed + (same progress-based stuck detection as the TTS runner, #50081). Child env is scrubbed of Hermes secrets (salvage of #56332) while still propagating delegated-child lineage markers when applicable. """ @@ -680,21 +685,89 @@ def _run_command_stt(command: str, timeout: float) -> subprocess.CompletedProces popen_kwargs["start_new_session"] = True proc = subprocess.Popen(command, **popen_kwargs, stdin=subprocess.DEVNULL) - try: - stdout, stderr = proc.communicate(timeout=timeout) - except subprocess.TimeoutExpired as exc: - _terminate_command_stt_process_tree(proc) + output_queue: "queue.Queue[tuple[str, Optional[str]]]" = queue.Queue() + chunks: Dict[str, list] = {"stdout": [], "stderr": []} + open_streams = {"stdout", "stderr"} + + def read_stream(name: str, stream: Any) -> None: + encoding = getattr(stream, "encoding", None) or "utf-8" + read1 = getattr(getattr(stream, "buffer", None), "read1", None) try: - stdout, stderr = proc.communicate(timeout=1) - except Exception: - stdout = getattr(exc, "output", None) - stderr = getattr(exc, "stderr", None) - raise subprocess.TimeoutExpired( - command, - timeout, - output=stdout, - stderr=stderr, - ) from exc + while True: + if read1 is None: + chunk = stream.read(65536) + else: + data = read1(65536) + chunk = data.decode(encoding, errors="replace") + if not chunk: + break + output_queue.put((name, chunk)) + finally: + output_queue.put((name, None)) + + readers = [ + threading.Thread( + target=read_stream, + args=("stdout", proc.stdout), + daemon=True, + ), + threading.Thread( + target=read_stream, + args=("stderr", proc.stderr), + daemon=True, + ), + ] + for reader in readers: + reader.start() + + deadline = time.monotonic() + timeout + timed_out = False + while open_streams: + remaining = deadline - time.monotonic() + if remaining <= 0: + timed_out = True + break + try: + name, chunk = output_queue.get(timeout=min(0.05, remaining)) + except queue.Empty: + continue + if chunk is None: + open_streams.discard(name) + continue + chunks[name].append(chunk) + deadline = time.monotonic() + timeout + + if not timed_out: + try: + proc.wait(timeout=max(0.0, deadline - time.monotonic())) + except subprocess.TimeoutExpired: + timed_out = True + + if timed_out: + _terminate_command_stt_process_tree(proc) + for reader in readers: + reader.join(timeout=0.5) + while True: + try: + name, chunk = output_queue.get_nowait() + except queue.Empty: + break + if chunk: + chunks[name].append(chunk) + stdout = "".join(chunks["stdout"]) + stderr = "".join(chunks["stderr"]) + try: + raise subprocess.TimeoutExpired(command, timeout) + except subprocess.TimeoutExpired as exc: + raise subprocess.TimeoutExpired( + command, + timeout, + output=stdout, + stderr=stderr, + ) from exc + + stdout = "".join(chunks["stdout"]) + stderr = "".join(chunks["stderr"]) if proc.returncode: raise subprocess.CalledProcessError(