Skip to content

symfonic.agent.cutover.typed_interrupts

typed_interrupts

Compatibility import for the native typed pause translator.

interrupt_events 

interrupt_events(event: AgentEvent) -> tuple[StreamEvent, ...]

Project one paused kernel event onto the typed pause vocabulary.

Returns () for anything that is not a pause, so the caller can hand every event through without branching on the kind twice.

Source code in src/symfonic/agent/typed_interrupts.py 
40
41
42
43
44
45
46
47
48
49
50
51
def interrupt_events(event: AgentEvent) -> tuple[StreamEvent, ...]:
    """Project one paused kernel event onto the typed pause vocabulary.

    Returns ``()`` for anything that is not a pause, so the caller can hand
    every event through without branching on the kind twice.
    """
    pending = event.interrupt
    if pending is None:
        return ()
    if event.kind == "ask_user":
        return (_ask_user(event, pending),)
    return (_generic(event, pending),)