Speech Feature Pipelines and Data Contracts

Questions: Are intervals closed or half-open? Should touching intervals merge? Are inputs sorted? Can intervals be empty? Invariant: after sorting, the output is disjoint and the last interval is the only candidate that can overlap the next interval. Test cases: empty input, one interval, nested intervals, touching intervals, small allowed gaps, and unsorted input.

def merge_audio_chunks(chunks, max_gap_ms=0):
    if max_gap_ms < 0:
        raise ValueError("max_gap_ms must be non-negative")
    if not chunks:
        return []

    ordered = sorted(chunks)
    for start, end in ordered:
        if start > end:
            raise ValueError("chunk start must be <= end")

    merged = [list(ordered[0])]

    for start, end in ordered[1:]:
        last = merged[-1]
        if start <= last[1] + max_gap_ms:
            last[1] = max(last[1], end)
        else:
            merged.append([start, end])

    return [tuple(item) for item in merged]


assert merge_audio_chunks([(30, 40), (0, 10), (10, 20)]) == [(0, 20), (30, 40)]
assert merge_audio_chunks([(0, 10), (13, 20)], max_gap_ms=3) == [(0, 20)]
try:
    merge_audio_chunks([(20, 10)])
    raise AssertionError("expected validation failure")
except ValueError:
    pass

Questions: Can private content be selected? Are scores stable under ties? Should one user dominate the queue? Common mistakes include ignoring consent, sorting the entire stream when a heap is enough, using raw transcript content in logs, and optimizing uncertainty while starving approved rare slices. A production queue should also require minimum slice support, cap or diversify per-user contributions, and avoid exposing raw transcripts before sending examples to reviewers.

from heapq import heappush, heappushpop
from math import isfinite


def rank_review_candidates(rows, k, min_slice_count=20):
    if not isinstance(k, int) or k < 0:
        raise ValueError("k must be a non-negative integer")
    if not isinstance(min_slice_count, int) or min_slice_count <= 0:
        raise ValueError("min_slice_count must be a positive integer")
    if k == 0:
        return []
    required = ("uncertainty", "slice_rarity", "regression", "slice_count", "example_id")
    heap = []
    for row in rows:
        if not row.get("consented"):
            continue
        missing = [name for name in required if name not in row]
        if missing:
            raise ValueError(f"missing candidate fields: {missing}")
        for name in ("uncertainty", "slice_rarity", "regression"):
            value = row[name]
            if not isinstance(value, (int, float)) or not isfinite(value) or not 0 <= value <= 1:
                raise ValueError(f"{name} must be a finite score between 0 and 1")
        if not isinstance(row["slice_count"], int) or row["slice_count"] < min_slice_count:
            continue
        score = (
            0.55 * row["uncertainty"]
            + 0.30 * row["slice_rarity"]
            + 0.15 * row["regression"]
        )
        item = (score, row["example_id"])
        if len(heap) < k:
            heappush(heap, item)
        else:
            heappushpop(heap, item)

    return [example_id for score, example_id in sorted(heap, reverse=True)]


rows = [
    {"example_id": "a", "consented": True, "uncertainty": 0.9, "slice_rarity": 0.2, "regression": 0.3, "slice_count": 80},
    {"example_id": "b", "consented": False, "uncertainty": 1.0, "slice_rarity": 1.0, "regression": 1.0, "slice_count": 100},
    {"example_id": "c", "consented": True, "uncertainty": 0.5, "slice_rarity": 0.9, "regression": 0.8, "slice_count": 40},
    {"example_id": "d", "consented": True, "uncertainty": 0.8, "slice_rarity": 1.0, "regression": 1.0, "slice_count": 3},
]
assert rank_review_candidates(rows, 2) == ["c", "a"]

from math import isfinite


def first_coverage_drop(values, window, threshold):
    if not isinstance(window, int) or window <= 0:
        raise ValueError("window must be a positive integer")
    if not isinstance(threshold, (int, float)) or not isfinite(threshold) or not 0 <= threshold <= 1:
        raise ValueError("threshold must be between 0 and 1")
    for value in values:
        if not isinstance(value, (int, float)) or not isfinite(value) or not 0 <= value <= 1:
            raise ValueError("coverage values must be finite rates between 0 and 1")
    if len(values) < window:
        return None

    total = sum(values[:window])
    if total / window < threshold:
        return 0

    for right in range(window, len(values)):
        total += values[right] - values[right - window]
        left = right - window + 1
        if total / window < threshold:
            return left

    return None


assert first_coverage_drop([0.99, 0.98, 0.70, 0.72, 0.74], 3, 0.85) == 1
assert first_coverage_drop([0.99, 0.98, 0.97], 2, 0.95) is None
try:
    first_coverage_drop([0.99, 1.2], 2, 0.95)
    raise AssertionError("expected validation failure")
except ValueError:
    pass

Production follow-up: page only when the drop affects critical fields or approved aggregate launch slices, deduplicate alerts by app version and ingestion path, and attach rollout context so on-call can identify whether the first bad hour matches a deployment.