diff --git a/agents/ROADMAP.md b/agents/ROADMAP.md
index 18f0d11..1891454 100644
--- a/agents/ROADMAP.md
+++ b/agents/ROADMAP.md
@@ -1,7 +1,7 @@
 # Durable — Roadmap
 
 **Last Updated:** 2026-10-23
-**Current release:** `durable` 1.0.0 and `durable_dashboard` 1.0.0 on Hex (2026-10-21)
+**Current release:** `durable` 1.1.0 (2026-10-23) and `durable_dashboard` 1.0.0 (2026-10-21) on Hex
 **Reference:** See `arch.md` for technical architecture, `WORKPLAN_ARCHIVED.md` for historical detail
 
 ---
@@ -50,6 +50,7 @@
 | [M05](milestones/M05-message-bus.md) | Message Bus | Partial | `Durable.PubSub` lifecycle events shipped. Adapter behaviour / pg_notify not built → backlog |
 | [M06](milestones/M06-phoenix-dashboard.md) | Phoenix Dashboard | Done | Shipped as its own package |
 | [M07](milestones/M07-scalability-adapters.md) | Scalability Adapters | Backlog | No demand for Redis yet |
+| [M08](milestones/M08-workflow-versioning.md) | Workflow Versioning | In Progress | Phase 1 (fail on a removed step) done; phase 2 (`version:` pinning) waits on the doc's open questions |
 
 ---
 
@@ -57,13 +58,9 @@
 
 In priority order:
 
-1. **Correctness follow-ups from v1:**
-   - The workflow deadline can't interrupt a step that's still running; it's only checked between steps.
-   - `wait_for_input` with `on_timeout: :fail` ends the run as `cancelled`, not `failed`. Fixing it changes behaviour, so ship it in a minor release with a changelog note.
-2. **Workflow versioning.** Needs a milestone doc. Changing a deployed workflow while runs are still in flight is the main open problem for production users.
-3. **DSL gaps:** waits inside `parallel` blocks, an `each`/foreach primitive (each iteration needs its own context, see `Durable.DSL.Step` docs), decisions inside `parallel`.
-4. **Testing follow-up:** wait timeouts can't be exercised by `Durable.Testing.drain/2` (the timeout worker only runs with the queue on). A helper that fires a run's pending timeouts would close it.
-5. **Formatter follow-up to #10:** the repo's own `durable/.formatter.exs` only *exports* `locals_without_parens`, so internal code and tests still get parens. Also add `:durable` to `import_deps` in `durable_dashboard` and `examples/phoenix_demo`.
+1. **Workflow versioning, phase 2** ([M08](milestones/M08-workflow-versioning.md)). Pin each run to the `version:` it started on. Needs the doc's three open questions answered first. Adds a migration.
+2. **DSL gaps:** waits inside `parallel` blocks, an `each`/foreach primitive (each iteration needs its own context, see `Durable.DSL.Step` docs), decisions inside `parallel`.
+3. **Testing follow-up:** wait timeouts can't be exercised by `Durable.Testing.drain/2` (the timeout worker only runs with the queue on). A helper that fires a run's pending timeouts would close it.
 
 ---
 
@@ -86,9 +83,10 @@ In priority order:
 2. Decisions are not supported inside parallel blocks
 3. No backward jumps in decision steps (forward-only by design)
 4. Context is single-level atomized (top-level keys only)
-5. No workflow versioning
+5. No workflow version pinning (M08)
 6. No foreach/loop DSL primitive (use Elixir's `Enum` functions)
 7. A resumed wait re-runs the step body from the top; side effects before the wait must be idempotent
 8. Only one `sleep/1` per step body
-9. The workflow deadline is checked between steps and can't interrupt a running step
-10. `Durable.Testing.drain/2` doesn't fire wait timeouts
+9. `Durable.Testing.drain/2` doesn't fire wait timeouts
+10. A step with a `timeout:`, or in a workflow with one, runs in its own process (`self()` isn't the worker)
+11. In-flight runs resume on the deployed definition; a removed current step fails the run (versioning: M08)
diff --git a/agents/milestones/M08-workflow-versioning.md b/agents/milestones/M08-workflow-versioning.md
new file mode 100644
index 0000000..5ccea63
--- /dev/null
+++ b/agents/milestones/M08-workflow-versioning.md
@@ -0,0 +1,158 @@
+# M08: Workflow Versioning
+
+**Status:** In Progress (phase 1 done, phase 2 needs the open questions answered)
+**Priority:** High
+**Effort:** 1 week
+**Dependencies:** None
+
+## Motivation
+
+A deploy changes the workflow code under runs that are still in flight. Today a
+run resumes against whatever definition is deployed, matched to its position by
+step name. Whether that's safe depends on the change:
+
+| Change | In-flight run today |
+|--------|---------------------|
+| Edit a step body | Runs the new body. Usually what you want (bug fixes). |
+| Add a step after the run's position | Runs it. |
+| Add a step before the run's position | Skips it. |
+| Remove a later step | Skips it. |
+| Rename or remove the step the run stopped at | **Was:** marked `completed`, every step after skipped, no error. **Now (phase 1):** fails with `definition_changed`. |
+| Change a `parallel`/`branch` block's steps | The block's id is a hash of its shape, so its step names change: same as a rename. |
+| A step expects context an earlier *new* step writes | Crashes at runtime, retried, fails. |
+
+The silent `completed` was a data-loss bug and is fixed. The rest needs a way to
+keep old runs on the old definition until they finish, which is what this
+milestone adds.
+
+## Scope
+
+### In Scope
+
+- Phase 1 (done): a run whose current step no longer exists fails with
+  `%{type: "definition_changed"}` and no compensations, so it stays retryable
+  once the step is restored.
+- Phase 2: explicit versions. Several definitions of one workflow side by side;
+  each run pinned to the version it started on.
+- A way to see which versions still have runs in flight, so old code can be
+  deleted safely.
+
+### Out of Scope
+
+- Replay-based patching (Temporal's `patched?`). Durable checkpoints steps
+  rather than replaying history, so there's nothing to keep deterministic.
+- Automatic versions from a structural hash. Every change would make a new
+  version, including the safe ones above, and old definitions still have to
+  stay in the code to be run.
+- Migrating a run from one version to another (moving its position by hand).
+
+## Architecture & Design
+
+### API
+
+```elixir
+defmodule MyApp.OrderWorkflow do
+  use Durable
+
+  # Runs started before v2 shipped finish on this. Delete it once
+  # `mix durable.status` shows none left.
+  workflow "order", version: 1 do
+    step :reserve, ...
+    step :approve, ...
+    step :ship, ...
+  end
+
+  workflow "order", version: 2 do
+    step :reserve, ...
+    step :review, ...
+    step :ship, ...
+  end
+end
+```
+
+- `version:` is a positive integer workflow option, default `1`. Every workflow
+  written so far is version 1, which matches every existing run.
+- Two `workflow` blocks with the same name and version fail the compile.
+- `Durable.start/3`, `call_workflow`, and schedules start the highest version.
+  `Durable.start(mod, input, version: 1)` picks one explicitly (for tests of the
+  old path).
+- `workflow_version/0` (in `Durable.Context`) returns the run's version, so one
+  definition can branch on it for small differences instead of being copied.
+
+### Storage
+
+- New column `workflow_executions.workflow_version integer NOT NULL DEFAULT 1`
+  (a new migration, shipped through `mix durable.gen.upgrade`). The default
+  covers existing rows.
+- Index `(workflow_module, workflow_name, workflow_version) WHERE status IN
+  (pending, running, waiting)` for the in-flight report.
+
+### Resolution
+
+- `__workflow_definition__(name)` keeps returning the latest version (the
+  dashboard and `Durable.Scheduler` rely on it);
+  `__workflow_definition__(name, version)` returns one version, and
+  `__workflow_versions__(name)` lists them.
+- The executor resolves `(module, name, version)`. A version that's gone fails
+  the run with `definition_changed`, like a missing step.
+- Parallel children copy the parent's version. A `call_workflow` child is a new
+  run and starts the callee's latest version.
+
+### Visibility
+
+- `Durable.Query.list_executions(version: 1)`.
+- `mix durable.status` shows in-flight runs per version.
+- Dashboard: the version on the run page; per-version in-flight counts on the
+  workflow page.
+- Telemetry: `:workflow_version` in run metadata (adding keys is allowed by the
+  telemetry contract).
+
+## Implementation Plan
+
+1. ~~Phase 1: fail on a missing current step~~ (done in this branch).
+2. DSL: `version:` option, per-version definition functions, duplicate check,
+   `__workflow_versions__/1`.
+3. Migration, schema, executor resolution, start/call/schedule pick latest,
+   parallel children inherit.
+4. `workflow_version/0`, query filter, `mix durable.status`, telemetry.
+5. Dashboard: version on the run page and in-flight counts per version.
+6. Guide: "Changing workflows" — which changes are safe in place, when to add a
+   version, how to retire one.
+
+## Testing Strategy
+
+- Two versions in one module: new runs start v2, a v1 run resumes and finishes
+  on v1.
+- A run whose version was deleted fails with `definition_changed`; restoring it
+  and `Durable.retry/2` finishes the run.
+- Parallel children run the parent's version.
+- Existing rows (no version set) resolve as version 1.
+- Duplicate `name`/`version` fails the compile.
+
+## Acceptance Criteria
+
+- [x] A deploy that removes a run's current step fails the run loudly and
+      recoverably instead of completing it.
+- [ ] A run always finishes on the version it started on, while that version is
+      in the code.
+- [ ] Operators can see which versions still have runs before deleting one.
+- [ ] Upgrading from 1.x needs only `mix durable.gen.upgrade` and a migrate.
+
+## Open Questions
+
+1. **API shape:** several `workflow "order", version: N` blocks in one module
+   (proposed), or one module per version (`MyApp.Order.V1`) tied together by a
+   name?
+2. **Release:** the new column means users must run a migration. Ship as 1.2.0
+   with an upgrade note, as the migration tooling was built for, or hold for
+   2.0?
+3. **Missing version:** fail the run (proposed), or fall back to the latest
+   version when the run's current step exists there too?
+
+## References
+
+- AWS Step Functions versions and aliases: executions are pinned to the version
+  they started on.
+- Restate deployments: old deployments keep serving in-flight invocations until
+  they drain.
+- `test/durable/definition_change_test.exs`: the phase 1 behaviour.
diff --git a/durable/.formatter.exs b/durable/.formatter.exs
index da888af..d353636 100644
--- a/durable/.formatter.exs
+++ b/durable/.formatter.exs
@@ -1,21 +1,22 @@
 # Used by "mix format"
+locals_without_parens = [
+  # Workflow DSL
+  workflow: 2,
+  workflow: 3,
+  # Step DSL
+  branch: 2,
+  compensate: 2,
+  compensate: 3,
+  decision: 2,
+  decision: 3,
+  parallel: 1,
+  parallel: 2,
+  step: 2,
+  step: 3
+]
+
 [
   inputs: ["{mix,.formatter}.exs", "{config,lib,test}/**/*.{ex,exs}"],
-  export: [
-    locals_without_parens: [
-      # Workflow DSL
-      workflow: 2,
-      workflow: 3,
-      # Step DSL
-      branch: 2,
-      compensate: 2,
-      compensate: 3,
-      decision: 2,
-      decision: 3,
-      parallel: 1,
-      parallel: 2,
-      step: 2,
-      step: 3
-    ]
-  ]
+  locals_without_parens: locals_without_parens,
+  export: [locals_without_parens: locals_without_parens]
 ]
diff --git a/durable/README.md b/durable/README.md
index 620ca25..43620ec 100644
--- a/durable/README.md
+++ b/durable/README.md
@@ -497,9 +497,9 @@ workflow "process_order", timeout: hours(2), queue: :orders do
 ```
 
 - `:timeout` — how long a run may take from when it's due to start, waits and
-  sleeps included. A run past it fails with a `"timeout"` error before its next
-  step and runs its compensations; a waiting run is woken to do so (within the
-  timeout worker's one-minute sweep). It can't interrupt a step that's running.
+  sleeps included. A run past it fails with a `"timeout"` error and runs its
+  compensations: a step still running is stopped, and a waiting run is woken
+  to fail (within the timeout worker's one-minute sweep).
 - `:queue` — the queue runs start on unless `Durable.start/3` names one.
 
 Unknown options fail the compile.
diff --git a/durable/guides/waiting.md b/durable/guides/waiting.md
index 62f7a9f..6cdfc8e 100644
--- a/durable/guides/waiting.md
+++ b/durable/guides/waiting.md
@@ -252,6 +252,7 @@ end
 | `:metadata` | Additional data for UI |
 | `:timeout` | Max wait time |
 | `:timeout_value` | Value on timeout |
+| `:on_timeout` | `:resume` (default) returns `:timeout_value`; `:fail` fails the run and runs its compensations |
 | `:fields` | Field definitions for forms |
 | `:choices` | Options for choice inputs |
 
diff --git a/durable/lib/durable/dsl/step.ex b/durable/lib/durable/dsl/step.ex
index ed2fd91..b58705c 100644
--- a/durable/lib/durable/dsl/step.ex
+++ b/durable/lib/durable/dsl/step.ex
@@ -37,9 +37,13 @@ defmodule Durable.DSL.Step do
   - `:retry` - Retry configuration
     - `:max_attempts` - Maximum retry attempts (default: 1)
     - `:backoff` - Backoff strategy: `:exponential`, `:linear`, `:constant`
-    - `:base` - Base for backoff calculation (default: 2)
-    - `:max_backoff` - Maximum backoff in ms (default: 3600000)
-  - `:timeout` - Step timeout in milliseconds
+    - `:base` - Base delay in ms (default: 1000)
+    - `:max_backoff` - Maximum backoff in ms (default: 30000)
+  - `:timeout` - How long one attempt may run, in milliseconds. A slower
+    attempt is stopped and fails with a `"timeout"` error, which is retried
+    like any error. A step with a timeout, or in a workflow with one, runs in
+    a separate process: its body sees the same context and captured logs,
+    but `self()` is not the worker.
   - `:compensate` - Name of compensation function for saga pattern
   - `:queue` - Override queue for this step
 
@@ -89,7 +93,7 @@ defmodule Durable.DSL.Step do
         type: :step,
         module: __MODULE__,
         body_fn: &(__MODULE__.unquote(func_name) / 1),
-        opts: unquote(Macro.escape(normalized_opts))
+        opts: unquote(quote_opts(normalized_opts))
       }
     end
   end
@@ -113,6 +117,14 @@ defmodule Durable.DSL.Step do
 
   defp normalize_retry_opts(opts), do: opts
 
+  # Option values are code: splice them in rather than escape them, so
+  # `timeout: seconds(30)` is a number in the workflow module, not the quoted
+  # call.
+  defp quote_opts(opts) when is_map(opts),
+    do: {:%{}, [], Enum.map(opts, fn {key, value} -> {key, quote_opts(value)} end)}
+
+  defp quote_opts(value), do: value
+
   @doc """
   Defines a decision step for conditional branching.
 
@@ -165,7 +177,7 @@ defmodule Durable.DSL.Step do
         type: :decision,
         module: __MODULE__,
         body_fn: &(__MODULE__.unquote(func_name) / 1),
-        opts: unquote(Macro.escape(normalized_opts))
+        opts: unquote(quote_opts(normalized_opts))
       }
     end
   end
@@ -227,7 +239,7 @@ defmodule Durable.DSL.Step do
         name: unquote(name),
         module: __MODULE__,
         body_fn: &(__MODULE__.unquote(func_name) / 1),
-        opts: unquote(Macro.escape(normalized_opts))
+        opts: unquote(quote_opts(normalized_opts))
       }
     end
   end
@@ -299,7 +311,7 @@ defmodule Durable.DSL.Step do
             type: :step,
             module: __MODULE__,
             body_fn: &(__MODULE__.unquote(func_name) / 1),
-            opts: unquote(Macro.escape(step_opts))
+            opts: unquote(quote_opts(step_opts))
           }
         end
       end)
@@ -515,7 +527,7 @@ defmodule Durable.DSL.Step do
             type: :step,
             module: __MODULE__,
             body_fn: &(__MODULE__.unquote(func_name) / 1),
-            opts: unquote(Macro.escape(step_opts))
+            opts: unquote(quote_opts(step_opts))
           }
         end
       end)
diff --git a/durable/lib/durable/dsl/workflow.ex b/durable/lib/durable/dsl/workflow.ex
index b067911..eaba5a6 100644
--- a/durable/lib/durable/dsl/workflow.ex
+++ b/durable/lib/durable/dsl/workflow.ex
@@ -22,9 +22,10 @@ defmodule Durable.DSL.Workflow do
 
   - `:timeout` - How long a run may take, in milliseconds, from when it is due
     to start — waits and sleeps included. A run past it fails with a
-    `"timeout"` error before its next step, running compensations like any
-    failure; a waiting run is woken to do so. The deadline is checked between
-    steps, so it can't interrupt a step that is still running.
+    `"timeout"` error, running compensations like any failure: a step still
+    running is stopped (and not retried), and a waiting run is woken to fail.
+    Steps in a run with a timeout run in their own process; see the step
+    `:timeout` option.
   - `:queue` - The queue runs start on unless `Durable.start/3` names one
 
   """
diff --git a/durable/lib/durable/executor.ex b/durable/lib/durable/executor.ex
index 60d6128..c9a66b6 100644
--- a/durable/lib/durable/executor.ex
+++ b/durable/lib/durable/executor.ex
@@ -482,12 +482,8 @@ defmodule Durable.Executor do
 
   defp past_deadline?(_execution), do: false
 
-  # The deadline passed before `step` began. Fail the run the way a failed step
-  # would — so its compensations run — and stop any child runs it waited on.
+  # The deadline passed before `step` began.
   defp time_out(execution, step, workflow_def, config) do
-    cancel_child_workflows(config, execution.id)
-    {:ok, exec} = update_current_step(config, execution, step.name)
-
     error = %{
       type: "timeout",
       message:
@@ -495,10 +491,34 @@ defmodule Durable.Executor do
           "before #{step.name}"
     }
 
+    fail_at(execution, step, error, workflow_def, config)
+  end
+
+  # The TimeoutWorker resumes a run whose `on_timeout: :fail` wait timed out
+  # with the error under `__fail__`, so the run fails here, in the executor,
+  # rather than being written off from outside.
+  defp resume_failure(%{context: %{"__fail__" => error}}), do: error
+  defp resume_failure(_execution), do: nil
+
+  # Fail the run at `step` the way a failed step would — so its compensations
+  # run — and stop any child runs it waited on.
+  defp fail_at(execution, step, error, workflow_def, config) do
+    cancel_child_workflows(config, execution.id)
+
+    {:ok, exec} =
+      execution
+      |> Ecto.Changeset.change(
+        current_step: Atom.to_string(step.name),
+        # A retried run must not fail again on the same marker.
+        context: Map.delete(execution.context || %{}, "__fail__")
+      )
+      |> cas_update!(config)
+
     handle_step_failure(exec, error, workflow_def, config)
   end
 
-  defp format_timeout(ms) when is_integer(ms) do
+  @doc false
+  def format_timeout(ms) when is_integer(ms) do
     Enum.find_value(
       [{86_400_000, "d"}, {3_600_000, "h"}, {60_000, "m"}, {1_000, "s"}],
       "#{ms}ms",
@@ -506,22 +526,32 @@ defmodule Durable.Executor do
     )
   end
 
-  defp format_timeout(_ms), do: "workflow"
+  def format_timeout(_ms), do: "workflow"
 
   defp execute_steps(steps, execution, config, initial_data) do
     # Get workflow definition for compensation lookup
     {:ok, workflow_def} = get_workflow_definition_from_execution(execution)
 
-    # Find the starting point (for resumed workflows)
+    # A resumed run picks up at its current step, found by name.
     {_already_completed, steps_to_run} =
-      if execution.current_step do
-        # Find and skip to current step
-        current_step_atom = String.to_existing_atom(execution.current_step)
-        Enum.split_while(steps, fn step -> step.name != current_step_atom end)
-      else
-        {[], steps}
-      end
+      Enum.split_while(steps, &(execution.current_step not in [nil, Atom.to_string(&1.name)]))
 
+    if steps_to_run == [] and execution.current_step != nil do
+      # The deployed definition no longer has the step this run stopped at.
+      # Running on from a guess would skip work, so fail, without
+      # compensating: restore the step and `Durable.retry/2` the run.
+      mark_failed(config, execution, %{
+        type: "definition_changed",
+        message:
+          "#{execution.workflow_name} has no step #{execution.current_step} any more; " <>
+            "this run stopped there before the workflow changed"
+      })
+    else
+      run_steps(steps, steps_to_run, execution, workflow_def, config, initial_data)
+    end
+  end
+
+  defp run_steps(steps, steps_to_run, execution, workflow_def, config, initial_data) do
     # Build step index for validation
     step_index = build_step_index(steps)
 
@@ -564,10 +594,15 @@ defmodule Durable.Executor do
          config,
          data
        ) do
-    if past_deadline?(execution) do
-      time_out(execution, step, workflow_def, config)
-    else
-      dispatch_step(step, remaining_steps, execution, step_index, workflow_def, config, data)
+    cond do
+      past_deadline?(execution) ->
+        time_out(execution, step, workflow_def, config)
+
+      error = resume_failure(execution) ->
+        fail_at(execution, step, error, workflow_def, config)
+
+      true ->
+        dispatch_step(step, remaining_steps, execution, step_index, workflow_def, config, data)
     end
   end
 
@@ -604,7 +639,7 @@ defmodule Durable.Executor do
     # Update current step
     {:ok, exec} = update_current_step(config, execution, step.name)
 
-    case StepRunner.execute(step, data, exec.id, config) do
+    case StepRunner.execute(step, data, exec.id, config, exec.deadline_at) do
       {:ok, new_data} ->
         # Save data as context and continue to next step
         # save_data_as_context merges orchestration keys from process dict
@@ -864,7 +899,7 @@ defmodule Durable.Executor do
        ) do
     {:ok, exec} = update_current_step(config, execution, step.name)
 
-    case StepRunner.execute(step, data, exec.id, config) do
+    case StepRunner.execute(step, data, exec.id, config, exec.deadline_at) do
       {:ok, new_data} ->
         {:ok, exec} = save_data_as_context(config, exec, new_data)
 
@@ -1051,7 +1086,9 @@ defmodule Durable.Executor do
       context: Map.put(parent_context, "__parallel_step", Atom.to_string(step.name)),
       parent_workflow_id: parent_exec.id,
       trigger: :parent,
-      current_step: Atom.to_string(step.name)
+      current_step: Atom.to_string(step.name),
+      # A branch is part of the run, so the run's deadline stops it too.
+      deadline_at: parent_exec.deadline_at
     }
 
     %WorkflowExecution{}
@@ -1091,7 +1128,7 @@ defmodule Durable.Executor do
       input = atomize_keys(execution.input)
       data = Map.merge(inherited, input)
 
-      case StepRunner.execute(step_def, data, execution.id, config) do
+      case StepRunner.execute(step_def, data, execution.id, config, execution.deadline_at) do
         {:ok, output_data} ->
           mark_completed(config, execution, output_data)
 
diff --git a/durable/lib/durable/executor/step_runner.ex b/durable/lib/durable/executor/step_runner.ex
index 4edd20d..03f508e 100644
--- a/durable/lib/durable/executor/step_runner.ex
+++ b/durable/lib/durable/executor/step_runner.ex
@@ -39,14 +39,16 @@ defmodule Durable.Executor.StepRunner do
   - `data` - The data to pass to the step (from previous step or workflow input)
   - `workflow_id` - The workflow execution ID
   - `config` - The Durable config
+  - `deadline` - When the run's `timeout:` runs out, if it has one. A step
+    still running then is stopped, and isn't retried.
 
   Returns `{:ok, data}` on success or `{:error, reason}` after all retries exhausted.
   """
-  @spec execute(Step.t(), map(), String.t(), Config.t()) :: result()
-  def execute(%Step{} = step, data, workflow_id, %Config{} = config) do
+  @spec execute(Step.t(), map(), String.t(), Config.t(), DateTime.t() | nil) :: result()
+  def execute(%Step{} = step, data, workflow_id, %Config{} = config, deadline \\ nil) do
     max_attempts = get_max_attempts(step)
     start_attempt = next_attempt(config, workflow_id, step)
-    execute_with_retry(step, data, workflow_id, start_attempt, max_attempts, config)
+    execute_with_retry(step, data, workflow_id, start_attempt, max_attempts, config, deadline)
   end
 
   # Seed the retry counter from durable state instead of always starting at 1.
@@ -82,7 +84,7 @@ defmodule Durable.Executor.StepRunner do
     (failed_count || 0) + 1
   end
 
-  defp execute_with_retry(step, data, workflow_id, attempt, max_attempts, config) do
+  defp execute_with_retry(step, data, workflow_id, attempt, max_attempts, config, deadline) do
     # Set current step for logging/observability
     Context.set_current_step(step.name)
 
@@ -107,7 +109,7 @@ defmodule Durable.Executor.StepRunner do
 
     result =
       :telemetry.span([:durable, :step], span_meta, fn ->
-        result = run_body(step, data)
+        result = run_body(step, data, time_limit(step, deadline))
         {result, Map.merge(span_meta, step_outcome(result))}
       end)
 
@@ -126,6 +128,7 @@ defmodule Durable.Executor.StepRunner do
       duration_ms: duration_ms,
       attempt: attempt,
       max_attempts: max_attempts,
+      deadline: deadline,
       config: config
     }
 
@@ -139,6 +142,70 @@ defmodule Durable.Executor.StepRunner do
     kind, reason -> {:error, Durable.Executor.caught_error(kind, reason, __STACKTRACE__)}
   end
 
+  # The step's own `timeout:` or what's left until the run's deadline,
+  # whichever is shorter; nil when neither applies.
+  defp time_limit(step, deadline) do
+    left = deadline && max(DateTime.diff(deadline, DateTime.utc_now(), :millisecond), 0)
+
+    [{step.opts[:timeout], :step}, {left, :run}]
+    |> Enum.filter(&elem(&1, 0))
+    |> Enum.min_by(&elem(&1, 0), fn -> nil end)
+  end
+
+  defp past?(nil), do: false
+  defp past?(deadline), do: DateTime.compare(DateTime.utc_now(), deadline) != :lt
+
+  # Keys Task sets for itself; everything else in the dictionary is copied.
+  @task_keys [:"$ancestors", :"$callers", :"$initial_call"]
+
+  defp run_body(step, data, nil), do: run_body(step, data)
+  defp run_body(step, _data, {0, limit}), do: {:error, timeout_error(step, limit)}
+
+  # A body with a time limit runs in a task that is killed when the limit
+  # passes. The task starts with a copy of this process's dictionary (context,
+  # log capture, telemetry metadata) and hands its own back, so the body sees
+  # no difference beyond `self()`.
+  defp run_body(step, data, {ms, limit}) do
+    dict = Process.get()
+
+    task =
+      Task.async(fn ->
+        put_dict(dict)
+        {run_body(step, data), Process.get()}
+      end)
+
+    case Task.yield(task, ms) || stop(task) do
+      {:ok, {result, dict}} ->
+        put_dict(dict)
+        result
+
+      {:stopped, dict} ->
+        put_dict(dict)
+        {:error, timeout_error(step, limit)}
+    end
+  end
+
+  # Keeps what the body logged before it was stopped.
+  defp stop(task) do
+    {:dictionary, dict} = Process.info(task.pid, :dictionary) || {:dictionary, []}
+    Task.shutdown(task, :brutal_kill) || {:stopped, dict}
+  end
+
+  defp put_dict(dict),
+    do: for({key, value} <- dict, key not in @task_keys, do: Process.put(key, value))
+
+  defp timeout_error(step, :step) do
+    %{
+      type: "timeout",
+      message:
+        "#{step.name} took longer than its " <>
+          "#{Durable.Executor.format_timeout(step.opts[:timeout])} timeout"
+    }
+  end
+
+  defp timeout_error(step, :run),
+    do: %{type: "timeout", message: "the run passed its timeout during #{step.name}"}
+
   defp step_span_meta(config, workflow_id, step, attempt, max_attempts) do
     %{
       durable: config.name,
@@ -234,13 +301,14 @@ defmodule Durable.Executor.StepRunner do
       duration_ms: duration_ms,
       attempt: attempt,
       max_attempts: max_attempts,
+      deadline: deadline,
       config: config
     } = ctx
 
     # Normalize error to map format for database storage
     normalized_error = normalize_error_for_storage(error)
 
-    if attempt < max_attempts do
+    if attempt < max_attempts and not past?(deadline) do
       {:ok, _} = fail_step_execution(config, step_exec, normalized_error, logs, duration_ms)
 
       retry_opts = get_retry_opts(step)
@@ -257,7 +325,15 @@ defmodule Durable.Executor.StepRunner do
 
       Backoff.pause(backoff_ms)
 
-      execute_with_retry(step, data, step_exec.workflow_id, attempt + 1, max_attempts, config)
+      execute_with_retry(
+        step,
+        data,
+        step_exec.workflow_id,
+        attempt + 1,
+        max_attempts,
+        config,
+        deadline
+      )
     else
       {:ok, _} = fail_step_execution(config, step_exec, normalized_error, logs, duration_ms)
       {:error, error}
diff --git a/durable/lib/durable/wait.ex b/durable/lib/durable/wait.ex
index 0b1804e..59d4ea0 100644
--- a/durable/lib/durable/wait.ex
+++ b/durable/lib/durable/wait.ex
@@ -321,7 +321,8 @@ defmodule Durable.Wait do
   - `:choices` - Choice options for choice input
   - `:timeout` - Timeout in milliseconds
   - `:timeout_value` - Value to return on timeout
-  - `:on_timeout` - `:resume` (default) or `:fail`
+  - `:on_timeout` - `:resume` (default) returns `:timeout_value`; `:fail`
+    fails the run with a `"timeout"` error, running its compensations
 
   ## Examples
 
diff --git a/durable/lib/durable/wait/timeout_worker.ex b/durable/lib/durable/wait/timeout_worker.ex
index c31ba06..e437291 100644
--- a/durable/lib/durable/wait/timeout_worker.ex
+++ b/durable/lib/durable/wait/timeout_worker.ex
@@ -150,62 +150,46 @@ defmodule Durable.Wait.TimeoutWorker do
   defp handle_input_timeout(config, pending_input) do
     on_timeout = pending_input.on_timeout || :resume
 
-    case on_timeout do
-      :resume ->
-        timeout_value = deserialize_timeout_value(pending_input.timeout_value)
-
-        resume_data = %{
-          pending_input.input_name => timeout_value,
-          :__timeout__ => true
-        }
-
-        # Atomic: mark the input :timeout AND transition the workflow back
-        # to :pending in a single transaction. Before this change, a crash
-        # between the two updates left the input as :timeout but the workflow
-        # stuck in :waiting forever (Bug M-2).
-        result =
-          atomic_resume_after_timeout(
-            config,
-            PendingInput.timeout_changeset(pending_input),
-            pending_input.workflow_id,
-            resume_data
-          )
-
-        case result do
-          {:ok, _} ->
-            Logger.info(
-              "Timeout handled for pending input #{pending_input.input_name} " <>
-                "in workflow #{pending_input.workflow_id} (resume)"
-            )
+    # Atomic: mark the input :timeout AND transition the workflow back to
+    # :pending in a single transaction. Before this change, a crash between
+    # the two updates left the input as :timeout but the workflow stuck in
+    # :waiting forever (Bug M-2).
+    case atomic_resume_after_timeout(
+           config,
+           PendingInput.timeout_changeset(pending_input),
+           pending_input.workflow_id,
+           input_timeout_data(on_timeout, pending_input)
+         ) do
+      {:ok, _} ->
+        Logger.info(
+          "Timeout handled for pending input #{pending_input.input_name} " <>
+            "in workflow #{pending_input.workflow_id} (#{on_timeout})"
+        )
 
-          {:error, stage, reason, _changes} ->
-            Logger.error(
-              "Failed input timeout transaction for #{pending_input.workflow_id}: " <>
-                "#{stage} → #{inspect(reason)}"
-            )
-        end
+      {:error, stage, reason, _changes} ->
+        Logger.error(
+          "Failed input timeout transaction for #{pending_input.workflow_id}: " <>
+            "#{stage} → #{inspect(reason)}"
+        )
+    end
+  end
 
-      :fail ->
-        # Mark input :timeout then cancel — same Multi pattern.
-        case atomic_cancel_after_timeout(
-               config,
-               PendingInput.timeout_changeset(pending_input),
-               pending_input.workflow_id,
-               "Timeout waiting for input: #{pending_input.input_name}"
-             ) do
-          {:ok, _} ->
-            Logger.info(
-              "Timeout handled for pending input #{pending_input.input_name} " <>
-                "in workflow #{pending_input.workflow_id} (fail)"
-            )
+  defp input_timeout_data(:resume, pending_input) do
+    %{
+      pending_input.input_name => deserialize_timeout_value(pending_input.timeout_value),
+      :__timeout__ => true
+    }
+  end
 
-          {:error, stage, reason, _changes} ->
-            Logger.error(
-              "Failed input timeout transaction for #{pending_input.workflow_id}: " <>
-                "#{stage} → #{inspect(reason)}"
-            )
-        end
-    end
+  # The executor fails the run at the waiting step, so compensations run and a
+  # parent run hears about it.
+  defp input_timeout_data(:fail, pending_input) do
+    %{
+      "__fail__" => %{
+        "type" => "timeout",
+        "message" => "Timeout waiting for input: #{pending_input.input_name}"
+      }
+    }
   end
 
   defp process_timed_out_events(config, now) do
@@ -319,29 +303,6 @@ defmodule Durable.Wait.TimeoutWorker do
     Repo.transaction(config, multi)
   end
 
-  defp atomic_cancel_after_timeout(config, pending_changeset, workflow_id, reason) do
-    multi =
-      Ecto.Multi.new()
-      |> Ecto.Multi.update(:pending, pending_changeset)
-      |> Ecto.Multi.run(:workflow, fn repo, _changes ->
-        case repo.get(WorkflowExecution, workflow_id) do
-          nil ->
-            {:error, :workflow_not_found}
-
-          %WorkflowExecution{} = exec ->
-            exec
-            |> WorkflowExecution.status_changeset(:cancelled, %{
-              error: %{"type" => "timeout", "message" => reason},
-              completed_at: DateTime.utc_now()
-            })
-            |> Ecto.Changeset.change(locked_by: nil, locked_at: nil)
-            |> repo.update()
-        end
-      end)
-
-    Repo.transaction(config, multi)
-  end
-
   defp process_timed_out_wait_groups(config, now) do
     # Find wait groups that have timed out
     query =
diff --git a/durable/test/durable/definition_change_test.exs b/durable/test/durable/definition_change_test.exs
new file mode 100644
index 0000000..89b90fb
--- /dev/null
+++ b/durable/test/durable/definition_change_test.exs
@@ -0,0 +1,70 @@
+defmodule Durable.DefinitionChangeTest do
+  @moduledoc """
+  A deploy can change a workflow under runs that are still in flight.
+  """
+  use Durable.DataCase, async: false
+
+  import Durable.Testing
+
+  alias Durable.Config
+  alias Durable.Storage.Schemas.WorkflowExecution
+
+  defmodule V1 do
+    use Durable
+    use Durable.Helpers
+    use Durable.Wait
+
+    workflow "order" do
+      step(:reserve, fn data -> {:ok, data} end)
+
+      step(:approve, fn data ->
+        wait_for_input("ok")
+        {:ok, data}
+      end)
+
+      step(:ship, fn data -> {:ok, assign(data, :shipped, true)} end)
+    end
+  end
+
+  # The next deploy renames :approve.
+  defmodule V2 do
+    use Durable
+    use Durable.Helpers
+    use Durable.Wait
+
+    workflow "order" do
+      step(:reserve, fn data -> {:ok, data} end)
+
+      step(:review, fn data ->
+        wait_for_input("ok")
+        {:ok, data}
+      end)
+
+      step(:ship, fn data -> {:ok, assign(data, :shipped, true)} end)
+    end
+  end
+
+  defp deploy(run, module) do
+    config = Config.get(Durable)
+
+    config.repo.update_all(from(w in WorkflowExecution, where: w.id == ^run.id),
+      set: [workflow_module: Atom.to_string(module)]
+    )
+  end
+
+  test "a run whose step was renamed fails instead of skipping to the end" do
+    run = run_workflow(V1)
+    deploy(run, V2)
+    :ok = Durable.provide_input(run.id, "ok", %{})
+
+    run = run |> drain() |> assert_status(:failed)
+    assert run.error["type"] == "definition_changed"
+    assert run.error["message"] =~ "no step approve"
+
+    # Putting the step back and retrying carries on from it.
+    deploy(run, V1)
+    {:ok, _} = Durable.retry(run.id)
+    run = run |> drain() |> assert_status(:completed)
+    assert run.context["shipped"]
+  end
+end
diff --git a/durable/test/durable/resume_edge_cases_test.exs b/durable/test/durable/resume_edge_cases_test.exs
index 54f7787..0df6e05 100644
--- a/durable/test/durable/resume_edge_cases_test.exs
+++ b/durable/test/durable/resume_edge_cases_test.exs
@@ -18,11 +18,7 @@ defmodule Durable.ResumeEdgeCasesTest do
   # ============================================================================
 
   describe "resume with corrupted state" do
-    # Note: Resume with non-existent current_step atom causes ArgumentError
-    # This is documented behavior - binary_to_existing_atom fails
-    # The test below documents this edge case
-
-    test "resume with non-existent current_step raises ArgumentError" do
+    test "resume with non-existent current_step fails the run" do
       config = Config.get(Durable)
       repo = config.repo
       {:ok, workflow_def} = ResumableWorkflow.__default_workflow__()
@@ -44,10 +40,10 @@ defmodule Durable.ResumeEdgeCasesTest do
         |> WorkflowExecution.changeset(attrs)
         |> repo.insert()
 
-      # Execute - should raise ArgumentError due to binary_to_existing_atom
-      assert_raise ArgumentError, fn ->
-        Executor.execute_workflow(execution.id, config)
-      end
+      Executor.execute_workflow(execution.id, config)
+
+      assert %{status: :failed, error: %{"type" => "definition_changed"}} =
+               repo.get!(WorkflowExecution, execution.id)
     end
 
     test "resume with empty context but populated step history" do
diff --git a/durable/test/durable/workflow_timeout_test.exs b/durable/test/durable/workflow_timeout_test.exs
index 82fed86..02ee2ef 100644
--- a/durable/test/durable/workflow_timeout_test.exs
+++ b/durable/test/durable/workflow_timeout_test.exs
@@ -6,7 +6,7 @@ defmodule Durable.WorkflowTimeoutTest do
 
   alias Durable.Config
   alias Durable.Executor
-  alias Durable.Storage.Schemas.{PendingInput, WorkflowExecution}
+  alias Durable.Storage.Schemas.{PendingInput, StepExecution, WorkflowExecution}
   alias Durable.Wait.TimeoutWorker
 
   defmodule SlowSaga do
@@ -14,12 +14,12 @@ defmodule Durable.WorkflowTimeoutTest do
     use Durable.Helpers
 
     workflow "slow_saga", timeout: 50, queue: :sagas do
-      step(:reserve, [compensate: :release], fn data ->
-        Process.sleep(80)
-        {:ok, assign(data, :reserved, true)}
-      end)
+      step(:reserve, [compensate: :release], fn data -> {:ok, assign(data, :reserved, true)} end)
 
-      step(:charge, fn data -> {:ok, assign(data, :charged, true)} end)
+      step(:charge, fn data ->
+        Process.sleep(1_000)
+        {:ok, assign(data, :charged, true)}
+      end)
 
       compensate(:release, fn data -> {:ok, assign(data, :released, true)} end)
     end
@@ -38,6 +38,41 @@ defmodule Durable.WorkflowTimeoutTest do
     end
   end
 
+  defmodule Timed do
+    use Durable
+    use Durable.Context
+    use Durable.Helpers
+
+    require Logger
+
+    workflow "timed", timeout: minutes(1) do
+      step(:remember, fn data ->
+        put_context(:note, "kept")
+        Logger.info("remembering")
+        {:ok, data}
+      end)
+
+      step(:fetch, [timeout: 20, retry: [max_attempts: 2, base: 1]], fn data ->
+        calls = Process.get(:fetch_calls, 0) + 1
+        Process.put(:fetch_calls, calls)
+        if calls == 1, do: Process.sleep(1_000)
+        {:ok, assign(data, :note, get_context(:note))}
+      end)
+    end
+  end
+
+  defmodule Ask do
+    use Durable
+    use Durable.Wait
+
+    workflow "ask" do
+      step(:ask, fn data ->
+        wait_for_input("ok", timeout: hours(1), on_timeout: :fail)
+        {:ok, data}
+      end)
+    end
+  end
+
   defp repo, do: Config.get(Durable).repo
   defp run(id), do: repo().get!(WorkflowExecution, id)
 
@@ -59,17 +94,73 @@ defmodule Durable.WorkflowTimeoutTest do
     end
   end
 
-  test "a run past its deadline fails before the next step and compensates" do
-    {:ok, id} = Executor.start_workflow(SlowSaga, %{}, inline: true)
+  test "step options are evaluated too: timeout: seconds(1) is a number" do
+    defmodule Opts do
+      use Durable
+
+      workflow "opts" do
+        step(:a, [timeout: seconds(1), retry: [base: seconds(2)]], fn data -> {:ok, data} end)
+      end
+    end
+
+    {:ok, %{steps: [step]}} = Opts.__workflow_definition__("opts")
+    assert step.opts == %{timeout: 1_000, retry: %{base: 2_000}}
+  end
+
+  test "the deadline stops a step that is still running, then compensates" do
+    {micros, {:ok, id}} =
+      :timer.tc(fn -> Executor.start_workflow(SlowSaga, %{}, inline: true) end)
+
     run = run(id)
 
+    assert micros < 500_000
     assert run.queue == "sagas"
     assert run.status == :compensated
-    assert %{"type" => "timeout", "message" => message} = run.error
-    assert message =~ "50ms timeout before charge"
+
+    assert run.error == %{
+             "type" => "timeout",
+             "message" => "the run passed its timeout during charge"
+           }
+
     refute run.context["charged"]
   end
 
+  test "a step past its own timeout is stopped and retried" do
+    {:ok, id} = Executor.start_workflow(Timed, %{}, inline: true)
+    run = run(id)
+
+    assert run.status == :completed
+    # put_context and logs inside a time-limited step reach the run
+    assert run.context["note"] == "kept"
+
+    [remember, first, second] =
+      repo().all(from(s in StepExecution, where: s.workflow_id == ^id, order_by: s.inserted_at))
+
+    assert [%{"message" => "remembering"}] = remember.logs
+    assert first.error["message"] == "fetch took longer than its 20ms timeout"
+    assert {first.attempt, second.attempt, second.status} == {1, 2, :completed}
+  end
+
+  test "an input wait with on_timeout: :fail fails the run, and a retry waits again" do
+    {:ok, id} = Executor.start_workflow(Ask, %{}, inline: true)
+    past = DateTime.add(DateTime.utc_now(), -1, :second)
+
+    repo().update_all(from(p in PendingInput, where: p.workflow_id == ^id),
+      set: [timeout_at: past]
+    )
+
+    start_supervised!({TimeoutWorker, config: Config.get(Durable), interval: 3_600_000})
+    TimeoutWorker.check_timeouts(Durable)
+    :sys.get_state(TimeoutWorker.worker_name(Durable))
+    Executor.execute_workflow(id, Config.get(Durable))
+
+    assert %{status: :failed, error: %{"message" => "Timeout waiting for input: ok"}} = run(id)
+
+    {:ok, ^id} = Durable.retry(id)
+    Executor.execute_workflow(id, Config.get(Durable))
+    assert %{status: :waiting} = run(id)
+  end
+
   test "a waiting run past its deadline is woken and failed" do
     {:ok, id} = Executor.start_workflow(Approval, %{}, inline: true)
     assert %{status: :waiting, deadline_at: %DateTime{}} = run(id)
diff --git a/durable_dashboard/.formatter.exs b/durable_dashboard/.formatter.exs
index 4872046..5c5f878 100644
--- a/durable_dashboard/.formatter.exs
+++ b/durable_dashboard/.formatter.exs
@@ -1,5 +1,5 @@
 [
-  import_deps: [:ecto, :ecto_sql, :phoenix, :phoenix_live_view],
+  import_deps: [:ecto, :ecto_sql, :phoenix, :phoenix_live_view, :durable],
   subdirectories: ["priv/*/migrations"],
   inputs: ["*.{ex,exs}", "{config,lib,test}/**/*.{ex,exs}"]
 ]
diff --git a/examples/phoenix_demo/.formatter.exs b/examples/phoenix_demo/.formatter.exs
index ef8840c..2835a6b 100644
--- a/examples/phoenix_demo/.formatter.exs
+++ b/examples/phoenix_demo/.formatter.exs
@@ -1,5 +1,5 @@
 [
-  import_deps: [:ecto, :ecto_sql, :phoenix],
+  import_deps: [:ecto, :ecto_sql, :phoenix, :durable],
   subdirectories: ["priv/*/migrations"],
   plugins: [Phoenix.LiveView.HTMLFormatter],
   inputs: ["*.{heex,ex,exs}", "{config,lib,test}/**/*.{heex,ex,exs}", "priv/*/seeds.exs"]