Orchestrata Ultimate › Plans › pipeline-event-convergence-design

Pipeline-Event Convergence — Design

Source: plans/pipeline-event-convergence-design.md at 8a2cb42ad · last changed 2026-08-19 · first committed 2026-08-05 · 415 lines.

Prose pending.

Plan text

Pipeline-Event Convergence — Design

Status: DESIGN ONLY (RUL-022 — no implementation in this artifact) Branch: feat/cxp-direct-transport · Date: 2026-08-05 · Author: analyst (design phase) Defect: two producers publish onto the UI topic pipeline.<thread_id> with two incompatible payload shapes. Producer B (kernel/pipeline_events.py::PipelineEventBus) is the SPEC-001-ratified taxonomy; Producer A (orchestrator/pipeline_kernel.py::PipelineKernelClient._emit_event) is an ad-hoc flat shape certified only by REQ-003-US-004-102, which contradicts SPEC-001. Every UI gadget and every shape-asserting test today matches Producer A. Producer A also carries a dead Redis fallback (pipeline:events) that nothing reads.

Every as-built claim below was grounded via the AST toolkit (scripts/serialize_codebase_ast.py {symbols,refs,source}) plus direct reads of the non-Python artifacts (gadget JS, specs, e2e). Specs are authoritative (RUL-013). Rulings consulted: RUL-002 (no mocks), RUL-003 (env from profile), RUL-006/024 (fix-on-sight; pre-existing is not a disposition — two extra defects found during grounding are IN scope: §1.5), RUL-009 (observed test runs), RUL-013 (specs win), RUL-022 (this document), RUL-023 (wired is done — §7).

Companion contract: plans/contracts/pipeline-event-convergence.mmd

1. AS-BUILT INVENTORY (verified, file:line)

1.1 Producer B — the ratified target

1.2 Producer A — the retirement target

1.3 Delivery plumbing (shared, out of the fork)

• gateway/server.py:1083-1097 WebSocketGateway.broadcast → protocol.py:115 wraps the payload in the {sequence, timestamp, payload} envelope.
• vizint-ui/bridge/websocket-bridge.js:232-252 unwraps params.payload and re-emits on the shared bus as ws.message {method: topic, params: <inner event dict>} — gadgets always see the raw inner dict. Envelope untouched by this epic.

1.4 Consumers (all match Producer A today)

1.5 Additional defects found while grounding (in scope per RUL-006/024)

2. DECISION LOG

D1 — Fold mechanism: real PipelineEvents over the existing _broadcast seam (option b)

PipelineKernelClient constructs genuine PipelineEvent objects and delivers event.to_payload() over its already-injected self._broadcast — it does not instantiate a PipelineEventBus.

Rejecting option (a) (client owns a bus + gateway Redis, forwarder delivers) on three verified grounds:

1. The forwarder is unconditional (gateway/server.py:731 → :633-641, gated only on the ring-buffer client), so a bus-writing fallback client is only dedup-safe if it stops broadcasting directly — at which point every fallback event pays a Redis round-trip + a pump poll window, and dies entirely when Redis is down (the direct path works Redis-less today).
2. The production construction site passes no Redis client (gateway/server.py:618), and PipelineEventBus's Redis calls are synchronous (emit :149-155, _forward_to_ui :193-205) — calling them from the gateway's event loop would block it; the gateway's own Redis client is async and incompatible.
3. Ordering: the direct path awaits each broadcast inline; the stream path adds a consumer hop. Stage-event ordering is a UI correctness property.

Dedup guarantee after this design: the fallback client only broadcasts (never writes the stream); the kernel bus only writes the stream (forwarder broadcasts). One event, one delivery, on either path. Packet 3a pins this with an exactly-once assertion.

D2 — Delete the dead fallback-fallback

PIPELINE_STREAM_KEY = "pipeline:events" (orchestrator/pipeline_kernel.py:63), the XADD branch (:503-513), the redis_client constructor parameter and self._redis attribute are all deleted. Evidence: zero readers repo-wide (§1.2); the only production constructor passes no Redis; cross-process delivery is Producer B's job (the kernel's own bus → gw:ringbuf:pipeline.events → forwarder). The three tests that exist solely to exercise the dead branch are deleted with it (Packet 3a).

D3 — Taxonomy: extend by 4 members (9 → 13); map start→INTENT and error→COMPLETE

The enum lacks homes for stage/paused/resumed/cancelled, and mapping them onto existing members would falsify their ratified semantics (ESCALATION carries from/to/subject, not status/duration_ms). Conversely start and error map faithfully onto existing members, so no START/ERROR members are added:

• pipeline.start → INTENT — the addendum's own gloss for INTENT is "intent submitted, thread created", which is exactly what _execute_pipeline :208-217 marks.
• pipeline.error → COMPLETE with data.success=False — the addendum's gloss for COMPLETE is "pipeline finished (success or failure)", and emit_complete already carries success: bool + reason (:410-429).

New members: STAGE = "pipeline.stage", PAUSED = "pipeline.paused", RESUMED = "pipeline.resumed", CANCELLED = "pipeline.cancelled". This is executed under the addendum's own drift rules: REQ-006-031's member list is amended to "exactly 13" (Packet 1d), the drift-guard test's member map / len / reflective-completeness blocks are extended (Packet 1c), and four new emit_* helpers keep the REQ-006-033 invariant (emitted set == enum set, no helper left behind). REQ-006-032's golden wire bytes are untouched (additive members only).

D4 — The shared serializer, named and placed

Two single-source seams, both in kernel/pipeline_events.py:

1. PipelineEvent.to_payload(self) -> dict[str, Any] — the canonical broadcast serialization, implemented as exactly asdict(self) so the REQ-006-032 golden-bytes test also pins it. PipelineEventBus.emit :138 switches from inline asdict(event) to event.to_payload(); PipelineKernelClient._emit_event uses the same call.
2. Module-level event factories build_intent_event, build_stage_event, build_complete_event, build_paused_event, build_resumed_event, build_cancelled_event — each returns a PipelineEvent with the canonical data shape. The bus helpers (emit_intent, emit_complete, and the four new ones) delegate to these factories; PipelineKernelClient constructs its events through the same factories. Data-shape drift between the two producers becomes structurally impossible.

D5 — Compat window: dual-emit legacy flat fields, fallback client only, one deploy cycle

• Who: only PipelineKernelClient._emit_event merges the legacy flat fields (_legacy_compat_fields(event) -> dict, gated by module flag _EMIT_LEGACY_COMPAT = True) into the canonical payload. The kernel bus never emits legacy fields — polluting the ratified stream would break the REQ-006-032 golden-bytes guard and poison the durable per-thread replay streams.
• Why the full flat set, not just an event alias: stale browser tabs read payload.stage, payload.status, etc. directly (§1.4); an alias alone would not keep them rendering. Merge is additive-only — the sole colliding keys (thread_id, timestamp) carry identical values on both shapes.
• How long: exactly one deploy cycle of the vizint UI + gateway (they ship from the same checkout; gadgets are static files, so a gateway restart + tab reload picks up Packet 2). Removal trigger: the first deploy AFTER this epic lands; executed as Packet 5 (delete _legacy_compat_fields + the flag), gated on pipeline-e2e.spec.js green while asserting the canonical shape only.

D6 — Normalize iteration to int in the forwarder decoder

_decode_pipeline_event (gateway/server.py:274-290) gains an int(...)-with-fallback-0 cast for iteration, closing the G2 path fork (§1.5). The direct path already delivers int. Consumers found reading iteration: none — the fix is contract hygiene ahead of Packet 2's data.* convention.

D7 — Shipping-shape proof: a real-infra Python test of the Producer-B pipeline, plus the e2e flip

The only shape assertion today (pipeline-e2e.spec.js:180) runs with GATEWAY_GRPC_READY_TIMEOUT_S='1' (:36) — it can only ever see the fallback producer. Two-part fix (Packet 3c/3d):

• 3c (new): tests/gateway/test_pipeline_forwarder_canonical_shape.py — real Redis, no doubles: a PipelineEventBus(redis, events_key=<test key>, ui_stream_key=<test stream>) emits via the typed helpers; the production pump _pump_pipeline_events_once(..., stream_key=<same test stream>) consumes; the captured broadcast payload is asserted canonical (event_type is a PipelineEventType value, iteration is int, data is a decoded dict, thread_id/volk/timestamp present). This exercises the exact two production halves that run when the real kernel is live, without the shared-key batch-flake the stream_key seam exists to avoid (gateway/server.py:158-162). Requires the small ui_stream_key injection parameter on PipelineEventBus.__init__ (Packet 1b), mirroring the existing events_key seam.
• 3d: flip :180 to params.event_type so the browser-level fallback path is asserted on the canonical field (it would pass on the compat field otherwise — flip it anyway so Packet 5 cannot regress silently).

D8 — Gadget dispatch rules (Packet 2 contract)

• Dispatch key: payload.event_type (never payload.event).
• Single COMPLETE branch: ok = payload.data.success === true; failure rendering when !ok (show payload.data.error when present, else payload.data.reason). The separate pipeline.error branches are deleted.
• Stage fields move to payload.data.* verbatim.
• New lifecycle branches (pipeline.paused/resumed/cancelled → thread status paused/running/cancelled) are added where the gadget tracks a per-thread status (DebugPanel, Orchestrator) — these events were already emitted by the workflow facade (SPEC-014 REQ-014-077/078/079) and silently dropped by every gadget.
• DebugPanel gains a pipeline.intent branch (thread.intent = payload.data.intent), replacing its dead payload.intent read in the stage branch.

3. EVENT-NAME + PAYLOAD MAPPING

3.1 Producer A → canonical (the Packet 1e fold)

Top-level on every event (from PipelineEvent): event_type, thread_id, volk, iteration (int, 0 here), timestamp (ISO-8601, dataclass default replaces _now_iso() — delete _now_iso if unused confirms no other caller).

3.2 Existing canonical data shapes (unchanged, for gadget follow-ups)

3.3 Per-gadget field mapping (Packet 2)

CXPFeed (cxp-feed/gadget.js:366-421):

DebugPanel (debug-panel/gadget.js:57-97):

Orchestrator (orchestrator/gadget.js:104-127 _onPipelineEvent(topic, data)):

4. PACKET BRIEFS (each ≤5-min dispatch, one-file-one-fix; dependency order as listed)

Atomicity constraint: 1a-1d must land in ONE commit — the enum/drift-guard/ addendum triangle (REQ-006-031's "exactly N" appears in all three) goes red in any partial state. 1e/1f may follow in separate commits. Packets 2 and 3 depend on 1; 2a/2b/2c and 3a/3b/3c/3d are mutually independent (parallel-safe). Packet 4 depends only on 1e. Packet 5 is deferred one deploy cycle.

Packet 1 — root fix (kernel taxonomy + producer fold)

• 1a src/orchestrata_ultimate/kernel/pipeline_events.py — add 4 enum members (STAGE/PAUSED/RESUMED/CANCELLED, §D3); add PipelineEvent.to_payload() (=asdict, §D4); add the 6 build_*_event module factories; switch emit :138 to event.to_payload(). Gate: python scripts/run_tests_streaming.py tests/kernel/test_pipeline_events_enum.py tests/kernel/test_pipeline_event_forwarder.py -j 1 (RED until 1c — expected), uv run ruff check, black.
• 1b same file — refactor emit_intent/emit_complete to delegate to their factories; add emit_stage/emit_paused/emit_resumed/emit_cancelled (data shapes §3.1); add ui_stream_key: str | None = None to __init__ and use it in _forward_to_ui for the global-stream XADD (default = PIPELINE_STREAM_KEY). Gate: same as 1a.
• 1c tests/kernel/test_pipeline_events_enum.py — extend the member map + len 9→13; add the 4 new helper calls to test_every_emit_method_emits_a_valid_enum_member (9→13) and test_emitted_types_match_enum; golden-bytes tests unchanged (must stay green byte-identically). Gate: that module green, observed.
• 1d specs/001-kernel/pipeline-event-type-addendum.md — amend REQ-006-031 to the 13-member list ("exactly 13"), append a dated amendment note naming this design doc and the fallback-producer convergence as the driver. Gate: python scripts/check_req_format.py.
• 1e src/orchestrata_ultimate/orchestrator/pipeline_kernel.py — replace all 10 emission sites per §3.1 using the build_*_event factories + event.to_payload(); _emit_event signature becomes async def _emit_event(self, event: PipelineEvent) -> None (topic from pipeline_topic(event.thread_id)); add _legacy_compat_fields + _EMIT_LEGACY_COMPAT (§D5); DELETE the Redis fallback branch, redis_client param, self._redis, PIPELINE_STREAM_KEY (:63), and _now_iso if now unused (verify: python scripts/serialize_codebase_ast.py unused). @traces reqs unchanged (REQ-003-US-004-102 survives; its text is amended in Packet 4). Gate: python scripts/run_tests_streaming.py tests/orchestrator/test_pipeline_kernel.py tests/gateway/handlers/test_workflow.py -j 1 (RED until Packet 3 — pair 1e with 3a/3b in one dispatch window if the 5-min budget allows, else accept the known-red handoff note to the next packet).
• 1f src/orchestrata_ultimate/gateway/server.py — _decode_pipeline_event: cast iteration to int (fallback 0) after decode (§D6); add the matching assertion in tests/kernel/test_pipeline_event_forwarder.py (seed :128 decodes to 0, an int). Gate: python scripts/run_tests_streaming.py tests/kernel/test_pipeline_event_forwarder.py -j 1.

Packet 2 — gadget convergence (one file each, mapping tables §3.3)

• 2a vizint-ui/gadgets/cxp-feed/gadget.js. Gate: cd vizint-ui && npx eslint gadgets/cxp-feed/gadget.js.
• 2b vizint-ui/gadgets/debug-panel/gadget.js. Gate: eslint likewise.
• 2c vizint-ui/gadgets/orchestrator/gadget.js. Gate: eslint likewise.

Packet 3 — test convergence

• 3a tests/orchestrator/test_pipeline_kernel.py — rewrite the shape asserts (:207, :239-242, :287-289, :324-363, :368-380) to canonical (§3.1): event_type values, data.* fields, top-level thread_id/volk/iteration/timestamp; add an exactly-once assertion (one broadcast per emission — D1's dedup pin); DELETE the three dead-branch tests (test_emit_event_falls_back_to_redis_stream_when_no_broadcast, test_emit_event_swallows_redis_xadd_exception) and drop the _redis assert from test_emit_event_noop_when_no_broadcast_and_no_redis. Gate: module green, observed.
• 3b tests/gateway/handlers/test_workflow.py — helper gate :73 → data.get("event_type") == "pipeline.intent"; asserts :141/:202/:229/:265 → event_type == pipeline.intent/paused/resumed/cancelled. Gate: module green.
• 3c NEW tests/gateway/test_pipeline_forwarder_canonical_shape.py — the shipping-shape proof (§D7): real Redis, PipelineEventBus(events_key=<test>, ui_stream_key=<test>) + _pump_pipeline_events_once(stream_key=<test>); asserts canonical payload incl. iteration: int and event_type in set(PipelineEventType); standard test_id isolation, no doubles (RUL-002). Gate: module green, observed.
• 3d vizint-ui/tests/e2e/pipeline-e2e.spec.js:180 → params.event_type. Gate: cd vizint-ui && npx playwright test tests/e2e/pipeline-e2e.spec.js (≈2 min, within the 5-min cap).

Packet 4 — REQ-003-US-004-102 amendment (specs/003-intent/checklists/requirements.md:260), replacement text:

REQ-003-US-004-102: _run_pipeline and the workflow-control facade publish their lifecycle events as SPEC-001 PipelineEvents — event_type drawn from the PipelineEventType taxonomy (specs/001-kernel/pipeline-event-type-addendum.md, REQ-006-031): pipeline.intent at run start, pipeline.stage per stage result, pipeline.complete on finish (success or failure, discriminated by data.success / data.error), and pipeline.paused / pipeline.resumed / pipeline.cancelled from pause/resume/cancel — serialized via the shared PipelineEvent.to_payload() and broadcast in-process on pipeline.<thread_id> through the gateway-injected broadcast callable. No bespoke event names; no parallel Redis stream.

Gate: python scripts/check_req_format.py, python scripts/trace_intel.py untraced-reqs.

Packet 5 — compat removal (NEXT deploy cycle, trigger per D5) — orchestrator/pipeline_kernel.py: delete _legacy_compat_fields + _EMIT_LEGACY_COMPAT; re-run 3a + 3d gates asserting canonical-only.

5. WIRING PLAN (RUL-023 / Hard Rule 8)

No new production module is introduced; every edit lands inside code that already has production importers, verified via refs/who-imports:

• Producer B delivery: gateway/server.py:731 start() → _init_pipeline_forwarder (:633-641) → _consume_pipeline_events (:841) — live in every gateway boot with Redis.
• Producer A successor: constructed at the single production site gateway/server.py:612-620 with broadcast=self.broadcast (delivery seam intact).
• Kernel-side emitters: kernel/pipeline.py:129, kernel/pipeline_escalation.py:42, kernel/pipeline_session.py:81 construct the bus (new enum members/helpers reachable the moment they land).
• UI: vizint-ui/bridge/websocket-bridge.js:239-251 fan-out (unchanged) → gadgets (served statically from the checkout; a gateway deploy + tab reload activates Packet 2 — the compat window D5 covers un-reloaded tabs).
• Deployment unit: kernel + gateway roll together as systemd services in the orchestrata-test app container (10.93.53.13, deploy/orchestrata-test/).

6. RISK REGISTER

7. OUT OF SCOPE / NAMED FOLLOW-UPS

• Retrospective stage events — _execute_pipeline emits stage events only after runner.run() returns (:224-238). Live per-stage streaming needs a stage callback on PipelineRunner; separate design.
• Gadget rendering of kernel-native events (escalation, swarm.*, review, test.result) — Packet 2 converges shape only; richer rendering is a UI feature.
• WS envelope ({sequence, timestamp, payload}) — explicitly not part of this fork.
• Kernel-side pause/resume/cancel emission — the real kernel does not yet emit the new lifecycle members (R7).

Related

• Plans
• Orchestrata Ultimate — project index
• Workflow
• Tasks