symfonic.platform.capture_boundary

capture_boundary

Fail-closed operator-attested fallback capture boundaries.

The report cannot observe ingress or a fleet. A true attestation here is an operator claim, not executable quiescence proof or retirement authority.

capture_boundary_evidence

capture_boundary_evidence(manifest: Mapping[str, Any] | None, *, start: datetime, end: datetime) -> tuple[list[str], int | None, int | None, int | None, int | None]

Return issues, fallback and native failure deltas and end active counts.

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

def capture_boundary_evidence(
    manifest: Mapping[str, Any] | None, *, start: datetime, end: datetime,
) -> tuple[list[str], int | None, int | None, int | None, int | None]:
    """Return issues, fallback and native failure deltas and end active counts."""
    if manifest is None:
        return [], None, None, None, None  # The caller reports missing_manifest.
    replicas = manifest.get("expected_replicas")
    captures = manifest.get("capture_boundaries")
    if (not isinstance(replicas, list) or not replicas
            or not all(isinstance(replica, str) for replica in replicas)
            or len(set(replicas)) != len(replicas)
            or not isinstance(captures, dict)
            or set(captures) != set(replicas)):
        return ["missing_capture_boundaries"], None, None, None, None

    journals = _journal_inventory(manifest.get("expected_native_journals"), replicas)
    issues: list[str] = []
    native_valid = journals is not None
    if journals is None:
        issues.append("missing_native_journal_inventory")
    total_failures = total_active_end = 0
    native_failures = native_active_end = 0
    counts_valid = True
    for replica in replicas:
        pair = captures[replica]
        if not isinstance(pair, dict):
            issues.append("missing_capture_boundaries")
            counts_valid = False
            continue
        snapshots: list[tuple[int, int, str] | None] = []
        native_snapshots: list[dict[str, tuple[int, int, str]] | None] = []
        times: list[datetime | None] = []
        for boundary in ("start", "end"):
            record = pair.get(boundary)
            if not isinstance(record, dict):
                issues.append(f"missing_capture_boundary_{boundary}")
                snapshots.append(None)
                native_snapshots.append(None)
                times.append(None)
                continue
            for claim in _CLAIMS:
                if record.get(claim) is not True:
                    issues.append(f"missing_{claim}_{boundary}")
            if "observed_at" not in record:
                issues.append(f"missing_capture_boundary_observed_at_{boundary}")
                times.append(None)
            else:
                try:
                    times.append(_observed_at(record["observed_at"]))
                except (ValueError, TypeError, OverflowError):
                    issues.append(f"invalid_capture_boundary_observed_at_{boundary}")
                    times.append(None)
            snapshot = _snapshot(record.get("snapshot"))
            if snapshot is None:
                issues.append(f"missing_capture_snapshot_{boundary}")
            elif snapshot[1]:
                issues.append(f"fallback_capture_in_flight_{boundary}")
            snapshots.append(snapshot)
            if journals is None:
                native_snapshots.append(None)
            else:
                native_snapshot = _native_snapshots(
                    record.get("native_snapshots"), list(journals[replica]),
                )
                if native_snapshot is None:
                    issues.append(f"missing_native_capture_snapshot_{boundary}")
                elif any(snapshot[1] for snapshot in native_snapshot.values()):
                    issues.append(f"native_capture_in_flight_{boundary}")
                native_snapshots.append(native_snapshot)
        observed_start, observed_end = times
        if observed_start is None or observed_end is None:
            counts_valid = False
        else:
            if observed_start >= observed_end:
                issues.append("capture_boundary_order_invalid")
                counts_valid = False
            if observed_start > start:
                issues.append("capture_boundary_start_after_window")
                counts_valid = False
            if observed_end < end:
                issues.append("capture_boundary_end_before_window")
                counts_valid = False
        before, after = snapshots
        if before is None or after is None:
            counts_valid = False
            continue
        total_active_end += after[1]
        if before[2] != after[2]:
            issues.append("fallback_capture_epoch_changed")
            counts_valid = False
        elif after[0] < before[0]:
            issues.append("fallback_capture_counter_reset")
            counts_valid = False
        else:
            delta = after[0] - before[0]
            total_failures += delta
            if delta:
                issues.append("fallback_capture_failed")
        native_before, native_after = native_snapshots
        if native_before is None or native_after is None or journals is None:
            native_valid = False
            continue
        for journal_id in journals[replica]:
            journal_before, journal_after = native_before[journal_id], native_after[journal_id]
            native_active_end += journal_after[1]
            if journal_before[2] != journal_after[2]:
                issues.append("native_capture_epoch_changed")
                native_valid = False
            elif journal_after[0] < journal_before[0]:
                issues.append("native_capture_counter_reset")
                native_valid = False
            else:
                native_delta = journal_after[0] - journal_before[0]
                native_failures += native_delta
                if native_delta:
                    issues.append("native_capture_failed")
    return (
        issues,
        total_failures if counts_valid else None,
        total_active_end if counts_valid else None,
        native_failures if counts_valid and native_valid else None,
        native_active_end if counts_valid and native_valid else None,
    )

native_journal_count_rows

native_journal_count_rows(manifest: Mapping[str, Any], counts: Mapping[tuple[str, str], int], bindings: Mapping[Path, tuple[str, str, str]]) -> list[dict[str, Any]]

Serialize row-derived counts against the validated operator inventory.

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

def native_journal_count_rows(
    manifest: Mapping[str, Any], counts: Mapping[tuple[str, str], int],
    bindings: Mapping[Path, tuple[str, str, str]],
) -> list[dict[str, Any]]:
    """Serialize row-derived counts against the validated operator inventory."""
    journals = manifest["expected_native_journals"]
    return [
        {"replica": replica, "journal_id": journal_id,
         "entrypoint": spec["entrypoint"], "declared_file_count": len(spec["files"]),
         "resolved_files": sorted(str(path) for path, owner in bindings.items()
                                  if owner[:2] == (replica, journal_id)),
         "count": counts.get((replica, journal_id), 0)}
        for replica in sorted(journals)
        for journal_id, spec in sorted(journals[replica].items())
    ]

native_journal_file_bindings

native_journal_file_bindings(manifest: Mapping[str, Any] | None, selected: set[Path], resolve: Callable[[Path], Path]) -> tuple[list[str], dict[Path, tuple[str, str, str]]]

Bind selected files to declared replica, journal and entrypoint labels.

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

def native_journal_file_bindings(
    manifest: Mapping[str, Any] | None, selected: set[Path],
    resolve: Callable[[Path], Path],
) -> tuple[list[str], dict[Path, tuple[str, str, str]]]:
    """Bind selected files to declared replica, journal and entrypoint labels."""
    if manifest is None:
        return [], {}
    replicas = manifest.get("expected_replicas")
    if not isinstance(replicas, list) or not all(isinstance(replica, str) for replica in replicas):
        return [], {}
    journals = _journal_inventory(manifest.get("expected_native_journals"), replicas)
    if journals is None:
        return [], {}  # The capture-boundary validator reports this.
    issues: list[str] = []
    bindings: dict[Path, tuple[str, str, str]] = {}
    for replica in replicas:
        for journal_id, spec in journals[replica].items():
            for name in spec["files"]:
                try:
                    path = resolve(Path(name))
                except (OSError, RuntimeError, ValueError):
                    issues.append("native_journal_file_inventory_mismatch")
                    continue
                if path not in selected:
                    issues.append("native_journal_file_inventory_mismatch")
                if path in bindings:
                    issues.append("native_journal_file_overlap")
                bindings[path] = replica, journal_id, spec["entrypoint"]
    return issues, bindings