fix: CoPilot review pass on PR #46

- compiled.py: align save-side schema_version read on
  self.state_cls (was type(post_state)); _maybe_save_checkpoint
  drops @staticmethod so it can access self. Symmetric with the
  resume-side check; subclass schema_versions no longer shadow
  the declared graph schema.
- compiled.py _apply_migration_step: re-raise CheckpointError
  subclasses unchanged before the bare-Exception wrap, so a
  canonical-category raise from a user migration propagates
  without being squashed as CheckpointStateMigrationFailed.
  Trim the wrap-message to the from/to identity + 'raised while
  migrating <label>' — __cause__ already carries the underlying
  exception; embedding type(exc).__name__: str(exc) duplicates
  info Python's traceback formatter shows anyway.
- builder.py with_state_migrations: pre-validate the full input
  list (against existing registry + against earlier entries in
  the same call) before mutating. A duplicate-pair raise in the
  middle of the list no longer leaves earlier entries
  half-registered. Update the singular with_state_migration
  docstring to reflect the canonical raise type
  (CheckpointStateMigrationChainAmbiguous, not ValueError) +
  flag the empty-to_version ValueError.
- migration.py register: reject empty to_version (un-declared
  sentinel is not a valid chain target per spec §10.2). Empty
  from_version stays valid for the documented bridging case.
- migration.py resolve_chain: raise
  CheckpointStateMigrationChainAmbiguous directly on
  multi-shortest-path detection (was ValueError). The registry's
  ambiguity contract is now one type regardless of when it
  surfaces (register vs resolve). compiled.py's wrap removed.
- protocol.py Checkpointer: clarify the supports_state_migration
  docstring on the Protocol-class-body vs runtime-attribute
  semantic — class-body '= False' is a typing-level signal, not
  a runtime guarantee. Engine uses getattr-with-default; backends
  SHOULD declare for Pyright honesty.
- events.py NodeEvent: document the synthetic-phases conventions
  (step=-1, dotted node_name, non-State pre_state on
  checkpoint_migrated) in the class docstring so third-party
  observer authors don't need to read the engine source.
- observer.py KNOWN_PHASES: expand the comment about the
  synthetic-phase opt-in and cross-reference the NodeEvent
  docstring.
- tests/conformance/test_state_migration.py: range docstring
  updated from 039-046 to 039-047 with a one-paragraph note on
  the chain-ambiguity surface 047 covers.
- tests/unit/test_state_migration.py: existing
  ambiguous-shortest-paths test asserts the canonical category
  (was ValueError); two new tests for the empty-to_version
  rejection + empty-from_version bridging acceptance; two new
  tests for with_state_migrations atomic pre-validation
  (no partial-state on internal-list and
  against-existing-registry duplicate cases).

Author: chris-colinsky

src/openarmature/checkpoint/migration.py

@@ -76,6 +76,21 @@ def __init__(self) -> None:
         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):
@@ -106,8 +121,13 @@ def resolve_chain(
         """Return an ordered chain of migrations bridging the two
         versions, or ``None`` if no chain exists.
 
-        Raises ``ValueError`` if multiple distinct shortest paths
-        exist (ambiguous chain per §10.12.2).
+        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 []
@@ -162,11 +182,18 @@ def resolve_chain(
             return None
         if len(shortest_paths) > 1:
             descriptions = [" → ".join([from_version, *(e.to_version for e in p)]) for p in shortest_paths]
-            raise ValueError(
+            # 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"
+                f"single canonical route",
+                from_version=from_version,
+                to_version=to_version,
             )
         return shortest_paths[0]
 

src/openarmature/checkpoint/protocol.py

@@ -160,6 +160,21 @@ class Checkpointer(Protocol):
     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
@@ -168,9 +183,11 @@ class Checkpointer(Protocol):
     # 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 with ``ClassVar[bool] = False``; pyright is happy with
-    # either shape because Protocol attribute conformance ignores the
-    # ClassVar marker on subclasses.
+    # 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:

src/openarmature/graph/builder.py

@@ -14,6 +14,7 @@
 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
 
@@ -274,9 +275,12 @@ def with_state_migration(
         state. The framework does not police purity (per §10.12.2),
         but violating it risks non-deterministic resume.
 
-        Raises ``ValueError`` at registration if the
-        ``(from_version, to_version)`` pair is already registered
-        (per §10.12.1 chain-ambiguity rule).
+        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(
@@ -290,7 +294,38 @@ def with_state_migration(
     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."""
+        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

src/openarmature/graph/compiled.py

@@ -46,10 +46,10 @@
 from pydantic import ValidationError
 
 from openarmature.checkpoint.errors import (
+    CheckpointError,
     CheckpointNotFound,
     CheckpointRecordInvalid,
     CheckpointSaveFailed,
-    CheckpointStateMigrationChainAmbiguous,
     CheckpointStateMigrationFailed,
     CheckpointStateMigrationMissing,
 )
@@ -274,10 +274,22 @@ def _apply_migration_step(
     """
     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} "
-            f"raised while migrating {label}: {type(exc).__name__}: {exc}",
+            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
@@ -419,24 +431,16 @@ async def _migrate_record(
                 f"support state migration",
             )
 
-        try:
-            chain = self.migration_registry.resolve_chain(
-                record.schema_version,
-                current_schema_version,
-            )
-        except ValueError as exc:
-            # MigrationRegistry signals multi-shortest-path ambiguity
-            # via ValueError. Per spec §10.10 / §10.12.2 (proposal 0018,
-            # spec v0.16.0), this routes to the canonical
-            # CheckpointStateMigrationChainAmbiguous category. Spec
-            # accepts load-time detection (this is the resume-side
-            # path); the duplicate-pair case raises the same category
-            # directly from MigrationRegistry.register at build time.
-            raise CheckpointStateMigrationChainAmbiguous(
-                str(exc),
-                from_version=record.schema_version,
-                to_version=current_schema_version,
-            ) from exc
+        # 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(
@@ -1505,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,
@@ -1574,13 +1585,15 @@ async def _maybe_save_checkpoint(
             # within-invocation order.
             last_saved_at=time.time(),
             # Per spec §10.2 (proposal 0014): read the user's
-            # state-schema version off the 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). The
-            # runtime type of ``post_state`` is the authoritative
-            # source (subclasses MAY override the ClassVar).
-            schema_version=cast("type[State]", type(post_state)).schema_version,
+            # 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)

src/openarmature/graph/events.py

@@ -122,6 +122,32 @@ 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

src/openarmature/graph/observer.py

@@ -95,8 +95,16 @@ 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"}``.
+# 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"})
 
 

tests/conformance/test_state_migration.py

@@ -1,4 +1,4 @@
-"""Run every spec state-migration conformance fixture (039-046) end-to-end.
+"""Run every spec state-migration conformance fixture (039-047) end-to-end.
 
 The fixtures live under
 ``spec/pipeline-utilities/conformance/`` as ``cases`` shapes; each
@@ -9,6 +9,14 @@
 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
@@ -26,7 +34,8 @@
 5. Calls ``invoke(resume_invocation=<seeded id>)`` and asserts
    against ``resume.expected`` (final state, migrations_run,
    invariants) OR ``resume.expected_error`` (category, carries,
-   cause).
+   cause) OR ``expected_chain_ambiguity_error`` at either case
+   top-level (build-time) or inside ``resume:`` (load-time).
 """
 
 from __future__ import annotations