diff --git a/CHANGELOG.md b/CHANGELOG.md
index ffc04d6..79b2e51 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -8,6 +8,9 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). The
 
 ### Added
 
+- **State migration for checkpointed graphs (proposal 0014, introduced in spec v0.15.0; refined by proposal 0018 in spec v0.16.0).** Saved checkpoints whose `schema_version` doesn't match the current state class now route through a registered migration chain instead of failing on resume. Surface: `State.schema_version: ClassVar[str] = ""` (declare a non-empty value to opt in), `GraphBuilder.with_state_migration(from_version, to_version, migrate)` and `with_state_migrations(*migrations)` for registration, `StateMigration` and `MigrationRegistry` types exported from `openarmature.checkpoint`. Chain resolution is BFS over the registered edges; the shortest path wins. Three new error categories: `CheckpointStateMigrationChainAmbiguous` (proposal 0018: duplicate `(from, to)` pair at registration time, or multiple distinct shortest paths between the saved and current versions at resume time), `CheckpointStateMigrationMissing` (no chain bridges the versions), and `CheckpointStateMigrationFailed` (a migration function raised). All non-transient. Post-migration deserialization failures still route to `CheckpointRecordInvalid` per §10.12.4. The same chain applies to each entry in `parent_states` in lockstep with the outer state per §10.12.2. Routing precedence per §10.10 (v0.16.0): chain-ambiguous → missing → failed → record-invalid.
+- **`Checkpointer.supports_state_migration` Protocol attribute.** Marks whether a backend can expose the structural intermediate form (a plain dict, JSON tree) the migration registry consumes. `SQLiteCheckpointer(serialization="json")` opts in; `SQLiteCheckpointer(serialization="pickle")` and `InMemoryCheckpointer` opt out. On version mismatch against a non-migration-eligible backend the engine raises `CheckpointRecordInvalid` per spec §10.12.1.
+- **`openarmature.checkpoint.migrate` OTel span (proposal 0014 §6 cross-ref).** Versioned resumes whose migration chain runs emit a zero-duration `openarmature.checkpoint.migrate` span on the OTel observer, parented under the invocation root span. Attributes: `openarmature.checkpoint.migrate.from_version`, `openarmature.checkpoint.migrate.to_version` (the final target), `openarmature.checkpoint.migrate.chain_length`. The §10.12.3 fast path (versions match, registry not consulted) emits no span. Engine-side: a synthetic `checkpoint_migrated` observer phase carries a `_MigrationSummary` payload from `_migrate_record` through to the OTel observer; the new phase is gated off default subscriptions (observers opt in explicitly via `phases={..., "checkpoint_migrated"}`).
 - **Prompt-management capability (proposal 0017, introduced in spec v0.15.0).** New `openarmature.prompts` subpackage. `PromptManager` composes one or more `PromptBackend`s, exposes `fetch` / `render` / `get`, applies the §8 fallback semantics (`prompt_store_unavailable` continues to the next backend; `prompt_not_found` stops the chain), and renders templates with Jinja2's `StrictUndefined` per §7. `Prompt` / `PromptResult` / `PromptGroup` are Pydantic models matching spec §3 / §4 / §9. Three error categories (`PromptNotFound`, `PromptRenderError`, `PromptStoreUnavailable`) with `PROMPT_TRANSIENT_CATEGORIES` exported for retry-middleware classifiers. `FilesystemPromptBackend` is the minimum local-filesystem reference backend (layout: `<root>/<label>/<name>.j2`; `version` derived from the first 16 hex chars of `template_hash`). New runtime dependency: `jinja2>=3.1`.
 - **`openarmature.prompts.context` — observability propagation per spec §11.** `with_active_prompt(result)` and `with_active_prompt_group(group)` context managers + `current_prompt_result()` / `current_prompt_group()` inspectors. When the OTel observer is active and an LLM call fires inside `with_active_prompt`, the `openarmature.llm.complete` span carries the normative `openarmature.prompt.*` attributes (`name`, `version`, `label`, `template_hash`, `rendered_hash`, `group_name`). Nesting is innermost-wins.
 - **Image content blocks for user messages (proposal 0015, introduced in spec v0.13.0).** `UserMessage.content` now accepts `str | list[ContentBlock]`. The block surface introduces `TextBlock`, `ImageBlock`, `ImageSourceURL`, `ImageSourceInline`, and the `ContentBlock` / `ImageSource` discriminated unions over the block / source `type` field. `ImageBlock` carries a `media_type` (required for inline sources; ignored for URL sources; typed as `str | None` so callers MAY pass any `image/*` type the bound model supports) and an optional `detail` hint (`"auto"` / `"low"` / `"high"`; `None` default omits the field from the wire so providers apply their own default). System, assistant, and tool messages stay text-string-only; image inputs are user-only in v1.
@@ -22,7 +25,9 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). The
 
 ### Changed
 
-- **Pinned spec version: 0.10.0 → 0.15.0.** Adopts the skip-ahead governance principle: the submodule jumps across v0.11.0–v0.15.0 (proposals 0009, 0011, 0014, 0015, 0016, 0017) in one bump. Only the surface introduced by proposal 0016 is implemented in this changelog entry; fixtures from 0011 / 0014 / 0015 / 0017 are marked deferred-skip in the conformance suite and unmark as their respective PRs land.
+- **Pinned spec version: 0.10.0 → 0.16.0.** Adopts the skip-ahead governance principle: the submodule jumps across v0.11.0–v0.16.0 (proposals 0009, 0011, 0014, 0015, 0016, 0017, 0018) in one bump. Only the surfaces introduced by proposals 0014–0017 are implemented in the batch's release; fixtures from 0011 are deferred-skip in the conformance suite and unmark with PR-5.
+- **`CheckpointRecord.schema_version` semantic shift (proposal 0014).** Previously a backend-internal record-shape version (`CHECKPOINT_SCHEMA_VERSION = "1"` constant), now the user-facing state-schema version per spec §10.2. The framework reads `type(state).schema_version` at save time. Pre-PR-4 records carrying `"1"` are reinterpreted as user-facing v1 identifiers; users with such records either declare `schema_version="1"` on their state class or discard the pre-PR-4 records. `SQLiteCheckpointer` no longer rejects records with non-default `schema_version` at the backend boundary; version-mismatch routing is now an engine concern at resume time. The `CHECKPOINT_SCHEMA_VERSION` module constant is removed; future record-shape evolution can add backend-private metadata fields if needed.
+- **`NodeEvent.pre_state` typed `Any` (was `State`).** Required by the new `checkpoint_migrated` phase which carries a `_MigrationSummary` payload rather than a `State` instance. Observer authors who type-narrowed `pre_state` to `State` should treat it as `Any` and narrow per-phase (e.g., `if event.phase == "completed": ...`). The `checkpoint_saved` phase already carried a State-flavored shape (not necessarily a typed `State` subclass instance), so this widens the declared type to match runtime reality rather than introducing a new constraint.
 
 ### Notes
 
diff --git a/docs/concepts/checkpointing.md b/docs/concepts/checkpointing.md
index a59dfd6..0cfb66c 100644
--- a/docs/concepts/checkpointing.md
+++ b/docs/concepts/checkpointing.md
@@ -149,6 +149,134 @@ multi-process), S3 (cross-region durability). For event-sourced
 runtimes (Temporal, DBOS, Restate, Inngest) the Protocol is the
 adapter layer.
 
+## State migrations
+
+When a checkpoint was saved against an earlier version of your state
+schema and the code has since evolved, the engine consults a
+**migration registry** to bridge the saved record into the current
+shape. Without migrations, a schema change invalidates every prior
+checkpoint; with one short registration per change, you keep your
+saved records working across releases.
+
+The wire-up is two pieces: declare a version on your state class,
+and register one migration per version bump.
+
+```python
+from typing import ClassVar
+from openarmature.graph import State, GraphBuilder
+from openarmature.checkpoint import SQLiteCheckpointer
+
+
+class MyState(State):
+    schema_version: ClassVar[str] = "v2"
+    x: int = 0
+    new_field: str = "default"      # added in v2
+
+
+def add_new_field_default(state: dict) -> dict:
+    return {**state, "new_field": "default"}
+
+
+graph = (
+    GraphBuilder(MyState)
+    .add_node(...)
+    .with_checkpointer(SQLiteCheckpointer("ck.db", serialization="json"))
+    .with_state_migration("v1", "v2", add_new_field_default)
+    .compile()
+)
+```
+
+On resume, the engine reads the saved record's `schema_version`. If
+it equals `MyState.schema_version`, the record loads via the §10.4
+fast path (no migration consulted). If it differs, the engine
+resolves a chain through the registry (BFS for the shortest path),
+applies each migration in order to the record's state, then
+deserializes the result into your current state class.
+
+### Chain resolution
+
+Registered migrations form a directed graph. Each
+`with_state_migration(a, b, fn)` is an edge from `a` to `b`. Chain
+resolution finds the shortest path between the saved version and the
+current version. Branching is fine: a v1 record can have one
+migration leading to v2 and another leading to v2-experimental;
+chain resolution picks the path that ends at the current declared
+version.
+
+Two ambiguity cases are configuration errors. Both surface as
+`CheckpointStateMigrationChainAmbiguous`:
+
+- **Duplicate edges.** Registering two migrations with the same
+  `(from_version, to_version)` pair raises at registration time so
+  the configuration error surfaces before any resume attempt.
+  Either delete one or pick distinct version identifiers.
+- **Multiple shortest paths.** A diamond like
+  `v1 → v2 → v4` and `v1 → v3 → v4` is ambiguous: both paths have
+  length 2. The engine raises during resume so the user can
+  register fewer migrations or pick a single canonical route.
+
+### The three new error categories
+
+- **`CheckpointStateMigrationChainAmbiguous`**: the registered
+  migration graph is ambiguous (duplicate `(from, to)` pair at
+  registration time, OR multiple distinct shortest paths between
+  the saved and current versions at resume time). Surfaces before
+  any migration function runs. Carries `from_version` and
+  `to_version` when known.
+- **`CheckpointStateMigrationMissing`**: the saved version doesn't
+  match the current version, and no chain bridges them. Carries
+  `from_version`, `to_version`, a count of registered migrations,
+  and a human-readable `registry_description` so operators see what
+  IS available.
+- **`CheckpointStateMigrationFailed`**: a user-supplied migration
+  function raised. Subsequent migrations in the chain don't run;
+  the resume fails. The migration's exception rides `__cause__`.
+
+Routing precedence on resume: chain-ambiguous → missing → failed →
+record-invalid.
+
+A third category, `CheckpointRecordInvalid`, continues to cover the
+**post**-migration case: a migration ran cleanly but produced
+output that the current state class can't deserialize (missing a
+required field, wrong type, etc.). The three categories are
+mutually exclusive on any given resume.
+
+### Backend support
+
+Not every backend can migrate. Migration needs the backend to expose
+a **structural intermediate form** of the loaded state (a plain
+dict, JSON tree, or similar) that's independent of the current
+state class.
+
+- **`SQLiteCheckpointer(serialization="json")`** can. JSON-encoded
+  state loads to a dict; the migration function operates on the
+  dict directly.
+- **`SQLiteCheckpointer(serialization="pickle")`** can NOT. Pickle
+  holds class identity and round-trips back to typed instances.
+- **`InMemoryCheckpointer`** can NOT. It holds live typed-state
+  references by reference; there's no serialization step.
+
+On version mismatch against a non-migration-eligible backend, the
+engine raises `CheckpointRecordInvalid` (not
+`CheckpointStateMigrationMissing`): the registry has no chance to
+bridge.
+
+### Parent-state migration
+
+Subgraph saves carry a `parent_states` chain of the outer-graph
+state captured at the moment of the inner save. On resume, the same
+migration chain applies to each entry in `parent_states` in lockstep
+with the outer state. The spec treats `parent_states` as carrying
+the same `schema_version` as the outer record (no per-parent
+version metadata in v1).
+
+### Migrations MUST be pure
+
+A migration function MUST be deterministic, with no I/O, no implicit
+state, no random or wall-clock-derived output. The framework
+doesn't enforce purity, but violating it breaks determinism
+guarantees for resume.
+
 ## When NOT to use checkpointing
 
 - **Pure pipelines that complete in seconds.** Restart-from-entry is
diff --git a/openarmature-spec b/openarmature-spec
index 644c2a5..bdfe13a 160000
--- a/openarmature-spec
+++ b/openarmature-spec
@@ -1 +1 @@
-Subproject commit 644c2a5448f9d5f00205e696cef08ddf7c61bae2
+Subproject commit bdfe13ad3bbeb83401eedb2b6e9ab51a15619ad3
diff --git a/pyproject.toml b/pyproject.toml
index f16133f..fd33d16 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -48,7 +48,7 @@ Repository = "https://github.com/LunarCommand/openarmature-python"
 Specification = "https://github.com/LunarCommand/openarmature-spec"
 
 [tool.openarmature]
-spec_version = "0.15.0"
+spec_version = "0.16.0"
 
 [dependency-groups]
 dev = [
diff --git a/src/openarmature/__init__.py b/src/openarmature/__init__.py
index 69af6a1..feef202 100644
--- a/src/openarmature/__init__.py
+++ b/src/openarmature/__init__.py
@@ -1,4 +1,4 @@
 """OpenArmature — workflow framework for LLM pipelines and tool-calling agents."""
 
 __version__ = "0.5.0"
-__spec_version__ = "0.15.0"
+__spec_version__ = "0.16.0"
diff --git a/src/openarmature/checkpoint/__init__.py b/src/openarmature/checkpoint/__init__.py
index 3acc363..ba13022 100644
--- a/src/openarmature/checkpoint/__init__.py
+++ b/src/openarmature/checkpoint/__init__.py
@@ -26,9 +26,12 @@
     CheckpointNotFound,
     CheckpointRecordInvalid,
     CheckpointSaveFailed,
+    CheckpointStateMigrationChainAmbiguous,
+    CheckpointStateMigrationFailed,
+    CheckpointStateMigrationMissing,
 )
+from .migration import MigrationRegistry, StateMigration
 from .protocol import (
-    CHECKPOINT_SCHEMA_VERSION,
     Checkpointer,
     CheckpointFilter,
     CheckpointRecord,
@@ -37,17 +40,21 @@
 )
 
 __all__ = [
-    "CHECKPOINT_SCHEMA_VERSION",
     "CheckpointError",
     "CheckpointFilter",
     "CheckpointNotFound",
     "CheckpointRecord",
     "CheckpointRecordInvalid",
     "CheckpointSaveFailed",
+    "CheckpointStateMigrationChainAmbiguous",
+    "CheckpointStateMigrationFailed",
+    "CheckpointStateMigrationMissing",
     "CheckpointSummary",
     "Checkpointer",
     "InMemoryCheckpointer",
+    "MigrationRegistry",
     "NodePosition",
     "SQLiteCheckpointer",
     "SerializationMode",
+    "StateMigration",
 ]
diff --git a/src/openarmature/checkpoint/backends/memory.py b/src/openarmature/checkpoint/backends/memory.py
index 1c4347b..1693264 100644
--- a/src/openarmature/checkpoint/backends/memory.py
+++ b/src/openarmature/checkpoint/backends/memory.py
@@ -28,8 +28,25 @@ class InMemoryCheckpointer:
     Pydantic state instance the engine produces is what comes back
     from :meth:`load` — no serialization round-trip. (This is the
     feature: tests can assert on the saved state's identity.)
+
+    **State-migration eligibility:** none. Per spec §10.12.1, a
+    backend supports migration only when it can expose a structural
+    intermediate form of the loaded state independent of the current
+    state class. This backend holds live typed instances by
+    reference, so a version mismatch on resume raises
+    ``CheckpointRecordInvalid`` rather than consulting the
+    migration registry.
     """
 
+    # Per spec §10.12.1: in-memory storage holds live typed-state
+    # references, so there's no class-independent intermediate form
+    # the migration registry could consume. Declared at the class
+    # level (not as a per-instance attribute) since the answer is
+    # constructor-independent; the Protocol declaration in
+    # ``protocol.py`` types this as ``bool`` (not ``ClassVar[bool]``)
+    # so Pyright accepts a class-attribute override here.
+    supports_state_migration: bool = False
+
     def __init__(self) -> None:
         self._records: dict[str, CheckpointRecord] = {}
         self._lock = asyncio.Lock()
diff --git a/src/openarmature/checkpoint/backends/sqlite.py b/src/openarmature/checkpoint/backends/sqlite.py
index d8065ba..588562c 100644
--- a/src/openarmature/checkpoint/backends/sqlite.py
+++ b/src/openarmature/checkpoint/backends/sqlite.py
@@ -42,7 +42,6 @@
 
 from ..errors import CheckpointRecordInvalid
 from ..protocol import (
-    CHECKPOINT_SCHEMA_VERSION,
     CheckpointFilter,
     CheckpointRecord,
     CheckpointSummary,
@@ -109,6 +108,13 @@ def __init__(
         self._serialization: SerializationMode = serialization
         self._lock = asyncio.Lock()
         self._initialized = False
+        # Per spec §10.12.1, a backend supports state migration only
+        # when it can expose a structural intermediate form of the
+        # loaded state that is independent of the current state
+        # class. JSON serialization satisfies this (loads to dicts);
+        # pickle holds class identity and round-trips to typed
+        # instances, so it cannot bridge a schema-version mismatch.
+        self.supports_state_migration: bool = serialization == "json"
 
     def _connect(self) -> sqlite3.Connection:
         conn = sqlite3.connect(self._path)
@@ -230,12 +236,12 @@ def _do() -> tuple[Any, ...] | None:
             schema_version,
             recorded_serialization,
         ) = row
-        if schema_version != CHECKPOINT_SCHEMA_VERSION:
-            raise CheckpointRecordInvalid(
-                invocation_id,
-                f"persisted schema_version={schema_version!r} does not match "
-                f"current {CHECKPOINT_SCHEMA_VERSION!r}",
-            )
+        # Note: per spec §10.12 (proposal 0014), version mismatches
+        # are no longer rejected at the backend boundary. The engine
+        # routes mismatches through the migration registry on resume
+        # (CheckpointStateMigrationMissing if no chain, else applies
+        # the chain). The backend just round-trips the version
+        # identifier as opaque data.
         state = self._decode(state_blob, recorded_serialization, invocation_id)
         position_dicts = self._decode(positions_blob, recorded_serialization, invocation_id)
         parent_states = self._decode(parent_states_blob, recorded_serialization, invocation_id)
diff --git a/src/openarmature/checkpoint/errors.py b/src/openarmature/checkpoint/errors.py
index 380b97e..073d031 100644
--- a/src/openarmature/checkpoint/errors.py
+++ b/src/openarmature/checkpoint/errors.py
@@ -17,6 +17,8 @@
 
 from __future__ import annotations
 
+from typing import Any
+
 
 class CheckpointError(Exception):
     """Base for all checkpoint errors. Each subclass carries a
@@ -56,10 +58,17 @@ def __init__(self, invocation_id: str, cause: BaseException) -> None:
 
 class CheckpointRecordInvalid(CheckpointError):
     """Raised when ``Checkpointer.load(X)`` returns a record whose
-    schema is incompatible with the current graph (state shape
-    mismatch, missing required fields, or
-    ``schema_version`` mismatch). Non-transient — the persisted
-    record was written by an incompatible version of the engine."""
+    schema is incompatible with the current graph: state shape
+    mismatch, missing required fields, OR a post-migration state
+    that fails to deserialize against the current state class (per
+    spec §10.12.4). Non-transient.
+
+    Note: raw ``schema_version`` mismatches no longer route here.
+    They now flow through ``CheckpointStateMigrationMissing`` (no
+    chain registered) or ``CheckpointStateMigrationFailed`` (chain
+    application raised) per spec §10.10's three-way category
+    distinction.
+    """
 
     category = "checkpoint_record_invalid"
 
@@ -68,9 +77,106 @@ def __init__(self, invocation_id: str, message: str) -> None:
         self.invocation_id = invocation_id
 
 
+class CheckpointStateMigrationMissing(CheckpointError):
+    """Raised on resume when the saved record's ``schema_version``
+    does not match the current state class's ``schema_version`` AND
+    no chain of registered migrations bridges the two. Non-transient
+    per spec §10.10 — the user MUST register a migration (or pin
+    their state to the saved version) for the resume to succeed.
+
+    Carries the saved-from / current-to versions and a description
+    of the registered migration set so the user can see what
+    migrations are available.
+    """
+
+    category = "checkpoint_state_migration_missing"
+
+    from_version: str
+    to_version: str
+    registered_migrations_count: int
+    registry_description: str
+
+    def __init__(
+        self,
+        *args: Any,
+        from_version: str,
+        to_version: str,
+        registered_migrations_count: int,
+        registry_description: str,
+    ) -> None:
+        super().__init__(*args)
+        self.from_version = from_version
+        self.to_version = to_version
+        self.registered_migrations_count = registered_migrations_count
+        self.registry_description = registry_description
+
+
+class CheckpointStateMigrationChainAmbiguous(CheckpointError):
+    """Raised when the registered migration graph is ambiguous per
+    spec §10.10 / §10.12 (proposal 0018, spec v0.16.0):
+
+    - Duplicate-pair case (§10.12.1): two migrations register with the
+      same ``(from_version, to_version)`` pair. Raised at registration
+      time so the user sees the ambiguity before any resume attempt.
+    - Multi-shortest-path case (§10.12.2): the registered migration
+      graph has multiple distinct shortest paths between the saved
+      and current versions (e.g., a diamond ``v1→v2→v4`` + ``v1→v3→v4``).
+      Spec accepts either compile-time detection (recommended) or
+      load-time detection (this impl runs the check inside BFS at
+      resume time).
+
+    Non-transient: retrying without changing the migration graph
+    will not succeed. Carries ``from_version`` / ``to_version`` when
+    known (always set for the duplicate-pair case, set on the resume
+    side too for multi-shortest-path detection).
+    """
+
+    category = "checkpoint_state_migration_chain_ambiguous"
+
+    from_version: str | None
+    to_version: str | None
+
+    def __init__(
+        self,
+        *args: Any,
+        from_version: str | None = None,
+        to_version: str | None = None,
+    ) -> None:
+        super().__init__(*args)
+        self.from_version = from_version
+        self.to_version = to_version
+
+
+class CheckpointStateMigrationFailed(CheckpointError):
+    """Raised on resume when a registered migration function raises
+    during chain application (per spec §10.12.2). The migration's
+    exception is preserved as ``__cause__``. Non-transient by
+    default: a buggy migration is deterministic, so retrying
+    without changing the migration code will not succeed.
+    """
+
+    category = "checkpoint_state_migration_failed"
+
+    from_version: str
+    to_version: str
+
+    def __init__(
+        self,
+        *args: Any,
+        from_version: str,
+        to_version: str,
+    ) -> None:
+        super().__init__(*args)
+        self.from_version = from_version
+        self.to_version = to_version
+
+
 __all__ = [
     "CheckpointError",
     "CheckpointNotFound",
     "CheckpointRecordInvalid",
     "CheckpointSaveFailed",
+    "CheckpointStateMigrationChainAmbiguous",
+    "CheckpointStateMigrationFailed",
+    "CheckpointStateMigrationMissing",
 ]
diff --git a/src/openarmature/checkpoint/migration.py b/src/openarmature/checkpoint/migration.py
new file mode 100644
index 0000000..0a1cb6e
--- /dev/null
+++ b/src/openarmature/checkpoint/migration.py
@@ -0,0 +1,217 @@
+"""State migration types and registry.
+
+Realizes pipeline-utilities §10.12 (proposal 0014). A
+``StateMigration`` describes one edge in the migration graph;
+``MigrationRegistry`` holds the ordered set and resolves chains
+via BFS. Ambiguity (duplicate ``(from, to)`` pairs OR multiple
+distinct shortest paths between the same source/sink) is a
+configuration-style error per §10.12.1 / §10.12.2.
+"""
+
+from __future__ import annotations
+
+from collections import deque
+from collections.abc import Callable, Iterator
+from dataclasses import dataclass
+from typing import Any
+
+from .errors import CheckpointStateMigrationChainAmbiguous
+
+
+@dataclass(frozen=True)
+class StateMigration:
+    """One edge in the migration graph.
+
+    ``migrate`` receives the most-deserialized form the backend can
+    expose that is still independent of the current state class
+    (a plain ``dict`` for JSON-backed backends). It MUST return a
+    value of the same kind, suitable for the next migration in the
+    chain (or for final deserialization into the current state class).
+
+    Migrations MUST be pure: deterministic, no I/O, no implicit
+    state. The framework does not police purity per spec §10.12.2
+    ("the contract is documented, not policed"); violating it
+    risks non-deterministic resume.
+    """
+
+    from_version: str
+    to_version: str
+    migrate: Callable[[Any], Any]
+
+
+class MigrationRegistry:
+    """Ordered set of registered migrations + BFS chain resolution.
+
+    Registration-time invariants:
+
+    - Two migrations with the same ``from_version`` AND
+      ``to_version`` raise ``CheckpointStateMigrationChainAmbiguous``
+      directly per spec §10.10 (proposal 0018) so the canonical
+      category surfaces at the registration boundary without any
+      wrapping by the builder.
+    - Two migrations with the same ``from_version`` and different
+      ``to_version`` are permitted (branched migration graph;
+      chain resolution picks a path or raises ambiguity if multiple
+      shortest paths exist).
+
+    Resolution-time semantics (per §10.12.2):
+
+    - BFS from ``record.schema_version`` to
+      ``current.schema_version``. BFS naturally finds the shortest
+      path.
+    - Empty registry on mismatch → no path → caller raises
+      ``CheckpointStateMigrationMissing``.
+    - Non-empty registry with no connecting path → same.
+    - Found a unique shortest path → return ordered list.
+    - Found multiple distinct shortest paths (same edge count,
+      different edge sequences) → raise ``ValueError`` internally;
+      ``CompiledGraph._migrate_record`` wraps the ``ValueError`` as
+      ``CheckpointStateMigrationChainAmbiguous`` at the resume
+      boundary. The internal ``ValueError`` keeps the registry
+      module dependency-light (no canonical-error import cycle).
+    """
+
+    def __init__(self) -> None:
+        self._migrations: dict[tuple[str, str], StateMigration] = {}
+        self._edges: dict[str, list[StateMigration]] = {}
+
+    def register(self, migration: StateMigration) -> None:
+        # Per spec §10.2 / proposal 0014: empty ``to_version`` would
+        # route migrations TO the "not declared" sentinel, which
+        # the spec calls out as not migration-eligible — incoherent
+        # as a chain target. Empty ``from_version`` stays valid:
+        # it's the documented bridging path for pre-declaration
+        # records (current class declares ``"v1"``; users register
+        # ``with_state_migration("", "v1", fn)`` to migrate records
+        # saved before the class declared a version). See Q4 in the
+        # proposal-0014 coord-thread plan for the matrix.
+        if not migration.to_version:
+            raise ValueError(
+                f"state migration to_version MUST be non-empty "
+                f"(got from_version={migration.from_version!r}, "
+                f"to_version={migration.to_version!r})"
+            )
+        key = (migration.from_version, migration.to_version)
+        if key in self._migrations:
+            # Per spec §10.10 / §10.12.1 (proposal 0018, spec v0.16.0):
+            # duplicate-pair detection raises the canonical category
+            # directly at registration time. The category surfaces
+            # before any resume attempt — neither the builder nor the
+            # caller needs to wrap.
+            raise CheckpointStateMigrationChainAmbiguous(
+                f"duplicate state migration {migration.from_version!r}→"
+                f"{migration.to_version!r} registered; chain would be ambiguous",
+                from_version=migration.from_version,
+                to_version=migration.to_version,
+            )
+        self._migrations[key] = migration
+        self._edges.setdefault(migration.from_version, []).append(migration)
+
+    def __iter__(self) -> Iterator[StateMigration]:
+        return iter(self._migrations.values())
+
+    def __len__(self) -> int:
+        return len(self._migrations)
+
+    def resolve_chain(
+        self,
+        from_version: str,
+        to_version: str,
+    ) -> list[StateMigration] | None:
+        """Return an ordered chain of migrations bridging the two
+        versions, or ``None`` if no chain exists.
+
+        Raises ``CheckpointStateMigrationChainAmbiguous`` if
+        multiple distinct shortest paths exist between
+        ``from_version`` and ``to_version`` (ambiguous chain per
+        spec §10.10 / §10.12.2 — proposal 0018 / spec v0.16.0).
+        Same canonical category as the duplicate-pair detection
+        in ``register``; one type for chain ambiguity regardless
+        of when it surfaces.
+        """
+        if from_version == to_version:
+            return []
+
+        # BFS that records every shortest-length path. If multiple
+        # paths share the minimum length, the chain is ambiguous.
+        # Standard BFS finds the shortest distance; the path-recording
+        # variant lets us detect ambiguity without a second pass.
+        # ``frontier`` items are (version, path_so_far).
+        frontier: deque[tuple[str, list[StateMigration]]] = deque()
+        frontier.append((from_version, []))
+        shortest_paths: list[list[StateMigration]] = []
+        shortest_length: int | None = None
+        # ``distances`` tracks the BFS layer at which each node was
+        # first seen. Frontier entries past the shortest_length layer
+        # are pruned.
+        distances: dict[str, int] = {from_version: 0}
+
+        while frontier:
+            version, path = frontier.popleft()
+            depth = len(path)
+            # Stop expanding once we've moved past the shortest target.
+            if shortest_length is not None and depth >= shortest_length:
+                continue
+            for edge in self._edges.get(version, []):
+                next_version = edge.to_version
+                next_path = path + [edge]
+                if next_version == to_version:
+                    if shortest_length is None:
+                        shortest_length = len(next_path)
+                    if len(next_path) == shortest_length:
+                        shortest_paths.append(next_path)
+                    continue
+                # Cycle-avoidance: a node revisited at the same or
+                # deeper BFS layer can't contribute to a strict-
+                # shortest path. Allow re-entry only when the new
+                # arrival is at the same layer as the first arrival
+                # (distinct shortest paths through the same node).
+                # NOTE: the strict-less-than comparison is load-
+                # bearing for multi-shortest-path detection — a
+                # diamond v1→v2→v4 + v1→v3→v4 lets BFS reach v4 via
+                # both v2 and v3 at layer 2, and both paths land in
+                # ``shortest_paths``. Tightening this to ``<=`` would
+                # break the ambiguity check.
+                prior_depth = distances.get(next_version)
+                if prior_depth is not None and prior_depth < depth + 1:
+                    continue
+                distances[next_version] = depth + 1
+                frontier.append((next_version, next_path))
+
+        if not shortest_paths:
+            return None
+        if len(shortest_paths) > 1:
+            descriptions = [" → ".join([from_version, *(e.to_version for e in p)]) for p in shortest_paths]
+            # Per spec §10.10 / §10.12.2 (proposal 0018, v0.16.0):
+            # raise the canonical category directly so the
+            # registry's ambiguity contract is one type regardless
+            # of when it surfaces (duplicate-pair at register time
+            # vs multi-shortest-path at resolve time).
+            raise CheckpointStateMigrationChainAmbiguous(
+                f"ambiguous migration chain from {from_version!r} to "
+                f"{to_version!r}: multiple distinct shortest paths exist "
+                f"({descriptions}); register fewer migrations or pick a "
+                f"single canonical route",
+                from_version=from_version,
+                to_version=to_version,
+            )
+        return shortest_paths[0]
+
+    def describe(self) -> str:
+        """Human-readable description of the registered set, used
+        in the ``CheckpointStateMigrationMissing`` error payload.
+        Empty registry returns ``"<no migrations registered>"``.
+
+        Output is registration-order (Python's dict preserves
+        insertion order). Diff-friendly test assertions should
+        not depend on the order across distinct registration
+        sequences; if cross-language conformance ever needs a
+        canonical order, a future change can sort by
+        ``(from_version, to_version)``.
+        """
+        if not self._migrations:
+            return "<no migrations registered>"
+        return "\n".join(f"{m.from_version} → {m.to_version}" for m in self._migrations.values())
+
+
+__all__ = ["MigrationRegistry", "StateMigration"]
diff --git a/src/openarmature/checkpoint/protocol.py b/src/openarmature/checkpoint/protocol.py
index 6ca11dd..3b4006a 100644
--- a/src/openarmature/checkpoint/protocol.py
+++ b/src/openarmature/checkpoint/protocol.py
@@ -22,11 +22,12 @@
 subgraph-internal nodes. Fan-out instance internal events do NOT
 produce records in the shipping version (atomic-restart contract).
 
-The :data:`CHECKPOINT_SCHEMA_VERSION` constant is the single source
-of truth for the persisted record shape — bump it whenever the field
-set changes incompatibly so older saved records surface as
-:class:`CheckpointRecordInvalid` on load rather than silently coercing
-to a mismatched shape.
+``CheckpointRecord.schema_version`` carries the user-facing
+state-schema identifier per spec §10.2 (proposal 0014 repurposes
+the field from the original backend-internal record-shape role).
+The framework reads ``type(state).schema_version`` at save time;
+on load, version mismatches route through the migration registry
+(per §10.12) rather than a strict equality check.
 """
 
 from __future__ import annotations
@@ -35,15 +36,6 @@
 from dataclasses import dataclass, field
 from typing import Any, Protocol
 
-# Persisted record shape version. Bump when CheckpointRecord's field
-# set or invariants change incompatibly — proposal 0009 (per-instance
-# fan-out resume) is the concrete near-term candidate, since it
-# populates ``fan_out_progress`` and the load path will need to
-# distinguish v1 records (where it's None) from v2 records (where it
-# carries instance-level progress data). Backends that reject mismatches
-# MUST raise CheckpointRecordInvalid per spec §10.10.
-CHECKPOINT_SCHEMA_VERSION = "1"
-
 
 # Spec: realizes pipeline-utilities §10.2 NodePosition. Field semantics
 # tied to graph-engine §6 NodeEvent (step shared; namespace here omits
@@ -108,7 +100,11 @@ class CheckpointRecord:
     completed_positions: tuple[NodePosition, ...]
     parent_states: tuple[Any, ...]
     last_saved_at: float
-    schema_version: str = CHECKPOINT_SCHEMA_VERSION
+    # Per spec §10.2 (proposal 0014): the user's state-schema
+    # version, read off ``type(state).schema_version`` at save time.
+    # Empty-string sentinel for state classes that don't declare a
+    # version; non-empty declares migration-eligibility.
+    schema_version: str = ""
     fan_out_progress: None = field(default=None)
 
 
@@ -154,8 +150,46 @@ class Checkpointer(Protocol):
     access). Each operation MUST be thread-safe (Python) /
     task-coroutine-safe (asyncio); backends with synchronous I/O
     typically wrap their work in ``asyncio.to_thread`` or equivalent.
+
+    ``supports_state_migration`` marks whether the backend can
+    expose a structural intermediate form of the loaded state (a
+    plain dict, JSON tree, or similar) that is independent of the
+    current state class. JSON-encoded backends naturally satisfy
+    this; backends that store live typed state instances or use
+    class-bound serialization (pickle) cannot. Per spec §10.12.1,
+    backends that cannot expose the intermediate MUST raise
+    ``CheckpointRecordInvalid`` on version mismatch even when
+    migrations are registered — the registry has no chance to bridge.
+
+    **Attribute-presence contract.** The class-body ``= False``
+    below is a typing-level signal, not a runtime guarantee:
+    ``typing.Protocol`` does not create an instance attribute on
+    a conforming class that doesn't declare it itself. Concrete
+    backends SHOULD declare ``supports_state_migration`` (either
+    at the class level like ``InMemoryCheckpointer`` does, or as
+    an ``__init__``-set instance attribute like
+    ``SQLiteCheckpointer`` does for the mode-dependent case) so
+    Pyright accepts the structural conformance and ``getattr``
+    sees the value. The engine's resume path reads the attribute
+    via ``getattr(checkpointer, "supports_state_migration",
+    False)``, so a third-party backend that omits the attribute
+    entirely is treated as non-migration-eligible without
+    raising — that's the runtime default the engine guarantees.
     """
 
+    # Declared as an instance attribute (not ``ClassVar``) so backends
+    # can compute it at construction time when the answer depends on
+    # constructor args. SQLiteCheckpointer is the concrete case:
+    # JSON-mode supports migration, pickle-mode doesn't, and the mode
+    # is a per-instance constructor arg. Backends with a static answer
+    # (InMemoryCheckpointer is always False) override at the class
+    # level. Pyright accepts either shape because Protocol attribute
+    # conformance ignores the ClassVar marker on subclasses. The
+    # class-body default below is for typing only; see the
+    # docstring's "Attribute-presence contract" section for the
+    # runtime ``getattr``-based safety net.
+    supports_state_migration: bool = False
+
     async def save(self, invocation_id: str, record: CheckpointRecord) -> None:
         """Persist ``record`` for ``invocation_id``. After return the
         record MUST be durable across process crashes for backends
@@ -186,7 +220,6 @@ async def delete(self, invocation_id: str) -> None:
 
 
 __all__ = [
-    "CHECKPOINT_SCHEMA_VERSION",
     "CheckpointFilter",
     "CheckpointRecord",
     "CheckpointSummary",
diff --git a/src/openarmature/graph/builder.py b/src/openarmature/graph/builder.py
index c930272..be5b8dd 100644
--- a/src/openarmature/graph/builder.py
+++ b/src/openarmature/graph/builder.py
@@ -14,6 +14,8 @@
 from types import GenericAlias, UnionType
 from typing import Any, Self, cast, get_args, get_origin
 
+from openarmature.checkpoint.errors import CheckpointStateMigrationChainAmbiguous
+from openarmature.checkpoint.migration import MigrationRegistry, StateMigration
 from openarmature.checkpoint.protocol import Checkpointer
 
 from .compiled import CompiledGraph
@@ -51,6 +53,10 @@ def __init__(self, state_cls: type[StateT]) -> None:
         # Optional Checkpointer attached at compile time; ``None`` is
         # the spec §10.1.1 default-off behavior.
         self._checkpointer: Checkpointer | None = None
+        # State-migration registry per pipeline-utilities §10.12
+        # (proposal 0014). Populated by ``with_state_migration(s)``;
+        # passed through to the compiled graph.
+        self._migration_registry: MigrationRegistry = MigrationRegistry()
 
     def add_node(
         self,
@@ -251,6 +257,79 @@ def with_checkpointer(self, checkpointer: Checkpointer) -> Self:
         self._checkpointer = checkpointer
         return self
 
+    def with_state_migration(
+        self,
+        from_version: str,
+        to_version: str,
+        migrate: Callable[[Any], Any],
+    ) -> Self:
+        """Register one state migration per pipeline-utilities §10.12.
+
+        On resume, when the saved record's ``schema_version`` does not
+        match the current state class's ``schema_version``, the engine
+        consults the registry for a chain that bridges the two and
+        applies it to the record's state (and to each entry in
+        ``parent_states``) before deserialization.
+
+        Migrations MUST be pure: deterministic, no I/O, no implicit
+        state. The framework does not police purity (per §10.12.2),
+        but violating it risks non-deterministic resume.
+
+        Raises ``CheckpointStateMigrationChainAmbiguous`` at
+        registration if the ``(from_version, to_version)`` pair is
+        already registered (per spec §10.10 / §10.12.1 — proposal
+        0018 / spec v0.16.0). Also raises ``ValueError`` if
+        ``to_version`` is the empty-string sentinel (the un-declared
+        marker per §10.2 is not a valid chain target).
+        """
+        self._migration_registry.register(
+            StateMigration(
+                from_version=from_version,
+                to_version=to_version,
+                migrate=migrate,
+            )
+        )
+        return self
+
+    def with_state_migrations(self, *migrations: StateMigration) -> Self:
+        """Register multiple migrations in one call. Convenience over
+        ``with_state_migration``; each entry is registered through the
+        same path and obeys the same ambiguity rule.
+
+        Pre-validates the full input list against the existing
+        registry + against earlier entries in the same call before
+        mutating, so a duplicate in the third entry cannot leave
+        the first two half-registered. If any duplicate ``(from,
+        to)`` pair is detected the call raises
+        ``CheckpointStateMigrationChainAmbiguous`` without mutating
+        the registry; otherwise all entries register atomically.
+        """
+        # Pre-validation pass: collect every (from, to) we're about
+        # to add, check both against the existing registry and
+        # against earlier entries in the input. Raise before
+        # mutating if anything collides.
+        seen_in_call: set[tuple[str, str]] = set()
+        for m in migrations:
+            key = (m.from_version, m.to_version)
+            if key in self._migration_registry._migrations:  # noqa: SLF001
+                raise CheckpointStateMigrationChainAmbiguous(
+                    f"duplicate state migration {m.from_version!r}→{m.to_version!r} already registered",
+                    from_version=m.from_version,
+                    to_version=m.to_version,
+                )
+            if key in seen_in_call:
+                raise CheckpointStateMigrationChainAmbiguous(
+                    f"duplicate state migration {m.from_version!r}→"
+                    f"{m.to_version!r} repeated in with_state_migrations call",
+                    from_version=m.from_version,
+                    to_version=m.to_version,
+                )
+            seen_in_call.add(key)
+        # Validation passed — commit them all.
+        for migration in migrations:
+            self._migration_registry.register(migration)
+        return self
+
     def add_middleware(self, middleware: Middleware) -> Self:
         """Register a per-graph middleware applied to every node in this graph.
 
@@ -343,6 +422,7 @@ def compile(self) -> CompiledGraph[StateT]:
             edges=edges_by_source,
             reducers=resolved,
             middleware=tuple(self._middleware),
+            migration_registry=self._migration_registry,
         )
         if self._checkpointer is not None:
             compiled.attach_checkpointer(self._checkpointer)
diff --git a/src/openarmature/graph/compiled.py b/src/openarmature/graph/compiled.py
index 6234a42..245af6c 100644
--- a/src/openarmature/graph/compiled.py
+++ b/src/openarmature/graph/compiled.py
@@ -28,6 +28,7 @@
 import uuid
 from collections.abc import Callable, Iterable, Mapping
 from dataclasses import dataclass, field
+from dataclasses import replace as dataclass_replace
 from typing import TYPE_CHECKING, Any, cast
 
 if TYPE_CHECKING:
@@ -45,12 +46,15 @@
 from pydantic import ValidationError
 
 from openarmature.checkpoint.errors import (
+    CheckpointError,
     CheckpointNotFound,
     CheckpointRecordInvalid,
     CheckpointSaveFailed,
+    CheckpointStateMigrationFailed,
+    CheckpointStateMigrationMissing,
 )
+from openarmature.checkpoint.migration import MigrationRegistry, StateMigration
 from openarmature.checkpoint.protocol import (
-    CHECKPOINT_SCHEMA_VERSION,
     Checkpointer,
     CheckpointRecord,
     NodePosition,
@@ -242,6 +246,55 @@ def _no_op_finalize(_edge_error: RuntimeGraphError | None) -> None:
     silently per proposal 0012 + fixture 013."""
 
 
+@dataclass(frozen=True)
+class _MigrationSummary:
+    """Per-resume migration-chain metadata threaded out of
+    ``_migrate_record`` so the engine can dispatch an
+    ``openarmature.checkpoint.migrate`` observer event after the
+    invocation context is built (per spec §6 cross-ref in proposal
+    0014). Carried on the synthetic ``NodeEvent.pre_state``
+    payload for ``phase="checkpoint_migrated"``; the OTel observer
+    reads it to emit the span.
+    """
+
+    from_version: str
+    to_version: str
+    chain_length: int
+
+
+def _apply_migration_step(
+    migration: StateMigration,
+    value: Any,
+    label: str,
+) -> Any:
+    """Apply one migration step to one value (outer state or one
+    parent-state entry). Wraps the user-supplied migration function's
+    raise as ``CheckpointStateMigrationFailed`` per spec §10.12.2.
+    The original exception rides ``__cause__``.
+    """
+    try:
+        return migration.migrate(value)
+    except CheckpointError:
+        # Preserve canonical category — if a migration raises a
+        # CheckpointError subclass itself (rare; migrations are
+        # spec-mandated pure per §10.12.2), propagate the original
+        # category rather than wrapping it as
+        # CheckpointStateMigrationFailed.
+        raise
+    except Exception as exc:
+        # Concise wrap-message intentionally. ``raise ... from exc``
+        # preserves the original exception on ``__cause__``;
+        # Python's traceback formatter surfaces it, so embedding the
+        # underlying ``type/str`` in this message would just
+        # duplicate information (and confuse the output when the
+        # underlying ``__str__`` is multi-line).
+        raise CheckpointStateMigrationFailed(
+            f"migration {migration.from_version!r}→{migration.to_version!r} raised while migrating {label}",
+            from_version=migration.from_version,
+            to_version=migration.to_version,
+        ) from exc
+
+
 @dataclass(frozen=True)
 class CompiledGraph[StateT: State]:
     """An immutable, executable graph produced by `GraphBuilder.compile()`.
@@ -272,6 +325,11 @@ class CompiledGraph[StateT: State]:
     # the user can swap the registered Checkpointer via
     # ``attach_checkpointer``. ``None`` when no backend is registered.
     _checkpointer_slot: list[Checkpointer | None] = field(default_factory=lambda: [None])
+    # State-migration registry (pipeline-utilities §10.12 / proposal
+    # 0014). Populated by ``GraphBuilder.with_state_migration(s)``;
+    # consulted on resume when the loaded record's ``schema_version``
+    # does not match the current state class's ``schema_version``.
+    migration_registry: MigrationRegistry = field(default_factory=MigrationRegistry)
 
     # ------------------------------------------------------------------
     # Observer registration (spec v0.6.0 §6)
@@ -329,6 +387,93 @@ def checkpointer(self) -> Checkpointer | None:
         """Currently-registered Checkpointer, or ``None``."""
         return self._checkpointer_slot[0]
 
+    # ------------------------------------------------------------------
+    # State migration (pipeline-utilities §10.12 / proposal 0014)
+    # ------------------------------------------------------------------
+
+    async def _migrate_record(
+        self,
+        record: CheckpointRecord,
+        checkpointer: Checkpointer,
+        invocation_id: str,
+        current_schema_version: str,
+    ) -> tuple[CheckpointRecord, _MigrationSummary]:
+        """Resolve a migration chain for ``record`` and apply it.
+
+        Returns ``(migrated_record, summary)``. ``migrated_record``
+        has ``state`` + ``parent_states`` mapped through the chain.
+        ``summary`` carries the chain's metadata so the caller can
+        dispatch a ``checkpoint_migrated`` observer event after the
+        invocation context exists (per the spec §6 cross-ref in
+        proposal 0014).
+
+        Caller is responsible for the post-migration deserialization
+        step (§10.12.4): if the migrated state cannot deserialize
+        against the current state class, the resulting failure
+        surfaces as ``CheckpointRecordInvalid``.
+
+        Spec §10.12.2 says "parent states MUST be treated as carrying
+        the same ``schema_version`` as the outer record." We apply
+        the same chain to every entry in ``parent_states`` lockstep
+        with the outer state. Future per-parent versioning would
+        need a spec follow-on.
+        """
+        # Eligibility check first per §10.12.1: backends that hold
+        # typed in-memory state or class-bound serialization cannot
+        # expose the class-independent intermediate the registry
+        # consumes. Mismatch + no eligibility → CheckpointRecordInvalid.
+        if not getattr(checkpointer, "supports_state_migration", False):
+            raise CheckpointRecordInvalid(
+                invocation_id,
+                f"persisted schema_version={record.schema_version!r} does not "
+                f"match current {current_schema_version!r}, and the active "
+                f"checkpointer ({type(checkpointer).__name__}) does not "
+                f"support state migration",
+            )
+
+        # resolve_chain raises CheckpointStateMigrationChainAmbiguous
+        # directly on multi-shortest-path detection per spec §10.10
+        # / §10.12.2 (proposal 0018, spec v0.16.0). No except-wrap
+        # needed here — the canonical category propagates straight
+        # through and the registry's exception contract is one type
+        # regardless of when ambiguity surfaces (register vs resolve).
+        chain = self.migration_registry.resolve_chain(
+            record.schema_version,
+            current_schema_version,
+        )
+
+        if chain is None:
+            raise CheckpointStateMigrationMissing(
+                f"no migration chain from {record.schema_version!r} to {current_schema_version!r}",
+                from_version=record.schema_version,
+                to_version=current_schema_version,
+                registered_migrations_count=len(self.migration_registry),
+                registry_description=self.migration_registry.describe(),
+            )
+
+        migrated_state: Any = record.state
+        migrated_parents: list[Any] = list(record.parent_states)
+        for migration in chain:
+            migrated_state = _apply_migration_step(migration, migrated_state, "state")
+            for i, parent in enumerate(migrated_parents):
+                migrated_parents[i] = _apply_migration_step(migration, parent, f"parent_states[{i}]")
+
+        # Per spec §6 cross-ref, the caller dispatches a synthetic
+        # ``checkpoint_migrated`` observer event using the summary
+        # below as soon as the invocation context exists. We can't
+        # dispatch from here because the context isn't built yet.
+        summary = _MigrationSummary(
+            from_version=record.schema_version,
+            to_version=current_schema_version,
+            chain_length=len(chain),
+        )
+        migrated = dataclass_replace(
+            record,
+            state=migrated_state,
+            parent_states=tuple(migrated_parents),
+        )
+        return migrated, summary
+
     async def drain(self) -> None:
         """Await delivery of every observer event produced by prior
         invocations of this graph.
@@ -423,6 +568,13 @@ async def invoke(
         resume_skip_set: frozenset[tuple[str, ...]] = frozenset()
         completed_positions: list[NodePosition] = []
         pending_resume_states: dict[int, Any] = {}
+        # Populated by ``_migrate_record`` during a version-mismatched
+        # resume; left ``None`` for the no-resume + versions-match
+        # paths. Dispatched as a synthetic ``checkpoint_migrated``
+        # observer event after the invocation context is built so
+        # the OTel observer can emit an
+        # ``openarmature.checkpoint.migrate`` span per spec §6.
+        migration_summary: _MigrationSummary | None = None
         if resume_invocation is not None:
             checkpointer = self._checkpointer_slot[0]
             if checkpointer is None:
@@ -433,11 +585,27 @@ async def invoke(
             record = await checkpointer.load(resume_invocation)
             if record is None:
                 raise CheckpointNotFound(resume_invocation)
-            if record.schema_version != CHECKPOINT_SCHEMA_VERSION:
-                raise CheckpointRecordInvalid(
+            # Per spec §10.12 (proposal 0014): version-mismatch resume.
+            # Routing precedence (per §10.10 + §10.12.1):
+            #   1. unsupported backend → CheckpointRecordInvalid.
+            #      Backends that hold typed in-memory state or
+            #      class-bound serialization can't expose the
+            #      class-independent intermediate the migration
+            #      registry needs.
+            #   2. no chain in the registry → CheckpointStateMigrationMissing.
+            #      Actionable: register a migration.
+            #   3. chain found but a migration raises →
+            #      CheckpointStateMigrationFailed.
+            #   4. post-migration state fails to deserialize →
+            #      CheckpointRecordInvalid (the §10.12.4 boundary).
+            # Order matters — do NOT swap eligibility and registry-lookup.
+            current_schema_version = self.state_cls.schema_version
+            if record.schema_version != current_schema_version:
+                record, migration_summary = await self._migrate_record(
+                    record,
+                    checkpointer,
                     resume_invocation,
-                    f"persisted schema_version={record.schema_version!r} "
-                    f"does not match current {CHECKPOINT_SCHEMA_VERSION!r}",
+                    current_schema_version,
                 )
             # The saved record's ``state`` is post-merge state at the
             # saving node's level (depth = len(parent_states)). For
@@ -516,6 +684,28 @@ async def invoke(
         # processed), remove it from the active set so long-running
         # services don't leak Task references between drain() calls.
         worker.add_done_callback(self._active_workers.discard)
+        # Per spec §6 cross-ref in proposal 0014: dispatch the
+        # ``checkpoint_migrated`` event as soon as the delivery
+        # worker is alive but before any node runs, so the OTel
+        # observer can emit an
+        # ``openarmature.checkpoint.migrate`` span ahead of the
+        # invocation's node spans. The synthetic event carries the
+        # ``_MigrationSummary`` on ``pre_state`` mirroring the
+        # ``checkpoint_saved`` convention (state-on-pre, post=None).
+        if migration_summary is not None:
+            _dispatch(
+                context,
+                NodeEvent(
+                    node_name="openarmature.checkpoint.migrate",
+                    namespace=("openarmature.checkpoint.migrate",),
+                    step=-1,
+                    phase="checkpoint_migrated",
+                    pre_state=migration_summary,
+                    post_state=None,
+                    error=None,
+                    parent_states=(),
+                ),
+            )
         try:
             return await self._invoke(starting_state, context)
         finally:
@@ -1319,8 +1509,15 @@ def _dispatch_completed(
             ),
         )
 
-    @staticmethod
+    # Instance method (not @staticmethod) so the save-time
+    # schema_version read goes through ``self.state_cls`` — matches
+    # the resume-side check, per spec §10.2's "framework reads
+    # schema_version from the state definition at save time"
+    # wording. Reading from ``type(post_state)`` would let a State
+    # subclass instance shadow the declared graph schema and
+    # trigger spurious migrations on resume.
     async def _maybe_save_checkpoint(
+        self,
         context: _InvocationContext,
         *,
         node_name: str,
@@ -1387,7 +1584,16 @@ async def _maybe_save_checkpoint(
             # ``step`` field on each NodePosition is the canonical
             # within-invocation order.
             last_saved_at=time.time(),
-            schema_version=CHECKPOINT_SCHEMA_VERSION,
+            # Per spec §10.2 (proposal 0014): read the user's
+            # state-schema version off the declared state class at
+            # save time. Empty-string sentinel when the user hasn't
+            # declared one — those records are not migration-eligible
+            # until they declare a non-empty version (per §10.2).
+            # ``self.state_cls`` is the authoritative source so the
+            # save-side read symmetrizes with the resume-side check
+            # (subclass schema_versions don't shadow the declared
+            # graph schema).
+            schema_version=self.state_cls.schema_version,
         )
         try:
             await checkpointer.save(context.invocation_id, record)
diff --git a/src/openarmature/graph/events.py b/src/openarmature/graph/events.py
index 599cb2b..cc05de9 100644
--- a/src/openarmature/graph/events.py
+++ b/src/openarmature/graph/events.py
@@ -14,7 +14,7 @@
 """
 
 from dataclasses import dataclass
-from typing import Literal
+from typing import Any, Literal
 
 from .errors import RuntimeGraphError
 from .state import State
@@ -122,13 +122,50 @@ class NodeEvent:
       be ``None``.
     - On ``completed`` events, exactly one of ``post_state`` and
       ``error`` is populated.
+
+    **Synthetic phases.** ``"checkpoint_saved"`` (pipeline-utilities
+    §10.8) and ``"checkpoint_migrated"`` (proposal 0014 §6
+    cross-ref) repurpose this dataclass for non-node events. Both
+    are opt-in via ``phases={...}`` on observer registration —
+    default subscriptions are ``{"started", "completed"}`` only, so
+    legacy observers never see them. Conventions on synthetic
+    events:
+
+    - ``checkpoint_saved``: ``pre_state`` carries the saved
+      post-merge state (still a real ``State`` instance for this
+      phase), ``post_state`` is ``None``. ``step`` matches the
+      saving node's step.
+    - ``checkpoint_migrated``: ``step=-1`` (no graph-step
+      sequencing — migrations run before any node fires).
+      ``node_name="openarmature.checkpoint.migrate"`` and
+      ``namespace=("openarmature.checkpoint.migrate",)`` are
+      dotted-pseudo identifiers, not real node names. ``pre_state``
+      carries a private ``_MigrationSummary`` dataclass with
+      ``from_version`` / ``to_version`` / ``chain_length``, NOT a
+      ``State`` instance. ``parent_states`` is the empty tuple.
+
+    Because ``pre_state`` is no longer guaranteed to be a ``State``
+    on the synthetic phases, its type is declared as ``Any`` and
+    observer authors who subscribe to those phases MUST narrow
+    per-phase before reading ``pre_state``.
     """
 
     node_name: str
     namespace: tuple[str, ...]
     step: int
-    phase: Literal["started", "completed", "checkpoint_saved"]
-    pre_state: State
+    phase: Literal[
+        "started",
+        "completed",
+        "checkpoint_saved",
+        # Synthetic phase per spec §6 cross-ref in proposal 0014:
+        # fires once at the start of a versioned resume to carry
+        # the migration chain's metadata. ``pre_state`` on this
+        # phase carries a ``_MigrationSummary`` (not a ``State``);
+        # the field type stays permissive on this dataclass and
+        # the OTel observer narrows defensively via ``isinstance``.
+        "checkpoint_migrated",
+    ]
+    pre_state: Any
     post_state: State | None
     error: RuntimeGraphError | None
     parent_states: tuple[State, ...]
diff --git a/src/openarmature/graph/observer.py b/src/openarmature/graph/observer.py
index b28b749..f36bf2a 100644
--- a/src/openarmature/graph/observer.py
+++ b/src/openarmature/graph/observer.py
@@ -95,9 +95,17 @@ async def __call__(self, event: NodeEvent, /) -> None: ...
 ALL_PHASES: frozenset[str] = frozenset({"started", "completed"})
 
 # All phase values the engine produces (per spec graph-engine §6 +
-# pipeline-utilities §10.8). Used by the registration-time validator
-# to reject typos like ``phases={"complete"}``.
-KNOWN_PHASES: frozenset[str] = frozenset({"started", "completed", "checkpoint_saved"})
+# pipeline-utilities §10.8 + proposal 0014 §6 cross-ref). Used by
+# the registration-time validator to reject typos like
+# ``phases={"complete"}``.
+#
+# The two synthetic phases (``checkpoint_saved`` and
+# ``checkpoint_migrated``) repurpose the ``NodeEvent`` shape for
+# non-node events — see the ``NodeEvent`` docstring for conventions.
+# Both are opt-in via explicit ``phases={...}``; the default
+# subscription ``ALL_PHASES`` above is ``{"started", "completed"}``
+# only, so legacy observers never receive them.
+KNOWN_PHASES: frozenset[str] = frozenset({"started", "completed", "checkpoint_saved", "checkpoint_migrated"})
 
 
 @dataclass(frozen=True)
diff --git a/src/openarmature/graph/state.py b/src/openarmature/graph/state.py
index 746dd36..6ac9520 100644
--- a/src/openarmature/graph/state.py
+++ b/src/openarmature/graph/state.py
@@ -16,6 +16,7 @@ class S(State):
 """
 
 from collections.abc import Mapping
+from typing import ClassVar
 
 from pydantic import BaseModel, ConfigDict
 
@@ -31,6 +32,16 @@ class State(BaseModel):
     # boundaries" intent.
     model_config = ConfigDict(frozen=True, extra="forbid")
 
+    # User-controlled state-schema version per pipeline-utilities
+    # spec §10.2 (proposal 0014). Empty-string sentinel for state
+    # classes that don't declare a version. The framework reads
+    # ``type(state).schema_version`` at save time and writes it
+    # onto the CheckpointRecord. Declaring a non-empty value opts
+    # the state class into the migration registry: on resume,
+    # records carrying a different version route through the
+    # registered migrations per spec §10.12.
+    schema_version: ClassVar[str] = ""
+
 
 def field_reducers(state_cls: type[State]) -> Mapping[str, list[Reducer]]:
     """Return `{field_name: [declared reducers]}` for each field on `state_cls`.
diff --git a/src/openarmature/observability/otel/observer.py b/src/openarmature/observability/otel/observer.py
index d94e6cf..3e25f80 100644
--- a/src/openarmature/observability/otel/observer.py
+++ b/src/openarmature/observability/otel/observer.py
@@ -258,6 +258,9 @@ async def __call__(self, event: NodeEvent) -> None:
         if event.phase == "checkpoint_saved":
             self._emit_checkpoint_save_span(event)
             return
+        if event.phase == "checkpoint_migrated":
+            self._emit_checkpoint_migrate_span(event)
+            return
         if event.phase == "started":
             # Idempotent — short-circuits inside ``_open_started_span``
             # if ``prepare_sync`` already opened the span synchronously
@@ -429,6 +432,61 @@ def _handle_completed(self, event: NodeEvent) -> None:
     # Special-event paths
     # ------------------------------------------------------------------
 
+    def _emit_checkpoint_migrate_span(self, event: NodeEvent) -> None:
+        """Spec pipeline-utilities §6 cross-ref (proposal 0014): emit a
+        zero-duration ``openarmature.checkpoint.migrate`` span when
+        a versioned resume's migration chain runs. The synthetic
+        event carries ``_MigrationSummary`` on ``pre_state``; this
+        handler reads ``from_version`` / ``to_version`` /
+        ``chain_length`` from the summary onto the span.
+
+        Emitted under the invocation's root span (no parent-node
+        context — the migration runs before any node fires), so
+        trace UIs surface it as the first child of the invocation.
+        """
+        from openarmature.graph.compiled import _MigrationSummary
+        from openarmature.observability.correlation import (
+            current_correlation_id,
+            current_invocation_id,
+        )
+
+        invocation_id = current_invocation_id()
+        if invocation_id is None:
+            return
+        summary = event.pre_state
+        if not isinstance(summary, _MigrationSummary):
+            # Defensive — the engine only sets pre_state to a
+            # _MigrationSummary on this phase. Skip if something
+            # else dispatched a checkpoint_migrated event.
+            return
+        # Open (or reuse) the invocation's root span and parent the
+        # migrate span under it. The migration runs before any node
+        # fires, so the root is the natural parent — no node span
+        # exists yet at this point in the invocation.
+        if invocation_id not in self._invocation_span:
+            cid = current_correlation_id() or ""
+            self._open_invocation_span(invocation_id, cid, event)
+        root_open = self._invocation_span.get(invocation_id)
+        parent_ctx: Any = None
+        if root_open is not None:
+            parent_ctx = set_span_in_context(root_open.span)
+        attrs: dict[str, Any] = {
+            "openarmature.checkpoint.migrate.from_version": summary.from_version,
+            "openarmature.checkpoint.migrate.to_version": summary.to_version,
+            "openarmature.checkpoint.migrate.chain_length": summary.chain_length,
+        }
+        cid = current_correlation_id()
+        if cid is not None:
+            attrs["openarmature.correlation_id"] = cid
+        span = self._tracer.start_span(
+            name="openarmature.checkpoint.migrate",
+            context=parent_ctx,
+            kind=SpanKind.INTERNAL,
+            attributes=attrs,
+        )
+        span.set_status(Status(StatusCode.OK))
+        span.end()
+
     def _emit_checkpoint_save_span(self, event: NodeEvent) -> None:
         """Spec pipeline-utilities §10.8 + observability §4.5: emit a
         zero-duration ``openarmature.checkpoint.save`` span attached
diff --git a/tests/conformance/harness/directives.py b/tests/conformance/harness/directives.py
index b2824f4..2a3b40f 100644
--- a/tests/conformance/harness/directives.py
+++ b/tests/conformance/harness/directives.py
@@ -56,16 +56,27 @@ class StateFieldSpec(_ForbidExtras):
     The ``alt_reducer`` knob exists only for ``graph-engine/007-compile-errors``'s
     ``conflicting_reducers`` case — fixtures intentionally declare two reducers
     on one field to verify the engine fails compile with the right category.
+
+    The ``required`` knob (used by the state-migration deserialization-
+    failure fixture 044) marks a field as having no default — Pydantic's
+    natural "required" shape. The default-when-omitted falls through
+    via the ``default`` field above.
     """
 
     type: str
     default: Any = None
     reducer: str | None = None
     alt_reducer: str | None = None
+    required: bool = False
 
 
 class StateSchema(_ForbidExtras):
     fields: dict[str, StateFieldSpec]
+    # User-facing state-schema version per pipeline-utilities §10.2
+    # (proposal 0014). The state-migration fixtures (039-046) declare
+    # this on each case's ``state`` block; non-migration fixtures
+    # omit it (defaults to empty-string sentinel).
+    schema_version: str = ""
 
 
 # ---------------------------------------------------------------------------
diff --git a/tests/conformance/test_fixture_parsing.py b/tests/conformance/test_fixture_parsing.py
index c9a145a..128b185 100644
--- a/tests/conformance/test_fixture_parsing.py
+++ b/tests/conformance/test_fixture_parsing.py
@@ -40,17 +40,10 @@ def _id(case: tuple[str, Path]) -> str:
     "pipeline-utilities/036-parallel-branches-with-branch-middleware-retry": "0011 parallel branches (PR-5)",
     "pipeline-utilities/037-parallel-branches-determinism": "0011 parallel branches (PR-5)",
     "pipeline-utilities/038-parallel-branches-compose-with-fan-out": "0011 parallel branches (PR-5)",
-    # proposal 0014 — state migration (PR-4)
-    "pipeline-utilities/039-state-migration-additive-field": "0014 state migration (PR-4)",
-    "pipeline-utilities/040-state-migration-chain": "0014 state migration (PR-4)",
-    "pipeline-utilities/041-state-migration-missing": "0014 state migration (PR-4)",
-    "pipeline-utilities/042-state-migration-versions-match-no-op": "0014 state migration (PR-4)",
-    "pipeline-utilities/043-state-migration-parent-states-migrated": "0014 state migration (PR-4)",
-    "pipeline-utilities/044-state-migration-post-migration-deserialization-fails": (
-        "0014 state migration (PR-4)"
-    ),
-    "pipeline-utilities/045-state-migration-no-path-in-registry": "0014 state migration (PR-4)",
-    "pipeline-utilities/046-state-migration-function-raises": "0014 state migration (PR-4)",
+    # proposal 0014's state-migration fixtures (039-046) were removed
+    # from this list as part of PR-4; the CasesFixture model already
+    # parses the seeded_record / migrations shape via its permissive
+    # extras (CaseSpec uses ``model_config = ConfigDict(extra="allow")``).
     # proposal 0015's llm-provider fixtures (009-020) were removed
     # from this list as part of PR-2; the typed harness parses the
     # content-block message shape via LlmCallSpec's permissive
diff --git a/tests/conformance/test_state_migration.py b/tests/conformance/test_state_migration.py
new file mode 100644
index 0000000..6f9f30a
--- /dev/null
+++ b/tests/conformance/test_state_migration.py
@@ -0,0 +1,465 @@
+"""Run every spec state-migration conformance fixture (039-047) end-to-end.
+
+The fixtures live under
+``spec/pipeline-utilities/conformance/`` as ``cases`` shapes; each
+case defines a state schema (with a ``schema_version``), an entry
+node and edges, a ``seeded_record:`` describing a checkpoint that
+was saved at a prior schema version, a ``migrations:`` list (each
+naming one of the harness mock functions), and a ``resume`` block
+specifying either an ``expected`` happy-path or an
+``expected_error`` raise.
+
+Fixture 047 (``state-migration-chain-ambiguous``, added in spec
+v0.16.0 / proposal 0018) exercises the
+``expected_chain_ambiguity_error`` harness primitive that accepts
+the canonical ``checkpoint_state_migration_chain_ambiguous``
+category at EITHER build time (duplicate-pair registration) or
+resume time (multi-shortest-path detection) per spec §10.12.2's
+compile-time-SHOULD / load-time-acceptable carve-out.
+
+The driver:
+
+1. Builds a State subclass via ``adapter.build_state_cls``, then
+   patches the generated class's ``schema_version`` ClassVar so
+   it matches the fixture's declared version.
+2. Builds a minimal graph (entry node + edge to END) via the
+   existing adapter primitives.
+3. Resolves each ``migrations[i].migrate`` name against the mock
+   library, wrapping every mock so the harness can count
+   invocations + record the ``v1->v2`` ordered list (for the
+   ``migrations_run`` / ``migration_count`` /
+   ``migration_order_matches_chain`` assertions).
+4. Seeds a ``CheckpointRecord`` via the configured backend
+   (SQLite in JSON mode) using a stable seeded ``invocation_id``.
+5. Calls ``invoke(resume_invocation=<seeded id>)`` and asserts
+   against ``resume.expected`` (final state, migrations_run,
+   invariants) OR ``resume.expected_error`` (category, carries,
+   cause) OR ``expected_chain_ambiguity_error`` at either case
+   top-level (build-time) or inside ``resume:`` (load-time).
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any, cast
+
+import pytest
+import yaml
+
+from openarmature.checkpoint import (
+    CheckpointError,
+    CheckpointRecord,
+    CheckpointRecordInvalid,
+    CheckpointStateMigrationChainAmbiguous,
+    CheckpointStateMigrationFailed,
+    CheckpointStateMigrationMissing,
+    SQLiteCheckpointer,
+)
+from openarmature.graph import END, GraphBuilder, State
+
+from .adapter import build_state_cls
+
+CONFORMANCE_DIR = (
+    Path(__file__).resolve().parents[2] / "openarmature-spec" / "spec" / "pipeline-utilities" / "conformance"
+)
+
+_FIXTURE_RANGE = range(39, 48)
+
+# ---------------------------------------------------------------------------
+# Migration mock library (harness-side; fixtures refer by name)
+# ---------------------------------------------------------------------------
+
+# Wrapped mocks all receive a dict (JSON-mode SQLite hands the engine the
+# structural intermediate form). Each returns a dict suitable as input
+# for the next migration in the chain (or for final deserialization).
+
+
+def _add_new_field_default(state: Any) -> Any:
+    out = dict(state)
+    out["new_field"] = "v2_default"
+    return out
+
+
+def _add_v2_field(state: Any) -> Any:
+    out = dict(state)
+    out["v2_field"] = "v2_default"
+    return out
+
+
+def _add_v3_field(state: Any) -> Any:
+    out = dict(state)
+    out["v3_field"] = "v3_default"
+    return out
+
+
+def _identity_passthrough(state: Any) -> Any:
+    # Used by 044 to verify post-migration deserialization-failure
+    # routing: the migration runs cleanly but produces output that
+    # the v2 state class can't deserialize (missing a required field).
+    return dict(state)
+
+
+def _raises_keyerror(_state: Any) -> Any:
+    # Used by 046 to verify CheckpointStateMigrationFailed routing.
+    raise KeyError("simulated buggy migration")
+
+
+def _should_not_run(_state: Any) -> Any:
+    # Used by 042 (versions-match no-op) to verify the engine does
+    # NOT consult the migration registry when versions match.
+    raise AssertionError("fixture 042 invariant violated: should_not_run was called despite version match")
+
+
+def _irrelevant(state: Any) -> Any:
+    # Used by 045 — migration is registered but the engine doesn't
+    # find a path to it. Returns input unchanged.
+    return dict(state)
+
+
+_MOCK_LIBRARY: dict[str, Any] = {
+    "add_new_field_default": _add_new_field_default,
+    "add_v2_field": _add_v2_field,
+    "add_v3_field": _add_v3_field,
+    "identity_passthrough": _identity_passthrough,
+    "raises_keyerror": _raises_keyerror,
+    "should_not_run": _should_not_run,
+    "irrelevant": _irrelevant,
+}
+
+
+class _MigrationTrace:
+    """Captures the order migrations were invoked in, for the
+    ``migrations_run`` / ``migration_count`` /
+    ``migration_order_matches_chain`` fixture assertions."""
+
+    def __init__(self) -> None:
+        self.order: list[str] = []
+
+    def wrap(self, name: str, fn: Any, from_v: str, to_v: str) -> Any:
+        label = f"{from_v}->{to_v}"
+
+        def _traced(state: Any) -> Any:
+            self.order.append(label)
+            return fn(state)
+
+        _traced.__name__ = f"_traced_{name}"
+        return _traced
+
+
+# ---------------------------------------------------------------------------
+# Fixture discovery
+# ---------------------------------------------------------------------------
+
+
+def _fixture_paths() -> list[Path]:
+    out: list[Path] = []
+    for p in sorted(CONFORMANCE_DIR.glob("[0-9][0-9][0-9]-*.yaml")):
+        try:
+            number = int(p.stem.split("-", 1)[0])
+        except ValueError:
+            continue
+        if number in _FIXTURE_RANGE:
+            out.append(p)
+    return out
+
+
+def _fixture_id(path: Path) -> str:
+    return path.stem
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _build_state_cls(state_spec: dict[str, Any], model_name: str) -> type[State]:
+    """Build a Pydantic State subclass from the fixture spec and stamp
+    its ``schema_version`` ClassVar with the declared version."""
+    fields_spec = state_spec.get("fields", {})
+    state_cls = build_state_cls(model_name, fields_spec)
+    # Stamp the per-fixture schema_version. The adapter's build_state_cls
+    # produces a fresh subclass via pydantic.create_model so it's safe
+    # to set the ClassVar after construction.
+    state_cls.schema_version = state_spec.get("schema_version", "")
+    return state_cls
+
+
+async def _seed_record(
+    checkpointer: SQLiteCheckpointer,
+    invocation_id: str,
+    seeded: dict[str, Any],
+) -> None:
+    """Persist a checkpoint record matching the fixture's
+    ``seeded_record:`` block. ``state`` and ``parent_states`` go
+    through as plain dicts (JSON-mode round-trip)."""
+    raw_positions: list[dict[str, Any]] = seeded.get("completed_positions", [])
+    from openarmature.checkpoint import NodePosition
+
+    positions = tuple(
+        NodePosition(
+            namespace=tuple(p.get("namespace", [])),
+            node_name=p["node_name"],
+            step=p["step"],
+            attempt_index=p.get("attempt_index", 0),
+            fan_out_index=p.get("fan_out_index"),
+        )
+        for p in raw_positions
+    )
+    record = CheckpointRecord(
+        invocation_id=invocation_id,
+        correlation_id=seeded.get("correlation_id", "seeded-corr"),
+        state=seeded["state"],
+        completed_positions=positions,
+        parent_states=tuple(seeded.get("parent_states", [])),
+        last_saved_at=0.0,
+        schema_version=seeded.get("schema_version", ""),
+    )
+    await checkpointer.save(invocation_id, record)
+
+
+# ---------------------------------------------------------------------------
+# Case runner
+# ---------------------------------------------------------------------------
+
+
+async def _run_one_case(case: dict[str, Any], tmp_path: Path) -> None:
+    """Run one fixture case end-to-end: build, seed, resume, assert.
+
+    Fixture 047 (chain-ambiguous) introduces a new primitive:
+    ``expected_chain_ambiguity_error`` that can land at EITHER the
+    case top level (registration-time detection: the migration set
+    raises on registration so the ``resume:`` block is unreachable)
+    OR inside ``resume:`` (load-time detection: registration
+    succeeds and BFS raises during the resume attempt). The driver
+    wraps both phases in try/except so the canonical category
+    surfaces from either timing per spec §10.12.2's compile-time-
+    SHOULD / load-time-acceptable carve-out.
+    """
+    state_cls = _build_state_cls(case["state"], model_name=f"Case_{case['name']}")
+
+    case_level_ambiguity = case.get("expected_chain_ambiguity_error")
+    resume_level_ambiguity = case.get("resume", {}).get("expected_chain_ambiguity_error")
+    expected_ambiguity = case_level_ambiguity or resume_level_ambiguity
+
+    # Minimal node body: apply ``update_pure`` to the state.
+    nodes_spec = case["nodes"]
+    edges_spec = case["edges"]
+    entry = case["entry"]
+
+    builder = GraphBuilder(state_cls)
+    for node_name, node_spec in nodes_spec.items():
+        update_pure = cast("dict[str, Any]", node_spec.get("update_pure", {}))
+
+        async def _node_body(_s: State, _u: dict[str, Any] = update_pure) -> dict[str, Any]:
+            return _u
+
+        builder.add_node(node_name, _node_body)
+    for edge in edges_spec:
+        target_raw = edge["to"]
+        target = END if target_raw == "END" else target_raw
+        builder.add_edge(edge["from"], target)
+    builder.set_entry(entry)
+
+    # Register migrations + wrap each in the trace recorder. The
+    # duplicate-pair ambiguity case raises during this loop.
+    trace = _MigrationTrace()
+    raised: BaseException | None = None
+    try:
+        for m in case.get("migrations", []):
+            mock = _MOCK_LIBRARY[m["migrate"]]
+            wrapped = trace.wrap(m["migrate"], mock, m["from_version"], m["to_version"])
+            builder.with_state_migration(m["from_version"], m["to_version"], wrapped)
+    except CheckpointError as exc:
+        raised = exc
+
+    if raised is not None:
+        if expected_ambiguity is not None:
+            _assert_ambiguity(expected_ambiguity, raised)
+            return
+        raise AssertionError(f"unexpected build-time raise: {raised!r}")
+
+    # Configure the SQLite backend in JSON mode (the migration-eligible
+    # backend per spec §10.12.1). One database file per case, isolated
+    # under tmp_path.
+    db_path = tmp_path / f"{case['name']}.db"
+    checkpointer = SQLiteCheckpointer(db_path, serialization="json")
+    builder.with_checkpointer(checkpointer)
+    compiled = builder.compile()
+
+    # Seed the prior record.
+    invocation_id = f"seeded-{case['name']}"
+    if "seeded_record" in case:
+        await _seed_record(checkpointer, invocation_id, case["seeded_record"])
+
+    # Resume + assert. Some fixtures (047 case 1, the duplicate-pair
+    # registration case) have no ``resume:`` block — the ambiguity
+    # fires at build time, above. For the resume-side cases the
+    # engine loads state from the checkpoint and never reads
+    # ``initial_state``; ``model_construct`` skips Pydantic
+    # validation so fixtures with required fields (044) can construct
+    # a placeholder without tripping the validator before the resume
+    # even starts.
+    resume = case.get("resume")
+    if resume is None:
+        if expected_ambiguity is not None:
+            raise AssertionError(
+                f"expected build-time chain-ambiguity raise ({expected_ambiguity!r}) "
+                f"but registration completed without raising"
+            )
+        return
+
+    initial_state = state_cls.model_construct()
+    final_state: Any = None
+    try:
+        if resume.get("from_seeded_record"):
+            final_state = await compiled.invoke(initial_state, resume_invocation=invocation_id)
+        else:
+            final_state = await compiled.invoke(initial_state)
+    except CheckpointError as exc:
+        raised = exc
+
+    if expected_ambiguity is not None:
+        _assert_ambiguity(expected_ambiguity, raised)
+        return
+    if "expected_error" in resume:
+        _assert_error(resume["expected_error"], resume.get("invariants", {}), raised)
+    elif "expected" in resume:
+        assert raised is None, f"expected success, got {raised!r}"
+        _assert_success(resume["expected"], resume.get("invariants", {}), final_state, trace)
+
+
+def _assert_ambiguity(expected_category: str, raised: BaseException | None) -> None:
+    """Verify the case raised the canonical chain-ambiguity category
+    at either build time or resume time."""
+    assert raised is not None, (
+        f"expected chain-ambiguity raise of category {expected_category!r}, got no exception"
+    )
+    actual_category = getattr(raised, "category", None)
+    assert actual_category == expected_category, (
+        f"expected category {expected_category!r}, got {actual_category!r} ({raised!r})"
+    )
+
+
+def _assert_error(
+    expected_error: dict[str, Any],
+    invariants: dict[str, Any],
+    raised: BaseException | None,
+) -> None:
+    assert raised is not None, (
+        f"expected raise of category {expected_error.get('category')!r}, got no exception"
+    )
+    actual_category = getattr(raised, "category", None)
+    assert actual_category == expected_error["category"], (
+        f"expected category {expected_error['category']!r}, got {actual_category!r} ({raised!r})"
+    )
+    for key, expected_value in expected_error.get("carries", {}).items():
+        actual_attr = getattr(raised, key, None)
+        assert actual_attr == expected_value, f"expected {key}={expected_value!r}, got {actual_attr!r}"
+    cause_spec = expected_error.get("cause")
+    if cause_spec is not None:
+        cause = raised.__cause__
+        assert cause is not None, "expected __cause__ to be populated"
+        assert type(cause).__name__ == cause_spec["exception_type"], (
+            f"expected __cause__ type {cause_spec['exception_type']!r}, got {type(cause).__name__!r}"
+        )
+    forbidden_categories = invariants.get("error_category_not", [])
+    for forbidden in forbidden_categories:
+        assert actual_category != forbidden, (
+            f"invariant violated: error category {forbidden!r} forbidden but raised"
+        )
+
+
+def _assert_success(
+    expected: dict[str, Any],
+    invariants: dict[str, Any],
+    final_state: Any,
+    trace: _MigrationTrace,
+) -> None:
+    expected_final_state = expected.get("final_state")
+    if expected_final_state is not None:
+        actual = final_state.model_dump()
+        for key, value in expected_final_state.items():
+            assert actual.get(key) == value, f"final_state.{key}: expected {value!r}, got {actual.get(key)!r}"
+
+    migrations_run = expected.get("migrations_run")
+    if migrations_run is not None:
+        # Per the spec's lockstep ordering, each migration step runs
+        # once for the outer state. The fixtures only seed parent_states
+        # on fixture 043; for the simpler fixtures, the order list
+        # tracks a single application of each step. fixture 043's
+        # lockstep semantics produce 2 invocations per migration step
+        # (once for outer, once for parent) — collapse the consecutive
+        # duplicates for the comparison so the order assertion stays
+        # "each step ran in chain order."
+        dedup_consecutive: list[str] = []
+        for label in trace.order:
+            if not dedup_consecutive or dedup_consecutive[-1] != label:
+                dedup_consecutive.append(label)
+        assert dedup_consecutive == migrations_run, (
+            f"migrations_run: expected {migrations_run!r}, got {dedup_consecutive!r} "
+            f"(raw invocation order: {trace.order!r})"
+        )
+
+    expected_count = invariants.get("migration_count")
+    if expected_count is not None:
+        # ``migration_count`` counts distinct migration steps in the
+        # chain (each step applied to outer + parents counts once per
+        # step). Collapse consecutive duplicates the same way.
+        dedup_consecutive_for_count: list[str] = []
+        for label in trace.order:
+            if not dedup_consecutive_for_count or dedup_consecutive_for_count[-1] != label:
+                dedup_consecutive_for_count.append(label)
+        assert len(dedup_consecutive_for_count) == expected_count, (
+            f"migration_count: expected {expected_count}, got "
+            f"{len(dedup_consecutive_for_count)} ({dedup_consecutive_for_count!r})"
+        )
+
+    if invariants.get("single_migration_invocation"):
+        # Each migration step ran once per state-or-parent-state entry.
+        # For fixture 039 there's no parent_states, so the dedup count
+        # equals the raw count and equals 1 (one migration step).
+        assert len(trace.order) == 1, (
+            f"single_migration_invocation: expected 1 invocation, got {len(trace.order)} ({trace.order!r})"
+        )
+
+    if invariants.get("migration_order_matches_chain"):
+        # The chain ordering is captured implicitly by the migrations_run
+        # comparison above; this is a redundant invariant the fixture
+        # surfaces explicitly. No-op here since migrations_run carries
+        # the assertion.
+        pass
+
+
+# ---------------------------------------------------------------------------
+# Parametrized driver
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize("fixture_path", _fixture_paths(), ids=_fixture_id)
+async def test_state_migration_fixture(fixture_path: Path, tmp_path: Path) -> None:
+    """Parametrize across 039-046; each case in a fixture's ``cases``
+    list runs as a sub-case under one parametrize id (matching how
+    test_checkpoint.py handles the cases-shape fixtures).
+    """
+    with fixture_path.open() as f:
+        spec = cast("dict[str, Any]", yaml.safe_load(f))
+    cases = spec.get("cases", [])
+    for case in cases:
+        case_name = case.get("name", "<unnamed>")
+        try:
+            await _run_one_case(case, tmp_path)
+        except AssertionError as exc:
+            raise AssertionError(f"case {case_name!r}: {exc}") from exc
+
+
+# Make sure the imports for the error categories are reachable from
+# the test module (pyright would flag them as unused otherwise; we
+# reference them via the spec-error-category strings inside the
+# fixtures, not by class identity, so the imports are load-bearing
+# for the package-public API surface verification).
+_ = (
+    CheckpointRecordInvalid,
+    CheckpointStateMigrationChainAmbiguous,
+    CheckpointStateMigrationFailed,
+    CheckpointStateMigrationMissing,
+)
diff --git a/tests/test_smoke.py b/tests/test_smoke.py
index 19f0ef0..f6fc125 100644
--- a/tests/test_smoke.py
+++ b/tests/test_smoke.py
@@ -9,7 +9,7 @@
 
 def test_package_versions() -> None:
     assert openarmature.__version__ == "0.5.0"
-    assert openarmature.__spec_version__ == "0.15.0"
+    assert openarmature.__spec_version__ == "0.16.0"
 
 
 def test_spec_version_matches_pyproject() -> None:
diff --git a/tests/unit/test_checkpoint.py b/tests/unit/test_checkpoint.py
index e9c9664..ae3fafb 100644
--- a/tests/unit/test_checkpoint.py
+++ b/tests/unit/test_checkpoint.py
@@ -20,7 +20,6 @@
 from pydantic import Field
 
 from openarmature.checkpoint import (
-    CHECKPOINT_SCHEMA_VERSION,
     Checkpointer,
     CheckpointFilter,
     CheckpointNotFound,
@@ -92,7 +91,10 @@ def test_checkpoint_record_default_schema_version() -> None:
         parent_states=(),
         last_saved_at=0.0,
     )
-    assert record.schema_version == CHECKPOINT_SCHEMA_VERSION
+    # Default schema_version is the empty-string sentinel per spec
+    # §10.2 (proposal 0014): records carry the user's state-schema
+    # version, which is "" until the state class declares one.
+    assert record.schema_version == ""
     assert record.fan_out_progress is None
 
 
@@ -305,13 +307,18 @@ async def test_schema_version_round_trips(tmp_path: Path) -> None:
     await cp.save("i", record)
     loaded = await cp.load("i")
     assert loaded is not None
-    assert loaded.schema_version == CHECKPOINT_SCHEMA_VERSION
-
-
-async def test_schema_version_mismatch_rejected_by_sqlite(tmp_path: Path) -> None:
-    """A persisted record with an unrecognized schema_version raises
-    CheckpointRecordInvalid on load — sentinel behavior so future
-    record-shape evolution doesn't silently coerce older records."""
+    # Records round-trip the user-facing schema_version verbatim per
+    # spec §10.2. With no version declared on the saved state, the
+    # default sentinel is "".
+    assert loaded.schema_version == ""
+
+
+async def test_schema_version_round_trips_through_sqlite_unchanged(tmp_path: Path) -> None:
+    """Per spec §10.12 (proposal 0014), the SQLite backend no longer
+    rejects records with non-default ``schema_version`` values — that
+    routing is now an engine concern at resume time. The backend
+    just round-trips the version identifier as opaque data so the
+    engine's migration registry has the chance to bridge it."""
     cp = SQLiteCheckpointer(tmp_path / "ck.db")
     record = CheckpointRecord(
         invocation_id="i",
@@ -320,11 +327,12 @@ async def test_schema_version_mismatch_rejected_by_sqlite(tmp_path: Path) -> Non
         completed_positions=(),
         parent_states=(),
         last_saved_at=1.0,
-        schema_version="999",  # not the current version
+        schema_version="999",  # an arbitrary user-facing identifier
     )
     await cp.save("i", record)
-    with pytest.raises(CheckpointRecordInvalid):
-        await cp.load("i")
+    loaded = await cp.load("i")
+    assert loaded is not None
+    assert loaded.schema_version == "999"
 
 
 # ---------------------------------------------------------------------------
@@ -424,6 +432,8 @@ class _AlwaysFailingCheckpointer:
     as :class:`CheckpointSaveFailed` and raises immediately to the
     caller of ``invoke()`` per the documented save-failure policy."""
 
+    supports_state_migration: bool = False
+
     async def save(self, invocation_id: str, record: CheckpointRecord) -> None:
         raise RuntimeError("simulated backend failure")
 
@@ -449,6 +459,8 @@ async def test_save_failure_raises_to_invoke_caller() -> None:
 
 
 class _CapturingCheckpointer:
+    supports_state_migration: bool = False
+
     def __init__(self) -> None:
         self.saves: list[CheckpointRecord] = []
 
diff --git a/tests/unit/test_observability_otel.py b/tests/unit/test_observability_otel.py
index 4c46fe9..9573fc5 100644
--- a/tests/unit/test_observability_otel.py
+++ b/tests/unit/test_observability_otel.py
@@ -22,6 +22,7 @@
 from __future__ import annotations
 
 import logging
+from pathlib import Path
 
 import pytest
 from pydantic import Field
@@ -217,6 +218,120 @@ async def test_failing_node_span_carries_error_status() -> None:
 # ---------------------------------------------------------------------------
 
 
+async def test_checkpoint_migrate_emits_span_with_chain_metadata(tmp_path: Path) -> None:
+    """Spec §6 cross-ref in proposal 0014: a versioned resume whose
+    migration chain runs SHOULD emit an
+    ``openarmature.checkpoint.migrate`` span carrying
+    ``from_version`` / ``to_version`` (final) / ``chain_length``."""
+    from openarmature.checkpoint import (
+        CheckpointRecord,
+        SQLiteCheckpointer,
+    )
+
+    # JSON-mode SQLite is migration-eligible (the dict-state form the
+    # registry consumes is what the load path produces).
+    cp = SQLiteCheckpointer(tmp_path / "ck.db", serialization="json")
+
+    class _MigState(State):
+        schema_version = "v2"
+        x: int = 0
+        new_field: str = "v2_default"
+
+    async def _noop(_s: _MigState) -> dict[str, int]:
+        return {}
+
+    # Seed a v1 record so the resume triggers the v1→v2 migration.
+    invocation_id = "mig-resume"
+    await cp.save(
+        invocation_id,
+        CheckpointRecord(
+            invocation_id=invocation_id,
+            correlation_id="cid",
+            state={"x": 9},
+            completed_positions=(),
+            parent_states=(),
+            last_saved_at=0.0,
+            schema_version="v1",
+        ),
+    )
+
+    exporter = InMemorySpanExporter()
+    observer = OTelObserver(span_processor=SimpleSpanProcessor(exporter))
+    g = (
+        GraphBuilder(_MigState)
+        .add_node("noop", _noop)
+        .add_edge("noop", END)
+        .set_entry("noop")
+        .with_checkpointer(cp)
+        .with_state_migration("v1", "v2", lambda s: {**s, "new_field": "v2_default"})
+        .compile()
+    )
+    g.attach_observer(observer, phases={"started", "completed", "checkpoint_migrated"})
+    await g.invoke(
+        _MigState.model_construct(),
+        resume_invocation=invocation_id,
+    )
+    await g.drain()
+    observer.shutdown()
+
+    migrate_spans = [s for s in exporter.get_finished_spans() if s.name == "openarmature.checkpoint.migrate"]
+    assert len(migrate_spans) == 1
+    span = migrate_spans[0]
+    attrs = dict(span.attributes or {})
+    assert attrs.get("openarmature.checkpoint.migrate.from_version") == "v1"
+    assert attrs.get("openarmature.checkpoint.migrate.to_version") == "v2"
+    assert attrs.get("openarmature.checkpoint.migrate.chain_length") == 1
+
+
+async def test_checkpoint_migrate_span_absent_on_version_match(tmp_path: Path) -> None:
+    """Spec §10.12.3 fast path: when the saved record's schema_version
+    equals the current state class's schema_version, the migration
+    registry is NOT consulted. The OTel observer MUST NOT emit a
+    ``openarmature.checkpoint.migrate`` span in that case."""
+    from openarmature.checkpoint import CheckpointRecord, SQLiteCheckpointer
+
+    cp = SQLiteCheckpointer(tmp_path / "ck.db", serialization="json")
+
+    class _MatchState(State):
+        schema_version = "v1"
+        x: int = 0
+
+    async def _noop(_s: _MatchState) -> dict[str, int]:
+        return {}
+
+    invocation_id = "match-resume"
+    await cp.save(
+        invocation_id,
+        CheckpointRecord(
+            invocation_id=invocation_id,
+            correlation_id="cid",
+            state={"x": 7},
+            completed_positions=(),
+            parent_states=(),
+            last_saved_at=0.0,
+            schema_version="v1",  # matches current class
+        ),
+    )
+
+    exporter = InMemorySpanExporter()
+    observer = OTelObserver(span_processor=SimpleSpanProcessor(exporter))
+    g = (
+        GraphBuilder(_MatchState)
+        .add_node("noop", _noop)
+        .add_edge("noop", END)
+        .set_entry("noop")
+        .with_checkpointer(cp)
+        .compile()
+    )
+    g.attach_observer(observer, phases={"started", "completed", "checkpoint_migrated"})
+    await g.invoke(_MatchState.model_construct(), resume_invocation=invocation_id)
+    await g.drain()
+    observer.shutdown()
+
+    migrate_spans = [s for s in exporter.get_finished_spans() if s.name == "openarmature.checkpoint.migrate"]
+    assert migrate_spans == []
+
+
 async def test_checkpoint_save_emits_zero_duration_span() -> None:
     """Spec §10.8: a checkpoint save SHOULD emit a §6-style observer
     event surfaced as a span. Our implementation emits a
diff --git a/tests/unit/test_state_migration.py b/tests/unit/test_state_migration.py
new file mode 100644
index 0000000..efd100e
--- /dev/null
+++ b/tests/unit/test_state_migration.py
@@ -0,0 +1,307 @@
+"""Focused unit tests for the state-migration surface.
+
+The conformance suite (``tests/conformance/test_state_migration.py``)
+covers the spec's behavioral surface end-to-end against fixtures
+039-046. These unit tests fill gaps the fixtures don't exercise
+directly: BFS edge cases on the registry, multi-shortest-path
+ambiguity detection, GraphBuilder ergonomics, and the error
+attribute carriage shape.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from openarmature.checkpoint import (
+    CheckpointStateMigrationChainAmbiguous,
+    CheckpointStateMigrationFailed,
+    CheckpointStateMigrationMissing,
+    MigrationRegistry,
+    StateMigration,
+)
+from openarmature.graph import END, GraphBuilder, State
+
+# ---------------------------------------------------------------------------
+# MigrationRegistry — basic registration + iteration + describe
+# ---------------------------------------------------------------------------
+
+
+def _id(x: int) -> int:
+    return x
+
+
+def test_registry_empty_describes_with_sentinel() -> None:
+    registry = MigrationRegistry()
+    assert len(registry) == 0
+    assert registry.describe() == "<no migrations registered>"
+
+
+def test_registry_lists_registered_in_order() -> None:
+    registry = MigrationRegistry()
+    registry.register(StateMigration(from_version="v1", to_version="v2", migrate=_id))
+    registry.register(StateMigration(from_version="v2", to_version="v3", migrate=_id))
+    assert len(registry) == 2
+    assert "v1 → v2" in registry.describe()
+    assert "v2 → v3" in registry.describe()
+
+
+def test_registry_rejects_empty_to_version() -> None:
+    """Per spec §10.2 / proposal 0014, empty to_version routes the
+    chain TO the "not declared" sentinel — incoherent. Registration
+    MUST reject it. Empty from_version stays valid (documented
+    bridging path for pre-declaration records)."""
+    registry = MigrationRegistry()
+    with pytest.raises(ValueError, match="to_version MUST be non-empty"):
+        registry.register(StateMigration(from_version="v1", to_version="", migrate=_id))
+
+
+def test_registry_accepts_empty_from_version_bridging_case() -> None:
+    """Empty from_version is the documented Q4 bridging path: pre-
+    declaration records carrying the empty sentinel migrate forward
+    to a newly-declared schema. The registration MUST succeed; BFS
+    treats the empty sentinel as a valid source node."""
+    registry = MigrationRegistry()
+    registry.register(StateMigration(from_version="", to_version="v1", migrate=_id))
+    chain = registry.resolve_chain("", "v1")
+    assert chain is not None
+    assert len(chain) == 1
+
+
+def test_registry_rejects_duplicate_edge() -> None:
+    registry = MigrationRegistry()
+    registry.register(StateMigration(from_version="v1", to_version="v2", migrate=_id))
+    # Per spec §10.10 / §10.12.1 (proposal 0018, v0.16.0), duplicate-
+    # pair detection raises the canonical category directly at
+    # registration time. ``CheckpointStateMigrationChainAmbiguous``
+    # inherits from ``CheckpointError`` (not ``ValueError``), so the
+    # canonical-category assertion is the right shape.
+    with pytest.raises(CheckpointStateMigrationChainAmbiguous, match="duplicate state migration"):
+        registry.register(StateMigration(from_version="v1", to_version="v2", migrate=_id))
+
+
+# ---------------------------------------------------------------------------
+# Chain resolution — empty, identity, single hop, multi hop
+# ---------------------------------------------------------------------------
+
+
+def test_resolve_chain_same_version_returns_empty_chain() -> None:
+    registry = MigrationRegistry()
+    chain = registry.resolve_chain("v1", "v1")
+    assert chain == []
+
+
+def test_resolve_chain_empty_registry_returns_none() -> None:
+    registry = MigrationRegistry()
+    assert registry.resolve_chain("v1", "v2") is None
+
+
+def test_resolve_chain_unrelated_registry_returns_none() -> None:
+    registry = MigrationRegistry()
+    registry.register(StateMigration(from_version="v3", to_version="v4", migrate=_id))
+    assert registry.resolve_chain("v1", "v2") is None
+
+
+def test_resolve_chain_single_hop() -> None:
+    registry = MigrationRegistry()
+    a_to_b = StateMigration(from_version="a", to_version="b", migrate=_id)
+    registry.register(a_to_b)
+    chain = registry.resolve_chain("a", "b")
+    assert chain == [a_to_b]
+
+
+def test_resolve_chain_multi_hop_in_order() -> None:
+    registry = MigrationRegistry()
+    a_to_b = StateMigration(from_version="a", to_version="b", migrate=_id)
+    b_to_c = StateMigration(from_version="b", to_version="c", migrate=_id)
+    c_to_d = StateMigration(from_version="c", to_version="d", migrate=_id)
+    # Register out of natural order to verify BFS doesn't depend on
+    # registration order.
+    registry.register(c_to_d)
+    registry.register(a_to_b)
+    registry.register(b_to_c)
+    chain = registry.resolve_chain("a", "d")
+    assert chain == [a_to_b, b_to_c, c_to_d]
+
+
+def test_resolve_chain_picks_shortest_when_unique() -> None:
+    """A short path exists alongside a longer one; BFS picks the short."""
+    registry = MigrationRegistry()
+    # Diamond with an extra step on one side.
+    registry.register(StateMigration(from_version="v1", to_version="v2", migrate=_id))
+    registry.register(StateMigration(from_version="v2", to_version="v3", migrate=_id))
+    # Long detour: v1 -> v1a -> v1b -> v3.
+    registry.register(StateMigration(from_version="v1", to_version="v1a", migrate=_id))
+    registry.register(StateMigration(from_version="v1a", to_version="v1b", migrate=_id))
+    registry.register(StateMigration(from_version="v1b", to_version="v3", migrate=_id))
+    chain = registry.resolve_chain("v1", "v3")
+    assert chain is not None
+    assert [(m.from_version, m.to_version) for m in chain] == [
+        ("v1", "v2"),
+        ("v2", "v3"),
+    ]
+
+
+def test_resolve_chain_ambiguous_shortest_paths_raises() -> None:
+    """Diamond with two distinct same-length paths is ambiguous per
+    spec §10.10 / §10.12.2 (proposal 0018). ``resolve_chain``
+    raises the canonical ``CheckpointStateMigrationChainAmbiguous``
+    directly — no boundary wrap needed at the resume site; the
+    registry's exception contract is one type regardless of when
+    ambiguity surfaces (register vs resolve)."""
+    registry = MigrationRegistry()
+    registry.register(StateMigration(from_version="v1", to_version="v2", migrate=_id))
+    registry.register(StateMigration(from_version="v1", to_version="v3", migrate=_id))
+    registry.register(StateMigration(from_version="v2", to_version="v4", migrate=_id))
+    registry.register(StateMigration(from_version="v3", to_version="v4", migrate=_id))
+    with pytest.raises(CheckpointStateMigrationChainAmbiguous, match="ambiguous migration chain") as exc_info:
+        registry.resolve_chain("v1", "v4")
+    assert exc_info.value.from_version == "v1"
+    assert exc_info.value.to_version == "v4"
+
+
+def test_chain_ambiguous_category_string() -> None:
+    """The canonical category string per spec §10.10 (proposal 0018)."""
+    exc = CheckpointStateMigrationChainAmbiguous("boom")
+    assert exc.category == "checkpoint_state_migration_chain_ambiguous"
+
+
+def test_chain_ambiguous_carries_identity_when_set() -> None:
+    exc = CheckpointStateMigrationChainAmbiguous(
+        "duplicate v1→v2 registered",
+        from_version="v1",
+        to_version="v2",
+    )
+    assert exc.from_version == "v1"
+    assert exc.to_version == "v2"
+
+
+def test_chain_ambiguous_carries_none_when_unset() -> None:
+    exc = CheckpointStateMigrationChainAmbiguous("boom")
+    assert exc.from_version is None
+    assert exc.to_version is None
+
+
+# ---------------------------------------------------------------------------
+# GraphBuilder ergonomics — singular + plural registration
+# ---------------------------------------------------------------------------
+
+
+class _Sv1(State):
+    schema_version = "v1"
+    x: int = 0
+
+
+def _build_minimal_graph() -> GraphBuilder[_Sv1]:
+    async def _noop(_s: _Sv1) -> dict[str, int]:
+        return {}
+
+    return GraphBuilder(_Sv1).add_node("noop", _noop).add_edge("noop", END).set_entry("noop")
+
+
+def test_builder_with_state_migration_singular() -> None:
+    builder = _build_minimal_graph()
+    builder.with_state_migration("v0", "v1", _id)
+    compiled = builder.compile()
+    assert len(compiled.migration_registry) == 1
+
+
+def test_builder_with_state_migrations_plural() -> None:
+    builder = _build_minimal_graph()
+    builder.with_state_migrations(
+        StateMigration(from_version="v0", to_version="v1", migrate=_id),
+        StateMigration(from_version="v1", to_version="v2", migrate=_id),
+    )
+    compiled = builder.compile()
+    assert len(compiled.migration_registry) == 2
+
+
+def test_builder_duplicate_registration_raises() -> None:
+    builder = _build_minimal_graph()
+    builder.with_state_migration("v0", "v1", _id)
+    with pytest.raises(CheckpointStateMigrationChainAmbiguous):
+        builder.with_state_migration("v0", "v1", _id)
+
+
+def test_builder_with_state_migrations_atomic_on_duplicate() -> None:
+    """The plural ``with_state_migrations`` pre-validates the full
+    input list before mutating. A duplicate in the middle of the
+    list MUST NOT leave the earlier entries half-registered."""
+    builder = _build_minimal_graph()
+    # Pre-seed one entry so the third in the plural call collides.
+    builder.with_state_migration("v0", "v1", _id)
+
+    with pytest.raises(CheckpointStateMigrationChainAmbiguous):
+        builder.with_state_migrations(
+            StateMigration(from_version="v1", to_version="v2", migrate=_id),
+            StateMigration(from_version="v2", to_version="v3", migrate=_id),
+            # Collides with the pre-seeded v0→v1.
+            StateMigration(from_version="v0", to_version="v1", migrate=_id),
+        )
+    # Registry still has only the pre-seed; the v1→v2 and v2→v3
+    # from the failed call MUST NOT have registered.
+    compiled = builder.compile()
+    keys = [(m.from_version, m.to_version) for m in compiled.migration_registry]
+    assert keys == [("v0", "v1")]
+
+
+def test_builder_with_state_migrations_atomic_on_internal_duplicate() -> None:
+    """A duplicate pair WITHIN the with_state_migrations input list
+    (not against an already-registered entry) also raises before
+    mutating."""
+    builder = _build_minimal_graph()
+    with pytest.raises(CheckpointStateMigrationChainAmbiguous):
+        builder.with_state_migrations(
+            StateMigration(from_version="v1", to_version="v2", migrate=_id),
+            # Same key as the entry above — internal duplicate.
+            StateMigration(from_version="v1", to_version="v2", migrate=_id),
+        )
+    # Nothing registered.
+    compiled = builder.compile()
+    assert list(compiled.migration_registry) == []
+
+
+# ---------------------------------------------------------------------------
+# Error attribute carriage
+# ---------------------------------------------------------------------------
+
+
+def test_migration_missing_carries_identity() -> None:
+    exc = CheckpointStateMigrationMissing(
+        "no chain",
+        from_version="v1",
+        to_version="v2",
+        registered_migrations_count=3,
+        registry_description="v3 → v4\nv5 → v6\nv7 → v8",
+    )
+    assert exc.from_version == "v1"
+    assert exc.to_version == "v2"
+    assert exc.registered_migrations_count == 3
+    assert "v3 → v4" in exc.registry_description
+
+
+def test_migration_failed_carries_identity() -> None:
+    exc = CheckpointStateMigrationFailed(
+        "boom",
+        from_version="v1",
+        to_version="v2",
+    )
+    assert exc.from_version == "v1"
+    assert exc.to_version == "v2"
+
+
+def test_migration_failed_preserves_cause_when_raised_from() -> None:
+    """The error's ``__cause__`` carries the original migration
+    exception when raised from a try/except. This mirrors the engine's
+    chain-application wrap."""
+    underlying = KeyError("missing field")
+    try:
+        try:
+            raise underlying
+        except KeyError as exc:
+            raise CheckpointStateMigrationFailed(
+                "boom",
+                from_version="v1",
+                to_version="v2",
+            ) from exc
+    except CheckpointStateMigrationFailed as final:
+        assert final.__cause__ is underlying