Address PR #71 review: error metadata + pairing + bool validation

Four CoPilot threads addressed:

1. CheckpointRecordInvalid (proposal 0029 count-drift raise) now
   references the LOADED record's invocation_id via
   `context.resume_invocation or context.invocation_id` — on resume
   the new run gets a freshly minted id but the error should point
   at the saved record being validated, not the new run.

2. _step_fan_out_node's CheckpointError branch now dispatches a
   `completed` event with a NodeException wrapper before re-raising
   the original CheckpointError. Preserves spec proposal 0012's
   started/completed pairing contract while keeping NodeEvent.error
   typed as RuntimeGraphError | None (CheckpointError is sibling-
   typed). Observers see complete pairing with the original
   CheckpointError on `__cause__`; the invoke() caller still
   receives the unwrapped checkpoint error via the bare `raise`.

3. SQLite checkpointer's `result_is_error` deserialization now
   distinguishes key-absent (pre-0027 records, default False for
   backward compat) from key-present (strictly validates it's a
   bool and raises CheckpointRecordInvalid on mismatch). The naive
   bool() coercion would have misclassified a corrupted record
   with `"result_is_error": "false"` (string) as True since
   non-empty strings are truthy. Threads invocation_id through
   _fan_out_progress_from_dict so the raise points at the correct
   persisted record.

4. result_present matcher comment + variable rename make the
   semantic explicit: "result is a non-None value" rather than the
   fuzzier "field present" framing. Documents the hypothetical
   `result = None` on a completed-success entry as out of scope —
   spec frames result as "the durable contribution" and no fixture
   exercises a None contribution on a completed entry.

Author: chris-colinsky

src/openarmature/checkpoint/backends/sqlite.py

@@ -113,10 +113,15 @@ def _fan_out_progress_to_dict(fp: FanOutProgress) -> dict[str, Any]:
     }
 
 
-def _fan_out_progress_from_dict(d: dict[str, Any]) -> FanOutProgress:
+def _fan_out_progress_from_dict(d: dict[str, Any], invocation_id: str) -> FanOutProgress:
     """Inverse of :func:`_fan_out_progress_to_dict` — rebuild a frozen
     :class:`FanOutProgress` from its dict shape, restoring positions
-    as :class:`NodePosition` instances."""
+    as :class:`NodePosition` instances.
+
+    ``invocation_id`` is threaded in so ``CheckpointRecordInvalid``
+    raises (e.g., a non-bool ``result_is_error`` field) can point at
+    the correct persisted record.
+    """
     instances: list[FanOutInstanceProgress] = []
     for inst in cast("list[dict[str, Any]]", d["instances"]):
         inner_positions = tuple(
@@ -129,16 +134,31 @@ def _fan_out_progress_from_dict(d: dict[str, Any]) -> FanOutProgress:
             )
             for p in cast("list[dict[str, Any]]", inst.get("completed_inner_positions", []))
         )
+        # Per proposal 0027: distinguish key-absent (pre-0027 records,
+        # default ``False`` for backward compat) from key-present
+        # (strictly validate it's a bool). The naive
+        # ``bool(inst.get("result_is_error", False))`` would coerce
+        # truthy non-bools (e.g. the string ``"false"``) to ``True``
+        # and misclassify resume routing — the JSON deserializer can
+        # in principle land a non-bool here from a corrupted record,
+        # and silently coercing it would route a success contribution
+        # through the errors_field bucket on resume.
+        if "result_is_error" in inst:
+            raw_rie = inst["result_is_error"]
+            if not isinstance(raw_rie, bool):
+                raise CheckpointRecordInvalid(
+                    invocation_id,
+                    f"fan_out_progress instance result_is_error must be bool, "
+                    f"got {type(raw_rie).__name__}: {raw_rie!r}",
+                )
+            result_is_error = raw_rie
+        else:
+            result_is_error = False
         instances.append(
             FanOutInstanceProgress(
                 state=inst["state"],
                 result=inst.get("result"),
-                # Pre-0027 records omit ``result_is_error``; the default
-                # ``False`` (matching the dataclass default) is correct
-                # for those — pre-0027 was atomic-restart-fan-out resume
-                # under proposal 0008, no error-record state machine, so
-                # the field's absence semantically means "not an error."
-                result_is_error=bool(inst.get("result_is_error", False)),
+                result_is_error=result_is_error,
                 completed_inner_positions=inner_positions,
             )
         )
@@ -381,7 +401,9 @@ def _do() -> tuple[Any, ...] | None:
                 recorded_serialization,
                 invocation_id,
             )
-            fan_out_progress = tuple(_fan_out_progress_from_dict(fp) for fp in fan_out_progress_dicts)
+            fan_out_progress = tuple(
+                _fan_out_progress_from_dict(fp, invocation_id) for fp in fan_out_progress_dicts
+            )
         return CheckpointRecord(
             invocation_id=invocation_id,
             correlation_id=correlation_id,

src/openarmature/graph/compiled.py

@@ -1628,17 +1628,35 @@ async def innermost(s: Any) -> Mapping[str, Any]:
                         fan_out_config=fan_out_event_config,
                     )
                     raise
-                except CheckpointError:
-                    # CheckpointError categories (proposal 0029's
-                    # count-drift raise, the migration-category errors,
-                    # save-failed) are sibling-typed to RuntimeGraphError
-                    # and propagate to the invoke() caller unwrapped so
-                    # callers can branch on ``e.category``. No completed
-                    # event is dispatched here — the `NodeEvent.error`
-                    # field is typed as ``RuntimeGraphError | None`` per
-                    # spec §6 and CheckpointError isn't in that
-                    # hierarchy. Matches ``_step_function_node``'s
-                    # CheckpointError branch (also raise-only).
+                except CheckpointError as e:
+                    # Spec proposal 0012's pairing contract requires
+                    # every started event have a paired completed
+                    # event. CheckpointError categories (notably
+                    # proposal 0029's count-drift raise) are sibling-
+                    # typed to RuntimeGraphError and propagate to the
+                    # invoke() caller unwrapped so callers can branch
+                    # on ``e.category``. To preserve pairing while
+                    # keeping ``NodeEvent.error`` typed as
+                    # ``RuntimeGraphError | None`` per spec §6, the
+                    # completed event carries a ``NodeException``
+                    # wrapper whose ``__cause__`` is the original
+                    # CheckpointError. The bare ``raise`` re-raises
+                    # the active exception (the CheckpointError, not
+                    # the wrapper) so the caller still sees the
+                    # checkpoint category. Mirrors the ``except
+                    # Exception`` branch below structurally; the
+                    # difference is what gets re-raised.
+                    wrapped = NodeException(node_name=current, cause=e, recoverable_state=s)
+                    self._dispatch_completed(
+                        context,
+                        current,
+                        namespace,
+                        step,
+                        s,
+                        error=wrapped,
+                        attempt_index=attempt_index,
+                        fan_out_config=fan_out_event_config,
+                    )
                     raise
                 except Exception as e:
                     wrapped = NodeException(node_name=current, cause=e, recoverable_state=s)

src/openarmature/graph/fan_out.py

@@ -196,8 +196,14 @@ async def run_with_context(
             # ``CheckpointSaveFailed`` imports below in this file).
             from openarmature.checkpoint.errors import CheckpointRecordInvalid  # noqa: PLC0415
 
+            # ``context.resume_invocation`` identifies the SAVED record
+            # being validated (per spec §10.4 step 3); ``context.invocation_id``
+            # is freshly minted for the resumed run (step 4). The fresh-
+            # run fallback is defensive only — the count-drift path can
+            # only fire on resume since fan_out_progress_state is empty
+            # on a fresh first run.
             raise CheckpointRecordInvalid(
-                context.invocation_id,
+                context.resume_invocation or context.invocation_id,
                 f"fan_out {self.name!r} at namespace {context.namespace_prefix!r}: "
                 f"saved instance_count={exec_state.instance_count} does not match "
                 f"resolved instance_count={instance_count} on resume",

tests/conformance/test_checkpoint.py

@@ -838,16 +838,26 @@ def _assert_fan_out_instance(
             f"actual={actual.result_is_error!r}, expected={expected['result_is_error']!r}"
         )
     if "result_present" in expected:
-        # Spec §10.11 (proposal 0027): assert the ``result`` field
-        # exists on the saved record without constraining its shape
-        # (the value remains impl-defined per §9.5). Pair with
+        # Spec §10.11 (proposal 0027): assert the captured
+        # contribution is a non-None value — the fixture's way of
+        # saying "an entry was recorded" without constraining its
+        # shape (the value remains impl-defined per §9.5). Pair with
         # ``result_is_error: true`` to assert "an error contribution
-        # was captured" without locking the test to one impl's error
-        # record format.
-        result_present_actual = actual.result is not None
-        assert result_present_actual == expected["result_present"], (
+        # was captured" portably across implementations whose
+        # error_record formats differ.
+        #
+        # Note on the semantic: ``FanOutInstanceProgress.result`` is
+        # a dataclass field that always exists as an attribute with
+        # a None default; "presence" here means "result is a
+        # non-None value." A hypothetical legitimate ``result =
+        # None`` on a completed-success entry would be treated as
+        # "not present" by this check, but spec frames ``result`` as
+        # "the durable contribution" and no fixture exercises a
+        # None contribution on a completed entry.
+        result_has_value = actual.result is not None
+        assert result_has_value == expected["result_present"], (
             f"fan_out_progress[{node_name!r}].instances[{idx}].result_present: "
-            f"actual={result_present_actual!r}, expected={expected['result_present']!r}"
+            f"actual={result_has_value!r}, expected={expected['result_present']!r}"
         )
     if "completed_inner_positions" in expected:
         positions_expected = cast("list[Mapping[str, Any]]", expected["completed_inner_positions"])