SPEC-065 — Event Stream Taxonomy

Canonical event topics and payload shape for SSE/subscribe consumers.

Source: docs/design/spec-065-event-stream-taxonomy.md

SPEC-065: Event stream taxonomy & subscription filters

Field Value
Status Shipped
Owner kris.ganjam@gmail.com
Date 2026-05-01
Pillar Cross-cutting
Implementation #451
Target repos: scidex-substrate (this doc, framework, event-name registry, subscription DSL parser, SSE backpressure). scidex-prism (live activity ticker subscribes — separate PR).
Masters: SPEC-001 §12 (event substrate), §13 (agent communication patterns), SPEC-019 (cutover SSE guarantees), SPEC-038 (the primary Prism consumer).

Adjacent context:

  • SPEC-001 §12 declared three event-name namespaces but the substrate today emits a mixed bag — some hypothesis_scored, some hypothesis.scored, some score_changed. 34+ distinct event types in flight, no central registry.

  • scidex.subscribe accepts only a list of event-type strings (with last-segment wildcard). No filter on agent, target, payload — clients over-subscribe and filter client-side.

  • Postgres LISTEN/NOTIFY is wired but a single subscriber connection holds one PG session. Fan-out to many SSE clients is naive (one PG listener per HTTP connection). Will not survive the SPEC-038 ticker at scale.

  • v1 had no SSE; SPEC-019 leans on substrate SSE for the ticker but does not pin envelope shape, ACK, or backpressure.

TL;DR

Anchored against north-star.md §6 “Treat every signal as a first-class audit trail”: every meaningful state change is already an event; the gap is no published taxonomy and no rich subscription filter. SPEC-065 finalises the SPEC-019 SSE contract: a formal three-tier event-type hierarchy (<artifact_type>.<lifecycle> / signal.<dimension>.recorded / link.<rel>.added / comment.posted / mission.assigned / market.resolved), a JSON subscription DSL filtering by type/agent/target/payload, one shared LISTEN/NOTIFY channel with server-side fan-out, versioned event envelopes, Last-Event-Id ACK, and 30-second slow-client drop. Net effect: Prism’s live ticker, agent subscribers, and external mirrors share one durable, filterable event substrate.

1. Problem statement

Four interlocking gaps:

  1. No event-name registry. Code paths invent their own names. hypothesis.scored vs hypothesis_score_changed vs score.updated. Subscribers ship blind regex matches that break silently when an emitter renames.

  2. Subscription DSL is too thin. types=['hypothesis.*'] is the entire surface. The most common needs — “events on artifacts I care about” / “events from agent X” — are not expressible.

  3. SSE fan-out doesn’t scale. Each HTTP/SSE client opens its own PG LISTEN. At 100 viewers, 100 idle PG sessions — fragile under max_connections.

  4. No backpressure contract. Slow clients accumulate buffer server-side. No documented eviction. A single slow client can stall the broadcast loop.

These add up to: SPEC-038 cannot ship reliably, SPEC-051 pantheon-debate subscribers cannot filter to “events on this debate”, SPEC-019 assumes SSE works at scale without specifying how.

2. Decision

2.1 Formal event taxonomy

Three top-level shapes plus a versioned envelope:

Shape Pattern Examples
Artifact lifecycle <artifact_type>.<lifecycle> hypothesis.created, hypothesis.promoted, pantheon_debate.frozen, paper.archived, wiki_page.merged
Signal recorded signal.<dimension>.recorded signal.upvote.recorded, signal.fund.recorded, signal.calibration_score.recorded
Relationship change link.<rel>.added / link.<rel>.removed link.supports.added, link.cites.added, link.contradicts.removed
Communication comment.posted, comment.reacted, comment.edited unchanged
Domain-specific mission.assigned, market.resolved, collider.opened, senate.quality_gate_evaluated, flag.raised one per documented domain action
System system.<noun>.<verb> system.lmsr_subsidy_exhausted, system.embed_backfill_completed

Names are lowercase snake_case with . as separator. * matches one segment; ** matches across segments (e.g. hypothesis.**). Names are stable — renaming requires a deprecation period during which old + new both fire.

A new module events/registry.py holds the canonical list as frozenset[str] plus a small Pydantic payload model per type. Emitters call events.publish(EVENT_NAME, …) where EVENT_NAME is imported. Emitting an unregistered name is a runtime error in tests, a warning in prod.

2.2 Versioned envelope

{
  "envelope_version": 1,
  "event_id":   14728316,                 // monotonic, gap-free per emitter
  "event_type": "hypothesis.promoted",
  "ref":        "hypothesis:h-7a9c@sha256:abc",
  "agent":      {"id":"agent-u-...", "kind":"ai_persona"},
  "session_id": "...",
  "occurred_at":"2026-05-01T12:34:56Z",
  "payload": { "from_state":"supported", "to_state":"promoted", "score":0.83 },
  "links":   [{"rel":"target","ref":"hypothesis:h-7a9c"}]
}

envelope_version lets future evolutions (e.g. adding delegation_chain) avoid breaking older subscribers.

2.3 Subscription DSL

{
  "types":   ["hypothesis.*", "signal.upvote.recorded"],
  "agents":  ["persona:mendel", "human:u-jane"],
  "targets": ["paper:p-123", "hypothesis:h-7a9c"],   // event.ref OR event.links[].ref
  "payload": {">=": [{"var":"score"}, 0.7]},          // JSONLogic over event.payload
  "since":   14728000,                                // event_id cursor
  "consumer_id": "ticker-prism-prod-1"                // durable cursor key
}

All four scoping fields are AND-combined. Within a field, OR-semantics. Empty / omitted fields are wildcards.

def matches(event, filter):
    if filter.types and not any(glob_match(t, event.event_type) for t in filter.types): return False
    if filter.agents and event.agent.id not in filter.agents: return False
    if filter.targets:
        target_refs = {event.ref} | {l.ref for l in event.links}
        if not (target_refs & set(filter.targets)): return False
    if filter.payload and not jsonlogic.eval(filter.payload, event.payload): return False
    return True

The same filter is accepted by scidex.poll (pull) and scidex.subscribe (push) — one DSL, two delivery modes per SPEC-001 §13.

2.4 SSE channel multiplexing

Replace per-client LISTEN with one server-process LISTEN scidex_events plus an in-process broker:

Postgres ──NOTIFY──> [single LISTEN coroutine] ──> [in-proc broker] ──┬──> SSE client A (filter f_A)
                                                                       ├──> SSE client B (filter f_B)
                                                                       └──> SSE client … (filter f_…)

The single LISTEN coroutine receives every notify, fetches the event row by id, evaluates each connected client’s filter, forwards to those whose filter matches. PG connection count is O(1) per substrate process regardless of viewer count.

Per-client outbound queue cap: 256 events. Overflow → client marked degraded.

2.5 Backpressure & ACK

  • SSE clients send Last-Event-Id on reconnect; server resumes by reading from the events table (durable replay).

  • During a live connection, server tracks the latest event the client has been written. If the gap exceeds 256 events OR last ACK was >30s ago, the connection closes with HTTP-event system.client_dropped_slow and a metric.

  • Clients reconnect, present Last-Event-Id, replay from cursor. No event is lost.

  • scidex.subscribe MCP/AsyncIterator semantics use the same per-iterator queue cap; if await next doesn’t run within 30s the iterator is closed.

2.6 Cursor durability

Per SPEC-001 §12, durable consumers register a consumer_id; cursor in consumer_subscriptions. On reconnect with the same consumer_id, replay continues. Cursors advance server-side after delivery confirmation. If silent for >7 days, cursor is GC’d; reconnect requires explicit since=.

3. Schema / verb changes (concrete)

  • New module: events/registry.py — frozenset[str] of all event types + payload Pydantic model per type.

  • New module: events/broker.py — single LISTEN coroutine + in-proc fan-out.

  • New module: events/filter.py — DSL parser + matcher.

  • Updated: scidex.subscribe and scidex.poll accept the new filter shape (back-compat: bare types=[...] rewritten to {types: [...]}).

  • Updated: every emitter call site imports event-type constants from events.registry (mechanical refactor).

  • Migration: 0053_consumer_subscriptions_filter.sql adds filter_json JSONB to consumer_subscriptions (nullable, back-compat).

  • Health: /health/deep reports events: { registered_types: 47, in_flight_subscriptions: 12, broker_lag_ms: 3 }.

No new top-level tables. The events table already exists.

4. Acceptance

  1. Registry coverage. Every emitter imports from events.registry. CI grep test fails on string-literal events.publish("...").

  2. Type taxonomy enforced. Emitting unregistered names logs a warning in prod, raises in tests.

  3. Filter DSL end-to-end. Subscribe {types:["hypothesis.*"], agents:["persona:mendel"]}. Mendel creates hypothesis → received. Darwin creates hypothesis → not. Mendel posts comment → not.

  4. Target filter. {targets:["paper:p-123"]} delivers link.cites.added with link.ref="paper:p-123"; unrelated link is not delivered.

  5. Payload filter. {types:["signal.upvote.recorded"], payload:{">=":[{"var":"weight"},5]}} delivers only high-weight upvotes.

  6. Fan-out scale. 100 concurrent SSE clients, distinct filters. Substrate process opens exactly 1 PG LISTEN. p95 broker latency <50 ms at 100 events/s ingest.

  7. Backpressure drop. Slow client (sleep on TCP write). After 30 s, server closes with system.client_dropped_slow. Reconnect with Last-Event-Id resumes correctly.

  8. Durable replay. Disconnect a consumer_id-registered subscriber; emit 100 events; reconnect; the 100 missed events replay in order from the cursor.

  9. Envelope versioned. Bump to envelope_version=2. Old subscribers (declared =1 accept) reject gracefully with a typed error rather than crashing.

5. Rollout

Phase 1 (1 PR, M): registry module + emitter refactor. No subscriber-visible change yet — internal tidy.

Phase 2 (1 PR, M): filter DSL + matcher. scidex.subscribe accepts the new shape (back-compat with bare list).

Phase 3 (1 PR, L): broker + single-LISTEN fan-out + backpressure. Behind SCIDEX_SSE_BROKER=1 flag; fall back to per-client LISTEN if disabled.

Phase 4 (1 PR, S): durable consumer cursor. SPEC-038 ticker becomes the first production subscriber.

Phase 5 (per-feature): SPEC-068 moderation surfaces, SPEC-051 pantheon-debate “live argument” feed, SPEC-018 economic-loop signal stream all ship subscribers using the new DSL.

Compatible with SPEC-019 cutover: Phases 1-3 pre-cutover; Phase 4 post-cutover; Phase 5 per-feature thereafter. Rollback is per-phase; PG LISTEN plumbing stays valid throughout.

Work Log

2026-05-14 — Phases 1–4 implementation

Phase 1 — merged (commit 2d197a3 / PR #451)

  • src/scidex_substrate/events/taxonomy.py — canonical event-type constants (25 types)

  • src/scidex_substrate/events/filter.py — SPEC-065 §2.3 DSL matcher (types/agents/targets)

  • src/scidex_substrate/events/__init__.py — module entry

  • src/scidex_substrate/migrations/0059_subscription_filters.sql — subscription_filters table

  • src/scidex_substrate/skill/verbs/poll.py — scidex.poll with SPEC-065 DSL routing

  • src/scidex_substrate/api/sse.py — SSE endpoint with full filter support

Phase 2 — payload JSONLogic filter [task:d0cdcde9-51a6-4ed7-8f15-68ec4051de39]

  • Added _jsonlogic_eval() to events/filter.py supporting var, >=, <=, >, <, ==, !=, and, or, !. Unknown operators are permissive (do not constrain).

  • matches() now evaluates filter["payload"] as a JSONLogic rule over event["payload"].

  • Updated _SPEC065_KEYS in poll.py to include "payload" and "since"/"consumer_id".

  • 20 new unit tests in tests/unit/test_event_filter_p2.py.

Phase 3 — single-LISTEN broker [task:d0cdcde9-51a6-4ed7-8f15-68ec4051de39]

  • Created src/scidex_substrate/events/broker.py:

    • EventBroker singleton: one PG LISTEN scidex_events per process.

    • Subscriber(filter_, since): per-client asyncio.Queue(maxsize=256) + slow-client detection.

    • _dispatch(): fan-out with backpressure (queue overflow → degraded) and 30 s slow-client eviction (is_slow() check on non-empty queue).

    • Reconnect loop: PG errors restart after 5 s without stalling the SSE service.

    • broker_enabled(): checks SCIDEX_SSE_BROKER=1.

  • Updated src/scidex_substrate/api/sse.py:

    • broker_stream(): replay missed events on connect, then drain from subscriber queue, emit system.client_dropped_slow on degradation, heartbeat at heartbeat_secs.

    • make_sse_endpoint(): routes to broker_stream when broker_enabled(), falls back to per-client event_stream (legacy, default).

  • 13 new unit tests in tests/unit/test_event_broker.py.

Phase 4 — durable cursor Already complete: consumer_subscriptions (with filter_json, cursor, last_activity_at) is in 0001_initial_schema.sql; scidex.poll implements full UPSERT cursor tracking. Phase 4 is production-ready; SPEC-038 ticker can register as a durable consumer via consumer_id on scidex.poll.

Phase 5 — per-feature (SPEC-068 moderation, SPEC-051 pantheon-debate, SPEC-018 economic-loop) are downstream and ship individually per the rollout plan.

6. Open questions

  • Glob vs regex. Globs are simpler but limited. Decision: ship globs first; revisit if real demand emerges.

  • Cross-substrate federation. A future federated read (SPEC-021) could surface as federated.<remote>.<event>. Out of scope; registry can absorb such names later.

  • Replay window. events grows unboundedly. Declare 30 days of online replay; rely on archival for older. Confirm with SPEC-008.

  • Per-agent concurrency. Should one agent have 100 concurrent subscriptions? Likely no for humans, yes for orchestrator. Tie to SPEC-004.

  • Event ordering across emitters. event_id is monotonic per emitter. Across emitters, follows insertion order. Subscribers should order on event_id, not occurred_at.

  • WebSocket variant. SSE is one-way. Defer; not required by current consumers.

  • Visibility filtering. Some events (moderation, COI per SPEC-068) should not be visible to all subscribers. Per-event visibility: public|reviewer_only|admin enforced by the broker. Detailed in SPEC-068.