NATS & Events -- Swiss AI Agent Protocol Reference

Look up NATS/event information. Topic or question via $ARGUMENTS.

Architecture Overview

The platform uses NATS as its central message bus with two tiers:

Key Rule: Control events go through JetStream (durable). Display events go through NATS Core (ephemeral).

Swiss AI Agent Protocol (SAAP)

Event Classification

All events inherit from BaseEvent and are classified into two categories:

Core Rule: Only ControlEvent types trigger agent @step() methods. Display events MUST NEVER influence workflow logic.

Event Hierarchy

BaseEvent
├── ControlEvent (workflow-driving)
│   └── ControlAndDisplayEvent (hybrid: workflow + UI)
│       ├── (capability calls and results, tool loop: see "Capability and Tool-Loop Events")
│       ├── StartEvent
│       │   └── UserMessageEvent (user chat message)
│       ├── StopEvent (run completed)
│       ├── SemanticEvent (OpenInference tracing)
│       │   ├── ExceptionEvent (error, halts run)
│       │   ├── LLMEvent (LLM call details)
│       │   │   └── LLMStopEvent (terminal LLM response)
│       │   ├── AgentEvent (agent tracing)
│       │   ├── ChainEvent (chain tracing)
│       │   ├── RetrieverEvent (RAG retrieval)
│       │   ├── RerankerEvent (reranking)
│       │   ├── GuardEvent (safety checks)
│       │   ├── ToolEvent (tool invocation)
│       │   └── EmbeddingEvent (vector generation)
│       ├── RouterEvent (LLM routing decision)
│       ├── BaseRetrieveMemoryEvent
│       │   ├── RetrieveUserMemoryEvent
│       │   └── RetrieveOrganizationMemoryEvent
│       ├── BaseStoreMemoryEvent
│       │   ├── StoreUserMemoryEvent
│       │   └── StoreOrganizationMemoryEvent
│       ├── HumanInTheLoopRequestEvent (pause for human input)
│       │   ├── HumanInTheLoopInputRequestEvent
│       │   ├── HumanInTheLoopConfirmationRequestEvent
│       │   └── HumanInTheLoopChatRequestEvent
│       ├── HumanInTheLoopResponseEvent[T] (human replied)
│       ├── AgentInTheLoopRequestEvent (delegate to agent)
│       ├── AgentInTheLoopResponseEvent (agent replied)
│       └── AgentInTheLoopExceptionEvent (agent failed)
│   ├── LanguageEvent
│   └── BotInTheLoopRequestEvent (ask via Slack/Teams)
│
├── DisplayEvent (observability-only)
│   ├── ThoughtEvent (agent reasoning)
│   ├── ChunkEvent (streaming text tokens)
│   ├── AttachedFileEvent (a file the model received, with its citation id)
│   ├── ToolLoopStatusEvent (status while the loop decides)
│   ├── CostEvent
│   │   └── LLMCostEvent (token costs)
│   └── (All ControlAndDisplayEvent types also publish as display)
│
├── ProcessEvent (process orchestration)
│   ├── WorkEvent (work completed)
│   │   ├── ProcessStartEvent
│   │   ├── ProcessStopEvent
│   │   ├── ProcessExceptionEvent
│   │   ├── HumanWorkEvent (human submitted form)
│   │   └── ProgramWorkEvent (program submitted data)
│   └── WorkRequestEvent (delegate work)
│       ├── HumanWorkRequestEvent (ask human with forms)
│       └── ProgramWorkRequestEvent (ask program)
│
├── ClassDiscoveryRequestEvent (query agent/process metadata)
├── AgentClassDiscoveryResponseEvent (agent metadata)
├── ProcessClassDiscoveryResponseEvent (process metadata)
└── SourceUpdatedEvent (pipeline trigger)

Semantic Events & OpenInference

SemanticEvent subclasses implement to_semantic_convention(), producing attributes compatible with the OpenInference specification. This enables export to Langfuse, Arize Phoenix, and any OpenTelemetry-compatible system.

from swiss_ai_hub.core.events.agent import RetrieverEvent

event = RetrieverEvent.from_nodes(retrieved_nodes)
otel_attributes = event.to_semantic_convention()
# Produces: openinference.span.kind=RETRIEVER, retrieval.documents.{i}.document.id, etc.

When to use semantic vs generic events: Prefer semantic events when observability matters. RetrieverEvent over a custom MyRetrieveEvent(ControlAndDisplayEvent) gives you OpenInference spans for free. For choosing which base event to use when building agents, see /scaffold-agent.

Events as Flow Carriers

Events serve two distinct purposes:

1. Data carriers: Transporting values between steps
2. Flow carriers: Controlling execution order independent of data

A step may depend on an event solely to ensure execution ordering, without requiring any data from it:

class PathA(Event):
    pass  # No fields - pure flow control

@step()
async def handle_path_a(self, _: PathA, original: StartEvent) -> StopEvent:
    # Underscore signals: "I need this event for flow control, not data"
    process(original.data)
    return StopEvent()

The _: EventType convention indicates dependency on an event's existence rather than its contents. This pattern is essential for:

• Conditional branching: Different steps execute based on which event type was emitted
• Sequencing: Ensuring step B waits for step A without needing A's output data
• Synchronization barriers: Waiting for a signal that work completed

Formal definition: An event e is a pure flow carrier if fields(e) = ∅. The execution constraint R(s) ⊆ Eₜ depends only on event type existence, not event content. Therefore, a step may depend on an event type T without accessing any value from instances of T.

Stop Event Constraint

Constraint: No step may depend on StopEvent or any subclass as an input. When a StopEvent is emitted, the run terminates. Subsequent steps are not scheduled. No events may be published after StopEvent.

# ILLEGAL: Depending on stop event
@step()
async def cleanup(self, stop: LLMStopEvent) -> CleanupEvent:
    ...  # Never executes

# CORRECT: Use non-stop event, then explicit stop
@step()
async def respond(self, event: Input) -> LLMEvent:  # Not LLMStopEvent
    return await displayer.display_llm_stream(..., as_stop_step=False)

@step()
async def cleanup(self, llm: LLMEvent) -> CleanupEvent:
    ...

@step(precondition=cleanup_complete)
async def finalize(self, cleanup: CleanupEvent) -> StopEvent:
    return StopEvent()

UserMessageEvent is a Chat UI Contract — Extend With Care

UserMessageEvent is the canonical entry point for chat interfaces (OpenWebUI, Teams, Slack, WebChat). Every chat UI that wants to drive an agent must know how to publish and render it. Every new field added to UserMessageEvent — or to a subclass that rides on the same "chat start" contract — raises the bar for every chat client in the ecosystem.

Rule: If the payload is agent-/domain-specific and the publisher is not a generic chat UI (e.g. a custom domain front-end that runs its own selection flow, or another agent delegating via AgentInTheLoop), do not subclass UserMessageEvent. Subclass StartEvent directly and have the agent accept UserMessageEvent | YourStartEvent.

Choosing the Right Base Event

Extended Event Reference

Guard Events

Utility Events

Capability and Tool-Loop Events

A capability (packages/agent/swiss_ai_hub/agent/capabilities/) is called with a request event and answers with a result event; see /scaffold-agent ("Step 4b") for the calling side. Requests are imperative, results are past participles. Protocol events are control and display: every one of them is a ControlAndDisplayEvent so it appears in the event history, and each has its own frontend component (/scaffold-event-display). Only high-frequency or purely cosmetic events stay DisplayEvent (ChunkEvent, AttachedFileEvent, ToolLoopStatusEvent, ToolLoopCondensedEvent, SandboxFileDisplayedEvent).

Tool-loop internals (events/agent/tool_loop/): ToolLoopIterationEvent per round, ToolCallsDecidedEvent (the model's choice), ToolApprovalRequestEvent / ToolApprovalResponseEvent (a HumanInTheLoopConfirmationRequestEvent and its response, shown as a yes/no in chat), ToolCallApprovedEvent, ToolResultEvent, ToolEvent per call and ToolLoopStatusEvent. The loop's messages and counters travel on the events (ToolLoopState); declined tools, cite_sources and remembered approvals live in RunContext (once-per-conversation approvals in ThreadContext). Code execution adds SandboxFileDisplayedEvent (events/agent/sandbox/) when an agent shows a sandbox file to the user.

Memory Events

Interaction Events

HITL Helper Classes (not events, but workflow utilities):

BaseEvent Core Fields

class BaseEvent(BaseModel):
    event_id: str = Field(default_factory=lambda: str(ObjectId()))
    created_at: int = Field(default_factory=time.time_ns)  # Nanosecond precision
    _event_name: str  # Computed: class name (used for deserialization)
    _parent_event_names: list[str]  # Computed: inheritance chain
    _jetstream_sequence: int | None  # JetStream sequence number (set by subscriber)

Auto-Registration: Every BaseEvent subclass auto-registers in _event_registry via __pydantic_init_subclass__. No manual registration needed.

Deserialization: BaseEvent.deserialize_event(data) looks up _event_name in registry, falls back to parent classes, preserves unknown fields.

Type Checking Properties:

event.is_control_event      # "ControlEvent" in _parent_event_names
event.is_display_event      # "DisplayEvent" in _parent_event_names
event.is_semantic_event     # "SemanticEvent" in _parent_event_names
event.is_start_event        # "StartEvent" in _parent_event_names
event.is_stop_event         # "StopEvent" in _parent_event_names
event.is_work_event         # "WorkEvent" in _parent_event_names
event.is_work_request_event # "WorkRequestEvent" in _parent_event_names
event.is_chunk_event        # isinstance check
event.is_hitl_response_event  # isinstance check

NATS Subject (Topic) System

Agent Subject Pattern

agent.{agent_class}.{agent_id}.{thread_id}.{display_id}.{run_id}.{event_type}.{event_name}.{event_id}
  0        1             2          3            4           5         6            7            8

Process Subject Pattern

process.{process_class}.{process_id}.{walkthrough_id}.{event_type}.{event_name}.{event_id}

RPC Subject Pattern

aihub.rpc.config.agent.{agent_class}.{agent_id}
aihub.rpc.config.process.{process_class}.{process_id}

Discovery Subject Pattern

class_discovery.agent.{agent_class}.*.request.{call_id}
class_discovery.agent.{agent_class}.*.response.{call_id}
instance_discovery.agent.{agent_class}.{agent_id}.*.request.{call_id}

Hierarchical Scoping (Thread > Display > Run)

Security: Users can only observe events from threads they're members of (ThreadEntity.users).

Stream Naming

Streams are named per agent class: agent_{agent_class}_stream (e.g., agent_RAGAgent_stream).

Topic Managers

Topic managers centralize subject construction. Never build subjects manually — use the appropriate manager.

TopicManager Hierarchy

TopicManager (base: RPC_TOPIC, CLASS_DISCOVERY_TOPIC, INSTANCE_DISCOVERY_TOPIC)
├── AgentTopicManager (all-agent subjects, discovery, RPC)
│   ├── AgentClassTopicManager(agent_class) (class-level, streams)
│   │   ├── AgentInstanceTopicManager(agent_class, agent_id) (instance-level)
│   │   │   └── AgentThreadTopicManager(agent_class, agent_id, thread_id, display_id, run_id)
│   │   └── (get_stream() → stream_name, stream_subject)
│   └── (get_agent_config_rpc_subject, get_subject_for_all_*_events)
└── ProcessTopicManager (process equivalents)
    └── ProcessClassTopicManager(process_class)

Common TopicManager Methods

# AgentTopicManager — global agent subjects
tm = AgentTopicManager()
tm.get_subject_for_all_events_in_agent()          # agent.*.*.*.*.*.*.*.*
tm.get_subject_for_all_display_events_in_agent()   # agent.*.*.*.*.*.display_event.*.*
tm.get_subject_for_all_control_events_in_agent()   # agent.*.*.*.*.*.control_event.*.*
tm.get_agent_config_rpc_subject("*", "*")           # aihub.rpc.config.agent.*.*
tm.get_agent_class_discovery_subject_request(call_id)  # class_discovery.agent.*.*.request.{call_id}

# AgentClassTopicManager — per agent class
ctm = AgentClassTopicManager(agent_class="RAGAgent")
ctm.get_stream()  # ("agent_RAGAgent_stream", "agent.RAGAgent.>")
ctm.get_subject_for_all_control_events()  # agent.RAGAgent.*.*.*.*.control_event.*.*

# AgentThreadTopicManager — per thread context
ttm = AgentThreadTopicManager(agent_class="RAGAgent", agent_id="wiki", thread_id="t1", display_id="d1", run_id="r1")
ttm.get_subject_for_control_event_in_thread(event_name="StartEvent", event_id="e1")
ttm.get_subject_for_display_event_in_thread(event_name="ChunkEvent", event_id="e2")

File locations:

• packages/core/swiss_ai_hub/core/topic_managers/topic_manager.py
• packages/core/swiss_ai_hub/core/topic_managers/agents/agent_topic_manager.py
• packages/core/swiss_ai_hub/core/topic_managers/agents/agent_class_topic_manager.py
• packages/core/swiss_ai_hub/core/topic_managers/agents/agent_instance_topic_manager.py
• packages/core/swiss_ai_hub/core/topic_managers/agents/agent_thread_topic_manager.py

Publishers

NCPublisher (NATS Core — Ephemeral)

from swiss_ai_hub.core.publishers import NCPublisher

publisher = NCPublisher("MyPublisher", nc)
await publisher.publish_event(event, subject)

Characteristics:

• Fire-and-forget, no retry
• Adds OpenTelemetry trace context headers
• Validates event-subject alignment (warns on mismatch)

JSPublisher (JetStream — Persistent)

from swiss_ai_hub.core.publishers import JSPublisher

publisher = JSPublisher("MyPublisher", js)
await publisher.ensure_stream_exists(stream_name, stream_subject)
await publisher.publish_event(event, subject, retries=10)

Characteristics:

• Retry with 1s backoff, up to 10 attempts (configurable)
• 5-second timeout per attempt
• UUID message ID for deduplication
• ACK confirmation with sequence number
• Raises RuntimeError after all retries exhausted

Publishing Decision Logic

# In dispatchers — the standard pattern:
if event.is_control_event:
    await self.js_publisher.publish_event(event, control_subject)   # JetStream
if event.is_display_event:
    await self.nc_publisher.publish_event(event, display_subject)   # NATS Core

Note: ControlAndDisplayEvent types are published to both channels.

Message Headers

from swiss_ai_hub.core.tracing import NATSMessageHeaders

headers = (
    NATSMessageHeaders()
    .with_trace_context()                          # OpenTelemetry W3C trace context
    .with_header("Nats-Msg-Id", str(uuid.uuid4()))  # Deduplication ID
    .to_dict()
)

Subscribers

NCSubscriber (NATS Core — Ephemeral)

from swiss_ai_hub.core.subscribers import NCSubscriber

subscriber = NCSubscriber(
    name="MySubscriber",
    nc=nc,
    subject="agent.*.*.*.*.*.display_event.*.*",
    event_cls=DisplayEvent,
    handler=my_handler,
)
await subscriber.start()
# ... later ...
await subscriber.stop()

Handler signature: async def handler(event: TEvent, topic: Topic) -> None

Characteristics:

• Ephemeral (no persistence, no replay)
• Non-blocking: spawns asyncio.Task per message
• Extracts trace context from headers

JSSubscriber (JetStream — Durable with Queue Groups)

from swiss_ai_hub.core.subscribers import JSSubscriber

subscriber = JSSubscriber(
    name="MySubscriber",
    nc=nc,
    subject="agent.RAGAgent.*.*.*.*.control_event.*.*",
    stream_name="agent_RAGAgent_stream",
    stream_subject="agent.RAGAgent.>",
    queue_group="my-worker-group",  # Load balancing
    event_cls=ControlEvent,
    handler=my_handler,
    js=js,
)
await subscriber.start()

Characteristics:

• Ensures stream exists before subscribing
• Queue groups for load-balanced delivery across instances
• Immediate ACK (at-least-once semantics)
• Semaphore limits concurrent handlers to 1000
• Sets event._jetstream_sequence from message metadata

Typed Subscriber Factories (Preferred)

Use the factory class methods instead of constructing subscribers directly:

from swiss_ai_hub.core.subscribers import AgentNCSubscriber
from swiss_ai_hub.core.subscribers import AgentJSSubscriber

# All display events from all agents (NATS Core)
sub = AgentNCSubscriber.for_all_agents_display_events(nc=nc, topic_manager=tm, handler=handler)

# All events from all agents (NATS Core)
sub = AgentNCSubscriber.for_all_agent_events(nc=nc, topic_manager=tm, handler=handler)

# Thread-scoped display events (NATS Core)
sub = AgentNCSubscriber.for_thread_display_events(nc=nc, topic_manager=thread_tm, handler=handler)

# All thread events including control (NATS Core)
sub = AgentNCSubscriber.for_all_thread_events(nc=nc, topic_manager=thread_tm, handler=handler)

# Discovery request events (NATS Core)
sub = AgentNCSubscriber.for_agent_class_discovery_request_events(nc=nc, topic_manager=tm, handler=handler)

# Control events for agent instance (JetStream, load-balanced)
sub = AgentJSSubscriber.for_agent_instance_control_events(
    nc=nc, topic_manager=instance_tm, handler=handler, queue_group="my-group", js=js
)

RPC (Request-Reply)

NCRequester (Client)

from swiss_ai_hub.core.requester import NCRequester

requester = NCRequester(
    name="AgentConfig",
    nc=nc,
    response_cls=FetchAgentConfigResponse,
    default_timeout_ms=5000,
)

response = await requester.request(
    FetchAgentConfigRequest(agent_class="RAGAgent", agent_id="wiki"),
    subject="aihub.rpc.config.agent.RAGAgent.wiki",
    timeout_ms=5000,
)

NCResponder (Server)

from swiss_ai_hub.core.responder import NCResponder

responder = NCResponder(
    name="AgentConfig",
    nc=nc,
    subject="aihub.rpc.config.agent.*.*",  # Wildcards for all requests
    request_cls=FetchAgentConfigRequest,
    handler=handle_config_request,
)
await responder.start()

Handler signature: async def handler(request: TRequest, subject: str) -> TResponse

Error handling: On exception, responds with {"error": str, "error_type": str}. Requester raises TimeoutError on no response.

High-Level RPC Client (Preferred)

from swiss_ai_hub.core.rpc import AgentConfigClient

client = AgentConfigClient(nc=nc, timeout_ms=5000)
config = await client.fetch_config(agent_class="RAGAgent", agent_id="wiki")

RPC Models

# packages/core/swiss_ai_hub/core/rpc/models.py
class FetchAgentConfigRequest(BaseModel):
    agent_class: str
    agent_id: str

class FetchAgentConfigResponse(BaseModel):
    agent_class: str
    agent_id: str
    config: dict[str, Any]
    found: bool = True
    error: str | None = None

JetStream Event Store

The JetStreamEventStore provides durable event storage with full-history replay.

Startup Sequence

1. Ensure stream exists via StreamManager
2. Subscribe to new control events (push consumer, durable)
3. Replay ALL historical events (pull consumer with DeliverPolicy.ALL)
4. Delete temporary replay consumer
5. Cache events in TTLCache(maxsize=100_000, ttl=30 days)

Stream Configuration

StreamConfig(
    name="agent_RAGAgent_stream",
    subjects=["agent.RAGAgent.>"],
    storage=StorageType.FILE,           # Persistent disk storage
    retention=RetentionPolicy.LIMITS,   # Bounded by count/age
    max_msgs=10_000_000,                # Max messages
    discard=DiscardPolicy.OLD,          # Evict oldest
    max_age=60 * 60 * 24 * 30,         # 30 days
    duplicate_window=60,                # 60s dedup window
)

JSPoller (Pull Consumer)

from swiss_ai_hub.core.polling import JSPoller

poller = JSPoller(js=js, stream_name="...", stream_subject="...", consumer_name="...")
await poller.ensure_consumer_exists(
    deliver_policy=DeliverPolicy.ALL,   # ALL, NEW, LAST_PER_SUBJECT
    ack_policy=AckPolicy.EXPLICIT,      # EXPLICIT, NONE, ALL
    max_deliver=3,
    filter_subject="...",
)

async for msg in poller.poll(batch_size=100, timeout=1.0):
    process(msg.event)
    await msg.ack()

Dispatcher Architecture

The BaseDispatcher orchestrates event processing using both publishers and the event store:

class BaseDispatcher(abc.ABC):
    nc_publisher: NCPublisher       # Display events (ephemeral)
    js_publisher: JSPublisher       # Control events (durable)
    event_store: JetStreamEventStore  # Event history + replay
    step_store: StepStore           # Step state in Redis (ephemeral)

AgentDispatcher Flow

1. Receive ControlEvent via JSSubscriber (queue group for load balancing)
2. Lookup which @step() methods accept this event type
3. Create fresh agent instance with RunContext and ThreadContext
4. Execute step method → returns new event(s)
5. Publish result:
   • ControlEvent → JetStream (durable)
   • DisplayEvent → NATS Core (ephemeral)
   • ControlAndDisplayEvent → both

ProcessDispatcher Flow

1. Receive WorkEvent (human/agent/program completed work)
2. Lookup which @process_step() methods accept this event type
3. Execute step → returns WorkRequestEvent (delegate to next entity)
4. Publish request to appropriate entity

Debugging: For diagnosing dispatcher issues (steps executing multiple times, steps never firing, events after stop), use /debug-agent which provides MCP-powered runtime inspection of NATS streams and MongoDB event stores.

Event Flow: End-to-End

User Query → Agent Response

Frontend → API Gateway (POST /agents/{class}/{id}/{event}/stream)
    ↓
API Gateway → ExternalAgentEventDistributor
    ↓  validates thread membership, creates run_id
    ↓  publishes UserMessageEvent as ControlEvent
NATS (JetStream) → agent.RAGAgent.wiki.thread1.display1.run1.control_event.UserMessageEvent.e1
    ↓
AgentDispatcher (JSSubscriber, queue group)
    ↓  receives ControlEvent, executes @step()
    ↓  agent publishes ChunkEvent (display), StopEvent (control+display)
NATS (Core) → agent.RAGAgent.wiki.thread1.display1.run1.display_event.ChunkEvent.e2
    ↓
API (NCSubscriber) → EventPersister (MongoDB) + WebSocketSender (UI)
    ↓
Frontend WebSocket ← ContextualizedAgentEvent { event, agent_class, thread_id, ... }

Parallel API Subscribers

The API creates two subscribers on startup:

SSE Streaming (OpenAI-compatible)

The API creates temporary per-request subscribers for SSE streams:

# ChatService creates temporary subscriber → queue → SSE generator
subscriber = AgentNCSubscriber.for_thread_display_events(nc=nc, topic_manager=ttm, handler=queue_handler)
await subscriber.start()

# SSE generator consumes from queue
async def sse_event_generator():
    while not stop_signal.is_set():
        chunk = await chunk_queue.get()
        yield f"data: {chunk.model_dump_json()}\n\n"

NATS Connection & Configuration

NatsSettings

# packages/core/swiss_ai_hub/core/infrastructure/nats/nats_settings.py
class NatsSettings(EnvironmentSettings):
    model_config = EnvironmentSettings.create_settings_config("NATS_")
    ENDPOINT: str                    # NATS_ENDPOINT (e.g., "nats://localhost:4222")
    TOKEN: SecretStr | None = None   # NATS_TOKEN (optional auth token)

    @classmethod
    async def create_client(cls) -> NATS:
        settings = cls()
        nc = NATS()
        await nc.connect(servers=[settings.ENDPOINT], token=settings.TOKEN.get_secret_value() if settings.TOKEN else None)
        return nc

FastAPI Dependency Injection

# packages/core/swiss_ai_hub/core/dependencies/use_nats.py
from fastapi import Request, WebSocket

def use_nats(request: Request) -> NATS:
    return request.app.state.nc

def use_nats_ws(request: WebSocket) -> NATS:
    return request.app.state.nc

Lifetime Manager (API Startup)

File: packages/api/swiss_ai_hub/api/runners/lifetime/lifetime_manager.py

Startup order:

1. MongoDB → Redis → Milvus → S3
2. NATS (NatsSettings.create_client()) → JetStream (nc.jetstream())
3. Event persisters (NCSubscriber for all agent + process events)
4. WebSocket infrastructure (NCSubscriber for display events)
5. Event distributors (ExternalAgentEventDistributor, ExternalProcessEventDistributor)
6. RPC responders (AgentConfigResponder, ProcessConfigResponder)
7. Discovery services
8. Database initialization

Shutdown: reverse order, NATS closed in finally block.

Environment Variables

NATS Server Config

Template: infra/deployment/templates/configs/nats-config.conf.j2

OpenTelemetry Tracing

All publishers and subscribers automatically propagate trace context via NATS headers.

Publisher Span Attributes

span.set_attribute("messaging.system", "nats.jetstream")  # or "nats"
span.set_attribute("messaging.destination", subject)
span.set_attribute("messaging.operation", "publish")
span.set_attribute("jetstream.sequence", ack.seq)       # JSPublisher only
span.set_attribute("jetstream.stream", ack.stream)       # JSPublisher only
span.set_attribute("jetstream.attempt", attempt)          # JSPublisher retry

Subscriber Span Attributes

span.set_attribute("messaging.system", "nats.jetstream")
span.set_attribute("messaging.source", msg.subject)
span.set_attribute("messaging.operation", "receive")
span.set_attribute("event.type", event.event_name)
span.set_attribute("jetstream.sequence", msg.metadata.sequence.stream)  # JSSubscriber
span.set_attribute("jetstream.acked", True)  # JSSubscriber

RPC Span Attributes

span.set_attribute("messaging.operation", "request")  # or "respond"
span.set_attribute("rpc.request_type", request.__class__.__name__)
span.set_attribute("rpc.response_type", response_cls.__name__)
span.set_attribute("rpc.success", True)

Trace Context Propagation

# Publishing — inject context
headers = NATSMessageHeaders().with_trace_context().to_dict()

# Subscribing — extract context
parent_context = NATSTraceContextPropagator.extract_and_activate_trace_context(msg.headers)
with tracer.start_as_current_span(..., context=parent_context):
    ...

Creating a New Event

Step 1: Define the Event Class

# packages/core/swiss_ai_hub/core/events/agent/my_feature/my_event.py
from swiss_ai_hub.core.events.agent.control.control_event import ControlEvent
from swiss_ai_hub.core.events.agent.display.display_event import DisplayEvent
from swiss_ai_hub.core.events.agent.control_and_display_event import ControlAndDisplayEvent

# Choose ONE base:
# - ControlEvent: workflow-driving only
# - DisplayEvent: UI/observability only
# - ControlAndDisplayEvent: both

class MyFeatureEvent(ControlAndDisplayEvent):
    """Signals that my feature completed."""

    _display_name: ClassVar[LocaleString] = LocaleString.from_i18n_path("lib.events.my_feature_event.name")
    _display_description: ClassVar[LocaleString] = LocaleString.from_i18n_path("lib.events.my_feature_event.description")

    result: str
    confidence: float

No registration needed — __pydantic_init_subclass__ auto-registers the class. Add the name and description under my_feature_event: in packages/core/swiss_ai_hub/core/i18n/translations/lib/events.{de,en,fr,it}.yml, export the class from swiss_ai_hub/core/events/agent/__init__.py (lazy __all__ + _LAZY_IMPORTS entries), and add a frontend component (/scaffold-event-display).

Choosing the right base class: See /scaffold-agent for a decision table on when to use ControlEvent vs ControlAndDisplayEvent vs DisplayEvent vs a semantic event.

Step 2: Use in Agent Step

@step()
async def my_step(self, ev: StartEvent) -> MyFeatureEvent:
    result = await do_work()
    return MyFeatureEvent(result=result, confidence=0.95)

Step 3: Subscribe to It

# The agent dispatcher handles this automatically for ControlEvents.
# For custom subscribers:
subscriber = NCSubscriber(
    name="MyFeatureHandler",
    nc=nc,
    subject="agent.*.*.*.*.*.control_event.MyFeatureEvent.*",
    event_cls=MyFeatureEvent,
    handler=my_handler,
)

Creating a New Publisher/Subscriber Pair

Custom Publisher

from swiss_ai_hub.core.publishers import JSPublisher
from swiss_ai_hub.core.publishers import NCPublisher

# For durable events:
js_pub = JSPublisher("MyServicePublisher", js)
await js_pub.ensure_stream_exists(stream_name, stream_subject)
await js_pub.publish_event(my_event, subject)

# For ephemeral events:
nc_pub = NCPublisher("MyServicePublisher", nc)
await nc_pub.publish_event(my_event, subject)

Custom Subscriber with Factory Pattern

class MyNCSubscriber(NCSubscriber[MyEvent]):
    @classmethod
    def for_all_my_events(
        cls,
        nc: NATS,
        topic_manager: MyTopicManager,
        handler: Callable[[MyEvent, MyTopic], Awaitable[None]],
        subscriber_name: str = "Unnamed",
    ):
        subject = topic_manager.get_subject_for_all_events()
        return cls(
            name=subscriber_name,
            nc=nc,
            subject=subject,
            event_cls=MyEvent,
            handler=handler,
        )

Key File Reference

Core Infrastructure

Publishers

Subscribers

RPC

Events

Topics & Streams

Dispatcher & Event Store

Tracing

Lifetime & Integration

Documentation

Formal Protocol Specification

The Swiss AI Agent Protocol is governed by formal dispatch rules and invariants.

Dispatch Rules

Invariants

• Fundamental Invariant: Steps declare data requirements, not execution order. Execution order is emergent from the event dependency graph.
• Race Condition Theorem: For a step with parameters (A, B?) where B is optional, arrival of A triggers immediate execution (R1). If B arrives later, R2 triggers re-execution. Use @precondition or ListOfSize to synchronize.
• List Parameter Theorem: A step s with parameter p: list[T] executes once for each event of type T that arrives. A list of length ≥ 1 satisfies the type constraint list[T]. Each subsequent arrival of type T creates a new state where the constraint is again satisfied with an updated list. Corollary: To execute exactly once after N events of type T, use FixedList(T, N) when N is compile-time constant, or a precondition checking len(list) >= expected_count when N is runtime-determined.
• Idempotency: The dispatcher skips re-execution if a step was already called with the exact same input events (tracked via StepStore.was_called_with_events()).

Formal Event Semantics

Control vs. Display Event Semantics:

An event e is a control event if ControlEvent ∈ ancestors(e). An event is a display event if DisplayEvent ∈ ancestors(e).

Dispatcher Trigger Rule: The dispatcher evaluates steps only when a control event is published. Display events are processed for UI/observability purposes but do not cause step re-evaluation.

Implication: For high-frequency updates (streaming chunks, progress indicators), use DisplayEvent to avoid dispatcher overhead. Reserve ControlEvent (or ControlAndDisplayEvent) for state transitions that should trigger downstream steps.

Events as Flow Carriers (Formal):

An event e is a pure flow carrier if fields(e) = ∅. The execution constraint R(s) ⊆ Eₜ depends only on event type existence, not event content. Therefore, a step may depend on an event type T without accessing any value from instances of T. The expression _: T in a step signature declares a flow dependency without data dependency.

Validity Conditions

A workflow W is valid iff:

1. Reachability: ∀s ∈ S, ∃ execution path from StartEvent to s
2. Termination: ∀ execution paths eventually reach StopEvent
3. No Stop Dependencies: ∀s ∈ S, StopEvent ∉ R(s)
4. Acyclicity: The event dependency graph is acyclic (bounded loops via RunContext are valid)

Source: packages/core/swiss_ai_hub/core/dispatcher/base_dispatcher.py, packages/agent/swiss_ai_hub/agent/dispatchers/agent_dispatcher.py

Conventions Checklist

• Events inherit from ControlEvent, DisplayEvent, or ControlAndDisplayEvent
• Event class names are unique (auto-registration enforced)
• Display events include _display_name and _display_description as ClassVar[LocaleString]
• Control events published via JSPublisher (durable)
• Display events published via NCPublisher (ephemeral)
• ControlAndDisplayEvent types published to both channels
• Never build NATS subjects manually — use TopicManager subclasses
• Publishers and subscribers have descriptive name (appears in OpenTelemetry spans)
• RPC handlers return response even on error (don't leave requester hanging)
• All handlers are async with signature (event: TEvent, topic: Topic) -> None
• Subscribers started in lifetime manager and stopped on shutdown
• Streams named {entity}_{class}_stream (e.g., agent_RAGAgent_stream)
• Queue groups used for JetStream subscribers that need load balancing