Lab 1: kill the agent five ways and count the model calls

Every exercise runs one scenario with demo.py. It starts the model stub and the agent, invokes WeatherAgent/run, waits until the weather tool has been called, kill -9s the agent one second into its six-second pause, restarts it two seconds later, and then reports the invocation's status, the stub's call counts and the journal. (Exercise 1.5 kills it during a model call instead.) The stub's model call #n lines print live as they happen.

A clean run needs 2 model calls and 1 weather call. Recorded outputs for exercises 1.1–1.5 are in observed/ in the lab repository. Yours should show the same outcomes, with different timestamps, pids and attempt numbers. The call counts should match too, except in 1.3, where the count depends on timing; working out why is part of that exercise.

The attempt N, last failure … lines are demo.py reading the invocation's retry counter every half second. They tell you which failures happened, but a number can be skipped or repeat the previous failure, so reason from the model calls and the journal, not from the attempt numbers.

Objective

Why: A recovered agent can repeat an external model call when its decision was never recorded. Comparing the model stub's counters with the invocation journal makes that cost visible (recorded runs; Durable steps).

By the end: Run the same agent with stable, varying, and journaled model decisions; identify the missing journal entry behind the extra call and RT0016; explain why a call interrupted before its result is recorded can still run again (recorded mid-call run).

Start with 1.1 and use the call counts and journal to answer each set of questions before moving on. Exercises 1.3 and 1.6 take longer and can be done after the main comparison.

Replay map — read this before 1.1

The same handler starts again after the crash. Use this map to predict which work repeats; then check it against the recorded naive run and Restate’s durable-steps guide.

BEFORE THE CRASH         JOURNAL AT CRASH           AFTER RESTART
ask_model()          →   no model entry         →   ask_model() runs again
ctx.run_typed(tool)  →   Run + result recorded  →   recorded result reused
ctx.sleep()          →   Sleep started          →   waits remaining time
                         ✕ agent process killed

The question for each line is whether it produced a recorded Restate operation. An input value, a constant, or a pure calculation from them needs no separate entry. A model answer can change between attempts, so it must be recorded before it controls the next operation; otherwise replay may issue a different operation and get RT0016 (error reference).

Boundary to remember: A completed, recorded ctx.run_typed result is reused on replay. A call still in flight can happen again because its result is absent from the journal. Exercise 1.5 measures that limit (recorded mid-call run).

Run the lab

On scimigo.com: the Run it on your computer panel above this guide starts each exercise through agent-runtime and shows its output there, with nothing to copy. The options below work without it.

You need: Docker and Python 3.10 or newer. The Restate server, agent, and model stub run on your machine. No cloud account or model API key is used.

Option A · local browser workspace

The local page has a button for each exercise, live output, status checks, and questions. The page uses port 3002 so it can run alongside the Temporal labs on port 3000. Run each block in its own terminal.

Terminal 1 · Restate server

git clone https://github.com/SciMigo/restate-durable-agent-demo.git
cd restate-durable-agent-demo
docker compose up -d

Leave Restate running between scenarios. If you already started it for this lab, this command keeps the same container; you do not need another server. Its UI is at http://127.0.0.1:19070/ui/.

Terminal 2 · local lab page

cd restate-durable-agent-demo
python3 lab_server.py

Terminal 2 should be in the cloned repository. If it opens in a different directory, use its absolute path with cd. Open the http://127.0.0.1:3002/ URL printed by the server, click Prepare this lab, then run exercise 1.1. Prepare installs the demo's packages in .venv; no separate Python install command is needed. Follow the output until it prints the call counts and journal, then answer 1.1's questions below.

If Chrome redirects the local URL to /en, the lab page also serves that path. If port 3002 is occupied, run python3 lab_server.py --port 3003 and open the new printed URL. Use http, not https.

Option B · run commands in terminals

From restate-durable-agent-demo, install the packages once and leave Restate running:

python3 -m venv .venv
./.venv/bin/pip install -r requirements.txt
docker compose up -d

Then run the command under each exercise from the same directory. Use ./.venv/bin/python, which works even if your shell has no python alias. For example:

./.venv/bin/python demo.py --agent naive --model stable

Run one scenario at a time because the model stub and agent use ports 8765 and 9080.

After all exercises, docker compose down resets the Restate server.

1.1 The extra model call

python demo.py --agent naive --model stable
  1. Find model call #2. Which decision did it return, and why was it made at all?
  2. The weather counter stays at 1. Which journal entry answered for get_weather on the replay?
  3. Match each of the six journal entries to a line of run() in agent.py. Which line has no entry?

1.2 When the second answer is different

python demo.py --agent naive --model varying
  1. Find the first last failure RT0016 line. What did the model call just before it answer?
  2. Which model call got the replay past the mismatch, and what did it answer?
  3. How many model calls did the run cost? How many of them were spent on attempts that failed? (Hint: an attempt can fail with RT0016 only if its first model call answers directly.)

1.3 When it stays different

python demo.py --agent naive --model drifted --wait 150

The invocation pauses after about 40 seconds; --wait 150 is only the longest demo.py will watch.

  1. Watch the attempts climb. What is the final status, and how many model calls did it take to get there?
  2. Read the last failure RT0016 message. Which journal index does it name, and what is at that index in the journal printed below it?
  3. Which setting in agent.py decided when the retries stopped? What would a longer policy have cost?
  4. Open the Restate UI at http://localhost:19070/ui/ and find the paused invocation.
  5. Run it again with --restart-delay 9, so the agent stays down for nine seconds instead of two. The model and the code are unchanged. Why did the count change, and would the count in 1.1 or 1.2 change the same way? (Recorded: observed/7-naive-drifted-restart-delay.txt.)

1.4 The fix

python demo.py --agent journaled --model varying
  1. Confirm model calls: 2 even though this stub changes its answers from call to call.
  2. Find Run call model in the journal. Which entry returned the first decision on the replay?
  3. In one sentence: why does the recorded answer make the replay's decision the same?

1.5 Kill it mid-call

python demo.py --agent journaled --model stable --kill-during model

The same journaled agent as in 1.4. This time the stub decides at once but holds its first answer back for four seconds, and demo.py kills the agent one second into that call. Predict first: how many model calls will the run make?

  1. Count the model calls. Which one did its work but never reached the journal?
  2. Compare this run's journal with the one from 1.4. Could you tell from the journal alone that a call was lost?
  3. ctx.run_typed stops repeating a call once Restate has recorded its result. What more would you need before retrying a call whose effect must not happen twice, such as charging a card?

1.6 (optional, advanced) Repairing an already-stuck invocation

Not needed to understand ctx.run: this one is about operating a service whose invocations are already stuck.

python recover.py recreates the paused invocation from 1.3, deploys the journaled agent on :9081 as a second deployment, and resumes the invocation on it. Then it kills the invocation and restarts it as new. Predict first: does the fixed code's first step match index 1 of the recorded journal?

  1. What status did the resume call return, and which deployment is the invocation pinned to now? Did the resumed replay call the model?
  2. Read the second RT0016 message. Which field differs, and what are its two values?
  3. What does restart-as-new answer while the invocation is paused, and what does kill record for the original?
  4. Count the restarted invocation's model calls and compare its journal with exercise 1.4's.
  5. The resume failed because the fix is not journal-compatible: it adds a step in front of index 1. Describe a fix that a resume could have rescued.

The recorded run is labs/observed/5-recover.txt.

What you learned

Questions you should be able to answer

  1. Which operation is absent from the naive journal, and how does that change the model call count after a crash?
  2. In the varying run, the journal's index 1 is the recorded weather call. What did the replay try to do at that index instead, and why did Restate stop it?
  3. Which journal entry lets the fixed run reuse its first model decision?
  4. Why did the mid-call run still use three model calls? What would a payment API need to make a retry safe (Sagas guide)?

Inspect the implementation and recorded evidence

7 recorded runs · 9 source files

Every file in the lab repository, as committed. The recorded runs are the outputs the exercises quote; run the same commands and compare.

Recorded runs

observed/1-naive-stable.txt
== agent: naive   model answers: stable
16:07:07  model stub on :8765, MODEL_ANSWERS=stable
invoked WeatherAgent/run   inv_11UJzoEzDOEo6F9iKqwIUpkO9lblvyitjr
16:07:08  model call #1  ->  {"tool": "get_weather", "city": "Berlin"}
16:07:08  weather call #1  get_weather(Berlin)
kill -9 agent (pid 1810632)
agent restarted (pid 1810834); Restate retries and replays the journal
  attempt 3, last failure RT0010 (service unreachable)
16:07:12  model call #2  ->  {"tool": "get_weather", "city": "Berlin"}
  attempt 4, last failure RT0010 (service unreachable)
16:07:14  model call #3  ->  {"answer": "Berlin right now: 18\u00b0C and cloudy."}

status: completed   model calls: 3   weather calls: 1
result: Berlin right now: 18°C and cloudy.
journal:
   0  Command: Input
   1  Command: Run  get_weather
   2  Notification: Run
   3  Command: Sleep  rate-limit pause
   4  Notification: Sleep
   5  Command: Output
UI: http://127.0.0.1:19070/ui/   invocation inv_11UJzoEzDOEo6F9iKqwIUpkO9lblvyitjr
observed/2-naive-varying.txt
== agent: naive   model answers: varying
16:04:17  model stub on :8765, MODEL_ANSWERS=varying
invoked WeatherAgent/run   inv_1cF1B9OB7hVa7cybq2osKc8mSDosPTYRIx
16:04:17  model call #1  ->  {"tool": "get_weather", "city": "Berlin"}
16:04:17  weather call #1  get_weather(Berlin)
kill -9 agent (pid 1799161)
agent restarted (pid 1799192); Restate retries and replays the journal
  attempt 3, last failure RT0010 (service unreachable)
16:04:21  model call #2  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 4, last failure RT0016 (journal mismatch)
16:04:23  model call #3  ->  {"tool": "get_weather", "city": "Berlin"}
  attempt 5, last failure RT0016 (journal mismatch)
16:04:23  model call #4  ->  {"answer": "Berlin right now: 18\u00b0C and cloudy."}

status: completed   model calls: 4   weather calls: 1
result: Berlin right now: 18°C and cloudy.
journal:
   0  Command: Input
   1  Command: Run  get_weather
   2  Notification: Run
   3  Command: Sleep  rate-limit pause
   4  Notification: Sleep
   5  Command: Output
UI: http://127.0.0.1:19070/ui/   invocation inv_1cF1B9OB7hVa7cybq2osKc8mSDosPTYRIx
observed/3-naive-drifted.txt
== agent: naive   model answers: drifted
16:06:18  model stub on :8765, MODEL_ANSWERS=drifted
invoked WeatherAgent/run   inv_16wT0XazNWHk3QxIhqQyiO3aqI2PJP59TM
16:06:18  model call #1  ->  {"tool": "get_weather", "city": "Berlin"}
16:06:18  weather call #1  get_weather(Berlin)
kill -9 agent (pid 1807391)
agent restarted (pid 1807463); Restate retries and replays the journal
  attempt 3, last failure RT0010 (service unreachable)
16:06:22  model call #2  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 4, last failure RT0016 (journal mismatch)
16:06:24  model call #3  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 5, last failure RT0016 (journal mismatch)
16:06:27  model call #4  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 6, last failure RT0016 (journal mismatch)
16:06:29  model call #5  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 7, last failure RT0016 (journal mismatch)
16:06:32  model call #6  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 8, last failure RT0016 (journal mismatch)
16:06:34  model call #7  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 9, last failure RT0016 (journal mismatch)
16:06:36  model call #8  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 10, last failure RT0016 (journal mismatch)
16:06:38  model call #9  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 11, last failure RT0016 (journal mismatch)
16:06:41  model call #10  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 12, last failure RT0016 (journal mismatch)
16:06:43  model call #11  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 13, last failure RT0016 (journal mismatch)
16:06:45  model call #12  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 14, last failure RT0016 (journal mismatch)
16:06:47  model call #13  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 15, last failure RT0016 (journal mismatch)
16:06:49  model call #14  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 16, last failure RT0016 (journal mismatch)
16:06:52  model call #15  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 17, last failure RT0016 (journal mismatch)
16:06:55  model call #16  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 18, last failure RT0016 (journal mismatch)
16:06:57  model call #17  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 19, last failure RT0016 (journal mismatch)
16:06:59  model call #18  ->  {"answer": "Berlin is probably mild this time of year."}

status: paused   model calls: 18   weather calls: 1
last failure RT0016:
  [570 Journal mismatch] Found a mismatch between the code paths taken during the previous execution and the paths taken during this execution.
  This typically happens when some parts of the code are non-deterministic.
   - The previous execution ran and recorded the following: 'handler return' (index '1')
   - The current execution attempts to perform the following: 'run'
journal:
   0  Command: Input
   1  Command: Run  get_weather
   2  Notification: Run
   3  Command: Sleep  rate-limit pause
   4  Notification: Sleep
UI: http://127.0.0.1:19070/ui/   invocation inv_16wT0XazNWHk3QxIhqQyiO3aqI2PJP59TM
observed/4-journaled-varying.txt
== agent: journaled   model answers: varying
16:04:41  model stub on :8765, MODEL_ANSWERS=varying
invoked WeatherAgent/run   inv_1exMup8JOBwU2zuoD2n0EVfstWXcCnyQbX
16:04:41  model call #1  ->  {"tool": "get_weather", "city": "Berlin"}
16:04:41  weather call #1  get_weather(Berlin)
kill -9 agent (pid 1800589)
agent restarted (pid 1801004); Restate retries and replays the journal
  attempt 3, last failure RT0010 (service unreachable)
  attempt 4, last failure RT0010 (service unreachable)
16:04:47  model call #2  ->  {"answer": "Berlin right now: 18\u00b0C and cloudy."}

status: completed   model calls: 2   weather calls: 1
result: Berlin right now: 18°C and cloudy.
journal:
   0  Command: Input
   1  Command: Run  call model
   2  Notification: Run
   3  Command: Run  get_weather
   4  Notification: Run
   5  Command: Sleep  rate-limit pause
   6  Notification: Sleep
   7  Command: Run  call model
   8  Notification: Run
   9  Command: Output
UI: http://127.0.0.1:19070/ui/   invocation inv_1exMup8JOBwU2zuoD2n0EVfstWXcCnyQbX
observed/5-recover.txt
== 1. a stuck invocation: naive agent, drifted model
16:43:43  model stub on :8765, MODEL_ANSWERS=drifted
invoked inv_1dFn1jZ5l9RI4YcvTavbL2Q9dkj0yZ3uGn on deployment dp_16R728fEzlWUoo1Y3SNvxi9
16:43:43  model call #1  ->  {"tool": "get_weather", "city": "Berlin"}
16:43:43  weather call #1  get_weather(Berlin)
kill -9 and restart; waiting for the retries to run out
16:43:47  model call #2  ->  {"answer": "Berlin is probably mild this time of year."}
16:43:49  model call #3  ->  {"answer": "Berlin is probably mild this time of year."}
16:43:52  model call #4  ->  {"answer": "Berlin is probably mild this time of year."}
16:43:54  model call #5  ->  {"answer": "Berlin is probably mild this time of year."}
16:43:56  model call #6  ->  {"answer": "Berlin is probably mild this time of year."}
16:43:58  model call #7  ->  {"answer": "Berlin is probably mild this time of year."}
16:44:00  model call #8  ->  {"answer": "Berlin is probably mild this time of year."}
16:44:02  model call #9  ->  {"answer": "Berlin is probably mild this time of year."}
16:44:05  model call #10  ->  {"answer": "Berlin is probably mild this time of year."}
16:44:07  model call #11  ->  {"answer": "Berlin is probably mild this time of year."}
16:44:09  model call #12  ->  {"answer": "Berlin is probably mild this time of year."}
16:44:11  model call #13  ->  {"answer": "Berlin is probably mild this time of year."}
16:44:14  model call #14  ->  {"answer": "Berlin is probably mild this time of year."}
16:44:16  model call #15  ->  {"answer": "Berlin is probably mild this time of year."}
16:44:19  model call #16  ->  {"answer": "Berlin is probably mild this time of year."}
16:44:21  model call #17  ->  {"answer": "Berlin is probably mild this time of year."}
16:44:23  model call #18  ->  {"answer": "Berlin is probably mild this time of year."}
status: paused   model calls so far: 18

== 2. deploy the fix as a new deployment and resume the paused invocation on it
PATCH /invocations/inv_1dFn1jZ5l9RI4YcvTavbL2Q9dkj0yZ3uGn/resume?deployment=dp_11aIP4h4jRqNAQfhLIZ9uGl  ->  200
pinned deployment: dp_11aIP4h4jRqNAQfhLIZ9uGl   last failure RT0016:
  [570 Journal mismatch] Found a mismatch between the code paths taken during the previous execution and the paths taken during this execution.
  This typically happens when some parts of the code are non-deterministic.
  - The mismatch happened while executing 'run' (index '1')
  - Difference:
     name: get_weather != call model
model calls since resume: 0
PATCH pause  ->  202; status: paused

== 3. restart it as new on the fixed deployment
PATCH restart-as-new while paused  ->  409 {'message': "The invocation 'inv_1dFn1jZ5l9RI4YcvTavbL2Q9dkj0yZ3uGn' is still running.", 'restate_code': None}
PATCH kill  ->  200; original: failure [409] killed
PATCH restart-as-new after kill  ->  200   new invocation inv_1dFn1jZ5l9RI66c4imrZWDwTEZhvXwLLfX
16:44:24  model call #1  ->  {"tool": "get_weather", "city": "Berlin"}
16:44:24  weather call #1  get_weather(Berlin)
16:44:30  model call #2  ->  {"answer": "Berlin right now: 18\u00b0C and cloudy."}
status: completed   pinned deployment: dp_11aIP4h4jRqNAQfhLIZ9uGl   model calls: 2
result: Berlin right now: 18°C and cloudy.
journal of the new invocation:
   0  Command: Input
   1  Command: Run  call model
   2  Notification: Run
   3  Command: Run  get_weather
   4  Notification: Run
   5  Command: Sleep  rate-limit pause
   6  Notification: Sleep
   7  Command: Run  call model
   8  Notification: Run
   9  Command: Output
observed/6-journaled-killed-mid-call.txt
== agent: journaled   model answers: stable
17:55:34  model stub on :8765, MODEL_ANSWERS=stable
invoked WeatherAgent/run   inv_1hxoDh7lblmW3m6su9nGNiXRYC1BLPRGnO
17:55:34  model call #1  ->  {"tool": "get_weather", "city": "Berlin"}
kill -9 agent (pid 2294216), before model call #1 returned its answer
agent restarted (pid 2294655); Restate retries and replays the journal
  attempt 3, last failure RT0010 (service unreachable)
17:55:38  model call #2  ->  {"tool": "get_weather", "city": "Berlin"}
17:55:38  weather call #1  get_weather(Berlin)
17:55:38  model call #1  answer not delivered: the caller is gone
17:55:44  model call #3  ->  {"answer": "Berlin right now: 18\u00b0C and cloudy."}

status: completed   model calls: 3   weather calls: 1
result: Berlin right now: 18°C and cloudy.
journal:
   0  Command: Input
   1  Command: Run  call model
   2  Notification: Run
   3  Command: Run  get_weather
   4  Notification: Run
   5  Command: Sleep  rate-limit pause
   6  Notification: Sleep
   7  Command: Run  call model
   8  Notification: Run
   9  Command: Output
UI: http://127.0.0.1:19070/ui/   invocation inv_1hxoDh7lblmW3m6su9nGNiXRYC1BLPRGnO
observed/7-naive-drifted-restart-delay.txt
$ python demo.py --agent naive --model drifted --wait 180 --restart-delay 0.2
== agent: naive   model answers: drifted
17:55:45  model stub on :8765, MODEL_ANSWERS=drifted
invoked WeatherAgent/run   inv_12cIQ4HWqxjR0Mz4106l5EswoH9EWrAJ9p
17:55:45  model call #1  ->  {"tool": "get_weather", "city": "Berlin"}
17:55:45  weather call #1  get_weather(Berlin)
kill -9 agent (pid 2295195)
agent restarted (pid 2295207); Restate retries and replays the journal
  attempt 1, last failure RT0010 (service unreachable)
17:55:47  model call #2  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 2, last failure RT0016 (journal mismatch)
  attempt 3, last failure RT0016 (journal mismatch)
17:55:48  model call #3  ->  {"answer": "Berlin is probably mild this time of year."}
17:55:49  model call #4  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 4, last failure RT0016 (journal mismatch)
17:55:51  model call #5  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 5, last failure RT0016 (journal mismatch)
17:55:53  model call #6  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 6, last failure RT0016 (journal mismatch)
17:55:55  model call #7  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 7, last failure RT0016 (journal mismatch)
17:55:57  model call #8  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 8, last failure RT0016 (journal mismatch)
17:56:00  model call #9  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 9, last failure RT0016 (journal mismatch)
17:56:02  model call #10  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 10, last failure RT0016 (journal mismatch)
17:56:04  model call #11  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 11, last failure RT0016 (journal mismatch)
17:56:07  model call #12  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 12, last failure RT0016 (journal mismatch)
17:56:09  model call #13  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 13, last failure RT0016 (journal mismatch)
17:56:11  model call #14  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 14, last failure RT0016 (journal mismatch)
17:56:14  model call #15  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 15, last failure RT0016 (journal mismatch)
17:56:16  model call #16  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 16, last failure RT0016 (journal mismatch)
17:56:19  model call #17  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 17, last failure RT0016 (journal mismatch)
17:56:21  model call #18  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 18, last failure RT0016 (journal mismatch)
17:56:23  model call #19  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 19, last failure RT0016 (journal mismatch)
17:56:25  model call #20  ->  {"answer": "Berlin is probably mild this time of year."}

status: paused   model calls: 20   weather calls: 1
last failure RT0016:
  Found a mismatch between the code paths taken during the previous execution and the paths taken during this execution.
  This typically happens when some parts of the code are non-deterministic.
   - The previous execution ran and recorded the following: 'handler return' (index '1')
   - The current execution attempts to perform the following: 'run'
journal:
   0  Command: Input
   1  Command: Run  get_weather
   2  Notification: Run
   3  Command: Sleep  rate-limit pause
   4  Notification: Sleep
UI: http://127.0.0.1:19070/ui/   invocation inv_12cIQ4HWqxjR0Mz4106l5EswoH9EWrAJ9p

$ python demo.py --agent naive --model drifted --wait 180 --restart-delay 2
== agent: naive   model answers: drifted
17:56:26  model stub on :8765, MODEL_ANSWERS=drifted
invoked WeatherAgent/run   inv_16df5t0MFTfz0CCVaMpDTifHrjfWt4NBFP
17:56:27  model call #1  ->  {"tool": "get_weather", "city": "Berlin"}
17:56:27  weather call #1  get_weather(Berlin)
kill -9 agent (pid 2298121)
agent restarted (pid 2298160); Restate retries and replays the journal
  attempt 3, last failure RT0010 (service unreachable)
17:56:30  model call #2  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 4, last failure RT0016 (journal mismatch)
17:56:32  model call #3  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 5, last failure RT0016 (journal mismatch)
17:56:35  model call #4  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 6, last failure RT0016 (journal mismatch)
17:56:37  model call #5  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 7, last failure RT0016 (journal mismatch)
17:56:39  model call #6  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 8, last failure RT0016 (journal mismatch)
17:56:41  model call #7  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 9, last failure RT0016 (journal mismatch)
17:56:44  model call #8  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 10, last failure RT0016 (journal mismatch)
17:56:46  model call #9  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 11, last failure RT0016 (journal mismatch)
17:56:48  model call #10  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 12, last failure RT0016 (journal mismatch)
17:56:51  model call #11  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 13, last failure RT0016 (journal mismatch)
17:56:53  model call #12  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 14, last failure RT0016 (journal mismatch)
17:56:55  model call #13  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 15, last failure RT0016 (journal mismatch)
17:56:57  model call #14  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 16, last failure RT0016 (journal mismatch)
17:57:00  model call #15  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 17, last failure RT0016 (journal mismatch)
17:57:02  model call #16  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 18, last failure RT0016 (journal mismatch)
17:57:04  model call #17  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 19, last failure RT0016 (journal mismatch)
17:57:06  model call #18  ->  {"answer": "Berlin is probably mild this time of year."}

status: paused   model calls: 18   weather calls: 1
last failure RT0016:
  Found a mismatch between the code paths taken during the previous execution and the paths taken during this execution.
  This typically happens when some parts of the code are non-deterministic.
   - The previous execution ran and recorded the following: 'handler return' (index '1')
   - The current execution attempts to perform the following: 'run'
journal:
   0  Command: Input
   1  Command: Run  get_weather
   2  Notification: Run
   3  Command: Sleep  rate-limit pause
   4  Notification: Sleep
UI: http://127.0.0.1:19070/ui/   invocation inv_16df5t0MFTfz0CCVaMpDTifHrjfWt4NBFP

$ python demo.py --agent naive --model drifted --wait 180 --restart-delay 5
== agent: naive   model answers: drifted
17:57:07  model stub on :8765, MODEL_ANSWERS=drifted
invoked WeatherAgent/run   inv_13Vk6Xhyl7tY1hlCKm410BBRHGCfmDvFZT
17:57:07  model call #1  ->  {"tool": "get_weather", "city": "Berlin"}
17:57:07  weather call #1  get_weather(Berlin)
kill -9 agent (pid 2300753)
agent restarted (pid 2301278); Restate retries and replays the journal
  attempt 5, last failure RT0010 (service unreachable)
17:57:16  model call #2  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 6, last failure RT0016 (journal mismatch)
17:57:18  model call #3  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 7, last failure RT0016 (journal mismatch)
17:57:20  model call #4  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 8, last failure RT0016 (journal mismatch)
17:57:22  model call #5  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 9, last failure RT0016 (journal mismatch)
17:57:25  model call #6  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 10, last failure RT0016 (journal mismatch)
17:57:27  model call #7  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 11, last failure RT0016 (journal mismatch)
17:57:29  model call #8  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 12, last failure RT0016 (journal mismatch)
17:57:32  model call #9  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 13, last failure RT0016 (journal mismatch)
17:57:34  model call #10  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 14, last failure RT0016 (journal mismatch)
17:57:36  model call #11  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 15, last failure RT0016 (journal mismatch)
17:57:39  model call #12  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 16, last failure RT0016 (journal mismatch)
17:57:41  model call #13  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 17, last failure RT0016 (journal mismatch)
17:57:43  model call #14  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 18, last failure RT0016 (journal mismatch)
17:57:45  model call #15  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 19, last failure RT0016 (journal mismatch)
17:57:48  model call #16  ->  {"answer": "Berlin is probably mild this time of year."}

status: paused   model calls: 16   weather calls: 1
last failure RT0016:
  Found a mismatch between the code paths taken during the previous execution and the paths taken during this execution.
  This typically happens when some parts of the code are non-deterministic.
   - The previous execution ran and recorded the following: 'handler return' (index '1')
   - The current execution attempts to perform the following: 'run'
journal:
   0  Command: Input
   1  Command: Run  get_weather
   2  Notification: Run
   3  Command: Sleep  rate-limit pause
   4  Notification: Sleep
UI: http://127.0.0.1:19070/ui/   invocation inv_13Vk6Xhyl7tY1hlCKm410BBRHGCfmDvFZT

$ python demo.py --agent naive --model drifted --wait 180 --restart-delay 9
== agent: naive   model answers: drifted
17:57:48  model stub on :8765, MODEL_ANSWERS=drifted
invoked WeatherAgent/run   inv_18yBtVQL1XCC0RXCWFfhL37cKaevGnB3D8
17:57:49  model call #1  ->  {"tool": "get_weather", "city": "Berlin"}
17:57:49  weather call #1  get_weather(Berlin)
kill -9 agent (pid 2303872)
agent restarted (pid 2304389); Restate retries and replays the journal
  attempt 6, last failure RT0010 (service unreachable)
17:58:00  model call #2  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 7, last failure RT0016 (journal mismatch)
17:58:02  model call #3  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 8, last failure RT0016 (journal mismatch)
17:58:04  model call #4  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 9, last failure RT0016 (journal mismatch)
17:58:06  model call #5  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 10, last failure RT0016 (journal mismatch)
17:58:08  model call #6  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 11, last failure RT0016 (journal mismatch)
17:58:11  model call #7  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 12, last failure RT0016 (journal mismatch)
17:58:13  model call #8  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 13, last failure RT0016 (journal mismatch)
17:58:15  model call #9  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 14, last failure RT0016 (journal mismatch)
17:58:17  model call #10  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 15, last failure RT0016 (journal mismatch)
17:58:20  model call #11  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 16, last failure RT0016 (journal mismatch)
17:58:22  model call #12  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 17, last failure RT0016 (journal mismatch)
17:58:25  model call #13  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 18, last failure RT0016 (journal mismatch)
17:58:27  model call #14  ->  {"answer": "Berlin is probably mild this time of year."}
  attempt 19, last failure RT0016 (journal mismatch)
17:58:30  model call #15  ->  {"answer": "Berlin is probably mild this time of year."}

status: paused   model calls: 15   weather calls: 1
last failure RT0016:
  Found a mismatch between the code paths taken during the previous execution and the paths taken during this execution.
  This typically happens when some parts of the code are non-deterministic.
   - The previous execution ran and recorded the following: 'handler return' (index '1')
   - The current execution attempts to perform the following: 'run'
journal:
   0  Command: Input
   1  Command: Run  get_weather
   2  Notification: Run
   3  Command: Sleep  rate-limit pause
   4  Notification: Sleep
UI: http://127.0.0.1:19070/ui/   invocation inv_18yBtVQL1XCC0RXCWFfhL37cKaevGnB3D8

Implementation

README.md
# Your agent paid for that model call twice

A runnable demo of durable execution on [Restate](https://restate.dev) for AI agents. A small weather agent asks a model, calls a tool, and pauses. We kill the process mid-run with `kill -9`. When it comes back, Restate replays the invocation's journal. What happens next depends on one line: whether the model call goes through `ctx.run`.

This is an independent teaching sample by SciMigo. It is not affiliated with or endorsed by Restate.

## What's here

| File | Role |
|---|---|
| `agent.py` | The Restate service `WeatherAgent/run`. `AGENT_MODE=naive` calls the model with a plain HTTP request; `AGENT_MODE=journaled` wraps the call in `ctx.run_typed`. Nothing else differs. |
| `model_stub.py` | A stand-in for the paid model API and the weather API. It counts every call, and it runs in its own process so the counts survive the agent being killed. |
| `demo.py` | Runs one scenario end to end: start, invoke, kill, restart, then report the outcome, the call counts and the journal. `--kill-during` and `--restart-delay` choose when the kill lands and how long the agent stays down. |
| `recover.py` | Takes the stuck invocation from scenario 3 and tries each way out: resume on a fixed deployment, restart-as-new, kill. |
| `docker-compose.yml` | Restate server, pinned to 1.7.10. |
| `observed/` | Output of each scenario, exactly as recorded. |

## Setup

Requires Docker and Python 3.10+.

```bash
python3 -m venv .venv && .venv/bin/pip install -r requirements.txt
docker compose up -d        # Restate: ingress on :18080, admin API and UI on :19070
```

The ports are moved off Restate's defaults (8080, 9070) to avoid clashing with anything already running. Point `demo.py` elsewhere with `RESTATE_INGRESS` and `RESTATE_ADMIN`. The Restate UI is at <http://localhost:19070/ui/>.

Each run ends with two links: the Restate UI, and the run's own invocation page in it
(`http://127.0.0.1:19070/ui/invocations/<id>`); `recover.py` also links the original invocation.
Output is colored in a terminal, or when `FORCE_COLOR` is set (agent-runtime sets it for lab actions,
and the scimigo.com lab page shows the colors); `NO_COLOR` turns colors off. Piped output stays plain text,
as recorded in `observed/`, which predates the two link lines: its runs end with a single
`UI: http://127.0.0.1:19070/ui/   invocation <id>` line.

The server keeps no volume, so `docker compose down` resets it.

## Browser lab (recommended)

The [SciMigo lab page](https://scimigo.com/en/learn/restate-durable-agents/01-the-model-call-you-pay-for-twice/lab) gives the full exercise. To run its scenarios with buttons on your own machine:

1. In terminal 1, from this repository, run `docker compose up -d`. Leave Restate running between scenarios. Its UI is at <http://127.0.0.1:19070/ui/>.
2. In terminal 2, from this repository, run `python3 lab_server.py`. Open the exact `http://127.0.0.1:3002/` address it prints. The local page uses Python's standard library, so there is nothing to install for the page itself.
3. Click **Prepare this lab**, then run exercises 1.1–1.5 in order. Prepare creates `.venv` and installs this repo's packages. Each button runs `demo.py` in the background, so watch **Live output** for its status, call counts, and journal. Run one scenario at a time. Exercise 1.6 is optional.

The page uses port 3002 so it can coexist with the Temporal course lab page on 3000. If Chrome rewrites the local address to `/en`, use the address printed in terminal 2 or open it in a fresh browser profile. The page also accepts `/en` to recover from that cached redirect. To choose another port, run `python3 lab_server.py --port 3003` and use the printed URL. Keep Restate running between scenarios; `docker compose up -d` on an already running server does not start another copy. Stop the page with Ctrl-C and, when finished with the lab, run `docker compose down` to reset Restate.

**Terminal route:** The setup commands above also support the `demo.py` commands in the scenario table. Run them with `.venv/bin/python` if your shell has no `python` command.

## The scenarios

In every run the agent first decides to call the tool, calls it, and starts a 6-second durable pause. By default `demo.py` kills the agent one second into that pause and restarts it two seconds later.

Run one scenario at a time. If you stop one early (Ctrl-C, or **Stop** on the lab page), `demo.py` (and `recover.py`) kills its unfinished invocation on the way out: left alone, its retries would reach the next scenario's agent and bill that run's stub.

The `attempt N, last failure …` lines are samples of the invocation's retry counter, read every half second while it runs. They show which failures happened, but a number can be skipped or repeat the previous failure, so count model calls, not attempt numbers.

| Command | What happens | Model calls (a clean run needs 2) |
|---|---|---|
| `python demo.py --agent naive --model stable` | The replay re-runs the unjournaled model call. The model gives the same decision, so replay continues and the run completes. The weather tool is **not** called again, because its result is in the journal. | **3** |
| `python demo.py --agent naive --model varying` | The re-run model call returns a *different* decision (answer directly). That contradicts the journal, which says the agent called the tool: **RT0016 journal mismatch**. Restate retries; this stub alternates, so the next retry happens to agree again, and the run completes. | **4** |
| `python demo.py --agent naive --model drifted --wait 150` | The model changed its mind for good. Every retry re-calls the model and fails with RT0016, until the handler's retry policy (20 attempts) pauses the invocation. | **18** in our run, then paused; the count depends on timing ([below](#how-many-calls-scenario-3-costs)) |
| `python demo.py --agent journaled --model varying` | The first model decision is recorded as `Run call model`. The replay returns the recorded decision without calling the model, and the run completes. | **2** |
| `python demo.py --agent journaled --model stable --kill-during model` | The kill lands while the first model call is in flight: the model has answered, but the answer never reached the journal. The replay calls the model again. | **3** |

Journal of the naive run: there is no entry for the model call, so nothing stops it from running again.

```
 0  Command: Input
 1  Command: Run  get_weather
 2  Notification: Run
 3  Command: Sleep  rate-limit pause
 4  Notification: Sleep
 5  Command: Output
```

Journal of the journaled run:

```
 0  Command: Input
 1  Command: Run  call model
 2  Notification: Run
 3  Command: Run  get_weather
 4  Notification: Run
 5  Command: Sleep  rate-limit pause
 6  Notification: Sleep
 7  Command: Run  call model
 8  Notification: Run
 9  Command: Output
```

The rule the demo teaches: **a replay has to issue the same Restate operations, in the same order, with the same inputs.** Anything that can come out differently on another attempt (a model answer, an HTTP response, the clock, a random number) has to be recorded before the handler branches on it: wrap model calls and network calls in `ctx.run`, and use the context's deterministic helpers (`ctx.random()`, `ctx.uuid()`, `ctx.time()`). Values that cannot change between attempts need no recording: the input, which is journal entry 0, constants, and anything computed from them. Code counts as constant only while its deployment does not change, which is why the `force: true` re-registration below is a demo shortcut. See [Durable steps](https://docs.restate.dev/develop/python/durable-steps) and [RT0016](https://docs.restate.dev/references/errors).

## What `ctx.run` does not promise

Scenario 5 kills the journaled agent one second into its first model call. The stub has decided, but holds the answer back for four seconds. Recorded in `observed/6-journaled-killed-mid-call.txt`:

```
17:55:34  model call #1  ->  {"tool": "get_weather", "city": "Berlin"}
kill -9 agent (pid 2294216), before model call #1 returned its answer
agent restarted (pid 2294655); Restate retries and replays the journal
  attempt 3, last failure RT0010 (service unreachable)
17:55:38  model call #2  ->  {"tool": "get_weather", "city": "Berlin"}
17:55:38  weather call #1  get_weather(Berlin)
17:55:38  model call #1  answer not delivered: the caller is gone
17:55:44  model call #3  ->  {"answer": "Berlin right now: 18\u00b0C and cloudy."}

status: completed   model calls: 3   weather calls: 1
```

The journal is the same ten entries as scenario 4's; nothing in it shows that a call was paid for and lost. A step counts as done once its result is in Restate's log ([Architecture](https://docs.restate.dev/references/architecture)); an attempt that dies before then runs the step again. So `ctx.run` never re-executes a step once Restate has recorded its result, but the external call may already have happened before that point: here the kill came while the answer was still on its way, and a crash after the answer arrives but before Restate records it runs the call again just the same. For an effect that must not happen twice, such as a payment, send an idempotency key the API honours ([Sagas](https://docs.restate.dev/guides/sagas)).

## How many calls scenario 3 costs

The drifted run paused after 18 model calls. That is this machine's number, not the demo's. The retry policy allows 20 attempts, and every attempt Restate makes while the agent is down fails with RT0010 without calling the model. How many fall in that window depends on how long the agent takes to come back. Same scenario, changing only `--restart-delay` (recorded in `observed/7-naive-drifted-restart-delay.txt`):

| `--restart-delay` | 0.2 s | 2 s (default) | 5 s | 9 s |
|---|---|---|---|---|
| Model calls before the pause | 20 | 18 | 16 | 15 |

The other scenarios' counts don't depend on timing: their extra calls happen on the first replay that reaches the agent, however many attempts it took to get there.

## Measured on

- `docker.restate.dev/restatedev/restate:1.7.10`
- `restate-sdk` 1.0.5 and `hypercorn` 0.18.0 on Python 3.12.3
- Linux, 2026-09-16

Timings and attempt numbers will vary. The outcomes, and the call counts of scenarios 1, 2, 4 and 5, reproduced on every run; scenario 3's count depends on timing (see above).

## Recovering a stuck invocation

`python recover.py` recreates scenario 3's paused invocation, then deploys the fixed (journaled) agent as a second deployment on :9081. Recorded in `observed/5-recover.txt`:

| Step | Result |
|---|---|
| `PATCH /invocations/{id}/resume?deployment=<fixed deployment>` | Accepted (200), and the invocation is re-pinned to the fixed deployment. It still fails with RT0016, because the fixed code's first step is `Run call model` while the recorded journal has `Run get_weather` at index 1. No model calls are made. |
| `PATCH /invocations/{id}/restart-as-new` while paused | Refused: 409, "The invocation … is still running." |
| `PATCH /invocations/{id}/kill`, then `restart-as-new` | The original completes as a failure (`[409] killed`). The new invocation starts from the original input on the latest deployment and completes with 2 model calls. |

Resuming without changing deployments behaves like the stuck retries: every attempt calls the model again and fails with RT0016.

The error for the resumed invocation is a same-type mismatch, and it names the difference directly:

```
[570 Journal mismatch] Found a mismatch between the code paths taken during the previous execution and the paths taken during this execution.
This typically happens when some parts of the code are non-deterministic.
- The mismatch happened while executing 'run' (index '1')
- Difference:
   name: get_weather != call model
```

**What this means:** a resume can only rescue an invocation when the new code is *journal-compatible*, meaning its replay issues the operations already recorded, in the same order. [Versioning](https://docs.restate.dev/services/versioning) lists a bug fixed inside a `ctx.run` as safe, and adding, removing or reordering operations as unsafe. This fix adds `Run call model` in front of the recorded `Run get_weather`, so however correct it is, it cannot replay this journal. The way out here is to kill the invocation and restart it as new, which re-runs everything from the input, including the tool call. The cheaper fix is to put the model call in `ctx.run` before anything gets stuck.

## Observations

- **The type-mismatch message swaps its two labels.** On the drifted run the SDK reports:

  ```
  - The previous execution ran and recorded the following: 'handler return' (index '1')
  - The current execution attempts to perform the following: 'run'
  ```

  The recorded journal has `Run get_weather` at index 1; it was this attempt that tried to return. The source agrees, in `restatedev/sdk-shared-core` v7.0.3, the core `restate-sdk` 1.0.5 is built on:
  - **Replay:** `PopJournalEntry` (`src/vm/transitions/journal.rs`) pops the recorded command and calls it `actual`. The command the handler is issuing now is `expected`.
  - **The call:** when the types differ, `RawMessage::decode_to` (`src/service_protocol/encoding.rs:80`) builds `CommandTypeMismatchError::new(index, <recorded type>, <current type>)`, filling `actual` and `expected` that way.
  - **The formatter:** `Display` for `CommandTypeMismatchError` (`src/vm/errors.rs:201-213`) prints `expected` as "previous execution ran and recorded" and `actual` as "current execution attempts".

  `main` has the same code as of 2026-09-16. The same-type message above (`CommandMismatchError`) prints a diff instead and is not affected. Reported upstream as [restatedev/sdk-shared-core#96](https://github.com/restatedev/sdk-shared-core/issues/96), with a fix in [#97](https://github.com/restatedev/sdk-shared-core/pull/97).
- **A paused invocation's failure moves to its journal events.** Once paused, `sys_invocation.last_failure` and `last_failure_error_code` are empty. The failure that caused the pause is kept on the invocation's `Paused` event in `sys_journal_events`, and `demo.py` reads it from there. The `TransientError` events beside it are not one per attempt: identical failures in a row are recorded once. A drifted run whose 17 attempts after the restart all failed with RT0016 held a single RT0016 `TransientError` event (2026-09-19).
- **RT0016 was retried, not failed immediately.** It followed the handler's retry policy (`on_max_attempts="pause"`) on this server version.

## Recording it by hand

`demo.py` is for reproducing the result. For a screen recording, three terminals read better:

```bash
# 1: the model stub, showing "model call #n" lines
MODEL_ANSWERS=stable .venv/bin/python model_stub.py

# 2: the agent
AGENT_MODE=naive .venv/bin/python agent.py

# 3: register, invoke, then kill terminal 2 during the pause and start it again
curl -s localhost:19070/deployments -H 'content-type: application/json' \
  -d '{"uri": "http://host.docker.internal:9080", "force": true}'
curl -s localhost:18080/WeatherAgent/run/send -H 'content-type: application/json' \
  -d '"What is the weather in Berlin?"'
```

`force: true` overwrites the registered deployment when the code changes between scenarios. That is a demo shortcut; in production, register a new deployment version instead ([Versioning](https://docs.restate.dev/services/versioning)).

## License

MIT © 2026 SciMigo
agent.py
"""A small weather agent on Restate: ask the model, maybe call a tool, answer.

    AGENT_MODE=naive      the model call is a plain HTTP request inside the handler
    AGENT_MODE=journaled  the model call goes through ctx.run_typed, so its result
                          is recorded in the invocation's journal

Everything else is identical. After the tool call the agent waits a few seconds
(a durable ctx.sleep, standing in for a rate-limit pause); the demo kills this
process during that wait, and Restate replays the handler when it comes back.
"""

import asyncio
import json
import os
import urllib.request
from datetime import timedelta

import restate
from restate.retry_policy import InvocationRetryPolicy

AGENT_MODE = os.environ.get("AGENT_MODE", "journaled")
STUB = os.environ.get("MODEL_STUB_URL", "http://127.0.0.1:8765")
PAUSE = timedelta(seconds=float(os.environ.get("AGENT_PAUSE_SECONDS", "6")))
PORT = int(os.environ.get("AGENT_PORT", "9080"))


def post(path: str, body: dict) -> dict:
    request = urllib.request.Request(
        STUB + path,
        data=json.dumps(body).encode(),
        headers={"content-type": "application/json"},
    )
    with urllib.request.urlopen(request, timeout=10) as response:
        return json.loads(response.read())


async def ask_model(messages: list[dict]) -> dict:
    return await asyncio.to_thread(post, "/model", {"messages": messages})


async def get_weather(city: str) -> str:
    result = await asyncio.to_thread(post, "/weather", {"city": city})
    return result["forecast"]


agent = restate.Service(
    "WeatherAgent",
    # Retry quickly once the process is back, so the replay is easy to watch.
    invocation_retry_policy=InvocationRetryPolicy(
        initial_interval=timedelta(milliseconds=500),
        exponentiation_factor=1.5,
        max_interval=timedelta(seconds=2),
        max_attempts=20,
        on_max_attempts="pause",
    ),
)


@agent.handler()
async def run(ctx: restate.Context, question: str) -> str:
    messages = [{"role": "user", "content": question}]

    for _ in range(4):
        if AGENT_MODE == "naive":
            decision = await ask_model(messages)  # not journaled: re-runs on every replay
        else:
            decision = await ctx.run_typed("call model", ask_model, messages=messages)

        if "answer" in decision:
            return decision["answer"]

        forecast = await ctx.run_typed("get_weather", get_weather, city=decision["city"])
        messages.append({"role": "tool", "content": forecast})
        await ctx.sleep(PAUSE, name="rate-limit pause")

    raise restate.TerminalError("agent did not finish in 4 steps")


app = restate.app(services=[agent])

if __name__ == "__main__":
    import hypercorn.asyncio
    import hypercorn.config

    config = hypercorn.config.Config()
    config.bind = [f"0.0.0.0:{PORT}"]
    print(f"WeatherAgent on :{PORT}, AGENT_MODE={AGENT_MODE}", flush=True)
    asyncio.run(hypercorn.asyncio.serve(app, config))
demo.py
"""Run one scenario end to end: start the agent, crash it mid-run, watch the replay.

    python demo.py --agent naive     --model stable    # the model is paid twice
    python demo.py --agent naive     --model varying   # ...and the replay diverges: RT0016
    python demo.py --agent naive     --model drifted   # ...and it stays broken until it pauses
    python demo.py --agent journaled --model varying   # the fix: one call, clean replay
    python demo.py --agent journaled --model stable --kill-during model
                                                       # ...but a call cut off mid-flight is paid again

By default the agent is killed one second into its durable pause and restarted two
seconds later. --kill-during model kills it while its first model call is waiting
for the answer instead; --restart-delay changes how long it stays down.

Needs the Restate server from docker-compose.yml (`docker compose up -d`).
The model stub's log lines ("model call #n") print live in this terminal.
"""

import argparse
import json
import os
import signal
import socket
import subprocess
import sys
import time
import urllib.error
import urllib.request

from term import style

HERE = os.path.dirname(os.path.abspath(__file__))
PY = sys.executable
INGRESS = os.environ.get("RESTATE_INGRESS", "http://127.0.0.1:18080")
ADMIN = os.environ.get("RESTATE_ADMIN", "http://127.0.0.1:19070")
STUB = "http://127.0.0.1:8765"
AGENT_URI_FOR_SERVER = os.environ.get("AGENT_URI", "http://host.docker.internal:9080")
QUESTION = "What is the weather in Berlin?"
SLOW_FIRST_ANSWER = 4  # seconds the stub holds back its first answer with --kill-during model


FAILURES = {"RT0010": "service unreachable", "RT0016": "journal mismatch"}
FAILURE_COLOR = {"RT0010": "yellow", "RT0016": "red"}


def say(line: str = "", *styles: str) -> None:
    """Print a line, in `styles` (see term.py) when colors are on."""
    print(style(line, *styles), flush=True)


def ui_links(invocation: str) -> None:
    """The Restate UI, and this invocation's page in it (both clickable in most terminals)."""
    say(f"UI:          {ADMIN}/ui/", "blue")
    say(f"invocation:  {ADMIN}/ui/invocations/{invocation}", "blue")


def http(method: str, url: str, body=None, timeout: float = 10):
    data = None if body is None else json.dumps(body).encode()
    request = urllib.request.Request(
        url, data=data, method=method,
        headers={"content-type": "application/json", "accept": "application/json"},
    )
    with urllib.request.urlopen(request, timeout=timeout) as response:
        raw = response.read()
        return json.loads(raw) if raw else None


def sql(query: str) -> list[dict]:
    return http("POST", f"{ADMIN}/query", {"query": query})["rows"]


def wait_for(check, timeout: float, interval: float = 0.25) -> bool:
    deadline = time.time() + timeout
    while time.time() < deadline:
        try:
            if check():
                return True
        except (urllib.error.URLError, ConnectionError, OSError):
            pass
        time.sleep(interval)
    return False


def port_open(port: int) -> bool:
    with socket.socket() as s:
        return s.connect_ex(("127.0.0.1", port)) == 0


def start_stub(model: str, slow_first: float = 0) -> subprocess.Popen:
    env = dict(os.environ, MODEL_ANSWERS=model, MODEL_SLOW_FIRST_SECONDS=str(slow_first))
    proc = subprocess.Popen([PY, os.path.join(HERE, "model_stub.py")], env=env)
    if not wait_for(lambda: http("GET", f"{STUB}/stats") is not None, 10):
        sys.exit("model stub did not start")
    return proc


def start_agent(mode: str, log, port: int = 9080) -> subprocess.Popen:
    env = dict(os.environ, AGENT_MODE=mode, AGENT_PORT=str(port))
    proc = subprocess.Popen([PY, os.path.join(HERE, "agent.py")], env=env, stdout=log, stderr=log)
    if not wait_for(lambda: port_open(port), 15):
        sys.exit("agent did not start; see .demo/agent.log")
    return proc


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("--agent", choices=["naive", "journaled"], required=True)
    parser.add_argument("--model", choices=["stable", "varying", "drifted"], default="stable")
    parser.add_argument("--wait", type=float, default=60, help="seconds to watch after the restart")
    parser.add_argument("--kill-during", choices=["pause", "model"], default="pause",
                        help="kill the agent in its durable pause (default), or while its first model call is in flight")
    parser.add_argument("--restart-delay", type=float, default=2, help="seconds between the kill and the restart")
    args = parser.parse_args()

    try:
        http("GET", f"{ADMIN}/health")
    except (urllib.error.URLError, OSError):
        sys.exit(f"Restate admin API not reachable at {ADMIN}. Run: docker compose up -d")
    for port in (8765, 9080):
        if port_open(port):
            sys.exit(f"port {port} is busy; stop the process using it first")

    os.makedirs(os.path.join(HERE, ".demo"), exist_ok=True)
    agent_log = open(os.path.join(HERE, ".demo", "agent.log"), "a")
    stub = agent = invocation = None
    row = {}
    try:
        say(f"== agent: {args.agent}   model answers: {args.model}", "bold")
        stub = start_stub(args.model, SLOW_FIRST_ANSWER if args.kill_during == "model" else 0)
        agent = start_agent(args.agent, agent_log)
        # force: this demo re-registers the same address with different code per scenario
        http("POST", f"{ADMIN}/deployments", {"uri": AGENT_URI_FOR_SERVER, "force": True})

        sent = http("POST", f"{INGRESS}/WeatherAgent/run/send", QUESTION)
        invocation = sent["invocationId"]
        say(f"invoked WeatherAgent/run   {invocation}")

        if args.kill_during == "model":
            if not wait_for(lambda: http("GET", f"{STUB}/stats")["model_calls"] >= 1, 20):
                sys.exit("the agent never called the model")
            time.sleep(1)  # the stub has the answer and is still holding it back
        else:
            if not wait_for(lambda: http("GET", f"{STUB}/stats")["weather_calls"] >= 1, 20):
                sys.exit("the agent never reached the weather tool")
            time.sleep(1)  # now inside the durable rate-limit pause

        agent.send_signal(signal.SIGKILL)
        agent.wait()
        during = ", before model call #1 returned its answer" if args.kill_during == "model" else ""
        say(f"kill -9 agent (pid {agent.pid}){during}", "bold", "red")
        time.sleep(args.restart_delay)
        agent = start_agent(args.agent, agent_log)
        say(f"agent restarted (pid {agent.pid}); Restate retries and replays the journal", "bold")

        seen, failure = None, None
        deadline = time.time() + args.wait
        while time.time() < deadline:
            rows = sql(
                "SELECT status, retry_count, last_failure_error_code, last_failure "
                f"FROM sys_invocation WHERE id = '{invocation}'"
            )
            row = rows[0] if rows else {}
            if row.get("status") in ("completed", "paused"):
                break
            code = row.get("last_failure_error_code")
            if code:
                failure = (code, row.get("last_failure") or "")
            if code and (row.get("retry_count"), code) != seen:
                seen = (row.get("retry_count"), code)
                say(f"  attempt {seen[0]}, last failure {code} ({FAILURES.get(code, 'see UI')})",
                    FAILURE_COLOR.get(code, "yellow"))
            time.sleep(0.5)

        if row.get("status") == "paused":
            # Pausing clears last_failure on sys_invocation; the failure that caused the
            # pause is kept on the invocation's Paused event in sys_journal_events.
            events = sql(
                f"SELECT event_json FROM sys_journal_events WHERE id = '{invocation}' "
                "AND event_type = 'Paused' ORDER BY appended_at DESC LIMIT 1"
            )
            if events:
                last = json.loads(events[0]["event_json"]).get("last_failure") or {}
                failure = (last.get("restate_doc_error_code"), last.get("error_message") or "")

        say()
        stats = http("GET", f"{STUB}/stats")
        completed = row.get("status") == "completed"
        say(f"status: {row.get('status')}   model calls: {stats['model_calls']}   weather calls: {stats['weather_calls']}",
            "bold", "green" if completed else "red")
        if completed:
            result = http("GET", f"{INGRESS}/restate/invocation/{invocation}/output")
            say(f"result: {result}", "green")
        elif failure:
            say(f"last failure {failure[0]}:", "bold", FAILURE_COLOR.get(failure[0], "red"))
            print("  " + failure[1].strip().replace("\n", "\n  "), flush=True)

        say("journal:", "bold")
        for entry in sql(
            f"SELECT index, entry_type, name FROM sys_journal WHERE id = '{invocation}' ORDER BY index"
        ):
            name = f"  {entry['name']}" if entry.get("name") else ""
            say(f"  {entry['index']:>2}  {entry['entry_type']}{name}",
                *(("cyan",) if entry.get("name") == "call model" else ()))
        ui_links(invocation)
    finally:
        # Stopped early (Ctrl-C, the lab page's Stop, --wait running out): the invocation is still
        # retrying, and its next attempt would reach whichever agent runs next and bill that run's
        # stub. Kill it so every scenario starts alone.
        if invocation and row.get("status") not in ("completed", "paused"):
            try:
                http("PATCH", f"{ADMIN}/invocations/{invocation}/kill")
                say(f"killed invocation {invocation}: it had not finished, so it cannot retry into the next run",
                    "yellow")
            except (urllib.error.URLError, OSError) as error:
                say(f"could not kill invocation {invocation} ({error}); kill it in the UI before the next run")
        for proc in (agent, stub):
            if proc and proc.poll() is None:
                # Hypercorn shuts down on SIGINT, not SIGTERM.
                proc.send_signal(signal.SIGINT)
                try:
                    proc.wait(timeout=5)
                except subprocess.TimeoutExpired:
                    proc.kill()
                    proc.wait()
        agent_log.close()


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        sys.exit(130)  # stopped on purpose; the finally block has already cleaned up
docker-compose.yml
services:
  restate:
    image: docker.restate.dev/restatedev/restate:1.7.10
    command: ["--node-name=restate-1"]
    ports:
      - "18080:8080"   # ingress: invoke handlers
      - "19070:9070"   # admin API and the Restate UI
    extra_hosts:
      # The agent runs on the host; this is how the server reaches it.
      - "host.docker.internal:host-gateway"
lab_server.py
#!/usr/bin/env python3
"""Local browser workspace for the Restate demo. Standard library only.

Run `docker compose up -d`, then `python3 lab_server.py` and open the printed
http://127.0.0.1 address. Only loopback requests are accepted.
"""
from __future__ import annotations

import argparse
import json
import os
import signal
import socket
import subprocess
import sys
import threading
import time
import urllib.error
import urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import urlsplit

ROOT = Path(__file__).resolve().parent
PAGE = (ROOT / "lab_page.html").read_bytes()
SCENARIOS = {
    "stable": ("1.1", "demo.py", "--agent", "naive", "--model", "stable"),
    "varying": ("1.2", "demo.py", "--agent", "naive", "--model", "varying"),
    "drifted": ("1.3", "demo.py", "--agent", "naive", "--model", "drifted", "--wait", "150"),
    "delayed": ("1.3 variation", "demo.py", "--agent", "naive", "--model", "drifted", "--wait", "150", "--restart-delay", "9"),
    "journaled": ("1.4", "demo.py", "--agent", "journaled", "--model", "varying"),
    "midcall": ("1.5", "demo.py", "--agent", "journaled", "--model", "stable", "--kill-during", "model"),
    "recovery": ("1.6", "recover.py"),
}
lock = threading.Lock()
current: subprocess.Popen | None = None
current_key = ""
current_log: Path | None = None
current_started = 0.0
prepared = False


def restate_ready() -> bool:
    try:
        with urllib.request.urlopen("http://127.0.0.1:19070/health", timeout=1) as response:
            return response.status == 200
    except (OSError, urllib.error.URLError):
        return False


def port_open(port: int) -> bool:
    with socket.socket() as sock:
        sock.settimeout(.3)
        return sock.connect_ex(("127.0.0.1", port)) == 0


def snapshot() -> dict:
    with lock:
        proc, key, log, started = current, current_key, current_log, current_started
    running = proc is not None and proc.poll() is None
    tail = ""
    if log and log.exists():
        with log.open("rb") as source:
            source.seek(max(0, log.stat().st_size - 40000))
            tail = source.read().decode("utf-8", errors="replace")
    return {"restate": restate_ready(), "prepared": prepared,
            "running": running, "scenario": key,
            "elapsed": round(time.time() - started) if running else None,
            "exitCode": None if running or proc is None else proc.returncode,
            "output": tail}


class Handler(BaseHTTPRequestHandler):
    def send_data(self, status: int, data: bytes, ctype: str) -> None:
        self.send_response(status)
        self.send_header("Content-Type", ctype)
        self.send_header("Content-Length", str(len(data)))
        self.send_header("Cache-Control", "no-store")
        self.send_header("X-Content-Type-Options", "nosniff")
        self.send_header("Content-Security-Policy", "default-src 'self'; style-src 'self' 'unsafe-inline'; script-src 'self' 'unsafe-inline'")
        self.end_headers()
        self.wfile.write(data)

    def send_json(self, status: int, value: dict) -> None:
        self.send_data(status, json.dumps(value).encode(), "application/json")

    def local_request(self) -> bool:
        host = self.headers.get("Host", "")
        if host not in {f"127.0.0.1:{self.server.server_port}", f"localhost:{self.server.server_port}"}:
            return False
        origin = self.headers.get("Origin")
        return not origin or urlsplit(origin).netloc == host and urlsplit(origin).scheme == "http"

    def do_GET(self) -> None:
        if not self.local_request():
            return self.send_json(403, {"error": "Open the printed local address."})
        if self.path in ("/", "/en", "/en/"):
            return self.send_data(200, PAGE, "text/html; charset=utf-8")
        if self.path == "/api/status":
            return self.send_json(200, snapshot())
        self.send_data(404, b"Not found", "text/plain")

    def do_POST(self) -> None:
        global current, current_key, current_log, current_started, prepared
        if not self.local_request():
            return self.send_json(403, {"error": "Open the printed local address."})
        try:
            length = int(self.headers.get("Content-Length", "0"))
            if length > 1024:
                raise ValueError("Request too large")
            body = json.loads(self.rfile.read(length) or b"{}")
            if not isinstance(body, dict):
                raise ValueError("Expected a JSON object")
            if self.path == "/api/prepare":
                with lock:
                    if current and current.poll() is None:
                        raise ValueError("Stop the current scenario before preparing again.")
                if sys.version_info < (3, 10):
                    raise ValueError("Python 3.10 or newer is required for the demo.")
                venv = ROOT / ".venv"
                if not (venv / "bin/python").exists():
                    subprocess.run([sys.executable, "-m", "venv", str(venv)], cwd=ROOT, check=True)
                install = subprocess.run([str(venv / "bin/python"), "-m", "pip", "install", "-q", "-r", "requirements.txt"],
                                         cwd=ROOT, capture_output=True, text=True, timeout=240,
                                         env={**os.environ, "PIP_DISABLE_PIP_VERSION_CHECK": "1"})
                if install.returncode:
                    raise RuntimeError("Package install failed: " + (install.stderr or install.stdout)[-2000:])
                prepared = True
                return self.send_json(200, {"message": "Ready. Python packages installed in .venv; Restate " + ("is ready." if restate_ready() else "is not running yet. Start Docker and refresh.")})
            if self.path == "/api/start":
                key = body.get("scenario")
                if key not in SCENARIOS:
                    raise ValueError("Choose a scenario card.")
                if not prepared and not (ROOT / ".venv/bin/python").exists():
                    raise ValueError("Click Prepare first.")
                if not restate_ready():
                    raise ValueError("Restate is not ready. In another terminal run: docker compose up -d")
                with lock:
                    if current and current.poll() is None:
                        raise ValueError(f"{current_key} is still running. Wait or stop it before starting another scenario.")
                    if any(port_open(port) for port in (8765, 9080, 9081)):
                        raise ValueError("A model stub or agent is still using port 8765, 9080, or 9081. Stop that run first.")
                    label, *args = SCENARIOS[key]
                    logs = ROOT / ".browser-runs"
                    logs.mkdir(exist_ok=True)
                    log = logs / f"{key}-{time.time_ns()}.log"
                    with log.open("wb") as output:
                        current = subprocess.Popen([str(ROOT / ".venv/bin/python"), *args], cwd=ROOT,
                            stdin=subprocess.DEVNULL, stdout=output, stderr=subprocess.STDOUT,
                            env={**os.environ, "PYTHONUNBUFFERED": "1"}, start_new_session=True)
                    current_key, current_log, current_started = key, log, time.time()
                return self.send_json(200, {"message": f"Started exercise {label}. Output updates below."})
            if self.path == "/api/stop":
                with lock:
                    proc = current
                if not proc or proc.poll() is not None:
                    raise ValueError("No scenario is running.")
                os.killpg(proc.pid, signal.SIGINT)
                try:
                    proc.wait(timeout=8)
                except subprocess.TimeoutExpired:
                    os.killpg(proc.pid, signal.SIGKILL)
                    proc.wait(timeout=3)
                return self.send_json(200, {"message": "Stopped the scenario and its child processes."})
            self.send_data(404, b"Not found", "text/plain")
        except (ValueError, RuntimeError, subprocess.SubprocessError, OSError) as error:
            self.send_json(400, {"error": str(error)})

    def log_message(self, *args) -> None:
        pass


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--port", type=int, default=3002, help="local port (default: 3002)")
    args = parser.parse_args()
    if port_open(args.port):
        parser.exit(1, f"Port {args.port} is in use. Run: python3 lab_server.py --port {args.port + 1}\n")
    server = ThreadingHTTPServer(("127.0.0.1", args.port), Handler)
    print(f"lab      http://127.0.0.1:{args.port}/", flush=True)
    print(f"restate  http://127.0.0.1:19070/  ({'ready' if restate_ready() else 'not running: docker compose up -d'})", flush=True)
    print("Ctrl-C to stop the page. A running scenario will keep going until it finishes or you click Stop.", flush=True)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\nLab page stopped.")
    finally:
        server.server_close()


if __name__ == "__main__":
    main()
model_stub.py
"""A stand-in for a paid model API and a weather API, with call counters.

The agent service is killed and restarted during the demo, so the counters live
here, in a separate process: they are the evidence of what ran twice.

    MODEL_ANSWERS=stable   the model gives the same decision for the same messages
    MODEL_ANSWERS=varying  like a real model at temperature > 0: the same messages
                           get a different decision on alternate calls
    MODEL_ANSWERS=drifted  the model changes its mind once and stays changed: after
                           the first call, the same messages always get a direct answer

    MODEL_SLOW_FIRST_SECONDS=4  the first model call decides at once but holds its answer
                           back this long, so the caller can die after the model has done
                           (and billed) the work and before the answer arrives

Endpoints:
    POST /model         {"messages": [...]} -> {"tool": "get_weather", "city": ...} or {"answer": ...}
    POST /weather       {"city": ...}       -> {"forecast": ...}
    GET  /stats         {"model_calls": n, "weather_calls": n}
    POST /reset         zero the counters
"""

import json
import os
import sys
import threading
import time
from datetime import datetime
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

from term import style

MODE = os.environ.get("MODEL_ANSWERS", "stable")
PORT = int(os.environ.get("MODEL_STUB_PORT", "8765"))
SLOW_FIRST = float(os.environ.get("MODEL_SLOW_FIRST_SECONDS") or 0)

_lock = threading.Lock()
_counts = {"model_calls": 0, "weather_calls": 0}


def log(line: str, *styles: str) -> None:
    print(f"{datetime.now():%H:%M:%S}  {style(line, *styles)}", flush=True)


def decide(messages: list[dict], call_number: int) -> dict:
    tool_result = next((m for m in messages if m.get("role") == "tool"), None)
    if tool_result is not None:
        return {"answer": f"Berlin right now: {tool_result['content']}."}
    changed_mind = (MODE == "varying" and call_number % 2 == 0) or (MODE == "drifted" and call_number > 1)
    if changed_mind:
        # Same question, different decision: skip the tool and answer directly.
        return {"answer": "Berlin is probably mild this time of year."}
    return {"tool": "get_weather", "city": "Berlin"}


class Handler(BaseHTTPRequestHandler):
    def _json(self, status: int, body: dict) -> None:
        data = json.dumps(body).encode()
        self.send_response(status)
        self.send_header("content-type", "application/json")
        self.send_header("content-length", str(len(data)))
        self.end_headers()
        self.wfile.write(data)

    def _body(self) -> dict:
        length = int(self.headers.get("content-length") or 0)
        return json.loads(self.rfile.read(length) or b"{}")

    def do_GET(self) -> None:
        if self.path == "/stats":
            with _lock:
                return self._json(200, dict(_counts))
        self._json(404, {"error": "not found"})

    def do_POST(self) -> None:
        body = self._body()
        if self.path == "/model":
            with _lock:
                _counts["model_calls"] += 1
                n = _counts["model_calls"]
            decision = decide(body.get("messages", []), n)
            log(f"model call #{n}  ->  {json.dumps(decision)}", "yellow")
            if n == 1 and SLOW_FIRST:
                time.sleep(SLOW_FIRST)
            try:
                return self._json(200, decision)
            except (BrokenPipeError, ConnectionResetError):
                log(f"model call #{n}  answer not delivered: the caller is gone", "red")
                return
        if self.path == "/weather":
            with _lock:
                _counts["weather_calls"] += 1
                n = _counts["weather_calls"]
            log(f"weather call #{n}  get_weather({body.get('city')})", "cyan")
            return self._json(200, {"forecast": "18°C and cloudy"})
        if self.path == "/reset":
            with _lock:
                _counts.update(model_calls=0, weather_calls=0)
            return self._json(200, dict(_counts))
        self._json(404, {"error": "not found"})

    def log_message(self, *args) -> None:  # silence the default access log
        pass


if __name__ == "__main__":
    server = ThreadingHTTPServer(("127.0.0.1", PORT), Handler)
    log(f"model stub on :{PORT}, MODEL_ANSWERS={MODE}", "dim")
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        sys.exit(0)
recover.py
"""What can you do with an invocation stuck on RT0016?

    python recover.py

1. Recreates the stuck invocation from scenario 3 (naive agent, drifted model):
   it retries on journal mismatch until it pauses.
2. Deploys the fixed (journaled) agent as a second deployment on :9081 and
   resumes the paused invocation on it.
3. Tries restart-as-new, kills the invocation, and restarts it as new on the
   fixed deployment.

Needs the Restate server from docker-compose.yml and free ports 8765, 9080, 9081.
"""

import json
import os
import signal
import subprocess
import sys
import time
import urllib.error
import urllib.request

from demo import (ADMIN, AGENT_URI_FOR_SERVER, HERE, INGRESS, QUESTION, STUB, port_open, say, sql,
                  start_agent, start_stub, ui_links, wait_for)

FIXED_URI = os.environ.get("FIXED_AGENT_URI", "http://host.docker.internal:9081")


def call(method: str, url: str, body=None):
    """Like demo.http, but returns (status, body) instead of raising on 4xx."""
    data = None if body is None else json.dumps(body).encode()
    request = urllib.request.Request(
        url, data=data, method=method,
        headers={"content-type": "application/json", "accept": "application/json"},
    )
    try:
        with urllib.request.urlopen(request, timeout=15) as response:
            raw = response.read()
            return response.status, json.loads(raw) if raw else None
    except urllib.error.HTTPError as error:
        raw = error.read()
        try:
            return error.code, json.loads(raw)
        except ValueError:
            return error.code, raw.decode()


def invocation(invocation_id: str) -> dict:
    rows = sql(
        "SELECT status, retry_count, last_failure_error_code, last_failure, pinned_deployment_id, "
        f"completion_result, completion_failure FROM sys_invocation WHERE id = '{invocation_id}'"
    )
    return rows[0] if rows else {}


def model_calls() -> int:
    return call("GET", f"{STUB}/stats")[1]["model_calls"]


def journal(invocation_id: str) -> None:
    for entry in sql(f"SELECT index, entry_type, name FROM sys_journal WHERE id = '{invocation_id}' ORDER BY index"):
        name = f"  {entry['name']}" if entry.get("name") else ""
        print(f"  {entry['index']:>2}  {entry['entry_type']}{name}", flush=True)


def stop(*procs) -> None:
    for proc in procs:
        if proc and proc.poll() is None:
            proc.send_signal(signal.SIGINT)
            try:
                proc.wait(timeout=5)
            except subprocess.TimeoutExpired:
                proc.kill()
                proc.wait()


def ok(status: int) -> str:
    """Green for a 2xx answer, red for a refusal."""
    return "green" if 200 <= status < 300 else "red"


def main() -> None:
    if call("GET", f"{ADMIN}/health")[0] != 200:
        sys.exit(f"Restate admin API not reachable at {ADMIN}. Run: docker compose up -d")
    for port in (8765, 9080, 9081):
        if port_open(port):
            sys.exit(f"port {port} is busy; stop the process using it first")

    os.makedirs(os.path.join(HERE, ".demo"), exist_ok=True)
    log = open(os.path.join(HERE, ".demo", "agent.log"), "a")
    stub = naive = fixed = None
    stuck = restarted = None
    try:
        say("== 1. a stuck invocation: naive agent, drifted model", "bold")
        stub = start_stub("drifted")
        naive = start_agent("naive", log)
        _, deployment_a = call("POST", f"{ADMIN}/deployments", {"uri": AGENT_URI_FOR_SERVER, "force": True})
        stuck = call("POST", f"{INGRESS}/WeatherAgent/run/send", QUESTION)[1]["invocationId"]
        say(f"invoked {stuck} on deployment {deployment_a['id']}")
        wait_for(lambda: call("GET", f"{STUB}/stats")[1]["weather_calls"] >= 1, 20)
        time.sleep(1)
        naive.send_signal(signal.SIGKILL)
        naive.wait()
        time.sleep(2)
        naive = start_agent("naive", log)
        say("kill -9 and restart; waiting for the retries to run out", "bold", "red")
        wait_for(lambda: invocation(stuck).get("status") == "paused", 150, 0.5)
        say(f"status: {invocation(stuck).get('status')}   model calls so far: {model_calls()}", "bold", "red")

        say()
        say("== 2. deploy the fix as a new deployment and resume the paused invocation on it", "bold")
        fixed = start_agent("journaled", log, port=9081)
        _, deployment_b = call("POST", f"{ADMIN}/deployments", {"uri": FIXED_URI, "force": True})
        before = model_calls()
        status, _ = call("PATCH", f"{ADMIN}/invocations/{stuck}/resume?deployment={deployment_b['id']}")
        say(f"PATCH /invocations/{stuck}/resume?deployment={deployment_b['id']}  ->  {status}", ok(status))
        failure = {}
        wait_for(lambda: failure.update(invocation(stuck)) or "Difference" in (failure.get("last_failure") or ""), 30, 0.5)
        say(f"pinned deployment: {failure.get('pinned_deployment_id')}   last failure {failure.get('last_failure_error_code')}:")
        print("  " + (failure.get("last_failure") or "").strip().replace("\n", "\n  "), flush=True)
        say(f"model calls since resume: {model_calls() - before}")
        status, _ = call("PATCH", f"{ADMIN}/invocations/{stuck}/pause")
        wait_for(lambda: invocation(stuck).get("status") == "paused", 20, 0.5)
        say(f"PATCH pause  ->  {status}; status: {invocation(stuck).get('status')}", ok(status))

        say()
        say("== 3. restart it as new on the fixed deployment", "bold")
        status, body = call("PATCH", f"{ADMIN}/invocations/{stuck}/restart-as-new")
        say(f"PATCH restart-as-new while paused  ->  {status} {body}", ok(status))
        status, _ = call("PATCH", f"{ADMIN}/invocations/{stuck}/kill")
        wait_for(lambda: invocation(stuck).get("status") == "completed", 20, 0.5)
        original = invocation(stuck)
        say(f"PATCH kill  ->  {status}; original: {original.get('completion_result')} {original.get('completion_failure')}", ok(status))
        call("POST", f"{STUB}/reset")
        status, body = call("PATCH", f"{ADMIN}/invocations/{stuck}/restart-as-new")
        restarted = body["new_invocation_id"]
        say(f"PATCH restart-as-new after kill  ->  {status}   new invocation {restarted}", ok(status))
        wait_for(lambda: invocation(restarted).get("status") == "completed", 60, 0.5)
        row = invocation(restarted)
        output = call("GET", f"{INGRESS}/restate/invocation/{restarted}/output")[1]
        say(f"status: {row.get('status')}   pinned deployment: {row.get('pinned_deployment_id')}   model calls: {model_calls()}",
            "bold", "green" if row.get("status") == "completed" else "red")
        say(f"result: {output}", "green")
        say("journal of the new invocation:", "bold")
        journal(restarted)
        ui_links(restarted)
        say(f"original:    {ADMIN}/ui/invocations/{stuck}", "blue")
    finally:
        # Stopped early (Ctrl-C, the lab page's Stop): an invocation still retrying would reach the
        # next scenario's agent and bill its stub. Kill it, as demo.py does; a paused one stays put.
        for leftover in (stuck, restarted):
            status = invocation(leftover).get("status") if leftover else None
            if status and status not in ("completed", "paused"):
                call("PATCH", f"{ADMIN}/invocations/{leftover}/kill")
                say(f"killed invocation {leftover}: it had not finished, so it cannot retry into the next run")
        stop(naive, fixed, stub)
        log.close()


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        sys.exit(130)  # stopped on purpose; the finally block has already cleaned up
requirements.txt
restate-sdk==1.0.5
hypercorn==0.18.0
term.py
"""Colors for the demo's output: in a terminal, or when FORCE_COLOR is set (agent-runtime sets it
for lab actions, and the scimigo.com lab page renders the colors). NO_COLOR turns them off.
Piped output without FORCE_COLOR stays plain text, as recorded in observed/.
"""

import os
import sys

ENABLED = (sys.stdout.isatty() or bool(os.environ.get("FORCE_COLOR"))) and not os.environ.get("NO_COLOR")

_CODES = {
    "bold": "1",
    "dim": "2",
    "red": "31",
    "green": "32",
    "yellow": "33",
    "blue": "34",
    "magenta": "35",
    "cyan": "36",
}


def style(text: str, *names: str) -> str:
    """`text` wrapped in the ANSI codes for `names` (e.g. "bold", "red"), or unchanged."""
    if not ENABLED or not names or not text:
        return text
    return "".join(f"\033[{_CODES[name]}m" for name in names) + text + "\033[0m"

Lab source: https://github.com/SciMigo/restate-durable-agent-demo at f515afa. Observed outputs recorded against restate-server 1.7.10 and restate-sdk 1.0.5 on 2026-09-16.