Add per-invocation observer event drain primitive

[chris-colinsky]

Summary

PR 2b of the v0.12.0 cycle. Implements proposal 0054 (spec graph-engine §6 Per-invocation drain, spec v0.46.0).

CompiledGraph.drain_events_for(invocation_id, *, timeout=5.0) gives a terminal node a way to block on the observer event stream before reading derived observer state. The classic pattern is the §9.4 queryable accumulator observer: events arrive on the deliver loop's queue and the accumulator's per-invocation bucket fills up as the loop processes them. A terminal persist node calling accumulator.get_bucket(...) without first awaiting the drain can read a partially-filled bucket because the deliver loop is still racing the node body. The new primitive synchronizes those two: the node body pauses until every event dispatched up to call time has been delivered, then reads.

Notable pieces

• Engine method in src/openarmature/graph/compiled.py. Validates timeout with the same not (timeout >= 0) idiom as drain(). Finds the worker context whose invocation_id matches; if none, returns an empty summary (already drained or never started). Snapshots dispatched at call time. Fast path: if delivered >= target, returns immediately. Slow path: creates a Future on the running loop, registers a (target, fut) pair on _DrainCounters.drain_wakers, and awaits via asyncio.wait_for. On TimeoutError: removes the Future from the wakers list and returns DrainSummary(undelivered_count=max(0, target - delivered), timeout_reached=True).
• Deliver-loop wake in src/openarmature/graph/observer.py. After counters.delivered += 1, the loop walks drain_wakers and fulfills any whose target has been reached. Guarded by if counters.drain_wakers: so the common-case (no in-flight drains) takes zero extra work. not fut.done() absorbs the race where asyncio.wait_for has already cancelled the Future on timeout.
• _DrainCounters field: drain_wakers: list[tuple[int, asyncio.Future[None]]]. Subgraph descents (fan-out instances, parallel-branches branches) share the parent's counters via the existing descend_into_subgraph plumbing, so a drain on the outermost invocation_id covers all inner events without extra wiring.
• Load-bearing divergence from drain(): per-invocation drain leaves the worker running on timeout. drain is shutdown semantics and cancels its workers; drain_events_for is in-flight synchronization and the graph keeps serving other invocations. Documented in the method docstring and pinned by test_drain_events_for_timeout_does_not_cancel_worker.

Behavior pins

Ten unit tests in tests/unit/test_drain.py mirror the spec fixtures' case shapes:

• test_drain_events_for_basic_synchronization (fixture 028)
• test_drain_events_for_snapshot_semantic_does_not_wait_for_own_completed_event (fixture 029): node body calls drain_events_for(current_invocation_id) from inside itself; outer asyncio.wait_for(invoke(...), timeout=5.0) would fire if the snapshot semantic were broken.
• test_drain_events_for_timeout_does_not_cancel_worker (fixture 030, the load-bearing divergence): tight timeout against very-slow observer; verifies the deliver loop still processes a fresh second invocation cleanly after the timed-out drain returns.
• test_drain_events_for_invocation_scope_isolation (fixture 031)
• test_drain_events_for_covers_fan_out_instance_events (fixture 032): 3-instance fan-out; drain on outer id awaits all three instance subgraph events.
• test_drain_events_for_covers_parallel_branches_events (fixture 033)
• test_drain_events_for_zero_timeout_is_non_blocking_check
• test_drain_events_for_unknown_invocation_returns_clean_summary
• test_drain_events_for_rejects_negative_timeout, test_drain_events_for_rejects_nan_timeout

The six spec fixtures (graph-engine/028-033) stay deferred from the cross-capability parser pending the upcoming conformance-adapter capability to ratify the new directive vocabulary (observers[].behavior: accumulate, nodes.<name>.invoke_drain_events_for, the per-node node_drain_summaries / node_accumulator_snapshots assertion blocks).

Conformance + docs

• conformance.toml: [proposals."0054"] flips not-yet → implemented/since="0.12.0". Comment block enumerates the ten pin tests and the worker-cancellation-divergence rationale.
• tests/conformance/test_fixture_parsing.py: 028-033 deferral rationale updates from "lands in PR 2b" to "fixture-shape models pending; contract pinned by unit tests".
• docs/concepts/observability.md: the !!! info "drain_events_for ships in v0.12.0..." callout (added in PR 129 as a future-promise hedge) disappears now that the primitive ships. The surrounding text grows a paragraph explaining the worker-cancellation divergence and the timeout discipline.
• CHANGELOG.md: ### Added under [Unreleased] gets a drain_events_for entry positioned before the existing get_invocation_metadata() entry from PR 129.
• src/openarmature/AGENTS.md: regenerated.

Coord context

Spec coord thread discuss-per-invocation-event-drain (02-spec-accepted-as-0054.md) cleared all five spec questions: §6 section placement, drain_events_for naming, no-mandated-default timeout (free to pick 5.0), reuse DrainSummary shape verbatim, and the resume interaction now covered by proposal 0021 (suspension). Spec also flagged the worker-cancellation divergence explicitly as a thing the Python implementation could regress on if shared with drain()'s code path — pinned by test_drain_events_for_timeout_does_not_cancel_worker.

Test plan

• uv run pytest tests/ — 1111 passed (was 1101 pre-PR), 307 skipped, 0 failures
• uv run pytest tests/unit/test_drain.py — 19 passed (was 9; +10 new)
• uv run pytest tests/unit/test_observer.py tests/unit/test_fan_out.py — no regressions on the touched subsystem
• uv run python scripts/check_conformance_manifest.py — 51/51 entries consistent
• uv run ruff check . + uv run ruff format --check . — clean
• uv run pyright src/ tests/ — 0 errors
• uv run mkdocs build --strict — clean

[Copilot]

Pull request overview

Implements proposal 0054’s per-invocation observer event drain so terminal nodes can synchronize with the observer delivery loop before reading derived observer state.

Changes:

• Add CompiledGraph.drain_events_for(invocation_id, *, timeout=5.0) with snapshot semantics and non-cancelling timeout behavior.
• Extend observer delivery loop bookkeeping to wake per-invocation drain waiters once delivered reaches the snapshotted target.
• Add unit tests and update conformance notes/docs/changelog to reflect the shipped primitive and deferred fixture parsing.

Reviewed changes

Copilot reviewed 7 out of 7 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
tests/unit/test_drain.py	Adds unit tests for drain_events_for behaviors (sync, snapshot, timeout, scope, fan-out/branches).
tests/conformance/test_fixture_parsing.py	Updates deferral rationale for 028–033 fixture shapes pending parser support.
src/openarmature/graph/observer.py	Adds per-invocation drain waker list and deliver-loop wakeup logic.
src/openarmature/graph/compiled.py	Adds drain_events_for API and docs alongside existing drain().
docs/concepts/observability.md	Updates observability docs to describe shipped primitive and timeout divergence from drain().
conformance.toml	Marks proposal 0054 as implemented and documents pinned behaviors/tests.
CHANGELOG.md	Adds an “Added” entry documenting drain_events_for and semantics.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.