Abilityai · vybe · Apr 21, 2026 · Apr 20, 2026 · Apr 21, 2026 · Apr 21, 2026
diff --git a/CLAUDE.md b/CLAUDE.md
@@ -128,6 +128,7 @@ Before adding endpoints, services, DB tables, or frontend views, review the Arch
 | `docs/memory/requirements.md` | **SINGLE SOURCE OF TRUTH** - All features |
 | @docs/memory/architecture.md | Current system design (~1000 lines max) |
 | `docs/memory/feature-flows.md` | Index of vertical slice docs |
+| `docs/planning/ORCHESTRATION_RELIABILITY_2026-04.md` | Active multi-sprint plan for execution/orchestration reliability. **Current focus: Tier 2.5 Simplification — #306 (push event bus) → #428/#429/#430.** Consult before touching `task_execution_service`, `slot_service`, `backlog_service`, `execution_queue`, or `cleanup_service`. |
 | GitHub Issues + Project Board | Prioritized task queue — **Trinity Roadmap** board (Todo/In Progress/Done), priority labels (P0-P3), Tier sub-priority (P1a/P1b/P1c) |
 
 ---
@@ -312,6 +313,7 @@ The **[abilities](https://github.com/abilityai/abilities)** repo is the canonica
 ## See Also
 
 - **SDLC & Development Workflow**: `docs/DEVELOPMENT_WORKFLOW.md` ← Start here for dev process
+- **Orchestration Reliability Plan**: `docs/planning/ORCHESTRATION_RELIABILITY_2026-04.md` ← Active direction for execution stack; read before extending orchestration primitives
 - **Full Architecture**: @docs/memory/architecture.md
 - **All Requirements**: `.claude/memory/requirements.md`
 - **Current Roadmap**: https://github.com/abilityai/trinity/issues

diff --git a/docs/memory/architecture.md b/docs/memory/architecture.md
@@ -173,6 +173,9 @@ Each agent runs as an isolated Docker container with standardized interfaces for
 - `scheduler_service.py` - APScheduler-based scheduling service
 - `cleanup_service.py` - Active watchdog reconciliation + passive stale recovery for executions, activities, and slots (CLEANUP-001, #129)
 
+*Real-time delivery:*
+- `event_bus.py` - Redis Streams transport for WebSocket delivery (`EventBus` publisher + `StreamDispatcher` consumer, reconnect replay via `last-event-id`, 3-failure client eviction, MAXLEN-trimmed stream) (RELIABILITY-003, #306)
+
 *Monitoring & Activities:*
 - `activity_service.py` - Activity tracking and timeline
 - `monitoring_service.py` - Fleet-wide health monitoring (MON-001)
@@ -256,6 +259,7 @@ Each agent runs as an isolated Docker container with standardized interfaces for
 - WebSocket client at `utils/websocket.js`
 - Auto-reconnect on disconnect
 - Status update broadcasts
+- Tracks `_eid` (Redis stream id) on every incoming message; reconnect URL appends `&last-event-id=<id>` so brief disconnects replay missed events. On `{type: "resync_required"}` the cursor is cleared and authoritative state is refetched via REST (RELIABILITY-003, #306)
 
 **Collaboration Dashboard:**
 - Vue Flow for node-based graph visualization
@@ -634,7 +638,7 @@ These are structural patterns that must be preserved. Breaking them causes casca
 
 9. **Channel Adapter ABC** — External messaging (Slack, Telegram) follows `adapters/base.py` → `ChannelAdapter` ABC with `NormalizedMessage` and `ChannelResponse`. New channels must implement this interface.
 
-10. **WebSocket Events for Real-Time** — All real-time updates go through WebSocket broadcast (`agent_activity`, `agent_collaboration`). Frontend subscribes via `utils/websocket.js`. Don't poll for state that should be pushed.
+10. **WebSocket Events for Real-Time** — All real-time updates go through WebSocket broadcast (`agent_activity`, `agent_collaboration`). Frontend subscribes via `utils/websocket.js`. Don't poll for state that should be pushed. Transport is the Redis Streams event bus in `services/event_bus.py` (RELIABILITY-003, #306) — `ConnectionManager` / `FilteredWebSocketManager` are thin shims that `XADD` to `trinity:events`; the `StreamDispatcher` runs one `XREAD BLOCK` per backend process and fans out to registered clients. New broadcast sites should continue calling the existing `manager.broadcast(...)` / `filtered_manager.broadcast_filtered(...)` API — do not bypass it to publish directly.
 
 11. **Docker as Source of Truth** — Agent container state comes from Docker labels (`trinity.*`), not from an in-memory registry. `docker_service.py` is the single point of Docker interaction.
 
@@ -1294,6 +1298,8 @@ Internal endpoints (`/api/internal/`) used by the scheduler and agent containers
 
 The `/ws` endpoint requires JWT authentication. Token provided via `?token=` query parameter or as first message (`Bearer <token>`, 5s timeout). Unauthenticated connections are rejected. The `/ws/events` endpoint requires MCP API key authentication (unchanged).
 
+**Reconnect replay (RELIABILITY-003, #306):** Both `/ws` and `/ws/events` accept an optional `?last-event-id=<stream_id>` query param. The value is regex-gated (`^\d+-\d+$`) by `validate_last_event_id()` in `services/event_bus.py` before reaching `XRANGE`; malformed input is ignored (no catchup). Catchup is capped at `REPLAY_GAP_LIMIT=5000` entries — a larger gap returns `{"type": "resync_required", "reason": "gap_too_large"}` instead of an unbounded `XRANGE`. Authorization (`accessible_agents` for `/ws/events`) is re-applied on replay, not just on live fan-out.
+
 ### Frontend XSS Protection (H-005)
 
 All markdown rendering in Vue components uses `DOMPurify` sanitization via `utils/markdown.js`. No direct `v-html` with unsanitized content.

diff --git a/docs/memory/feature-flows.md b/docs/memory/feature-flows.md
@@ -11,6 +11,7 @@
 
 | Date | ID | Feature | Flow |
 |------|-----|---------|------|
+| 2026-04-21 | RELIABILITY-003 (#306) | WebSocket event bus on Redis Streams — replaces in-process broadcast with XADD/XREAD, adds reconnect replay via `?last-event-id=`, 3-failure eviction, MAXLEN trim (tunable) | [websocket-event-bus.md](feature-flows/websocket-event-bus.md) |
 | 2026-04-20 | #420 | Scheduler sync loop fix — `update_schedule_run_times` no longer bumps `updated_at`, stopping the self-triggering re-register of every schedule per tick | [scheduler-service.md](feature-flows/scheduler-service.md) |
 | 2026-04-20 | #418 | Inter-agent timeout honors per-agent `execution_timeout_seconds` — removed 600s hardcoded defaults in MCP `chat_with_agent`/`fan_out` tools and fan-out service; HTTP client ceiling bumped to platform max (7200s) | [fan-out.md](feature-flows/fan-out.md), [mcp-orchestration.md](feature-flows/mcp-orchestration.md), [parallel-headless-execution.md](feature-flows/parallel-headless-execution.md) |
 | 2026-04-19 | #211 | Auto-propagate global GitHub PAT to running agents on update — per-agent PAT holders and agents without `GITHUB_PAT` in `.env` are skipped; delete does NOT propagate | [github-sync.md](feature-flows/github-sync.md), [platform-settings.md](feature-flows/platform-settings.md) |
@@ -276,6 +277,7 @@
 | OpenTelemetry Integration | [opentelemetry-integration.md](feature-flows/opentelemetry-integration.md) | OTel metrics export |
 | Async Docker Operations | [async-docker-operations.md](feature-flows/async-docker-operations.md) | Non-blocking Docker SDK wrappers |
 | Cleanup Service | [cleanup-service.md](feature-flows/cleanup-service.md) | Active watchdog reconciliation + passive stale recovery for executions, activities, and slots (CLEANUP-001, #129) |
+| WebSocket Event Bus | [websocket-event-bus.md](feature-flows/websocket-event-bus.md) | Redis Streams transport for `/ws` + `/ws/events` with reconnect replay, per-client eviction, `MAXLEN` trim (RELIABILITY-003 / #306) |
 
 ### Templates & Pages
 

diff --git a/docs/memory/feature-flows/activity-stream.md b/docs/memory/feature-flows/activity-stream.md
@@ -146,6 +146,13 @@ class ActivityService:
 | 202-212 | `_notify_subscribers()` | Extensibility for plugins |
 | 214-244 | `_get_action_description()` | Human-readable descriptions |
 
+> **Transport**: `websocket_manager.broadcast(...)` and
+> `filtered_websocket_manager.broadcast_filtered(...)` are thin shims over the
+> Redis Streams event bus introduced in #306 — events are durably logged and
+> replayable on reconnect. See
+> [websocket-event-bus.md](websocket-event-bus.md). The activity service's
+> event shape is unchanged.
+
 **Activity Creation Flow** (`activity_service.py:46-107`)
 ```python
 async def track_activity(

diff --git a/docs/memory/feature-flows/websocket-event-bus.md b/docs/memory/feature-flows/websocket-event-bus.md
@@ -0,0 +1,250 @@
+# Feature: WebSocket Event Bus (Redis Streams)
+
+## Overview
+
+Real-time WebSocket delivery backed by a Redis Stream. Replaces the legacy
+in-process `ConnectionManager.broadcast(...)` + `except: pass` pattern with a
+durable event log so:
+
+- A momentary WebSocket disconnect no longer drops events — reconnecting with
+  `?last-event-id=<stream_id>` replays missed events from the stream.
+- Failed sends to a dead socket are logged and the client is evicted after
+  3 consecutive failures instead of silently accumulating zombie connections.
+- The stream has bounded memory (`XADD MAXLEN ~10000`, env-tunable) and
+  provides the substrate for later work: agent-push completion (#428/#429) and
+  heartbeat push (#307) will reuse the same stream primitive.
+
+Issue: [#306](https://github.com/abilityai/trinity/issues/306) — RELIABILITY-003.
+Positioned as the keystone of Tier 2.5 simplification in
+[`docs/planning/ORCHESTRATION_RELIABILITY_2026-04.md`](../../planning/ORCHESTRATION_RELIABILITY_2026-04.md).
+
+## User Story
+
+As a Trinity operator watching the Collaboration Dashboard or a Trinity Connect
+listener, I want real-time events to be reliable across short disconnects so
+my UI doesn't end up with timeline bars stuck on "started" or missed
+collaboration edges after a laptop sleep.
+
+## Entry Points
+
+- **Publisher shim (legacy API preserved)**: `src/backend/main.py`
+  - `ConnectionManager.broadcast(message)` — `/ws` broadcast. Accepts a
+    JSON-encoded string (legacy) or a dict. Internally calls
+    `event_bus.publish(message, scope=SCOPE_ALL)`.
+  - `FilteredWebSocketManager.broadcast_filtered(event)` — `/ws/events`
+    broadcast with per-user `accessible_agents` filter. Internally calls
+    `event_bus.publish(event, scope=SCOPE_SCOPED)`.
+- **Publisher core**: `src/backend/services/event_bus.py:EventBus.publish`
+- **WebSocket endpoints**:
+  - `GET /ws?token=<jwt>&last-event-id=<stream_id>` — `src/backend/main.py:634+`
+  - `GET /ws/events?token=trinity_mcp_...&last-event-id=<stream_id>` —
+    `src/backend/main.py:697+`
+- **Frontend WebSocket clients**:
+  - Main tab: `src/frontend/src/utils/websocket.js`
+  - Collaboration dashboard: `src/frontend/src/stores/network.js` (separate WS
+    connection)
+
+---
+
+## Architecture
+
+### Design Decisions
+
+**1. Two layers: producer (EventBus) + consumer (StreamDispatcher)**
+- `EventBus.publish()` is fire-and-forget: events land in a bounded
+  `asyncio.Queue(10_000)` and a background writer drains to Redis. Broadcast
+  call sites never block on Redis latency.
+- `StreamDispatcher` runs a single `XREAD BLOCK` coroutine per backend process
+  and fans out in-memory to registered WebSocket clients.
+- Rationale: 50+ per-connection `XREAD` calls would waste pool connections;
+  one reader with in-memory fan-out is both cheaper and simpler.
+
+**2. Single stream with `scope` field, not two streams**
+- Stream key: `trinity:events`
+- Every XADD carries `scope: "all" | "scoped"` and optional `agent_name`
+- `_event_is_visible(slot, scope, agent_name)` enforces:
+  - `/ws` clients (`scope=SCOPE_ALL`) see only `scope=all` events
+  - `/ws/events` clients (`scope=SCOPE_SCOPED`) see `scope=scoped` events,
+    filtered by `accessible_agents` (admins see all)
+- Rationale: Keeps the auth boundary in one place; avoids the 8 dual-broadcast
+  call sites having to `XADD` twice.
+
+**3. Per-client bounded queue, never await send from fan-out**
+- Each client slot has `asyncio.Queue(maxsize=256)`
+- `_fanout` does `put_nowait`; on `QueueFull`, flags the client for
+  `resync_required` and drops the event
+- A dedicated `_client_consumer` coroutine drains the queue and does the
+  actual `websocket.send_*`
+- Rationale: A slow client can't block fan-out for others (head-of-line
+  blocking prevention).
+
+**4. Reconnect replay with trim-race detection**
+- Client stores `lastEventId` in-memory (not localStorage — page reload wipes
+  stores anyway, so only sub-session reconnect matters)
+- On reconnect the client sends `?last-event-id=<id>`
+- Dispatcher snapshots `_last_stream_id` at registration time, then runs
+  `XRANGE (<id> max=<snapshot>` — the cap prevents catchup from overlapping
+  with live fan-out, which would otherwise double-deliver (#306 review C1)
+- Trimmed cursor → `{type: "resync_required", reason: "trimmed"}`; frontend
+  clears cursor and refetches authoritative state via REST
+
+**5. 3-failure eviction**
+- `EVICT_AFTER_FAILURES=3` consecutive send exceptions → close socket,
+  remove from dispatcher
+- Replaces the legacy `except: pass` silent-drop
+
+**6. Graceful degradation when Redis is unavailable**
+- `EventBus._xadd` catches errors, closes and reconnects the client on next
+  call
+- In-memory fallback buffer (`_FALLBACK_BUFFER_MAX=1024`) holds events for a
+  brief Redis outage; drops silently when full
+- Trade-off: no live updates when Redis is down; rest of Trinity is already
+  broken in that case (Redis is a hard dependency for credentials,
+  rate-limiting, sessions)
+
+---
+
+## Flow
+
+### 1. Publish path
+```
+broadcast site
+   │ manager.broadcast(str|dict)  or  filtered_manager.broadcast_filtered(dict)
+   ▼
+EventBus.publish(event, scope)
+   │ wraps into envelope: {payload, scope, agent_name}
+   │ put_nowait onto self._outbound (asyncio.Queue, cap 10_000)
+   ▼
+EventBus._writer_loop (background task)
+   │ drains outbound queue
+   ▼
+redis.xadd(STREAM_KEY, fields, maxlen=10000, approximate=True)
+```
+
+### 2. Consume path (live)
+```
+StreamDispatcher._reader_loop (single background task per process)
+   │ redis.xread({STREAM_KEY: self._last_stream_id}, block=5000, count=100)
+   │ updates self._last_stream_id
+   ▼
+StreamDispatcher._fanout(entry_id, fields)
+   │ deserialize + inject _eid into payload
+   ▼
+for each slot in self._clients:
+   │ if _event_is_visible(slot, scope, agent_name):
+   │     slot.queue.put_nowait((entry_id, payload))
+   │     # QueueFull → mark resync_pending, enqueue resync_required marker
+   ▼
+_client_consumer (one per WS connection)
+   │ await queue.get()
+   │ await slot.send_func(payload)   # websocket.send_text / send_json
+   │ on Exception: failure_count++; evict at EVICT_AFTER_FAILURES
+```
+
+### 3. Reconnect replay
+```
+browser reconnects with ?last-event-id=<id>
+   │
+   ▼ validate_last_event_id(raw)   # regex: ^\d+-\d+$ (security gate)
+   │
+   ▼ manager.connect(ws, last_event_id=validated_id)
+   │ snapshot catchup_max = dispatcher._last_stream_id  (before adding to _clients)
+   │ register slot → fan-out starts delivering events > snapshot
+   ▼
+_catchup(client_id, slot, last_event_id, catchup_max)
+   │ if gap > REPLAY_GAP_LIMIT (5000) → queue resync_required("gap_too_large")
+   │ else XRANGE(STREAM_KEY, min="(<id>", max=catchup_max, count=5001)
+   │ check for trim race: if oldest_id > last_event_id → resync_required("trimmed")
+   ▼ for each entry matching scope/access → put_nowait onto slot.queue
+```
+
+### 4. Frontend lastEventId contract
+- **Set**: on every incoming message with `_eid`, store in module-scoped var
+- **Send**: on (re)connect, append `&last-event-id=<id>` to WS URL
+- **Clear**: on `{type: "resync_required"}` message, null out and call
+  authoritative REST refetchers (`fetchAgents()`, `fetchHistoricalCollaborations()`,
+  `fetchPendingCount()`)
+
+---
+
+## Security
+
+- **Authentication**: unchanged. `/ws` requires JWT; `/ws/events` requires
+  `trinity_mcp_*` API key. Consumer tasks inherit connection-level auth.
+- **Input validation**: `?last-event-id=` is regex-gated
+  (`EID_PATTERN = ^\d+-\d+$`) in `validate_last_event_id()` before reaching
+  `XRANGE`. Malformed input → `None` → no catchup.
+- **Authorization**: `_event_is_visible` is applied on both live fan-out AND
+  catchup replay. `accessible_agents` is read at event-delivery time (not
+  cached at connect), so changes via `update_accessible_agents` take effect
+  immediately.
+- **Replay DoS ceiling**: `REPLAY_GAP_LIMIT=5000`. Larger gap → reject with
+  `resync_required` instead of an unbounded XRANGE.
+- **Stream memory**: `REDIS_STREAM_MAXLEN` env var, default 10_000.
+
+See `docs/security-reports/cso-2026-04-21-diff-306.md` (if saved) for the
+`/cso --diff` audit.
+
+---
+
+## Observability
+
+Log events (via stdlib `logging`, captured by Vector):
+- `event_bus: writer error: <err>; backoff=<s>` — Redis XADD failure
+- `event_bus: Redis unavailable (<err>); publish will degrade` — on startup
+- `event_bus: outbound queue full, dropping event` — publisher side saturation
+- `stream_dispatcher: reader crashed: <err>; restart in <s>` — supervised
+  reader restart
+- `stream_dispatcher: client <id-prefix> queue full, marking resync` — slow
+  consumer
+- `stream_dispatcher: send failed for <id-prefix> (N/3): <err>` — per-attempt
+- `stream_dispatcher: evicting client <id-prefix> after 3 failures`
+
+---
+
+## Key Files
+
+| Layer | File | Role |
+|-------|------|------|
+| Service | `src/backend/services/event_bus.py` | `EventBus`, `StreamDispatcher`, scope helpers, `validate_last_event_id` |
+| Entry | `src/backend/main.py:112-200` | `ConnectionManager` / `FilteredWebSocketManager` shims over the bus |
+| Entry | `src/backend/main.py:634+`, `main.py:697+` | `/ws` and `/ws/events` endpoints with `?last-event-id=` support |
+| Entry | `src/backend/main.py:285-294` | Lifespan `event_bus.start()` + `stream_dispatcher.start()` |
+| Entry | `src/backend/main.py:538-547` | Lifespan 2s graceful drain on shutdown |
+| Client | `src/frontend/src/utils/websocket.js` | Main WebSocket client; `_eid` capture, reconnect replay, `resync_required` → REST refetch |
+| Client | `src/frontend/src/stores/network.js:535+` | Collaboration dashboard WS client; same contract, separate `lastEventId` |
+| Tests | `tests/test_event_bus.py` | 23 unit tests — envelope, scope visibility, eviction, slow-consumer resync, monotonic cursor guard, catchup trim detection |
+
+## Constants
+
+| Name | Value | Purpose |
+|------|-------|---------|
+| `STREAM_KEY` | `"trinity:events"` | Redis stream name |
+| `STREAM_MAXLEN` | env `REDIS_STREAM_MAXLEN`, default 10000 | Approximate trim target |
+| `CLIENT_QUEUE_MAXSIZE` | 256 | Per-client buffer; overflow triggers resync |
+| `EVICT_AFTER_FAILURES` | 3 | Consecutive send failures before eviction |
+| `REPLAY_GAP_LIMIT` | 5000 | Max replay size before forced resync |
+| `_FALLBACK_BUFFER_MAX` | 1024 | In-process buffer during Redis outage |
+
+---
+
+## Scope Discipline (#306 vs #428/#429/#307)
+
+This flow covers **WebSocket delivery only**. The following are intentionally
+deferred (see the orchestration reliability plan):
+
+- **#428 (CAPACITY-CONSOLIDATE)** — replaces `ExecutionQueue`/`SlotService`/
+  `BacklogService` with a single `CapacityManager`. Will consume agent
+  completion events from a dedicated stream.
+- **#429 (CLEANUP-COLLAPSE)** — retires the 9-path cleanup pyramid once
+  agent-push completion is authoritative. Gated on ≥2 weeks of push in
+  production with zero observed orphans.
+- **#307 (RELIABILITY-004)** — flips agent heartbeat from 30s polling to 5s
+  push. Reuses the stream primitive established here.
+- **#408** — dissolves once agent-push completion retires the 1h blocking
+  HTTP call in `TaskExecutionService`.
+
+The additive-first migration rule: new paths ship alongside old ones, old
+code is deleted only after proof. The legacy `manager.broadcast(...)` call
+signature is preserved across all 33 broadcast sites, so no call-site change
+was needed for this issue.