symfonic.platform.dispatch_capture

dispatch_capture

Bound real dispatch observers to selected journals and capture boundaries.

dispatch_capture_evidence

dispatch_capture_evidence(manifest: Mapping[str, Any] | None, selected: set[Path], resolve: Callable[[Path], Path], *, start: datetime, end: datetime) -> tuple[list[str], int | None, int | None, set[Path], dict[frozenset[Path], int] | None]

Require every replica's journal and measured observer snapshot pair.

Source code in src/symfonic/platform/dispatch_capture.py

def dispatch_capture_evidence(
    manifest: Mapping[str, Any] | None,
    selected: set[Path],
    resolve: Callable[[Path], Path],
    *,
    start: datetime,
    end: datetime,
) -> tuple[list[str], int | None, int | None, set[Path],
           dict[frozenset[Path], int] | None]:
    """Require every replica's journal and measured observer snapshot pair."""
    if manifest is None:
        return [], None, None, set(), None
    replicas = manifest.get("expected_replicas")
    if (
        not isinstance(replicas, list)
        or not replicas
        or not all(isinstance(replica, str) for replica in replicas)
        or len(set(replicas)) != len(replicas)
    ):
        return ["missing_dispatch_journal_inventory"], None, None, set(), None
    journals = _inventory(manifest.get("expected_dispatch_journals"), replicas)
    if journals is None:
        if "expected_dispatch_journals" not in manifest:
            return [], None, None, set(), None  # Old reports remain readable; zero needs health.
        return ["missing_dispatch_journal_inventory"], None, None, set(), None
    issues: list[str] = []
    bound: set[Path] = set()
    for replica in replicas:
        for spec in journals[replica].values():
            for name in spec["files"]:
                try:
                    path = resolve(Path(name))
                except (OSError, RuntimeError, ValueError):
                    issues.append("dispatch_journal_file_inventory_mismatch")
                    continue
                if path not in selected:
                    issues.append("dispatch_journal_file_inventory_mismatch")
                if path in bound:
                    issues.append("dispatch_journal_file_overlap")
                bound.add(path)
    captures = manifest.get("capture_boundaries")
    if not isinstance(captures, dict) or set(captures) != set(replicas):
        return issues + ["missing_dispatch_capture_boundaries"], None, None, bound, None
    failures = active_end = 0
    expected: dict[frozenset[Path], int] = {}
    valid = not issues
    for replica in replicas:
        pair = captures[replica]
        if not isinstance(pair, dict):
            issues.append("missing_dispatch_capture_boundaries")
            valid = False
            continue
        snapshots = []
        for boundary, limit in (("start", start), ("end", end)):
            record = pair.get(boundary)
            value = record.get("dispatch_snapshots") if isinstance(record, dict) else None
            parsed = _snapshots(value, list(journals[replica]))
            if parsed is None:
                issues.append(f"missing_dispatch_capture_snapshot_{boundary}")
                valid = False
            elif any(snapshot[1] for snapshot in parsed.values()):
                issues.append(f"dispatch_capture_in_flight_{boundary}")
                valid = False
            snapshots.append(parsed)
            try:
                observed = _observed_at(record["observed_at"])
            except (KeyError, TypeError, ValueError, OverflowError):
                valid = False  # The shared boundary validator reports the time issue.
            else:
                if (boundary == "start" and observed > limit) or (
                    boundary == "end" and observed < limit
                ):
                    valid = False
        before, after = snapshots
        if before is None or after is None:
            continue
        for journal_id in journals[replica]:
            old, new = before[journal_id], after[journal_id]
            active_end += new[1]
            if old[2] != new[2]:
                issues.append("dispatch_capture_epoch_changed")
                valid = False
            elif new[0] < old[0] or new[3] < old[3]:
                issues.append("dispatch_capture_counter_reset")
                valid = False
            else:
                try:
                    files = frozenset(
                        resolve(Path(name)) for name in journals[replica][journal_id]["files"]
                    )
                except (OSError, RuntimeError, ValueError):
                    issues.append("dispatch_journal_file_inventory_mismatch")
                    valid = False
                    continue
                expected[files] = new[3] - old[3]
                delta = new[0] - old[0]
                failures += delta
                if delta:
                    issues.append("dispatch_capture_failed")
                    valid = False
    return (issues, failures if valid or failures else None,
            active_end if valid else None, bound, expected if valid else None)

dispatch_count_issues

dispatch_count_issues(expected: dict[frozenset[Path], int] | None, observed: Mapping[Path, int]) -> tuple[list[str], int | None]

Reconcile each declared observer with rows in all its selected rotations.

Source code in src/symfonic/platform/dispatch_capture.py

def dispatch_count_issues(
    expected: dict[frozenset[Path], int] | None, observed: Mapping[Path, int],
) -> tuple[list[str], int | None]:
    """Reconcile each declared observer with rows in all its selected rotations."""
    if expected is None:
        return [], None
    if any(sum(observed.get(path, 0) for path in files) != count
           for files, count in expected.items()):
        return ["dispatch_record_count_mismatch"], None
    return [], sum(expected.values())