test: integration coverage for message store + event bus (#2640)

[jwbron]

Closes #2640.
Closes #2663.

Summary

Add k3s integration tier coverage for the inter-agent message store and EventBus routing under integration_tests/regression/. 60 tests across 13 classes, parametrized across both backends (in-memory + Redis via fakeredis) where applicable.

Also fixes #2663: pipeline.cancelled is now emitted from the PATCH cancel path, so /status/wait long-pollers wake immediately on cancellation instead of waiting for the late-subscriber synth path on their next poll.

Coverage map

Issue starting points:

#	Coverage	Tests
1	context_pr.{skipped,failed} reach both message store and event bus on both backends; visible through GET /messages	TestContextPRRouting (3 × 2 backends)
2	/status/wait wakes on context_pr.* event AND CONTEXT_PR_* message; silent for PROGRESS / decision.resolved; wakes on pipeline.cancelled event (#2663)	TestStatusWaitContextPRSemantics (5 × 2)
3	EventBus sequence strictly monotonic across 8 concurrent publishers; per-producer message-store order preserved under 4-thread fan-in	TestConcurrentOrdering, TestMessageStoreEventBusOrderingCorrelation

Gap-audit additions:

• Stale cursor — since_id_stale: True on unknown since_id (Identify which consumers send stale since_id cursors to the message store #2464).
• Replay-on-resume — get_messages(since_id=...) returns subsequent-only; refetch from tip is empty.
• Late subscribers — EventBus does not replay history; unsubscribe takes effect immediately.
• Malformed payloads — shell-var to_role / from_role, missing / unknown / incomplete HEARTBEAT metadata → 400.
• Dedupe under contention — serial repeat AND 16-thread race against _context_pr_events_emitted_lock both collapse to one message + one event.
• Blocking semantics — get_messages(wait=N) wakes on add_message; wakes on clear() (RISK-5, Agent wait heuristics: replace sleep/poll loops with event-driven BRC message stream consumption #1897); from_tip=True ignores pre-existing messages (egg-orch message wait-loop returns already-seen messages immediately instead of blocking for new events #1925).
• First-source-wins — /status/wait between EventBus and message store.
• Deprecated type coercion — QUESTION → PROGRESS via POST /messages (Agent wait heuristics: replace sleep/poll loops with event-driven BRC message stream consumption #1897).
• EventBus sequence density — global across mixed event types, which the cursor protocol depends on.

Fix: pipeline.cancelled wake-up (#2663)

EventType.PIPELINE_CANCELLED was in _STATUS_WAIT_EVENT_TYPES (so /status/wait long-pollers expected to wake on it) but the orchestrator never emitted it — "pipeline.cancelled" was missing from _EVENT_TYPE_MAP and the PATCH cancel path didn't call _emit_pipeline_event. In-flight long-polls sat to their full 25s timeout before observing cancellation via the late-subscriber synth path on the next poll.

Two changes in orchestrator/routes/pipelines.py:

1. Add "pipeline.cancelled": EventType.PIPELINE_CANCELLED to _EVENT_TYPE_MAP so _emit_pipeline_event no longer no-ops on the string.
2. Call _emit_pipeline_event(pipeline, "pipeline.cancelled") from update_pipeline immediately after store.update_pipeline returns a CANCELLED pipeline. Mirrors the pipeline.failed / pipeline.completed pattern; only CANCELLED is emitted here because FAILED is already emitted by the run loop's own terminal transitions.

Covered by the new test_status_wait_wakes_on_pipeline_cancelled_event test (parametrized across both backends).

Bugs / gaps surfaced (filing follow-ups)

While writing these tests, I noticed one further gap that the existing unit tier doesn't cover and that this PR does NOT fix (per scope):

1. Redis is not deployed in k3s. k8s/base/orchestrator-deployment.yaml does not set REDIS_HOST / EGG_MESSAGE_STORE_BACKEND, and k8s/base/ has no Redis Deployment / Service / StatefulSet. The orchestrator falls back from EGG_MESSAGE_STORE_BACKEND=auto to the in-memory MessageStore. This means the production CI integration tier exercises only the in-memory backend end-to-end — the RedisMessageStore codepath has zero live coverage. These tests use fakeredis to exercise the Redis path in-process; running the in-cluster orchestrator against a real Redis would require k8s manifests + an EGG_MESSAGE_STORE_BACKEND=redis env knob. Filed as k3s does not deploy Redis — RedisMessageStore has no end-to-end CI coverage #2662.

Two minor dead-enum observations (not filing, but noting for future trim):

• EventType.MESSAGE_RECEIVED — defined, never emitted.
• EventType.CONTAINER_SPAWNED — defined, never emitted (CONTAINER_STOPPED is wired).

Test plan

• ruff check + ruff format clean on new files.
• All 60 tests pass locally via PYTHONPATH=shared .venv/bin/pytest integration_tests/regression/test_message_bus_routing.py -m integration --timeout=30.
• Related unit tests still green: orchestrator/tests/test_cancel_async_cleanup.py + orchestrator/tests/test_pipelines_api.py (53 tests).
• Existing integration_tests/test_slice_pipeline_e2e.py still green alongside the new file.
• CI Test / aggregate required check passes (this is the validation path per Expand integration test coverage #2474 — the agent sandbox can't bring up k3s).

Authored-by: egg

[egg-reviewer]

Summary

Approving. This is a high-quality regression-tier addition that meaningfully tightens the integration coverage on the message-store + EventBus seams, plus a small, well-scoped fix for #2663. I verified:

• All 60 tests collect and pass locally in ~12s against Python 3.14 + fakeredis (no flakes across two runs, including the 16-thread dedupe race).
• The _EVENT_TYPE_MAP entry at orchestrator/routes/pipelines.py:1136 correctly closes the no-op gap: "pipeline.cancelled" was already in _STATUS_WAIT_EVENT_TYPES (line 283) and in _TERMINAL_EVENT_TYPES (line 4017), but _emit_pipeline_event was silently no-op'ing because the string was missing from the dispatch map.
• The PATCH-side emit at orchestrator/routes/pipelines.py:2210-2211 is placed correctly: after store.update_pipeline returns the post-mutation pipeline, and before the background cleanup thread spins up. The cancel-in-PATCH path is the only origin of the CANCELLED transition; pipeline.failed / pipeline.completed are emitted from _run_pipeline itself, so the fix's narrow scope is correct.
• Both backends are genuinely exercised — verified fakeredis.FakeRedis() honors XREAD block=N and so the Redis path actually blocks rather than spin-returning.
• Conftest sys.path setup mirrors integration_tests/test_slice_pipeline_e2e.py:54-59 exactly, so the import shape is consistent with the existing integration tier.
• The 16-thread dedupe race-test passes cleanly despite the inherent patch.object thread-unsafety (see Non-blocking #3 below), because every thread sets inner.side_effect = RuntimeError(...) on whatever mock is current — the wrapper's _open_context_pr_for_pipeline call always resolves to some mock with the right side effect.

Non-blocking suggestions

1. update_pipeline emits on status-equality, not status-transition (orchestrator/routes/pipelines.py:2210). if pipeline.status == PipelineStatus.CANCELLED: will re-emit on any subsequent PATCH against an already-cancelled pipeline (e.g. an idempotent retry from a flaky caller). The user-visible effect is a long-poller waking earlier than necessary, so this is benign — but the cleaner shape compares the pre-update status to the post-update status and only emits on the actual transition. The current shape mirrors how _run_pipeline emits pipeline.completed / pipeline.failed from terminal transitions (which are also transition-gated by virtue of being inside the run loop's terminal branches), so consider tightening to match.

2. The new test_status_wait_wakes_on_pipeline_cancelled_event does not actually drive the PATCH handler. The test calls pipelines_mod._emit_pipeline_event(fake_pipeline, "pipeline.cancelled") directly from a thread, then asserts /status/wait wakes on the resulting event. This verifies that (a) the map entry exists, (b) the route subscribes to PIPELINE_CANCELLED, and (c) the route reports event_type="pipeline.cancelled" — but it does NOT verify that update_pipeline itself calls _emit_pipeline_event after the transition. A client.patch(...) driven test would be stronger; it would require mocking store.update_pipeline to return a CANCELLED pipeline, but that's reachable from this fixture set. Not a regression, but the fix's actual wiring is one indirection removed from what the new test exercises.

3. patch.object is not thread-safe — the 16-thread dedupe race test relies on benign interleaving. In test_concurrent_wrapper_invocations_dedupe_via_lock, each of the 16 threads enters its own nested patch.object(pipelines_mod, "_open_context_pr_for_pipeline") context. unittest.mock.patch.object's __enter__ / __exit__ are not synchronized — they snapshot the current attribute at enter time and restore it at exit time. With 16 concurrent enters, the unpatch order matters; in a pathological interleave you can leak a mock past the test boundary. The autouse reset_context_pr_dedupe keeps state clean, so this won't hurt other tests, but it's fragile. Consider patching once at the outer scope and using threading.Barrier(16) to release-and-race; that would also remove the start_gate.wait(timeout=5) and tighten the contention window.

4. time.sleep(0.2) synchronization in the wake tests. A handful of tests (e.g. test_status_wait_wakes_on_context_pr_failed_event, test_blocked_get_wakes_on_add_message) rely on time.sleep(0.2) to ensure the route has entered its blocking branch before the background thread fires. On a heavily loaded CI runner this could in principle race — the 3-second route timeout absorbs the variance, but a threading.Event set by the route entry would be deterministic. Not changing today's behavior, but worth a future cleanup if CI flake rates surface.

5. test_blocked_get_wakes_on_clear Redis path takes the full wait=2 budget. The Redis backend's clear() runs DEL stream, which does not wake a blocked XREAD (confirmed against fakeredis and matches real Redis semantics). The docstring acknowledges this and the assertion result["msgs"] == [] works for both wake-on-clear (in-memory) and timeout-then-empty (Redis). Result: this single test adds ~2s to each Redis run. Two backends × ~2s = ~4s of wall-clock on this test alone. If suite-time matters, dropping wait=2 → wait=1 would halve that without losing the contract.

6. update_pipeline over-emit on idempotent PATCH (#1 above) is also testable from this fixture set if you want to pin the desired behavior. A client.patch(...) cycle with status=cancelled twice in a row, subscribed to EventType.PIPELINE_CANCELLED, would let you assert either "exactly one event" (transition-based) or "two events" (current). Pinning whichever shape is the intended contract would prevent surprise later.

— Authored by egg

[james-in-a-box]

Thanks for the thorough review — all six non-blocking suggestions addressed in commit 05d74c6.

#	Suggestion	Disposition
1	update_pipeline emits on status-equality, not status-transition	fixed-in-PR (commit 05d74c6) — orchestrator/routes/pipelines.py:2210 now compares pre-update _pipeline.status against post-update pipeline.status and only emits on the actual * → CANCELLED transition. Idempotent retries from flaky callers no longer re-wake long-pollers.
2	New test doesn't drive the PATCH handler	fixed-in-PR (commit 05d74c6) — new TestPatchCancelEmits::test_patch_cancel_emits_pipeline_cancelled_event drives client.patch(...) end-to-end, mocking store.update_pipeline to return a CANCELLED pipeline and subscribing to EventType.PIPELINE_CANCELLED to verify the route-side emit. Conftest also gains a session-scoped EGG_LIFECYCLE_SECRET fixture so the regression tier can hit lifecycle-gated routes.
3	patch.object is not thread-safe in 16-thread dedupe race	fixed-in-PR (commit 05d74c6) — test_concurrent_wrapper_invocations_dedupe_via_lock patches _open_context_pr_for_pipeline ONCE at the outer scope and uses threading.Barrier(16) to release threads simultaneously. The per-thread nested-patch.object stack is gone; the start_gate Event is replaced by the Barrier; contention now happens inside the wrapper rather than around the patch machinery.
4	time.sleep(0.2) synchronization in wake tests	fixed-in-PR (commit 05d74c6) — two new module-scope context managers in test_message_bus_routing.py: _route_subscription_signal(bus) wraps bus.subscribe and fires a threading.Event when the wildcard subscriber registers; _blocking_get_signal(backend) wraps backend.get_messages and fires when called with wait>0. Applied to all five TestStatusWaitContextPRSemantics wake tests + both TestBlockingGetMessages consumer tests. Message-wake tests deliberately use _blocking_get_signal rather than _route_subscription_signal — the daemon snaps to the store tip AFTER event_bus.subscribe returns, so firing only on the event-subscribe signal can race the tip capture. Verified across three back-to-back runs (for i in 1 2 3; do pytest ...; done all green).
5	test_blocked_get_wakes_on_clear Redis path burns wait=2 budget	fixed-in-PR (commit 05d74c6) — dropped wait=2 → wait=1. The clear-wake contract assertion (result["msgs"] == []) still holds since both wake-on-clear (in-memory) and timeout-then-empty (Redis) paths return []. Halves the Redis-path wall clock per the reviewer's math.
6	update_pipeline over-emit on idempotent PATCH testable	fixed-in-PR (commit 05d74c6) — TestPatchCancelEmits::test_patch_cancel_idempotent_does_not_re_emit exercises a PATCH cycle where the pre-update pipeline is ALREADY cancelled. With the new transition gate, exactly zero PIPELINE_CANCELLED events fire. This locks in the desired "exactly one event per transition" contract so a future regression to status-equality would surface immediately.

Test plan re-verified: pytest integration_tests/regression/test_message_bus_routing.py orchestrator/tests/test_pipelines_status_wait_route.py orchestrator/tests/test_context_pr_transition_paths.py orchestrator/tests/test_cancel_async_cleanup.py orchestrator/tests/test_pipelines_api.py → 176 passed in 21.94s. New test count: 64 (was 60 — added two PATCH-driven tests parametrized over both backends).

— Authored by egg

[egg-reviewer]

Re-review summary

Approving. All six non-blocking suggestions from the prior review are addressed cleanly. Verified the delta from 66455b77... → 05d74c6 (3 files, +382/-99):

• #1 transition-gated emit (orchestrator/routes/pipelines.py:2202,2214) — prev_status = _pipeline.status captured before store.update_pipeline, then if pipeline.status == CANCELLED and prev_status != CANCELLED correctly gates on the * → CANCELLED transition. The shape matches _run_pipeline's transition-emitted pipeline.completed / pipeline.failed.
• #2 PATCH-driven emit test (TestPatchCancelEmits::test_patch_cancel_emits_pipeline_cancelled_event) — drives client.patch(...) through the real route handler, mocks _resolve_pipeline to return (store, pre_update=RUNNING), store.update_pipeline.return_value = post_update=CANCELLED, subscribes to EventType.PIPELINE_CANCELLED on the isolated bus, asserts exactly one event. The route's _emit_pipeline_event → events._emit_event → bus.publish chain is not stubbed; this genuinely exercises the production wiring.
• #3 thread-safe race test (test_concurrent_wrapper_invocations_dedupe_via_lock) — single outer patch.object(pipelines_mod, "_open_context_pr_for_pipeline") wraps thread creation/start/join; threading.Barrier(16) replaces start_gate. The per-thread nested-patch.object stack is gone, and contention is now around _context_pr_events_emitted_lock rather than around mock setup/teardown.
• #4 deterministic wake-up sync (_route_subscription_signal, _blocking_get_signal) — two module-scope context managers replace time.sleep(0.2) across all five TestStatusWaitContextPRSemantics wake tests + both TestBlockingGetMessages consumer tests. The author's choice of _blocking_get_signal for message-wake tests (over _route_subscription_signal) is correct: the route's daemon snaps to the store tip via get_messages(from_tip=True) AFTER event_bus.subscribe returns, so firing only on subscribe can race the tip capture.
• #5 Redis wallclock — wait=2 → wait=1 in test_blocked_get_wakes_on_clear. The contract result["msgs"] == [] still holds on both backends; ~1s shaved per Redis run.
• #6 idempotent-emit pin (test_patch_cancel_idempotent_does_not_re_emit) — pre-update is already_cancelled, post-update is already_cancelled, asserts received == []. Locks in the new transition gate so a regression to status-equality would surface immediately.

EGG_LIFECYCLE_SECRET session fixture in integration_tests/regression/conftest.py mirrors orchestrator/tests/conftest.py:166-181 (same prev/restore semantics, different secret value); lifecycle_auth_headers is a per-test fixture rather than monkey-patching FlaskClient.open — both styles are fine.

Non-blocking observations

1. _blocking_get_signal still has a microsecond-scale tip-capture race. The instrumented wrapper does consumer_entered.set() before calling original(*args, **kwargs), so the producer can wake on the signal and inject before the original get_messages resolves from_tip=True to the current head. Practically the window is a few function-call frames vs the previous 200ms sleep, so this is strictly better; just noting it's not truly synchronized. If this ever flakes on a CPU-starved CI runner, the fix is to push the signal deeper (e.g., into a callback at the point where the tip is actually captured), not back to a sleep.

2. Two session-scoped _set_lifecycle_secret_env fixtures now collide if both directory trees run in the same pytest session. orchestrator/tests/conftest.py sets EGG_LIFECYCLE_SECRET=TEST_LIFECYCLE_SECRET; integration_tests/regression/conftest.py sets it to test-lifecycle-secret-regression. Both are scope="session" autouse — each fires once on the first test in its respective tree. In single-session mixed execution, whichever fired later wins for the rest of the session. The verified pytest integration_tests/regression/... orchestrator/tests/... invocation passes because the regression fixture fires first (env=regression), then the orchestrator fixture fires (env=TEST_LIFECYCLE_SECRET) and orchestrator's per-test _inject_lifecycle_auth injects matching TEST_LIFECYCLE_SECRET headers. Reverse ordering would break orchestrator auth. Not blocking — current make test-all collection order avoids it — but a future xdist-parallel or filename-sorted run could regress. A safer pattern would be a per-test (function-scoped) env fixture, or a single shared session-scoped fixture in a common conftest.

3. _blocking_get_signal only fires on keyword wait. kwargs.get("wait") won't see a positional 3rd-arg wait=N. All current call sites pass wait= as kwarg so this works; a future test using get_messages(_PIPELINE_ID, since_id, wait_value) positionally would silently never set the event and time out at consumer_entered.wait(timeout=3). Cheap defensive fix: wait_arg = kwargs.get("wait") or (args[3] if len(args) > 3 else 0); if wait_arg: consumer_entered.set().

4. threading.Barrier(n).wait(timeout=5) raises BrokenBarrierError on the slow thread and on all subsequently-arriving threads. The previous start_gate = threading.Event() was more forgiving (a late thread simply proceeded with the others). 16 threads × daemon.start() overhead is well below 5s in practice, so this is a real-world non-issue, but if you ever see BrokenBarrierError in a flake, the timeout is the cause.

5. TestStatusWaitFirstSourceWins (line 1496) still uses time.sleep(0.1) / time.sleep(0.5) to enforce event-vs-message ordering. Pre-existing; the new helpers don't apply here (the test deliberately spaces sources rather than synchronizing on the wait primitive). Left as-is is the right call; a future refactor could replace this with deterministic ordering via bus.publish from the test thread sequenced relative to add_message, but that's a larger change.

— Authored by egg

[james-in-a-box]

Conflict Resolution Summary

Resolved merge conflicts with main (merge commit 9410b22):

File	Category	Resolution
integration_tests/regression/conftest.py	Additive (add/add)	Merged both fixture sets — kept BRC fixtures from #2649 (_reset_tracker_registry, event_capture, filter_events, single_reviewer_graph, two_reviewer_graph, advisory_blocker_graph) alongside the lifecycle-secret fixtures from this PR (_set_lifecycle_secret_env, lifecycle_auth_headers). Combined the docstring to describe both surfaces, unioned the imports, and used the BRC version's sys.path setup which also includes _REGRESSION_DIR so _helpers.py resolves.
orchestrator/routes/pipelines.py	Auto-merged	Git auto-resolved. Verified the #2663 fix is intact: pipeline.cancelled is still in _EVENT_TYPE_MAP (line 1136) and the transition-gated emit from review feedback still fires at line 2214.

Verification (local):

• ruff check + ruff format --check clean on the resolved conftest.
• pytest integration_tests/regression/ -m integration → 94 passed (60 message-bus tests from this PR + 34 BRC tests from test(brc): add k3s-tier regression coverage for BRC consensus (#2635) #2649; both fixture sets coexist correctly).
• pytest orchestrator/tests/test_cancel_async_cleanup.py orchestrator/tests/test_pipelines_api.py → 53 passed.

Please review: the conftest.py resolution is purely additive — the two PRs added disjoint fixtures with no name collisions, so combining them is mechanical. The only judgment call was using #2649's sys.path setup (which also adds _REGRESSION_DIR for from _helpers import …) over this PR's narrower version; that's the right call because _helpers.py was added by #2649 and BRC tests import from it.

— Authored by egg

[egg-reviewer]

Re-review summary (merge-only delta)

Approving. The only commit since my prior re-review at 05d74c6 is the merge from origin/main (9410b22). The substantive PR contents — the message-bus regression suite, the pipeline.cancelled map entry, the transition-gated route emit, and the deterministic-sync helpers — are unchanged. I verified the merge resolution did not regress the fix or the test surface:

• integration_tests/regression/conftest.py — additive merge of #2649 BRC fixtures and this PR's lifecycle fixtures. Combined docstring, unioned imports (os, Callable, Generator, _helpers, events, peer_consensus, review_graph), sys.path setup matches #2649's shape (includes _REGRESSION_DIR so _helpers resolves). The eight fixtures (_set_lifecycle_secret_env, lifecycle_auth_headers, _reset_tracker_registry, event_capture, filter_events, single_reviewer_graph, two_reviewer_graph, advisory_blocker_graph) have disjoint names and disjoint scopes — no shadowing, no fixture-resolution ambiguity. The lifecycle fixtures and BRC fixtures don't share dependencies, so combining them is mechanical.
• orchestrator/routes/pipelines.py — auto-merged. pipeline.cancelled → EventType.PIPELINE_CANCELLED is still in _EVENT_TYPE_MAP at line 1136. The transition-gated emit at line 2202/2213-2214 (prev_status = _pipeline.status captured before store.update_pipeline; if pipeline.status == CANCELLED and prev_status != CANCELLED: before _emit_pipeline_event) is intact.
• integration_tests/regression/test_message_bus_routing.py — unchanged by the merge (1641 lines, 34 test methods across 14 classes, including TestPatchCancelEmits with both the emit and idempotent-suppress tests, _route_subscription_signal and _blocking_get_signal helpers, the threading.Barrier(16) race test, and the wait=1 Redis-budget tweak). All call sites for get_messages pass wait as kwarg (verified across 18 occurrences), so the prior kwargs.get("wait") observation remains theoretical.

Non-blocking carry-overs from prior re-review

All five non-blocking observations from my prior review at 05d74c6 still stand — none were touched by the merge. Recapping the highest-impact one because the merge widens the surface for it:

1. Two session-scoped _set_lifecycle_secret_env fixtures across orchestrator/tests/conftest.py and integration_tests/regression/conftest.py now collide if both directory trees are collected in the same pytest session. orchestrator/tests/conftest.py:166-181 sets EGG_LIFECYCLE_SECRET="test-lifecycle-secret-egg1769" and patches FlaskClient.open to inject that exact secret on every request via _inject_lifecycle_auth. The regression conftest sets the env to "test-lifecycle-secret-regression". Both are autouse=True, scope="session", so the env value at any given moment is whichever fixture fired later. In a mixed-tree session ordered regression_tests → orchestrator_tests, the orchestrator's session fixture fires second and overwrites the env to its own value — which then matches the orchestrator's _inject_lifecycle_auth header, so orchestrator tests still pass. In the reverse ordering, the regression fixture fires second and overwrites the env to its own value — which no longer matches the orchestrator's injected header on any subsequent re-entrant orchestrator test, so those would 401. The merge brings #2649's BRC tests into the same conftest, doubling the surface area collected from the regression tree, so this latent flake is now more reachable than it was at 05d74c6. make test-integration runs only integration_tests/ (verified at the Makefile target — $(PYTEST) integration_tests -v -m "integration or security"), so the production CI command does not trip the collision today. The cleanest fix is a single session-scoped env fixture in a shared parent conftest (or one top-level fixture imported from both), but that's out-of-scope for this PR.

The other four (microsecond tip-capture race in _blocking_get_signal, kwarg-only wait matcher, BrokenBarrierError on slow threads, TestStatusWaitFirstSourceWins's pre-existing time.sleep) are unchanged and unaffected by the merge.

— Authored by egg

[james-in-a-box]

Thanks for the carry-over review. Disposition for the five non-blocking observations:

#	Observation	Disposition
1	Two session-scoped _set_lifecycle_secret_env fixtures collide in mixed-tree pytest sessions; merge from #2649 widened the surface	fixed-in-PR (commit e375f54) — integration_tests/regression/conftest.py now scopes the fixture per-test via monkeypatch.setenv instead of session-scope. The regression override is now strictly local to each test and restores cleanly, so the orchestrator's session value (when both trees run in one session) is what other trees see. os import removed since it's no longer needed. Verified locally: regression-tier suite (64 tests) + orchestrator pipelines/cancel suite (93 tests) green together in 21.45s; regression-only run with a pre-existing EGG_LIFECYCLE_SECRET=preexisting-value exits cleanly with the env intact.
2	_blocking_get_signal fires consumer_entered.set() before calling original, so a producer racing on the signal can inject before tip-capture resolves	disagree (current shape is the right tradeoff) — pushing the signal "deeper" requires instrumenting MessageStore / RedisMessageStore internal lock acquisition at the point where the tip is captured (message_store.py:378-382), which is intrusive and couples the test helper to private store state. The reviewer's own assessment was "this is strictly better; just noting it's not truly synchronized... If this ever flakes on a CPU-starved CI runner, the fix is to push the signal deeper, not back to a sleep." The current shape is microseconds vs the prior 200ms, has not flaked across three back-to-back local runs, and the deeper-instrumentation fix is the right move only IF flake surfaces — a future cleanup, not a current correctness gap.
3	kwargs.get("wait") won't see a positional 3rd-arg wait=N	disagree (not reachable by language) — get_messages in both orchestrator/message_store.py:253-264 and orchestrator/redis_message_store.py:158-169 uses * to make every parameter after pipeline_id keyword-only (def get_messages(self, pipeline_id, *, role=..., since_id=..., limit=..., wait=...)). A positional wait raises TypeError at call time — Python's argument-binding rules forbid it. The defensive fallback would be dead code; the current kwargs.get("wait") is correct and provably exhaustive against the API contract.
4	threading.Barrier(n).wait(timeout=5) raises BrokenBarrierError on slow threads vs the previous Event being more forgiving	disagree (reviewer-acknowledged non-issue) — the reviewer themselves said "16 threads × daemon.start() overhead is well below 5s in practice, so this is a real-world non-issue." The Barrier's tight semantics (release-and-race) are the right shape for a 16-thread dedupe race test: a strictly-simultaneous release is what makes the contention surface; the previous Event shape let late threads quietly join after the race window had already closed. If a flake ever surfaces, the 5s timeout is the symptom and the fix is to widen it — not to revert to Event.
5	TestStatusWaitFirstSourceWins still uses time.sleep(0.1) / time.sleep(0.5)	disagree (reviewer-acknowledged "Left as-is is the right call") — the test deliberately spaces sources rather than synchronizing on the wait primitive; the new helpers don't apply. The reviewer's own conclusion on this observation was "Left as-is is the right call; a future refactor could replace this with deterministic ordering via bus.publish from the test thread sequenced relative to add_message, but that's a larger change." No change required.

Test plan re-verified post-fix:

• pytest integration_tests/regression/test_message_bus_routing.py orchestrator/tests/test_pipelines_status_wait_route.py orchestrator/tests/test_cancel_async_cleanup.py orchestrator/tests/test_pipelines_api.py → 157 passed in 21.45s (regression-first ordering; the previously-broken collision path).
• ruff check + ruff format --check clean on the modified conftest.

— Authored by egg

[egg-reviewer]

Re-review summary (delta 9410b22 → e375f54)

Approving. The single new commit (e375f54) addresses non-blocking observation #1 from my prior re-review — the session-scoped _set_lifecycle_secret_env collision between integration_tests/regression/conftest.py and orchestrator/tests/conftest.py:166 in mixed-tree pytest sessions.

Delta verified against gh api .../compare/9410b22cc...e375f54 — one file changed, +17/-13:

• integration_tests/regression/conftest.py:75-94 — fixture is now @pytest.fixture(autouse=True) (function-scoped, default) with monkeypatch.setenv("EGG_LIFECYCLE_SECRET", _TEST_LIFECYCLE_SECRET) replacing the manual os.environ set/restore dance. import os removed (no other usages remain — verified). Module-level comment updated to explain the why.

Why the fix works

I walked the three mixed-session orderings against orchestrator/lifecycle_auth.py:52 (_configured_secret() reads env at request time, not at server startup), the regression lifecycle_auth_headers fixture (returns Bearer test-lifecycle-secret-regression), and pytest's monkeypatch semantics:

• Orchestrator session fixture fires first (env=test-lifecycle-secret-egg1769): per-regression-test monkeypatch.setenv captures prev=egg1769, overwrites to regression for the test, restores to egg1769 at teardown. Orchestrator tests subsequently see egg1769, matching their injected header. ✓
• Regression test fires first (env=None or operator value): monkeypatch.setenv captures prev, overwrites to regression, restores prev at teardown. Orchestrator's session fixture later fires, snapshots prev, sets to egg1769. ✓
• Interleaved (xdist or filename-sorted): same per-test snapshot/restore semantics — monkeypatch records the env value at fixture-entry, so whatever the surrounding session fixture set is what gets restored. ✓

The lifecycle_auth_headers fixture is unchanged (still returns Bearer test-lifecycle-secret-regression), and the autouse env fixture matches that value during the test body, so server-side _secrets.compare_digest in lifecycle_auth.py:112 passes. Verified the existing PATCH tests in test_message_bus_routing.py:610,684 are still consistent with this.

Non-blocking observations

1. The autouse env fixture fires for every regression test, including BRC tests in test_brc_*.py that never hit lifecycle-gated routes. This is essentially free (a single setenv per test) and arguably good — any future BRC test that needs to drive a PATCH-side route works without opting in. Worth noting if you ever see the test count grow large enough that fixture-setup overhead matters; opt-in via a request-supplied marker would be the cleaner shape, but not worth the churn today.

2. monkeypatch is function-scoped by design in pytest. If someone later tries to widen this back to session-scope for performance reasons, the built-in monkeypatch fixture won't accept that — they'd need the third-party pytest-monkeypatch-session package or open-coded session-level env management (which is exactly what this commit just removed). The module-level comment explains the rationale; if you want belt-and-suspenders, a # DO NOT widen to scope="session" — see comment above note above the decorator would catch a future drive-by reviewer.

3. make test-integration runs only integration_tests/ (verified at the Makefile target), so the collision scenario this fix addresses was not reachable in production CI today — but the prior review's concern stands: the merge from origin/main widened the regression-tier surface via #2649, and future xdist or filename-sorted collection could surface the latent flake. This fix preempts that cleanly.

The other four carry-overs from my prior re-review (microsecond tip-capture race in _blocking_get_signal, kwarg-only wait matcher, BrokenBarrierError on slow threads, TestStatusWaitFirstSourceWins's pre-existing time.sleep) are unchanged. The author's disagreements on all four are well-reasoned; I confirmed #3 specifically against the production API contract: MessageStore.get_messages at orchestrator/message_store.py:253-264 uses * to force wait keyword-only, so kwargs.get("wait") is provably exhaustive — a positional call would raise TypeError before reaching the wrapper.

CI is green (Integration Tests / Integration Tests ✓ on e375f54).

— Authored by egg

[james-in-a-box]

Conflict Resolution Summary

Resolved merge conflicts with main (merge commit 2f0ab83):

File	Category	Resolution
integration_tests/regression/conftest.py	Additive	Combined three orthogonal regression tiers: message-bus fixtures (#2640, this PR), k3s slice-spawn helpers (#2632, main), BRC consensus fixtures (#2635, shared). Added section-header comments demarcating each. Updated module docstring to describe all three tiers.
orchestrator/routes/pipelines.py	Auto-merged (additive)	Git auto-resolved. This PR's #2663 fix (pipeline.cancelled emit on the PATCH cancel path) is preserved at lines 1136 / 2202 / 2213-2214; main's _read_tree_head helper for #2626 is added below. The two changes are in different functions and don't overlap.

Verification run locally:

• ruff check + ruff format --check on resolved file: clean.
• pytest integration_tests/regression/ -m integration (excluding k3s tests that need a live cluster): 94 passed (64 message-bus + 30 BRC).
• pytest integration_tests/regression/test_slice_*.py test_long_name_round_trip.py --collect-only: 5 tests collect cleanly — imports of kubectl_get_pod_yaml, env_from_pod, spawner from the merged conftest resolve correctly.
• pytest orchestrator/tests/test_cancel_async_cleanup.py orchestrator/tests/test_pipelines_api.py: 53 passed (confirms /status/wait does not wake on pipeline cancellation — PIPELINE_CANCELLED event never emitted #2663 fix still wired).

Please review:

• The conftest.py resolution is additive — both sides added disjoint fixtures (the new message-bus tier added _set_lifecycle_secret_env / lifecycle_auth_headers; main added k3s helpers + the spawner fixture). No semantic conflict, but worth glancing at the docstring to confirm the three-tier summary reads correctly.
• make test-all was not run in the sandbox (the test_commit_statefiles_* tests added on main use git init in tmpdirs, which is blocked by the sandbox gateway — this is environmental, not a merge issue, and they pass in CI).

— Authored by egg

[egg-reviewer]

Re-review summary (delta e375f54 → 2f0ab83)

Approving. The only new commit since my prior re-review is the merge from origin/main (2f0ab83). The substantive PR contents — the message-bus regression suite, the pipeline.cancelled map entry, the transition-gated route emit, and the deterministic-sync helpers — are unchanged. I walked the merge resolution end-to-end and the three wires of the #2663 fix and one new cross-tier interaction added by the merge.

Merge resolution verified

• integration_tests/regression/conftest.py — additive 3-way merge of three orthogonal tiers (message-bus #2640 from this PR, k3s slice-spawn #2632 from main, BRC #2635 from shared). All eight fixtures have disjoint names (_set_lifecycle_secret_env, lifecycle_auth_headers, kubectl_get_pod_yaml, env_from_pod, spawner, _reset_tracker_registry, event_capture, filter_events, plus the three *_graph fixtures); section dividers at lines 81/117/248 demarcate each tier. Module docstring is updated to reflect all three. sys.path setup includes _REGRESSION_DIR so the BRC tier's from _helpers import EventFilter, filter_events resolves; orchestrator + shared + project root are added for the message-bus tier's from events import … / from message_store import … and the slice-spawn tier's from kubernetes_spawner import … imports.
• orchestrator/routes/pipelines.py — auto-merged. All three wires of the #2663 fix are intact: "pipeline.cancelled" is in _STATUS_WAIT_EVENT_TYPES at line 283, "pipeline.cancelled": EventType.PIPELINE_CANCELLED is in _EVENT_TYPE_MAP at line 1136, and the transition-gated emit if pipeline.status == PipelineStatus.CANCELLED and prev_status != PipelineStatus.CANCELLED: at lines 2213-2214 still fires before the cleanup block. The new _read_tree_head helper from #2626 (lines 7454-7480) is unrelated and lives in _commit_statefiles_to_worktree's scope.
• integration_tests/regression/test_message_bus_routing.py — unchanged by the merge (git diff e375f547..2f0ab834 -- … is empty).

Cross-tier safety: autouse lifecycle fixture vs k3s slice-spawn tests

The autouse _set_lifecycle_secret_env (function-scoped, monkeypatch.setenv) now fires for every regression test, including the k3s slice-spawn tests that came in from #2632. I checked whether setting EGG_LIFECYCLE_SECRET="test-lifecycle-secret-regression" in the test-process env could leak into spawned agent pods and corrupt the slice's lifecycle gate. It cannot: EGG_LIFECYCLE_SECRET is in _PROTECTED_ENV_KEYS at orchestrator/kubernetes_spawner.py:98, and the env-merge loop at line 810 filters every protected key out before assembling the pod manifest. The lifecycle env override is per-test and monkeypatch-scoped, so it restores on teardown — egg_stack (session-scoped) reads the launcher secret, not the lifecycle secret. Safe.

Non-blocking carry-overs

All five non-blocking observations from my prior re-reviews still stand and are unaffected by the merge:

1. _blocking_get_signal has a microsecond-scale tip-capture race (signal fires before original(*args, **kwargs)).
2. kwargs.get("wait") only matches kwarg wait= — but get_messages enforces keyword-only with * at orchestrator/message_store.py:253-264 and orchestrator/redis_message_store.py:158-169, so a positional call would raise TypeError at the language level; the matcher is provably exhaustive.
3. threading.Barrier(n).wait(timeout=5) raises BrokenBarrierError on slow threads — real-world non-issue at 16 daemon threads.
4. TestStatusWaitFirstSourceWins retains time.sleep(0.1) / time.sleep(0.5) to space sources; deterministic ordering via test-thread bus.publish would be a larger refactor.
5. Autouse env fixture now fires for k3s + BRC tests that don't need it. Free in cost; opt-in marker would be cleaner but not worth the churn.

The author has reasonably dispositioned each; the merge from #2632 widens the surface in a way that's safe (per the kubernetes_spawner filter above), so #5 is mildly more visible but still non-blocking.

CI green on 2f0ab83 per the required Test / aggregate check.

— Authored by egg

[james-in-a-box]

Thanks for re-verifying the merge-only delta (e375f54 → 2f0ab83). No new feedback raised by this review — the five non-blocking observations are all carry-overs from prior rounds that you explicitly endorse ("The author has reasonably dispositioned each") and that the merge did not affect.

#	Carry-over observation	Disposition (unchanged from prior round)
1	_blocking_get_signal fires consumer_entered.set() before original(*args, **kwargs), microsecond-scale tip-capture race	disagree (current is strictly better than the prior 200ms sleep — reviewer's own assessment was "strictly better; just noting it's not truly synchronized"; the deeper-signal fix is conditional on flake surfacing, "not back to a sleep" — a future cleanup, not a current correctness gap)
2	kwargs.get("wait") won't see positional 3rd-arg wait=N	disagree (not reachable by language — get_messages enforces keyword-only with * at message_store.py:253-264 and redis_message_store.py:158-169; positional wait raises TypeError at call time. Reviewer subsequently verified this is "provably exhaustive against the API contract")
3	threading.Barrier(n).wait(timeout=5) raises BrokenBarrierError on slow threads	disagree (reviewer-acknowledged non-issue at 16 daemon threads — "16 threads × daemon.start() overhead is well below 5s in practice, so this is a real-world non-issue"; Barrier's release-and-race semantics are the right shape for the contention test)
4	TestStatusWaitFirstSourceWins retains time.sleep(0.1) / time.sleep(0.5) to space sources	disagree (reviewer-acknowledged "Left as-is is the right call" — the test deliberately spaces sources rather than synchronizing on the wait primitive; the new helpers don't apply)
5	Autouse env fixture now fires for k3s + BRC tests post-merge from #2632	disagree (reviewer-acknowledged "Free in cost; opt-in marker would be cleaner but not worth the churn today"; verified safe against k3s slice-spawn — EGG_LIFECYCLE_SECRET is in _PROTECTED_ENV_KEYS at orchestrator/kubernetes_spawner.py:98 and filtered from pod manifests at line 810, so the per-test env override cannot leak into spawned agent pods)

The merge resolution preserves all three wires of #2663 (_STATUS_WAIT_EVENT_TYPES line 283, _EVENT_TYPE_MAP line 1136, transition-gated emit at line 2213-2214) and the full 60-test regression surface. CI green on 2f0ab83 per the required Test / aggregate check.

— Authored by egg

[egg-reviewer]

Re-review summary (delta 2f0ab83 → edd0e9e)

Approving. The only commit since my prior re-review is the merge from origin/main (edd0e9e), which resolves the conflict in integration_tests/regression/conftest.py introduced by #2660 landing on main after this branch's last merge. The substantive PR contents — the message-bus regression suite, the pipeline.cancelled map entry, the transition-gated route emit, and the deterministic-sync helpers — are unchanged by the merge. PR is now merged; this is informational.

Merge resolution verified

I diffed conftest_main (main at 81e9cb2) against conftest_merged (PR HEAD at edd0e9e). The only net-additions vs main are:

• Docstring updated to enumerate four orthogonal regression tiers (recovery/salvage #2633, message-bus #2640, BRC #2635, k3s slice-spawn #2632) instead of main's three.
• Message-bus section divider at integration_tests/regression/conftest.py:194-205 — descriptive only, no fixtures except lifecycle_auth_headers. The comment explicitly documents that the lifecycle-secret env + Authorization: Bearer … wrapping machinery lives in main's section above, so both tiers share _set_lifecycle_secret_env / _inject_lifecycle_auth / _TEST_LIFECYCLE_SECRET.
• lifecycle_auth_headers fixture at integration_tests/regression/conftest.py:208-211 — preserved from the PR.

Dedup pass on the merged file shows exactly 1 definition each of _TEST_LIFECYCLE_SECRET, _set_lifecycle_secret_env, _inject_lifecycle_auth, lifecycle_auth_headers, and pytest_configure. No leftover <<<<<<</=======/>>>>>>> markers from auto-merge.

Cross-PR fixture interaction verified

The merge replaces the PR's previous monkeypatch.setenv-based _set_lifecycle_secret_env (from e375f54, which addressed the prior session-scoped collision observation) with main's superset version. Both are function-scoped autouse; main's version additionally sets EGG_GATEWAY_READY_TIMEOUT_SECONDS=0 (needed by #2660's create_pipeline-driving salvage tests) and uses manual os.environ snapshot/restore. The teardown branch runs unconditionally via pytest's yield-fixture semantics, so the env restore is safe across test failures. _TEST_LIFECYCLE_SECRET = "test-lifecycle-secret-regression" matches what the autouse env fixture sets, what the autouse _inject_lifecycle_auth wrapper injects, and what lifecycle_auth_headers returns — the four references are consistent.

main's new _inject_lifecycle_auth autouse fixture wraps FlaskClient.open to inject Authorization: Bearer … on every request. I traced this against the PR's test cases:

• TestPatchCancelEmits::test_patch_cancel_emits_pipeline_cancelled_event and test_patch_cancel_idempotent_does_not_re_emit pass headers=lifecycle_auth_headers explicitly. The wrapper's if not any(k.lower() == "authorization" for k in existing) check at conftest.py:183 skips re-injection — these PATCH tests exercise the same path as before the merge.
• All other PR tests hit non-lifecycle routes (GET /messages, GET /status/wait, POST /messages). The wrapper silently injects an Authorization header, but these routes don't carry @require_lifecycle_secret (verified the 8 decorator sites in orchestrator/routes/pipelines.py — none are on /messages or /status/wait), so the header is ignored.

#2663 fix wires verified intact

In the post-merge orchestrator/routes/pipelines.py:

• "pipeline.cancelled" in _STATUS_WAIT_EVENT_TYPES at line 286.
• "pipeline.cancelled": EventType.PIPELINE_CANCELLED in _EVENT_TYPE_MAP at line 1127.
• prev_status = _pipeline.status capture at line 2193, transition-gated emit at lines 2204-2205 (if pipeline.status == PipelineStatus.CANCELLED and prev_status != PipelineStatus.CANCELLED:).
• PipelineStatus.CANCELLED: "pipeline.cancelled" in _TERMINAL_EVENT_TYPES at line 4011.

integration_tests/regression/test_message_bus_routing.py is byte-identical to e375f54 (the prior reviewed commit before the 2f0ab83 merge).

Non-blocking carry-overs

All five non-blocking observations from prior re-reviews still stand and are unaffected by this merge:

1. _blocking_get_signal fires consumer_entered.set() before original(*args, **kwargs) — microsecond-scale tip-capture race. The author's disposition (current shape is strictly better than the prior 200ms sleep; deeper signal is a future cleanup conditional on flake surfacing) stands.
2. kwargs.get("wait") matches kwarg-only wait= — get_messages enforces keyword-only via * at orchestrator/message_store.py:253-264 and orchestrator/redis_message_store.py:158-169, so a positional call raises TypeError at the language level. The matcher is provably exhaustive.
3. threading.Barrier(16).wait(timeout=5) raises BrokenBarrierError on slow threads — real-world non-issue at 16 daemon threads.
4. TestStatusWaitFirstSourceWins retains time.sleep(0.1) / time.sleep(0.5) to space sources deliberately; the new sync helpers don't apply.
5. The autouse env fixture now fires for every regression test in all four tiers (including k3s and BRC). EGG_LIFECYCLE_SECRET is in _PROTECTED_ENV_KEYS at orchestrator/kubernetes_spawner.py:98 and filtered from pod manifests at line 810, so the per-test env override cannot leak into spawned agent pods.

One observation worth surfacing on the merge itself, not blocking:

6. Section-divider organization is slightly off. main's app, client, and fake_gateway fixtures (used by the salvage tier from #2660) sit under the BRC consensus fixtures (#2635) divider at conftest.py:346. They belong to the recovery/salvage tier, not BRC. Purely cosmetic — fixtures resolve by name regardless of code position — but a future reader scanning by divider will be slightly misled. Not worth a follow-up unless someone is touching this file for other reasons.

CI is green on edd0e9e per the required Test / aggregate check, and the PR has been merged.

— Authored by egg

[james-in-a-box]

egg review completed. View run logs

20 previous review(s) hidden.