symfonic.platform.route_attempt_capture

route_attempt_capture

Reconcile independent route-attempt observers with their selected journal rows.

route_attempt_capture_evidence

route_attempt_capture_evidence(manifest, selected, resolve, *, start, end)

Use the same strict inventory/epoch/delivery rules as dispatch capture.

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

def route_attempt_capture_evidence(manifest, selected, resolve, *, start, end):
    """Use the same strict inventory/epoch/delivery rules as dispatch capture."""
    if manifest is None or "expected_route_attempt_journals" not in manifest:
        return [], None, None, set(), None
    translated = dict(manifest)
    translated["expected_dispatch_journals"] = manifest["expected_route_attempt_journals"]
    boundaries = manifest.get("capture_boundaries")
    if isinstance(boundaries, dict):
        translated["capture_boundaries"] = {
            replica: {
                boundary: (dict(record, dispatch_snapshots=record.get("route_attempt_snapshots"))
                           if isinstance(record, dict) else record)
                for boundary, record in pair.items()
            } if isinstance(pair, dict) else pair
            for replica, pair in boundaries.items()
        }
    issues, failures, active, bound, expected = dispatch_capture_evidence(
        translated, selected, resolve, start=start, end=end,
    )
    return [_code(code) for code in issues], failures, active, bound, expected

route_attempt_report

route_attempt_report(groups: Mapping[tuple[str, str], int], observed, capture, *, declared: bool) -> tuple[dict[str, Any], list[str]]

Keep readable older reports; only measured, reconciled capture is healthy.

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

def route_attempt_report(
    groups: Mapping[tuple[str, str], int], observed, capture, *, declared: bool,
) -> tuple[dict[str, Any], list[str]]:
    """Keep readable older reports; only measured, reconciled capture is healthy."""
    issues, failures, active, _, expected = capture
    count_issues, delivered = dispatch_count_issues(expected, observed)
    issues = issues + [_code(code) for code in count_issues]
    total = sum(groups.values())
    if declared and not total:
        issues.append("empty_route_attempt_journal")
    return {
        "kernel_route_attempt_count": total,
        "kernel_route_attempt_groups": [
            {"entrypoint": entry, "capability": capability, "count": count}
            for (entry, capability), count in sorted(groups.items())
        ],
        "route_attempt_capture_failure_count": failures,
        "route_attempt_capture_in_flight_end": active,
        "route_attempt_capture_delivered_count": delivered,
        "route_attempt_capture_status": (
            "healthy" if failures == 0 and active == 0 and not issues else "incomplete"
        ),
    }, issues