As-Run Reconciliation and Discrepancy Handling

Every linear break the automation plays out ends with a confession: the playout system emits an as-run log recording what actually aired, when it started, and for how long — and that record almost never matches the schedule byte-for-byte. A spot slid four seconds because the preceding segment ran long, a thirty stretched to a thirty-two on encode, an entire pod vanished behind an Emergency Alert System crawl. Reconciliation is the subsystem that reads both logs, matches each aired event back to the spot it was supposed to be, and classifies the outcome — aired-as-scheduled, shifted, short, long, preempted, or missing — so that billing invoices only what truly ran and the traffic desk gets a clean queue of make-good candidates. This page treats that comparison as a deterministic engineering problem within Broadcast Traffic Architecture & Taxonomy: typed records in, a matching key, a classifier, and an auditable result out. It is written for two readers at once — traffic managers who need the plain-language reasoning behind each classification, and Python automation builders who need a deployable, strictly typed module.

Reconciliation sits at the seam between playout and revenue. It consumes the same normalized identifiers defined by the canonical traffic field data dictionary, it must agree with the billing keys produced by billing code normalization, and every MISSING or PREEMPTED it emits is the trigger that arms make-good routing for preemptions. Get the matching wrong and the station either bills for spots that never aired — an FCC and contractual exposure — or writes off revenue it legitimately earned.

Concept & Data Model #

Reconciliation compares two ordered event streams for a single channel and broadcast day. The scheduled stream is the intended log; the as-run stream is playout telemetry. The engine pairs them on a deterministic matching key, classifies each pair, and packages the outcome. Four entities carry a spot through that pipeline, each with a stable schema so classification stays decoupled from ingestion and billing.

The classification vocabulary is fixed, because each value drives a different downstream action. The DiscrepancyType enum carries AIRED_AS_SCHEDULED (matched within tolerance — fully billable), SHIFTED (aired outside the start-time tolerance but at full duration — billable, flagged), SHORT (aired for materially less than booked — partial credit), LONG (aired over — billable at the booked length, overrun noted), PREEMPTED (a scheduled spot displaced by a higher-priority event such as EAS or a network break), MISSING (scheduled but never found in the as-run log — not billable, make-good candidate), and UNMATCHED_ASRUN (an aired event with no scheduled counterpart — an ingest or trafficking error to investigate). Only AIRED_AS_SCHEDULED, SHIFTED, and LONG are billable at full rate; SHORT bills pro-rata; PREEMPTED and MISSING bill nothing and route to recovery.

The deterministic matching key is the crux. Playout systems rarely echo the traffic system’s spot_id, so the engine cannot join on it. Instead it derives a key from attributes both logs agree on — the house_number (the station’s stable creative identifier) scoped to the broadcast day and a coarse time bucket — and resolves ties by nearest aired_at. That keeps matching stable when a break is reordered but its members all air, and it degrades predictably when they do not.

Figure — Reconciliation matches each scheduled spot to an as-run event on a deterministic key, then classifies the pair by start-time drift and duration delta; unmatched spots become make-good candidates and orphan as-run events are flagged for investigation.

Implementation Approach #

Three design decisions shape a production reconciliation engine, and each is a trade-off worth stating explicitly.

Match on stable attributes, not on the traffic spot_id. Playout automation and the traffic system are different vendors with different primary keys, so the join key must be derived from what both agree on. The engine keys on house_number scoped to the broadcast day, then disambiguates duplicates within that day by nearest aired_at. This is the same discipline that lets billing code normalization reconcile a house number back to an ISCI and an ad-ID: identity is computed from business attributes, never inherited from one system’s surrogate key.

Classify with explicit tolerances, not exact equality. No spot airs at its scheduled millisecond, and encoders round durations. A reconciliation that demands exact matches would flag every airing as a discrepancy and drown the traffic desk in noise. The engine instead applies a start-time tolerance (a spot that drifts within it is AIRED_AS_SCHEDULED, beyond it is SHIFTED) and a duration tolerance expressed as a fraction of booked length (so a thirty tolerates more absolute drift than a ten). Tolerances are configuration owned by traffic managers, not constants baked into the classifier, and the mechanics of tuning them are covered in detecting as-run vs scheduled discrepancies.

Deterministic and idempotent. Reconciliation runs at end of broadcast day, but it also re-runs when a late as-run correction arrives or an auditor replays an archived log months later. The same two logs must always produce the same classifications and the same billable flags, so the engine holds no hidden state and derives every result purely from its inputs. That property is what lets a billing dispute be settled by re-running the reconciliation against the archived logs and getting a byte-identical answer.

Before any of this runs, both logs must be shaped to the canonical field contract. Preemption context — was this pod displaced by an EAS emergency interrupt or by a network break signalled over SCTE-35? — is resolved upstream and attached to the as-run stream, so the classifier can distinguish a deliberate PREEMPTED from a genuine MISSING.

Production Python Implementation #

The module below is deployable as a library or a batch worker. It uses Pydantic v2 for schema enforcement, strict type hints throughout, a deterministic matching key, configurable tolerances, and structured logging in the traffic-ops format timestamp | level | module | spot_id.

from __future__ import annotations

import logging
from datetime import datetime, timezone
from enum import Enum
from typing import Optional

from pydantic import BaseModel, Field, field_validator

# --- Structured logging: timestamp | level | module | spot_id ---------------
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
)
logger = logging.getLogger("asrun_reconciler")


def _log(level: int, spot_id: str, msg: str) -> None:
    # Prepend the spot_id so every audit line is greppable by placement.
    logger.log(level, "%s | %s", spot_id, msg)


class DiscrepancyType(str, Enum):
    AIRED_AS_SCHEDULED = "aired_as_scheduled"  # Matched within tolerance
    SHIFTED = "shifted"                        # Full length, start out of tolerance
    SHORT = "short"                            # Materially under booked length
    LONG = "long"                              # Aired over booked length
    PREEMPTED = "preempted"                    # Displaced by EAS / network break
    MISSING = "missing"                        # Scheduled, never aired
    UNMATCHED_ASRUN = "unmatched_asrun"        # Aired, never scheduled


# Full-rate billable classifications; SHORT bills pro-rata, the rest bill zero.
_FULL_RATE = {
    DiscrepancyType.AIRED_AS_SCHEDULED,
    DiscrepancyType.SHIFTED,
    DiscrepancyType.LONG,
}


class ScheduledSpot(BaseModel):
    spot_id: str
    house_number: str                    # Station-stable creative id; the join key
    isci: str                            # Industry standard commercial identifier
    sponsor_id: str
    scheduled_at: datetime               # Intended airing (UTC)
    expected_duration_ms: int = Field(gt=0)
    preemptible: bool = True             # False for firm, non-preemptible units

    @field_validator("scheduled_at")
    @classmethod
    def _utc_aware(cls, v: datetime) -> datetime:
        # Naive datetimes silently corrupt drift math across a DST boundary.
        if v.tzinfo is None:
            raise ValueError("scheduled_at must be timezone-aware (UTC)")
        return v.astimezone(timezone.utc)


class AsRunEntry(BaseModel):
    event_id: str
    house_number: str
    aired_at: datetime                   # Actual airing start (UTC)
    actual_duration_ms: int = Field(ge=0)
    verified: bool = True                # Playout confirmed frames were emitted
    preemption_reason: Optional[str] = None  # e.g. "eas_activation", "network_break"

    @field_validator("aired_at")
    @classmethod
    def _utc_aware(cls, v: datetime) -> datetime:
        if v.tzinfo is None:
            raise ValueError("aired_at must be timezone-aware (UTC)")
        return v.astimezone(timezone.utc)


class Discrepancy(BaseModel):
    spot_id: str
    discrepancy_type: DiscrepancyType
    start_drift_ms: int = 0              # signed: positive = aired late
    duration_delta_ms: int = 0          # signed: positive = aired long
    detail: str = ""


class ReconciliationResult(BaseModel):
    spot_id: str
    matched_event_id: Optional[str]
    discrepancy: Discrepancy
    billable: bool
    billable_duration_ms: int           # what billing should invoice against
    decided_at: datetime = Field(
        default_factory=lambda: datetime.now(timezone.utc)
    )


class ReconciliationConfig(BaseModel):
    # Start-time drift within this bound is not a discrepancy at all.
    start_tolerance_ms: int = 2_000
    # Duration is allowed to vary by this fraction of the booked length.
    duration_tolerance_pct: float = Field(default=0.05, gt=0, lt=1)
    # Matching bucket: two spots for one house number inside this window
    # are disambiguated by nearest aired_at rather than treated as distinct.
    match_bucket_ms: int = 900_000       # 15 minutes


class Reconciler:
    """Deterministic as-run vs scheduled reconciliation for one channel-day."""

    def __init__(self, config: Optional[ReconciliationConfig] = None) -> None:
        self.config = config or ReconciliationConfig()

    @staticmethod
    def _broadcast_key(house_number: str, when: datetime) -> str:
        # Broadcast day rolls at local 06:00; here we key on the UTC date the
        # traffic system already assigned, keeping the join deterministic.
        return f"{house_number}:{when.date().isoformat()}"

    def _match(
        self, spot: ScheduledSpot, candidates: list[AsRunEntry]
    ) -> Optional[AsRunEntry]:
        # Among unconsumed as-run events sharing the house/day key, pick the
        # one whose aired_at is closest to the scheduled time (ties -> earliest).
        key = self._broadcast_key(spot.house_number, spot.scheduled_at)
        pool = [
            e for e in candidates
            if self._broadcast_key(e.house_number, e.aired_at) == key
        ]
        if not pool:
            return None
        return min(pool, key=lambda e: abs(
            (e.aired_at - spot.scheduled_at).total_seconds()
        ))

    def _classify(
        self, spot: ScheduledSpot, entry: AsRunEntry
    ) -> Discrepancy:
        start_drift = int(
            (entry.aired_at - spot.scheduled_at).total_seconds() * 1000
        )
        duration_delta = entry.actual_duration_ms - spot.expected_duration_ms
        dur_tol = int(spot.expected_duration_ms * self.config.duration_tolerance_pct)

        # A verified airing that is materially short of the booked length only
        # earns a pro-rata credit; classify SHORT before SHIFTED so a spot that
        # is both late and truncated bills correctly.
        if entry.actual_duration_ms == 0 or not entry.verified:
            return Discrepancy(spot_id=spot.spot_id,
                               discrepancy_type=DiscrepancyType.MISSING,
                               start_drift_ms=start_drift,
                               duration_delta_ms=duration_delta,
                               detail="zero-length or unverified airing")
        if duration_delta < -dur_tol:
            return Discrepancy(spot_id=spot.spot_id,
                               discrepancy_type=DiscrepancyType.SHORT,
                               start_drift_ms=start_drift,
                               duration_delta_ms=duration_delta,
                               detail=f"aired {abs(duration_delta)}ms under booked")
        if duration_delta > dur_tol:
            return Discrepancy(spot_id=spot.spot_id,
                               discrepancy_type=DiscrepancyType.LONG,
                               start_drift_ms=start_drift,
                               duration_delta_ms=duration_delta,
                               detail=f"aired {duration_delta}ms over booked")
        if abs(start_drift) > self.config.start_tolerance_ms:
            return Discrepancy(spot_id=spot.spot_id,
                               discrepancy_type=DiscrepancyType.SHIFTED,
                               start_drift_ms=start_drift,
                               duration_delta_ms=duration_delta,
                               detail=f"start drifted {start_drift}ms")
        return Discrepancy(spot_id=spot.spot_id,
                           discrepancy_type=DiscrepancyType.AIRED_AS_SCHEDULED,
                           start_drift_ms=start_drift,
                           duration_delta_ms=duration_delta)

    def _billable(
        self, spot: ScheduledSpot, disc: Discrepancy
    ) -> tuple[bool, int]:
        # LONG bills at the booked length (never invoice an overrun); SHORT
        # bills pro-rata against what actually aired; full-rate types bill full.
        if disc.discrepancy_type in _FULL_RATE:
            return True, spot.expected_duration_ms
        if disc.discrepancy_type is DiscrepancyType.SHORT:
            aired = spot.expected_duration_ms + disc.duration_delta_ms
            return True, max(aired, 0)
        return False, 0  # MISSING / PREEMPTED / UNMATCHED_ASRUN

    def reconcile(
        self, scheduled: list[ScheduledSpot], as_run: list[AsRunEntry]
    ) -> list[ReconciliationResult]:
        results: list[ReconciliationResult] = []
        remaining = list(as_run)  # consumed as matches are made

        for spot in scheduled:
            match = self._match(spot, remaining)
            if match is None:
                # No airing at all. If the pod carried a preemption reason on a
                # sibling event, prefer PREEMPTED; else it is genuinely MISSING.
                dtype = (DiscrepancyType.PREEMPTED
                         if spot.preemptible and self._pod_preempted(spot, as_run)
                         else DiscrepancyType.MISSING)
                disc = Discrepancy(spot_id=spot.spot_id, discrepancy_type=dtype,
                                   detail="no matching as-run event")
                _log(logging.WARNING, spot.spot_id,
                     f"{dtype.value} | no as-run match | make-good candidate")
                results.append(ReconciliationResult(
                    spot_id=spot.spot_id, matched_event_id=None,
                    discrepancy=disc, billable=False, billable_duration_ms=0))
                continue

            remaining.remove(match)  # one as-run event serves one scheduled spot
            disc = self._classify(spot, match)
            billable, bill_ms = self._billable(spot, disc)
            _log(logging.INFO, spot.spot_id,
                 f"{disc.discrepancy_type.value} | drift={disc.start_drift_ms}ms "
                 f"delta={disc.duration_delta_ms}ms billable={billable}")
            results.append(ReconciliationResult(
                spot_id=spot.spot_id, matched_event_id=match.event_id,
                discrepancy=disc, billable=billable,
                billable_duration_ms=bill_ms))

        # Any as-run event never consumed aired without a schedule row.
        for orphan in remaining:
            _log(logging.ERROR, orphan.event_id,
                 "unmatched_asrun | aired with no scheduled spot")
            results.append(ReconciliationResult(
                spot_id=f"ASRUN:{orphan.event_id}", matched_event_id=orphan.event_id,
                discrepancy=Discrepancy(
                    spot_id=f"ASRUN:{orphan.event_id}",
                    discrepancy_type=DiscrepancyType.UNMATCHED_ASRUN,
                    detail="no scheduled counterpart"),
                billable=False, billable_duration_ms=0))
        return results

    @staticmethod
    def _pod_preempted(spot: ScheduledSpot, as_run: list[AsRunEntry]) -> bool:
        # A spot is preempted (not merely missing) when another event in the
        # same broadcast day carries an explicit preemption reason, evidence the
        # pod was displaced rather than dropped by a trafficking error.
        day = spot.scheduled_at.date()
        return any(e.preemption_reason and e.aired_at.date() == day
                   for e in as_run)

The reconciler never mutates the source logs and never talks to billing directly. It returns a list of ReconciliationResult envelopes; a separate report builder (see the companion guide on generating discrepancy reports for billing) aggregates those into credits, make-good candidates, and audit rows. Keeping the classifier pure means a disputed invoice can be re-adjudicated by replaying the archived logs.

Validation & Edge Cases #

Broadcast operations produce boundary conditions that a naive reconciler mishandles. Each of the following is exercised by the implementation above.

• DST and timezone drift. Start-time drift is a difference of two datetimes; a naive local timestamp crossing a spring-forward boundary distorts the delta by a full hour and turns every spot in the affected hour into a false SHIFTED. The Pydantic validators reject naive datetimes outright and normalize everything to UTC before any subtraction happens.
• Sports overrun sliding a whole break. An overrun does not truncate one spot; it pushes an entire pod later. Because matching keys on house number and day rather than exact time, and disambiguates by nearest airing, a uniformly shifted pod still matches spot-for-spot — each member simply classifies as SHIFTED with a similar positive drift, not as a cascade of MISSING.
• Preemption versus genuine absence. A spot with no as-run event is MISSING only if nothing displaced it. When a sibling event in the day carries a preemption_reason — an EAS activation or a network break signalled over SCTE-35 — the classifier prefers PREEMPTED, which carries different contractual weight even though both route to make-good.
• Zero-duration and unverified airings. An as-run event with actual_duration_ms == 0 or verified == False means playout logged the intent but not the frames. The classifier treats it as MISSING rather than SHORT, because a spot that emitted no video is not a partial airing — it did not air.
• Duplicate house numbers in one day. The same creative can be booked several times per day. The matcher consumes each as-run event exactly once (remaining.remove(match)), so two scheduled airings of one house number claim two distinct as-run events and never double-match a single airing.

Integration Points #

Upstream, both logs are shaped to the canonical contract before reconciliation runs. The scheduled log comes from the traffic system; the as-run log is playout telemetry mapped into AsRunEntry by an ingestion adapter that attaches preemption context. Field names and constraints follow the canonical traffic field data dictionary, so the reconciler works against a single canonical shape rather than vendor-specific payloads.

Downstream, each ReconciliationResult is serialized to a wire message that billing and make-good routing both consume. The contract is deliberately small and carries the billable_duration_ms so billing never has to re-derive it:

{
  "message_type": "reconciliation.result.v1",
  "spot_id": "SP-2026-0716-0442",
  "matched_event_id": "AR-CH7-20260716-1837",
  "discrepancy_type": "short",
  "start_drift_ms": 1400,
  "duration_delta_ms": -4000,
  "billable": true,
  "billable_duration_ms": 26000,
  "house_number": "H7-QSR-030",
  "isci": "ABCD1234030H",
  "decided_at": "2026-07-17T05:12:44Z"
}

A PREEMPTED or MISSING result on this wire is exactly the signal that arms make-good routing for preemptions: the routing engine reads the unfulfilled obligation and searches for a compliant replacement window. The billable and billable_duration_ms fields, meanwhile, feed the billing export so an invoice reflects only what aired at what length — consistent with the join keys established in billing code normalization.

Compliance & Audit Considerations #

Reconciliation is compliance-critical because it decides what a station may invoice, and a mis-classification either overstates revenue or forfeits it. Three obligations apply directly.

Bill only what aired. Invoicing a MISSING spot is not merely a billing error; for a political or issue advertisement it is a public-file discrepancy an auditor will find. The engine’s billable flag is the authoritative gate — nothing classified MISSING, PREEMPTED, or UNMATCHED_ASRUN reaches an invoice, and SHORT airings bill strictly against the length that actually ran.

Immutable, replayable audit trail. Every classification is emitted on the timestamp | level | module | spot_id log line and should be shipped to append-only storage alongside both source logs. Because reconciliation is deterministic, an auditor or a disputing advertiser can replay the archived scheduled and as-run logs months later and reproduce the identical ReconciliationResult, which is what SOC 2 reproducibility and an FCC public-inspection review require.

Lineage back to the airing. Each result carries matched_event_id, tying the billing decision to the exact as-run event that justified it. That linkage is what lets a station defend an invoice line: the house number, the ISCI, the aired timestamp, and the measured duration are all reconstructable from one immutable record, and they agree with the identifiers defined in the broadcast traffic architecture taxonomy.

Troubleshooting & Common Errors #

The ERR_CLOCK_SKEW case deserves particular attention. When an entire daypart classifies as SHIFTED with a near-constant drift, the fault is almost never the spots — it is an unsynchronized clock on one of the two source systems. Widening the start tolerance to silence it would mask genuine shifts elsewhere in the day; the correct fix is upstream, and the tuning trade-offs are covered in detecting as-run vs scheduled discrepancies.

Related #

• Broadcast Traffic Architecture & Taxonomy — the parent architecture that defines the identifiers, contracts, and audit guarantees this reconciliation engine inherits.
• Detecting As-Run vs Scheduled Discrepancies — the step-by-step matching and classification procedure with tunable start-time and duration tolerances.
• Generating Discrepancy Reports for Billing — how reconciliation results aggregate into credits, make-good candidates, and audit rows for the billing run.
• Automating Make-Good Routing for Preemptions — the routing subsystem that a MISSING or PREEMPTED classification arms with an unfulfilled obligation.