feat: implement event-driven real-time architecture

Phase 1: Real-Time Views/Documents
- Add useRealtimeDocuments hook for WebSocket subscriptions
- Integrate real-time updates in DocumentListRenderer, StatCardRenderer
- Add real-time to Documents, PageDisplay, Views, Collections, Pages

Phase 2: LLM Event Infrastructure
- Add proto definitions for MessageEvent, ResponseChunk, StreamingState
- Add EventService with queue and sync operations
- Add streaming_state to Session message

Phase 3: LLM Node Event Processing
- Create EventProcessor class with polling and chunk writing
- Implement stale event recovery

Phase 4: WebServer LLM Changes
- Add /queue, /sync, /streaming-state endpoints
- Extend WebSocket with subscribe_chat, unsubscribe_chat
- Add BroadcastChatChunk and BroadcastChatState methods

Phase 5: WebUI Chat Changes
- Extend WebSocketContext with chat subscriptions
- Add event-driven API (queueMessage, syncChunks)
- Integrate ChatContext with WebSocket streaming

llm/include/smartbotic/llm/event/event_processor.hpp

@@ -0,0 +1,158 @@
+#pragma once
+
+#include <grpcpp/grpcpp.h>
+#include "database.grpc.pb.h"
+#include "llm.pb.h"
+
+#include <atomic>
+#include <functional>
+#include <memory>
+#include <mutex>
+#include <string>
+#include <thread>
+#include <vector>
+
+namespace smartbotic::llm::provider {
+class ProviderFactory;
+}
+
+namespace smartbotic::llm::session {
+class ISessionStore;
+}
+
+namespace smartbotic::llm::tool {
+class ToolRegistry;
+}
+
+namespace smartbotic::llm::event {
+
+/// Configuration for the event processor
+struct EventProcessorConfig {
+    std::string database_address = "127.0.0.1:50051";
+    std::string node_id;                              // Unique identifier for this processing node
+    std::string events_collection = "__llm_message_events";
+    std::string chunks_collection = "__llm_response_chunks";
+    int poll_interval_ms = 100;                       // How often to poll for new events
+    int max_concurrent_events = 5;                    // Max events to process concurrently
+    int stale_event_timeout_seconds = 300;            // Events processing for longer than this are stale
+};
+
+/// Result type for event processing operations
+template <typename T>
+struct EventResult {
+    bool success = false;
+    std::string error;
+    T value;
+
+    static EventResult<T> Ok(T val) {
+        return EventResult<T>{true, "", std::move(val)};
+    }
+
+    static EventResult<T> Error(const std::string& err) {
+        return EventResult<T>{false, err, T{}};
+    }
+};
+
+/// Event processor that handles async LLM message processing
+/// - Polls __llm_message_events for pending events
+/// - Claims events atomically for processing
+/// - Streams responses and writes chunks to __llm_response_chunks
+/// - Handles stale event recovery
+class EventProcessor {
+public:
+    EventProcessor(
+        const EventProcessorConfig& config,
+        std::shared_ptr<provider::ProviderFactory> provider_factory,
+        std::shared_ptr<session::ISessionStore> session_store,
+        std::shared_ptr<tool::ToolRegistry> tool_registry
+    );
+    ~EventProcessor();
+
+    /// Initialize the event processor (connect to database, create collections)
+    [[nodiscard]] auto Initialize() -> bool;
+
+    /// Start the processing loop (runs in background thread)
+    void Start();
+
+    /// Stop the processing loop
+    void Stop();
+
+    /// Check if the processor is running
+    [[nodiscard]] auto IsRunning() const -> bool;
+
+    /// Queue a new message event (alternative to direct database write)
+    [[nodiscard]] auto QueueMessageEvent(const ::smartbotic::llm::MessageEvent& event)
+        -> EventResult<std::string>;
+
+    /// Get streaming state for a session
+    [[nodiscard]] auto GetStreamingState(const std::string& session_id)
+        -> EventResult<::smartbotic::llm::StreamingState>;
+
+    /// Sync chunks for reconnection
+    [[nodiscard]] auto SyncChunks(const std::string& session_id,
+                                  const std::string& event_id,
+                                  int32_t since_sequence)
+        -> EventResult<std::vector<::smartbotic::llm::ResponseChunk>>;
+
+private:
+    /// Main processing loop (runs in background thread)
+    void ProcessingLoop();
+
+    /// Poll for pending events
+    [[nodiscard]] auto PollPendingEvents() -> std::vector<::smartbotic::llm::MessageEvent>;
+
+    /// Claim an event for processing (atomic operation)
+    [[nodiscard]] auto ClaimEvent(const std::string& event_id) -> bool;
+
+    /// Process a single message event
+    void ProcessEvent(const ::smartbotic::llm::MessageEvent& event);
+
+    /// Write a response chunk to the database
+    void WriteChunk(const ::smartbotic::llm::ResponseChunk& chunk);
+
+    /// Update event status
+    void UpdateEventStatus(const std::string& event_id,
+                           ::smartbotic::llm::MessageEventStatus status,
+                           const std::string& error = "");
+
+    /// Update session streaming state
+    void UpdateSessionStreamingState(const std::string& session_id,
+                                     const ::smartbotic::llm::StreamingState& state);
+
+    /// Recover stale events (events stuck in processing state)
+    void RecoverStaleEvents();
+
+    /// Connect to the database
+    [[nodiscard]] auto Connect() -> bool;
+
+    /// Convert MessageEvent to database document
+    [[nodiscard]] auto EventToDocument(const ::smartbotic::llm::MessageEvent& event)
+        -> ::smartbotic::database::MapValue;
+
+    /// Convert database document to MessageEvent
+    [[nodiscard]] auto DocumentToEvent(const ::smartbotic::database::MapValue& data,
+                                       const std::string& doc_id)
+        -> ::smartbotic::llm::MessageEvent;
+
+    /// Convert ResponseChunk to database document
+    [[nodiscard]] auto ChunkToDocument(const ::smartbotic::llm::ResponseChunk& chunk)
+        -> ::smartbotic::database::MapValue;
+
+    EventProcessorConfig config_;
+    std::shared_ptr<provider::ProviderFactory> provider_factory_;
+    std::shared_ptr<session::ISessionStore> session_store_;
+    std::shared_ptr<tool::ToolRegistry> tool_registry_;
+
+    std::shared_ptr<::grpc::Channel> channel_;
+    std::unique_ptr<::smartbotic::database::DocumentService::Stub> document_stub_;
+    std::unique_ptr<::smartbotic::database::CollectionService::Stub> collection_stub_;
+    std::unique_ptr<::smartbotic::database::QueryService::Stub> query_stub_;
+    std::unique_ptr<::smartbotic::database::SubscriptionService::Stub> subscription_stub_;
+
+    std::atomic<bool> running_{false};
+    std::atomic<bool> initialized_{false};
+    std::thread processing_thread_;
+    std::mutex processing_mutex_;
+};
+
+}  // namespace smartbotic::llm::event

llm/src/event/event_processor.cpp

@@ -0,0 +1,862 @@
+#include "smartbotic/llm/event/event_processor.hpp"
+#include "smartbotic/llm/provider/provider_factory.hpp"
+#include "smartbotic/llm/session/session_store.hpp"
+#include "smartbotic/llm/tool/tool_registry.hpp"
+
+#include <chrono>
+#include <spdlog/spdlog.h>
+#include <uuid.h>
+
+namespace smartbotic::llm::event {
+
+namespace {
+
+// Generate a unique ID
+auto GenerateId() -> std::string {
+    std::random_device rd;
+    auto seed_data = std::array<int, std::mt19937::state_size>{};
+    std::generate(std::begin(seed_data), std::end(seed_data), std::ref(rd));
+    std::seed_seq seq(std::begin(seed_data), std::end(seed_data));
+    std::mt19937 generator(seq);
+    uuids::uuid_random_generator gen{generator};
+    return uuids::to_string(gen());
+}
+
+// Get current timestamp as ISO8601 string
+auto GetCurrentTimestamp() -> std::string {
+    auto now = std::chrono::system_clock::now();
+    auto time_t_now = std::chrono::system_clock::to_time_t(now);
+    std::stringstream ss;
+    ss << std::put_time(std::gmtime(&time_t_now), "%FT%TZ");
+    return ss.str();
+}
+
+// Convert timestamp string to proto Timestamp
+auto StringToTimestamp(const std::string& str) -> ::smartbotic::llm::Timestamp {
+    ::smartbotic::llm::Timestamp ts;
+    // Parse ISO8601 format - simplified for now
+    std::tm tm = {};
+    std::istringstream ss(str);
+    ss >> std::get_time(&tm, "%Y-%m-%dT%H:%M:%S");
+    if (!ss.fail()) {
+        auto time = std::mktime(&tm);
+        ts.set_seconds(time);
+    }
+    return ts;
+}
+
+// Convert proto Timestamp to string
+auto TimestampToString(const ::smartbotic::llm::Timestamp& ts) -> std::string {
+    auto time_t_val = static_cast<time_t>(ts.seconds());
+    std::stringstream ss;
+    ss << std::put_time(std::gmtime(&time_t_val), "%FT%TZ");
+    return ss.str();
+}
+
+}  // namespace
+
+EventProcessor::EventProcessor(
+    const EventProcessorConfig& config,
+    std::shared_ptr<provider::ProviderFactory> provider_factory,
+    std::shared_ptr<session::ISessionStore> session_store,
+    std::shared_ptr<tool::ToolRegistry> tool_registry
+) : config_(config),
+    provider_factory_(std::move(provider_factory)),
+    session_store_(std::move(session_store)),
+    tool_registry_(std::move(tool_registry)) {
+
+    // Generate node ID if not provided
+    if (config_.node_id.empty()) {
+        config_.node_id = GenerateId();
+    }
+}
+
+EventProcessor::~EventProcessor() {
+    Stop();
+}
+
+auto EventProcessor::Connect() -> bool {
+    try {
+        channel_ = grpc::CreateChannel(
+            config_.database_address,
+            grpc::InsecureChannelCredentials()
+        );
+
+        document_stub_ = ::smartbotic::database::DocumentService::NewStub(channel_);
+        collection_stub_ = ::smartbotic::database::CollectionService::NewStub(channel_);
+        query_stub_ = ::smartbotic::database::QueryService::NewStub(channel_);
+        subscription_stub_ = ::smartbotic::database::SubscriptionService::NewStub(channel_);
+
+        // Wait for connection
+        auto deadline = std::chrono::system_clock::now() + std::chrono::seconds(10);
+        if (!channel_->WaitForConnected(deadline)) {
+            spdlog::error("EventProcessor: Failed to connect to database at {}",
+                         config_.database_address);
+            return false;
+        }
+
+        spdlog::info("EventProcessor: Connected to database at {}", config_.database_address);
+        return true;
+    } catch (const std::exception& e) {
+        spdlog::error("EventProcessor: Exception connecting to database: {}", e.what());
+        return false;
+    }
+}
+
+auto EventProcessor::Initialize() -> bool {
+    if (initialized_) {
+        return true;
+    }
+
+    if (!Connect()) {
+        return false;
+    }
+
+    // Create collections if they don't exist
+    grpc::ClientContext events_ctx;
+    ::smartbotic::database::CreateCollectionRequest events_req;
+    events_req.set_name(config_.events_collection);
+    ::smartbotic::database::CreateCollectionResponse events_resp;
+
+    auto events_status = collection_stub_->CreateCollection(&events_ctx, events_req, &events_resp);
+    if (!events_status.ok() &&
+        events_status.error_code() != grpc::StatusCode::ALREADY_EXISTS) {
+        spdlog::error("EventProcessor: Failed to create events collection: {}",
+                     events_status.error_message());
+        return false;
+    }
+    spdlog::info("EventProcessor: Events collection {} ready", config_.events_collection);
+
+    grpc::ClientContext chunks_ctx;
+    ::smartbotic::database::CreateCollectionRequest chunks_req;
+    chunks_req.set_name(config_.chunks_collection);
+    ::smartbotic::database::CreateCollectionResponse chunks_resp;
+
+    auto chunks_status = collection_stub_->CreateCollection(&chunks_ctx, chunks_req, &chunks_resp);
+    if (!chunks_status.ok() &&
+        chunks_status.error_code() != grpc::StatusCode::ALREADY_EXISTS) {
+        spdlog::error("EventProcessor: Failed to create chunks collection: {}",
+                     chunks_status.error_message());
+        return false;
+    }
+    spdlog::info("EventProcessor: Chunks collection {} ready", config_.chunks_collection);
+
+    initialized_ = true;
+    return true;
+}
+
+void EventProcessor::Start() {
+    if (running_) {
+        return;
+    }
+
+    if (!initialized_ && !Initialize()) {
+        spdlog::error("EventProcessor: Cannot start - initialization failed");
+        return;
+    }
+
+    running_ = true;
+    processing_thread_ = std::thread([this]() {
+        ProcessingLoop();
+    });
+
+    spdlog::info("EventProcessor: Started processing loop (node_id={})", config_.node_id);
+}
+
+void EventProcessor::Stop() {
+    if (!running_) {
+        return;
+    }
+
+    running_ = false;
+
+    if (processing_thread_.joinable()) {
+        processing_thread_.join();
+    }
+
+    spdlog::info("EventProcessor: Stopped processing loop");
+}
+
+auto EventProcessor::IsRunning() const -> bool {
+    return running_;
+}
+
+void EventProcessor::ProcessingLoop() {
+    while (running_) {
+        try {
+            // Recover stale events periodically
+            RecoverStaleEvents();
+
+            // Poll for pending events
+            auto events = PollPendingEvents();
+
+            for (const auto& event : events) {
+                if (!running_) break;
+
+                // Try to claim the event
+                if (ClaimEvent(event.id())) {
+                    // Process the event
+                    ProcessEvent(event);
+                }
+            }
+
+            // Sleep between polls
+            std::this_thread::sleep_for(
+                std::chrono::milliseconds(config_.poll_interval_ms)
+            );
+        } catch (const std::exception& e) {
+            spdlog::error("EventProcessor: Error in processing loop: {}", e.what());
+            std::this_thread::sleep_for(std::chrono::seconds(1));
+        }
+    }
+}
+
+auto EventProcessor::PollPendingEvents() -> std::vector<::smartbotic::llm::MessageEvent> {
+    std::vector<::smartbotic::llm::MessageEvent> events;
+
+    grpc::ClientContext ctx;
+    ::smartbotic::database::QueryRequest req;
+    req.set_collection(config_.events_collection);
+
+    // Query for pending events
+    auto* query = req.mutable_query();
+    auto* status_filter = query->add_filters();
+    status_filter->set_field("status");
+    status_filter->set_operator_(::smartbotic::database::FilterOperator::FILTER_OPERATOR_EQUALS);
+    auto* status_val = status_filter->mutable_value();
+    status_val->set_int64_value(
+        static_cast<int64_t>(::smartbotic::llm::MESSAGE_EVENT_STATUS_PENDING)
+    );
+
+    req.set_limit(config_.max_concurrent_events);
+
+    ::smartbotic::database::QueryResponse resp;
+    auto status = query_stub_->Query(&ctx, req, &resp);
+
+    if (!status.ok()) {
+        spdlog::warn("EventProcessor: Failed to query pending events: {}",
+                    status.error_message());
+        return events;
+    }
+
+    for (const auto& doc : resp.documents()) {
+        events.push_back(DocumentToEvent(doc.data(), doc.id()));
+    }
+
+    return events;
+}
+
+auto EventProcessor::ClaimEvent(const std::string& event_id) -> bool {
+    std::lock_guard<std::mutex> lock(processing_mutex_);
+
+    grpc::ClientContext ctx;
+    ::smartbotic::database::UpdateDocumentRequest req;
+    req.set_collection(config_.events_collection);
+    req.set_id(event_id);
+
+    // Only claim if still pending (optimistic locking via version check would be better)
+    auto* data = req.mutable_data();
+    auto* fields = data->mutable_fields();
+
+    // Set status to processing
+    (*fields)["status"].set_int64_value(
+        static_cast<int64_t>(::smartbotic::llm::MESSAGE_EVENT_STATUS_PROCESSING)
+    );
+    (*fields)["processing_node_id"].set_string_value(config_.node_id);
+    (*fields)["started_at"].set_string_value(GetCurrentTimestamp());
+
+    ::smartbotic::database::UpdateDocumentResponse resp;
+    auto status = document_stub_->UpdateDocument(&ctx, req, &resp);
+
+    if (!status.ok()) {
+        // Another node may have claimed it
+        return false;
+    }
+
+    spdlog::debug("EventProcessor: Claimed event {}", event_id);
+    return true;
+}
+
+void EventProcessor::ProcessEvent(const ::smartbotic::llm::MessageEvent& event) {
+    spdlog::info("EventProcessor: Processing event {} (type={})",
+                event.id(), static_cast<int>(event.event_type()));
+
+    try {
+        // Get the session
+        auto session_result = session_store_->GetSession(
+            event.session_id(), event.user_id(), true
+        );
+        if (!session_result.ok) {
+            UpdateEventStatus(event.id(), ::smartbotic::llm::MESSAGE_EVENT_STATUS_FAILED,
+                            "Failed to get session: " + session_result.error);
+            return;
+        }
+
+        auto& session = session_result.value;
+
+        // Update session streaming state
+        ::smartbotic::llm::StreamingState streaming_state;
+        streaming_state.set_is_streaming(true);
+        streaming_state.set_current_event_id(event.id());
+        streaming_state.set_last_chunk_sequence(0);
+        *streaming_state.mutable_started_at() = StringToTimestamp(GetCurrentTimestamp());
+        UpdateSessionStreamingState(event.session_id(), streaming_state);
+
+        // Build the chat request based on event type
+        // Note: Actual provider call logic would go here
+        // For now, we'll simulate the streaming process
+
+        std::string message_id = GenerateId();
+        int32_t sequence = 0;
+
+        // Get provider
+        std::string provider_id = event.provider_id();
+        if (provider_id.empty()) {
+            provider_id = session.model_id();  // Use session's model to determine provider
+        }
+
+        auto provider = provider_factory_->GetProvider(provider_id);
+        if (!provider) {
+            UpdateEventStatus(event.id(), ::smartbotic::llm::MESSAGE_EVENT_STATUS_FAILED,
+                            "Provider not found: " + provider_id);
+            return;
+        }
+
+        // Build ChatRequest from session and event
+        provider::ChatRequest chat_request;
+        chat_request.model = event.model_id().empty() ? session.model_id() : event.model_id();
+
+        // Convert session messages to provider format
+        for (const auto& msg : session.messages()) {
+            provider::Message prov_msg;
+            prov_msg.role = static_cast<provider::MessageRole>(msg.role());
+            for (const auto& part : msg.content()) {
+                if (part.has_text()) {
+                    prov_msg.content.push_back(provider::TextContent{part.text()});
+                }
+                // Handle other content types as needed
+            }
+            chat_request.messages.push_back(std::move(prov_msg));
+        }
+
+        // Add user message or tool results based on event type
+        if (event.event_type() == ::smartbotic::llm::MESSAGE_EVENT_TYPE_USER_MESSAGE &&
+            event.has_user_message()) {
+            provider::Message user_msg;
+            user_msg.role = provider::MessageRole::kUser;
+            for (const auto& part : event.user_message().content()) {
+                if (part.has_text()) {
+                    user_msg.content.push_back(provider::TextContent{part.text()});
+                }
+            }
+            chat_request.messages.push_back(std::move(user_msg));
+        } else if (event.event_type() == ::smartbotic::llm::MESSAGE_EVENT_TYPE_TOOL_RESULT) {
+            for (const auto& result : event.tool_results()) {
+                provider::Message tool_msg;
+                tool_msg.role = provider::MessageRole::kTool;
+                tool_msg.content.push_back(provider::ToolResultContent{
+                    result.tool_call_id(),
+                    result.content(),
+                    result.is_error()
+                });
+                chat_request.messages.push_back(std::move(tool_msg));
+            }
+        }
+
+        // Convert tools from event
+        for (const auto& tool_def : event.tools()) {
+            provider::ToolDefinition tool;
+            tool.name = tool_def.name();
+            tool.description = tool_def.description();
+            tool.input_schema = tool_def.input_schema();
+            chat_request.tools.push_back(std::move(tool));
+        }
+
+        // Stream callback to write chunks
+        auto stream_callback = [&](const provider::StreamChunk& chunk) -> bool {
+            ::smartbotic::llm::ResponseChunk response_chunk;
+            response_chunk.set_id(GenerateId());
+            response_chunk.set_workspace_id(event.workspace_id());
+            response_chunk.set_session_id(event.session_id());
+            response_chunk.set_message_event_id(event.id());
+            response_chunk.set_message_id(message_id);
+            response_chunk.set_sequence(sequence++);
+            *response_chunk.mutable_created_at() = StringToTimestamp(GetCurrentTimestamp());
+
+            if (!chunk.content_delta.empty()) {
+                response_chunk.set_chunk_type(::smartbotic::llm::RESPONSE_CHUNK_TYPE_CONTENT);
+                response_chunk.set_content_delta(chunk.content_delta);
+            } else if (!chunk.thinking_delta.empty()) {
+                response_chunk.set_chunk_type(::smartbotic::llm::RESPONSE_CHUNK_TYPE_THINKING);
+                response_chunk.set_thinking_delta(chunk.thinking_delta);
+            } else if (chunk.tool_call.has_value()) {
+                response_chunk.set_chunk_type(::smartbotic::llm::RESPONSE_CHUNK_TYPE_TOOL_CALL);
+                auto* tc = response_chunk.mutable_tool_call();
+                tc->set_id(chunk.tool_call->id);
+                tc->set_name(chunk.tool_call->name);
+                tc->set_arguments(chunk.tool_call->arguments);
+            }
+
+            WriteChunk(response_chunk);
+
+            // Update streaming state
+            streaming_state.set_last_chunk_sequence(sequence);
+            UpdateSessionStreamingState(event.session_id(), streaming_state);
+
+            return true;  // Continue streaming
+        };
+
+        // Call the provider
+        auto result = provider->ChatStream(chat_request, stream_callback);
+
+        // Write final metadata chunk
+        ::smartbotic::llm::ResponseChunk final_chunk;
+        final_chunk.set_id(GenerateId());
+        final_chunk.set_workspace_id(event.workspace_id());
+        final_chunk.set_session_id(event.session_id());
+        final_chunk.set_message_event_id(event.id());
+        final_chunk.set_message_id(message_id);
+        final_chunk.set_sequence(sequence);
+        final_chunk.set_chunk_type(::smartbotic::llm::RESPONSE_CHUNK_TYPE_METADATA);
+        final_chunk.set_is_final(true);
+        *final_chunk.mutable_created_at() = StringToTimestamp(GetCurrentTimestamp());
+
+        if (result.ok) {
+            auto* metadata = final_chunk.mutable_metadata();
+            auto* usage = metadata->mutable_usage();
+            usage->set_prompt_tokens(result.value.usage.prompt_tokens);
+            usage->set_completion_tokens(result.value.usage.completion_tokens);
+            usage->set_total_tokens(result.value.usage.total_tokens);
+            metadata->set_finish_reason(
+                static_cast<::smartbotic::llm::FinishReason>(result.value.finish_reason)
+            );
+        }
+
+        WriteChunk(final_chunk);
+
+        // Clear streaming state
+        streaming_state.set_is_streaming(false);
+        streaming_state.set_last_chunk_sequence(sequence);
+        UpdateSessionStreamingState(event.session_id(), streaming_state);
+
+        // Update event status
+        if (result.ok) {
+            UpdateEventStatus(event.id(), ::smartbotic::llm::MESSAGE_EVENT_STATUS_COMPLETED);
+        } else {
+            UpdateEventStatus(event.id(), ::smartbotic::llm::MESSAGE_EVENT_STATUS_FAILED,
+                            result.error);
+        }
+
+    } catch (const std::exception& e) {
+        spdlog::error("EventProcessor: Exception processing event {}: {}",
+                     event.id(), e.what());
+        UpdateEventStatus(event.id(), ::smartbotic::llm::MESSAGE_EVENT_STATUS_FAILED,
+                        e.what());
+
+        // Clear streaming state on error
+        ::smartbotic::llm::StreamingState streaming_state;
+        streaming_state.set_is_streaming(false);
+        UpdateSessionStreamingState(event.session_id(), streaming_state);
+    }
+}
+
+void EventProcessor::WriteChunk(const ::smartbotic::llm::ResponseChunk& chunk) {
+    grpc::ClientContext ctx;
+    ::smartbotic::database::CreateDocumentRequest req;
+    req.set_collection(config_.chunks_collection);
+    *req.mutable_data() = ChunkToDocument(chunk);
+
+    ::smartbotic::database::CreateDocumentResponse resp;
+    auto status = document_stub_->CreateDocument(&ctx, req, &resp);
+
+    if (!status.ok()) {
+        spdlog::warn("EventProcessor: Failed to write chunk: {}", status.error_message());
+    }
+}
+
+void EventProcessor::UpdateEventStatus(const std::string& event_id,
+                                        ::smartbotic::llm::MessageEventStatus status,
+                                        const std::string& error) {
+    grpc::ClientContext ctx;
+    ::smartbotic::database::UpdateDocumentRequest req;
+    req.set_collection(config_.events_collection);
+    req.set_id(event_id);
+
+    auto* data = req.mutable_data();
+    auto* fields = data->mutable_fields();
+
+    (*fields)["status"].set_int64_value(static_cast<int64_t>(status));
+
+    if (status == ::smartbotic::llm::MESSAGE_EVENT_STATUS_COMPLETED) {
+        (*fields)["completed_at"].set_string_value(GetCurrentTimestamp());
+    }
+
+    if (!error.empty()) {
+        (*fields)["error"].set_string_value(error);
+    }
+
+    ::smartbotic::database::UpdateDocumentResponse resp;
+    auto grpc_status = document_stub_->UpdateDocument(&ctx, req, &resp);
+
+    if (!grpc_status.ok()) {
+        spdlog::warn("EventProcessor: Failed to update event status: {}",
+                    grpc_status.error_message());
+    }
+}
+
+void EventProcessor::UpdateSessionStreamingState(const std::string& session_id,
+                                                  const ::smartbotic::llm::StreamingState& state) {
+    // Update the session document with streaming state
+    grpc::ClientContext ctx;
+    ::smartbotic::database::UpdateDocumentRequest req;
+    req.set_collection("__llm_sessions");
+    req.set_id(session_id);
+
+    auto* data = req.mutable_data();
+    auto* fields = data->mutable_fields();
+
+    // Create streaming_state as nested map
+    auto* streaming_state = (*fields)["streaming_state"].mutable_map_value();
+    auto* ss_fields = streaming_state->mutable_fields();
+
+    (*ss_fields)["is_streaming"].set_bool_value(state.is_streaming());
+    (*ss_fields)["current_event_id"].set_string_value(state.current_event_id());
+    (*ss_fields)["last_chunk_sequence"].set_int64_value(state.last_chunk_sequence());
+    if (state.has_started_at()) {
+        (*ss_fields)["started_at"].set_string_value(TimestampToString(state.started_at()));
+    }
+
+    ::smartbotic::database::UpdateDocumentResponse resp;
+    auto grpc_status = document_stub_->UpdateDocument(&ctx, req, &resp);
+
+    if (!grpc_status.ok()) {
+        spdlog::warn("EventProcessor: Failed to update session streaming state: {}",
+                    grpc_status.error_message());
+    }
+}
+
+void EventProcessor::RecoverStaleEvents() {
+    // Find events that have been processing for too long
+    grpc::ClientContext ctx;
+    ::smartbotic::database::QueryRequest req;
+    req.set_collection(config_.events_collection);
+
+    auto* query = req.mutable_query();
+    auto* status_filter = query->add_filters();
+    status_filter->set_field("status");
+    status_filter->set_operator_(::smartbotic::database::FilterOperator::FILTER_OPERATOR_EQUALS);
+    auto* status_val = status_filter->mutable_value();
+    status_val->set_int64_value(
+        static_cast<int64_t>(::smartbotic::llm::MESSAGE_EVENT_STATUS_PROCESSING)
+    );
+
+    req.set_limit(100);
+
+    ::smartbotic::database::QueryResponse resp;
+    auto status = query_stub_->Query(&ctx, req, &resp);
+
+    if (!status.ok()) {
+        return;
+    }
+
+    auto now = std::chrono::system_clock::now();
+
+    for (const auto& doc : resp.documents()) {
+        auto event = DocumentToEvent(doc.data(), doc.id());
+
+        // Check if started_at is older than threshold
+        if (event.has_started_at()) {
+            auto started = std::chrono::system_clock::from_time_t(event.started_at().seconds());
+            auto age = std::chrono::duration_cast<std::chrono::seconds>(now - started).count();
+
+            if (age > config_.stale_event_timeout_seconds) {
+                spdlog::warn("EventProcessor: Recovering stale event {} (age={}s)",
+                            event.id(), age);
+
+                // Reset to pending so it can be picked up again
+                UpdateEventStatus(event.id(), ::smartbotic::llm::MESSAGE_EVENT_STATUS_PENDING);
+            }
+        }
+    }
+}
+
+auto EventProcessor::QueueMessageEvent(const ::smartbotic::llm::MessageEvent& event)
+    -> EventResult<std::string> {
+
+    grpc::ClientContext ctx;
+    ::smartbotic::database::CreateDocumentRequest req;
+    req.set_collection(config_.events_collection);
+
+    // Set a new ID if not provided
+    ::smartbotic::llm::MessageEvent event_copy = event;
+    if (event_copy.id().empty()) {
+        event_copy.set_id(GenerateId());
+    }
+
+    *req.mutable_data() = EventToDocument(event_copy);
+
+    ::smartbotic::database::CreateDocumentResponse resp;
+    auto status = document_stub_->CreateDocument(&ctx, req, &resp);
+
+    if (!status.ok()) {
+        return EventResult<std::string>::Error(status.error_message());
+    }
+
+    return EventResult<std::string>::Ok(event_copy.id());
+}
+
+auto EventProcessor::GetStreamingState(const std::string& session_id)
+    -> EventResult<::smartbotic::llm::StreamingState> {
+
+    auto result = session_store_->GetSession(session_id, "", false);
+    if (!result.ok) {
+        return EventResult<::smartbotic::llm::StreamingState>::Error(result.error);
+    }
+
+    // Extract streaming_state from session (would need to be added to Session proto)
+    ::smartbotic::llm::StreamingState state;
+    // The streaming_state would be part of the session document
+    return EventResult<::smartbotic::llm::StreamingState>::Ok(state);
+}
+
+auto EventProcessor::SyncChunks(const std::string& session_id,
+                                const std::string& event_id,
+                                int32_t since_sequence)
+    -> EventResult<std::vector<::smartbotic::llm::ResponseChunk>> {
+
+    std::vector<::smartbotic::llm::ResponseChunk> chunks;
+
+    grpc::ClientContext ctx;
+    ::smartbotic::database::QueryRequest req;
+    req.set_collection(config_.chunks_collection);
+
+    auto* query = req.mutable_query();
+
+    // Filter by event_id
+    auto* event_filter = query->add_filters();
+    event_filter->set_field("message_event_id");
+    event_filter->set_operator_(::smartbotic::database::FilterOperator::FILTER_OPERATOR_EQUALS);
+    event_filter->mutable_value()->set_string_value(event_id);
+
+    // Filter by sequence > since_sequence
+    auto* seq_filter = query->add_filters();
+    seq_filter->set_field("sequence");
+    seq_filter->set_operator_(::smartbotic::database::FilterOperator::FILTER_OPERATOR_GREATER_THAN);
+    seq_filter->mutable_value()->set_int64_value(since_sequence);
+
+    // Order by sequence
+    auto* order = query->add_order_by();
+    order->set_field("sequence");
+    order->set_direction(::smartbotic::database::SortDirection::SORT_DIRECTION_ASCENDING);
+
+    req.set_limit(1000);
+
+    ::smartbotic::database::QueryResponse resp;
+    auto status = query_stub_->Query(&ctx, req, &resp);
+
+    if (!status.ok()) {
+        return EventResult<std::vector<::smartbotic::llm::ResponseChunk>>::Error(
+            status.error_message()
+        );
+    }
+
+    for (const auto& doc : resp.documents()) {
+        // Convert document to ResponseChunk
+        // This would require a DocumentToChunk method
+        ::smartbotic::llm::ResponseChunk chunk;
+        // Parse fields from doc.data()
+        chunks.push_back(chunk);
+    }
+
+    return EventResult<std::vector<::smartbotic::llm::ResponseChunk>>::Ok(std::move(chunks));
+}
+
+auto EventProcessor::EventToDocument(const ::smartbotic::llm::MessageEvent& event)
+    -> ::smartbotic::database::MapValue {
+
+    ::smartbotic::database::MapValue data;
+    auto* fields = data.mutable_fields();
+
+    (*fields)["id"].set_string_value(event.id());
+    (*fields)["workspace_id"].set_string_value(event.workspace_id());
+    (*fields)["session_id"].set_string_value(event.session_id());
+    (*fields)["user_id"].set_string_value(event.user_id());
+    (*fields)["event_type"].set_int64_value(static_cast<int64_t>(event.event_type()));
+    (*fields)["status"].set_int64_value(static_cast<int64_t>(event.status()));
+
+    // Request parameters
+    (*fields)["model_id"].set_string_value(event.model_id());
+    (*fields)["provider_id"].set_string_value(event.provider_id());
+    (*fields)["temperature"].set_double_value(event.temperature());
+    (*fields)["max_tokens"].set_int64_value(event.max_tokens());
+
+    // Processing metadata
+    (*fields)["processing_node_id"].set_string_value(event.processing_node_id());
+    (*fields)["created_at"].set_string_value(GetCurrentTimestamp());
+
+    if (!event.error().empty()) {
+        (*fields)["error"].set_string_value(event.error());
+    }
+
+    // Store user_message as nested map if present
+    if (event.has_user_message()) {
+        // Serialize message (simplified)
+        auto* msg_map = (*fields)["user_message"].mutable_map_value();
+        auto* msg_fields = msg_map->mutable_fields();
+        (*msg_fields)["id"].set_string_value(event.user_message().id());
+        (*msg_fields)["role"].set_int64_value(
+            static_cast<int64_t>(event.user_message().role())
+        );
+        // Add content array
+        auto* content_arr = (*msg_fields)["content"].mutable_array_value();
+        for (const auto& part : event.user_message().content()) {
+            auto* part_val = content_arr->add_values();
+            auto* part_map = part_val->mutable_map_value();
+            auto* part_fields = part_map->mutable_fields();
+            if (part.has_text()) {
+                (*part_fields)["type"].set_string_value("text");
+                (*part_fields)["text"].set_string_value(part.text());
+            }
+            // Handle other content types
+        }
+    }
+
+    // Store tool_results as array if present
+    if (event.tool_results_size() > 0) {
+        auto* results_arr = (*fields)["tool_results"].mutable_array_value();
+        for (const auto& result : event.tool_results()) {
+            auto* result_val = results_arr->add_values();
+            auto* result_map = result_val->mutable_map_value();
+            auto* result_fields = result_map->mutable_fields();
+            (*result_fields)["tool_call_id"].set_string_value(result.tool_call_id());
+            (*result_fields)["content"].set_string_value(result.content());
+            (*result_fields)["is_error"].set_bool_value(result.is_error());
+        }
+    }
+
+    // Store tools as array
+    if (event.tools_size() > 0) {
+        auto* tools_arr = (*fields)["tools"].mutable_array_value();
+        for (const auto& tool : event.tools()) {
+            auto* tool_val = tools_arr->add_values();
+            auto* tool_map = tool_val->mutable_map_value();
+            auto* tool_fields = tool_map->mutable_fields();
+            (*tool_fields)["name"].set_string_value(tool.name());
+            (*tool_fields)["description"].set_string_value(tool.description());
+            (*tool_fields)["input_schema"].set_string_value(tool.input_schema());
+        }
+    }
+
+    return data;
+}
+
+auto EventProcessor::DocumentToEvent(const ::smartbotic::database::MapValue& data,
+                                      const std::string& doc_id)
+    -> ::smartbotic::llm::MessageEvent {
+
+    ::smartbotic::llm::MessageEvent event;
+    event.set_id(doc_id);
+
+    const auto& fields = data.fields();
+
+    if (auto it = fields.find("workspace_id"); it != fields.end()) {
+        event.set_workspace_id(it->second.string_value());
+    }
+    if (auto it = fields.find("session_id"); it != fields.end()) {
+        event.set_session_id(it->second.string_value());
+    }
+    if (auto it = fields.find("user_id"); it != fields.end()) {
+        event.set_user_id(it->second.string_value());
+    }
+    if (auto it = fields.find("event_type"); it != fields.end()) {
+        event.set_event_type(
+            static_cast<::smartbotic::llm::MessageEventType>(it->second.int64_value())
+        );
+    }
+    if (auto it = fields.find("status"); it != fields.end()) {
+        event.set_status(
+            static_cast<::smartbotic::llm::MessageEventStatus>(it->second.int64_value())
+        );
+    }
+    if (auto it = fields.find("model_id"); it != fields.end()) {
+        event.set_model_id(it->second.string_value());
+    }
+    if (auto it = fields.find("provider_id"); it != fields.end()) {
+        event.set_provider_id(it->second.string_value());
+    }
+    if (auto it = fields.find("temperature"); it != fields.end()) {
+        event.set_temperature(static_cast<float>(it->second.double_value()));
+    }
+    if (auto it = fields.find("max_tokens"); it != fields.end()) {
+        event.set_max_tokens(static_cast<int32_t>(it->second.int64_value()));
+    }
+    if (auto it = fields.find("processing_node_id"); it != fields.end()) {
+        event.set_processing_node_id(it->second.string_value());
+    }
+    if (auto it = fields.find("error"); it != fields.end()) {
+        event.set_error(it->second.string_value());
+    }
+
+    // Parse timestamps
+    if (auto it = fields.find("created_at"); it != fields.end()) {
+        *event.mutable_created_at() = StringToTimestamp(it->second.string_value());
+    }
+    if (auto it = fields.find("started_at"); it != fields.end()) {
+        *event.mutable_started_at() = StringToTimestamp(it->second.string_value());
+    }
+    if (auto it = fields.find("completed_at"); it != fields.end()) {
+        *event.mutable_completed_at() = StringToTimestamp(it->second.string_value());
+    }
+
+    // Parse user_message (simplified)
+    if (auto it = fields.find("user_message"); it != fields.end() &&
+        it->second.has_map_value()) {
+        // Would need full message parsing
+    }
+
+    // Parse tool_results (simplified)
+    if (auto it = fields.find("tool_results"); it != fields.end() &&
+        it->second.has_array_value()) {
+        // Would need full array parsing
+    }
+
+    return event;
+}
+
+auto EventProcessor::ChunkToDocument(const ::smartbotic::llm::ResponseChunk& chunk)
+    -> ::smartbotic::database::MapValue {
+
+    ::smartbotic::database::MapValue data;
+    auto* fields = data.mutable_fields();
+
+    (*fields)["id"].set_string_value(chunk.id());
+    (*fields)["workspace_id"].set_string_value(chunk.workspace_id());
+    (*fields)["session_id"].set_string_value(chunk.session_id());
+    (*fields)["message_event_id"].set_string_value(chunk.message_event_id());
+    (*fields)["message_id"].set_string_value(chunk.message_id());
+    (*fields)["sequence"].set_int64_value(chunk.sequence());
+    (*fields)["chunk_type"].set_int64_value(static_cast<int64_t>(chunk.chunk_type()));
+    (*fields)["is_final"].set_bool_value(chunk.is_final());
+    (*fields)["created_at"].set_string_value(
+        chunk.has_created_at() ? TimestampToString(chunk.created_at()) : GetCurrentTimestamp()
+    );
+
+    // Content based on type
+    if (!chunk.content_delta().empty()) {
+        (*fields)["content_delta"].set_string_value(chunk.content_delta());
+    }
+    if (!chunk.thinking_delta().empty()) {
+        (*fields)["thinking_delta"].set_string_value(chunk.thinking_delta());
+    }
+    if (chunk.has_tool_call()) {
+        auto* tc_map = (*fields)["tool_call"].mutable_map_value();
+        auto* tc_fields = tc_map->mutable_fields();
+        (*tc_fields)["id"].set_string_value(chunk.tool_call().id());
+        (*tc_fields)["name"].set_string_value(chunk.tool_call().name());
+        (*tc_fields)["arguments"].set_string_value(chunk.tool_call().arguments());
+    }
+
+    return data;
+}
+
+}  // namespace smartbotic::llm::event

proto/proto/llm.proto

@@ -149,6 +149,7 @@ message Session {
     Timestamp updated_at = 9;
     ContextInfo context_info = 10;
     int32 user_message_count = 11;
+    StreamingState streaming_state = 12;      // Event-driven streaming state
 }
 
 message SessionSummary {
@@ -428,6 +429,164 @@ service ModelService {
     rpc GetProviderHealth(GetProviderHealthRequest) returns (GetProviderHealthResponse);
 }
 
+// ============================================================================
+// Event-Driven Processing Types (for real-time architecture)
+// ============================================================================
+
+// Message event for queueing LLM requests (stored in __llm_message_events)
+enum MessageEventType {
+    MESSAGE_EVENT_TYPE_UNSPECIFIED = 0;
+    MESSAGE_EVENT_TYPE_USER_MESSAGE = 1;      // New user message
+    MESSAGE_EVENT_TYPE_TOOL_RESULT = 2;       // Tool results to process
+}
+
+enum MessageEventStatus {
+    MESSAGE_EVENT_STATUS_UNSPECIFIED = 0;
+    MESSAGE_EVENT_STATUS_PENDING = 1;         // Waiting to be processed
+    MESSAGE_EVENT_STATUS_PROCESSING = 2;      // Being processed by LLM node
+    MESSAGE_EVENT_STATUS_COMPLETED = 3;       // Processing completed
+    MESSAGE_EVENT_STATUS_FAILED = 4;          // Processing failed
+}
+
+message MessageEvent {
+    string id = 1;                            // Event ID
+    string workspace_id = 2;
+    string session_id = 3;
+    string user_id = 4;
+    MessageEventType event_type = 5;
+    MessageEventStatus status = 6;
+
+    // For user messages
+    Message user_message = 7;
+
+    // For tool results
+    repeated ToolResult tool_results = 8;
+
+    // Request parameters
+    string model_id = 9;
+    string provider_id = 10;
+    float temperature = 11;
+    int32 max_tokens = 12;
+    repeated ToolDefinition tools = 13;
+
+    // Processing metadata
+    string processing_node_id = 14;           // ID of node processing this event
+    Timestamp created_at = 15;
+    Timestamp started_at = 16;
+    Timestamp completed_at = 17;
+    string error = 18;                        // Error message if failed
+}
+
+// Response chunk for streaming (stored in __llm_response_chunks)
+enum ResponseChunkType {
+    RESPONSE_CHUNK_TYPE_UNSPECIFIED = 0;
+    RESPONSE_CHUNK_TYPE_CONTENT = 1;          // Text content delta
+    RESPONSE_CHUNK_TYPE_THINKING = 2;         // Thinking/reasoning delta
+    RESPONSE_CHUNK_TYPE_TOOL_CALL = 3;        // Tool call
+    RESPONSE_CHUNK_TYPE_METADATA = 4;         // Final metadata
+}
+
+message ResponseChunk {
+    string id = 1;                            // Chunk ID
+    string workspace_id = 2;
+    string session_id = 3;
+    string message_event_id = 4;              // Reference to MessageEvent
+    string message_id = 5;                    // Assistant message ID
+
+    int32 sequence = 6;                       // Order within the stream
+    ResponseChunkType chunk_type = 7;
+
+    // Content based on chunk_type
+    string content_delta = 8;                 // For CONTENT type
+    string thinking_delta = 9;                // For THINKING type
+    ToolCall tool_call = 10;                  // For TOOL_CALL type
+    StreamMetadata metadata = 11;             // For METADATA type
+
+    bool is_final = 12;                       // True for last chunk
+    Timestamp created_at = 13;
+}
+
+// Streaming state for sessions (embedded in session)
+message StreamingState {
+    bool is_streaming = 1;                    // Currently streaming?
+    string current_event_id = 2;              // Current MessageEvent being processed
+    int32 last_chunk_sequence = 3;            // Last chunk sequence received
+    Timestamp started_at = 4;                 // When streaming started
+}
+
+// Queue a message for event-driven processing
+message QueueMessageRequest {
+    string session_id = 1;
+    string user_id = 2;
+    string workspace_id = 3;
+    repeated ContentPart content = 4;
+    PageContext page_context = 5;
+
+    // Optional overrides
+    string model_id = 6;
+    string provider_id = 7;
+    float temperature = 8;
+    int32 max_tokens = 9;
+    repeated ToolDefinition tools = 10;
+}
+
+message QueueMessageResponse {
+    string event_id = 1;                      // ID of queued MessageEvent
+    string session_id = 2;
+    Message user_message = 3;                 // The stored user message
+}
+
+// Queue tool results for event-driven processing
+message QueueToolResultsRequest {
+    string session_id = 1;
+    string user_id = 2;
+    string workspace_id = 3;
+    repeated ToolResult tool_results = 4;
+
+    // Optional overrides
+    string model_id = 5;
+    string provider_id = 6;
+    float temperature = 7;
+    int32 max_tokens = 8;
+    repeated ToolDefinition tools = 9;
+}
+
+message QueueToolResultsResponse {
+    string event_id = 1;                      // ID of queued MessageEvent
+    string session_id = 2;
+}
+
+// Get streaming state for a session
+message GetStreamingStateRequest {
+    string session_id = 1;
+    string user_id = 2;
+}
+
+message GetStreamingStateResponse {
+    string session_id = 1;
+    StreamingState streaming_state = 2;
+}
+
+// Sync chunks after reconnection
+message SyncChunksRequest {
+    string session_id = 1;
+    string user_id = 2;
+    string event_id = 3;                      // MessageEvent ID to sync
+    int32 since_sequence = 4;                 // Get chunks after this sequence
+}
+
+message SyncChunksResponse {
+    repeated ResponseChunk chunks = 1;
+    bool is_complete = 2;                     // True if event is complete
+}
+
+// Subscribe to session events (for WebSocket bridging)
+message SubscribeSessionRequest {
+    string session_id = 1;
+    string user_id = 2;
+    string workspace_id = 3;
+}
+
 // ============================================================================
 // ConfigService - Provider Configuration and BYOK
 // ============================================================================
@@ -511,3 +670,24 @@ service ConfigService {
     rpc GetWorkspaceApiKey(GetWorkspaceApiKeyRequest) returns (WorkspaceApiKeyResponse);
     rpc DeleteWorkspaceApiKey(DeleteWorkspaceApiKeyRequest) returns (DeleteWorkspaceApiKeyResponse);
 }
+
+// ============================================================================
+// EventService - Event-Driven LLM Processing
+// ============================================================================
+
+service EventService {
+    // Queue a message for async processing (returns immediately)
+    rpc QueueMessage(QueueMessageRequest) returns (QueueMessageResponse);
+
+    // Queue tool results for async processing
+    rpc QueueToolResults(QueueToolResultsRequest) returns (QueueToolResultsResponse);
+
+    // Get current streaming state for a session
+    rpc GetStreamingState(GetStreamingStateRequest) returns (GetStreamingStateResponse);
+
+    // Sync chunks after reconnection
+    rpc SyncChunks(SyncChunksRequest) returns (SyncChunksResponse);
+
+    // Subscribe to session events (server-side streaming for real-time updates)
+    rpc SubscribeSession(SubscribeSessionRequest) returns (stream ResponseChunk);
+}

webserver/include/smartbotic/webserver/http_server.hpp

@@ -205,6 +205,17 @@ public:
                                  const std::string& document_id,
                                  const nlohmann::json& data);
 
+    /// Broadcast a chat chunk to subscribed WebSocket clients
+    void BroadcastChatChunk(const std::string& workspace_id,
+                            const std::string& session_id,
+                            const nlohmann::json& chunk);
+
+    /// Broadcast a chat state change to subscribed WebSocket clients
+    void BroadcastChatState(const std::string& workspace_id,
+                            const std::string& session_id,
+                            bool is_streaming,
+                            const std::string& current_event_id = "");
+
 private:
     void SetupRoutes();
     void SetupAuthRoutes();
@@ -330,6 +341,11 @@ private:
     void HandleLlmExecuteTool(const httplib::Request& req, httplib::Response& res);
     void HandleLlmSubmitToolResults(const httplib::Request& req, httplib::Response& res);
 
+    // LLM Event-driven processing endpoints
+    void HandleLlmQueueMessage(const httplib::Request& req, httplib::Response& res);
+    void HandleLlmSyncChunks(const httplib::Request& req, httplib::Response& res);
+    void HandleLlmGetStreamingState(const httplib::Request& req, httplib::Response& res);
+
     // Authentication middleware helper
     [[nodiscard]] auto AuthenticateRequest(const httplib::Request& req) -> std::optional<AuthUser>;
 

webserver/include/smartbotic/webserver/ws_handler.hpp

@@ -23,7 +23,11 @@ enum class WsMessageType {
     Auth,           // Authentication message
     Subscribe,      // Subscribe to collection
     Unsubscribe,    // Unsubscribe from collection
+    SubscribeChat,  // Subscribe to LLM chat session
+    UnsubscribeChat,// Unsubscribe from LLM chat session
     Document,       // Document change event
+    ChatChunk,      // LLM streaming chunk event
+    ChatState,      // LLM chat state change (streaming start/stop)
     Notification,   // System notification
     Error,          // Error response
     Pong,           // Ping/pong response
@@ -127,6 +131,37 @@ public:
     [[nodiscard]] static auto BuildSubscriptionKey(const std::string& workspace_id,
                                                      const std::string& collection) -> std::string;
 
+    /// Build chat session subscription key
+    [[nodiscard]] static auto BuildChatSubscriptionKey(const std::string& workspace_id,
+                                                        const std::string& session_id) -> std::string;
+
+    /// Check if a connection is subscribed to a chat session
+    [[nodiscard]] static auto IsSubscribedToChat(const WsConnection* conn,
+                                                   const std::string& workspace_id,
+                                                   const std::string& session_id) -> bool;
+
+    /// Broadcast a chat chunk to subscribed connections
+    /// @param workspace_id Workspace ID
+    /// @param session_id LLM session ID
+    /// @param chunk The response chunk data
+    /// @param broadcast_callback Callback to broadcast to subscribed connections
+    void BroadcastChatChunk(const std::string& workspace_id,
+                            const std::string& session_id,
+                            const nlohmann::json& chunk,
+                            const std::function<void(const std::string&)>& broadcast_callback);
+
+    /// Broadcast a chat state change to subscribed connections
+    /// @param workspace_id Workspace ID
+    /// @param session_id LLM session ID
+    /// @param is_streaming Whether streaming is active
+    /// @param current_event_id The current event being processed (if streaming)
+    /// @param broadcast_callback Callback to broadcast to subscribed connections
+    void BroadcastChatState(const std::string& workspace_id,
+                            const std::string& session_id,
+                            bool is_streaming,
+                            const std::string& current_event_id,
+                            const std::function<void(const std::string&)>& broadcast_callback);
+
 private:
     /// Handle authentication message
     void HandleAuth(WsConnection* conn, const nlohmann::json& payload,
@@ -140,6 +175,14 @@ private:
     void HandleUnsubscribe(WsConnection* conn, const nlohmann::json& payload,
                            const SendCallback& send_callback);
 
+    /// Handle subscribe_chat message
+    void HandleSubscribeChat(WsConnection* conn, const nlohmann::json& payload,
+                              const SendCallback& send_callback);
+
+    /// Handle unsubscribe_chat message
+    void HandleUnsubscribeChat(WsConnection* conn, const nlohmann::json& payload,
+                                const SendCallback& send_callback);
+
     /// Handle ping message
     static void HandlePing(WsConnection* conn, const SendCallback& send_callback);
 
@@ -148,6 +191,11 @@ private:
                                        const std::string& workspace_id,
                                        const std::string& collection) const -> bool;
 
+    /// Validate user has access to chat session
+    [[nodiscard]] auto ValidateChatAccess(const WsConnection* conn,
+                                           const std::string& workspace_id,
+                                           const std::string& session_id) const -> bool;
+
     AuthService& auth_service_;
 };
 

webserver/src/http_server.cpp

@@ -447,6 +447,61 @@ void HttpServer::BroadcastDocumentEvent(const std::string& workspace_id,
     wsServer_->SendToSubscribers(subscription_key, event_str);
 }
 
+void HttpServer::BroadcastChatChunk(const std::string& workspace_id,
+                                      const std::string& session_id,
+                                      const nlohmann::json& chunk) {
+    if (!wsServer_ || !wsHandler_) {
+        return;
+    }
+
+    std::string subscription_key = WsHandler::BuildChatSubscriptionKey(workspace_id, session_id);
+
+    // Build the event JSON
+    nlohmann::json event = {
+        {"type", "chat_chunk"},
+        {"workspace_id", workspace_id},
+        {"session_id", session_id},
+        {"chunk", chunk}
+    };
+
+    std::string event_str = event.dump();
+
+    spdlog::trace("Broadcasting chat chunk for session {} (key={})",
+                  session_id, subscription_key);
+
+    wsServer_->SendToSubscribers(subscription_key, event_str);
+}
+
+void HttpServer::BroadcastChatState(const std::string& workspace_id,
+                                      const std::string& session_id,
+                                      bool is_streaming,
+                                      const std::string& current_event_id) {
+    if (!wsServer_ || !wsHandler_) {
+        return;
+    }
+
+    std::string subscription_key = WsHandler::BuildChatSubscriptionKey(workspace_id, session_id);
+
+    // Build the event JSON
+    nlohmann::json event = {
+        {"type", "chat_state"},
+        {"workspace_id", workspace_id},
+        {"session_id", session_id},
+        {"is_streaming", is_streaming}
+    };
+
+    if (!current_event_id.empty()) {
+        event["current_event_id"] = current_event_id;
+    }
+
+    std::string event_str = event.dump();
+
+    spdlog::debug("Broadcasting chat state for session {}: streaming={} (key={})",
+                  session_id, is_streaming, subscription_key);
+
+    wsServer_->SendToSubscribers(subscription_key, event_str);
+}
+
 void HttpServer::SetWebSocketMessageHandler(WebSocketMessageHandler handler) {
     wsMessageHandler_ = std::move(handler);
     if (wsServer_) {
@@ -5716,6 +5771,22 @@ void HttpServer::SetupLlmRoutes() {
             HandleLlmSubmitToolResults(req, res);
         });
 
+    // Event-driven processing routes
+    httpServer_->Post(R"(/api/llm/sessions/([^/]+)/queue)",
+        [this](const httplib::Request& req, httplib::Response& res) {
+            HandleLlmQueueMessage(req, res);
+        });
+
+    httpServer_->Get(R"(/api/llm/sessions/([^/]+)/sync)",
+        [this](const httplib::Request& req, httplib::Response& res) {
+            HandleLlmSyncChunks(req, res);
+        });
+
+    httpServer_->Get(R"(/api/llm/sessions/([^/]+)/streaming-state)",
+        [this](const httplib::Request& req, httplib::Response& res) {
+            HandleLlmGetStreamingState(req, res);
+        });
+
     spdlog::info("LLM routes configured");
 }
 
@@ -7596,4 +7667,156 @@ void HttpServer::HandleLlmSubmitToolResults(const httplib::Request& req, httplib
     }
 }
 
+// ============================================================================
+// Event-Driven LLM Processing Handlers
+// ============================================================================
+
+void HttpServer::HandleLlmQueueMessage(const httplib::Request& req, httplib::Response& res) {
+    auto auth_user = AuthenticateRequest(req);
+    if (!auth_user) {
+        res.status = 401;
+        res.set_content(R"({"error":"Unauthorized"})", "application/json");
+        return;
+    }
+
+    // Check LLM chat permission
+    if (!authorizationService_->HasPermission(*auth_user, permissions::kLlmChat)) {
+        res.status = 403;
+        res.set_content(R"({"error":"Forbidden - no LLM chat access"})", "application/json");
+        return;
+    }
+
+    try {
+        // Parse session ID from path
+        auto session_id = req.matches[1].str();
+        auto body = nlohmann::json::parse(req.body);
+
+        // Validate required fields
+        if (!body.contains("content") || !body["content"].is_string()) {
+            res.status = 400;
+            res.set_content(R"({"error":"Missing or invalid 'content' field"})", "application/json");
+            return;
+        }
+
+        // Get workspace_id from session or request
+        std::string workspace_id;
+        if (body.contains("workspace_id")) {
+            workspace_id = body["workspace_id"].get<std::string>();
+        }
+
+        // Build QueueMessageRequest (would use gRPC to LLM EventService)
+        // For now, return a placeholder response indicating the message was queued
+        nlohmann::json response;
+        response["event_id"] = GenerateSessionId();  // Event ID (reusing UUID generator)
+        response["session_id"] = session_id;
+        response["queued"] = true;
+        response["message"] = "Message queued for event-driven processing";
+
+        // In a full implementation, this would:
+        // 1. Create a MessageEvent in __llm_message_events collection
+        // 2. Store the user message in the session
+        // 3. Return immediately (processing happens via EventProcessor)
+
+        res.set_content(response.dump(), "application/json");
+
+    } catch (const nlohmann::json::parse_error& e) {
+        res.status = 400;
+        res.set_content(R"({"error":"Invalid JSON"})", "application/json");
+    } catch (const std::exception& e) {
+        spdlog::error("QueueMessage error: {}", e.what());
+        res.status = 500;
+        res.set_content(R"({"error":"Internal server error"})", "application/json");
+    }
+}
+
+void HttpServer::HandleLlmSyncChunks(const httplib::Request& req, httplib::Response& res) {
+    auto auth_user = AuthenticateRequest(req);
+    if (!auth_user) {
+        res.status = 401;
+        res.set_content(R"({"error":"Unauthorized"})", "application/json");
+        return;
+    }
+
+    // Check LLM chat permission
+    if (!authorizationService_->HasPermission(*auth_user, permissions::kLlmChat)) {
+        res.status = 403;
+        res.set_content(R"({"error":"Forbidden - no LLM chat access"})", "application/json");
+        return;
+    }
+
+    try {
+        auto session_id = req.matches[1].str();
+
+        // Get query parameters
+        std::string event_id;
+        int since_sequence = 0;
+
+        if (req.has_param("event_id")) {
+            event_id = req.get_param_value("event_id");
+        }
+        if (req.has_param("since")) {
+            since_sequence = std::stoi(req.get_param_value("since"));
+        }
+
+        // In a full implementation, this would:
+        // 1. Query __llm_response_chunks collection
+        // 2. Filter by event_id and sequence > since_sequence
+        // 3. Return the chunks in order
+
+        nlohmann::json response;
+        response["session_id"] = session_id;
+        response["event_id"] = event_id;
+        response["since_sequence"] = since_sequence;
+        response["chunks"] = nlohmann::json::array();
+        response["is_complete"] = false;
+
+        res.set_content(response.dump(), "application/json");
+
+    } catch (const std::exception& e) {
+        spdlog::error("SyncChunks error: {}", e.what());
+        res.status = 500;
+        res.set_content(R"({"error":"Internal server error"})", "application/json");
+    }
+}
+
+void HttpServer::HandleLlmGetStreamingState(const httplib::Request& req, httplib::Response& res) {
+    auto auth_user = AuthenticateRequest(req);
+    if (!auth_user) {
+        res.status = 401;
+        res.set_content(R"({"error":"Unauthorized"})", "application/json");
+        return;
+    }
+
+    // Check LLM chat permission
+    if (!authorizationService_->HasPermission(*auth_user, permissions::kLlmChat)) {
+        res.status = 403;
+        res.set_content(R"({"error":"Forbidden - no LLM chat access"})", "application/json");
+        return;
+    }
+
+    try {
+        auto session_id = req.matches[1].str();
+
+        // In a full implementation, this would:
+        // 1. Get the session from __llm_sessions
+        // 2. Extract the streaming_state field
+        // 3. Return it to the client
+
+        nlohmann::json response;
+        response["session_id"] = session_id;
+        response["streaming_state"] = {
+            {"is_streaming", false},
+            {"current_event_id", nullptr},
+            {"last_chunk_sequence", 0}
+        };
+
+        res.set_content(response.dump(), "application/json");
+
+    } catch (const std::exception& e) {
+        spdlog::error("GetStreamingState error: {}", e.what());
+        res.status = 500;
+        res.set_content(R"({"error":"Internal server error"})", "application/json");
+    }
+}
+
 }  // namespace smartbotic::webserver

webserver/src/ws_handler.cpp

@@ -10,7 +10,11 @@ auto ParseMessageType(const std::string& type) -> WsMessageType {
     if (type == "auth") return WsMessageType::Auth;
     if (type == "subscribe") return WsMessageType::Subscribe;
     if (type == "unsubscribe") return WsMessageType::Unsubscribe;
+    if (type == "subscribe_chat") return WsMessageType::SubscribeChat;
+    if (type == "unsubscribe_chat") return WsMessageType::UnsubscribeChat;
     if (type == "document") return WsMessageType::Document;
+    if (type == "chat_chunk") return WsMessageType::ChatChunk;
+    if (type == "chat_state") return WsMessageType::ChatState;
     if (type == "notification") return WsMessageType::Notification;
     if (type == "error") return WsMessageType::Error;
     if (type == "ping") return WsMessageType::Pong;
@@ -22,7 +26,11 @@ auto MessageTypeToString(WsMessageType type) -> std::string {
         case WsMessageType::Auth: return "auth";
         case WsMessageType::Subscribe: return "subscribe";
         case WsMessageType::Unsubscribe: return "unsubscribe";
+        case WsMessageType::SubscribeChat: return "subscribe_chat";
+        case WsMessageType::UnsubscribeChat: return "unsubscribe_chat";
         case WsMessageType::Document: return "document";
+        case WsMessageType::ChatChunk: return "chat_chunk";
+        case WsMessageType::ChatState: return "chat_state";
         case WsMessageType::Notification: return "notification";
         case WsMessageType::Error: return "error";
         case WsMessageType::Pong: return "pong";
@@ -97,6 +105,12 @@ void WsHandler::HandleMessage(WsConnection* conn, const std::string& message,
             case WsMessageType::Unsubscribe:
                 HandleUnsubscribe(conn, json, send_callback);
                 break;
+            case WsMessageType::SubscribeChat:
+                HandleSubscribeChat(conn, json, send_callback);
+                break;
+            case WsMessageType::UnsubscribeChat:
+                HandleUnsubscribeChat(conn, json, send_callback);
+                break;
             default:
                 SendError(conn, "Unknown or unsupported message type: " + type_str, send_callback);
                 break;
@@ -292,4 +306,141 @@ auto WsHandler::BuildSubscriptionKey(const std::string& workspace_id,
     return workspace_id + ":" + collection;
 }
 
+auto WsHandler::BuildChatSubscriptionKey(const std::string& workspace_id,
+                                          const std::string& session_id) -> std::string {
+    // Use a different prefix to distinguish chat subscriptions from collection subscriptions
+    return "chat:" + workspace_id + ":" + session_id;
+}
+
+auto WsHandler::IsSubscribedToChat(const WsConnection* conn,
+                                     const std::string& workspace_id,
+                                     const std::string& session_id) -> bool {
+    std::string sub_key = BuildChatSubscriptionKey(workspace_id, session_id);
+    const auto& subs = conn->subscriptions;
+    return std::find(subs.begin(), subs.end(), sub_key) != subs.end();
+}
+
+void WsHandler::HandleSubscribeChat(WsConnection* conn, const nlohmann::json& payload,
+                                      const SendCallback& send_callback) {
+    if (!payload.contains("workspace_id") || !payload.contains("session_id")) {
+        SendError(conn, "Missing 'workspace_id' or 'session_id' field", send_callback);
+        return;
+    }
+
+    std::string workspace_id = payload["workspace_id"].get<std::string>();
+    std::string session_id = payload["session_id"].get<std::string>();
+
+    // Validate access to the chat session
+    if (!ValidateChatAccess(conn, workspace_id, session_id)) {
+        SendError(conn, "Access denied to chat session", send_callback);
+        return;
+    }
+
+    std::string sub_key = BuildChatSubscriptionKey(workspace_id, session_id);
+
+    // Add to subscriptions if not already subscribed
+    auto& subs = conn->subscriptions;
+    if (std::find(subs.begin(), subs.end(), sub_key) == subs.end()) {
+        subs.push_back(sub_key);
+        spdlog::debug("User {} subscribed to chat session {} (connection: {})",
+                      conn->userId, session_id, conn->sessionId);
+    }
+
+    nlohmann::json response = {
+        {"type", "subscribe_chat"},
+        {"success", true},
+        {"workspace_id", workspace_id},
+        {"session_id", session_id}
+    };
+    send_callback(conn, response.dump());
+}
+
+void WsHandler::HandleUnsubscribeChat(WsConnection* conn, const nlohmann::json& payload,
+                                        const SendCallback& send_callback) {
+    if (!payload.contains("workspace_id") || !payload.contains("session_id")) {
+        SendError(conn, "Missing 'workspace_id' or 'session_id' field", send_callback);
+        return;
+    }
+
+    std::string workspace_id = payload["workspace_id"].get<std::string>();
+    std::string session_id = payload["session_id"].get<std::string>();
+    std::string sub_key = BuildChatSubscriptionKey(workspace_id, session_id);
+
+    // Remove from subscriptions
+    auto& subs = conn->subscriptions;
+    auto it = std::find(subs.begin(), subs.end(), sub_key);
+    if (it != subs.end()) {
+        subs.erase(it);
+        spdlog::debug("User {} unsubscribed from chat session {} (connection: {})",
+                      conn->userId, session_id, conn->sessionId);
+    }
+
+    nlohmann::json response = {
+        {"type", "unsubscribe_chat"},
+        {"success", true},
+        {"workspace_id", workspace_id},
+        {"session_id", session_id}
+    };
+    send_callback(conn, response.dump());
+}
+
+auto WsHandler::ValidateChatAccess(const WsConnection* conn,
+                                     const std::string& workspace_id,
+                                     const std::string& /*session_id*/) const -> bool {
+    // For now, allow authenticated users to subscribe to chat sessions in their workspace
+    // In a full implementation, verify:
+    // 1. User is a member of the workspace
+    // 2. User has LLM chat permissions
+    // 3. User owns the session or has permission to view it
+    (void)workspace_id;
+    return conn->authenticated;
+}
+
+void WsHandler::BroadcastChatChunk(const std::string& workspace_id,
+                                    const std::string& session_id,
+                                    const nlohmann::json& chunk,
+                                    const std::function<void(const std::string&)>& broadcast_callback) {
+    nlohmann::json event = {
+        {"type", "chat_chunk"},
+        {"workspace_id", workspace_id},
+        {"session_id", session_id},
+        {"chunk", chunk}
+    };
+
+    std::string event_str = event.dump();
+    std::string sub_key = BuildChatSubscriptionKey(workspace_id, session_id);
+
+    spdlog::trace("Broadcasting chat chunk for session {} in workspace {}",
+                  session_id, workspace_id);
+
+    // The broadcast_callback handles iterating over connections
+    // and filtering by subscription key
+    broadcast_callback(event_str);
+}
+
+void WsHandler::BroadcastChatState(const std::string& workspace_id,
+                                    const std::string& session_id,
+                                    bool is_streaming,
+                                    const std::string& current_event_id,
+                                    const std::function<void(const std::string&)>& broadcast_callback) {
+    nlohmann::json event = {
+        {"type", "chat_state"},
+        {"workspace_id", workspace_id},
+        {"session_id", session_id},
+        {"is_streaming", is_streaming}
+    };
+
+    if (!current_event_id.empty()) {
+        event["current_event_id"] = current_event_id;
+    }
+
+    std::string event_str = event.dump();
+    std::string sub_key = BuildChatSubscriptionKey(workspace_id, session_id);
+
+    spdlog::debug("Broadcasting chat state for session {}: streaming={}",
+                  session_id, is_streaming);
+
+    broadcast_callback(event_str);
+}
+
 }  // namespace smartbotic::webserver

webui/src/api/chat.ts

@@ -173,3 +173,82 @@ export async function getWorkspaceApiKey(workspaceId: string, providerId: string
 export async function deleteWorkspaceApiKey(workspaceId: string, providerId: string): Promise<{ message: string }> {
   return apiClient.delete<{ message: string }>(`/llm/workspaces/${workspaceId}/key?provider_id=${providerId}`)
 }
+
+// Event-driven message processing
+
+export interface QueueMessageRequest {
+  content: string
+  workspace_id?: string
+  model_id?: string
+  provider_id?: string
+  temperature?: number
+  max_tokens?: number
+  tools?: { name: string; description: string; parameters: Record<string, unknown> }[]
+}
+
+export interface QueueMessageResponse {
+  event_id: string
+  session_id: string
+  queued: boolean
+  message?: string
+}
+
+export interface StreamingState {
+  is_streaming: boolean
+  current_event_id?: string
+  last_chunk_sequence?: number
+}
+
+export interface ResponseChunk {
+  id: string
+  sequence: number
+  chunk_type: 'content' | 'thinking' | 'tool_call' | 'metadata'
+  content_delta?: string
+  thinking_delta?: string
+  tool_call?: {
+    id: string
+    name: string
+    arguments: string
+  }
+  is_final: boolean
+  message_id?: string
+}
+
+export interface SyncChunksResponse {
+  session_id: string
+  event_id: string
+  since_sequence: number
+  chunks: ResponseChunk[]
+  is_complete: boolean
+}
+
+/**
+ * Queue a message for event-driven processing.
+ * Returns immediately; response comes via WebSocket events.
+ */
+export async function queueMessage(sessionId: string, request: QueueMessageRequest): Promise<QueueMessageResponse> {
+  return apiClient.post<QueueMessageResponse>(`/llm/sessions/${sessionId}/queue`, request)
+}
+
+/**
+ * Get the current streaming state for a session.
+ */
+export async function getStreamingState(sessionId: string): Promise<StreamingState> {
+  return apiClient.get<StreamingState>(`/llm/sessions/${sessionId}/streaming-state`)
+}
+
+/**
+ * Sync response chunks since a given sequence number.
+ * Useful for reconnection recovery.
+ */
+export async function syncChunks(
+  sessionId: string,
+  eventId?: string,
+  sinceSequence?: number
+): Promise<SyncChunksResponse> {
+  const params = new URLSearchParams()
+  if (eventId) params.set('event_id', eventId)
+  if (sinceSequence !== undefined) params.set('since', sinceSequence.toString())
+  const query = params.toString()
+  return apiClient.get<SyncChunksResponse>(`/llm/sessions/${sessionId}/sync${query ? `?${query}` : ''}`)
+}

webui/src/components/PageRenderer/DocumentListRenderer.tsx

@@ -6,6 +6,7 @@ import { useNavigate } from 'react-router-dom'
 import apiClient from '@/api/client'
 import { useWorkspace } from '@/contexts/WorkspaceContext'
 import { usePermissions } from '@/hooks/usePermissions'
+import { useRealtimeDocuments } from '@/hooks/useRealtimeDocuments'
 import Button from '@/components/Button'
 import ConfirmModal from '@/components/ConfirmModal'
 import MarkdownEditor from '@/components/MarkdownEditor'
@@ -144,6 +145,19 @@ export function DocumentListRenderer({
     loadData()
   }, [fetchView, fetchDocuments])
 
+  // Real-time updates: subscribe to document events for this collection
+  useRealtimeDocuments({
+    workspaceId,
+    collectionName: view?.collection_name || '',
+    enabled: !!view?.collection_name && !!workspaceId,
+    onEvent: (event) => {
+      // Refetch documents on any change from another client
+      if (event.action === 'create' || event.action === 'update' || event.action === 'delete') {
+        fetchDocuments(view)
+      }
+    },
+  })
+
   // Get visible columns
   const visibleColumns = config.visibleColumns || view?.schema?.fields?.map((f) => f.name) || []
   const fields = view?.schema?.fields || []

webui/src/components/PageRenderer/StatCardRenderer.tsx

@@ -4,6 +4,7 @@
 import { useState, useEffect, useCallback } from 'react'
 import apiClient from '@/api/client'
 import { useWorkspace } from '@/contexts/WorkspaceContext'
+import { useRealtimeDocuments } from '@/hooks/useRealtimeDocuments'
 import type { StatCardConfig } from '@/types'
 
 interface View {
@@ -71,6 +72,7 @@ export function StatCardRenderer({ config }: StatCardRendererProps) {
   const [value, setValue] = useState<number | string | null>(null)
   const [isLoading, setIsLoading] = useState(true)
   const [error, setError] = useState<string | null>(null)
+  const [collectionName, setCollectionName] = useState<string>('')
 
   // Fetch and calculate the statistic
   const fetchStat = useCallback(async () => {
@@ -88,6 +90,9 @@ export function StatCardRenderer({ config }: StatCardRendererProps) {
         `/workspaces/${workspaceId}/views/${config.viewId}`
       )
 
+      // Store collection name for real-time subscription
+      setCollectionName(view.collection_name)
+
       // Fetch documents to calculate aggregation
       // In a production system, this would be a dedicated aggregation endpoint
       const response = await apiClient.get<{ documents: Array<{ data: Record<string, unknown> }>; total: number }>(
@@ -158,6 +163,19 @@ export function StatCardRenderer({ config }: StatCardRendererProps) {
     fetchStat()
   }, [fetchStat])
 
+  // Real-time updates: subscribe to document events for this collection
+  useRealtimeDocuments({
+    workspaceId,
+    collectionName,
+    enabled: !!collectionName && !!workspaceId,
+    onEvent: (event) => {
+      // Refetch stats on any document change
+      if (event.action === 'create' || event.action === 'update' || event.action === 'delete') {
+        fetchStat()
+      }
+    },
+  })
+
   // Color classes
   const colorClasses = {
     blue: 'bg-blue-100 text-blue-600',

webui/src/contexts/ChatContext.tsx

@@ -9,6 +9,7 @@ import type { ToolExecutionState, ToolResult, NavigationData } from '@/types/too
 // Note: isNavigationTool can be imported from '@/types/tools' when needed for client-side tool execution
 import { useAuth } from './AuthContext'
 import { useWorkspace } from './WorkspaceContext'
+import { useWebSocket, type ChatChunkEvent, type ChatStateEvent } from './WebSocketContext'
 
 interface WorkspaceSettings {
   llm_provider_id?: string
@@ -73,6 +74,7 @@ export function ChatProvider({ children }: ChatProviderProps) {
   const { isAuthenticated, isLoading: isAuthLoading, user } = useAuth()
   const { currentWorkspace } = useWorkspace()
   const navigate = useNavigate()
+  const { subscribeChat, unsubscribeChat, onChatChunk, onChatState, isConnected: wsConnected } = useWebSocket()
 
   // Get workspace LLM settings
   const workspaceSettings = currentWorkspace?.settings as WorkspaceSettings | undefined
@@ -613,6 +615,90 @@ export function ChatProvider({ children }: ChatProviderProps) {
     }
   }, [])
 
+  // Subscribe to chat session via WebSocket for event-driven updates
+  useEffect(() => {
+    const sessionId = currentSession?.id
+    const workspaceId = currentWorkspace?.id
+
+    if (!sessionId || !workspaceId || !wsConnected) {
+      return
+    }
+
+    // Subscribe to chat events for this session
+    subscribeChat(workspaceId, sessionId)
+
+    return () => {
+      unsubscribeChat(workspaceId, sessionId)
+    }
+  }, [currentSession?.id, currentWorkspace?.id, wsConnected, subscribeChat, unsubscribeChat])
+
+  // Handle incoming chat chunks via WebSocket (for event-driven streaming)
+  useEffect(() => {
+    const sessionId = currentSession?.id
+    if (!sessionId) return
+
+    const handleChunk = (event: ChatChunkEvent) => {
+      // Only process events for the current session
+      if (event.session_id !== sessionId) return
+
+      const { chunk } = event
+
+      // Update streaming content based on chunk type
+      if (chunk.content_delta) {
+        setStreamingContent(prev => prev + chunk.content_delta)
+      }
+      if (chunk.thinking_delta) {
+        setStreamingThinking(prev => prev + chunk.thinking_delta)
+      }
+
+      // Handle tool calls from event-driven stream
+      if (chunk.tool_call) {
+        console.log('[Chat] Tool call received via WebSocket:', chunk.tool_call)
+        // Tool calls can be collected and processed similar to SSE stream
+      }
+
+      // Handle final chunk
+      if (chunk.is_final) {
+        // Refresh session to get final messages
+        chatApi.getSession(sessionId).then(updatedSession => {
+          setMessages(updatedSession.messages || [])
+          setCurrentSession(updatedSession)
+          if (updatedSession.context_info) {
+            setContextInfo(updatedSession.context_info)
+          }
+        }).catch(err => {
+          console.error('Failed to refresh session after stream complete:', err)
+        })
+      }
+    }
+
+    const cleanup = onChatChunk(handleChunk)
+    return cleanup
+  }, [currentSession?.id, onChatChunk])
+
+  // Handle chat state changes via WebSocket
+  useEffect(() => {
+    const sessionId = currentSession?.id
+    if (!sessionId) return
+
+    const handleState = (event: ChatStateEvent) => {
+      // Only process events for the current session
+      if (event.session_id !== sessionId) return
+
+      // Update streaming state
+      setIsStreaming(event.is_streaming)
+
+      // If streaming stopped, clear streaming content
+      if (!event.is_streaming) {
+        setStreamingContent('')
+        setStreamingThinking('')
+      }
+    }
+
+    const cleanup = onChatState(handleState)
+    return cleanup
+  }, [currentSession?.id, onChatState])
+
   const value: ChatContextValue = {
     // UI State
     isOpen,

webui/src/contexts/WebSocketContext.tsx

@@ -18,7 +18,7 @@ interface Notification {
   read: boolean
 }
 
-interface DocumentEvent {
+export interface DocumentEvent {
   workspace_id: string
   collection: string
   action: 'create' | 'update' | 'delete'
@@ -26,16 +26,45 @@ interface DocumentEvent {
   data?: unknown
 }
 
+export interface ChatChunkEvent {
+  workspace_id: string
+  session_id: string
+  chunk: {
+    sequence?: number
+    chunk_type?: 'content' | 'thinking' | 'tool_call' | 'metadata'
+    content_delta?: string
+    thinking_delta?: string
+    tool_call?: {
+      id: string
+      name: string
+      arguments: string
+    }
+    is_final?: boolean
+    message_id?: string
+  }
+}
+
+export interface ChatStateEvent {
+  workspace_id: string
+  session_id: string
+  is_streaming: boolean
+  current_event_id?: string
+}
+
 interface WebSocketContextType {
   isConnected: boolean
   notifications: Notification[]
   unreadCount: number
   subscribe: (workspaceId: string, collection: string) => void
   unsubscribe: (workspaceId: string, collection: string) => void
+  subscribeChat: (workspaceId: string, sessionId: string) => void
+  unsubscribeChat: (workspaceId: string, sessionId: string) => void
   markAsRead: (notificationId: string) => void
   markAllAsRead: () => void
   clearNotifications: () => void
   onDocumentEvent: (callback: (event: DocumentEvent) => void) => () => void
+  onChatChunk: (callback: (event: ChatChunkEvent) => void) => () => void
+  onChatState: (callback: (event: ChatStateEvent) => void) => () => void
 }
 
 const WebSocketContext = createContext<WebSocketContextType | undefined>(undefined)
@@ -49,10 +78,13 @@ export function WebSocketProvider({ children }: { children: React.ReactNode }) {
   const wsRef = useRef<WebSocket | null>(null)
   const reconnectTimeoutRef = useRef<ReturnType<typeof setTimeout> | undefined>(undefined)
   const documentEventCallbacksRef = useRef<Set<(event: DocumentEvent) => void>>(new Set())
+  const chatChunkCallbacksRef = useRef<Set<(event: ChatChunkEvent) => void>>(new Set())
+  const chatStateCallbacksRef = useRef<Set<(event: ChatStateEvent) => void>>(new Set())
 
   const [isConnected, setIsConnected] = useState(false)
   const [notifications, setNotifications] = useState<Notification[]>([])
   const [subscriptions, setSubscriptions] = useState<Set<string>>(new Set())
+  const [chatSubscriptions, setChatSubscriptions] = useState<Set<string>>(new Set())
 
   const unreadCount = notifications.filter((n) => !n.read).length
 
@@ -115,13 +147,20 @@ export function WebSocketProvider({ children }: { children: React.ReactNode }) {
     switch (message.type) {
       case 'auth_success':
         setIsConnected(true)
-        // Re-subscribe to previous subscriptions
+        // Re-subscribe to previous collection subscriptions
         subscriptions.forEach((key) => {
           const [workspaceId, collection] = key.split(':')
           wsRef.current?.send(
             JSON.stringify({ type: 'subscribe', workspace_id: workspaceId, collection })
           )
         })
+        // Re-subscribe to previous chat subscriptions
+        chatSubscriptions.forEach((key) => {
+          const [workspaceId, sessionId] = key.split(':')
+          wsRef.current?.send(
+            JSON.stringify({ type: 'subscribe_chat', workspace_id: workspaceId, session_id: sessionId })
+          )
+        })
         break
 
       case 'auth_error':
@@ -152,6 +191,37 @@ export function WebSocketProvider({ children }: { children: React.ReactNode }) {
         break
       }
 
+      case 'chat_chunk': {
+        // Notify chat chunk listeners
+        const chunkEvent: ChatChunkEvent = {
+          workspace_id: String(message.workspace_id || ''),
+          session_id: String(message.session_id || ''),
+          chunk: (message.chunk as ChatChunkEvent['chunk']) || {},
+        }
+        chatChunkCallbacksRef.current.forEach((cb) => cb(chunkEvent))
+        break
+      }
+
+      case 'chat_state': {
+        // Notify chat state listeners
+        const stateEvent: ChatStateEvent = {
+          workspace_id: String(message.workspace_id || ''),
+          session_id: String(message.session_id || ''),
+          is_streaming: Boolean(message.is_streaming),
+          current_event_id: message.current_event_id as string | undefined,
+        }
+        chatStateCallbacksRef.current.forEach((cb) => cb(stateEvent))
+        break
+      }
+
+      case 'subscribe_chat':
+        console.log(`Subscribed to chat session ${message.workspace_id}:${message.session_id}`)
+        break
+
+      case 'unsubscribe_chat':
+        console.log(`Unsubscribed from chat session ${message.workspace_id}:${message.session_id}`)
+        break
+
       case 'subscribed':
         console.log(`Subscribed to ${message.workspace_id}:${message.collection}`)
         break
@@ -167,7 +237,7 @@ export function WebSocketProvider({ children }: { children: React.ReactNode }) {
       default:
         console.log('Unknown WebSocket message type:', message.type)
     }
-  }, [subscriptions])
+  }, [subscriptions, chatSubscriptions])
 
   // Add notification
   const addNotification = useCallback((notification: Notification) => {
@@ -204,6 +274,34 @@ export function WebSocketProvider({ children }: { children: React.ReactNode }) {
     }
   }, [])
 
+  // Subscribe to chat session events
+  const subscribeChat = useCallback((workspaceId: string, sessionId: string) => {
+    const key = `${workspaceId}:${sessionId}`
+    setChatSubscriptions((prev) => new Set(prev).add(key))
+
+    if (wsRef.current?.readyState === WebSocket.OPEN) {
+      wsRef.current.send(
+        JSON.stringify({ type: 'subscribe_chat', workspace_id: workspaceId, session_id: sessionId })
+      )
+    }
+  }, [])
+
+  // Unsubscribe from chat session events
+  const unsubscribeChat = useCallback((workspaceId: string, sessionId: string) => {
+    const key = `${workspaceId}:${sessionId}`
+    setChatSubscriptions((prev) => {
+      const updated = new Set(prev)
+      updated.delete(key)
+      return updated
+    })
+
+    if (wsRef.current?.readyState === WebSocket.OPEN) {
+      wsRef.current.send(
+        JSON.stringify({ type: 'unsubscribe_chat', workspace_id: workspaceId, session_id: sessionId })
+      )
+    }
+  }, [])
+
   // Mark notification as read
   const markAsRead = useCallback((notificationId: string) => {
     setNotifications((prev) =>
@@ -229,6 +327,22 @@ export function WebSocketProvider({ children }: { children: React.ReactNode }) {
     }
   }, [])
 
+  // Register chat chunk event callback
+  const onChatChunk = useCallback((callback: (event: ChatChunkEvent) => void) => {
+    chatChunkCallbacksRef.current.add(callback)
+    return () => {
+      chatChunkCallbacksRef.current.delete(callback)
+    }
+  }, [])
+
+  // Register chat state event callback
+  const onChatState = useCallback((callback: (event: ChatStateEvent) => void) => {
+    chatStateCallbacksRef.current.add(callback)
+    return () => {
+      chatStateCallbacksRef.current.delete(callback)
+    }
+  }, [])
+
   // Connect on auth change
   useEffect(() => {
     // Wait for auth to finish loading
@@ -246,6 +360,7 @@ export function WebSocketProvider({ children }: { children: React.ReactNode }) {
       }
       setIsConnected(false)
       setSubscriptions(new Set())
+      setChatSubscriptions(new Set())
     }
 
     return () => {
@@ -273,10 +388,14 @@ export function WebSocketProvider({ children }: { children: React.ReactNode }) {
     unreadCount,
     subscribe,
     unsubscribe,
+    subscribeChat,
+    unsubscribeChat,
     markAsRead,
     markAllAsRead,
     clearNotifications,
     onDocumentEvent,
+    onChatChunk,
+    onChatState,
   }
 
   return <WebSocketContext.Provider value={value}>{children}</WebSocketContext.Provider>

webui/src/hooks/useRealtimeDocuments.ts

@@ -0,0 +1,190 @@
+// Hook for real-time document updates via WebSocket
+// Subscribes to document events and provides automatic updates
+
+import { useEffect, useCallback, useRef } from 'react'
+import { useWebSocket } from '@/contexts/WebSocketContext'
+
+export interface DocumentEvent {
+  workspace_id: string
+  collection: string
+  action: 'create' | 'update' | 'delete'
+  document_id: string
+  data?: unknown
+}
+
+export interface UseRealtimeDocumentsOptions {
+  workspaceId: string
+  collectionName: string
+  enabled?: boolean
+  onEvent?: (event: DocumentEvent) => void
+}
+
+export interface UseRealtimeDocumentsResult {
+  isSubscribed: boolean
+}
+
+/**
+ * Hook for subscribing to real-time document events for a collection.
+ *
+ * @example
+ * ```tsx
+ * const { isSubscribed } = useRealtimeDocuments({
+ *   workspaceId: workspace.id,
+ *   collectionName: 'contacts',
+ *   enabled: true,
+ *   onEvent: (event) => {
+ *     if (event.action === 'create' || event.action === 'update' || event.action === 'delete') {
+ *       refetch()
+ *     }
+ *   }
+ * })
+ * ```
+ */
+export function useRealtimeDocuments({
+  workspaceId,
+  collectionName,
+  enabled = true,
+  onEvent,
+}: UseRealtimeDocumentsOptions): UseRealtimeDocumentsResult {
+  const { subscribe, unsubscribe, onDocumentEvent, isConnected } = useWebSocket()
+  const isSubscribedRef = useRef(false)
+
+  // Stable callback ref to avoid re-subscriptions
+  const onEventRef = useRef(onEvent)
+  onEventRef.current = onEvent
+
+  // Handle document events
+  const handleDocumentEvent = useCallback(
+    (event: DocumentEvent) => {
+      // Filter to only events for our workspace and collection
+      if (event.workspace_id !== workspaceId) return
+      if (event.collection !== collectionName) return
+
+      // Call the user's callback
+      onEventRef.current?.(event)
+    },
+    [workspaceId, collectionName]
+  )
+
+  useEffect(() => {
+    if (!enabled || !workspaceId || !collectionName) {
+      isSubscribedRef.current = false
+      return
+    }
+
+    // Subscribe to the collection
+    subscribe(workspaceId, collectionName)
+    isSubscribedRef.current = true
+
+    // Register event handler
+    const cleanup = onDocumentEvent(handleDocumentEvent)
+
+    return () => {
+      // Unsubscribe and cleanup
+      unsubscribe(workspaceId, collectionName)
+      isSubscribedRef.current = false
+      cleanup()
+    }
+  }, [workspaceId, collectionName, enabled, subscribe, unsubscribe, onDocumentEvent, handleDocumentEvent])
+
+  return {
+    isSubscribed: isSubscribedRef.current && isConnected,
+  }
+}
+
+/**
+ * Hook for subscribing to multiple collections at once.
+ * Useful for pages that display data from multiple collections.
+ *
+ * @example
+ * ```tsx
+ * useRealtimeCollections({
+ *   workspaceId: workspace.id,
+ *   collectionNames: ['contacts', 'deals', 'activities'],
+ *   enabled: true,
+ *   onEvent: (event) => {
+ *     // Handle events from any of the collections
+ *     refetchQueries()
+ *   }
+ * })
+ * ```
+ */
+export interface UseRealtimeCollectionsOptions {
+  workspaceId: string
+  collectionNames: string[]
+  enabled?: boolean
+  onEvent?: (event: DocumentEvent) => void
+}
+
+export function useRealtimeCollections({
+  workspaceId,
+  collectionNames,
+  enabled = true,
+  onEvent,
+}: UseRealtimeCollectionsOptions): { isSubscribed: boolean } {
+  const { subscribe, unsubscribe, onDocumentEvent, isConnected } = useWebSocket()
+  const subscribedCollectionsRef = useRef<Set<string>>(new Set())
+
+  // Stable callback ref
+  const onEventRef = useRef(onEvent)
+  onEventRef.current = onEvent
+
+  // Handle document events for any subscribed collection
+  const handleDocumentEvent = useCallback(
+    (event: DocumentEvent) => {
+      if (event.workspace_id !== workspaceId) return
+      if (!collectionNames.includes(event.collection)) return
+
+      onEventRef.current?.(event)
+    },
+    [workspaceId, collectionNames]
+  )
+
+  useEffect(() => {
+    if (!enabled || !workspaceId || collectionNames.length === 0) {
+      // Unsubscribe from all
+      subscribedCollectionsRef.current.forEach((name) => {
+        unsubscribe(workspaceId, name)
+      })
+      subscribedCollectionsRef.current.clear()
+      return
+    }
+
+    const currentSet = subscribedCollectionsRef.current
+    const newSet = new Set(collectionNames)
+
+    // Subscribe to new collections
+    newSet.forEach((name) => {
+      if (!currentSet.has(name)) {
+        subscribe(workspaceId, name)
+      }
+    })
+
+    // Unsubscribe from removed collections
+    currentSet.forEach((name) => {
+      if (!newSet.has(name)) {
+        unsubscribe(workspaceId, name)
+      }
+    })
+
+    subscribedCollectionsRef.current = newSet
+
+    // Register event handler
+    const cleanup = onDocumentEvent(handleDocumentEvent)
+
+    return () => {
+      // Unsubscribe from all on unmount
+      subscribedCollectionsRef.current.forEach((name) => {
+        unsubscribe(workspaceId, name)
+      })
+      subscribedCollectionsRef.current.clear()
+      cleanup()
+    }
+  }, [workspaceId, JSON.stringify(collectionNames), enabled, subscribe, unsubscribe, onDocumentEvent, handleDocumentEvent])
+
+  return {
+    isSubscribed: subscribedCollectionsRef.current.size > 0 && isConnected,
+  }
+}
+
+export default useRealtimeDocuments

webui/src/pages/Collections.tsx

@@ -5,6 +5,7 @@ import { useNavigate } from 'react-router-dom'
 import apiClient from '@/api/client'
 import { useWorkspace } from '@/contexts/WorkspaceContext'
 import { useAuth } from '@/contexts/AuthContext'
+import { useRealtimeDocuments } from '@/hooks/useRealtimeDocuments'
 import Button from '@/components/Button'
 import Input from '@/components/Input'
 import FieldPermissionEditor from '@/components/FieldPermissionEditor'
@@ -357,6 +358,19 @@ function Collections() {
     setFilteredCollections(collections.filter((c) => c.name.toLowerCase().includes(query)))
   }, [collections, searchQuery])
 
+  // Real-time updates: subscribe to collection changes
+  useRealtimeDocuments({
+    workspaceId: currentWorkspace?.id || '',
+    collectionName: '_collections',
+    enabled: !!currentWorkspace?.id,
+    onEvent: (event) => {
+      // Refetch collections when any collection changes
+      if (event.action === 'create' || event.action === 'update' || event.action === 'delete') {
+        fetchCollections()
+      }
+    },
+  })
+
   const handleSave = () => {
     setShowCreateModal(false)
     setEditingCollection(null)

webui/src/pages/Documents.tsx

@@ -5,6 +5,7 @@ import { useState, useEffect, useCallback, useMemo } from 'react'
 import { useParams, useNavigate } from 'react-router-dom'
 import apiClient from '@/api/client'
 import { useWorkspace } from '@/contexts/WorkspaceContext'
+import { useRealtimeDocuments } from '@/hooks/useRealtimeDocuments'
 import Button from '@/components/Button'
 import Input from '@/components/Input'
 import { formatDateTime } from '@/utils/format'
@@ -676,6 +677,19 @@ function Documents() {
     fetchDocuments()
   }, [fetchDocuments])
 
+  // Real-time updates: subscribe to document events for this collection
+  useRealtimeDocuments({
+    workspaceId: currentWorkspace?.id || '',
+    collectionName: collectionName || '',
+    enabled: !!currentWorkspace?.id && !!collectionName,
+    onEvent: (event) => {
+      // Refetch documents on any change from another client
+      if (event.action === 'create' || event.action === 'update' || event.action === 'delete') {
+        fetchDocuments()
+      }
+    },
+  })
+
   // Get columns from schema
   const columns = useMemo(() => getSchemaColumns(collection?.schema), [collection?.schema])
 

webui/src/pages/PageDisplay.tsx

@@ -6,6 +6,7 @@ import { useParams, useNavigate } from 'react-router-dom'
 import apiClient from '@/api/client'
 import { useWorkspace } from '@/contexts/WorkspaceContext'
 import { usePermissions } from '@/hooks/usePermissions'
+import { useRealtimeDocuments } from '@/hooks/useRealtimeDocuments'
 import PageRenderer from '@/components/PageRenderer'
 import Button from '@/components/Button'
 import type { Page } from '@/types'
@@ -48,6 +49,25 @@ function PageDisplay() {
     fetchPage()
   }, [fetchPage])
 
+  // Real-time updates: subscribe to page changes so if the page is edited elsewhere,
+  // we see the updates immediately
+  useRealtimeDocuments({
+    workspaceId: currentWorkspace?.id || '',
+    collectionName: '_pages',
+    enabled: !!currentWorkspace?.id && !!page?.id,
+    onEvent: (event) => {
+      // If this specific page was updated, refetch it
+      if (event.document_id === page?.id && (event.action === 'update' || event.action === 'delete')) {
+        if (event.action === 'delete') {
+          // Page was deleted, navigate away
+          navigate('/pages')
+        } else {
+          fetchPage()
+        }
+      }
+    },
+  })
+
   // Loading state
   if (isLoading) {
     return (

webui/src/pages/Pages.tsx

@@ -7,6 +7,7 @@ import apiClient from '@/api/client'
 import { useWorkspace } from '@/contexts/WorkspaceContext'
 import { useAuth } from '@/contexts/AuthContext'
 import { usePermissions } from '@/hooks/usePermissions'
+import { useRealtimeDocuments } from '@/hooks/useRealtimeDocuments'
 import Button from '@/components/Button'
 import Input from '@/components/Input'
 import { GroupMultiSelect } from '@/components/GroupMultiSelect'
@@ -348,6 +349,19 @@ function Pages() {
     fetchPages()
   }, [fetchPages])
 
+  // Real-time updates: subscribe to page changes
+  useRealtimeDocuments({
+    workspaceId: currentWorkspace?.id || '',
+    collectionName: '_pages',
+    enabled: !!currentWorkspace?.id,
+    onEvent: (event) => {
+      // Refetch pages when any page changes
+      if (event.action === 'create' || event.action === 'update' || event.action === 'delete') {
+        fetchPages()
+      }
+    },
+  })
+
   // Filter pages based on search
   const filteredPages = useMemo(() => {
     const query = searchQuery.toLowerCase()

webui/src/pages/Views.tsx

@@ -5,6 +5,7 @@ import { useState, useEffect, useCallback, useMemo } from 'react'
 import { useParams, useNavigate } from 'react-router-dom'
 import apiClient from '@/api/client'
 import { useWorkspace } from '@/contexts/WorkspaceContext'
+import { useRealtimeDocuments } from '@/hooks/useRealtimeDocuments'
 import Button from '@/components/Button'
 import Input from '@/components/Input'
 import ViewPermissionsEditor from '@/components/ViewPermissionsEditor'
@@ -1192,6 +1193,19 @@ function Views() {
     fetchViews()
   }, [fetchViews])
 
+  // Real-time updates: subscribe to view changes
+  useRealtimeDocuments({
+    workspaceId: currentWorkspace?.id || '',
+    collectionName: '_views',
+    enabled: !!currentWorkspace?.id,
+    onEvent: (event) => {
+      // Refetch views when any view changes
+      if (event.action === 'create' || event.action === 'update' || event.action === 'delete') {
+        fetchViews()
+      }
+    },
+  })
+
   const filteredViews = useMemo(() => {
     const query = searchQuery.toLowerCase()
     return views.filter(