symfonic.agent.cutover.dispatch_outcomes

dispatch_outcomes

Opt-in observations of real invocation dispatch, never turn payloads.

This is not a population certificate. A host must explicitly install a sink; an absent sink records nothing and changes no route. The observer counts sink failures so an instrumented host can detect capture loss without failing turns.

DispatchOutcomeEvent dataclass

DispatchOutcomeEvent(entrypoint: str, outcome: str, capability: str, reason_code: str = 'none', selected_route: str = 'none')

Closed, metadata-only result of one actual dispatch attempt.

DispatchOutcomeObserver

DispatchOutcomeObserver(sink: Callable[[DispatchOutcomeEvent], None] | None)

Observe each selected body or approved preflight refusal once.

Source code in src/symfonic/agent/cutover/dispatch_outcomes.py

def __init__(self, sink: Callable[[DispatchOutcomeEvent], None] | None) -> None:
    self._sink = sink
    self._failures = 0
    self._in_flight = 0
    self._delivered = 0
    self._epoch = uuid4().hex
    self._lock = Lock()

capture_counts property

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

Return health and successful sink calls from one atomic reading.

capture_snapshot property

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

Return measured health only when a dispatch sink is installed.

direct_legacy

direct_legacy(entrypoint: str, board: object) -> None

Classify an existing legacy route without reading free-text reasons.

Source code in src/symfonic/agent/cutover/dispatch_outcomes.py

def direct_legacy(self, entrypoint: str, board: object) -> None:
    """Classify an existing legacy route without reading free-text reasons."""
    if self._sink is None:
        return
    with self._capture():
        try:
            capability = CAPABILITIES[entrypoint]
            if board.rollback_reason(capability) is not None:
                reason = "rollback"
            elif board.legacy_pin.holds(capability):
                reason = "legacy_pin"
            else:
                reason = "route_not_flipped"
        except Exception:  # noqa: BLE001 - telemetry cannot alter the turn
            with self._lock:
                self._failures += 1
            logger.warning("dispatch outcome classification failed for %s", entrypoint)
            return
        self.record(entrypoint, "direct_legacy", reason)

guard

guard(entrypoint: str, function: Callable, *args: object, **kwargs: object)

Observe a known pre-dispatch refusal without changing its exception.

Source code in src/symfonic/agent/cutover/dispatch_outcomes.py

def guard(self, entrypoint: str, function: Callable, *args: object, **kwargs: object):
    """Observe a known pre-dispatch refusal without changing its exception."""
    with self.preflight(entrypoint):
        return function(*args, **kwargs)

pre_body

pre_body(entrypoint: str, selected_route: str) -> Iterator[None]

Account for preparation failing after a route is selected.

Source code in src/symfonic/agent/cutover/dispatch_outcomes.py

@contextmanager
def pre_body(self, entrypoint: str, selected_route: str) -> Iterator[None]:
    """Account for preparation failing after a route is selected."""
    try:
        yield
    except (Exception, CancelledError):
        self.record(
            entrypoint,
            "pre_body_failure",
            "pre_body_exception",
            selected_route=selected_route,
        )
        raise

preflight

preflight(entrypoint: str) -> Iterator[None]

Only the known cutover-refusal family is an approved rejection.

Source code in src/symfonic/agent/cutover/dispatch_outcomes.py

@contextmanager
def preflight(self, entrypoint: str) -> Iterator[None]:
    """Only the known cutover-refusal family is an approved rejection."""
    try:
        yield
    except CutoverPathError:
        self.record(entrypoint, "approved_rejection", "preflight_refusal")
        raise
    except (Exception, CancelledError):
        self.record(entrypoint, "pre_body_failure", "pre_body_exception")
        raise