prepare-sync: precise warn text + spec-derived test bodies

PR-C.3 review fixup.

- observer.py: rewrite the awaitable-from-prepare_sync warning.
  The old text claimed "did NOT run", but a user returning an
  ``asyncio.Task`` / ``Future`` may have work in flight on the
  loop — just not awaited at the prepare_sync call site. The
  contract violation is "no guarantee the prep completes before
  the node body," not "definitely doesn't run." Reworded to that
  shape and included ``type(result).__name__`` so the user
  can see which awaitable they returned at a glance.

- tests/conformance/test_observability.py: sub-case 1's driver
  hardcoded the YAML message bodies ("node a executing" /
  "node b executing") for record filtering and lookup, even
  though it had already read ``emits_log.message`` from the
  spec to drive the node body. That duplicated spec data and
  made the test brittle to fixture wording changes. Derive a
  ``node_emit_messages`` map from ``nodes_spec`` up front; use
  the values for both record filtering and ``by_body`` indexing.
  Sub-case 2 already worked this way (uses ``outer_emit`` /
  ``inner_emit`` derived from spec); sub-case 1 now matches.

Author: chris-colinsky

src/openarmature/graph/observer.py

@@ -435,10 +435,12 @@ def _dispatch(context: _InvocationContext, event: NodeEvent) -> None:
                             stacklevel=2,
                         )
                 warnings.warn(
-                    "observer prepare_sync returned an awaitable; "
-                    "prepare_sync MUST be sync (define as `def`, not "
-                    "`async def`). The synchronous prep work did NOT "
-                    "run; log correlation will miss this node's span.",
+                    f"observer prepare_sync returned an awaitable "
+                    f"({type(result).__name__}); prepare_sync MUST be sync "
+                    f"(define as `def`, not `async def`). The returned "
+                    f"awaitable will not be awaited and is NOT guaranteed "
+                    f"to complete before the node body starts; log "
+                    f"correlation may miss this node's span.",
                     stacklevel=2,
                 )
     context.queue.put_nowait(_QueuedItem(event=event, observers=observers))

tests/conformance/test_observability.py

@@ -1515,6 +1515,13 @@ async def _run_fixture_010_nested_trace(case: Mapping[str, Any]) -> None:
 
     nodes_spec = cast("dict[str, Any]", case["nodes"])
     correlation_id = cast("str", case["caller_correlation_id"])
+    # Spec YAML is the single source of truth for the log bodies; derive
+    # them up front rather than hard-coding so a fixture rename doesn't
+    # silently break the driver's record filtering.
+    node_emit_messages: dict[str, str] = {
+        name: cast("str", cast("dict[str, Any]", nodes_spec[name])["emits_log"]["message"])
+        for name in nodes_spec
+    }
 
     class _S(State):
         x: int = 0
@@ -1554,18 +1561,21 @@ async def body(_s: _S) -> dict[str, Any]:
         log_provider.force_flush()
 
         records = log_exporter.get_finished_logs()
-        # Filter to OUR test logger so concurrent test setup noise
-        # doesn't contaminate the assertions.
-        ours = [r for r in records if str(r.log_record.body) in {"node a executing", "node b executing"}]
+        # Filter to OUR test loggers so concurrent test setup noise
+        # doesn't contaminate the assertions. Expected message set
+        # comes from the spec YAML, not hard-coded strings.
+        expected_messages = set(node_emit_messages.values())
+        ours = [r for r in records if str(r.log_record.body) in expected_messages]
         assert len(ours) == 2, (
             f"expected 2 log records (one per node body); got {len(ours)}: "
             f"{[str(r.log_record.body) for r in ours]}"
         )
 
-        # Group by body for predictable lookup.
+        # Group by body for predictable lookup, indexing by the spec's
+        # emit-message values.
         by_body = {str(r.log_record.body): r for r in ours}
-        a_log = by_body["node a executing"]
-        b_log = by_body["node b executing"]
+        a_log = by_body[node_emit_messages["a"]]
+        b_log = by_body[node_emit_messages["b"]]
 
         # Invariant: all_logs_same_trace_id.
         trace_ids = {a_log.log_record.trace_id, b_log.log_record.trace_id}