Timestamp Alignment & Correction

Within the Transcription & Speaker Diarization pipeline, timestamp alignment and correction is the deterministic refinement stage that fuses two probabilistic streams — word-level transcription timing and speaker turn boundaries — into a single validated, frame-accurate schema. It protects the SLA the rest of the system silently depends on: sub-100ms temporal precision on every chapter marker, caption cue, and editorial cut point, held constant across long-form media and variable acoustic conditions. When this stage is weak, drift accumulates invisibly until subtitles slide out of sync, automated chapters land mid-sentence, and editors burn hours scrubbing. This component turns the upstream models’ best guesses into markers an automated rendering engine can trust without human review.

Prerequisites & Environment

The alignment worker assumes Python 3.10+ (for structural pattern matching on segment event types and precise type hints on the timing models) and a pinned audio toolchain baked into the container image rather than installed from the host package manager — librosa, scipy, and numpy version skew silently changes RMS framing and peak indices, which makes “deterministic” output drift across nodes. This stage runs strictly downstream of inference: it consumes the word-level timestamps emitted by the Whisper Large V3 integration and the turn boundaries emitted by speaker diarization with Pyannote, and must not execute until both upstream streams for a given media_id have converged.

# Python 3.10+
# numpy==1.26.4           # memmap audio buffers + vectorized energy math
# librosa==0.10.1         # RMS / spectral-flux feature extraction
# scipy==1.12.0           # find_peaks for VAD boundary snapping
# pydantic==2.6.4         # payload-contract validation
# soundfile==0.12.1       # streaming slice reads (libsndfile backend)
# structlog==24.1.0       # JSON structured logging
# prometheus-client==0.20.0  # drift + MAE metrics

CPU-bound and memory-sensitive, the alignment worker belongs on a separate worker class from GPU inference so that feature extraction never competes with the acoustic models for VRAM — the same hardware-affinity discipline the async transcription queue management layer enforces when it dispatches completion events to this stage. The environment variables below govern everything dynamic; frame rates, codec assumptions, and snap windows that must stay reproducible across a release live in version-controlled config, not the shell.

This component consumes codec-normalized audio guaranteed by the Media Ingestion & Format Architecture boundary, so it can assume a known sample rate and a single fetchable object per media_id and spend its cycles on correction rather than transcode validation.

Architecture & Queue Topology

The alignment worker is a stateful consumer in the processing graph, triggered by a completion event rather than a polling loop. It leases one media_id at a time, fetches the normalized audio and both timing streams, runs a fixed sequence of deterministic transformations, and emits a corrected manifest. The internal flow has three serial passes — payload validation, acoustic boundary snapping, and cross-stream reconciliation — followed by frame quantization and emission. Each pass is pure with respect to its inputs, which is what makes a redelivered event safe to reprocess: identical inputs yield byte-identical outputs.

Heavy audio never traverses the event bus. The worker fetches directly from object storage and maps the file with numpy.memmap, reading only the 30-second windows it needs, so a four-hour interview holds a stable heap footprint instead of loading whole into RAM. Idempotency is anchored on media_id plus alignment_version: the worker checks a centralized ledger before emitting, so a re-fired completion event finds the existing manifest and re-acknowledges instead of recomputing.

Step-by-Step Implementation

Each step ends with a verification command you can run before wiring the next one.

1. Validate and normalize the inbound timing payload. The worker rejects anything that violates the timing contract before it touches audio — non-monotonic ordering, speaker-turn overlap beyond MAX_OVERLAP_MS, or confidence below the floor are deterministic faults, not transient ones:

# Python 3.10+
# pydantic==2.6.4
from pydantic import BaseModel, Field, model_validator
from typing import Literal

class Word(BaseModel):
    text: str
    start_ms: int = Field(ge=0)
    end_ms: int = Field(gt=0)

class Segment(BaseModel):
    start_ms: int = Field(ge=0)
    end_ms: int = Field(gt=0)
    speaker_id: str
    confidence: float = Field(ge=0.0, le=1.0)
    words: list[Word]

class AlignmentPayload(BaseModel):
    media_id: str
    sample_rate: Literal[16000, 44100, 48000]
    alignment_version: str
    segments: list[Segment]

    @model_validator(mode="after")
    def monotonic(self) -> "AlignmentPayload":
        last = -1
        for seg in self.segments:
            if seg.start_ms < last:
                raise ValueError(f"non-monotonic segment at {seg.start_ms}ms")
            last = seg.end_ms
        return self

Verify the contract rejects a malformed payload before any audio I/O:

python -c "from align import AlignmentPayload; \
AlignmentPayload.model_validate_json(open('payload.json').read()); print('OK')"

2. Snap boundaries to voice activity with a bounded search window. Probabilistic boundaries drift into silence or mid-syllable. Snapping forces each start/end to the nearest energy peak inside a ±SNAP_WINDOW_MS window, loading only the audio slice it needs:

# Python 3.10+
# numpy==1.26.4 ; librosa==0.10.1 ; scipy==1.12.0
import numpy as np
import librosa
from scipy.signal import find_peaks

def snap_boundary_to_vad(audio_path: str, target_ms: float,
                         sr: int = 48000, window_ms: int = 150) -> float:
    """Snap a timestamp to the nearest RMS energy peak within ±window_ms.

    Loads only [0, target_ms + window_ms] of audio to bound memory; returns
    target_ms unchanged when no peak is found so the result is deterministic.
    """
    load_duration = (target_ms + window_ms) / 1000.0
    y, sr = librosa.load(audio_path, sr=sr, mono=True, duration=load_duration)

    hop_length = 960  # ~20ms at 48kHz — keep fixed for reproducibility
    rms = librosa.feature.rms(y=y, frame_length=2048, hop_length=hop_length)[0]

    frame_target = int((target_ms / 1000.0) * sr / hop_length)
    search_frames = int(window_ms / 1000.0 * sr / hop_length)
    slice_start = max(0, frame_target - search_frames)
    slice_end = min(len(rms), frame_target + search_frames)
    local_energy = rms[slice_start:slice_end]
    if local_energy.size == 0:
        return target_ms

    peaks, _ = find_peaks(local_energy,
                          height=np.percentile(local_energy, 75), distance=2)
    if len(peaks) == 0:
        return target_ms

    center_in_slice = frame_target - slice_start
    best_peak = peaks[np.argmin(np.abs(peaks - center_in_slice))]
    return (slice_start + best_peak) * hop_length / sr * 1000.0

Apply a 6dB hysteresis threshold so background noise never triggers a false snap: only accept the adjustment when the energy delta across the boundary exceeds the threshold inside a 50ms window. Verify snapping is stable across two identical runs:

python -c "from align import snap_boundary_to_vad as s; \
a=s('media.wav',12450); b=s('media.wav',12450); \
assert a==b, 'non-deterministic snap'; print(round(a,2))"

3. Reconcile word timing against diarization turns with DTW. Dynamic time warping aligns the transcribed word sequence against the diarization turn sequence, minimizing temporal distortion while preserving order, so speaker labels never bleed across a boundary:

# Python 3.10+
# numpy==1.26.4 ; scipy==1.12.0
import numpy as np
from scipy.spatial.distance import cdist

def dtw_reconcile(word_centers_ms: np.ndarray,
                  turn_centers_ms: np.ndarray, step_penalty: float = 1.0):
    """Greedy DTW path between word and turn midpoints.

    step_penalty is a fixed scalar (not sampled) so the warp path is identical
    on every worker — a requirement for reproducible CI validation.
    """
    cost = cdist(word_centers_ms[:, None], turn_centers_ms[:, None], "cityblock")
    n, m = cost.shape
    acc = np.full((n + 1, m + 1), np.inf)
    acc[0, 0] = 0.0
    for i in range(1, n + 1):
        for j in range(1, m + 1):
            acc[i, j] = cost[i - 1, j - 1] + min(
                acc[i - 1, j] + step_penalty,
                acc[i, j - 1] + step_penalty,
                acc[i - 1, j - 1],
            )
    return acc[n, m]  # total warp distance; expose per-segment for telemetry

Segments below CONFIDENCE_FLOOR are not force-snapped — they are flagged for the deferral path described under Resilience. Verify the warp distance is finite and reproducible:

python -c "import numpy as np; from align import dtw_reconcile; \
print(dtw_reconcile(np.array([100.,300.]), np.array([110.,290.])))"

4. Quantize corrected markers to the target frame rate. Editorial consumers need SMPTE-compatible markers, so the worker rounds each corrected boundary to the nearest frame at TARGET_FPS and applies a configurable lead/lag compensation for decoder buffering:

# Python 3.10+
def quantize_to_frame(ms: float, fps: int = 24, lead_frames: int = 0) -> int:
    """Round a millisecond marker to the nearest frame boundary, then shift
    by lead_frames (negative = earlier) to absorb decoder/render latency."""
    frame_ms = 1000.0 / fps
    frame_idx = round(ms / frame_ms) + lead_frames
    return int(max(0, frame_idx) * frame_ms)

Verify a marker lands exactly on a 24fps frame grid:

python -c "from align import quantize_to_frame; \
v=quantize_to_frame(13120, 24); assert v % round(1000/24) == 0; print(v)"

5. Emit a drift-annotated alignment manifest. The worker publishes a structured event carrying the corrected segments plus the per-segment delta between raw and corrected boundaries, so downstream stages and telemetry consume the same record:

# Python 3.10+
import json

def emit_aligned(r, media_id: str, manifest_uri: str, mae_ms: float) -> None:
    event = {
        "media_id": media_id,
        "manifest_uri": manifest_uri,
        "mean_abs_drift_ms": round(mae_ms, 2),
        "pipeline_stage": "aligned",   # consumers route off this flag
    }
    r.xadd("events.alignment", {"event": json.dumps(event)})

Verify the event reaches the bus the chapter and caption consumers subscribe to:

redis-cli XREVRANGE events.alignment + - COUNT 1

Data Contracts

The handoff into and out of the alignment stage is governed by an explicit schema. Validating it at the boundary is what makes reprocessing idempotent and downstream rendering predictable regardless of upstream model variation. The inbound payload carries word-level timing, speaker turns, and confidence; the outbound manifest adds frame-quantized markers and a drift annotation.

A payload missing monotonic ordering, carrying speaker-turn overlap beyond MAX_OVERLAP_MS, or scoring below CONFIDENCE_FLOOR is rejected at the boundary rather than half-processed. The alignment_version is part of the idempotency key so that bumping the transform set forces a clean recompute instead of mixing old and new corrections in the same manifest.

Resilience Patterns

Every alignment job is a pure transformation that can fail wholly but never partially. Idempotency is anchored on media_id plus alignment_version: the worker checks the ledger before emitting, so a redelivered completion event under at-least-once delivery absorbs into a re-acknowledgement instead of a second recompute. This makes the event bus from the async transcription queue management layer safe to retry.

Failure classification decides the path. Transient faults — an object-storage timeout on audio fetch, a memory-mapped read interrupted by a host eviction — retry with bounded exponential backoff and jitter. Deterministic faults — a payload that fails validation, a corrupted audio header, an unsupported sample rate — route straight to the dead-letter queue with the raw upstream payload preserved and a pipeline_stage: "skipped" flag, so a downstream consumer can fall back to uncorrected timestamps rather than block. That quarantine-and-replay machinery is specified once in the retry logic and dead-letter queues patterns and reused here rather than reinvented.

Low-confidence segments are a distinct case from failure. When source material suffers compression artifacts, room resonance, or cross-talk, the worker defers instead of force-snapping: segments below CONFIDENCE_FLOOR are routed to a secondary verification queue — a higher-precision diarization pass or a lightweight human-in-the-loop review — while high-confidence segments around them are still emitted. Graceful degradation means a single ambiguous turn never stalls the whole manifest. The cross-host orchestration of these deferral and retry pools belongs to Celery task routing for video jobs, which this stage feeds.

Observability & Debugging

Operational visibility comes from structured logs correlated by media_id and a small set of high-signal metrics, not from scraping worker stdout. The single most important signal is drift telemetry — the delta between raw model timestamps and corrected boundaries — because a sudden rise in mean absolute error is the earliest evidence of upstream model degradation or an audio-format change. Instrument these metrics:

• align_drift_ms — histogram of per-boundary raw→corrected delta (the core health signal)
• align_mae_ms{media_id} — mean absolute error per asset; alert on a step change vs. the rolling baseline
• boundary_collisions_total — counter of segments violating MIN_SEGMENT_MS or MAX_OVERLAP_MS
• deferred_segments_total — counter of sub-confidence segments routed to secondary verification
• align_rss_bytes — worker resident memory; alert above 80% of the cgroup ceiling

The error messages you will actually see map to concrete root causes:

Performance Tuning

Throughput is governed by I/O discipline and bounded memory, not raw CPU. Map audio with numpy.memmap and process in 30-second sliding windows so a two-hour interview at 48kHz holds a flat heap instead of paging — whole-file loads are the most common cause of MemoryError on long-form assets. Keep hop_length and frame_length fixed across the fleet; changing RMS framing between releases silently shifts peak indices and breaks the byte-identical-output guarantee that CI relies on.

Pin every stochastic knob — VAD percentile thresholds, the DTW step_penalty — to a fixed value rather than sampling, so all worker nodes produce the same warp path during validation. Cap each worker’s RSS with a cgroup limit so a pathological file is OOM-killed and DLQ’d instead of swapping the host, and tie horizontal scaling to event-bus depth rather than CPU utilization so a burst of completion events grows queue depth instead of thrashing. Containerize with pinned ffmpeg, librosa, and numpy versions in a multi-stage build that strips debug symbols, and consult the official librosa documentation for feature-extraction parameter specifics. The frame-accurate downstream consumer of these markers is the aligning speaker labels with video cuts workflow, which maps corrected turns onto editorial cut boundaries.

Frequently Asked Questions

Related

• Up to the parent overview: Transcription & Speaker Diarization
• The boundary this stage feeds: aligning speaker labels with video cuts
• Upstream word-level timing source: Whisper Large V3 integration
• Upstream speaker-turn source: speaker diarization with Pyannote
• The bus that triggers this stage: async transcription queue management
• Where deterministic failures are quarantined: retry logic and dead-letter queues