diff --git a/CLAUDE.md b/CLAUDE.md index 07e3b246d..dc91f14d6 100644 --- 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 index cab109e9b..3dc3a1432 100644 --- 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=` 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 `, 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=` 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 index cc74035af..475b5fcb6 100644 --- 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 index 5c6a719eb..f6c72a1f3 100644 --- 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 new file mode 100644 index 000000000..acc8f2cc4 --- /dev/null +++ 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=` 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=&last-event-id=` — `src/backend/main.py:634+` + - `GET /ws/events?token=trinity_mcp_...&last-event-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=` +- Dispatcher snapshots `_last_stream_id` at registration time, then runs + `XRANGE ( max=` — 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= + │ + ▼ 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="(", 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=` 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: ; backoff=` — Redis XADD failure +- `event_bus: Redis unavailable (); publish will degrade` — on startup +- `event_bus: outbound queue full, dropping event` — publisher side saturation +- `stream_dispatcher: reader crashed: ; restart in ` — supervised + reader restart +- `stream_dispatcher: client queue full, marking resync` — slow + consumer +- `stream_dispatcher: send failed for (N/3): ` — per-attempt +- `stream_dispatcher: evicting client 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. diff --git a/docs/planning/ORCHESTRATION_RELIABILITY_2026-04.md b/docs/planning/ORCHESTRATION_RELIABILITY_2026-04.md index c4d3c1abc..5dc4e86cf 100644 --- a/docs/planning/ORCHESTRATION_RELIABILITY_2026-04.md +++ b/docs/planning/ORCHESTRATION_RELIABILITY_2026-04.md @@ -1,9 +1,11 @@ # Orchestration & Multi-Agent Reliability Plan -**Date:** 2026-04-13 +**Date:** 2026-04-13 (revised 2026-04-20) **Status:** Proposed sequencing for execution-time orchestration, event subscriptions, and multi-agent reliability. -**Progress:** Sprint A — **7/7 complete**. Sprint B — **1/1 complete**. Sprint C — **3/5 complete**: #260 (PR #316), #271 (PR #332), #264 (PR #334). **Next: #294.** +**Progress:** Sprint A — **7/7 complete**. Sprint B — **1/1 complete**. Sprint C — **3/5 complete**: #260 (PR #316), #271 (PR #332), #264 (PR #334). **#294 and #291 paused pending #306.** **Next: #306 (push event bus).** + +**2026-04-20 revision:** After reviewing the accumulated orchestration surface (three queue abstractions, nine cleanup paths, twelve status-column writers, seven dispatch sites), the next priority shifted from finishing Sprint C to **push-based completion (#306) + consolidation** — see *Tier 2.5 — Simplification* below. The cleanup pyramid is load-bearing, so simplification is **additive-first**: new paths ship alongside old ones and the watchdog is retired only after push has soaked. --- @@ -24,9 +26,11 @@ Shipping #260 on top of today's foundation would produce a *persistent* backlog ``` Sprint A (unblock): #95 ✅, #285 ✅, #226 ✅, #286 ✅, #61 ✅, #132 ✅, #56 ✅ ← COMPLETE Sprint B (trace): #305 ✅ ← COMPLETE -Sprint C (orchestrate): #260 ✅ → #271 ✅ → #264 ✅ → #294 → #291 -Sprint D (push telemetry): #306, #307 -Sprint E (scale): #24, #18 +Sprint C (orchestrate): #260 ✅ → #271 ✅ → #264 ✅ → [#294 PAUSED] → [#291 PAUSED] +Sprint D (simplify): #306 → #428 (CAPACITY-CONSOLIDATE) → #429 (CLEANUP-COLLAPSE) → #430 (PROCESS-ENGINE-DECISION) + (and #408 dissolves once #306 lands) +Sprint E (telemetry): #307 +Sprint F (scale): #24, #18 ``` `#95` lands alone because every other Tier 0 fix layers on top of the unified executor. The remaining Tier 0 issues are independent and can parallelize once `#95` ships. @@ -97,9 +101,9 @@ Sprint E (scale): #24, #18 |---|-------|---------------| | ~~#260~~ ✅ | ~~Persistent task backlog (BACKLOG-001)~~ | **Shipped** in PR #316. SQLite-backed FIFO backlog with `status=queued`. Drain via `BacklogService.try_drain_one()` called on slot release. 24h stale expiry. Depth cap configurable per-agent. | | ~~#271~~ ✅ | ~~Retry mechanism for scheduled executions~~ | **Shipped** in PR #332. Configurable `max_retries` (0-5, default 1) and `retry_delay_seconds` (30-600, default 60). Rate-limited (429) failures use 2x delay. Retries persist to DB and survive scheduler restart via `_recover_pending_retries()`. New status: `pending_retry`. | -| #294 | Business task validation (VALIDATE-001) | Clean-context auditor session after execution. Reuses unified executor (#95). Writes `business_status` separate from technical `status`. | +| #294 ⏸️ | Business task validation (VALIDATE-001) **— PAUSED 2026-04-20** | Clean-context auditor session after execution. Re-examine after #306: a second full Claude session per task is a 2x cost feature that may be subsumable by cheaper primitives (output schemas, post-hoc validators) running in-process. | | ~~#264~~ ✅ | ~~Self-execute during chat (SELF-EXEC-001)~~ | **Shipped** in PR #334. Detects source==target, sets `X-Self-Task` header, optionally injects result back into chat session via `inject_result` parameter. Uses backlog for overflow when at capacity. | -| #291 | Agent webhook triggers (WEBHOOK-001) | External → agent dispatch. HMAC-signed URL. **Distinct from existing process-engine webhooks** (`routers/triggers.py`) which trigger BPMN process executions. Before building, decide: reuse the process-engine trigger surface (lower surface area) or ship a parallel agent-scoped trigger surface (clearer mental model, but exactly the parallel-paths problem this plan exists to fix). Default recommendation: reuse, with an `agent_task` shortcut process. | +| #291 ⏸️ | Agent webhook triggers (WEBHOOK-001) **— PAUSED 2026-04-20** | External → agent dispatch. The "reuse process-engine triggers vs. parallel surface" decision is easier after #430 (PROCESS-ENGINE-DECISION) (Tier 2.5) resolves whether the engine stays at all. Re-open after that. | ### Architectural shift @@ -130,21 +134,67 @@ Retry and validation are **not new infrastructure** — they're just new trigger --- -## Tier 3 — Push telemetry (Sprint D) +## Tier 2.5 — Simplification (Sprint D) — **NEXT** + +**Goal:** Collapse the three-queues / nine-cleanup-paths / twelve-status-writers pyramid that has accumulated across Sprints A–C. The pyramid exists because dispatch is HTTP-blocking and agent state is reconciled from three sources (Redis + DB + agent). Fix those two roots and most of the pyramid falls away. -**Goal:** Move the remaining polling loops to push, now that the executor and queue are stable. +**Premise from the 2026-04-20 review:** Each new Sprint C primitive welds itself into the pyramid; adding #294 / #291 on top first would make consolidation strictly harder. Reorder: simplify before extending. + +### Sequencing within Sprint D + +``` +#306 (push bus) ─► soak ≥2 weeks ─► #428 (CAPACITY-CONSOLIDATE) ─► #429 (CLEANUP-COLLAPSE) + └► #430 (PROCESS-ENGINE-DECISION) (parallel) +``` | # | Title | Why it's here | |---|-------|---------------| -| #306 | Redis Streams event bus for WebSocket (RELIABILITY-003) | Replaces in-process `ConnectionManager.broadcast()` (`main.py:125-130`, currently `except: pass`). `XADD`/`XREAD` with reconnect replay. Bigger surface than #260 itself — explicit `lastEventId` work on the frontend. | -| #307 | Agent heartbeat push (RELIABILITY-004) | Flip 30s polling (`monitoring_service.py:654`) → 5s push. Feeds monitoring + (future) circuit breaker. Uses existing Redis. | +| #306 | **Redis Streams event bus (RELIABILITY-003)** — keystone | Enables push-based completion from agents. Retires the 1h blocking HTTP call in `TaskExecutionService` (dissolves #408). Replaces in-process `ConnectionManager.broadcast()` (`main.py:125-130`, currently `except: pass`) with `XADD`/`XREAD` + reconnect replay. Explicit `lastEventId` work on the frontend. | +| **NEW** | **#428 (CAPACITY-CONSOLIDATE)** | Merge `ExecutionQueue` + `SlotService` + `BacklogService` into one `CapacityManager` with `(max_concurrent, overflow_policy)` config. `/chat` = `(1, queue_in_memory)`. `/task` = `(N, queue_persistent)`. Depends on #306 so the drain/TTL logic has the event consumer to lean on. | +| **NEW** | **#429 (CLEANUP-COLLAPSE)** | Once agent is authoritative for "is this running?" (via push), retire Phase 1/1b/1c/3 reconciliation. Slot TTL disappears — capacity is recomputed from DB, not TTL'd. Target: 9 paths → 1 periodic `DB ⟷ agent./api/running` sync. **Do not ship until #306 has been in prod ≥2 weeks with zero observed orphans.** | +| **NEW** | **#430 (PROCESS-ENGINE-DECISION)** | Today `process_engine/engine/handlers/agent_task.py` bypasses `TaskExecutionService` entirely (architecture.md marks engine as "dormant, out of scope"). Ship one of: (a) fold `agent_task` through TES, or (b) delete the engine. Sitting in the middle means every orchestration invariant has a silent exception. | + +### Architectural shift + +**Before (today):** Three queue primitives, nine cleanup paths, twelve status writers, seven dispatch sites, HTTP connection held up to 3610s. Each new trigger type (retry, validation, webhook, event sub, self-exec) adds its own reconciliation wrinkle. FAILED→SUCCESS races patched by Phase 3 re-verify. + +**After Sprint D:** One `CapacityManager`. Dispatch is a <5s HTTP 202; agent pushes completion via Redis Stream. Backend consumer writes the result once. One reconciliation loop (agent is source of truth). No TTL math. New trigger types add zero new cleanup paths. + +### Additive-first migration (regression mitigation) + +The watchdog pyramid is load-bearing *right now*. The migration must not trade known bugs for unknown ones: + +1. **#306 ships alongside** the existing HTTP path — both active, push is opt-in per agent initially. +2. **#428 (CAPACITY-CONSOLIDATE)** lands behind a feature flag per agent, or class-by-class, with old Queue/Slot/Backlog classes kept until all callers have moved. +3. **#429 (CLEANUP-COLLAPSE) is the riskiest and must not ship early.** Gate it on "#306 in prod ≥2 weeks, zero orphan observations." +4. Every PR must leave the system in a shippable state — no multi-PR in-between states where both old and new paths are partially wired. + +Worst case: new paths break and we fall back to existing paths. Old code gets deleted after proof, not before. + +### Verification gates before exiting Tier 2.5 + +- Push completion success rate ≥99.9% over 2 weeks (tracked via stream consumer metrics). +- Zero orphan recoveries from Phase 0 watchdog during soak period. +- Grep for direct `SlotService` / `ExecutionQueue` / `BacklogService` instantiation returns zero hits outside `CapacityManager` and its tests. +- Single writer per `schedule_executions.status` transition, verifiable by audit. +- #408 closeable as a dissolved symptom (no code change on #408 itself). + +--- + +## Tier 3 — Remaining push telemetry (Sprint E) + +**Goal:** Finish the polling-to-push migration that #306 started. + +| # | Title | Why it's here | +|---|-------|---------------| +| #307 | Agent heartbeat push (RELIABILITY-004) | Flip 30s polling (`monitoring_service.py:654`) → 5s push. Feeds monitoring + (future) circuit breaker. Uses the Redis stream established in #306. | ### Considerations - **Redis memory**: stream trim via `MAXLEN ~10000`. Without this, a burst of activity blows up Redis. - **Backward compat**: WebSocket event shape must not change. Frontend needs `lastEventId` support but old events should still render. -## Tier 4 — Scale (later) +## Tier 4 — Scale (Sprint F, later) | # | Title | |---|-------| @@ -212,6 +262,38 @@ APScheduler fire-and-forget, async status consumer WebSocket ◄── Redis Streams (XADD/XREAD) with replay ``` +### Aspirational (after Tier 2.5) + +``` +All entry paths ─► TaskExecutionService (true single funnel, process engine folded in or gone) + │ + ▼ + ┌─────────────────────────────────┐ + │ CapacityManager │ + │ (subsumes Queue + Slot + │ + │ Backlog; one class, one │ + │ TTL reasoner, one counter) │ + └──────────────┬──────────────────┘ + │ HTTP POST /api/task → 202 (short) + ▼ + AGENT CONTAINER (authoritative) + runs task, owns lifecycle + │ + │ XADD agent-events-stream + ▼ + ┌─────────────────────────────────┐ + │ Event Consumer (backend) │ + │ XREAD lastId → persist result │ + │ → release capacity → drain │ + │ → WebSocket fan-out │ + └─────────────────────────────────┘ + +Recovery: ONE periodic sync (DB ⟷ agent./api/running; agent wins). +Writers to schedule_executions.status: ~4 (create, start, finish, external-cancel). +No TTL math. No multi-phase cleanup pyramid. FAILED→SUCCESS race impossible +(single-writer event consumer owns the terminal transition). +``` + ### After Tier 2 (partial — #260, #271, #264 shipped) ``` @@ -250,5 +332,9 @@ New triggers, all funnel into the same executor: 5. ~~Confirm scope cuts for #260: FIFO-only v1, depth 50 default, 24h expiry.~~ ✅ Shipped with these cuts in PR #316. 6. ~~Rescope #132 against `src/scheduler/service.py`~~ — ✅ Shipped in PR #328. 7. ~~Re-estimate #56~~ — ✅ Shipped in PR #329. -8. Decide #291 direction: reuse process-engine triggers (recommended) vs. parallel agent-scoped trigger surface. -9. **Next:** Pick up #294 (validation). #264 shipped in PR #334. +8. ~~Decide #291 direction~~ — **Paused 2026-04-20 pending #430 (PROCESS-ENGINE-DECISION).** +9. ~~**Next:** Pick up #294 (validation).~~ — **Paused 2026-04-20 pending #306.** +10. **Next (2026-04-20):** Pick up **#306 (Redis Streams event bus)** — keystone for Tier 2.5 simplification. +11. **Follow-up:** Create and rank the three new issues from Tier 2.5: #428 (CAPACITY-CONSOLIDATE), #429 (CLEANUP-COLLAPSE), #430 (PROCESS-ENGINE-DECISION). +12. **After #306 lands:** 2-week soak period; instrument push success rate and orphan count; *then* schedule #428 (CAPACITY-CONSOLIDATE). +13. **Re-evaluate #408** once #306 is live — expected outcome: close as dissolved (no direct code change needed). diff --git a/src/backend/main.py b/src/backend/main.py index 58ae6b196..d43ab64eb 100644 --- a/src/backend/main.py +++ b/src/backend/main.py @@ -16,7 +16,7 @@ import json import os from datetime import datetime -from typing import List +from typing import Dict, List, Optional from contextlib import asynccontextmanager from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Depends, Request, Query @@ -109,89 +109,91 @@ logger = logging.getLogger(__name__) -class ConnectionManager: - """WebSocket connection manager for broadcasting events.""" +# Redis Streams event bus replaces the old in-process broadcast. Legacy manager +# classes are kept as thin shims so the 33 existing broadcast call sites don't +# change. See docs/memory/feature-flows/websocket-event-bus.md and +# services/event_bus.py (RELIABILITY-003 / #306). +from services.event_bus import ( + event_bus, + stream_dispatcher, + SCOPE_ALL, + SCOPE_SCOPED, +) - def __init__(self): - self.active_connections: List[WebSocket] = [] - async def connect(self, websocket: WebSocket): - await websocket.accept() - self.active_connections.append(websocket) +class ConnectionManager: + """Thin shim over ``event_bus``: preserves the legacy broadcast-a-JSON-string API. - def disconnect(self, websocket: WebSocket): - if websocket in self.active_connections: - self.active_connections.remove(websocket) + Connections themselves are tracked by ``StreamDispatcher``; ``connect()`` + here returns a client id so callers can ``disconnect()`` without juggling + the dispatcher directly.""" - async def broadcast(self, message: str): - for connection in self.active_connections: - try: - await connection.send_text(message) - except: - pass + def __init__(self) -> None: + self._client_ids: Dict[WebSocket, str] = {} + async def connect(self, websocket: WebSocket, last_event_id: Optional[str] = None) -> None: + await websocket.accept() + async def _send(payload: dict) -> None: + await websocket.send_text(json.dumps(payload)) + client_id = await stream_dispatcher.register( + websocket, + scope=SCOPE_ALL, + send_func=_send, + last_event_id=last_event_id, + ) + self._client_ids[websocket] = client_id -class FilteredWebSocketManager: - """ - WebSocket manager that filters events based on user's accessible agents. + def disconnect(self, websocket: WebSocket) -> None: + client_id = self._client_ids.pop(websocket, None) + if client_id: + stream_dispatcher.unregister(client_id) - Used by /ws/events endpoint for external listeners (Trinity Connect). - Events are filtered server-side based on user's owned and shared agents. - """ + async def broadcast(self, message) -> None: + """Publish an event for all /ws consumers. - def __init__(self): - from typing import Dict, Set - self.connections: Dict[WebSocket, Dict] = {} # ws -> {email, is_admin, accessible_agents} + Accepts either a JSON-encoded string (legacy signature) or a dict + (preferred going forward).""" + await event_bus.publish(message, scope=SCOPE_ALL) - async def connect(self, websocket: WebSocket, email: str, is_admin: bool, accessible_agents: List[str]): - """Register a new connection with its accessible agents.""" - self.connections[websocket] = { - "email": email, - "is_admin": is_admin, - "accessible_agents": set(accessible_agents) - } - def disconnect(self, websocket: WebSocket): - """Remove a connection.""" - self.connections.pop(websocket, None) - - def update_accessible_agents(self, websocket: WebSocket, accessible_agents: List[str]): - """Update the accessible agents list for a connection.""" - if websocket in self.connections: - self.connections[websocket]["accessible_agents"] = set(accessible_agents) - - async def broadcast_filtered(self, event: dict): - """ - Broadcast event only to users who can access the event's agent. - - Extracts agent name from various event fields and checks if - each connected user can access that agent. - """ - # Extract agent name from event (different fields for different event types) - agent_name = ( - event.get("agent_name") or - event.get("agent") or - event.get("name") or # agent_started/agent_stopped events - event.get("source_agent") or - (event.get("details") or {}).get("source_agent") or - (event.get("details") or {}).get("target_agent") +class FilteredWebSocketManager: + """Thin shim over ``event_bus`` for /ws/events (Trinity Connect).""" + + def __init__(self) -> None: + self._client_ids: Dict[WebSocket, str] = {} + + async def connect( + self, + websocket: WebSocket, + email: str, + is_admin: bool, + accessible_agents: List[str], + last_event_id: Optional[str] = None, + ) -> None: + async def _send(payload: dict) -> None: + await websocket.send_json(payload) + client_id = await stream_dispatcher.register( + websocket, + scope=SCOPE_SCOPED, + send_func=_send, + is_admin=is_admin, + accessible_agents=accessible_agents, + last_event_id=last_event_id, ) + self._client_ids[websocket] = client_id - if not agent_name: - return # Can't filter without agent name + def disconnect(self, websocket: WebSocket) -> None: + client_id = self._client_ids.pop(websocket, None) + if client_id: + stream_dispatcher.unregister(client_id) - disconnected = [] - for websocket, info in self.connections.items(): - # Admin sees all, otherwise check accessible agents - if info["is_admin"] or agent_name in info["accessible_agents"]: - try: - await websocket.send_json(event) - except Exception: - disconnected.append(websocket) + def update_accessible_agents(self, websocket: WebSocket, accessible_agents: List[str]) -> None: + client_id = self._client_ids.get(websocket) + if client_id: + stream_dispatcher.update_accessible_agents(client_id, accessible_agents) - # Clean up disconnected clients - for ws in disconnected: - self.disconnect(ws) + async def broadcast_filtered(self, event: dict) -> None: + await event_bus.publish(event, scope=SCOPE_SCOPED) manager = ConnectionManager() @@ -280,6 +282,17 @@ async def lifespan(app: FastAPI): # Set up structured JSON logging (captured by Vector) setup_logging() + # Start Redis Streams event bus + dispatcher (RELIABILITY-003 / #306). + # Must start before the WebSocket endpoints begin accepting clients so the + # first connection has a live dispatcher to register with. + try: + await event_bus.start() + await stream_dispatcher.start() + logger.info("Redis Streams event bus started (maxlen=%d)", + int(os.getenv("REDIS_STREAM_MAXLEN", "10000"))) + except Exception as e: + logger.error(f"Event bus startup failed (broadcasts will degrade): {e}") + # Report OpenTelemetry status (RELIABILITY-002) if _otel_enabled: sample_rate = float(os.getenv("OTEL_SAMPLE_RATE", "0.1")) @@ -523,6 +536,16 @@ async def _backlog_maintenance_loop(): except Exception as e: print(f"Error closing agent HTTP client pool: {e}") + # Drain event bus + stop dispatcher last so late-lifecycle broadcasts + # (e.g. "agent_stopped" emitted during service shutdown) still land on + # the stream. 2s drain window per #306. + try: + await stream_dispatcher.stop() + await event_bus.stop(drain_timeout=2.0) + print("Event bus and stream dispatcher stopped") + except Exception as e: + print(f"Error stopping event bus/dispatcher: {e}") + # Create FastAPI app app = FastAPI( @@ -610,7 +633,11 @@ async def add_security_headers(request: Request, call_next): # WebSocket endpoint @app.websocket("/ws") -async def websocket_endpoint(websocket: WebSocket, token: str = Query(default=None)): +async def websocket_endpoint( + websocket: WebSocket, + token: str = Query(default=None), + last_event_id: Optional[str] = Query(default=None, alias="last-event-id"), +): """ WebSocket endpoint for real-time updates. @@ -620,9 +647,15 @@ async def websocket_endpoint(websocket: WebSocket, token: str = Query(default=No Connections without a valid token are rejected before websocket.accept() to prevent any unauthenticated data leakage. + + Reconnect replay (#306): clients may pass ``last-event-id=`` + to receive events missed during a disconnect. Malformed or too-old ids + produce a ``{"type": "resync_required"}`` message — the client must then + fetch current state via REST. """ from jose import JWTError, jwt as jose_jwt from config import SECRET_KEY, ALGORITHM + from services.event_bus import validate_last_event_id # Reject immediately if no token provided — before accept() if not token: @@ -641,7 +674,7 @@ async def websocket_endpoint(websocket: WebSocket, token: str = Query(default=No return # Token validated — now accept the connection - await manager.connect(websocket) + await manager.connect(websocket, last_event_id=validate_last_event_id(last_event_id)) try: while True: @@ -663,7 +696,8 @@ async def websocket_endpoint(websocket: WebSocket, token: str = Query(default=No @app.websocket("/ws/events") async def websocket_events_endpoint( websocket: WebSocket, - token: str = Query(None, description="MCP API key for authentication") + token: str = Query(None, description="MCP API key for authentication"), + last_event_id: Optional[str] = Query(None, alias="last-event-id"), ): """ WebSocket endpoint for external event listeners (Trinity Connect). @@ -686,6 +720,7 @@ async def websocket_events_endpoint( - "refresh" -> refreshes accessible agents list """ from database import db + from services.event_bus import validate_last_event_id # Validate MCP API key if not token or not token.startswith("trinity_mcp_"): @@ -713,8 +748,11 @@ async def websocket_events_endpoint( "message": "Listening for events. Events filtered to your accessible agents." }) - # Add to filtered connections manager - await filtered_manager.connect(websocket, user_email, is_admin, accessible_agents) + # Add to filtered connections manager — enables reconnect replay via #306. + await filtered_manager.connect( + websocket, user_email, is_admin, accessible_agents, + last_event_id=validate_last_event_id(last_event_id), + ) try: while True: diff --git a/src/backend/services/event_bus.py b/src/backend/services/event_bus.py new file mode 100644 index 000000000..8bf3e7f0e --- /dev/null +++ b/src/backend/services/event_bus.py @@ -0,0 +1,603 @@ +""" +Redis Streams event bus for WebSocket delivery (RELIABILITY-003 / #306). + +Replaces the in-process ``ConnectionManager.broadcast()`` + ``except: pass`` pattern +with a durable event log. Provides reconnect replay via per-client ``last-event-id`` +and graceful degradation when Redis is unavailable. + +Design +------ +* Producers call ``event_bus.publish(event, scope=...)`` which XADDs to + ``trinity:events`` with an approximate MAXLEN trim. +* A single ``StreamDispatcher`` coroutine per backend process reads the stream + with ``XREAD BLOCK`` and fans out in-memory to registered WebSocket clients. +* Each client has a bounded ``asyncio.Queue(256)``. If the queue is full the + dispatcher drops and schedules a ``resync_required`` message — slow clients + never block others. +* After 3 consecutive send failures a client is evicted and its socket is closed. +* Reconnect flow: client supplies ``?last-event-id=``; the dispatcher runs + a one-shot ``XRANGE`` catchup, then joins the live fan-out. If the requested + id is older than the stream's earliest entry (trimmed) the client receives + ``{"type": "resync_required"}`` and must fetch current state via REST. + +Scope discipline (#306) +----------------------- +This module is the WebSocket delivery layer. Agent-push completion, heartbeat +push, and capacity consolidation (#307 / #428 / #429) will reuse the same +stream primitive in later sprints — see +``docs/planning/ORCHESTRATION_RELIABILITY_2026-04.md``. +""" +from __future__ import annotations + +import asyncio +import json +import logging +import os +import re +import time +import uuid +from dataclasses import dataclass, field +from typing import Any, Callable, Dict, List, Optional, Set + +try: + import redis.asyncio as aioredis + from redis.exceptions import ResponseError as RedisResponseError +except Exception: # pragma: no cover — redis is a hard dependency + aioredis = None + RedisResponseError = Exception + +from config import REDIS_URL + +logger = logging.getLogger(__name__) + +STREAM_KEY = "trinity:events" +STREAM_MAXLEN = int(os.getenv("REDIS_STREAM_MAXLEN", "10000")) +CLIENT_QUEUE_MAXSIZE = 256 +EVICT_AFTER_FAILURES = 3 +REPLAY_GAP_LIMIT = 5000 # reject replays larger than this; force resync +EID_PATTERN = re.compile(r"^\d+-\d+$") + +SCOPE_ALL = "all" +SCOPE_SCOPED = "scoped" + +# In-memory fallback used when Redis is unavailable at publish time. +# Capped to avoid unbounded growth in degraded mode. +_FALLBACK_BUFFER_MAX = 1024 + + +def _serialize(event: Any) -> Dict[str, str]: + """Convert an event (dict or already-json-encoded str) into the flat + string map Redis XADD expects.""" + if isinstance(event, str): + try: + parsed = json.loads(event) + except (json.JSONDecodeError, TypeError): + parsed = {"raw": event} + return {"data": json.dumps(parsed)} + return {"data": json.dumps(event)} + + +def _deserialize(fields: Dict[str, str]) -> Dict[str, Any]: + raw = fields.get("data") if isinstance(fields, dict) else None + if not raw: + return {} + try: + return json.loads(raw) + except (json.JSONDecodeError, TypeError): + return {"raw": raw} + + +@dataclass +class _ClientSlot: + """Per-WebSocket state held by the dispatcher.""" + + ws: Any + scope: str # SCOPE_ALL or SCOPE_SCOPED + send_func: Callable # async fn(dict) that serializes appropriately + is_admin: bool = False + accessible_agents: Set[str] = field(default_factory=set) + queue: asyncio.Queue = field(default_factory=lambda: asyncio.Queue(maxsize=CLIENT_QUEUE_MAXSIZE)) + last_delivered_id: str = "0-0" + failure_count: int = 0 + consumer_task: Optional[asyncio.Task] = None + resync_pending: bool = False + + +def _event_is_visible(slot: _ClientSlot, event_scope: str, agent_name: Optional[str]) -> bool: + """Apply the scope/filter contract that replaces FilteredWebSocketManager. + + ``scope=SCOPE_ALL`` events reach every ``/ws`` client. ``scope=SCOPE_SCOPED`` + events reach admin /ws/events listeners and non-admins with access to the + named agent. Matches the legacy ``broadcast_filtered`` semantics.""" + if slot.scope == SCOPE_ALL: + return event_scope == SCOPE_ALL + # slot.scope == SCOPE_SCOPED + if event_scope != SCOPE_SCOPED: + return False + if slot.is_admin: + return True + if not agent_name: + return False + return agent_name in slot.accessible_agents + + +class EventBus: + """Publisher side of the stream. + + One instance per backend process. ``publish`` is fire-and-forget: it enqueues + to an internal ``asyncio.Queue`` and a background writer drains to Redis, so + callers never block on Redis latency and a Redis flap doesn't stall broadcast + sites like chat/activity.""" + + def __init__(self) -> None: + self._outbound: asyncio.Queue = asyncio.Queue(maxsize=10_000) + self._fallback: List[Dict[str, Any]] = [] + self._writer_task: Optional[asyncio.Task] = None + self._redis: Optional["aioredis.Redis"] = None + self._ready = False + + async def start(self) -> None: + if self._writer_task is not None: + return + self._writer_task = asyncio.create_task(self._writer_loop(), name="event_bus_writer") + + async def stop(self, drain_timeout: float = 2.0) -> None: + """Drain pending publishes, then close Redis. Called from lifespan shutdown.""" + if self._writer_task: + try: + await asyncio.wait_for(self._outbound.join(), timeout=drain_timeout) + except asyncio.TimeoutError: + logger.warning("event_bus: drain timeout after %.1fs; %d events may be lost", + drain_timeout, self._outbound.qsize()) + self._writer_task.cancel() + try: + await self._writer_task + except (asyncio.CancelledError, Exception): + pass + self._writer_task = None + if self._redis is not None: + try: + await self._redis.aclose() + except Exception: + pass + self._redis = None + + async def publish( + self, + event: Any, + scope: str = SCOPE_ALL, + agent_name: Optional[str] = None, + ) -> None: + """Publish an event. Non-blocking; drops oldest on overflow. + + ``event`` may be a dict or a JSON-encoded string (for the legacy + ``ConnectionManager.broadcast(str)`` call sites). ``scope`` and + ``agent_name`` are stored alongside the event payload so consumers can + filter without inspecting the payload.""" + if isinstance(event, str): + try: + payload = json.loads(event) + except (json.JSONDecodeError, TypeError): + payload = {"raw": event} + else: + payload = event + + if not isinstance(payload, dict): + payload = {"value": payload} + + # Infer agent_name from payload if not provided — matches the legacy + # FilteredWebSocketManager heuristic. + if scope == SCOPE_SCOPED and agent_name is None: + details = payload.get("details") if isinstance(payload.get("details"), dict) else {} + agent_name = ( + payload.get("agent_name") + or payload.get("agent") + or payload.get("name") + or payload.get("source_agent") + or details.get("source_agent") + or details.get("target_agent") + ) + + envelope = {"payload": payload, "scope": scope, "agent_name": agent_name or ""} + + try: + self._outbound.put_nowait(envelope) + except asyncio.QueueFull: + # Evict oldest to make room (favour liveness over completeness under storm). + try: + _ = self._outbound.get_nowait() + self._outbound.task_done() + except asyncio.QueueEmpty: + pass + try: + self._outbound.put_nowait(envelope) + except asyncio.QueueFull: + logger.warning("event_bus: outbound queue full, dropping event") + + async def _writer_loop(self) -> None: + backoff = 1.0 + while True: + try: + envelope = await self._outbound.get() + try: + await self._xadd(envelope) + backoff = 1.0 # reset on success + finally: + self._outbound.task_done() + except asyncio.CancelledError: + raise + except Exception as e: + logger.warning("event_bus: writer error: %s; backoff=%.1fs", e, backoff) + await asyncio.sleep(min(backoff, 30.0)) + backoff = min(backoff * 2.0, 30.0) + + async def _get_redis(self) -> Optional["aioredis.Redis"]: + if self._redis is not None: + return self._redis + if aioredis is None: + return None + try: + self._redis = aioredis.from_url(REDIS_URL, decode_responses=True) + await self._redis.ping() + self._ready = True + return self._redis + except Exception as e: + logger.warning("event_bus: Redis unavailable (%s); publish will degrade", e) + self._redis = None + return None + + async def _xadd(self, envelope: Dict[str, Any]) -> None: + redis = await self._get_redis() + if redis is None: + # Keep a small rolling buffer so very early events survive a brief + # Redis outage; drop silently when saturated. + if len(self._fallback) < _FALLBACK_BUFFER_MAX: + self._fallback.append(envelope) + return + + fields = { + "payload": json.dumps(envelope["payload"]), + "scope": envelope["scope"], + "agent_name": envelope["agent_name"] or "", + } + try: + await redis.xadd(STREAM_KEY, fields, maxlen=STREAM_MAXLEN, approximate=True) + except Exception as e: + # Force reconnect on next call. + logger.warning("event_bus: XADD failed: %s", e) + try: + await redis.aclose() + except Exception: + pass + self._redis = None + raise + + +class StreamDispatcher: + """Consumer side of the stream. + + Maintains a ``clients`` map and runs a single ``XREAD BLOCK`` loop per + backend process. Events are put_nowait'd into each client's bounded queue; + each client has a consumer task that dequeues and sends. + """ + + def __init__(self) -> None: + self._clients: Dict[str, _ClientSlot] = {} + self._reader_task: Optional[asyncio.Task] = None + self._redis: Optional["aioredis.Redis"] = None + self._last_stream_id: str = "$" # start at live tip on boot + self._lock = asyncio.Lock() + self._shutting_down = False + + async def start(self) -> None: + if self._reader_task is not None: + return + self._reader_task = asyncio.create_task(self._supervised_reader(), name="stream_dispatcher") + + async def stop(self) -> None: + self._shutting_down = True + for slot in list(self._clients.values()): + if slot.consumer_task: + slot.consumer_task.cancel() + if self._reader_task: + self._reader_task.cancel() + try: + await self._reader_task + except (asyncio.CancelledError, Exception): + pass + self._reader_task = None + if self._redis is not None: + try: + await self._redis.aclose() + except Exception: + pass + self._redis = None + + async def register( + self, + ws: Any, + scope: str, + send_func: Callable, + *, + is_admin: bool = False, + accessible_agents: Optional[List[str]] = None, + last_event_id: Optional[str] = None, + ) -> str: + """Register a WebSocket client. Returns the client_id used for lookups. + + If ``last_event_id`` is provided and valid, a catch-up XRANGE is queued + before the live fan-out begins. On failure (malformed id, huge gap, or + trim race) a ``resync_required`` event is queued and the client resumes + from the live tip.""" + slot = _ClientSlot( + ws=ws, + scope=scope, + send_func=send_func, + is_admin=is_admin, + accessible_agents=set(accessible_agents or []), + ) + client_id = str(uuid.uuid4()) + slot.consumer_task = asyncio.create_task( + self._client_consumer(client_id, slot), name=f"ws_consumer_{client_id[:8]}" + ) + + # Snapshot the reader's position BEFORE adding the client to _clients + # so catchup's upper bound can't overlap with live fan-out. Without this + # snapshot, ``_catchup`` calling ``XRANGE max="+"`` concurrently with + # ``_fanout`` could double-deliver events (see #306 review C1). + catchup_max = self._last_stream_id if self._last_stream_id != "$" else None + + self._clients[client_id] = slot + + if last_event_id: + asyncio.create_task( + self._catchup(client_id, slot, last_event_id, catchup_max) + ) + + return client_id + + def unregister(self, client_id: str) -> None: + slot = self._clients.pop(client_id, None) + if slot and slot.consumer_task: + slot.consumer_task.cancel() + + def update_accessible_agents(self, client_id: str, accessible_agents: List[str]) -> None: + slot = self._clients.get(client_id) + if slot: + slot.accessible_agents = set(accessible_agents) + + def client_count(self) -> int: + return len(self._clients) + + # ------------------------------------------------------------------ reader + + async def _supervised_reader(self) -> None: + """Restart the XREAD loop with exponential backoff on unexpected errors.""" + backoff = 1.0 + while not self._shutting_down: + try: + await self._reader_loop() + # Normal exit shouldn't happen; treat as error. + backoff = min(backoff * 2.0, 30.0) + except asyncio.CancelledError: + raise + except Exception as e: + logger.error("stream_dispatcher: reader crashed: %s; restart in %.1fs", e, backoff) + await asyncio.sleep(backoff) + backoff = min(backoff * 2.0, 30.0) + + async def _reader_loop(self) -> None: + redis = await self._get_redis() + if redis is None: + await asyncio.sleep(5.0) + return + + while not self._shutting_down: + try: + response = await redis.xread( + {STREAM_KEY: self._last_stream_id}, block=5000, count=100 + ) + except asyncio.CancelledError: + raise + except Exception as e: + logger.warning("stream_dispatcher: xread error: %s", e) + try: + await redis.aclose() + except Exception: + pass + self._redis = None + raise # let supervisor back off + + if not response: + continue + + for _stream_name, entries in response: + for entry_id, fields in entries: + self._last_stream_id = entry_id + await self._fanout(entry_id, fields) + + async def _fanout(self, entry_id: str, fields: Dict[str, str]) -> None: + payload = _deserialize({"data": fields.get("payload", "")}) + scope = fields.get("scope", SCOPE_ALL) + agent_name = fields.get("agent_name") or None + + # Inject stream id into the payload so frontend reconnect logic can + # persist it. Additive — existing handlers ignore unknown fields. + payload = dict(payload) + payload["_eid"] = entry_id + + for client_id, slot in list(self._clients.items()): + if not _event_is_visible(slot, scope, agent_name): + continue + try: + slot.queue.put_nowait((entry_id, payload)) + except asyncio.QueueFull: + # Slow client — drop and require resync. + if not slot.resync_pending: + slot.resync_pending = True + logger.warning("stream_dispatcher: client %s queue full, marking resync", + client_id[:8]) + try: + slot.queue.put_nowait((entry_id, {"type": "resync_required", + "reason": "slow_consumer", + "_eid": entry_id})) + except asyncio.QueueFull: + pass # will be evicted by consumer on next failure + + # ------------------------------------------------------------------ client + + async def _client_consumer(self, client_id: str, slot: _ClientSlot) -> None: + """Drain this client's queue, handle send failures and eviction.""" + try: + while True: + entry_id, payload = await slot.queue.get() + try: + await slot.send_func(payload) + slot.failure_count = 0 + # Monotonic guard (#306 review C1): even though catchup's + # range is capped to avoid overlap with live fan-out, keep + # this defensive check so any future ordering bug can't + # regress the client's cursor. Never advances backwards. + if payload.get("type") != "resync_required" and _id_greater_than( + entry_id, slot.last_delivered_id + ): + slot.last_delivered_id = entry_id + if payload.get("type") == "resync_required": + slot.resync_pending = False + except asyncio.CancelledError: + raise + except Exception as e: + slot.failure_count += 1 + logger.info( + "stream_dispatcher: send failed for %s (%d/%d): %s", + client_id[:8], slot.failure_count, EVICT_AFTER_FAILURES, e, + ) + if slot.failure_count >= EVICT_AFTER_FAILURES: + logger.warning("stream_dispatcher: evicting client %s after %d failures", + client_id[:8], EVICT_AFTER_FAILURES) + try: + close = getattr(slot.ws, "close", None) + if close: + await close(code=1011, reason="broadcast failure eviction") + except Exception: + pass + self._clients.pop(client_id, None) + return + except asyncio.CancelledError: + pass + + async def _catchup( + self, + client_id: str, + slot: _ClientSlot, + last_event_id: str, + catchup_max: Optional[str] = None, + ) -> None: + """One-shot replay from (last_event_id, catchup_max]. + + ``catchup_max`` is the dispatcher's ``_last_stream_id`` at the moment + the client was registered. Capping XRANGE at this point prevents + overlap with live fan-out (see #306 review C1). Falls back to ``"+"`` + only when the reader hasn't read any events yet — in that case fan-out + itself has delivered nothing, so no overlap is possible.""" + if not EID_PATTERN.match(last_event_id): + await self._queue_resync(slot, "invalid_last_event_id") + return + + redis = await self._get_redis() + if redis is None: + await self._queue_resync(slot, "redis_unavailable") + return + + # Exclusive-start form ``(`` avoids re-delivering last_event_id. + # See https://redis.io/commands/xrange/ + start = f"({last_event_id}" + end = catchup_max if catchup_max else "+" + try: + entries = await redis.xrange(STREAM_KEY, min=start, max=end, count=REPLAY_GAP_LIMIT + 1) + except Exception as e: + logger.warning("stream_dispatcher: xrange failed: %s", e) + await self._queue_resync(slot, "replay_error") + return + + if len(entries) > REPLAY_GAP_LIMIT: + await self._queue_resync(slot, "gap_too_large") + return + + # Check for trim race: if the stream's earliest id is > last_event_id the + # cursor is stale and events were trimmed. + if entries: + # entries sorted ascending; the first id > last_event_id means some + # entries between last_event_id and the first replayed one may have + # been trimmed. Detect by comparing against the stream's oldest id. + try: + oldest = await redis.xrange(STREAM_KEY, min="-", max="+", count=1) + if oldest: + oldest_id = oldest[0][0] + # last_event_id must be >= the oldest-minus-one; if the oldest + # id is numerically greater than last_event_id, history has + # been trimmed past the cursor. + if _id_greater_than(oldest_id, last_event_id): + await self._queue_resync(slot, "trimmed") + return + except Exception: + pass + + for entry_id, fields in entries: + payload = _deserialize({"data": fields.get("payload", "")}) + scope = fields.get("scope", SCOPE_ALL) + agent_name = fields.get("agent_name") or None + if not _event_is_visible(slot, scope, agent_name): + continue + payload = dict(payload) + payload["_eid"] = entry_id + try: + slot.queue.put_nowait((entry_id, payload)) + except asyncio.QueueFull: + await self._queue_resync(slot, "queue_overflow_during_catchup") + return + + async def _queue_resync(self, slot: _ClientSlot, reason: str) -> None: + slot.resync_pending = True + try: + slot.queue.put_nowait(("0-0", {"type": "resync_required", "reason": reason})) + except asyncio.QueueFull: + pass + + async def _get_redis(self) -> Optional["aioredis.Redis"]: + if self._redis is not None: + return self._redis + if aioredis is None: + return None + try: + self._redis = aioredis.from_url(REDIS_URL, decode_responses=True) + await self._redis.ping() + return self._redis + except Exception as e: + logger.warning("stream_dispatcher: Redis unavailable (%s)", e) + self._redis = None + return None + + +def _id_greater_than(a: str, b: str) -> bool: + """Compare two Redis stream ids in ``-`` form.""" + try: + a_ms, a_seq = (int(x) for x in a.split("-")) + b_ms, b_seq = (int(x) for x in b.split("-")) + except (ValueError, AttributeError): + return False + return (a_ms, a_seq) > (b_ms, b_seq) + + +def validate_last_event_id(raw: Optional[str]) -> Optional[str]: + """Return ``raw`` if it is a well-formed Redis stream id, else ``None``.""" + if not raw: + return None + if not EID_PATTERN.match(raw): + return None + return raw + + +# Module-level singletons used by main.py and 33 legacy broadcast call sites. +event_bus = EventBus() +stream_dispatcher = StreamDispatcher() diff --git a/src/frontend/src/stores/network.js b/src/frontend/src/stores/network.js index e337a9c73..e8f984b95 100644 --- a/src/frontend/src/stores/network.js +++ b/src/frontend/src/stores/network.js @@ -33,6 +33,9 @@ export const useNetworkStore = defineStore('network', () => { const websocketHeartbeatInterval = ref(null) // Interval ID for WebSocket keepalive ping const runningToggleLoading = ref({}) // Map of agent name -> boolean (loading state for start/stop) const schedules = ref([]) // Enabled schedules for timeline markers + // #306 Redis Streams reconnect replay: remember the last event id we saw so + // a network blip replays missed events instead of dropping them. + const lastEventId = ref(null) // View mode state (graph vs timeline) - default to timeline, persist to localStorage const savedViewMode = localStorage.getItem('trinity-dashboard-view') @@ -542,7 +545,10 @@ export const useNetworkStore = defineStore('network', () => { return } - const wsUrl = `${window.location.protocol === 'https:' ? 'wss:' : 'ws:'}//${window.location.host}/ws?token=${encodeURIComponent(token)}` + let wsUrl = `${window.location.protocol === 'https:' ? 'wss:' : 'ws:'}//${window.location.host}/ws?token=${encodeURIComponent(token)}` + if (lastEventId.value) { + wsUrl += `&last-event-id=${encodeURIComponent(lastEventId.value)}` + } // Prevent duplicate connections if (websocket.value?.readyState === WebSocket.OPEN) { @@ -564,7 +570,18 @@ export const useNetworkStore = defineStore('network', () => { websocket.value.onmessage = (event) => { try { const data = JSON.parse(event.data) + if (data._eid) { + lastEventId.value = data._eid + } + if (data.type === 'resync_required') { + // #306: stream cursor was trimmed (or reject), reload authoritative state. + console.warn('[Collaboration] resync_required:', data.reason) + lastEventId.value = null + try { fetchAgents() } catch (_) {} + try { fetchHistoricalCollaborations() } catch (_) {} + return + } if (data.type === 'agent_collaboration') { handleCollaborationEvent(data) } else if (data.type === 'agent_status') { diff --git a/src/frontend/src/utils/websocket.js b/src/frontend/src/utils/websocket.js index dbe6468d3..fdfb55c05 100644 --- a/src/frontend/src/utils/websocket.js +++ b/src/frontend/src/utils/websocket.js @@ -5,6 +5,9 @@ import { useOperatorQueueStore } from '../stores/operatorQueue' const ws = ref(null) const isConnected = ref(false) +// #306 Redis Streams reconnect replay: track the last Redis stream id we saw +// so a brief disconnect replays missed events rather than dropping them. +let lastEventId = null export function useWebSocket() { const agentsStore = useAgentsStore() @@ -20,7 +23,10 @@ export function useWebSocket() { return } - const wsUrl = `${window.location.protocol === 'https:' ? 'wss:' : 'ws:'}//${window.location.host}/ws?token=${encodeURIComponent(token)}` + let wsUrl = `${window.location.protocol === 'https:' ? 'wss:' : 'ws:'}//${window.location.host}/ws?token=${encodeURIComponent(token)}` + if (lastEventId) { + wsUrl += `&last-event-id=${encodeURIComponent(lastEventId)}` + } ws.value = new WebSocket(wsUrl) ws.value.onopen = () => { @@ -31,6 +37,9 @@ export function useWebSocket() { ws.value.onmessage = (event) => { try { const data = JSON.parse(event.data) + if (data._eid) { + lastEventId = data._eid + } handleMessage(data) } catch (error) { console.error('Failed to parse WebSocket message:', error) @@ -64,6 +73,16 @@ export function useWebSocket() { } const handleMessage = (data) => { + // #306 reconnect replay: server signals "your last-event-id was trimmed, + // full-refetch to rehydrate." Clear the cursor so the next reconnect + // starts live and refetch anything event-driven. + if (data.type === 'resync_required') { + console.warn('[WebSocket] resync_required:', data.reason) + lastEventId = null + try { agentsStore.fetchAgents && agentsStore.fetchAgents() } catch (_) {} + try { notificationsStore.fetchPendingCount && notificationsStore.fetchPendingCount() } catch (_) {} + return + } switch (data.event) { case 'agent_created': // Add to list (createAgent() no longer pushes to avoid race conditions) diff --git a/tests/registry.json b/tests/registry.json index c9e7ef7ce..b67be9c94 100644 --- a/tests/registry.json +++ b/tests/registry.json @@ -173,6 +173,13 @@ "added": "2026-04-19", "categories": ["backend", "unit", "telegram", "access-control"], "description": "Unit tests for Telegram /login access gate: verified+shared returns ready, open_access returns ready, verified+restricted upserts access_request and returns 'pending approval', upsert failures don't break the reply, invalid code short-circuits gate" + }, + { + "file": "test_event_bus.py", + "feature": "RELIABILITY-003 / #306", + "added": "2026-04-21", + "categories": ["backend", "unit", "websocket", "redis-streams"], + "description": "Unit tests for Redis Streams event bus (#306): last-event-id validation + id comparison, scope visibility (SCOPE_ALL vs SCOPE_SCOPED with accessible_agents), EventBus XADD envelope (dict + legacy JSON string + inferred agent_name), Redis-unavailable fallback buffer, StreamDispatcher 3-failure client eviction, slow-consumer queue overflow triggers resync marker, update_accessible_agents mutation, invalid last-event-id queues resync_required" } ] } diff --git a/tests/test_event_bus.py b/tests/test_event_bus.py new file mode 100644 index 000000000..0875d8094 --- /dev/null +++ b/tests/test_event_bus.py @@ -0,0 +1,439 @@ +""" +Unit tests for Redis Streams event bus (#306 / RELIABILITY-003). +Related flow: docs/memory/feature-flows/websocket-event-bus.md + +Covers: +- Envelope shape (payload/scope/agent_name fields) +- last-event-id validation + id comparison +- Scope visibility rules (SCOPE_ALL vs SCOPE_SCOPED with accessible_agents) +- Serialize/deserialize round-trip +- Graceful degradation when Redis is unavailable +- EventBus outbound-queue overflow drops oldest +- StreamDispatcher client queue + failure eviction +- Fallback buffer cap + +These tests don't require the backend to be running — they import +``services.event_bus`` directly and exercise its logic. +""" +from __future__ import annotations + +import asyncio +import importlib.util +import os +import sys + +import pytest + +# ``services/__init__.py`` auto-imports docker_service and the rest of the +# backend graph, so we load ``event_bus.py`` directly via a spec — isolates the +# unit under test from Docker / database setup. Minimal shim: only config.py is +# also needed, and it's stdlib-only. + +_BACKEND_PATH = os.path.abspath( + os.path.join(os.path.dirname(__file__), "..", "src", "backend") +) +if _BACKEND_PATH not in sys.path: + sys.path.insert(0, _BACKEND_PATH) + + +def _load_module(name, relpath): + spec = importlib.util.spec_from_file_location( + name, os.path.join(_BACKEND_PATH, relpath) + ) + module = importlib.util.module_from_spec(spec) + sys.modules[name] = module + spec.loader.exec_module(module) + return module + + +# Load config first (event_bus imports REDIS_URL from it). +_load_module("config", "config.py") +event_bus_mod = _load_module("services.event_bus", "services/event_bus.py") + +CLIENT_QUEUE_MAXSIZE = event_bus_mod.CLIENT_QUEUE_MAXSIZE +EVICT_AFTER_FAILURES = event_bus_mod.EVICT_AFTER_FAILURES +EventBus = event_bus_mod.EventBus +SCOPE_ALL = event_bus_mod.SCOPE_ALL +SCOPE_SCOPED = event_bus_mod.SCOPE_SCOPED +STREAM_KEY = event_bus_mod.STREAM_KEY +StreamDispatcher = event_bus_mod.StreamDispatcher +_ClientSlot = event_bus_mod._ClientSlot +_event_is_visible = event_bus_mod._event_is_visible +_id_greater_than = event_bus_mod._id_greater_than +validate_last_event_id = event_bus_mod.validate_last_event_id + + +pytestmark = pytest.mark.unit + + +# ---------------------------------------------------------------- pure helpers + + +class TestEventIdValidation: + def test_valid_id_accepted(self): + assert validate_last_event_id("1700000000000-0") == "1700000000000-0" + assert validate_last_event_id("1-42") == "1-42" + + def test_missing_id_returns_none(self): + assert validate_last_event_id(None) is None + assert validate_last_event_id("") is None + + def test_malformed_id_rejected(self): + assert validate_last_event_id("not-an-id") is None + assert validate_last_event_id("123") is None + assert validate_last_event_id("123-abc") is None + # CSV/injection attempts should not parse. + assert validate_last_event_id("1-0; DROP") is None + assert validate_last_event_id("$") is None + + +class TestIdComparison: + def test_greater_by_ms(self): + assert _id_greater_than("2-0", "1-999") is True + + def test_greater_by_seq(self): + assert _id_greater_than("1-2", "1-1") is True + + def test_equal_is_not_greater(self): + assert _id_greater_than("1-1", "1-1") is False + + def test_lesser_is_not_greater(self): + assert _id_greater_than("1-0", "2-0") is False + + def test_malformed_returns_false(self): + assert _id_greater_than("junk", "1-0") is False + + +# ----------------------------------------------------------- scope visibility + + +class _FakeWS: + async def close(self, *args, **kwargs): + pass + + +def _make_slot(scope, *, is_admin=False, agents=None): + return _ClientSlot( + ws=_FakeWS(), + scope=scope, + send_func=lambda *_: None, + is_admin=is_admin, + accessible_agents=set(agents or []), + ) + + +class TestScopeVisibility: + def test_all_scope_sees_all_events(self): + slot = _make_slot(SCOPE_ALL) + assert _event_is_visible(slot, SCOPE_ALL, "agent-a") is True + + def test_all_scope_ignores_scoped_events(self): + slot = _make_slot(SCOPE_ALL) + assert _event_is_visible(slot, SCOPE_SCOPED, "agent-a") is False + + def test_scoped_admin_sees_any_agent(self): + slot = _make_slot(SCOPE_SCOPED, is_admin=True) + assert _event_is_visible(slot, SCOPE_SCOPED, "agent-a") is True + assert _event_is_visible(slot, SCOPE_SCOPED, "agent-z") is True + + def test_scoped_user_sees_only_accessible_agents(self): + slot = _make_slot(SCOPE_SCOPED, agents=["agent-a", "agent-b"]) + assert _event_is_visible(slot, SCOPE_SCOPED, "agent-a") is True + assert _event_is_visible(slot, SCOPE_SCOPED, "agent-c") is False + + def test_scoped_event_without_agent_name_not_visible(self): + slot = _make_slot(SCOPE_SCOPED, agents=["agent-a"]) + assert _event_is_visible(slot, SCOPE_SCOPED, None) is False + + def test_scoped_user_ignores_all_events(self): + slot = _make_slot(SCOPE_SCOPED, agents=["agent-a"]) + # Events on SCOPE_ALL should stay on /ws; /ws/events must not replay them. + assert _event_is_visible(slot, SCOPE_ALL, "agent-a") is False + + +# ----------------------------------------------------------------- EventBus + + +class _FakeRedis: + """Minimal in-memory stand-in for ``redis.asyncio.Redis`` that records XADD + calls and exposes them for assertions. Used to isolate EventBus from a real + Redis in unit tests.""" + + def __init__(self): + self.xadd_calls = [] + self.closed = False + + async def ping(self): + return True + + async def xadd(self, key, fields, maxlen=None, approximate=False): + self.xadd_calls.append( + {"key": key, "fields": fields, "maxlen": maxlen, "approximate": approximate} + ) + return f"{len(self.xadd_calls)}-0" + + async def aclose(self): + self.closed = True + + +@pytest.fixture +def fake_redis_bus(monkeypatch): + """Return (bus, fake_redis) with the bus's Redis replaced by a fake.""" + bus = EventBus() + fake = _FakeRedis() + + async def _get_redis(): + bus._ready = True + bus._redis = fake + return fake + + monkeypatch.setattr(bus, "_get_redis", _get_redis) + return bus, fake + + +@pytest.mark.asyncio +async def test_publish_dict_xadds_envelope(fake_redis_bus): + bus, fake = fake_redis_bus + await bus.start() + try: + await bus.publish({"type": "agent_started", "agent_name": "a"}, scope=SCOPE_ALL) + await asyncio.wait_for(bus._outbound.join(), timeout=2.0) + finally: + await bus.stop(drain_timeout=1.0) + + assert len(fake.xadd_calls) == 1 + call = fake.xadd_calls[0] + assert call["key"] == STREAM_KEY + assert call["maxlen"] > 0 and call["approximate"] is True + assert "payload" in call["fields"] + assert call["fields"]["scope"] == SCOPE_ALL + + +@pytest.mark.asyncio +async def test_publish_accepts_json_string(fake_redis_bus): + """Legacy ConnectionManager.broadcast() passes a JSON-encoded string.""" + bus, fake = fake_redis_bus + await bus.start() + try: + await bus.publish('{"type": "agent_stopped", "name": "a"}', scope=SCOPE_ALL) + await asyncio.wait_for(bus._outbound.join(), timeout=2.0) + finally: + await bus.stop(drain_timeout=1.0) + + import json + + payload = json.loads(fake.xadd_calls[0]["fields"]["payload"]) + assert payload["type"] == "agent_stopped" + + +@pytest.mark.asyncio +async def test_publish_infers_agent_name_for_scoped(fake_redis_bus): + bus, fake = fake_redis_bus + await bus.start() + try: + await bus.publish( + {"type": "agent_activity", "agent_name": "ruby", "details": {}}, + scope=SCOPE_SCOPED, + ) + await asyncio.wait_for(bus._outbound.join(), timeout=2.0) + finally: + await bus.stop(drain_timeout=1.0) + + assert fake.xadd_calls[0]["fields"]["agent_name"] == "ruby" + + +@pytest.mark.asyncio +async def test_publish_without_redis_buffers_to_fallback(monkeypatch): + bus = EventBus() + + async def _no_redis(): + return None + + monkeypatch.setattr(bus, "_get_redis", _no_redis) + await bus.start() + try: + for i in range(5): + await bus.publish({"type": "test", "i": i}, scope=SCOPE_ALL) + await asyncio.wait_for(bus._outbound.join(), timeout=2.0) + finally: + await bus.stop(drain_timeout=1.0) + + # All events land in fallback buffer; never raises. + assert len(bus._fallback) == 5 + + +# -------------------------------------------------------- StreamDispatcher + + +@pytest.mark.asyncio +async def test_client_eviction_after_consecutive_failures(): + """Matches the AC: failed sends evict the client after N consecutive failures.""" + dispatcher = StreamDispatcher() + + failures = {"count": 0} + + class _BrokenWS: + async def close(self, *args, **kwargs): + pass + + async def _broken_send(payload): + failures["count"] += 1 + raise RuntimeError("broken pipe") + + client_id = await dispatcher.register( + ws=_BrokenWS(), scope=SCOPE_ALL, send_func=_broken_send + ) + slot = dispatcher._clients[client_id] + + # Feed enough events to trigger eviction. + for i in range(EVICT_AFTER_FAILURES + 1): + slot.queue.put_nowait((f"{i}-0", {"type": "x"})) + + # Give the consumer task time to drain. + for _ in range(20): + await asyncio.sleep(0.01) + if client_id not in dispatcher._clients: + break + + assert client_id not in dispatcher._clients + assert failures["count"] >= EVICT_AFTER_FAILURES + + +@pytest.mark.asyncio +async def test_client_queue_overflow_triggers_resync_marker(): + """Slow consumer shouldn't block fan-out; overflow should request resync.""" + dispatcher = StreamDispatcher() + delivered = [] + release = asyncio.Event() + + class _WS: + async def close(self, *args, **kwargs): + pass + + async def _slow_send(payload): + # Block the consumer until ``release`` is set so the queue saturates. + if payload.get("type") != "resync_required": + await release.wait() + delivered.append(payload) + + client_id = await dispatcher.register( + ws=_WS(), scope=SCOPE_ALL, send_func=_slow_send + ) + + # Push one event (the consumer will block on release). + slot = dispatcher._clients[client_id] + slot.queue.put_nowait(("1-0", {"type": "x"})) + + # Simulate the dispatcher's fan-out on an already-full queue. Fill the queue + # past its capacity — the first put succeeds, subsequent ones go through the + # same code path as _fanout (put_nowait → QueueFull → resync_required). + fills_attempted = CLIENT_QUEUE_MAXSIZE + 10 + filled = 0 + for i in range(fills_attempted): + try: + slot.queue.put_nowait((f"{i+2}-0", {"type": "y"})) + filled += 1 + except asyncio.QueueFull: + break + + # At least one put_nowait must have hit QueueFull — meaning the slow client + # scenario is genuinely reproduced. + assert filled < fills_attempted + + # Release the consumer and let it drain. + release.set() + for _ in range(20): + await asyncio.sleep(0.01) + if delivered: + break + + # Clean up + dispatcher.unregister(client_id) + assert len(delivered) >= 1 + + +@pytest.mark.asyncio +async def test_update_accessible_agents_mutates_slot(): + dispatcher = StreamDispatcher() + + class _WS: + async def close(self, *args, **kwargs): + pass + + async def _send(_): + pass + + client_id = await dispatcher.register( + ws=_WS(), scope=SCOPE_SCOPED, send_func=_send, accessible_agents=["a"] + ) + dispatcher.update_accessible_agents(client_id, ["a", "b"]) + assert dispatcher._clients[client_id].accessible_agents == {"a", "b"} + dispatcher.unregister(client_id) + + +@pytest.mark.asyncio +async def test_consumer_last_delivered_id_is_monotonic(): + """Regression guard for #306 review C1: even if the queue receives events + out of order (e.g. a catchup batch interleaved with live fan-out), the + client's ``last_delivered_id`` must never advance backwards.""" + dispatcher = StreamDispatcher() + + sent = [] + + class _WS: + async def close(self, *args, **kwargs): + pass + + async def _send(payload): + sent.append(payload) + + client_id = await dispatcher.register( + ws=_WS(), scope=SCOPE_ALL, send_func=_send + ) + slot = dispatcher._clients[client_id] + + # Simulate out-of-order delivery: live event 200, then catchup event 100. + slot.queue.put_nowait(("200-0", {"type": "live"})) + slot.queue.put_nowait(("100-0", {"type": "catchup"})) + + for _ in range(30): + await asyncio.sleep(0.01) + if len(sent) >= 2: + break + + # Both delivered, but cursor stuck at the higher id. + assert len(sent) == 2 + assert slot.last_delivered_id == "200-0" + dispatcher.unregister(client_id) + + +@pytest.mark.asyncio +async def test_register_with_invalid_last_event_id_queues_resync(): + """Malformed cursor should not crash; should trigger a resync marker.""" + dispatcher = StreamDispatcher() + + class _WS: + async def close(self, *args, **kwargs): + pass + + sent = [] + + async def _send(payload): + sent.append(payload) + + client_id = await dispatcher.register( + ws=_WS(), + scope=SCOPE_ALL, + send_func=_send, + last_event_id="definitely-not-a-valid-id", + ) + + # Give catchup task a chance to run. + for _ in range(10): + await asyncio.sleep(0.01) + if sent: + break + + assert any(p.get("type") == "resync_required" for p in sent) + assert any(p.get("reason") == "invalid_last_event_id" for p in sent) + dispatcher.unregister(client_id)