How to Test AI Agent Crash Recovery Locally with Dapr Workflow

Most advice on testing AI agents is about the answers, such as evals and golden datasets. Crash recovery needs its own test. If your agent runs as a durable workflow, a restart should resume the same run without repeating finished steps or side effects. You can prove that on your laptop, but only if the crash happens at exactly the same place every time, and pressing Ctrl+C half a second late does not give you that. In this post you put the crash in the code, run a pytest integration test against a Dapr Workflow agent, and check the outcome by counting lines in files. For background on how replay skips completed steps, read the durable agents guide.

The commands in this post run unchanged in bash and in Windows PowerShell, except the manual run in the verification section, which shows both.

The agent triages support tickets in two steps. The summarize activity stands in for your LLM call, and post_comment stands in for the tool that writes to the ticket. Replace both with something you can count: summarize appends the workflow ID to llm_calls.txt, and post_comment appends one line to ledger.txt. One comment is one line, so correct means one line. DATA_DIR lets every test run write to its own folder.

import os import sys from pathlib import Path import dapr.ext.workflow as wf wfr = wf. WorkflowRuntime () DATA_DIR = Path ( os. getenv ( " DATA_DIR ", ". " )) @wfr.workflow ( name = " triage " ) def triage ( ctx: wf.

DaprWorkflowContext, ticket: dict ): summary = yield ctx. call_activity ( summarize, input = ticket ) yield ctx. call_activity ( post_comment, input = summary ) return " done " @wfr.activity ( name = " summarize " ) def summarize ( ctx: wf. WorkflowActivityContext, ticket: dict ): with open ( DATA_DIR / " llm_calls.txt ", " a " ) as f: f. write ( f " { ctx. workflow_id } \n " ) # your LLM call goes here return f " Summary: { ticket [ ' title ' ] } " @wfr.activity ( name = " post_comment " ) def post_comment ( ctx: wf. WorkflowActivityContext, summary: str ): with open ( DATA_DIR / " ledger.txt ", " a " ) as f: f. write ( f " { ctx. workflow_id }: { summary } \n " ) return " posted " if __name__ == " __main__ ": mode, instance_id = sys. argv [ 1 ], sys. argv [ 2 ] wfr. start () client = wf.

DaprWorkflowClient () if mode == " start ": client. schedule_new_workflow ( workflow = triage, input = { " title ": " Login page 500s " }, instance_id = instance_id, ) state = client. wait_for_workflow_completion ( instance_id, timeout_in_seconds = 60 ) print ( f " STATUS= { state. runtime_status. name } ", flush = True ) wfr. shutdown () Enter fullscreen mode Exit fullscreen mode Save this as agent.py. The start mode schedules a new workflow instance and the resume mode only waits for an existing one, which is what the restart needs.

You can stop the process with Ctrl+C or kill, but you will press it half a second late, after the step has finished, and the run proves nothing. A crash you cannot place exactly is a crash you cannot repeat, and a test you cannot repeat never makes it into CI.

Instead, add a function that exits the process when the CRASH_AT environment variable names the current step. os._exit(1) skips cleanup and finally blocks, so it behaves like a power cut. Add this below the DATA_DIR line in agent.py:

CRASH_AT = os. getenv ( " CRASH_AT " ) def crash_point ( name ): if CRASH_AT == name: print ( f " crashing at { name } ", flush = True ) os. _exit ( 1 ) # no cleanup, no finally blocks Enter fullscreen mode Exit fullscreen mode Then place two crash points in post_comment. The first fires before the write, and the second fires after the write but before the activity returns:

@wfr.activity ( name = " post_comment " ) def post_comment ( ctx: wf. WorkflowActivityContext, summary: str ): crash_point ( " before_post " ) with open ( DATA_DIR / " ledger.txt ", " a " ) as f: f. write ( f " { ctx. workflow_id }: { summary } \n " ) crash_point ( " after_post " ) return " posted " Enter fullscreen mode Exit fullscreen mode Write the crash recovery test The test starts the agent under dapr run as a subprocess, once with the crash switch on and once with it off. The first run must print the crash message and must not complete. The second run must report COMPLETED, and the files must hold the expected counts. Save this as test_crash_recovery.py next to agent.py:

import os import subprocess import sys import uuid from pathlib import Path import pytest pytestmark = pytest. mark. integration AGENT = Path ( __file__ ). parent / " agent.py " def run_agent ( mode, instance_id, app_id, data_dir, crash_at = None ): env = { ** os. environ, " DATA_DIR ": str ( data_dir )} if crash_at: env [ " CRASH_AT " ] = crash_at return subprocess. run ( [ " dapr ", " run ", " --app-id ", app_id, " -- ", sys. executable, str ( AGENT ), mode, instance_id ], env = env, capture_output = True, text = True, timeout = 120, ) def count_lines ( path ): return len ( path. read_text (). splitlines ()) if path. exists () else 0 def test_crash_before_post ( tmp_path ): suffix = uuid. uuid4 (). hex [: 8 ] instance_id, app_id = f " ticket- { suffix } ", f " triage- { suffix } " crashed = run_agent ( " start ", instance_id, app_id, tmp_path, crash_at = " before_post " ) assert " crashing at before_post " in crashed. stdout assert " STATUS=COMPLETED " not in crashed. stdout resumed = run_agent ( " resume ", instance_id, app_id, tmp_path ) assert " STATUS=COMPLETED " in resumed. stdout, resumed. stdout [ - 2000:] assert count_lines ( tmp_path / " ledger.txt " ) == 1 assert count_lines ( tmp_path / " llm_calls.txt " ) == 1 Enter fullscreen mode Exit fullscreen mode Every test generates a new instance ID and app ID, so state left behind by earlier runs cannot make a test pass for the wrong reason. Both runs share the same app ID because the workflow state belongs to that app. Each test also writes to its own tmp_path, so the line counts start at zero.

Register the integration marker in pytest.ini, so these slower tests stay out of your fast unit run:

[pytest] markers = integration: runs the agent under dapr run and needs a local Dapr install Enter fullscreen mode Exit fullscreen mode Handle a crash after the side effect The two crash points show different things.

With before_post, summarize has already completed and been recorded. On restart it is not run again, so llm_calls.txt keeps one line (no second LLM call and no second token bill), and the ledger ends with one line.

With after_post, the comment was written but the activity never returned, so the runtime has no record that it finished. It runs post_comment again and the ledger gets two lines. That is correct behavior for an activity that crashed mid-flight, which is why a side-effecting activity needs a "did I already do this?" check keyed on ctx.workflow_id. Add one to agent.py:

IDEMPOTENT = os. getenv ( " IDEMPOTENT " ) == " 1 " def already_posted ( workflow_id ): ledger = DATA_DIR / " ledger.txt " return ledger. exists () and any ( line. startswith ( f " { workflow_id }: " ) for line in ledger. read_text (). splitlines () ) @wfr.activity ( name = " post_comment " ) def post_comment ( ctx: wf. WorkflowActivityContext, summary: str ): if IDEMPOTENT and already_posted ( ctx. workflow_id ): return " already posted " crash_point ( " before_post " ) with open ( DATA_DIR / " ledger.txt ", " a " ) as f: f. write ( f " { ctx. workflow_id }: { summary } \n " ) crash_point ( " after_post " ) return " posted " Enter fullscreen mode Exit fullscreen mode The IDEMPOTENT switch only exists so the test can show both outcomes. In your own agent the check is always on, and the key is usually the idempotency key of the system you write to, such as your ticketing tool.

Now parametrize the test over the three cases. Pass the new idempotent option through run_agent, and replace test_crash_before_post:

def run_agent ( mode, instance_id, app_id, data_dir, crash_at = None, idempotent = False ): env = { ** os. environ, " DATA_DIR ": str ( data_dir ), " IDEMPOTENT ": " 1 " if idempotent else " 0 " } if crash_at: env [ " CRASH_AT " ] = crash_at return subprocess. run ( [ " dapr ", " run ", " --app-id ", app_id, " -- ", sys. executable, str ( AGENT ), mode, instance_id ], env = env, capture_output = True, text = True, timeout = 120, ) @pytest.mark.parametrize ( " crash_at, idempotent, expected_comments ", [ ( " before_post ", False, 1 ), ( " after_post ", False, 2 ), # at-least-once: the comment is written twice ( " after_post ", True, 1 ), ], ) def test_crash_recovery ( tmp_path, crash_at, idempotent, expected_comments ): suffix = uuid. uuid4 (). hex [: 8 ] instance_id, app_id = f " ticket- { suffix } ", f " triage- { suffix } " crashed = run_agent ( " start ", instance_id, app_id, tmp_path, crash_at, idempotent ) assert f " crashing at { crash_at } " in crashed. stdout assert " STATUS=COMPLETED " not in crashed. stdout resumed = run_agent ( " resume ", instance_id, app_id, tmp_path, idempotent = idempotent ) assert " STATUS=COMPLETED " in resumed. stdout, resumed. stdout [ - 2000:] assert count_lines ( tmp_path / " ledger.txt " ) == expected_comments assert count_lines ( tmp_path / " llm_calls.txt " ) == 1 # summarize is never replayed Enter fullscreen mode Exit fullscreen mode The second case asserts two lines on purpose. It documents the at-least-once behavior, so a future change to the runtime or to your activity cannot alter it without a failing test.

pytest -m integration -v Enter fullscreen mode Exit fullscreen mode All three cases should pass in about 20 seconds:

test_crash_recovery.py::test_crash_recovery[before_post-False-1] PASSED test_crash_recovery.py::test_crash_recovery[after_post-False-2] PASSED test_crash_recovery.py::test_crash_recovery[after_post-True-1] PASSED Enter fullscreen mode Exit fullscreen mode To watch a single run in detail, start the Dapr Dev Dashboard in a second terminal with diagrid-dev-dashboard and open http://localhost:9090. Then crash and restart the agent by hand, using a fresh instance ID. In bash on macOS or Linux:

ID = ticket- $RANDOM CRASH_AT = before_post dapr run --app-id triage -- python agent.py start $ID dapr run --app-id triage -- python agent.py resume $ID wc -l ledger.txt llm_calls.txt Enter fullscreen mode Exit fullscreen mode In Windows PowerShell, set the environment variable for the first run and remove it before the restart:

$ID = "ticket- $( Get-Random ) " $ env: CRASH_AT = "before_post" dapr run --app-id triage -- python agent.py start $ID Remove-Item Env:CRASH_AT dapr run --app-id triage -- python agent.py resume $ID ( Get-Content ledger.txt | Measure-Object -Line ). Lines ( Get-Content llm_calls.txt | Measure-Object -Line ). Lines Enter fullscreen mode Exit fullscreen mode Select the instance in the dashboard to see its event history: which activity completed before the crash, and which one was scheduled again after the restart. To run the same check in CI, give the runner Docker, run dapr init, and add pytest -m integration next to your evals.

If a test passes suspiciously fast, check the instance ID and app ID first. Workflow state persists in your local state store, so resuming an instance that finished in an earlier run reports COMPLETED right away, even when nothing crashed and nothing was written. The unique IDs in the test prevent this. When you run the commands by hand, use a fresh $ID every time.

The dapr run output can show actor retry warnings while the restarted app registers with the placement service. In these runs the workflow still completed, so keep the 60 second wait timeout generous instead of treating those warnings as a failure.

Crash recovery is now a test that fails when recovery breaks. A named crash point, a restart with the switch off, and a ledger you can count tell you what your agent did after it died. Both crash points should end at one line: before_post because completed steps are never replayed, and after_post because your activity checks whether it already did the work. Add the pytest run to CI next to your evals, and open the Dev Dashboard whenever you want to see the event history behind a result. The durable agents guide explains checkpointing and replay in more detail, and Dapr University has courses on durable execution and Dapr.