Add exception transitions

[elijahbenizzy]

Currently, exceptions will break the control flow of an action, stopping the program early. Thus, if an exception is expected, the program will stop early. We will be adding the ability to conditionally transition based on exceptions, which will allow you to transition to an error-handling (or retry) action that does not need the full outputs of the prior action.

Here is what it would look liek in the current API:

@action(reads=["attempts"], writes=["output", "attempts"])
def some_flaky_action(state: State, max_retries: int=3) -> Tuple[dict, State]:
    result = {"output": None, "attempts": state["attempts"] + 1}
    try:
        result["output"] = call_some_api(...)
    excecpt APIException as e:
        if state["attempts"] >= max_retries:
           raise e
    return result, state.update(**result)

One could imagine adding it as a condition (a few possibilities)

@action(reads=[], writes=["output"])
def some_flaky_action(state: State) -> Tuple[dict, State]:
    result = {"output": call_some_api(...)}
    return result, state.update(**result)

builder.with_actions(
   some_flaky_action=some_flaky_action
).with_transitions(
   (
      "some_flaky_action",
      "some_flaky_action",
      error(APIException) # infinite retries
      error(APIException, max=3) # 3 visits to this edge then it gets reset if this is not chosen
      # That's stored in state
)

[skrawcz]

Could also directly annotate it:

@action(reads=[], writes=["output"], error=(APIException, max=3))
def some_flaky_action(state: State) -> Tuple[dict, State]:
    result = {"output": call_some_api(...)}
    return result, state.update(**result)

Under the hood it would:

1. rerun the function with the same state inputs, except for incrementing an invocation counter.

and then it would display a loop back to the node based on error and then otherwise error the program out...

[skrawcz]

Or if it errors you want to set some state value.

@action(reads=[], writes=["output"], error={"write_to": "error_key"})
def some_flaky_action(state: State) -> Tuple[dict, State]:
    result = {"output": call_some_api(...)}
    return result, state.update(**result)

[skrawcz]

But I think what's missing here is

1. if some action errors
2. I want it to go to some specific "action" that I have defined on error
3. what's the best way to do that

Right now you have to manually define all the edges. But there could be a more systematic way.

[drdraad]

Hello

I have the following use case:

1. I want to transition to a default action (let's call it RESET) when an exception occurs in any state action.

2. This RESET action would reset the state and transition back to the initial state.

3. Before transitioning, I'd like to catch the exception, modify the state to indicate that an exception occurred, and include a stack trace message for user communication.

The challenge I'm facing is that manually adding transitions from hundreds of states to the RESET state is impractical. Ideally, I'm looking for a catchall transition mechanism, something along the lines of (*, "RESET", when(exception_occurred=True))

A global exception handler might be a good approach:

def global_exception_handler(state: State, e: Exception) -> State:
   return state.update(
       exception_occurred=True,
       error_message=str(e),
       stack_trace=traceback.format_exc()
   )


builder.with_global_exception_handler(
   handler=global_exception_handler,
   target_state="RESET"
)

[elijahbenizzy]

Hello

I have the following use case:

1. I want to transition to a default action (let's call it RESET) when an exception occurs in any state action.
2. This RESET action would reset the state and transition back to the initial state.
3. Before transitioning, I'd like to catch the exception, modify the state to indicate that an exception occurred, and include a stack trace message for user communication.

The challenge I'm facing is that manually adding transitions from hundreds of states to the RESET state is impractical. Ideally, I'm looking for a catchall transition mechanism, something along the lines of (*, "RESET", when(exception_occurred=True))

A global exception handler might be a good approach:

def global_exception_handler(state: State, e: Exception) -> State:
   return state.update(
       exception_occurred=True,
       error_message=str(e),
       stack_trace=traceback.format_exc()
   )


builder.with_global_exception_handler(
   handler=global_exception_handler,
   target_state="RESET"
)

Some extra ideas from Slack:

@skrawcz says:

Yeah wildcard transition specifications seem like a good idea:

("*", "RESET", when(exception=True))

seems reasonable -- they would have to be at the end though. Yeah interesting point <@599951704330338304> -- also I think we can expose state fields (E.G. exception) that suppresses and stores it. So:

@action(reads=..., writes=..., on_error=capture_as(field="exception")))
def ...
    ...

or on application build, for everything

app = (
    ApplicationBuilder()...
    .with_error_handling=capture_as(field="exception"))
    # then everything transitions to it
    .with_transitions("*", error_handler, expr("exception is not None"))
    .build