From f57aed1a7d7aaa9e70804ce4f39ece68500efbeb Mon Sep 17 00:00:00 2001 From: Tachikoma Date: Fri, 5 Jun 2026 13:50:13 +0800 Subject: [PATCH] feat(core): add get_prior_action() to Application Adds the symmetric counterpart to get_next_action(). The data already lives in state under __PRIOR_STEP -- this exposes it as a public API. Returns None when no step has been executed yet (app at entrypoint). Closes #502 --- burr/core/application.py | 14 ++++++++++++++ tests/core/test_application.py | 32 ++++++++++++++++++++++++++++++++ 2 files changed, 46 insertions(+) diff --git a/burr/core/application.py b/burr/core/application.py index 25bce4a10..0da3ab943 100644 --- a/burr/core/application.py +++ b/burr/core/application.py @@ -2034,6 +2034,20 @@ def _set_state(self, new_state: State[ApplicationStateType]): def get_next_action(self) -> Optional[Action]: return self._graph.get_next_node(self._state.get(PRIOR_STEP), self._state, self.entrypoint) + def get_prior_action(self) -> Optional[Action]: + """Returns the action that was executed in the prior step. + + This is the symmetric counterpart to :py:meth:`get_next_action`. + Returns ``None`` if no step has been executed yet (i.e. the application + is still at its entrypoint and has not run any action). + + :return: The last-executed :py:class:`Action`, or ``None`` if no step has run. + """ + prior_step = self._state.get(PRIOR_STEP) + if prior_step is None: + return None + return self._graph.get_action(prior_step) + def update_state(self, new_state: State[ApplicationStateType]): """Updates state -- this is meant to be called if you need to do anything with the state. For example: diff --git a/tests/core/test_application.py b/tests/core/test_application.py index 9313cefc9..8ce0b67d4 100644 --- a/tests/core/test_application.py +++ b/tests/core/test_application.py @@ -3128,6 +3128,38 @@ def test_app_get_next_step(): assert app.get_next_action().name == "counter_1" +def test_app_get_prior_step(): + counter_action_1 = base_counter_action.with_name("counter_1") + counter_action_2 = base_counter_action.with_name("counter_2") + counter_action_3 = base_counter_action.with_name("counter_3") + app = Application( + state=State(), + entrypoint="counter_1", + partition_key="test", + uid="test-123", + sequence_id=0, + graph=Graph( + actions=[counter_action_1, counter_action_2, counter_action_3], + transitions=[ + Transition(counter_action_1, counter_action_2, default), + Transition(counter_action_2, counter_action_3, default), + Transition(counter_action_3, counter_action_1, default), + ], + ), + ) + # uninitialized -- no prior step yet + assert app.get_prior_action() is None + app.step() + # ran counter_1 + assert app.get_prior_action().name == "counter_1" + app.step() + # ran counter_2 + assert app.get_prior_action().name == "counter_2" + app.step() + # ran counter_3 + assert app.get_prior_action().name == "counter_3" + + def test_application_builder_complete(): app = ( ApplicationBuilder()