graph: address PR #5 review

CodeQL noise (10 threads): replace bare `...` Protocol/stub bodies
with explicit forms. `Observer.__call__` gets a docstring +
`raise NotImplementedError` (so pyright accepts the declared return
type, same pattern as Protocol bodies fixed in #4); test stub
observers get `pass`. Add an explanatory comment to
`RemoveHandle.remove`'s `try/except ValueError: pass` documenting
the idempotency intent the docstring already promises.

Logic (2 threads):

  - `_active_workers` switched from list to set and each per-
    invocation worker now registers add_done_callback(
    self._active_workers.discard). Long-running services that
    never call drain() no longer accumulate completed Task
    references indefinitely. drain() simplifies to a one-line
    asyncio.gather over a snapshot of the set.

  - SubgraphNode branch in _invoke now wraps non-RuntimeGraphError
    exceptions from node.run(state, context=context) as
    NodeException tagged with the wrapper's name. Projection
    errors (project_in / project_out) and subgraph state-class
    init errors (e.g. Pydantic ValidationError) were previously
    propagating raw, bypassing the spec §4 error contract. Already-
    wrapped errors from inside the subgraph's _invoke pass through
    unchanged so the inner node's identity stays attached. Adds
    test_subgraph_projection_error_wrapped_as_node_exception
    covering the case.

Author: chris-colinsky

src/openarmature/graph/compiled.py

@@ -114,7 +114,10 @@ class CompiledGraph[StateT: State]:
     # dataclass: the list reference is fixed but its contents change.
     # Parameterized factories so pyright infers the element types.
     _attached_observers: list[Observer] = field(default_factory=list[Observer])
-    _active_workers: list[asyncio.Task[None]] = field(default_factory=list[asyncio.Task[None]])
+    # `set` (not list) so a per-task `add_done_callback(self._active_workers.discard)`
+    # auto-removes completed workers — long-running services that never call
+    # drain() don't accumulate completed Task references indefinitely.
+    _active_workers: set[asyncio.Task[None]] = field(default_factory=set[asyncio.Task[None]])
 
     # ------------------------------------------------------------------
     # Observer registration (spec v0.3.0 §6)
@@ -151,11 +154,10 @@ async def drain(self) -> None:
         """
         if not self._active_workers:
             return
-        pending = list(self._active_workers)
-        await asyncio.gather(*pending, return_exceptions=True)
-        for w in pending:
-            if w in self._active_workers:
-                self._active_workers.remove(w)
+        # Snapshot the set: each worker's done-callback removes itself
+        # from `_active_workers`, so iterating it directly while gather
+        # awaits would mutate during iteration.
+        await asyncio.gather(*list(self._active_workers), return_exceptions=True)
 
     # ------------------------------------------------------------------
     # Public invocation
@@ -188,7 +190,11 @@ async def invoke(
             invocation_scoped=invocation_scoped,
         )
         worker = asyncio.create_task(deliver_loop(queue))
-        self._active_workers.append(worker)
+        self._active_workers.add(worker)
+        # Auto-prune: when the worker completes (after the sentinel is
+        # 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)
         try:
             return await self._invoke(initial_state, context)
         finally:
@@ -225,9 +231,18 @@ async def _invoke(
                 # Subgraph wrappers are transparent to the observer protocol
                 # (per fixture 013): no event is dispatched for the wrapper
                 # itself, the step counter does not advance for it, and any
-                # exception bubbling up from the subgraph's _invoke is
-                # already wrapped with the inner node's identity.
-                partial = await node.run(state, context=context)
+                # `RuntimeGraphError` bubbling up from the subgraph's
+                # _invoke is already wrapped with the inner node's identity
+                # — pass it through. Other exceptions (projection errors,
+                # subgraph state-class init errors) escape the spec §4
+                # categories, so we wrap them as NodeException tagged with
+                # the wrapper's name.
+                try:
+                    partial = await node.run(state, context=context)
+                except RuntimeGraphError:
+                    raise
+                except Exception as e:
+                    raise NodeException(node_name=current, cause=e, recoverable_state=state) from e
                 state = _merge_partial(state, partial, self.reducers, current)
             else:
                 state = await self._step_function_node(node, current, state, context)

src/openarmature/graph/observer.py

@@ -44,7 +44,9 @@ class Observer(Protocol):
     `event`, `_event`, or any other name.
     """
 
-    async def __call__(self, event: NodeEvent, /) -> None: ...
+    async def __call__(self, event: NodeEvent, /) -> None:
+        """Receive a single node-boundary event."""
+        raise NotImplementedError
 
 
 @dataclass(frozen=True)
@@ -64,6 +66,9 @@ def remove(self) -> None:
         try:
             self._observers.remove(self._observer)
         except ValueError:
+            # Idempotency: the observer is already detached. Per the
+            # docstring, a second .remove() call is a no-op rather than
+            # an error.
             pass
 
 

tests/unit/test_observer.py

@@ -174,8 +174,11 @@ async def test_dispatch_skips_when_no_observers_for_depth() -> None:
 
 
 async def test_dispatch_enqueues_with_full_observer_chain_in_order() -> None:
-    async def graph_obs(_event: NodeEvent) -> None: ...
-    async def invocation_obs(_event: NodeEvent) -> None: ...
+    async def graph_obs(_event: NodeEvent) -> None:
+        pass
+
+    async def invocation_obs(_event: NodeEvent) -> None:
+        pass
 
     queue: asyncio.Queue[_QueuedItem | None] = asyncio.Queue()
     ctx = _InvocationContext(queue=queue, graph_attached=(graph_obs,), invocation_scoped=(invocation_obs,))
@@ -192,9 +195,14 @@ async def invocation_obs(_event: NodeEvent) -> None: ...
 
 
 async def test_descend_extends_chain_namespace_and_parent_states() -> None:
-    async def outer_obs(_event: NodeEvent) -> None: ...
-    async def sub_obs(_event: NodeEvent) -> None: ...
-    async def invocation_obs(_event: NodeEvent) -> None: ...
+    async def outer_obs(_event: NodeEvent) -> None:
+        pass
+
+    async def sub_obs(_event: NodeEvent) -> None:
+        pass
+
+    async def invocation_obs(_event: NodeEvent) -> None:
+        pass
 
     queue: asyncio.Queue[_QueuedItem | None] = asyncio.Queue()
     outer = _InvocationContext(queue=queue, graph_attached=(outer_obs,), invocation_scoped=(invocation_obs,))
@@ -227,7 +235,8 @@ async def test_take_step_shares_counter_across_descended_contexts() -> None:
 
 
 def test_remove_handle_detaches_observer() -> None:
-    async def obs(_event: NodeEvent) -> None: ...
+    async def obs(_event: NodeEvent) -> None:
+        pass
 
     observers: list[Observer] = [obs]
     handle = RemoveHandle(_observers=observers, _observer=obs)
@@ -238,7 +247,8 @@ async def obs(_event: NodeEvent) -> None: ...
 
 
 def test_remove_handle_is_idempotent() -> None:
-    async def obs(_event: NodeEvent) -> None: ...
+    async def obs(_event: NodeEvent) -> None:
+        pass
 
     observers: list[Observer] = [obs]
     handle = RemoveHandle(_observers=observers, _observer=obs)

tests/unit/test_runtime_errors.py

@@ -96,3 +96,55 @@ async def node_a(_state: Any) -> dict[str, Any]:
     err = excinfo.value
     assert err.category == "state_validation_error"
     assert "undeclared" in err.fields
+
+
+async def test_subgraph_projection_error_wrapped_as_node_exception() -> None:
+    """Errors from a subgraph's projection (project_in / project_out) are
+    NOT spec §4 categories on their own. The engine wraps them as
+    NodeException tagged with the subgraph wrapper's name so callers see
+    a uniform error contract."""
+
+    from openarmature.graph import NodeException, ProjectionStrategy
+
+    class Inner(State):
+        x: int = 0
+
+    async def _inner_node(_s: Inner) -> dict[str, Any]:
+        return {}
+
+    inner_g = GraphBuilder(Inner).add_node("i", _inner_node).add_edge("i", END).set_entry("i").compile()
+
+    # Parameter names match the ProjectionStrategy Protocol exactly so
+    # pyright's strict structural conformance check passes.
+    class BoomProjection:
+        def project_in(self, parent_state: S, subgraph_state_cls: type[Inner]) -> Inner:
+            raise RuntimeError("project_in boom")
+
+        def project_out(
+            self,
+            subgraph_final_state: Inner,
+            parent_state: S,
+            subgraph_state_cls: type[Inner],
+        ) -> dict[str, Any]:
+            return {}
+
+    _: ProjectionStrategy[S, Inner] = BoomProjection()
+
+    g = (
+        GraphBuilder(S)
+        .add_subgraph_node("sub", inner_g, projection=BoomProjection())
+        .add_edge("sub", END)
+        .set_entry("sub")
+        .compile()
+    )
+
+    with pytest.raises(NodeException) as excinfo:
+        await g.invoke(S())
+
+    err = excinfo.value
+    assert err.category == "node_exception"
+    # The wrapper's name, not the inner node's — projection is at the
+    # boundary, not inside the subgraph.
+    assert err.node_name == "sub"
+    assert isinstance(err.__cause__, RuntimeError)
+    assert str(err.__cause__) == "project_in boom"