--- name: nats-events description: >- Reference for NATS messaging, JetStream, and the Swiss AI Agent Protocol. Use when user says 'how to publish an event', 'create a subscriber', 'NATS subject format', 'event hierarchy', 'add a new event type', 'JetStream consumer setup', 'RPC pattern', 'Control vs Display event', 'TopicManager usage', 'how events flow between services', or 'NATS connection config'. Covers events, pub/sub, RPC, topics, streams, dispatchers, and tracing. arguments: - name: topic description: Topic or question (e.g., "publish event", "RPC pattern", "event hierarchy", "JetStream consumer") --- # 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: | Tier | Protocol | Durability | Use Case | | ------------- | --------------------------------- | ----------------------------- | ------------------------------------------- | | **NATS Core** | `nc.publish()` / `nc.subscribe()` | Ephemeral (fire-and-forget) | Display events, discovery, real-time UI | | **JetStream** | `js.publish()` / `js.subscribe()` | Persistent (30-day retention) | Control events, workflow state, audit trail | **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: | Category | Base Class | Purpose | Transport | Failure Impact | | ----------------- | ------------------------ | -------------------------- | ------------- | ------------------- | | **Control Event** | `ControlEvent` | Drives workflow execution | JetStream | Breaks agent logic | | **Display Event** | `DisplayEvent` | Observability / UI updates | NATS Core | No workflow impact | | **Both** | `ControlAndDisplayEvent` | Workflow + user-visible | Both channels | Depends on consumer | **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) │ ├── 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) │ ├── 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](https://github.com/Arize-ai/openinference). This enables export to Langfuse, Arize Phoenix, and any OpenTelemetry-compatible system. | SemanticEvent Subclass | OpenInference Span Kind | Key Attributes | | ---------------------- | ----------------------- | ------------------------------------------- | | `LLMEvent` | `LLM` | Token counts, messages, model name | | `LLMStopEvent` | `LLM` (terminal) | Combines LLM data with workflow termination | | `RetrieverEvent` | `RETRIEVER` | Document IDs, content, scores | | `RerankerEvent` | `RERANKER` | Input/output nodes, reranking scores | | `EmbeddingEvent` | `EMBEDDING` | Vectors, model info, dimensions | | `ToolEvent` | `TOOL` | Tool name, parameters, result | | `GuardEvent` | `GUARDRAIL` | Guard result, accepted/rejected | | `ChainEvent` | `CHAIN` | Chain execution trace | | `AgentEvent` | `AGENT` | Agent invocation trace | ```python from swiss_ai_hub.core.nats.events.semantic 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: ```python 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`. ```python # 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 | If your event represents... | Inherit from | Benefits | | ------------------------------- | ------------------------------------------------------------- | ------------------------------------------- | | Workflow start condition | `StartEvent` | Recognized as entry point | | User message from a **chat UI** | `UserMessageEvent` | Chat history, locale, user identity | | Workflow termination | `StopEvent` | Signals completion | | Error/failure | `ExceptionEvent` | Error handling patterns | | LLM invocation result | `LLMEvent` | Token counts, messages, OpenInference spans | | LLM terminal response | `LLMStopEvent` | Combines LLM data with workflow termination | | Document retrieval | `RetrieverEvent` | Retrieved nodes, OpenInference spans | | Reranking operation | `RerankerEvent` | Input/output nodes, OpenInference spans | | Embedding generation | `EmbeddingEvent` | Vectors, model info, OpenInference spans | | Tool/function call | `ToolEvent` | Tool name, parameters, OpenInference spans | | Guardrail check | `GuardEvent` | Guard result, OpenInference spans | | Human approval needed | `HumanInTheLoopRequestEvent` | Workflow suspension, UI prompt | | Agent delegation | `AgentInTheLoopRequestEvent` | Cross-agent communication | | Memory retrieval | `RetrieveUserMemoryEvent` / `RetrieveOrganizationMemoryEvent` | Memory search results | | Memory storage | `StoreUserMemoryEvent` / `StoreOrganizationMemoryEvent` | Memory persistence confirmation | | Streaming text chunk | `ChunkEvent` | Real-time UI updates (display-only) | | Agent thought/reasoning | `ThoughtEvent` | Transparency display (display-only) | | Cost information | `LLMCostEvent` | Token usage, pricing (display-only) | | Generic workflow state | `ControlAndDisplayEvent` | Triggers dispatcher + UI display | | Generic UI update | `DisplayEvent` | UI-only, no dispatcher overhead | ### Extended Event Reference #### Guard Events | Event | Base | Purpose | | -------------------------------- | ------------ | ---------------------------- | | `GuardAcceptEvent` | `GuardEvent` | Guard passed | | `GuardRejectionEvent` | `GuardEvent` | Guard rejected | | `AgentSuitabilityAcceptEvent` | `GuardEvent` | Agent can handle request | | `AgentSuitabilityRejectEvent` | `GuardEvent` | Agent cannot handle request | | `ContextSufficientAcceptEvent` | `GuardEvent` | Sufficient context available | | `ContextInsufficientRejectEvent` | `GuardEvent` | Insufficient context | | `SensitiveInfoAcceptEvent` | `GuardEvent` | No sensitive info detected | | `SensitiveInfoRejectEvent` | `GuardEvent` | Sensitive info detected | #### Utility Events | Event | Base | Purpose | | ---------------------------------- | ------------------------ | ---------------------- | | `RouterEvent` | `ControlAndDisplayEvent` | LLM routing decision | | `LanguageEvent` | `ControlEvent` | Language detection | | `LimitChatHistoryEvent` | `ControlEvent` | Truncated history | | `StandaloneQuestionCondenserEvent` | `ControlEvent` | Question reformulation | #### Memory Events | Event | Base | Purpose | | --------------------------------- | ------------------------ | ---------------------------- | | `BaseRetrieveMemoryEvent` | `ControlAndDisplayEvent` | Memory retrieval base | | `RetrieveUserMemoryEvent` | above | User-scoped memory retrieval | | `RetrieveOrganizationMemoryEvent` | above | Org-scoped memory retrieval | | `BaseStoreMemoryEvent` | `ControlAndDisplayEvent` | Memory storage base | | `StoreUserMemoryEvent` | above | User-scoped memory storage | | `StoreOrganizationMemoryEvent` | above | Org-scoped memory storage | | `AddMemoryToChatHistoryEvent` | `ControlEvent` | Extended chat history | #### Interaction Events | Event | Base | Purpose | | ---------------------------------------- | ---------------------------- | ----------------------------------- | | `HumanInTheLoopInputRequestEvent` | `HumanInTheLoopRequestEvent` | Popup with text input field | | `HumanInTheLoopConfirmationRequestEvent` | `HumanInTheLoopRequestEvent` | Yes/No button selection | | `HumanInTheLoopChatRequestEvent` | `HumanInTheLoopRequestEvent` | Chat message (fallback) | | `HumanInTheLoopResponseEvent[T]` | `ControlAndDisplayEvent` | Human replied (generic T) | | `AgentInTheLoopRequestEvent` | `ControlAndDisplayEvent` | Delegate to another agent | | `AgentInTheLoopResponseEvent` | `ControlAndDisplayEvent` | Delegated agent result | | `AgentInTheLoopExceptionEvent` | `ControlAndDisplayEvent` | Delegated agent failure | | `BotInTheLoopRequestEvent` | `ControlEvent` | Send message to Teams/Slack channel | | `BotInTheLoopResponseEvent` | `ControlEvent` | Response from Teams/Slack user | **HITL Helper Classes** (not events, but workflow utilities): | Helper | UI Behavior | Response Type | | ---------------------------- | ----------------------------------------------------- | ------------- | | `HumanInTheLoopInput` | Popup dialog for free-form text entry | `str` | | `HumanInTheLoopConfirmation` | Yes/No button selection | `bool` | | `HumanInTheLoopChat` | Message in chat stream (fallback for simple UIs/APIs) | `str` | ### BaseEvent Core Fields ```python 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**: ```python 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 ``` | Segment | Example | Purpose | | ------------- | --------------------------------- | --------------------------------- | | `agent_class` | `RAGAgent` | Agent blueprint type | | `agent_id` | `wiki_agent` | Specific agent instance | | `thread_id` | `t948a201...` | Conversation context (ObjectId) | | `display_id` | `d135bfc9...` | UI display grouping (ObjectId) | | `run_id` | `r4fg68bb...` | Single execution trace (ObjectId) | | `event_type` | `control_event` / `display_event` | Event classification | | `event_name` | `ChunkEvent` | Event class name | | `event_id` | `e423...` | Unique event instance ID | ### 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) | Scope | Purpose | Access Control | | ----------- | ---------------------------------------------- | ------------------------------------------- | | **Thread** | Conversation context, long-lived (days/months) | Users granted access at thread level | | **Display** | UI grouping, can span multiple agents | Shared or isolated between delegated agents | | **Run** | Single execution (StartEvent → StopEvent) | Isolated per workflow invocation | **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 ```python # 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) ```python from swiss_ai_hub.core.nats.publishers.NCPublisher 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) ```python from swiss_ai_hub.core.nats.publishers.JSPublisher 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 ```python # 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 ```python from swiss_ai_hub.core.nats.tracing.NATSMessageHeaders 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) ```python from swiss_ai_hub.core.nats.subscribers.NCSubscriber 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) ```python from swiss_ai_hub.core.nats.subscribers.JSSubscriber 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: ```python from swiss_ai_hub.core.nats.subscribers.agent.AgentNCSubscriber import AgentNCSubscriber from swiss_ai_hub.core.nats.subscribers.agent.AgentJSSubscriber 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) ```python from swiss_ai_hub.core.nats.requester.NCRequester 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) ```python from swiss_ai_hub.core.nats.responder.NCResponder 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) ```python from swiss_ai_hub.core.nats.rpc.AgentConfigClient import AgentConfigClient client = AgentConfigClient(nc=nc, timeout_ms=5000) config = await client.fetch_config(agent_class="RAGAgent", agent_id="wiki") ``` ### RPC Models ```python # 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 ```python 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) ```python from swiss_ai_hub.core.nats.polling.JSPoller 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: ```python 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: | Subscriber | Subject | Handler | Purpose | | --------------------- | -------------------------------------------------- | ------------------------------- | ---------------------- | | `AgentEventPersister` | `agent.*.*.*.*.*.*.*.*` (all events) | `persister.persist_agent_event` | MongoDB audit log | | `WebSockets` | `agent.*.*.*.*.*.display_event.*.*` (display only) | `ws_sender.send_event` | Real-time UI streaming | ### SSE Streaming (OpenAI-compatible) The API creates **temporary per-request subscribers** for SSE streams: ```python # 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 ```python # packages/core/swiss_ai_hub/core/infrastructure/nats/NatsSettings.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 ```python # 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 | Variable | Default | Purpose | | --------------- | -------- | ------------------------------------------ | | `NATS_ENDPOINT` | Required | Server URL (e.g., `nats://localhost:4222`) | | `NATS_TOKEN` | Optional | Token authentication | ### NATS Server Config **Template**: `deployment/templates/configs/nats-config.conf.j2` | Setting | Dev | Prod | | ---------------------------- | ----- | ------ | | `max_payload` | 1MB | 2MB | | `max_connections` | 64 | 256 | | `max_subscriptions` | 1000 | 5000 | | `max_pending` | 512MB | 2GB | | JetStream `max_memory_store` | 512MB | 2GB | | JetStream `max_file_store` | 10GB | 50GB | | JetStream `sync_interval` | 1m | 2m | | JetStream `domain` | `dev` | `prod` | ______________________________________________________________________ ## OpenTelemetry Tracing All publishers and subscribers automatically propagate trace context via NATS headers. ### Publisher Span Attributes ```python 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 ```python 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 ```python 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 ```python # 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 ```python # 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] = from_i18n_path("events.my_feature.display_name") _display_description: ClassVar[LocaleString] = from_i18n_path("events.my_feature.description") result: str confidence: float ``` **No registration needed** — `__pydantic_init_subclass__` auto-registers the class. **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 ```python @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 ```python # 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 ```python from swiss_ai_hub.core.nats.publishers.JSPublisher import JSPublisher from swiss_ai_hub.core.nats.publishers.NCPublisher 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 ```python 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 | File | Purpose | | ---------------------------------------------------------------------- | ---------------------- | | `packages/core/swiss_ai_hub/core/infrastructure/nats/nats_settings.py` | NATS connection config | | `packages/core/swiss_ai_hub/core/dependencies/use_nats.py` | FastAPI DI | ### Publishers | File | Purpose | | ------------------------------------------------------------------ | -------------------- | | `packages/core/swiss_ai_hub/core/publishers/abstract_publisher.py` | Publisher base class | | `packages/core/swiss_ai_hub/core/publishers/nc_publisher.py` | NATS Core publisher | | `packages/core/swiss_ai_hub/core/publishers/js_publisher.py` | JetStream publisher | ### Subscribers | File | Purpose | | -------------------------------------------------------------------------- | ----------------------------- | | `packages/core/swiss_ai_hub/core/subscribers/abstract_subscriber.py` | Subscriber base class | | `packages/core/swiss_ai_hub/core/subscribers/nc_subscriber.py` | NATS Core subscriber | | `packages/core/swiss_ai_hub/core/subscribers/js_subscriber.py` | JetStream subscriber | | `packages/core/swiss_ai_hub/core/subscribers/agent/agent_nc_subscriber.py` | Agent NC subscriber factories | | `packages/core/swiss_ai_hub/core/subscribers/agent/agent_js_subscriber.py` | Agent JS subscriber factories | ### RPC | File | Purpose | | ------------------------------------------------------------- | --------------------------- | | `packages/core/swiss_ai_hub/core/requester/nc_requester.py` | RPC client | | `packages/core/swiss_ai_hub/core/responder/nc_responder.py` | RPC server | | `packages/core/swiss_ai_hub/core/rpc/agent_config_client.py` | Agent config RPC client | | `packages/core/swiss_ai_hub/core/rpc/models.py` | RPC request/response models | | `packages/api/swiss_ai_hub/api/rpc/agent_config_responder.py` | Agent config RPC server | ### Events | File | Purpose | | ----------------------------------------------------------------------------------- | --------------------------------------- | | `packages/core/swiss_ai_hub/core/events/base_event.py` | Event base + registry + deserialization | | `packages/core/swiss_ai_hub/core/events/agent/control/control_event.py` | Workflow event base | | `packages/core/swiss_ai_hub/core/events/agent/display/display_event.py` | UI event base | | `packages/core/swiss_ai_hub/core/events/agent/control_and_display_event.py` | Hybrid event base | | `packages/core/swiss_ai_hub/core/events/agent/control/start/start_event.py` | Run start | | `packages/core/swiss_ai_hub/core/events/agent/control/stop/stop_event.py` | Run stop | | `packages/core/swiss_ai_hub/core/events/agent/control/exception/exception_event.py` | Error | | `packages/core/swiss_ai_hub/core/events/agent/display/chunk_event.py` | Streaming text | | `packages/core/swiss_ai_hub/core/events/agent/display/thought_event.py` | Agent reasoning | | `packages/core/swiss_ai_hub/core/events/agent/user/` | User chat message events | ### Topics & Streams | File | Purpose | | ------------------------------------------------------------------------------ | --------------------- | | `packages/core/swiss_ai_hub/core/topics/topic.py` | Topic base + registry | | `packages/core/swiss_ai_hub/core/topics/agents/agent_instance_topic.py` | Full agent topic | | `packages/core/swiss_ai_hub/core/topic_managers/topic_manager.py` | Subject builder base | | `packages/core/swiss_ai_hub/core/topic_managers/agents/agent_topic_manager.py` | Agent subjects | | `packages/core/swiss_ai_hub/core/streams/stream_manager.py` | Stream creation | ### Dispatcher & Event Store | File | Purpose | | ----------------------------------------------------------------------------------- | ------------------ | | `packages/core/swiss_ai_hub/core/dispatcher/base_dispatcher.py` | Dispatcher base | | `packages/core/swiss_ai_hub/core/dispatcher/stores/event/jet_stream_event_store.py` | Event store | | `packages/core/swiss_ai_hub/core/polling/js_poller.py` | Pull consumer | | `packages/agent/swiss_ai_hub/agent/dispatchers/agent_dispatcher.py` | Agent dispatcher | | `packages/process/swiss_ai_hub/process/dispatchers/process_dispatcher.py` | Process dispatcher | ### Tracing | File | Purpose | | -------------------------------------------------------------------------- | --------------------- | | `packages/core/swiss_ai_hub/core/tracing/nats_message_headers.py` | Header builder | | `packages/core/swiss_ai_hub/core/tracing/nats_trace_context_propagator.py` | W3C trace propagation | ### Lifetime & Integration | File | Purpose | | --------------------------------------------------------------------------------- | ----------------------- | | `packages/api/swiss_ai_hub/api/runners/lifetime/lifetime_manager.py` | API startup/shutdown | | `packages/api/swiss_ai_hub/api/sockets/sender/web_socket_sender.py` | NATS → WebSocket bridge | | `packages/api/swiss_ai_hub/api/sockets/manager/web_socket_manager.py` | WebSocket connections | | `packages/core/swiss_ai_hub/core/distributor/external_agent_event_distributor.py` | API → NATS bridge | | `packages/agent/swiss_ai_hub/agent/runners/agent_runner.py` | Agent NATS bootstrap | ### Documentation | File | Purpose | | --------------------------------------------------------------------------- | --------------------------- | | `docs/docs/2_platform/2_architecture/3_swiss_ai_agent_protocol/index.en.md` | Protocol spec | | `deployment/templates/configs/nats-config.conf.j2` | NATS server config template | ______________________________________________________________________ ## Formal Protocol Specification The Swiss AI Agent Protocol is governed by formal dispatch rules and invariants. ### Dispatch Rules | Rule | Name | Statement | | ---- | ------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | R1 | Minimum Viable Input | A step executes the moment ALL required inputs are satisfied — not before, not after | | R2 | Re-execution on New Data | If a step receives a new event matching a parameter it already consumed, it executes again with the new data | | R3 | List Parameter Semantics | `list[EventType]` triggers on each new event arrival — a list of length 1 satisfies `list[T]`, causing re-execution on each new event. Use `FixedList(E, N)` for deterministic fan-in | | R4 | StopEvent Constraint | `StopEvent` MUST be the last event emitted by a run — nothing may follow it | | R5 | Precondition Override | `@precondition` adds a callable guard checked AFTER R1 is satisfied — can delay or block execution | | R6 | Event Persistence | Every `ControlEvent` is persisted in JetStream and replayed on dispatcher restart | ### 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