feat: transcribe chunks with configurable parallel workers

Process up to eight chunks concurrently by default and add --jobs 1..64,
with --jobs 1 retaining sequential operation. Give each worker separate
encoding and request files, release payloads after completion, and merge
responses by original chunk index regardless of completion order.

Use a bounded worker scheduler that handles jobs finishing before wait is
called. Stop scheduling on an observed worker failure and terminate worker
process groups on failure or interruption before removing temporary files.
Include jobs interrupted between launch and scheduler registration in cleanup.
Preserve all-or-nothing stdout for both text and SRT output.

Document the Bash 5.1 requirement, concurrency option, resource usage, and
limits of cancelling requests already accepted by a provider.

Test Plan:
- python3 -B -m unittest -v: all 52 tests passed.
- Mocked provider tests verified eight concurrent requests by default,
  configurable limits, rolling worker replacement, isolated payloads,
  ordered text, and identical sequential/parallel SRT output.
- Verified immediate encoder failure, concurrent API failure, SIGINT and
  SIGTERM, child-process cancellation, cleanup, and no partial stdout.
- bash -n transcribe, ./transcribe --help, and
  git diff --cached --check: passed.
- No live provider request; credentials are unavailable in this environment.

test_transcribe.py

@@ -2,8 +2,10 @@ import json
 import os
 from pathlib import Path
 import shutil
+import signal
 import subprocess
 import tempfile
+import time
 import unittest
 
 from test_chunks import make_audio
@@ -29,18 +31,159 @@ class TranscribeTests(unittest.TestCase):
                         TRANSCRIBE_TEST_LOG=str(self.log), TRANSCRIBE_TEST_MODE="ok")
         self.audio = self.path / "input with spaces.wav"
 
-    def run_cli(self, *args, mode="ok", duration=70):
+    def run_cli(self, *args, mode="ok", duration=70, env_overrides=None):
         make_audio(self.audio, duration)
         return subprocess.run([str(ROOT / "transcribe"), *args, str(self.audio)],
-                              env=dict(self.env, TRANSCRIBE_TEST_MODE=mode),
+                              env=dict(self.env, TRANSCRIBE_TEST_MODE=mode, **(env_overrides or {})),
                               capture_output=True, text=True, timeout=30)
 
     def requests(self):
-        return [json.loads(line) for line in self.log.read_text().splitlines()]
+        return sorted((json.loads(line) for line in self.log.read_text().splitlines()),
+                      key=lambda request: request["index"])
+
+    def events(self):
+        path = self.log.with_suffix(".events.jsonl")
+        return [json.loads(line) for line in path.read_text().splitlines()] if path.exists() else []
+
+    def reset_logs(self):
+        self.log.unlink(missing_ok=True)
+        self.log.with_suffix(".events.jsonl").unlink(missing_ok=True)
+
+    def assert_processes_stopped(self):
+        for event in self.events():
+            for pid in (event["pid"], event.get("child_pid")):
+                if pid is None:
+                    continue
+                try:
+                    # A killed descendant can await reaping by init; it must
+                    # not still be running or holding our temporary files.
+                    state = Path(f"/proc/{pid}/stat").read_text().rsplit(")", 1)[1].split()[0]
+                    self.assertEqual(state, "Z", f"Process {pid} survived worker cleanup")
+                except FileNotFoundError:
+                    pass
 
     def assert_cleaned(self):
         self.assertTrue(all(not Path(r["staging"]).exists() for r in self.requests()))
 
+    def test_parallel_default_and_job_limit_with_ordered_output(self):
+        for options, limit, duration in (((), 8, 500), (("--jobs", "2"), 2, 180),
+                                         (("--jobs", "1"), 1, 130)):
+            with self.subTest(options=options):
+                self.reset_logs()
+                result = self.run_cli(
+                    *options, "--chunk-minutes", "1", mode="parallel", duration=duration,
+                    env_overrides={"TRANSCRIBE_TEST_BARRIER": str(limit)},
+                )
+                self.assertEqual(result.returncode, 0, result.stderr)
+                requests = self.requests()
+                self.assertEqual(result.stdout.split(), [f"Part{i}." for i in range(len(requests))])
+                active = maximum = 0
+                events = sorted(self.events(), key=lambda event: event["time"])
+                for event in events:
+                    active += 1 if event["event"] == "start" else -1
+                    maximum = max(maximum, active)
+                self.assertEqual(active, 0)
+                self.assertEqual(maximum, limit)
+                self.assertEqual(len({r["work"] for r in requests}), len(requests))
+                if limit > 1:
+                    first_finished = next(e for e in events if e["event"] == "finish")
+                    self.assertNotEqual(first_finished["index"], 0)
+                    refilled = next(e["time"] for e in events if e["event"] == "start" and e["index"] == limit)
+                    slow_finished = next(e["time"] for e in events if e["event"] == "finish" and e["index"] == 0)
+                    self.assertLess(refilled, slow_finished)
+                self.assert_processes_stopped()
+                self.assert_cleaned()
+
+    def test_parallel_srt_matches_sequential_output(self):
+        serial = self.run_cli("--jobs", "1", "--chunk-minutes", "1", "--srt", mode="parallel", duration=130)
+        self.assertEqual(serial.returncode, 0, serial.stderr)
+        self.reset_logs()
+        parallel = self.run_cli(
+            "--jobs", "3", "--chunk-minutes", "1", "--srt", mode="parallel", duration=130,
+            env_overrides={"TRANSCRIBE_TEST_BARRIER": "3"},
+        )
+        self.assertEqual(parallel.returncode, 0, parallel.stderr)
+        self.assertEqual(parallel.stdout, serial.stdout)
+        self.assertEqual([block.splitlines()[0] for block in parallel.stdout.strip().split("\n\n")],
+                         ["1", "2", "3"])
+        self.assert_cleaned()
+
+    def test_parallel_failure_cancels_workers_and_their_children(self):
+        result = self.run_cli(
+            "--jobs", "2", "--chunk-minutes", "1", mode="parallel-error", duration=180,
+            env_overrides={"TRANSCRIBE_TEST_BARRIER": "2"},
+        )
+        self.assertNotEqual(result.returncode, 0)
+        self.assertEqual(result.stdout, "")
+        self.assertIn("simulated concurrent failure", result.stderr)
+        self.assertIn("stopping remaining workers", result.stderr)
+        self.assertEqual({e["index"] for e in self.events() if e["event"] == "start"}, {0, 1})
+        self.assertTrue(any(e["event"] == "child" for e in self.events()))
+        self.assert_processes_stopped()
+        self.assert_cleaned()
+
+    def test_interruptions_stop_workers_before_removing_staging(self):
+        for signum in (signal.SIGINT, signal.SIGTERM):
+            with self.subTest(signal=signum):
+                self.reset_logs()
+                make_audio(self.audio, 180)
+                process = subprocess.Popen(
+                    [str(ROOT / "transcribe"), "--jobs", "2", "--chunk-minutes", "1", str(self.audio)],
+                    env=dict(self.env, TRANSCRIBE_TEST_MODE="blocked"),
+                    stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, start_new_session=True,
+                )
+                try:
+                    deadline = time.monotonic() + 5
+                    while time.monotonic() < deadline:
+                        events = self.events()
+                        if (sum(e["event"] == "start" for e in events) >= 2 and
+                                any(e["event"] == "child" for e in events)):
+                            break
+                        time.sleep(0.01)
+                    else:
+                        self.fail("Workers did not start")
+                    process.send_signal(signum)
+                    stdout, stderr = process.communicate(timeout=5)
+                    self.assertEqual(process.returncode, 128 + signum, stderr)
+                    self.assertEqual(stdout, "")
+                    self.assert_processes_stopped()
+                    self.assertTrue(all(not Path(e["staging"]).exists() for e in self.events()))
+                finally:
+                    if process.poll() is None:
+                        process.kill()
+                    for event in self.events():
+                        try:
+                            os.killpg(event["group"], signal.SIGKILL)
+                        except ProcessLookupError:
+                            pass
+                    process.communicate(timeout=5)
+
+    def test_immediate_encoder_failure_is_not_lost_by_worker_waiting(self):
+        encoder = self.bin / "ffmpeg"
+        encoder.write_text(
+            "#!/usr/bin/env python3\nimport os, sys\n"
+            "if '-ss' in sys.argv:\n    sys.exit(42)\n"
+            f"os.execv({shutil.which('ffmpeg')!r}, ['ffmpeg', *sys.argv[1:]])\n"
+        )
+        encoder.chmod(0o755)
+        result = self.run_cli("--jobs", "64", "--chunk-minutes", "1", duration=500)
+        self.assertEqual(result.returncode, 42, result.stderr)
+        self.assertEqual(result.stdout, "")
+        self.assertIn("stopping remaining workers", result.stderr)
+        self.assertFalse(self.log.exists())
+
+    def test_invalid_jobs_fail_before_provider_access(self):
+        for value in ("0", "65", "-1", "1.5", "abc", "9999999999999999999999"):
+            with self.subTest(value=value):
+                result = self.run_cli("--jobs", value, duration=1)
+                self.assertNotEqual(result.returncode, 0)
+                self.assertIn("integer between 1 and 64", result.stderr)
+        result = subprocess.run([str(ROOT / "transcribe"), "--jobs"], env=self.env,
+                                capture_output=True, text=True, timeout=5)
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn("integer between 1 and 64", result.stderr)
+        self.assertFalse(self.log.exists())
+
     def assert_diarization(self, enabled):
         for request in self.requests():
             self.assertEqual(request["provider"], {