feat(observability): openarmature.checkpoint.migrate OTel span

Lands the §6 cross-ref piece of proposal 0014 that the previous
commit deferred. Versioned resumes whose migration chain runs
now emit a zero-duration `openarmature.checkpoint.migrate` span
on the OTel observer, parented under the invocation root.

- New _MigrationSummary frozen dataclass in graph/compiled.py
  carries the chain metadata (from_version, to_version,
  chain_length) out of _migrate_record back to invoke().
- _migrate_record return type becomes
  tuple[CheckpointRecord, _MigrationSummary]; invoke() captures
  the summary across the migration → context-creation gap.
- After context + delivery-worker setup but before the main
  execution loop, invoke() dispatches a synthetic NodeEvent
  with phase='checkpoint_migrated' carrying the summary on
  pre_state. Mirrors the checkpoint_saved pattern (state-on-pre,
  post=None) so the OTel observer's existing routing shape
  picks it up cleanly.
- New phase 'checkpoint_migrated' added to NodeEvent's Literal
  union and to KNOWN_PHASES (rejects typos in observer.phases
  subscriptions). NodeEvent.pre_state typed as Any so the
  permissive _MigrationSummary payload fits; the OTel observer
  narrows defensively via isinstance.
- OTelObserver._emit_checkpoint_migrate_span emits the span
  with the three normative attributes. Parented under the
  invocation root span (opened lazily if not yet present —
  versioned resumes are the first event on a new invocation_id,
  before any node-span emission). Mirrors _emit_checkpoint_save_span.
- Two OTel unit tests: one positive (versioned resume emits the
  span with the right attrs) and one negative (version-match
  fast path emits no span, per spec §10.12.3).

CHANGELOG entry promoted from 'deferred to follow-on' to the
landed shape. Phase gated off default subscriptions so legacy
observers don't surface a new event without opt-in.

Author: chris-colinsky

CHANGELOG.md

@@ -10,6 +10,7 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). The
 
 - **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.

src/openarmature/graph/compiled.py

@@ -246,6 +246,22 @@ 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,
@@ -369,15 +385,20 @@ async def _migrate_record(
         checkpointer: Checkpointer,
         invocation_id: str,
         current_schema_version: str,
-    ) -> CheckpointRecord:
+    ) -> tuple[CheckpointRecord, _MigrationSummary]:
         """Resolve a migration chain for ``record`` and apply it.
 
-        Returns the record with ``state`` + ``parent_states`` mapped
-        through the chain. 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``.
+        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
@@ -433,22 +454,21 @@ async def _migrate_record(
             for i, parent in enumerate(migrated_parents):
                 migrated_parents[i] = _apply_migration_step(migration, parent, f"parent_states[{i}]")
 
-        # TODO(observability): emit an ``openarmature.checkpoint.migrate``
-        # span per spec §6 cross-ref. Deferred to a follow-on because
-        # ``_migrate_record`` runs before the invocation's
-        # ``_InvocationContext`` is created (the engine needs the
-        # migrated state shape to build the context), so the existing
-        # ``_dispatch``-based observer pathway is not yet available
-        # here. A natural fix is to dispatch a synthetic
-        # ``checkpoint_migrated`` event as the first event of the
-        # invocation after context creation. The chain metadata
-        # captured for that event: ``from_version``, ``to_version``
-        # (final target), ``chain_length = len(chain)``.
-        return dataclass_replace(
+        # 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
@@ -544,6 +564,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:
@@ -570,7 +597,7 @@ async def invoke(
             # 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 = await self._migrate_record(
+                record, migration_summary = await self._migrate_record(
                     record,
                     checkpointer,
                     resume_invocation,
@@ -653,6 +680,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:

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
@@ -127,8 +127,19 @@ class NodeEvent:
     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, ...]

src/openarmature/graph/observer.py

@@ -97,7 +97,7 @@ async def __call__(self, event: NodeEvent, /) -> None: ...
 # 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"})
+KNOWN_PHASES: frozenset[str] = frozenset({"started", "completed", "checkpoint_saved", "checkpoint_migrated"})
 
 
 @dataclass(frozen=True)

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