optmised main and step 1 2 and 4

main.py

@@ -1,3 +1,953 @@
+# # uvicorn main:app --host 0.0.0.0 --port 8000
+# """
+# FastAPI service for the 5-step subtitle pipeline.
+
+# Project storage contract:
+
+# src/
+# ├── input/
+# │   └── <project_id>/
+# │       └── file.<original-extension>
+# └── output/
+#     └── <project_id>/
+#         ├── audios/
+#         ├── diarization/
+#         ├── segments/
+#         ├── transcribe/
+#         ├── subtitles/
+#         ├── service/
+#         │   ├── status.json
+#         │   └── pipeline.log
+#         └── ...
+
+# API:
+
+# POST /projects
+#     Start a new project from a remote HTTP/HTTPS media URL.
+
+# GET /projects/{project_id}
+#     Return current persistent project status.
+
+# GET /projects/{project_id}/logs
+#     Return recent project log lines.
+
+# POST /projects/{project_id}/resume
+#     Resume the failed/interrupted project from the first incomplete step.
+
+# GET /health
+#     Health check.
+
+# The processing job is intentionally single-worker inside this process.
+# That is important for your CPU-heavy ML pipeline because Step 2/4 should not
+# be duplicated into multiple independent OS processes.
+
+# For production across multiple machines/processes, move the job execution to
+# a durable queue such as Celery/RQ/Arq with Redis. FastAPI's BackgroundTasks
+# are suitable for smaller/simple background work but FastAPI itself documents
+# a queue/worker system as preferable for heavy background computation.
+# """
+
+# from __future__ import annotations
+
+# import asyncio
+# import json
+# import logging
+# import os
+# import re
+# import shutil
+# import tempfile
+# import threading
+# import time
+# from concurrent.futures import ThreadPoolExecutor
+# from datetime import datetime, timezone
+# from email.message import Message
+# from pathlib import Path
+# from typing import Any
+# from urllib.parse import urlparse, unquote
+
+# import httpx
+# from fastapi import FastAPI, HTTPException, status
+# from fastapi.responses import FileResponse
+# from pydantic import BaseModel, Field, HttpUrl
+
+# from src.pipeline.step_001_separate import run_audio_separation
+# from src.pipeline.step_002_diarize import run_diarization
+# from src.pipeline.step_003_segment import run_segment_preparation
+# from src.pipeline.step_004_transcribe import run_transcription
+# from src.pipeline.step_005_original_subtitle import run_step_5
+
+# # ============================================================================
+# # CONSOLE & FILE LOGGING SETUP
+# # ============================================================================
+
+# logging.basicConfig(
+#     level=logging.INFO,
+#     format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
+# )
+# console_logger = logging.getLogger("subtitle-pipeline-console")
+
+# # ============================================================================
+# # CONFIGURATION
+# # ============================================================================
+
+# BASE_DIR = Path(__file__).resolve().parent
+# SRC_DIR = BASE_DIR / "src"
+
+# INPUT_ROOT = SRC_DIR / "input"
+# OUTPUT_ROOT = SRC_DIR / "output"
+
+# MAX_DOWNLOAD_BYTES = int(
+#     os.getenv(
+#         "MAX_DOWNLOAD_BYTES",
+#         str(4 * 1024 * 1024 * 1024),
+#     )
+# )
+
+# DOWNLOAD_CHUNK_BYTES = 1024 * 1024  # 1 MiB
+# HTTP_TIMEOUT_SECONDS = float(
+#     os.getenv(
+#         "HTTP_TIMEOUT_SECONDS",
+#         "120",
+#     )
+# )
+
+# MAX_LOG_LINES = int(
+#     os.getenv(
+#         "MAX_LOG_LINES",
+#         "5000",
+#     )
+# )
+
+# MAX_STATUS_LOG_LINES = int(
+#     os.getenv(
+#         "MAX_STATUS_LOG_LINES",
+#         "100",
+#     )
+# )
+
+# MAX_CONCURRENT_PIPELINES = 1
+
+# SUPPORTED_EXTENSIONS = {
+#     ".mp3", ".wav", ".flac", ".m4a", ".aac", ".ogg", ".opus", ".wma",
+#     ".mp4", ".mkv", ".mov", ".avi", ".webm", ".m4v", ".mpeg", ".mpg",
+#     ".ts", ".mts", ".m2ts", ".3gp",
+# }
+
+# EXTENSION_BY_CONTENT_TYPE = {
+#     "audio/mpeg": ".mp3",
+#     "audio/wav": ".wav",
+#     "audio/x-wav": ".wav",
+#     "audio/flac": ".flac",
+#     "audio/x-flac": ".flac",
+#     "audio/mp4": ".m4a",
+#     "audio/aac": ".aac",
+#     "audio/ogg": ".ogg",
+#     "audio/opus": ".opus",
+#     "video/mp4": ".mp4",
+#     "video/quicktime": ".mov",
+#     "video/x-matroska": ".mkv",
+#     "video/webm": ".webm",
+#     "video/mpeg": ".mpg",
+# }
+
+# MEDIA_MAGIC_FALLBACK_EXTENSION = ".bin"
+
+# STEPS = [
+#     {"number": 1, "name": "audio_separation"},
+#     {"number": 2, "name": "speaker_diarization"},
+#     {"number": 3, "name": "speaker_segment_preparation"},
+#     {"number": 4, "name": "speaker_aware_transcription"},
+#     {"number": 5, "name": "original_subtitle_generation"},
+# ]
+
+# # ============================================================================
+# # FASTAPI APP
+# # ============================================================================
+
+# app = FastAPI(
+#     title="Subtitle Pipeline Service",
+#     version="1.0.0",
+#     description=(
+#         "Project-based FastAPI orchestration for audio separation, "
+#         "speaker diarization, segmentation, transcription, and subtitles."
+#     ),
+# )
+
+# # ============================================================================
+# # GLOBAL JOB EXECUTOR
+# # ============================================================================
+
+# PIPELINE_EXECUTOR = ThreadPoolExecutor(
+#     max_workers=MAX_CONCURRENT_PIPELINES,
+#     thread_name_prefix="subtitle-pipeline",
+# )
+
+# ACTIVE_PROJECTS: set[str] = set()
+# ACTIVE_PROJECTS_LOCK = threading.Lock()
+# PROJECT_EVENTS: dict[str, asyncio.Event] = {}
+# PROJECT_EVENTS_LOCK = threading.Lock()
+
+# # ============================================================================
+# # REQUEST MODELS
+# # ============================================================================
+
+# class CreateProjectRequest(BaseModel):
+#     project_id: str = Field(
+#         ...,
+#         min_length=1,
+#         max_length=128,
+#         description="Directory-safe project identifier.",
+#     )
+#     file_url: HttpUrl = Field(
+#         ...,
+#         description="Public/signed HTTP(S) URL of the source media file.",
+#     )
+
+# class ResumeProjectResponse(BaseModel):
+#     project_id: str
+#     status: str
+#     resumed_from_step: int | None
+#     message: str
+
+# # ============================================================================
+# # LOGGING / PERSISTENT STATE
+# # ============================================================================
+
+# class ProjectLogger:
+#     def __init__(self, project_id: str) -> None:
+#         self.project_id = project_id
+#         self.project_dir = OUTPUT_ROOT / project_id
+#         self.service_dir = self.project_dir / "service"
+#         self.log_path = self.service_dir / "pipeline.log"
+#         self.status_path = self.service_dir / "status.json"
+
+#         self.service_dir.mkdir(parents=True, exist_ok=True)
+#         self._lock = threading.Lock()
+
+#         self.logger = logging.getLogger(f"subtitle-project.{project_id}")
+#         self.logger.setLevel(logging.INFO)
+#         self.logger.propagate = False
+
+#         if not self.logger.handlers:
+#             handler = logging.FileHandler(self.log_path, encoding="utf-8")
+#             handler.setFormatter(
+#                 logging.Formatter("%(asctime)s | %(levelname)s | %(message)s")
+#             )
+#             self.logger.addHandler(handler)
+
+#     def write(self, message: str, level: int = logging.INFO) -> None:
+#         with self._lock:
+#             self.logger.log(level, message)
+#             # Detailed Console Output with Timestamp
+#             timestamp = datetime.now(timezone.utc).isoformat()
+#             console_msg = f"[{timestamp}] [Project: {self.project_id}] {message}"
+#             if level == logging.ERROR:
+#                 console_logger.error(console_msg)
+#             elif level == logging.WARNING:
+#                 console_logger.warning(console_msg)
+#             else:
+#                 console_logger.info(console_msg)
+
+#     def recent_lines(self, limit: int = MAX_STATUS_LOG_LINES) -> list[str]:
+#         if not self.log_path.exists():
+#             return []
+#         with self.log_path.open("r", encoding="utf-8", errors="replace") as handle:
+#             lines = handle.readlines()
+#         return lines[-limit:]
+
+# # ============================================================================
+# # PROJECT STATE UTILITIES
+# # ============================================================================
+
+# def utc_now() -> str:
+#     return datetime.now(timezone.utc).isoformat()
+
+# def _project_state_path(project_id: str) -> Path:
+#     return OUTPUT_ROOT / project_id / "service" / "status.json"
+
+# def _default_step_state() -> dict[str, Any]:
+#     return {
+#         "status": "pending",
+#         "started_at": None,
+#         "completed_at": None,
+#         "error": None,
+#         "result": None,
+#     }
+
+# def _initial_state(project_id: str, source_url: str) -> dict[str, Any]:
+#     return {
+#         "schema_version": "1.0",
+#         "project_id": project_id,
+#         "source_url": source_url,
+#         "source_file": None,
+#         "status": "queued",
+#         "current_step": None,
+#         "current_step_name": None,
+#         "last_completed_step": 0,
+#         "created_at": utc_now(),
+#         "updated_at": utc_now(),
+#         "completed_at": None,
+#         "error": None,
+#         "steps": {
+#             str(step["number"]): {**step, **_default_step_state()}
+#             for step in STEPS
+#         },
+#         "logs": [],
+#     }
+
+# def _read_state(project_id: str) -> dict[str, Any] | None:
+#     path = _project_state_path(project_id)
+#     if not path.exists():
+#         return None
+#     try:
+#         with path.open("r", encoding="utf-8") as handle:
+#             return json.load(handle)
+#     except json.JSONDecodeError:
+#         return None
+
+# def _atomic_write_json(path: Path, payload: dict[str, Any]) -> None:
+#     path.parent.mkdir(parents=True, exist_ok=True)
+#     temp = path.with_name(f".{path.name}.tmp-{os.getpid()}")
+#     try:
+#         with temp.open("w", encoding="utf-8") as handle:
+#             json.dump(payload, handle, indent=2, ensure_ascii=False)
+#             handle.write("\n")
+#         os.replace(temp, path)
+#     finally:
+#         temp.unlink(missing_ok=True)
+
+# def _write_state(state: dict[str, Any]) -> None:
+#     state["updated_at"] = utc_now()
+#     state["logs"] = state.get("logs", [])[-MAX_STATUS_LOG_LINES:]
+#     _atomic_write_json(_project_state_path(state["project_id"]), state)
+
+#     with PROJECT_EVENTS_LOCK:
+#         event = PROJECT_EVENTS.get(state["project_id"])
+#     if event:
+#         event.set()
+
+# def _record_event(
+#     state: dict[str, Any],
+#     logger: ProjectLogger,
+#     message: str,
+#     *,
+#     level: int = logging.INFO,
+# ) -> None:
+#     logger.write(message, level=level)
+#     logs = state.setdefault("logs", [])
+#     logs.append({"timestamp": utc_now(), "message": message})
+#     state["updated_at"] = utc_now()
+#     _write_state(state)
+
+# # ============================================================================
+# # PROJECT ID / PATH SAFETY
+# # ============================================================================
+
+# PROJECT_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
+
+# def validate_project_id(project_id: str) -> str:
+#     value = str(project_id).strip()
+#     if not PROJECT_ID_PATTERN.fullmatch(value):
+#         raise HTTPException(
+#             status_code=status.HTTP_400_BAD_REQUEST,
+#             detail=(
+#                 "project_id must contain only letters, numbers, "
+#                 "dot, underscore, and hyphen, and must not exceed 128 characters."
+#             ),
+#         )
+#     return value
+
+# def project_paths(project_id: str) -> dict[str, Path]:
+#     project_id = validate_project_id(project_id)
+#     root = OUTPUT_ROOT / project_id
+#     return {
+#         "project_dir": root,
+#         "input_dir": INPUT_ROOT / project_id,
+#         "output_dir": root,
+#         "audios": root / "audios",
+#         "diarization": root / "diarization",
+#         "segments": root / "segments",
+#         "transcribe": root / "transcribe",
+#         "subtitles": root / "subtitles",
+#         "service": root / "service",
+#     }
+
+# # ============================================================================
+# # URL DOWNLOAD
+# # ============================================================================
+
+# def _extension_from_url(url: str) -> str | None:
+#     parsed = urlparse(url)
+#     name = unquote(Path(parsed.path).name)
+#     suffix = Path(name).suffix.lower()
+#     if suffix in SUPPORTED_EXTENSIONS:
+#         return suffix
+#     return None
+
+# def _extension_from_content_type(content_type: str | None) -> str | None:
+#     if not content_type:
+#         return None
+#     value = content_type.split(";", 1)[0].strip().lower()
+#     return EXTENSION_BY_CONTENT_TYPE.get(value)
+
+# def _extension_from_content_disposition(header: str | None) -> str | None:
+#     if not header:
+#         return None
+#     message = Message()
+#     message["content-disposition"] = header
+#     filename = message.get_filename()
+#     if not filename:
+#         return None
+#     suffix = Path(filename).suffix.lower()
+#     if suffix in SUPPORTED_EXTENSIONS:
+#         return suffix
+#     return None
+
+# def _sniff_extension(first_bytes: bytes) -> str | None:
+#     if first_bytes.startswith(b"RIFF") and first_bytes[8:12] == b"WAVE":
+#         return ".wav"
+#     if first_bytes.startswith(b"ID3"):
+#         return ".mp3"
+#     if first_bytes.startswith(b"OggS"):
+#         return ".ogg"
+#     if first_bytes.startswith(b"fLaC"):
+#         return ".flac"
+#     if len(first_bytes) >= 12:
+#         if first_bytes[4:8] == b"ftyp":
+#             return ".mp4"
+#     if first_bytes.startswith(b"\x1a\x45\xdf\xa3"):
+#         return ".mkv"
+#     return None
+
+# def _safe_download_destination(project_id: str, extension: str) -> Path:
+#     paths = project_paths(project_id)
+#     input_dir = paths["input_dir"]
+#     input_dir.mkdir(parents=True, exist_ok=True)
+#     return input_dir / f"file{extension}"
+
+# def download_source_file(
+#     project_id: str,
+#     source_url: str,
+#     logger: ProjectLogger,
+# ) -> Path:
+#     logger.write(f"Downloading source media: {source_url}")
+#     with httpx.Client(
+#         follow_redirects=True,
+#         timeout=httpx.Timeout(HTTP_TIMEOUT_SECONDS),
+#         headers={"User-Agent": "subtitle-pipeline/1.0 (media downloader)"},
+#     ) as client:
+#         try:
+#             with client.stream("GET", source_url) as response:
+#                 response.raise_for_status()
+#                 content_type = response.headers.get("content-type")
+#                 content_length_header = response.headers.get("content-length")
+
+#                 if content_length_header:
+#                     try:
+#                         content_length = int(content_length_header)
+#                     except ValueError:
+#                         content_length = None
+
+#                     if content_length is not None and content_length > MAX_DOWNLOAD_BYTES:
+#                         raise ValueError("Remote media exceeds MAX_DOWNLOAD_BYTES.")
+
+#                 extension = (
+#                     _extension_from_content_disposition(response.headers.get("content-disposition"))
+#                     or _extension_from_url(source_url)
+#                     or _extension_from_content_type(content_type)
+#                 )
+
+#                 if not extension:
+#                     extension = MEDIA_MAGIC_FALLBACK_EXTENSION
+
+#                 destination = _safe_download_destination(project_id, extension)
+#                 partial = destination.with_suffix(destination.suffix + ".part")
+#                 partial.unlink(missing_ok=True)
+
+#                 total = 0
+#                 sniff = bytearray()
+
+#                 with partial.open("wb") as output:
+#                     for chunk in response.iter_bytes(chunk_size=DOWNLOAD_CHUNK_BYTES):
+#                         if not chunk:
+#                             continue
+#                         if len(sniff) < 64:
+#                             needed = 64 - len(sniff)
+#                             sniff.extend(chunk[:needed])
+#                         total += len(chunk)
+#                         if total > MAX_DOWNLOAD_BYTES:
+#                             raise ValueError("Remote media exceeded MAX_DOWNLOAD_BYTES.")
+#                         output.write(chunk)
+
+#                 if destination.suffix == MEDIA_MAGIC_FALLBACK_EXTENSION:
+#                     sniffed = _sniff_extension(bytes(sniff))
+#                     if sniffed:
+#                         sniffed_destination = destination.with_suffix(sniffed)
+#                         sniffed_destination.unlink(missing_ok=True)
+#                         os.replace(partial, sniffed_destination)
+#                         destination = sniffed_destination
+#                     else:
+#                         os.replace(partial, destination)
+#                 else:
+#                     os.replace(partial, destination)
+
+#                 if destination.stat().st_size <= 0:
+#                     raise ValueError("Downloaded media file is empty.")
+
+#                 logger.write(
+#                     f"Download complete: {destination} ({destination.stat().st_size:,} bytes)"
+#                 )
+#                 return destination
+
+#         except Exception:
+#             try:
+#                 if "partial" in locals():
+#                     partial.unlink(missing_ok=True)
+#             except Exception:
+#                 pass
+#             raise
+
+# # ============================================================================
+# # PIPELINE EXECUTION
+# # ============================================================================
+
+# def _result_to_jsonable(value: Any) -> Any:
+#     if value is None:
+#         return None
+#     if isinstance(value, (str, int, float, bool)):
+#         return value
+#     if isinstance(value, Path):
+#         return str(value)
+#     if isinstance(value, dict):
+#         return {str(key): _result_to_jsonable(item) for key, item in value.items()}
+#     if isinstance(value, (list, tuple)):
+#         return [_result_to_jsonable(item) for item in value]
+#     return str(value)
+
+# def _run_step_1(project_id: str, source_file: Path) -> dict[str, Any]:
+#     paths = project_paths(project_id)
+#     vocal_path, instrument_path = run_audio_separation(
+#         input_file_path=source_file,
+#         output_dir=paths["audios"],
+#     )
+#     return {"vocal_path": str(vocal_path), "instrument_path": str(instrument_path)}
+
+# def _run_step_2(project_id: str) -> dict[str, Any]:
+#     return run_diarization(project_id)
+
+# def _run_step_3(project_id: str) -> dict[str, Any]:
+#     return run_segment_preparation(project_id)
+
+# def _run_step_4(project_id: str) -> dict[str, Any]:
+#     return run_transcription(project_id)
+
+# def _run_step_5(project_id: str) -> dict[str, Any]:
+#     return run_step_5(project_id)
+
+# def _step_function(number: int):
+#     return {
+#         1: _run_step_1,
+#         2: _run_step_2,
+#         3: _run_step_3,
+#         4: _run_step_4,
+#         5: _run_step_5,
+#     }[number]
+
+# def _first_incomplete_step(state: dict[str, Any]) -> int | None:
+#     for step in STEPS:
+#         step_state = state["steps"][str(step["number"])]
+#         if step_state.get("status") != "success":
+#             return step["number"]
+#     return None
+
+# def _mark_step_started(state: dict[str, Any], step_number: int) -> None:
+#     step_state = state["steps"][str(step_number)]
+#     step_state["status"] = "running"
+#     step_state["started_at"] = utc_now()
+#     step_state["completed_at"] = None
+#     step_state["error"] = None
+
+#     state["status"] = "running"
+#     state["current_step"] = step_number
+#     state["current_step_name"] = STEPS[step_number - 1]["name"]
+#     _write_state(state)
+
+# def _mark_step_success(state: dict[str, Any], step_number: int, result: Any) -> None:
+#     step_state = state["steps"][str(step_number)]
+#     step_state["status"] = "success"
+#     step_state["completed_at"] = utc_now()
+#     step_state["error"] = None
+#     step_state["result"] = _result_to_jsonable(result)
+
+#     state["last_completed_step"] = max(state.get("last_completed_step", 0), step_number)
+#     state["error"] = None
+#     _write_state(state)
+
+# def _mark_step_failed(state: dict[str, Any], step_number: int, error: Exception) -> None:
+#     step_state = state["steps"][str(step_number)]
+#     step_state["status"] = "failed"
+#     step_state["completed_at"] = utc_now()
+#     step_state["error"] = str(error)
+
+#     state["status"] = "failed"
+#     state["error"] = str(error)
+#     state["current_step"] = step_number
+#     state["current_step_name"] = STEPS[step_number - 1]["name"]
+#     _write_state(state)
+
+# def _run_pipeline_job(project_id: str) -> None:
+#     logger = ProjectLogger(project_id)
+#     state = _read_state(project_id)
+
+#     if not state:
+#         logger.write("Cannot start pipeline: project state missing.", logging.ERROR)
+#         return
+
+#     source_file_value = state.get("source_file")
+#     if not source_file_value:
+#         state["status"] = "failed"
+#         state["error"] = "source_file missing from project state."
+#         _write_state(state)
+#         logger.write("Pipeline aborted: source_file missing.", logging.ERROR)
+#         return
+
+#     source_file = Path(source_file_value)
+#     if not source_file.exists():
+#         state["status"] = "failed"
+#         state["error"] = f"Source file not found: {source_file}"
+#         _write_state(state)
+#         logger.write(state["error"], logging.ERROR)
+#         return
+
+#     start_step = _first_incomplete_step(state)
+#     if start_step is None:
+#         state["status"] = "completed"
+#         state["current_step"] = None
+#         state["current_step_name"] = None
+#         state["completed_at"] = state.get("completed_at") or utc_now()
+#         _write_state(state)
+#         logger.write("Project already completed; nothing to do.")
+#         return
+
+#     _record_event(state, logger, f"Pipeline starting from Step {start_step}.")
+
+#     try:
+#         for number in range(start_step, 6):
+#             step = STEPS[number - 1]
+#             step_name = step["name"]
+
+#             _mark_step_started(state, number)
+#             _record_event(state, logger, f"STEP {number:03d} STARTED | {step_name}")
+
+#             try:
+#                 if number == 1:
+#                     result = _step_function(number)(project_id, source_file)
+#                 else:
+#                     result = _step_function(number)(project_id)
+#             except Exception as exc:
+#                 _mark_step_failed(state, number, exc)
+#                 _record_event(
+#                     state,
+#                     logger,
+#                     f"STEP {number:03d} FAILED | {step_name} | {exc}",
+#                     level=logging.ERROR,
+#                 )
+#                 return
+
+#             _mark_step_success(state, number, result)
+#             _record_event(state, logger, f"STEP {number:03d} COMPLETED | {step_name}")
+
+#         state["status"] = "completed"
+#         state["current_step"] = None
+#         state["current_step_name"] = None
+#         state["completed_at"] = utc_now()
+#         state["error"] = None
+#         _write_state(state)
+#         _record_event(state, logger, "PIPELINE COMPLETED SUCCESSFULLY.")
+
+#     finally:
+#         with ACTIVE_PROJECTS_LOCK:
+#             ACTIVE_PROJECTS.discard(project_id)
+
+# def _submit_pipeline(project_id: str) -> None:
+#     with ACTIVE_PROJECTS_LOCK:
+#         if project_id in ACTIVE_PROJECTS:
+#             raise RuntimeError("Project is already running.")
+#         ACTIVE_PROJECTS.add(project_id)
+
+#     try:
+#         PIPELINE_EXECUTOR.submit(_run_pipeline_job, project_id)
+#     except Exception:
+#         with ACTIVE_PROJECTS_LOCK:
+#             ACTIVE_PROJECTS.discard(project_id)
+#         raise
+
+# # ============================================================================
+# # API ROUTES
+# # ============================================================================
+
+# @app.get("/")
+# async def root() -> dict[str, Any]:
+#     console_logger.info(f"[{utc_now()}] Root endpoint '/' accessed.")
+#     with ACTIVE_PROJECTS_LOCK:
+#         active_count = len(ACTIVE_PROJECTS)
+#     return {
+#         "message": "Subtitle Pipeline Service is active and running.",
+#         "version": "1.0.0",
+#         "active_projects_count": active_count,
+#         "docs_url": "/docs",
+#         "health_check": "/health"
+#     }
+
+# @app.get("/health")
+# async def health() -> dict[str, Any]:
+#     with ACTIVE_PROJECTS_LOCK:
+#         active = sorted(ACTIVE_PROJECTS)
+#     console_logger.info(f"[{utc_now()}] Health check endpoint accessed.")
+#     return {
+#         "status": "ok",
+#         "service": "subtitle-pipeline",
+#         "active_projects": active,
+#         "max_concurrent_pipelines": MAX_CONCURRENT_PIPELINES,
+#         "output_root": str(OUTPUT_ROOT.resolve()),
+#     }
+
+# @app.post("/projects", status_code=status.HTTP_202_ACCEPTED)
+# async def create_project(request: CreateProjectRequest) -> dict[str, Any]:
+#     project_id = validate_project_id(request.project_id)
+#     source_url = str(request.file_url)
+#     paths = project_paths(project_id)
+
+#     paths["service"].mkdir(parents=True, exist_ok=True)
+#     existing = _read_state(project_id)
+
+#     if existing:
+#         raise HTTPException(
+#             status_code=status.HTTP_409_CONFLICT,
+#             detail=f"Project '{project_id}' already exists. Use GET for status or POST /resume.",
+#         )
+
+#     logger = ProjectLogger(project_id)
+#     state = _initial_state(project_id, source_url)
+#     _write_state(state)
+#     logger.write(f"Project created: {project_id}")
+
+#     try:
+#         source_file = await asyncio.to_thread(
+#             download_source_file,
+#             project_id,
+#             source_url,
+#             logger,
+#         )
+
+#         state = _read_state(project_id)
+#         if not state:
+#             raise RuntimeError("Project state disappeared after download.")
+
+#         state["source_file"] = str(source_file.resolve())
+#         state["status"] = "queued"
+#         state["error"] = None
+#         _write_state(state)
+
+#         _record_event(state, logger, "Source media downloaded successfully. Submitting pipeline.")
+#         _submit_pipeline(project_id)
+
+#     except Exception as exc:
+#         state = _read_state(project_id) or state
+#         state["status"] = "failed"
+#         state["error"] = str(exc)
+#         _write_state(state)
+#         logger.write(f"PROJECT CREATION FAILED | {exc}", level=logging.ERROR)
+#         raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc
+
+#     return {
+#         "project_id": project_id,
+#         "status": "queued",
+#         "source_file": state["source_file"],
+#         "status_url": f"/projects/{project_id}",
+#         "logs_url": f"/projects/{project_id}/logs",
+#         "resume_url": f"/projects/{project_id}/resume",
+#     }
+
+# @app.get("/projects/{project_id}")
+# async def get_project(project_id: str) -> dict[str, Any]:
+#     project_id = validate_project_id(project_id)
+#     state = _read_state(project_id)
+
+#     if not state:
+#         raise HTTPException(
+#             status_code=status.HTTP_404_NOT_FOUND,
+#             detail=f"Project '{project_id}' does not exist.",
+#         )
+
+#     with ACTIVE_PROJECTS_LOCK:
+#         running = project_id in ACTIVE_PROJECTS
+
+#     state["runtime"] = {"running_in_this_process": running}
+#     return state
+
+# @app.get("/projects/{project_id}/logs")
+# async def get_project_logs(project_id: str, limit: int = 100) -> dict[str, Any]:
+#     project_id = validate_project_id(project_id)
+#     state = _read_state(project_id)
+
+#     if not state:
+#         raise HTTPException(
+#             status_code=status.HTTP_404_NOT_FOUND,
+#             detail=f"Project '{project_id}' does not exist.",
+#         )
+
+#     limit = max(1, min(limit, MAX_LOG_LINES))
+#     logger = ProjectLogger(project_id)
+
+#     return {
+#         "project_id": project_id,
+#         "log_file": str(logger.log_path.resolve()),
+#         "lines": logger.recent_lines(limit),
+#     }
+
+# @app.post("/projects/{project_id}/resume", response_model=ResumeProjectResponse)
+# async def resume_project(project_id: str) -> ResumeProjectResponse:
+#     project_id = validate_project_id(project_id)
+#     state = _read_state(project_id)
+
+#     if not state:
+#         raise HTTPException(
+#             status_code=status.HTTP_404_NOT_FOUND,
+#             detail=f"Project '{project_id}' does not exist.",
+#         )
+
+#     if not state.get("source_file"):
+#         raise HTTPException(
+#             status_code=status.HTTP_409_CONFLICT,
+#             detail="Cannot resume because the source media file is missing from project state.",
+#         )
+
+#     if not Path(state["source_file"]).exists():
+#         raise HTTPException(
+#             status_code=status.HTTP_409_CONFLICT,
+#             detail="Cannot resume because the downloaded source file no longer exists.",
+#         )
+
+#     with ACTIVE_PROJECTS_LOCK:
+#         if project_id in ACTIVE_PROJECTS:
+#             raise HTTPException(
+#                 status_code=status.HTTP_409_CONFLICT,
+#                 detail="Project is already running.",
+#             )
+
+#     start_step = _first_incomplete_step(state)
+#     if start_step is None:
+#         return ResumeProjectResponse(
+#             project_id=project_id,
+#             status="completed",
+#             resumed_from_step=None,
+#             message="Project is already complete.",
+#         )
+
+#     step_state = state["steps"][str(start_step)]
+#     step_state["status"] = "pending"
+#     step_state["error"] = None
+#     step_state["started_at"] = None
+#     step_state["completed_at"] = None
+
+#     state["status"] = "queued"
+#     state["error"] = None
+#     state["current_step"] = start_step
+#     state["current_step_name"] = STEPS[start_step - 1]["name"]
+#     _write_state(state)
+
+#     logger = ProjectLogger(project_id)
+#     _record_event(
+#         state,
+#         logger,
+#         f"RESUME REQUESTED | restarting from Step {start_step:03d} | {STEPS[start_step - 1]['name']}",
+#     )
+
+#     try:
+#         _submit_pipeline(project_id)
+#     except RuntimeError as exc:
+#         raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) from exc
+
+#     return ResumeProjectResponse(
+#         project_id=project_id,
+#         status="queued",
+#         resumed_from_step=start_step,
+#         message=f"Project queued for resume from Step {start_step:03d}.",
+#     )
+
+# @app.get("/projects/{project_id}/subtitle")
+# async def get_project_subtitle(project_id: str):
+#     project_id = validate_project_id(project_id)
+#     state = _read_state(project_id)
+
+#     if not state:
+#         raise HTTPException(
+#             status_code=status.HTTP_404_NOT_FOUND,
+#             detail={
+#                 "status": "not_found",
+#                 "message": f"Project '{project_id}' does not exist.",
+#             },
+#         )
+
+#     step_5 = state.get("steps", {}).get("5", {})
+#     step_5_result = step_5.get("result") or {}
+#     srt_path_value = step_5_result.get("srt_path")
+
+#     if srt_path_value:
+#         srt_path = Path(srt_path_value)
+#         if srt_path.exists() and srt_path.is_file():
+#             return FileResponse(
+#                 path=str(srt_path),
+#                 media_type="application/x-subrip",
+#                 filename=srt_path.name,
+#                 headers={"Content-Disposition": f'attachment; filename="{srt_path.name}"'},
+#             )
+
+#     subtitles_dir = OUTPUT_ROOT / project_id / "subtitles"
+#     if subtitles_dir.exists():
+#         srt_files = sorted(
+#             subtitles_dir.glob("*.srt"),
+#             key=lambda path: path.stat().st_mtime,
+#             reverse=True,
+#         )
+#         if srt_files:
+#             srt_path = srt_files[0]
+#             return FileResponse(
+#                 path=str(srt_path),
+#                 media_type="application/x-subrip",
+#                 filename=srt_path.name,
+#                 headers={"Content-Disposition": f'attachment; filename="{srt_path.name}"'},
+#             )
+
+#     logger = ProjectLogger(project_id)
+#     return {
+#         "status": "not_ready",
+#         "message": "Subtitle file has not been created yet.",
+#         "project_id": project_id,
+#         "project_status": state.get("status"),
+#         "current_step": state.get("current_step"),
+#         "current_step_name": state.get("current_step_name"),
+#         "last_completed_step": state.get("last_completed_step", 0),
+#         "subtitle_exists": False,
+#         "error": state.get("error"),
+#         "logs": logger.recent_lines(20),
+#     }
+
+#     '''
+
+
+# curl -X POST \
+#   "https://friendly-space-train-xrvvggqpg95jc57j-8000.app.github.dev/projects" \
+#   -H "Content-Type: application/json" \
+#   -d '{
+#     "project_id": "test-conversation-002",
+#     "file_url": "https://huggingface.co/datasets/zhangjinyang/TalkingHead-1KH-audio/resolve/main/val_cropped_clips/1lSejjfNHpw_0075_S1_E728_L671_T47_R1471_B847.mp4"
+#   }'
+
+#   '''
+
+
 # uvicorn main:app --host 0.0.0.0 --port 8000
 """
 FastAPI service for the 5-step subtitle pipeline.
@@ -50,16 +1000,18 @@ a queue/worker system as preferable for heavy background computation.
 from __future__ import annotations
 
 import asyncio
+import copy
 import json
 import logging
 import os
 import re
 import shutil
+import sys
 import tempfile
 import threading
 import time
 from concurrent.futures import ThreadPoolExecutor
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 from email.message import Message
 from pathlib import Path
 from typing import Any
@@ -68,6 +1020,7 @@ from urllib.parse import urlparse, unquote
 import httpx
 from fastapi import FastAPI, HTTPException, status
 from fastapi.responses import FileResponse
+from fastapi.middleware.cors import CORSMiddleware
 from pydantic import BaseModel, Field, HttpUrl
 
 from src.pipeline.step_001_separate import run_audio_separation
@@ -116,6 +1069,171 @@ MAX_STATUS_LOG_LINES = int(
     )
 )
 
+# Webhook (optional, sent per-project) delivery timeout.
+WEBHOOK_TIMEOUT_SECONDS = float(
+    os.getenv(
+        "WEBHOOK_TIMEOUT_SECONDS",
+        "15",
+    )
+)
+
+# Whether logs are also mirrored to stdout (colorized) in addition to the
+# per-project log file on disk.
+LOG_TO_CONSOLE = os.getenv(
+    "LOG_TO_CONSOLE",
+    "true",
+).strip().lower() in {
+    "1",
+    "true",
+    "yes",
+}
+
+# ============================================================================
+# TIMEZONE (INDIAN STANDARD TIME) — all human-facing log timestamps use IST.
+# ============================================================================
+
+IST_TZ = timezone(
+    timedelta(
+        hours=5,
+        minutes=30,
+    ),
+    name="IST",
+)
+
+
+def ist_now() -> datetime:
+    return datetime.now(
+        IST_TZ
+    )
+
+
+def ist_now_iso() -> str:
+    return ist_now().isoformat()
+
+
+def ist_now_display() -> str:
+    return ist_now().strftime(
+        "%Y-%m-%d %H:%M:%S.%f"
+    )[:-3] + " IST"
+
+
+# ============================================================================
+# ANSI COLORS — used to make per-step / per-event log lines colorful.
+# ============================================================================
+
+class Ansi:
+    RESET = "\033[0m"
+    BOLD = "\033[1m"
+    DIM = "\033[2m"
+
+    BLACK = "\033[30m"
+    RED = "\033[31m"
+    GREEN = "\033[32m"
+    YELLOW = "\033[33m"
+    BLUE = "\033[34m"
+    MAGENTA = "\033[35m"
+    CYAN = "\033[36m"
+    WHITE = "\033[37m"
+
+    BRIGHT_RED = "\033[91m"
+    BRIGHT_GREEN = "\033[92m"
+    BRIGHT_YELLOW = "\033[93m"
+    BRIGHT_BLUE = "\033[94m"
+    BRIGHT_MAGENTA = "\033[95m"
+    BRIGHT_CYAN = "\033[96m"
+    BRIGHT_WHITE = "\033[97m"
+
+
+# Each pipeline step gets its own distinct color so its log lines are
+# instantly recognizable when scanning pipeline.log or the console.
+STEP_COLORS: dict[int, str] = {
+    1: Ansi.BRIGHT_CYAN,
+    2: Ansi.BRIGHT_MAGENTA,
+    3: Ansi.BRIGHT_YELLOW,
+    4: Ansi.BRIGHT_BLUE,
+    5: Ansi.BRIGHT_GREEN,
+}
+
+# Event-type colors (STARTED / COMPLETED / FAILED / etc.) layered on top of
+# the step color.
+EVENT_COLORS: dict[str, str] = {
+    "STARTED": Ansi.BLUE,
+    "RUNNING": Ansi.CYAN,
+    "COMPLETED": Ansi.GREEN,
+    "SUCCESS": Ansi.GREEN,
+    "FAILED": Ansi.RED,
+    "ERROR": Ansi.RED,
+    "INFO": Ansi.WHITE,
+    "WEBHOOK": Ansi.MAGENTA,
+    "RESUME": Ansi.YELLOW,
+    "PROJECT": Ansi.BRIGHT_WHITE,
+}
+
+LEVEL_COLORS: dict[int, str] = {
+    logging.DEBUG: Ansi.WHITE,
+    logging.INFO: Ansi.CYAN,
+    logging.WARNING: Ansi.YELLOW,
+    logging.ERROR: Ansi.RED,
+    logging.CRITICAL: Ansi.BOLD + Ansi.BRIGHT_RED,
+}
+
+
+def _colorize(
+    text: str,
+    color: str,
+) -> str:
+    return f"{color}{text}{Ansi.RESET}"
+
+
+def step_log_message(
+    step_number: int | None,
+    event_type: str,
+    message: str,
+) -> str:
+    """
+    Build a colorful, per-step log message such as:
+
+        [STEP 002] STARTED | speaker_diarization
+
+    Colors are ANSI escape codes. They render nicely in any terminal that
+    tails pipeline.log (e.g. `tail -f` / `cat`) and in the console handler.
+    They are harmless (ignored/visible as-is) in non-ANSI viewers.
+    """
+    step_color = (
+        STEP_COLORS.get(
+            step_number,
+            Ansi.WHITE,
+        )
+        if step_number is not None
+        else Ansi.BRIGHT_WHITE
+    )
+
+    event_color = EVENT_COLORS.get(
+        event_type,
+        Ansi.WHITE,
+    )
+
+    step_tag = (
+        _colorize(
+            f"[STEP {step_number:03d}]",
+            Ansi.BOLD + step_color,
+        )
+        if step_number is not None
+        else _colorize(
+            "[PROJECT]",
+            Ansi.BOLD + Ansi.BRIGHT_WHITE,
+        )
+    )
+
+    event_tag = _colorize(
+        event_type,
+        Ansi.BOLD + event_color,
+    )
+
+    return (
+        f"{step_tag} {event_tag} | {message}"
+    )
+
 # Only one whole pipeline job executes at a time in this process.
 #
 # This is deliberate. Step 2/4 are heavy CPU model workloads, and concurrent
@@ -201,6 +1319,14 @@ app = FastAPI(
     ),
 )
 
+# Configure CORS to allow requests from anywhere
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],       # Allows requests from any origin
+    allow_credentials=True,    # Allows cookies/auth headers
+    allow_methods=["*"],       # Allows all HTTP methods (GET, POST, etc.)
+    allow_headers=["*"],       # Allows all headers
+)
 
 # ============================================================================
 # GLOBAL JOB EXECUTOR
@@ -239,6 +1365,18 @@ class CreateProjectRequest(BaseModel):
         description="Public/signed HTTP(S) URL of the source media file.",
     )
 
+    webhook_url: HttpUrl | None = Field(
+        default=None,
+        description=(
+            "Optional HTTP(S) webhook URL. If present, the service will "
+            "POST a JSON payload to it whenever a pipeline step completes "
+            "(with that step's detailed logs plus all previous logs for "
+            "this project), and once more when the whole pipeline "
+            "completes (with all logs and the subtitle file path/"
+            "download link)."
+        ),
+    )
+
 
 class ResumeProjectResponse(BaseModel):
     project_id: str
@@ -251,6 +1389,79 @@ class ResumeProjectResponse(BaseModel):
 # LOGGING / PERSISTENT STATE
 # ============================================================================
 
+class ISTColorFormatter(logging.Formatter):
+    """
+    Logging formatter that:
+      1. Renders timestamps in Indian Standard Time (IST, UTC+05:30).
+      2. Colorizes the level name using ANSI escape codes so pipeline.log
+         (and the console mirror) is colorful and easy to scan.
+
+    The per-step / per-event coloring (STARTED / COMPLETED / FAILED, etc.)
+    is applied separately, in the message text itself, via
+    `step_log_message()`, so a single log line ends up carrying both a
+    level color and a step/event color.
+    """
+
+    def __init__(
+        self,
+        fmt: str,
+        colorize: bool = True,
+    ) -> None:
+        super().__init__(fmt)
+        self.colorize = colorize
+
+    def formatTime(
+        self,
+        record: logging.LogRecord,
+        datefmt: str | None = None,
+    ) -> str:
+        dt = datetime.fromtimestamp(
+            record.created,
+            tz=IST_TZ,
+        )
+
+        if datefmt:
+            return dt.strftime(
+                datefmt
+            )
+
+        return (
+            dt.strftime(
+                "%Y-%m-%d %H:%M:%S.%f"
+            )[:-3]
+            + " IST"
+        )
+
+    def format(
+        self,
+        record: logging.LogRecord,
+    ) -> str:
+        if not self.colorize:
+            return super().format(
+                record
+            )
+
+        # Work on a shallow copy so we never mutate the shared LogRecord
+        # (multiple handlers format the same record).
+        colored_record = copy.copy(
+            record
+        )
+
+        level_color = LEVEL_COLORS.get(
+            colored_record.levelno,
+            Ansi.WHITE,
+        )
+
+        colored_record.levelname = _colorize(
+            f"{colored_record.levelname:<8}",
+            level_color,
+        )
+
+        return super().format(
+            colored_record
+        )
+
+
 class ProjectLogger:
     """
     File-backed project logger.
@@ -300,16 +1511,31 @@ class ProjectLogger:
         self.logger.propagate = False
 
         if not self.logger.handlers:
-            handler = logging.FileHandler(
+            file_handler = logging.FileHandler(
                 self.log_path,
                 encoding="utf-8",
             )
-            handler.setFormatter(
-                logging.Formatter(
+            file_handler.setFormatter(
+                ISTColorFormatter(
                     "%(asctime)s | %(levelname)s | %(message)s",
+                    colorize=True,
                 )
             )
-            self.logger.addHandler(handler)
+            self.logger.addHandler(file_handler)
+
+            if LOG_TO_CONSOLE:
+                console_handler = logging.StreamHandler(
+                    sys.stdout
+                )
+                console_handler.setFormatter(
+                    ISTColorFormatter(
+                        "%(asctime)s | %(levelname)s | "
+                        f"[{project_id}] "
+                        "%(message)s",
+                        colorize=True,
+                    )
+                )
+                self.logger.addHandler(console_handler)
 
     def write(
         self,
@@ -322,6 +1548,31 @@ class ProjectLogger:
                 message,
             )
 
+    def write_step(
+        self,
+        step_number: int | None,
+        event_type: str,
+        message: str,
+        level: int = logging.INFO,
+    ) -> str:
+        """
+        Write a colorful, per-step log line (both to file and, if enabled,
+        console) and return the fully-composed message text so callers can
+        reuse the exact same text (e.g. for state["logs"] / webhooks).
+        """
+        composed = step_log_message(
+            step_number,
+            event_type,
+            message,
+        )
+
+        self.write(
+            composed,
+            level=level,
+        )
+
+        return composed
+
     def recent_lines(
         self,
         limit: int = MAX_STATUS_LOG_LINES,
@@ -340,6 +1591,41 @@ class ProjectLogger:
 
         return lines[-limit:]
 
+    def all_lines(
+        self,
+    ) -> list[str]:
+        """
+        Full project log, unbounded. Used for the "all previous logs
+        related to this project" payload sent to the webhook, and for the
+        final completed-pipeline webhook event.
+        """
+        if not self.log_path.exists():
+            return []
+
+        with self.log_path.open(
+            "r",
+            encoding="utf-8",
+            errors="replace",
+        ) as handle:
+            return handle.readlines()
+
+    def lines_for_step(
+        self,
+        step_number: int,
+    ) -> list[str]:
+        """
+        Just the lines related to a single step (STARTED/COMPLETED/FAILED
+        and anything else logged with that step's [STEP NNN] tag).
+        Used for the "this step's detailed logs" webhook payload.
+        """
+        marker = f"STEP {step_number:03d}]"
+
+        return [
+            line
+            for line in self.all_lines()
+            if marker in line
+        ]
+
 
 # ============================================================================
 # PROJECT STATE
@@ -375,12 +1661,14 @@ def _default_step_state() -> dict[str, Any]:
 def _initial_state(
     project_id: str,
     source_url: str,
+    webhook_url: str | None = None,
 ) -> dict[str, Any]:
     return {
         "schema_version": "1.0",
         "project_id": project_id,
         "source_url": source_url,
         "source_file": None,
+        "webhook_url": webhook_url,
         "status": "queued",
         "current_step": None,
         "current_step_name": None,
@@ -502,7 +1790,7 @@ def _record_event(
 
     logs.append(
         {
-            "timestamp": utc_now(),
+            "timestamp_ist": ist_now_iso(),
             "message": message,
         }
     )
@@ -715,8 +2003,11 @@ def download_source_file(
     A .part file is used until the download completes. This makes resume/retry
     safe: an interrupted download is never mistaken for a usable source.
     """
-    logger.write(
-        f"Downloading source media: {source_url}"
+    logger.write_step(
+        None,
+        "PROJECT",
+        f"Downloading source media: {source_url} "
+        f"| started_at={ist_now_display()}",
     )
 
     with httpx.Client(
@@ -869,10 +2160,13 @@ def download_source_file(
                         "Downloaded media file is empty."
                     )
 
-                logger.write(
+                logger.write_step(
+                    None,
+                    "PROJECT",
                     "Download complete: "
                     f"{destination} "
-                    f"({destination.stat().st_size:,} bytes)"
+                    f"({destination.stat().st_size:,} bytes) "
+                    f"| completed_at={ist_now_display()}",
                 )
 
                 return destination
@@ -947,6 +2241,181 @@ def _result_to_jsonable(
     )
 
 
+# ============================================================================
+# OPTIONAL PROJECT WEBHOOK
+# ============================================================================
+
+def _subtitle_download_info(
+    project_id: str,
+    state: dict[str, Any],
+) -> dict[str, Any]:
+    """
+    Resolve the generated subtitle file path (if any) plus the existing
+    download route for it, for inclusion in webhook payloads.
+    """
+    step_5 = state.get(
+        "steps",
+        {},
+    ).get(
+        "5",
+        {},
+    )
+
+    step_5_result = (
+        step_5.get("result")
+        or {}
+    )
+
+    srt_path_value = step_5_result.get(
+        "srt_path"
+    )
+
+    subtitle_path: str | None = None
+
+    if srt_path_value and Path(
+        srt_path_value
+    ).exists():
+        subtitle_path = str(
+            Path(
+                srt_path_value
+            ).resolve()
+        )
+    else:
+        subtitles_dir = (
+            OUTPUT_ROOT
+            / project_id
+            / "subtitles"
+        )
+
+        if subtitles_dir.exists():
+            srt_files = sorted(
+                subtitles_dir.glob(
+                    "*.srt"
+                ),
+                key=lambda path: path.stat().st_mtime,
+                reverse=True,
+            )
+
+            if srt_files:
+                subtitle_path = str(
+                    srt_files[0].resolve()
+                )
+
+    return {
+        "subtitle_file_path": subtitle_path,
+        "subtitle_ready": subtitle_path is not None,
+        # This route already exists in this API and streams the .srt file.
+        "subtitle_download_url": (
+            f"/projects/{project_id}/subtitle"
+        ),
+    }
+
+
+def _send_webhook(
+    state: dict[str, Any],
+    logger: ProjectLogger,
+    event: str,
+    *,
+    step_number: int | None = None,
+    step_state: dict[str, Any] | None = None,
+) -> None:
+    """
+    Best-effort webhook delivery.
+
+    Fired:
+      - after every step completes (success or failure): includes that
+        step's own detailed logs, plus every log line for the project so
+        far ("previous logs too, related to that project").
+      - once more when the whole pipeline finishes: includes ALL logs for
+        the project plus the subtitle file path / download link.
+
+    Failures here are logged but NEVER raised — a webhook outage must not
+    break the pipeline.
+    """
+    webhook_url = state.get(
+        "webhook_url"
+    )
+
+    if not webhook_url:
+        return
+
+    project_id = state[
+        "project_id"
+    ]
+
+    payload: dict[str, Any] = {
+        "event": event,
+        "project_id": project_id,
+        "sent_at_ist": ist_now_iso(),
+        "status": state.get("status"),
+        "current_step": state.get("current_step"),
+        "current_step_name": state.get("current_step_name"),
+        "last_completed_step": state.get(
+            "last_completed_step",
+            0,
+        ),
+        "error": state.get("error"),
+        "logs_so_far": logger.all_lines(),
+    }
+
+    if step_number is not None:
+        payload["step"] = {
+            "number": step_number,
+            "name": STEPS[step_number - 1]["name"],
+            "status": (step_state or {}).get("status"),
+            "started_at": (step_state or {}).get("started_at"),
+            "completed_at": (step_state or {}).get("completed_at"),
+            "error": (step_state or {}).get("error"),
+            "result": (step_state or {}).get("result"),
+        }
+        payload["step_logs"] = logger.lines_for_step(
+            step_number
+        )
+
+    if event == "pipeline_completed":
+        payload.update(
+            _subtitle_download_info(
+                project_id,
+                state,
+            )
+        )
+        payload["completed_at"] = state.get(
+            "completed_at"
+        )
+
+    try:
+        with httpx.Client(
+            timeout=httpx.Timeout(
+                WEBHOOK_TIMEOUT_SECONDS
+            ),
+        ) as client:
+            response = client.post(
+                webhook_url,
+                json=payload,
+            )
+
+        logger.write_step(
+            step_number,
+            "WEBHOOK",
+            (
+                f"Webhook delivered | event={event} | "
+                f"url={webhook_url} | "
+                f"response_status={response.status_code}"
+            ),
+        )
+
+    except Exception as exc:
+        logger.write_step(
+            step_number,
+            "WEBHOOK",
+            (
+                f"Webhook delivery FAILED | event={event} | "
+                f"url={webhook_url} | error={exc}"
+            ),
+            level=logging.WARNING,
+        )
+
+
 def _run_step_1(
     project_id: str,
     source_file: Path,
@@ -1228,7 +2697,12 @@ def _run_pipeline_job(
     _record_event(
         state,
         logger,
-        f"Pipeline starting from Step {start_step}.",
+        step_log_message(
+            None,
+            "PROJECT",
+            f"Pipeline starting from Step {start_step:03d} "
+            f"at {ist_now_display()}.",
+        ),
     )
 
     try:
@@ -1252,7 +2726,11 @@ def _run_pipeline_job(
             _record_event(
                 state,
                 logger,
-                f"STEP {number:03d} STARTED | {step_name}",
+                step_log_message(
+                    number,
+                    "STARTED",
+                    f"{step_name} | started_at={ist_now_display()}",
+                ),
             )
 
             try:
@@ -1280,13 +2758,25 @@ def _run_pipeline_job(
                 _record_event(
                     state,
                     logger,
-                    (
-                        f"STEP {number:03d} FAILED | "
-                        f"{step_name} | {exc}"
+                    step_log_message(
+                        number,
+                        "FAILED",
+                        f"{step_name} | error={exc} | "
+                        f"failed_at={ist_now_display()}",
                     ),
                     level=logging.ERROR,
                 )
 
+                # Notify the optional webhook: this step's detailed logs +
+                # every previous log line for this project.
+                _send_webhook(
+                    state,
+                    logger,
+                    event="step_failed",
+                    step_number=number,
+                    step_state=state["steps"][str(number)],
+                )
+
                 return
 
             _mark_step_success(
@@ -1298,7 +2788,21 @@ def _run_pipeline_job(
             _record_event(
                 state,
                 logger,
-                f"STEP {number:03d} COMPLETED | {step_name}",
+                step_log_message(
+                    number,
+                    "COMPLETED",
+                    f"{step_name} | completed_at={ist_now_display()}",
+                ),
+            )
+
+            # Notify the optional webhook: this step's detailed logs +
+            # every previous log line for this project so far.
+            _send_webhook(
+                state,
+                logger,
+                event="step_completed",
+                step_number=number,
+                step_state=state["steps"][str(number)],
             )
 
         state["status"] = "completed"
@@ -1314,7 +2818,20 @@ def _run_pipeline_job(
         _record_event(
             state,
             logger,
-            "PIPELINE COMPLETED SUCCESSFULLY.",
+            step_log_message(
+                None,
+                "SUCCESS",
+                "PIPELINE COMPLETED SUCCESSFULLY at "
+                f"{ist_now_display()}.",
+            ),
+        )
+
+        # Final webhook: all logs for the project + subtitle file path /
+        # download link.
+        _send_webhook(
+            state,
+            logger,
+            event="pipeline_completed",
         )
 
     finally:
@@ -1361,6 +2878,42 @@ def _submit_pipeline(
 # API ROUTES
 # ============================================================================
 
+@app.get(
+    "/",
+)
+async def root() -> dict[str, Any]:
+    """
+    Root endpoint — simple service banner + a directory of available
+    routes, so hitting the base URL is never a 404.
+    """
+    with ACTIVE_PROJECTS_LOCK:
+        active = sorted(
+            ACTIVE_PROJECTS
+        )
+
+    return {
+        "service": "Subtitle Pipeline Service",
+        "version": app.version,
+        "status": "ok",
+        "message": (
+            "Subtitle Pipeline API is running. "
+            "See /docs for interactive API documentation."
+        ),
+        "server_time_ist": ist_now_display(),
+        "active_projects": active,
+        "endpoints": {
+            "root": "GET /",
+            "health": "GET /health",
+            "docs": "GET /docs",
+            "create_project": "POST /projects",
+            "project_status": "GET /projects/{project_id}",
+            "project_logs": "GET /projects/{project_id}/logs",
+            "resume_project": "POST /projects/{project_id}/resume",
+            "download_subtitle": "GET /projects/{project_id}/subtitle",
+        },
+    }
+
+
 @app.get(
     "/health",
 )
@@ -1398,6 +2951,14 @@ async def create_project(
         request.file_url
     )
 
+    webhook_url = (
+        str(
+            request.webhook_url
+        )
+        if request.webhook_url
+        else None
+    )
+
     paths = project_paths(
         project_id
     )
@@ -1431,14 +2992,25 @@ async def create_project(
     state = _initial_state(
         project_id,
         source_url,
+        webhook_url,
     )
 
     _write_state(
         state
     )
 
-    logger.write(
-        f"Project created: {project_id}"
+    logger.write_step(
+        None,
+        "PROJECT",
+        (
+            f"Project created: {project_id} | source_url={source_url} | "
+            "webhook_url="
+            + (
+                webhook_url
+                if webhook_url
+                else "(none)"
+            )
+        ),
     )
 
     try:
@@ -1471,9 +3043,11 @@ async def create_project(
         _record_event(
             state,
             logger,
-            (
+            step_log_message(
+                None,
+                "PROJECT",
                 "Source media downloaded successfully. "
-                "Submitting pipeline."
+                f"Submitting pipeline at {ist_now_display()}.",
             ),
         )
 
@@ -1495,9 +3069,12 @@ async def create_project(
             state
         )
 
-        logger.write(
-            f"PROJECT CREATION FAILED | {exc}",
-            logging.ERROR,
+        logger.write_step(
+            None,
+            "FAILED",
+            f"PROJECT CREATION FAILED | {exc} | "
+            f"failed_at={ist_now_display()}",
+            level=logging.ERROR,
         )
 
         raise HTTPException(
@@ -1706,9 +3283,12 @@ async def resume_project(
     _record_event(
         state,
         logger,
-        (
-            f"RESUME REQUESTED | restarting from Step "
-            f"{start_step:03d} | {STEPS[start_step - 1]['name']}"
+        step_log_message(
+            start_step,
+            "RESUME",
+            f"Restarting from Step {start_step:03d} | "
+            f"{STEPS[start_step - 1]['name']} | "
+            f"requested_at={ist_now_display()}",
         ),
     )
 
@@ -1837,5 +3417,4 @@ async def get_project_subtitle(project_id: str):
 #     PIPELINE_EXECUTOR.shutdown(
 #         wait=False,
 #         cancel_futures=False,
-#     )
-
+#     )

src/pipeline/step_001_separate.py

@@ -20,44 +20,52 @@ from dataclasses import dataclass
 from pathlib import Path
 from typing import Final
 
-from audio_separator.separator import Separator
+LOGGER = logging.getLogger(__name__)
 
 
 # ---------------------------------------------------------------------------
-# Configuration
+# Dynamic Thread & Core Configuration (Executed BEFORE C++ Runtime Imports)
 # ---------------------------------------------------------------------------
 
-# Exact registered model name in audio-separator. 
-# This triggers automatic download and uses ONNX for 10x-15x faster CPU execution.
-DEFAULT_MODEL: Final[str] = "UVR-MDX-NET-Voc_FT.onnx"
-DEFAULT_SAMPLE_RATE: Final[int] = 44_100
-DEFAULT_NORMALIZATION: Final[float] = 0.9
-DEFAULT_MODEL_DIR: Final[Path] = Path("models/audio-separator")
-DEFAULT_FFMPEG_AUDIO_CHANNELS: Final[int] = 2
+def _detect_physical_cores() -> int:
+    """Dynamically detect physical CPU cores to prevent thread thrashing."""
+    try:
+        import psutil
+        physical_cores = psutil.cpu_count(logical=False)
+        if physical_cores and physical_cores > 0:
+            return physical_cores
+    except Exception:
+        pass
 
-VOCAL_OUTPUT_NAME: Final[str] = "vocal.wav"
-INSTRUMENT_OUTPUT_NAME: Final[str] = "instrument.wav"
+    # Fallback using logical count / 2 (standard hyperthreading ratio)
+    logical_cores = os.cpu_count()
+    if logical_cores and logical_cores > 1:
+        return max(2, logical_cores // 2)
 
-LOGGER = logging.getLogger(__name__)
+    return 2  # Safe default core limit
 
 
 def _configure_cpu_threads() -> int:
-    """Configure PyTorch, OpenMP, and ONNX Runtime thread pools for CPU execution."""
-    cpu_count = os.cpu_count() or 1
-    str_cpus = str(cpu_count)
-
-    # Force CPU thread pools to match actual hardware core count
-    os.environ["OMP_NUM_THREADS"] = str_cpus
-    os.environ["MKL_NUM_THREADS"] = str_cpus
-    os.environ["OPENBLAS_NUM_THREADS"] = str_cpus
-    os.environ["NUMEXPR_NUM_THREADS"] = str_cpus
-    os.environ["VECLIB_MAXIMUM_THREADS"] = str_cpus
-    os.environ["NUMPY_NUM_THREADS"] = str_cpus
+    """Configure PyTorch, OpenMP, HuggingFace, and ONNX Runtime thread pools for CPU execution."""
+    cores = _detect_physical_cores()
+    str_cores = str(cores)
+
+    # Force C/C++ underlying thread pools to match physical CPU core count
+    os.environ["OMP_NUM_THREADS"] = str_cores
+    os.environ["MKL_NUM_THREADS"] = str_cores
+    os.environ["OPENBLAS_NUM_THREADS"] = str_cores
+    os.environ["NUMEXPR_NUM_THREADS"] = str_cores
+    os.environ["VECLIB_MAXIMUM_THREADS"] = str_cores
+    os.environ["NUMPY_NUM_THREADS"] = str_cores
+
+    # ONNX & Hugging Face threading control
+    os.environ["OMP_WAIT_POLICY"] = "PASSIVE"
+    os.environ["TOKENIZERS_PARALLELISM"] = "false"
 
     try:
         import torch
 
-        torch.set_num_threads(cpu_count)
+        torch.set_num_threads(cores)
         try:
             torch.set_num_interop_threads(1)
         except RuntimeError:
@@ -65,8 +73,29 @@ def _configure_cpu_threads() -> int:
     except Exception as exc:
         LOGGER.warning("Could not configure PyTorch CPU threads: %s", exc)
 
-    LOGGER.info("CPU optimization: using %d logical CPU cores", cpu_count)
-    return cpu_count
+    LOGGER.info("CPU optimization: dynamically locked execution to %d physical CPU cores", cores)
+    return cores
+
+
+# Initialize thread locks prior to loading heavy model dependencies
+PHYSICAL_CPU_CORES = _configure_cpu_threads()
+
+# Lazy import after environment variables are set
+from audio_separator.separator import Separator  # noqa: E402
+
+
+# ---------------------------------------------------------------------------
+# Configuration
+# ---------------------------------------------------------------------------
+
+DEFAULT_MODEL: Final[str] = "UVR-MDX-NET-Voc_FT.onnx"
+DEFAULT_SAMPLE_RATE: Final[int] = 44_100
+DEFAULT_NORMALIZATION: Final[float] = 0.9
+DEFAULT_MODEL_DIR: Final[Path] = Path("models/audio-separator")
+DEFAULT_FFMPEG_AUDIO_CHANNELS: Final[int] = 2
+
+VOCAL_OUTPUT_NAME: Final[str] = "vocal.wav"
+INSTRUMENT_OUTPUT_NAME: Final[str] = "instrument.wav"
 
 
 @dataclass(frozen=True)
@@ -171,7 +200,7 @@ def _extract_to_wav(input_path: Path, temp_dir: Path, sample_rate: int) -> Path:
         "-vn",
         "-ac", str(DEFAULT_FFMPEG_AUDIO_CHANNELS),
         "-ar", str(sample_rate),
-        "-threads", "0",
+        "-threads", str(PHYSICAL_CPU_CORES),
         "-c:a", "pcm_s16le",
         str(output_path),
     ]
@@ -213,7 +242,7 @@ def _make_separator(
             mdx_params={
                 "hop_length": 1024,
                 "segment_size": 256,
-                "overlap": 0.25,  # Drastically speeds up execution
+                "overlap": 0.25,  # Drastically speeds up CPU execution
                 "batch_size": 1,
                 "enable_denoise": False,
             },
@@ -334,7 +363,6 @@ def run_audio_separation(
     input_file_path: str | os.PathLike[str],
     output_dir: str | os.PathLike[str],
 ) -> tuple[str, str]:
-    _configure_cpu_threads()
 
     input_path = Path(input_file_path).expanduser().resolve()
     output_path = Path(output_dir).expanduser().resolve()
@@ -373,7 +401,6 @@ def run_audio_separation(
         model_to_use = os.getenv("AUDIO_SEPARATOR_MODEL", DEFAULT_MODEL)
 
         try:
-            # audio-separator handles the download internally if the model is missing
             separator.load_model(model_filename=model_to_use)
         except TypeError:
             separator.load_model(model_to_use)
@@ -382,7 +409,7 @@ def run_audio_separation(
                 f"Failed to load separation model '{model_to_use}'."
             ) from exc
 
-        LOGGER.info("Separating with %s", model_to_use)
+        LOGGER.info("Separating with %s on %d physical cores", model_to_use, PHYSICAL_CPU_CORES)
 
         try:
             try:

src/pipeline/step_004_transcribe.py

@@ -1,5 +1,5 @@
 """
-Pipeline Step 004 - Speaker-Aware Transcription.
+Pipeline Step 004 - Speaker-Aware Transcription (Optimized Concurrency).
 
 Project layout used by this pipeline:
 
@@ -27,8 +27,8 @@ Step 004:
     2. Uses every prepared speaker WAV referenced by the manifest.
     3. Uses faster-whisper + CTranslate2.
     4. Uses Whisper medium.
-    5. Uses INT8 CPU inference.
-    6. Uses the available physical CPU cores by default.
+    5. Uses INT8 CPU inference with optimized multi-worker concurrency.
+    6. Uses the available physical CPU cores by default, split efficiently.
     7. Detects language automatically when WHISPER_LANGUAGE is unset.
     8. Stores language and confidence in the output.
     9. Produces word-level timestamps.
@@ -46,19 +46,6 @@ Important:
 
 from __future__ import annotations
 
-# ============================================================================
-# CPU THREAD ENVIRONMENT
-# ============================================================================
-#
-# CTranslate2 owns the model execution threads. Environment variables are set
-# before importing faster-whisper so native math libraries do not silently
-# create competing thread pools.
-#
-# Prefer physical cores. On virtualized environments os.cpu_count() often
-# reports logical CPUs, while physical-core count is usually the better default
-# for one heavy CPU inference job.
-# ============================================================================
-
 import gc
 import json
 import logging
@@ -66,8 +53,10 @@ import math
 import os
 import time
 from collections import Counter
+from concurrent.futures import ThreadPoolExecutor, as_completed
 from dataclasses import dataclass
 from pathlib import Path
+from threading import Lock
 from typing import Any, Final
 
 try:
@@ -103,26 +92,21 @@ def _detect_cpu_threads() -> int:
 
 CPU_THREADS: Final[int] = _detect_cpu_threads()
 
-os.environ.setdefault("OMP_NUM_THREADS", str(CPU_THREADS))
-os.environ.setdefault("MKL_NUM_THREADS", str(CPU_THREADS))
-os.environ.setdefault("OPENBLAS_NUM_THREADS", str(CPU_THREADS))
-os.environ.setdefault("NUMEXPR_NUM_THREADS", str(CPU_THREADS))
+# Optimize concurrent execution: split physical cores across parallel workers
+# to maximize CPU throughput without thread contention.
+NUM_WORKERS: Final[int] = max(1, min(CPU_THREADS // 4, 4))
+THREADS_PER_WORKER: Final[int] = max(1, CPU_THREADS // NUM_WORKERS)
+
+os.environ.setdefault("OMP_NUM_THREADS", str(THREADS_PER_WORKER))
+os.environ.setdefault("MKL_NUM_THREADS", str(THREADS_PER_WORKER))
+os.environ.setdefault("OPENBLAS_NUM_THREADS", str(THREADS_PER_WORKER))
+os.environ.setdefault("NUMEXPR_NUM_THREADS", str(THREADS_PER_WORKER))
 os.environ.setdefault("TOKENIZERS_PARALLELISM", "false")
 
 # ============================================================================
 # CONFIGURATION
 # ============================================================================
 
-# This file lives at:
-#   subtitle/src/pipeline/step_004_transcribe.py
-#
-# parents[1] is:
-#   subtitle/src
-#
-# Therefore every Step 4 project is rooted at:
-#   subtitle/src/output/<project_id>
-#
-# This is the path rule used by Step 3 as well.
 SRC_ROOT: Final[Path] = Path(__file__).resolve().parents[1]
 DEFAULT_OUTPUT_ROOT: Final[Path] = SRC_ROOT / "output"
 
@@ -136,9 +120,6 @@ COMPUTE_TYPE: Final[str] = os.getenv(
     "int8",
 )
 
-# One model copy per process. CTranslate2 uses CPU threads inside that model.
-NUM_WORKERS: Final[int] = 1
-
 BEAM_SIZE: Final[int] = max(
     1,
     int(os.getenv("WHISPER_BEAM_SIZE", "5")),
@@ -148,20 +129,15 @@ WORD_TIMESTAMPS: Final[bool] = True
 
 TASK: Final[str] = "transcribe"
 
-# Empty => automatic language detection.
 REQUESTED_LANGUAGE: Final[str | None] = (
     os.getenv("WHISPER_LANGUAGE") or None
 )
 
-# Set to 1 only for genuinely multilingual/code-switching content.
 DETECT_LANGUAGE_PER_CHUNK: Final[bool] = (
     os.getenv("DETECT_LANGUAGE_PER_CHUNK", "0").lower()
     in {"1", "true", "yes", "on"}
 )
 
-# VAD is optional. For these short speaker-isolated chunks it is useful for
-# long internal silence and hallucination control, but it is not needed to
-# reconstruct the original timeline because Step 003 supplies the offset.
 ENABLE_VAD: Final[bool] = (
     os.getenv("WHISPER_VAD", "1").lower()
     in {"1", "true", "yes", "on"}
@@ -174,12 +150,10 @@ VAD_MIN_SILENCE_MS: Final[int] = int(
     )
 )
 
-# Whisper/faster-whisper decoding safeguards.
 NO_SPEECH_THRESHOLD: Final[float] = 0.6
 LOG_PROB_THRESHOLD: Final[float] = -1.0
 COMPRESSION_RATIO_THRESHOLD: Final[float] = 2.4
 
-# Native faster-whisper language detection.
 LANGUAGE_DETECTION_SEGMENTS: Final[int] = max(
     1,
     int(
@@ -197,8 +171,6 @@ LANGUAGE_DETECTION_THRESHOLD: Final[float] = float(
     )
 )
 
-# If auto-detection on a chunk is below this probability, keep detecting on
-# later chunks instead of permanently locking the project to a weak guess.
 LANGUAGE_LOCK_THRESHOLD: Final[float] = float(
     os.getenv(
         "LANGUAGE_LOCK_THRESHOLD",
@@ -206,7 +178,6 @@ LANGUAGE_LOCK_THRESHOLD: Final[float] = float(
     )
 )
 
-# Manifest is checkpointed after every processed chunk.
 CHECKPOINT_EVERY_CHUNKS: Final[int] = 1
 
 FORCE_RETRANSCRIBE: Final[bool] = (
@@ -247,7 +218,6 @@ def _validate_project_id(project_id: str) -> str:
     if value in {".", ".."}:
         raise ValueError("Invalid project_id.")
 
-    # A project ID is a directory name, never a nested path.
     if Path(value).name != value:
         raise ValueError(
             "project_id must be a single directory-safe name."
@@ -494,7 +464,6 @@ def _resolve_audio_path(
     if path.is_absolute():
         return path.resolve()
 
-    # Step 3 stores relative audio paths relative to project_dir.
     project_dir = (
         manifest_path.parent.parent
     )
@@ -506,7 +475,6 @@ def _resolve_audio_path(
     if candidate.exists():
         return candidate.resolve()
 
-    # Backward-compatible fallback.
     return (
         manifest_path.parent / path
     ).resolve()
@@ -539,7 +507,7 @@ def _load_model() -> Any:
         "cpu_threads=%d num_workers=%d",
         MODEL_NAME,
         COMPUTE_TYPE,
-        CPU_THREADS,
+        THREADS_PER_WORKER,
         NUM_WORKERS,
     )
 
@@ -552,8 +520,8 @@ def _load_model() -> Any:
             MODEL_NAME,
             device="cpu",
             compute_type=COMPUTE_TYPE,
-            cpu_threads=CPU_THREADS,
-            num_workers=NUM_WORKERS,
+            cpu_threads=THREADS_PER_WORKER,
+            num_workers=1,
         )
     except Exception as exc:
         raise RuntimeError(
@@ -624,14 +592,6 @@ def _transcribe_one(
     audio_path: Path,
     language: str | None,
 ):
-    """
-    Run one speaker chunk.
-
-    When language=None, faster-whisper performs native Whisper language
-    detection and returns info.language / info.language_probability.
-
-    Word timestamps are requested because Step 005 needs them.
-    """
     kwargs: dict[str, Any] = {
         "language": language,
         "task": TASK,
@@ -1002,28 +962,12 @@ def _summarize_languages(
 def run_transcription(
     project_id: str,
 ) -> dict[str, Any]:
-    """
-    Run Step 004 for one project.
-
-    Only project_id is required.
-
-    Input:
-        src/output/<project_id>/segments/segments_manifest.json
-
-    Output:
-        src/output/<project_id>/transcribe/transcription.json
-        src/output/<project_id>/transcribe/step_004_manifest.json
-
-    Also updates:
-        src/output/<project_id>/segments/segments_manifest.json
-    """
     started = time.perf_counter()
 
     paths = _get_paths(
         project_id
     )
 
-    # Fail early and clearly before loading the model.
     _validate_file(
         paths.segments_manifest,
         "Step 003 segments_manifest.json",
@@ -1097,8 +1041,6 @@ def run_transcription(
         REQUESTED_LANGUAGE
     )
 
-    # When auto-detecting, do not immediately lock the entire project to a
-    # low-confidence first chunk.
     project_language: str | None = requested_language
     project_language_probability: float | None = None
     language_votes: list[tuple[str, float]] = []
@@ -1109,92 +1051,32 @@ def run_transcription(
     total_words = 0
     total_inference_seconds = 0.0
 
-    # ========================================================================
-    # CHUNK LOOP
-    # ========================================================================
-
-    for index, chunk in enumerate(
-        chunks,
-        start=1,
-    ):
-        chunk_id = chunk.get(
-            "chunk_id",
-            index,
-        )
+    manifest_lock = Lock()
+    completed_counter = 0
 
-        speaker = str(
-            chunk.get(
-                "speaker",
-                "UNKNOWN",
-            )
-        )
-
-        # ----------------------------------------------------
-        # Resume
-        # ----------------------------------------------------
-
-        if (
-            not FORCE_RETRANSCRIBE
-            and _chunk_successful(
-                chunk
-            )
-        ):
-            skipped += 1
-
-            previous_language = (
-                chunk.get(
-                    "transcription",
-                    {},
-                ).get(
-                    "language"
-                )
-            )
+    def process_chunk(index: int, chunk: dict[str, Any]) -> dict[str, Any]:
+        chunk_id = chunk.get("chunk_id", index)
+        speaker = str(chunk.get("speaker", "UNKNOWN"))
 
+        if not FORCE_RETRANSCRIBE and _chunk_successful(chunk):
+            previous_language = chunk.get("transcription", {}).get("language")
             previous_probability = _safe_float(
-                chunk.get(
-                    "transcription",
-                    {},
-                ).get(
-                    "language_probability",
-                    0.0,
-                )
-            )
-
-            if previous_language:
-                language_votes.append(
-                    (
-                        str(
-                            previous_language
-                        ),
-                        previous_probability,
-                    )
-                )
-
-            LOGGER.info(
-                "[%d/%d] chunk=%s speaker=%s -> SKIP",
-                index,
-                len(chunks),
-                chunk_id,
-                speaker,
+                chunk.get("transcription", {}).get("language_probability", 0.0)
             )
-            continue
-
-        # ----------------------------------------------------
-        # Resolve chunk audio
-        # ----------------------------------------------------
-
-        audio = chunk.get(
-            "audio",
-            {}
-        )
+            return {
+                "status": "skipped",
+                "index": index,
+                "chunk_id": chunk_id,
+                "speaker": speaker,
+                "language": previous_language,
+                "probability": previous_probability,
+            }
 
-        audio_path_value = audio.get(
-            "path"
-        )
+        audio = chunk.get("audio", {})
+        audio_path_value = audio.get("path")
 
         if not audio_path_value:
-            failed += 1
-
+            error = "Missing audio.path in segments_manifest.json."
             chunk["transcription"] = {
                 "status": "failed",
                 "completed": False,
@@ -1203,41 +1085,14 @@ def run_transcription(
                 "language_probability": None,
                 "words": [],
                 "word_count": 0,
-                "error": (
-                    "Missing audio.path "
-                    "in segments_manifest.json."
-                ),
+                "error": error,
             }
+            return {"status": "failed", "index": index, "chunk_id": chunk_id, "error": error}
 
-            _checkpoint_manifest(
-                paths.segments_manifest,
-                manifest,
-            )
-            continue
-
-        audio_path = _resolve_audio_path(
-            paths.segments_manifest,
-            str(
-                audio_path_value
-            ),
-        )
+        audio_path = _resolve_audio_path(paths.segments_manifest, str(audio_path_value))
 
         if not audio_path.exists():
-            failed += 1
-
-            error = (
-                "Chunk audio not found: "
-                f"{audio_path}"
-            )
-
-            LOGGER.error(
-                "[%d/%d] chunk=%s -> %s",
-                index,
-                len(chunks),
-                chunk_id,
-                error,
-            )
-
+            error = f"Chunk audio not found: {audio_path}"
             chunk["transcription"] = {
                 "status": "failed",
                 "completed": False,
@@ -1248,157 +1103,40 @@ def run_transcription(
                 "word_count": 0,
                 "error": error,
             }
+            return {"status": "failed", "index": index, "chunk_id": chunk_id, "error": error}
 
-            _checkpoint_manifest(
-                paths.segments_manifest,
-                manifest,
-            )
-            continue
-
-        # ----------------------------------------------------
-        # Source timeline from Step 003
-        # ----------------------------------------------------
-
-        timeline = chunk.get(
-            "source_timeline",
-            {}
-        )
-
-        source_start = max(
-            0.0,
-            _safe_float(
-                timeline.get(
-                    "start",
-                    0.0,
-                )
-            ),
-        )
-
-        source_end = max(
-            source_start,
-            _safe_float(
-                timeline.get(
-                    "end",
-                    source_start,
-                )
-            ),
-        )
-
-        LOGGER.info(
-            "[%d/%d] chunk=%s speaker=%s %.3f -> %.3f",
-            index,
-            len(chunks),
-            chunk_id,
-            speaker,
-            source_start,
-            source_end,
-        )
-
-        # ----------------------------------------------------
-        # Choose language for this chunk
-        # ----------------------------------------------------
+        timeline = chunk.get("source_timeline", {})
+        source_start = max(0.0, _safe_float(timeline.get("start", 0.0)))
+        source_end = max(source_start, _safe_float(timeline.get("end", source_start)))
 
         if DETECT_LANGUAGE_PER_CHUNK:
             language_for_chunk = None
         else:
-            language_for_chunk = project_language
-
-            # Auto detection has not yet found a sufficiently confident
-            # project language.
-            if (
-                requested_language is None
-                and (
-                    project_language is None
-                    or (
-                        project_language_probability is not None
-                        and project_language_probability
-                        < LANGUAGE_LOCK_THRESHOLD
-                    )
-                )
-            ):
-                language_for_chunk = None
+            language_for_chunk = requested_language
+            if requested_language is None and project_language is not None:
+                language_for_chunk = project_language
 
         inference_started = time.perf_counter()
-
         try:
             decoded_segments, info = _transcribe_one(
                 model,
                 audio_path,
                 language_for_chunk,
             )
+            inference_seconds = time.perf_counter() - inference_started
 
-            inference_seconds = (
-                time.perf_counter()
-                - inference_started
-            )
-
-            text = _extract_text(
-                decoded_segments
-            )
+            text = _extract_text(decoded_segments)
+            detected_language, language_probability = _language_from_info(info)
+            final_chunk_language = detected_language or project_language
 
-            detected_language, language_probability = (
-                _language_from_info(
-                    info
-                )
-            )
-
-            if detected_language:
-                language_votes.append(
-                    (
-                        detected_language,
-                        language_probability or 0.0,
-                    )
-                )
-
-                # Lock the project language only after sufficient confidence.
-                if (
-                    requested_language is None
-                    and not DETECT_LANGUAGE_PER_CHUNK
-                    and (
-                        project_language is None
-                        or (
-                            language_probability is not None
-                            and language_probability
-                            > (
-                                project_language_probability
-                                or 0.0
-                            )
-                        )
-                    )
-                ):
-                    project_language = detected_language
-                    project_language_probability = (
-                        language_probability
-                    )
-
-            final_chunk_language = (
-                detected_language
-                or project_language
-            )
-
-            words = _extract_word_timestamps(
-                decoded_segments,
-                source_start,
-                source_end,
-            )
-
-            transcript_segments = _extract_segments(
-                decoded_segments,
-                source_start,
-                source_end,
-            )
+            words = _extract_word_timestamps(decoded_segments, source_start, source_end)
+            transcript_segments = _extract_segments(decoded_segments, source_start, source_end)
 
             word_probabilities = [
-                float(
-                    item["probability"]
-                )
-                for item in words
-                if "probability" in item
+                float(item["probability"]) for item in words if "probability" in item
             ]
-
             average_word_probability = (
-                sum(word_probabilities)
-                / len(word_probabilities)
+                sum(word_probabilities) / len(word_probabilities)
                 if word_probabilities
                 else None
             )
@@ -1409,20 +1147,9 @@ def run_transcription(
                 "text": text,
                 "language": final_chunk_language,
                 "language_probability": (
-                    round(
-                        language_probability,
-                        4,
-                    )
+                    round(language_probability, 4)
                     if language_probability is not None
-                    else (
-                        round(
-                            project_language_probability,
-                            4,
-                        )
-                        if project_language_probability
-                        is not None
-                        else None
-                    )
+                    else None
                 ),
                 "model": MODEL_NAME,
                 "engine": "faster-whisper",
@@ -1436,73 +1163,33 @@ def run_transcription(
                 "word_count": len(words),
                 "segments": transcript_segments,
                 "average_word_probability": (
-                    round(
-                        average_word_probability,
-                        4,
-                    )
-                    if average_word_probability
-                    is not None
+                    round(average_word_probability, 4)
+                    if average_word_probability is not None
                     else None
                 ),
-                "inference_time_seconds": round(
-                    inference_seconds,
-                    3,
-                ),
+                "inference_time_seconds": round(inference_seconds, 3),
             }
 
             chunk["subtitle"] = {
                 "ready": bool(words),
-                "absolute_start": round(
-                    source_start,
-                    3,
-                ),
-                "absolute_end": round(
-                    source_end,
-                    3,
-                ),
-                "word_timestamps_available": bool(
-                    words
-                ),
+                "absolute_start": round(source_start, 3),
+                "absolute_end": round(source_end, 3),
+                "word_timestamps_available": bool(words),
             }
 
-            successful += 1
-            total_words += len(words)
-            total_inference_seconds += (
-                inference_seconds
-            )
-
-            LOGGER.info(
-                "[%d/%d] OK chunk=%s words=%d language=%s "
-                "prob=%.3f inference=%.2fs",
-                index,
-                len(chunks),
-                chunk_id,
-                len(words),
-                final_chunk_language or "unknown",
-                language_probability
-                if language_probability is not None
-                else (
-                    project_language_probability
-                    or 0.0
-                ),
-                inference_seconds,
-            )
+            return {
+                "status": "success",
+                "index": index,
+                "chunk_id": chunk_id,
+                "speaker": speaker,
+                "words_len": len(words),
+                "inference_seconds": inference_seconds,
+                "language": detected_language,
+                "probability": language_probability,
+            }
 
         except Exception as exc:
-            inference_seconds = (
-                time.perf_counter()
-                - inference_started
-            )
-
-            failed += 1
-
-            LOGGER.exception(
-                "[%d/%d] FAILED chunk=%s",
-                index,
-                len(chunks),
-                chunk_id,
-            )
-
+            inference_seconds = time.perf_counter() - inference_started
             chunk["transcription"] = {
                 "status": "failed",
                 "completed": False,
@@ -1517,55 +1204,102 @@ def run_transcription(
                 "words": [],
                 "word_count": 0,
                 "segments": [],
-                "inference_time_seconds": round(
-                    inference_seconds,
-                    3,
-                ),
+                "inference_time_seconds": round(inference_seconds, 3),
                 "error": str(exc),
             }
-
             chunk["subtitle"] = {
                 "ready": False,
-                "absolute_start": round(
-                    source_start,
-                    3,
-                ),
-                "absolute_end": round(
-                    source_end,
-                    3,
-                ),
+                "absolute_start": round(source_start, 3),
+                "absolute_end": round(source_end, 3),
                 "word_timestamps_available": False,
             }
+            return {"status": "failed_exception", "index": index, "chunk_id": chunk_id, "error": str(exc)}
 
-        finally:
-            gc.collect()
+    # ========================================================================
+    # CONCURRENT CHUNK EXECUTION
+    # ========================================================================
 
-        # ----------------------------------------------------
-        # Checkpoint after every chunk
-        # ----------------------------------------------------
+    with ThreadPoolExecutor(max_workers=NUM_WORKERS) as executor:
+        futures = {
+            executor.submit(process_chunk, index, chunk): (index, chunk)
+            for index, chunk in enumerate(chunks, start=1)
+        }
 
-        if (
-            index % CHECKPOINT_EVERY_CHUNKS == 0
-        ):
-            manifest["step_004_progress"] = {
-                "processed": index,
-                "total": len(chunks),
-                "successful": successful,
-                "failed": failed,
-                "skipped": skipped,
-            }
+        for future in as_completed(futures):
+            res = future.result()
+            status = res["status"]
+            index = res["index"]
+            chunk_id = res["chunk_id"]
+
+            with manifest_lock:
+                completed_counter += 1
+
+                if status == "success":
+                    successful += 1
+                    total_words += res["words_len"]
+                    total_inference_seconds += res["inference_seconds"]
+                    lang = res["language"]
+                    prob = res["probability"]
+                    if lang:
+                        language_votes.append((lang, prob or 0.0))
+                        if (
+                            requested_language is None
+                            and not DETECT_LANGUAGE_PER_CHUNK
+                            and (
+                                project_language is None
+                                or (prob is not None and prob > (project_language_probability or 0.0))
+                            )
+                        ):
+                            project_language = lang
+                            project_language_probability = prob
+
+                    LOGGER.info(
+                        "[%d/%d] OK chunk=%s words=%d language=%s prob=%.3f inference=%.2fs",
+                        index,
+                        len(chunks),
+                        chunk_id,
+                        res["words_len"],
+                        res["language"] or project_language or "unknown",
+                        prob or project_language_probability or 0.0,
+                        res["inference_seconds"],
+                    )
 
-            _checkpoint_manifest(
-                paths.segments_manifest,
-                manifest,
-            )
+                elif status == "skipped":
+                    skipped += 1
+                    if res["language"]:
+                        language_votes.append((str(res["language"]), res["probability"]))
+                    LOGGER.info(
+                        "[%d/%d] chunk=%s speaker=%s -> SKIP",
+                        index,
+                        len(chunks),
+                        chunk_id,
+                        res["speaker"],
+                    )
 
-        if (
-            index % MEMORY_LOG_INTERVAL == 0
-        ):
-            _log_memory(
-                f"after chunk {index}/{len(chunks)}"
-            )
+                else:
+                    failed += 1
+                    LOGGER.error(
+                        "[%d/%d] chunk=%s -> %s",
+                        index,
+                        len(chunks),
+                        chunk_id,
+                        res.get("error", "Unknown error"),
+                    )
+
+                if completed_counter % CHECKPOINT_EVERY_CHUNKS == 0:
+                    manifest["step_004_progress"] = {
+                        "processed": completed_counter,
+                        "total": len(chunks),
+                        "successful": successful,
+                        "failed": failed,
+                        "skipped": skipped,
+                    }
+                    _checkpoint_manifest(paths.segments_manifest, manifest)
+
+                if completed_counter % MEMORY_LOG_INTERVAL == 0:
+                    _log_memory(f"after chunk {completed_counter}/{len(chunks)}")
+
+            gc.collect()
 
     # =========================================================================
     # FINAL LANGUAGE
@@ -1664,7 +1398,6 @@ def run_transcription(
         ),
     }
 
-    # Remove temporary progress marker after completion.
     manifest.pop(
         "step_004_progress",
         None,
@@ -1992,4 +1725,4 @@ if __name__ == "__main__":
         LOGGER.exception(
             "Step 004 failed."
         )
-        raise SystemExit(1)
+        raise SystemExit(1)