feat: add QuickwitSink module and config for ADF log shipping

Feature-gated behind quickwit Cargo feature.
- quickwit.rs: async batched NDJSON shipper (mpsc channel + tokio background task)
- config.rs: QuickwitConfig struct with defaults (endpoint, index_id, batch_size, flush_interval)
- Cargo.toml: quickwit feature flag with optional reqwest dependency

Refs #330, #333

Author: AlexMikhalev

crates/terraphim_orchestrator/Cargo.toml

@@ -7,6 +7,10 @@ description = "AI Dark Factory orchestrator wiring spawner, router, supervisor i
 license = "Apache-2.0"
 repository = "https://github.com/terraphim/terraphim-ai"
 
+[features]
+default = []
+quickwit = ["dep:reqwest"]
+
 [dependencies]
 # Terraphim internal crates
 terraphim_spawner = { path = "../terraphim_spawner", version = "1.0.0" }
@@ -36,6 +40,9 @@ toml = "0.8"
 handlebars = "6.3"
 regex = "1"
 
+# Quickwit integration (optional)
+reqwest = { workspace = true, optional = true }
+
 [dev-dependencies]
 tokio-test = "0.4"
 tempfile = "3.27"

crates/terraphim_orchestrator/src/config.rs

@@ -74,6 +74,10 @@ pub struct OrchestratorConfig {
     /// Mention-driven dispatch configuration.
     #[serde(default)]
     pub mentions: Option<MentionConfig>,
+    /// Quickwit log shipping configuration (only available with quickwit feature).
+    #[cfg(feature = "quickwit")]
+    #[serde(default)]
+    pub quickwit: Option<QuickwitConfig>,
 }
 
 /// Configuration for posting agent output to Gitea issues.
@@ -107,6 +111,65 @@ fn default_poll_modulo() -> u64 {
     2
 }
 
+/// Quickwit log shipping configuration.
+#[cfg(feature = "quickwit")]
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct QuickwitConfig {
+    /// Whether Quickwit logging is enabled.
+    #[serde(default = "default_quickwit_enabled")]
+    pub enabled: bool,
+    /// Quickwit API endpoint.
+    #[serde(default = "default_quickwit_endpoint")]
+    pub endpoint: String,
+    /// Index ID for log ingestion.
+    #[serde(default = "default_quickwit_index_id")]
+    pub index_id: String,
+    /// Maximum documents per batch.
+    #[serde(default = "default_quickwit_batch_size")]
+    pub batch_size: usize,
+    /// Seconds between forced flushes.
+    #[serde(default = "default_quickwit_flush_interval_secs")]
+    pub flush_interval_secs: u64,
+}
+
+#[cfg(feature = "quickwit")]
+impl Default for QuickwitConfig {
+    fn default() -> Self {
+        Self {
+            enabled: default_quickwit_enabled(),
+            endpoint: default_quickwit_endpoint(),
+            index_id: default_quickwit_index_id(),
+            batch_size: default_quickwit_batch_size(),
+            flush_interval_secs: default_quickwit_flush_interval_secs(),
+        }
+    }
+}
+
+#[cfg(feature = "quickwit")]
+fn default_quickwit_enabled() -> bool {
+    false
+}
+
+#[cfg(feature = "quickwit")]
+fn default_quickwit_endpoint() -> String {
+    "http://127.0.0.1:7280".to_string()
+}
+
+#[cfg(feature = "quickwit")]
+fn default_quickwit_index_id() -> String {
+    "adf-logs".to_string()
+}
+
+#[cfg(feature = "quickwit")]
+fn default_quickwit_batch_size() -> usize {
+    100
+}
+
+#[cfg(feature = "quickwit")]
+fn default_quickwit_flush_interval_secs() -> u64 {
+    5
+}
+
 /// Lightweight reference to an SFIA skill code and level.
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
 pub struct SfiaSkillRef {

crates/terraphim_orchestrator/src/quickwit.rs

@@ -0,0 +1,240 @@
+//! Quickwit log shipping for ADF orchestrator
+//!
+//! Feature-gated behind `quickwit` feature flag.
+//! Provides async log shipping to Quickwit search engine.
+
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+use tokio::sync::mpsc;
+use tracing::{error, warn};
+
+/// Log document structure for Quickwit ingestion
+#[derive(Debug, Clone, Serialize, Deserialize, Default)]
+pub struct LogDocument {
+    /// ISO 8601 timestamp
+    pub timestamp: String,
+    /// Log level (INFO, WARN, ERROR)
+    pub level: String,
+    /// Name of the agent
+    pub agent_name: String,
+    /// Agent layer (Safety, Core, Growth)
+    pub layer: String,
+    /// Source of the log (orchestrator, stdout, stderr)
+    pub source: String,
+    /// Log message
+    pub message: String,
+    /// Model used (if applicable)
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub model: Option<String>,
+    /// Persona used (if applicable)
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub persona: Option<String>,
+    /// Exit code (if applicable)
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub exit_code: Option<i32>,
+    /// Trigger event (if applicable)
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub trigger: Option<String>,
+    /// Wall time in seconds (if applicable)
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub wall_time_secs: Option<f64>,
+    /// Flow name (if applicable)
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub flow_name: Option<String>,
+    /// Extra fields (if applicable)
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub extra: Option<serde_json::Value>,
+}
+
+/// Error type for QuickwitSink operations
+#[derive(Debug)]
+pub enum QuickwitError {
+    /// Failed to send document to the channel
+    SendError(String),
+    /// HTTP client error
+    HttpError(String),
+    /// Serialization error
+    SerializationError(String),
+}
+
+impl std::fmt::Display for QuickwitError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            QuickwitError::SendError(msg) => write!(f, "Send error: {}", msg),
+            QuickwitError::HttpError(msg) => write!(f, "HTTP error: {}", msg),
+            QuickwitError::SerializationError(msg) => write!(f, "Serialization error: {}", msg),
+        }
+    }
+}
+
+impl std::error::Error for QuickwitError {}
+
+impl<T> From<mpsc::error::SendError<T>> for QuickwitError {
+    fn from(_: mpsc::error::SendError<T>) -> Self {
+        QuickwitError::SendError("Channel closed".to_string())
+    }
+}
+
+/// Async log shipper for Quickwit
+#[derive(Debug, Clone)]
+pub struct QuickwitSink {
+    tx: mpsc::Sender<LogDocument>,
+}
+
+impl QuickwitSink {
+    /// Create a new QuickwitSink and spawn the background ingestion task
+    ///
+    /// # Arguments
+    /// * `endpoint` - Quickwit API endpoint (e.g., "http://127.0.0.1:7280")
+    /// * `index_id` - Index name (e.g., "adf-logs")
+    /// * `batch_size` - Maximum documents per batch
+    /// * `flush_interval_secs` - Seconds between forced flushes
+    pub fn new(
+        endpoint: String,
+        index_id: String,
+        batch_size: usize,
+        flush_interval_secs: u64,
+    ) -> Self {
+        let (tx, mut rx) = mpsc::channel::<LogDocument>(4096);
+        let client = reqwest::Client::new();
+        let ingest_url = format!("{}/api/v1/{}/ingest", endpoint, index_id);
+
+        // Spawn background task for batching and sending
+        tokio::spawn(async move {
+            let mut buffer: Vec<LogDocument> = Vec::with_capacity(batch_size);
+            let mut last_flush = tokio::time::Instant::now();
+            let flush_interval = tokio::time::Duration::from_secs(flush_interval_secs);
+
+            loop {
+                let timeout = tokio::time::sleep_until(last_flush + flush_interval);
+                tokio::pin!(timeout);
+
+                tokio::select! {
+                    Some(doc) = rx.recv() => {
+                        buffer.push(doc);
+                        if buffer.len() >= batch_size {
+                            Self::flush_batch(&client, &ingest_url, &mut buffer).await;
+                            last_flush = tokio::time::Instant::now();
+                        }
+                    }
+                    _ = &mut timeout => {
+                        if !buffer.is_empty() {
+                            Self::flush_batch(&client, &ingest_url, &mut buffer).await;
+                            last_flush = tokio::time::Instant::now();
+                        }
+                    }
+                    else => {
+                        // Channel closed, flush remaining and exit
+                        if !buffer.is_empty() {
+                            Self::flush_batch(&client, &ingest_url, &mut buffer).await;
+                        }
+                        break;
+                    }
+                }
+            }
+        });
+
+        Self { tx }
+    }
+
+    /// Send a log document to the sink
+    ///
+    /// Returns an error if the channel is closed
+    pub async fn send(&self, doc: LogDocument) -> Result<(), QuickwitError> {
+        self.tx.send(doc).await.map_err(|e| e.into())
+    }
+
+    /// Shutdown the sink, flushing any pending documents
+    ///
+    /// Note: This drops the sender, causing the background task to flush
+    /// remaining documents and exit
+    pub fn shutdown(self) {
+        // Dropping the sender will cause the background task to exit
+        // after flushing any remaining documents
+        drop(self.tx);
+    }
+
+    /// Flush a batch of documents to Quickwit
+    ///
+    /// Fire-and-forget: errors are logged but not propagated
+    async fn flush_batch(
+        client: &reqwest::Client,
+        url: &str,
+        buffer: &mut Vec<LogDocument>,
+    ) {
+        if buffer.is_empty() {
+            return;
+        }
+
+        // Build NDJSON payload
+        let mut ndjson = String::new();
+        for doc in buffer.drain(..) {
+            match serde_json::to_string(&doc) {
+                Ok(json) => {
+                    ndjson.push_str(&json);
+                    ndjson.push('\n');
+                }
+                Err(e) => {
+                    warn!(error = %e, "failed to serialize log document");
+                }
+            }
+        }
+
+        // Send to Quickwit
+        match client
+            .post(url)
+            .header("Content-Type", "application/x-ndjson")
+            .body(ndjson)
+            .send()
+            .await
+        {
+            Ok(response) => {
+                if !response.status().is_success() {
+                    let status = response.status();
+                    let body = response.text().await.unwrap_or_default();
+                    warn!(status = %status, body = %body, "Quickwit ingest failed");
+                }
+            }
+            Err(e) => {
+                warn!(error = %e, "failed to send logs to Quickwit");
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_log_document_serialization() {
+        let doc = LogDocument {
+            timestamp: "2024-01-01T00:00:00Z".to_string(),
+            level: "INFO".to_string(),
+            agent_name: "test-agent".to_string(),
+            layer: "Safety".to_string(),
+            source: "orchestrator".to_string(),
+            message: "test message".to_string(),
+            model: Some("gpt-4".to_string()),
+            persona: None,
+            exit_code: Some(0),
+            trigger: None,
+            wall_time_secs: Some(1.5),
+            flow_name: None,
+            extra: None,
+        };
+
+        let json = serde_json::to_string(&doc).unwrap();
+        assert!(json.contains("test-agent"));
+        assert!(json.contains("INFO"));
+        assert!(json.contains("gpt-4"));
+        // None fields should be skipped
+        assert!(!json.contains("persona"));
+    }
+
+    #[test]
+    fn test_quickwit_error_display() {
+        let err = QuickwitError::HttpError("connection refused".to_string());
+        assert_eq!(err.to_string(), "HTTP error: connection refused");
+    }
+}