Skip to content

symfonic.agent.cutover.route_attempts

route_attempts

Opt-in attempt capture with measured delivery health, separate from dispatch.

RouteAttemptObserver 

RouteAttemptObserver(sink: Callable[[KernelRouteAttemptEvent], None] | None)

Atomic capture counters, independent of admission and dispatch outcomes.

Source code in src/symfonic/agent/cutover/route_attempts.py 
17
18
19
20
21
def __init__(self, sink: Callable[[KernelRouteAttemptEvent], None] | None) -> None:
    self._sink = sink
    self._failures = self._in_flight = self._delivered = 0
    self._epoch = uuid4().hex
    self._lock = Lock()

capture_counts property 

capture_counts: tuple[int, int, str, int] | None

Return measured health only when a sink is installed.

record 

record(entrypoint: str) -> None

Observe once immediately after the actual invocation selects Kernel.

Source code in src/symfonic/agent/cutover/route_attempts.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def record(self, entrypoint: str) -> None:
    """Observe once immediately after the actual invocation selects Kernel."""
    if self._sink is None:
        return
    with self._lock:
        self._in_flight += 1
    delivered = False
    try:
        self._sink(KernelRouteAttemptEvent(entrypoint, CAPABILITIES[entrypoint]))
        delivered = True
    except Exception:  # noqa: BLE001 - telemetry cannot change admission
        logger.warning("Kernel route attempt capture failed")
    finally:
        with self._lock:
            self._in_flight -= 1
            if delivered:
                self._delivered += 1
            else:
                self._failures += 1