Add instrumentation and feedback collection infrastructure

Cargo.toml

@@ -38,6 +38,7 @@ default-members = [
   "crates/node-host",
   "crates/nostr",
   "crates/oauth",
+  "crates/observability",
   "crates/onboarding",
   "crates/openclaw-import",
   "crates/plugins",
@@ -106,6 +107,7 @@ members = [
   "crates/node-host",
   "crates/nostr",
   "crates/oauth",
+  "crates/observability",
   "crates/onboarding",
   "crates/openclaw-import",
   "crates/plugins",
@@ -212,7 +214,7 @@ sqlx = { features = ["migrate", "runtime-tokio", "sqlite"], version = "0.8" }
 # HTTP client
 reqwest = { features = ["form", "json", "multipart", "query", "socks", "stream"], version = "0.13" }
 # UUID
-uuid = { features = ["v4"], version = "1" }
+uuid = { features = ["v4", "v5"], version = "1" }
 # Async
 async-stream = "0.3"
 async-trait  = "0.1"
@@ -401,6 +403,7 @@ moltis-network-filter   = { default-features = false, path = "crates/network-fil
 moltis-node-host        = { path = "crates/node-host" }
 moltis-nostr            = { path = "crates/nostr" }
 moltis-oauth            = { path = "crates/oauth" }
+moltis-observability    = { default-features = false, path = "crates/observability" }
 moltis-onboarding       = { path = "crates/onboarding" }
 moltis-openclaw-import  = { path = "crates/openclaw-import" }
 moltis-plugins          = { path = "crates/plugins" }

README.md

@@ -126,7 +126,7 @@ Verify releases with `gh attestation verify <artifact> -R moltis-org/moltis` or
 - **Safer Agent Editing** — Automatic checkpoints before built-in skill and memory mutations, restore tooling, session branching
 - **Extensibility** — MCP servers (stdio + HTTP/SSE), skill system, 15 lifecycle hook events with circuit breaker, destructive command guard
 - **Security** — Encryption-at-rest vault (XChaCha20-Poly1305 + Argon2id), password + passkey + API key auth, sandbox isolation, SSRF/CSWSH protection
-- **Operations** — Cron scheduling, OpenTelemetry tracing, Prometheus metrics, cloud deploy (Fly.io, DigitalOcean), Tailscale integration, managed SSH deploy keys, host-pinned remote targets, live tool inventory in Settings, and CLI/web remote-exec doctor flows
+- **Operations** — Cron scheduling, agent instrumentation (Langfuse for LLM traces, OTLP for Grafana/Datadog), Prometheus metrics, cloud deploy (Fly.io, DigitalOcean), Tailscale integration, managed SSH deploy keys, host-pinned remote targets, live tool inventory in Settings, and CLI/web remote-exec doctor flows
 
 ## How It Works
 

crates/agents/Cargo.toml

@@ -13,7 +13,8 @@ async-stream    = { workspace = true }
 async-trait     = { workspace = true }
 futures         = { workspace = true }
 include_dir     = { workspace = true }
-moltis-common   = { workspace = true }
+moltis-common        = { workspace = true }
+moltis-observability = { workspace = true }
 moltis-config   = { workspace = true }
 moltis-sessions = { workspace = true }
 moltis-skills   = { workspace = true }

crates/agents/src/model/mod.rs

@@ -5,8 +5,8 @@ pub use moltis_config::schema::{AgentToolControls, ReasoningEffort, ToolChoice};
 
 mod types;
 pub use types::{
-    CompletionResponse, MAX_CAPTURED_PROVIDER_RAW_EVENTS, ModelMetadata, TOOL_CALL_METADATA_KEYS,
-    ToolCall, ToolCallArgumentDiagnostic, ToolCallArgumentSource, Usage,
+    CompletionResponse, InputTokenAccounting, MAX_CAPTURED_PROVIDER_RAW_EVENTS, ModelMetadata,
+    TOOL_CALL_METADATA_KEYS, ToolCall, ToolCallArgumentDiagnostic, ToolCallArgumentSource, Usage,
     push_capped_provider_raw_event,
 };
 
@@ -20,7 +20,9 @@ pub use convert::{
 };
 
 mod stream;
-pub use stream::{LlmProvider, StreamEvent};
+pub use stream::{
+    LlmProvider, ProviderAttemptEvent, ProviderIdentity, StreamEvent, TrackedStreamEvent,
+};
 
 #[cfg(test)]
 fn document_absolute_path_from_media_ref(media_ref: &str) -> String {
@@ -242,6 +244,24 @@ mod tests {
         assert_eq!(total.cache_write_tokens, 44);
     }
 
+    #[test]
+    fn usage_normalizes_inclusive_input_into_exclusive_buckets() {
+        let usage = Usage::from_input_tokens(InputTokenAccounting::Inclusive, 1_000, 50, 800, 20);
+
+        assert_eq!(usage.input_tokens, 180);
+        assert_eq!(usage.cache_read_tokens, 800);
+        assert_eq!(usage.cache_write_tokens, 20);
+    }
+
+    #[test]
+    fn usage_preserves_anthropic_exclusive_input() {
+        let usage = Usage::from_input_tokens(InputTokenAccounting::Exclusive, 180, 50, 800, 20);
+
+        assert_eq!(usage.input_tokens, 180);
+        assert_eq!(usage.cache_read_tokens, 800);
+        assert_eq!(usage.cache_write_tokens, 20);
+    }
+
     // ── to_openai_value ──────────────────────────────────────────────
 
     #[test]

crates/agents/src/model/stream.rs

@@ -8,6 +8,40 @@ use super::{
     types::{CompletionResponse, ModelMetadata, Usage},
 };
 
+/// Concrete provider and model selected for one completion attempt.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ProviderIdentity {
+    pub provider: String,
+    pub model: String,
+}
+
+impl ProviderIdentity {
+    #[must_use]
+    pub fn new(provider: impl Into<String>, model: impl Into<String>) -> Self {
+        Self {
+            provider: provider.into(),
+            model: model.into(),
+        }
+    }
+}
+
+/// Lifecycle events for attempts made by a provider wrapper.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum ProviderAttemptEvent {
+    Started(ProviderIdentity),
+    Failed {
+        identity: ProviderIdentity,
+        error: String,
+    },
+}
+
+/// Event returned by the runner-facing tracked streaming API.
+#[derive(Debug, Clone)]
+pub enum TrackedStreamEvent {
+    Attempt(ProviderAttemptEvent),
+    Event(StreamEvent),
+}
+
 // ── Stream events ───────────────────────────────────────────────────────────
 
 /// Events emitted during streaming LLM completion.
@@ -72,6 +106,26 @@ pub trait LlmProvider: Send + Sync {
         self.complete(messages, tools).await
     }
 
+    /// Complete while reporting each concrete provider attempt.
+    async fn complete_with_options_tracked(
+        &self,
+        messages: &[ChatMessage],
+        tools: &[serde_json::Value],
+        options: &AgentToolControls,
+        on_attempt: &mut (dyn FnMut(ProviderAttemptEvent) + Send),
+    ) -> anyhow::Result<CompletionResponse> {
+        let identity = ProviderIdentity::new(self.name(), self.id());
+        on_attempt(ProviderAttemptEvent::Started(identity.clone()));
+        let result = self.complete_with_options(messages, tools, options).await;
+        if let Err(error) = &result {
+            on_attempt(ProviderAttemptEvent::Failed {
+                identity,
+                error: error.to_string(),
+            });
+        }
+        result
+    }
+
     /// Whether this provider supports tool/function calling.
     /// Defaults to false; providers that handle the `tools` parameter
     /// in `complete()` should override this to return true.
@@ -134,6 +188,24 @@ pub trait LlmProvider: Send + Sync {
         self.stream_with_tools(messages, tools)
     }
 
+    /// Stream while reporting the concrete provider/model selected for the attempt.
+    fn stream_with_tools_and_options_tracked(
+        &self,
+        messages: Vec<ChatMessage>,
+        tools: Vec<serde_json::Value>,
+        options: AgentToolControls,
+    ) -> Pin<Box<dyn Stream<Item = TrackedStreamEvent> + Send + '_>> {
+        let attempt = TrackedStreamEvent::Attempt(ProviderAttemptEvent::Started(
+            ProviderIdentity::new(self.name(), self.id()),
+        ));
+        Box::pin(
+            tokio_stream::once(attempt).chain(
+                self.stream_with_tools_and_options(messages, tools, options)
+                    .map(TrackedStreamEvent::Event),
+            ),
+        )
+    }
+
     /// Configured reasoning effort for this provider instance, if any.
     ///
     /// Providers that support extended thinking (Anthropic, OpenAI o-series)

crates/agents/src/model/types.rs

@@ -71,13 +71,48 @@ pub const TOOL_CALL_METADATA_KEYS: &[&str] = &["thought_signature"];
 
 #[derive(Debug, Clone, Default)]
 pub struct Usage {
+    /// Fresh input tokens, excluding cache reads and cache writes.
     pub input_tokens: u32,
     pub output_tokens: u32,
     pub cache_read_tokens: u32,
     pub cache_write_tokens: u32,
 }
 
+/// Provider-reported input-token accounting semantics.
+///
+/// OpenAI-style APIs report cached tokens as part of their input total, while
+/// Anthropic reports fresh input, cache reads, and cache writes separately.
+/// Provider adapters must identify which contract they received so `Usage`
+/// always exposes mutually exclusive buckets.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum InputTokenAccounting {
+    Inclusive,
+    Exclusive,
+}
+
 impl Usage {
+    #[must_use]
+    pub const fn from_input_tokens(
+        accounting: InputTokenAccounting,
+        reported_input_tokens: u32,
+        output_tokens: u32,
+        cache_read_tokens: u32,
+        cache_write_tokens: u32,
+    ) -> Self {
+        let input_tokens = match accounting {
+            InputTokenAccounting::Inclusive => reported_input_tokens
+                .saturating_sub(cache_read_tokens)
+                .saturating_sub(cache_write_tokens),
+            InputTokenAccounting::Exclusive => reported_input_tokens,
+        };
+        Self {
+            input_tokens,
+            output_tokens,
+            cache_read_tokens,
+            cache_write_tokens,
+        }
+    }
+
     #[must_use]
     pub fn saturating_add(&self, other: &Self) -> Self {
         Self {