0.5.0 (2/5): login for agents, and listen, speak and debug streaming

conftest.py

@@ -55,6 +55,19 @@ def _clean_deepgram_env(monkeypatch):
             monkeypatch.delenv(key, raising=False)
 
 
+@pytest.fixture(autouse=True)
+def _speak_tests_run_as_a_person(request, monkeypatch):
+    """Speak tests run as a person by default: the agent rules key off hard
+    agent markers, which a developer's own shell may set (CLAUDECODE=1).
+    Tests that need agent mode patch ``is_ai_agent`` themselves."""
+    if "deepctl-cmd-speak" not in request.path.parts:
+        return
+    from deepctl_core.output import AI_AGENT_ENV_MARKERS
+
+    for name in (*AI_AGENT_ENV_MARKERS, "OR_APP_NAME", "OR_SITE_URL"):
+        monkeypatch.delenv(name, raising=False)
+
+
 @pytest.fixture(autouse=True)
 def _disable_telemetry(monkeypatch):
     """Keep tests that call `main()` from initializing real telemetry.
@@ -64,3 +77,14 @@ def _disable_telemetry(monkeypatch):
     """
     monkeypatch.setenv("DEEPCTL_TELEMETRY_DISABLED", "1")
     monkeypatch.setenv("DEEPCTL_TELEMETRY_DSN", "http://k@127.0.0.1:9/1")
+
+
+@pytest.fixture(autouse=True)
+def _restore_sigint():
+    """`main()` installs its own SIGINT handler; put pytest's back."""
+    import signal
+
+    before = signal.getsignal(signal.SIGINT)
+    yield
+    if before is not None:
+        signal.signal(signal.SIGINT, before)

packages/deepctl-cmd-debug-audio/README.md

@@ -38,7 +38,6 @@ pipx run deepctl --help
 - `click>=8.2.0,<9`
 - `rich>=13.0.0,<16`
 - `pydantic>=2.0.0,<3`
-- `ffmpeg-python>=0.2.0`
 - `httpx>=0.24.0,<1`
 
 ## License

packages/deepctl-cmd-debug-audio/pyproject.toml

@@ -26,7 +26,6 @@ dependencies = [
     "click>=8.2.0,<9",
     "rich>=13.0.0,<16",
     "pydantic>=2.0.0,<3",
-    "ffmpeg-python>=0.2.0",
     "httpx>=0.24.0,<1",
 ]
 

packages/deepctl-cmd-debug-audio/src/deepctl_cmd_debug_audio/command.py

@@ -8,7 +8,6 @@
 from typing import Any
 from urllib.parse import urlparse
 
-import ffmpeg  # type: ignore[import-untyped]
 from deepctl_core import (
     AuthManager,
     BaseCommand,
@@ -45,6 +44,32 @@ def _ffprobe_input(file_path: str) -> str:
 console = get_console()
 status_console = get_status_console()
 
+# Seconds a single ffprobe run may take before it is abandoned.
+FFPROBE_TIMEOUT_S = 30
+
+
+def _probe_input(path: str) -> list[str]:
+    """ffprobe's input arguments, always last on the command line.
+
+    The input is always a local file (URLs are downloaded first), so ffprobe
+    may open it only through the file protocol: tcp:, concat:, pipe: and the
+    rest are refused. It is named as ``file:<absolute path>``
+    (``_ffprobe_input``), so ffprobe reads it as a file and never as an
+    option, whatever it is called.
+    """
+    return ["-protocol_whitelist", "file", _ffprobe_input(path)]
+
+
+def _run_ffprobe(cmd: list[str]) -> "subprocess.CompletedProcess[bytes]":
+    """Run ffprobe from an argv list: no shell, no stdin, bounded in time."""
+    return subprocess.run(
+        cmd,
+        stdin=subprocess.DEVNULL,
+        capture_output=True,
+        timeout=FFPROBE_TIMEOUT_S,
+        check=False,
+    )
+
 
 def _human_output() -> bool:
     """True when stdout is for a person, not a parser."""
@@ -239,17 +264,37 @@ def run_ffprobe(
                 # value stays one argument, and run ffprobe directly
                 cmd = [ffprobe_bin, "-v", "quiet", "-print_format", "json"]
                 cmd.extend(shlex.split(custom_args))
-                cmd.append(_ffprobe_input(file_path))
+                cmd.extend(_probe_input(file_path))
 
-                result = subprocess.run(cmd, capture_output=True, text=True)
+                result = _run_ffprobe(cmd)
                 if result.returncode != 0:
-                    raise Exception(f"ffprobe failed: {result.stderr}")
-
-                return dict(json.loads(result.stdout))
-            else:
-                # Use python-ffmpeg for standard probe
-                probe = ffmpeg.probe(_ffprobe_input(file_path), cmd=ffprobe_bin)
-                return dict(probe)
+                    stderr = result.stderr.decode("utf-8", errors="replace")
+                    raise Exception(f"ffprobe failed: {stderr}")
+
+                return dict(json.loads(result.stdout.decode("utf-8")))
+            # The standard probe: the argv ffmpeg-python's probe() built,
+            # plus the file-only protocol whitelist.
+            cmd = [
+                ffprobe_bin,
+                "-show_format",
+                "-show_streams",
+                "-of",
+                "json",
+                *_probe_input(file_path),
+            ]
+            probe = _run_ffprobe(cmd)
+            if probe.returncode != 0:
+                # stderr is captured, so say what ffprobe said: its last line.
+                said = [
+                    line.strip()
+                    for line in (probe.stderr or b"")
+                    .decode("utf-8", "replace")
+                    .splitlines()
+                    if line.strip()
+                ]
+                why = said[-1][:300] if said else f"exit {probe.returncode}"
+                raise Exception(f"ffprobe couldn't read the file: {why}")
+            return dict(json.loads(probe.stdout.decode("utf-8")))
         except Exception as e:
             raise Exception(f"Failed to probe file: {e!s}")