security(transcribe): cap chunk count to bound Whisper cost/output

agent_reach/transcribe.py

@@ -26,6 +26,9 @@
 # Whisper API limit is 25MB; leave headroom for multipart overhead.
 SIZE_LIMIT_BYTES = 24 * 1024 * 1024
 CHUNK_SECONDS = 600  # 10 min — small enough that boundary cuts rarely lose meaning
+# Bound the number of chunks transcribed per call so a multi-hour or hostile
+# source cannot run up unbounded Whisper API cost / output. ~4 hours of audio.
+MAX_CHUNKS = 24
 
 PROVIDERS = {
     "groq": {
@@ -239,11 +242,22 @@ def transcribe(
     else:
         chunks = chunk_audio(compressed, work_dir)
 
+    truncated = len(chunks) > MAX_CHUNKS
+    if truncated:
+        chunks = chunks[:MAX_CHUNKS]
+
     pieces: List[str] = []
     for chunk in chunks:
         text = _transcribe_with_fallback(chunk, order, cfg)
         pieces.append(text.strip())
-    return "\n".join(p for p in pieces if p)
+    result = "\n".join(p for p in pieces if p)
+    if truncated:
+        minutes = MAX_CHUNKS * CHUNK_SECONDS // 60
+        result += (
+            f"\n\n[transcript truncated: source exceeded the {MAX_CHUNKS}-chunk "
+            f"(~{minutes} min) limit]"
+        )
+    return result
 
 
 def _transcribe_with_fallback(chunk: Path, order: List[str], config: Config) -> str:

tests/test_transcribe.py

@@ -216,6 +216,33 @@ def test_chunks_concatenated_with_newlines(
         )
         assert text == "part one\npart two"
 
+    def test_chunk_count_is_capped(self, monkeypatch, fake_config, tmp_path, chunk_file):
+        fake_config.set("groq_api_key", "gsk_test")
+        big = tmp_path / "compressed.m4a"
+        big.write_bytes(b"x" * (tr.SIZE_LIMIT_BYTES + 1))
+        monkeypatch.setattr(tr, "compress_audio", lambda src, out_dir: big)
+
+        # Produce more chunks than the cap allows.
+        many = []
+        for i in range(tr.MAX_CHUNKS + 5):
+            c = tmp_path / f"chunk_{i:03d}.m4a"
+            c.write_bytes(b"a")
+            many.append(c)
+        monkeypatch.setattr(tr, "chunk_audio", lambda src, out_dir: many)
+
+        calls = {"n": 0}
+
+        def fake_post(*a, **k):
+            calls["n"] += 1
+            return FakeResponse(200, "seg")
+
+        monkeypatch.setattr(tr.requests, "post", fake_post)
+
+        text = tr.transcribe(str(chunk_file), out_dir=tmp_path / "work", config=fake_config)
+        # Only MAX_CHUNKS Whisper calls were made — cost is bounded.
+        assert calls["n"] == tr.MAX_CHUNKS
+        assert "truncated" in text
+
     def test_no_provider_configured_fails_fast(self, fake_config, chunk_file):
         with pytest.raises(tr.NoProviderConfigured):
             tr.transcribe(str(chunk_file), config=fake_config)