@ondeinference / onde / commits / 615d7a1

feat(pulse): dual-write telemetry to GresIQ document gateway (ADR-0003 M2)

Mirror each pulse event into JSONB collections (pulse_edges, pulse_models, pulse_deployments, pulse_inference_events) via the M1 document gateway, in addition to the semantic pulse/* routes that still serve reads. Adds onde_app_id as a plain doc field (no FK), a ChatEngine::with_app_id constructor, and the ONDE_PULSE_DUAL_WRITE toggle. Re-pins smbcloud-gresiq-sdk to the published 0.4.7 (adds upsert_document/get_collection); drops the local [patch]. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Seto Elkahfi committed Jun 28, 2026 at 22:22 UTC 615d7a13028bd7324dc269342c1f20a614ea5acf
5 files changed +212 -20
Cargo.lock
+15 -15
@@ -1209,7 +1209,7 @@ dependencies = [
1209 "libc",
1210 "option-ext",
1211 "redox_users",
1212 - "windows-sys 0.59.0",
1212 + "windows-sys 0.61.2",
1213 ]
1214
1215 [[package]]
@@ -1368,7 +1368,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
1368 checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb"
1369 dependencies = [
1370 "libc",
1371 - "windows-sys 0.59.0",
1371 + "windows-sys 0.61.2",
1372 ]
1373
1374 [[package]]
@@ -3012,7 +3012,7 @@ version = "0.50.3"
3012 source = "registry+https://github.com/rust-lang/crates.io-index"
3013 checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5"
3014 dependencies = [
3015 - "windows-sys 0.59.0",
3015 + "windows-sys 0.61.2",
3016 ]
3017
3018 [[package]]
@@ -3851,7 +3851,7 @@ dependencies = [
3851 "once_cell",
3852 "socket2",
3853 "tracing",
3854 - "windows-sys 0.59.0",
3854 + "windows-sys 0.60.2",
3855 ]
3856
3857 [[package]]
@@ -4291,7 +4291,7 @@ dependencies = [
4291 "errno",
4292 "libc",
4293 "linux-raw-sys",
4294 - "windows-sys 0.59.0",
4294 + "windows-sys 0.61.2",
4295 ]
4296
4297 [[package]]
@@ -4350,7 +4350,7 @@ dependencies = [
4350 "security-framework",
4351 "security-framework-sys",
4352 "webpki-root-certs",
4353 - "windows-sys 0.59.0",
4353 + "windows-sys 0.61.2",
4354 ]
4355
4356 [[package]]
@@ -4885,9 +4885,9 @@ checksum = "b7c388c1b5e93756d0c740965c41e8822f866621d41acbdf6336a6a168f8840c"
4885
4886 [[package]]
4887 name = "smbcloud-gresiq-sdk"
4888 -version = "0.4.6"
4888 +version = "0.4.7"
4889 source = "registry+https://github.com/rust-lang/crates.io-index"
4890 -checksum = "1884fb7315f8723d2414a2ce29bbbc50d3ec454f5b52ead44bf5e6066c05440c"
4890 +checksum = "45cac3688bdeba866bfbc46b89ed8f4e39be7d6818cde24347776042db567e1d"
4891 dependencies = [
4892 "anyhow",
4893 "log",
@@ -4901,9 +4901,9 @@ dependencies = [
4901
4902 [[package]]
4903 name = "smbcloud-model"
4904 -version = "0.4.6"
4904 +version = "0.4.7"
4905 source = "registry+https://github.com/rust-lang/crates.io-index"
4906 -checksum = "516bb29427592a6f80cd3b72b3cb16304340b79e92ed363d8d01ed8be9b1ed34"
4906 +checksum = "e77a4fba4ede67959cca5b3531f4e01f9b03be559117018dee132a5709a44ab7"
4907 dependencies = [
4908 "chrono",
4909 "log",
@@ -4919,9 +4919,9 @@ dependencies = [
4919
4920 [[package]]
4921 name = "smbcloud-network"
4922 -version = "0.4.6"
4922 +version = "0.4.7"
4923 source = "registry+https://github.com/rust-lang/crates.io-index"
4924 -checksum = "a86ef82a19e3b0788e9d3ef4c4ab00c96ff95fd68da42b572897a4baf82d733b"
4924 +checksum = "ba309eda605816132dfa079eb3d47d41dbcea65ca9c36efb6a4032cb6f18c25b"
4925 dependencies = [
4926 "clap 4.6.1",
4927 "log",
@@ -5406,7 +5406,7 @@ dependencies = [
5406 "getrandom 0.4.2",
5407 "once_cell",
5408 "rustix",
5409 - "windows-sys 0.59.0",
5409 + "windows-sys 0.61.2",
5410 ]
5411
5412 [[package]]
@@ -5437,7 +5437,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
5437 checksum = "230a1b821ccbd75b185820a1f1ff7b14d21da1e442e22c0863ea5f08771a8874"
5438 dependencies = [
5439 "rustix",
5440 - "windows-sys 0.59.0",
5440 + "windows-sys 0.61.2",
5441 ]
5442
5443 [[package]]
@@ -6601,7 +6601,7 @@ version = "0.1.11"
6601 source = "registry+https://github.com/rust-lang/crates.io-index"
6602 checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22"
6603 dependencies = [
6604 - "windows-sys 0.59.0",
6604 + "windows-sys 0.61.2",
6605 ]
6606
6607 [[package]]
Cargo.toml
+3 -1
@@ -89,7 +89,9 @@ tsync = "2"
89 uniffi = { version = "=0.31.0", features = ["tokio"] }
90
91 [dependencies.smbcloud-gresiq-sdk]
92 -version = "0.4"
92 +# 0.4.7 adds the document-gateway methods (upsert_document/get_collection) the
93 +# M2 dual-write path uses. See ADR-0003 M2.
94 +version = "0.4.7"
95 # path = "../smbcloud-cli/crates/smbcloud-gresiq-sdk"
96
97 [build-dependencies]
src/inference/engine.rs
+30 -2
@@ -243,6 +243,10 @@ enum HistoryEntry {
243 pub struct ChatEngine {
244 inner: Mutex<Option<LoadedModel>>,
245 pulse: std::sync::OnceLock<Option<crate::pulse::PulseClient>>,
246 + /// The Onde app UUID this engine reports telemetry for. Threaded into every
247 + /// pulse event so the document store associates rows with the right app.
248 + /// `None` for open-ended SDK consumers (UniFFI `OndeChatEngine`).
249 + onde_app_id: Option<String>,
250 }
251
252 #[cfg(any(
@@ -258,12 +262,23 @@ pub struct ChatEngine {
262 impl ChatEngine {
263 // ── Construction ─────────────────────────────────────────────────────
264
261 - /// Create a new engine with no model loaded.
265 + /// Create a new engine with no model loaded and no Onde app association.
266 + /// Pulse telemetry is still enabled, but events carry no `onde_app_id`.
267 pub fn new() -> Self {
268 + Self::with_app_id(None)
269 + }
270 +
271 + /// Create a new engine that reports telemetry for a specific Onde app.
272 + /// The app UUID is threaded into every pulse event so the document store
273 + /// associates rows with the right app. Consumer apps that know their ID
274 + /// call `ChatEngine::with_app_id(Some(smbcloud::onde_app_id()))`; the
275 + /// UniFFI `OndeChatEngine` keeps calling `new()` (→ `None`).
276 + pub fn with_app_id(onde_app_id: Option<String>) -> Self {
277 crate::install_panic_hook_once();
278 Self {
279 inner: Mutex::new(None),
280 pulse: std::sync::OnceLock::new(),
281 + onde_app_id,
282 }
283 }
284
@@ -279,7 +294,14 @@ impl ChatEngine {
294 let environment = Self::pulse_environment();
295 let edge_id =
296 std::env::var("ONDE_EDGE_ID").unwrap_or_else(|_| "onde-unknown".to_string());
282 - let client = crate::pulse::PulseClient::new(environment, edge_id);
297 + // Prefer the constructor-supplied app id; fall back to the
298 + // ONDE_APP_ID env var so UniFFI consumers can opt in without a
299 + // new constructor signature.
300 + let onde_app_id = self
301 + .onde_app_id
302 + .clone()
303 + .or_else(|| std::env::var("ONDE_APP_ID").ok().filter(|id| !id.is_empty()));
304 + let client = crate::pulse::PulseClient::new(environment, edge_id, onde_app_id);
305
306 match &client {
307 Some(_) => log::info!(
@@ -1795,6 +1817,12 @@ impl ChatEngine {
1817 Self
1818 }
1819
1820 + /// Stub mirror of the real `with_app_id`. The unsupported-platform engine
1821 + /// ignores telemetry, so the app id is dropped.
1822 + pub fn with_app_id(_onde_app_id: Option<String>) -> Self {
1823 + Self
1824 + }
1825 +
1826 pub async fn load_gguf_model(
1827 &self,
1828 _config: GgufModelConfig,
src/pulse/client.rs
+121 -2
@@ -23,6 +23,7 @@ const EMBEDDED_API_SECRET_SECRET_PRODUCTION: Option<&str> =
23 pub struct PulseClient {
24 inner: GresiqClient,
25 edge_id: String,
26 + onde_app_id: Option<String>,
27 }
28
29 impl PulseClient {
@@ -39,6 +40,23 @@ impl PulseClient {
40 )
41 }
42
43 + /// ADR-0003 M2 dual-write toggle. When enabled (the default), every pulse
44 + /// event is *also* written to the GresIQ document gateway
45 + /// (`/gresiq/v1/collections/:collection`) in addition to the semantic
46 + /// `pulse/*` routes, which still serve reads until the M5 cut-over. Set
47 + /// `ONDE_PULSE_DUAL_WRITE=0` (or `false`/`no`/`off`) to suppress it.
48 + fn dual_write_enabled() -> bool {
49 + !matches!(
50 + std::env::var("ONDE_PULSE_DUAL_WRITE")
51 + .ok()
52 + .as_deref()
53 + .map(str::trim)
54 + .map(str::to_ascii_lowercase)
55 + .as_deref(),
56 + Some("0") | Some("false") | Some("no") | Some("off")
57 + )
58 + }
59 +
60 /// Build a pulse client using the GresIQ credentials embedded in the SDK.
61 ///
62 /// Returns `None` if the SDK was compiled without `GRESIQ_API_KEY` /
@@ -47,7 +65,15 @@ impl PulseClient {
65 ///
66 /// `edge_id` is a stable device identifier (installation UUID).
67 /// Pass an empty string to default to `"onde-unknown"`.
50 - pub fn new(environment: Environment, edge_id: String) -> Option<Self> {
68 + ///
69 + /// `onde_app_id` is the Onde app UUID (from the ondeinference.com
70 + /// dashboard) that owns this telemetry. `None` is fine — events then carry
71 + /// no app association (open-source/direct Rust consumers).
72 + pub fn new(
73 + environment: Environment,
74 + edge_id: String,
75 + onde_app_id: Option<String>,
76 + ) -> Option<Self> {
77 if Self::disabled_by_env() {
78 return None;
79 }
@@ -96,7 +122,11 @@ impl PulseClient {
122 }
123 };
124
99 - Some(PulseClient { inner, edge_id })
125 + Some(PulseClient {
126 + inner,
127 + edge_id,
128 + onde_app_id,
129 + })
130 }
131
132 /// Writes the model-load event to the pulse/model_loaded table and
@@ -124,10 +154,12 @@ impl PulseClient {
154 model_id,
155 model_name,
156 load_duration_ms,
157 + onde_app_id: client.onde_app_id.clone(),
158 };
159 if let Err(error) = client.inner.insert("pulse/model_loaded", &event).await {
160 log::warn!("pulse: model_loaded failed: {}", error);
161 }
162 + client.dual_write_model_loaded(&event).await;
163 };
164
165 if tokio::runtime::Handle::try_current().is_ok() {
@@ -181,10 +213,12 @@ impl PulseClient {
213 request_id,
214 duration_ms,
215 status,
216 + onde_app_id: client.onde_app_id.clone(),
217 };
218 if let Err(error) = client.inner.insert("pulse/inference_event", &event).await {
219 log::warn!("pulse: inference_event failed: {}", error);
220 }
221 + client.dual_write_inference(&event).await;
222 };
223
224 if let Ok(handle) = tokio::runtime::Handle::try_current() {
@@ -210,4 +244,89 @@ impl PulseClient {
244 }
245 });
246 }
247 +
248 + // ── ADR-0003 M2 dual-write to the document gateway ───────────────────────
249 + //
250 + // These mirror each semantic pulse event into JSONB collections under the
251 + // Onde gresiq_app. The relational `pulse/*` routes still own reads until the
252 + // M5 cut-over, so these writes are fire-and-forget: a failure is logged and
253 + // never affects inference. `onde_app_id` rides as a plain doc field — there
254 + // is no FK in the document store.
255 +
256 + /// Mirror a model-load into `pulse_edges`, `pulse_models`, and
257 + /// `pulse_deployments`. Each collection is a separate upsert (the
258 + /// FK-ordered chain of the semantic route becomes client-orchestrated
259 + /// multi-insert; the document store has no relationships to order).
260 + async fn dual_write_model_loaded(&self, event: &ModelLoadedEvent) {
261 + if !Self::dual_write_enabled() {
262 + return;
263 + }
264 +
265 + self.write_doc(
266 + "pulse_edges",
267 + Some(&event.edge_id),
268 + &serde_json::json!({ "edge_id": event.edge_id }),
269 + )
270 + .await;
271 +
272 + // One model assignment per app (the relational table is unique on
273 + // app_id); key by onde_app_id when present, else by the model itself.
274 + let model_key = event
275 + .onde_app_id
276 + .clone()
277 + .unwrap_or_else(|| event.model_id.clone());
278 + self.write_doc(
279 + "pulse_models",
280 + Some(&model_key),
281 + &serde_json::json!({
282 + "slug": event.model_id,
283 + "model_id": event.model_id,
284 + "model_name": event.model_name,
285 + "onde_app_id": event.onde_app_id,
286 + }),
287 + )
288 + .await;
289 +
290 + let deployment_key = format!("{}:{}", event.edge_id, event.model_id);
291 + self.write_doc(
292 + "pulse_deployments",
293 + Some(&deployment_key),
294 + &serde_json::json!({
295 + "edge_id": event.edge_id,
296 + "model_id": event.model_id,
297 + "load_duration_ms": event.load_duration_ms,
298 + "onde_app_id": event.onde_app_id,
299 + }),
300 + )
301 + .await;
302 + }
303 +
304 + /// Mirror an inference event into the append-only `pulse_inference_events`
305 + /// collection (no doc_key → server-generated UUID).
306 + async fn dual_write_inference(&self, event: &InferenceEvent) {
307 + if !Self::dual_write_enabled() {
308 + return;
309 + }
310 +
311 + self.write_doc(
312 + "pulse_inference_events",
313 + None,
314 + &serde_json::json!({
315 + "edge_id": event.edge_id,
316 + "model_id": event.model_id,
317 + "request_id": event.request_id,
318 + "duration_ms": event.duration_ms,
319 + "status": event.status,
320 + "onde_app_id": event.onde_app_id,
321 + }),
322 + )
323 + .await;
324 + }
325 +
326 + /// Upsert (or append, when `key` is `None`) one document, logging failures.
327 + async fn write_doc(&self, collection: &str, key: Option<&str>, doc: &serde_json::Value) {
328 + if let Err(error) = self.inner.upsert_document(collection, key, doc).await {
329 + log::warn!("pulse: dual-write to {} failed: {}", collection, error);
330 + }
331 + }
332 }
src/pulse/events.rs
+43
@@ -13,6 +13,12 @@ pub struct ModelLoadedEvent {
13 /// How long the load took. This is `elapsed.as_millis() as u64`,
14 /// where `elapsed` is what `load_gguf_model` returns.
15 pub load_duration_ms: u64,
16 + /// The Onde app that loaded this model (UUID string from the
17 + /// ondeinference.com dashboard). `None` for SDK builds without app
18 + /// credentials or for direct Rust consumers. In the document store this
19 + /// is a plain field on the `pulse_models` doc, not a foreign key.
20 + #[serde(skip_serializing_if = "Option::is_none")]
21 + pub onde_app_id: Option<String>,
22 }
23
24 /// What gets sent to GresIQ after each `send_message` or `generate`.
@@ -29,4 +35,41 @@ pub struct InferenceEvent {
35 /// `"success"`, `"cancelled"`, or `"error"`. Currently always
36 /// `"success"` here — inference errors throw before we reach this point.
37 pub status: String,
38 + /// Same Onde app as the corresponding `ModelLoadedEvent`. `None` for SDK
39 + /// builds without app credentials or direct Rust consumers.
40 + #[serde(skip_serializing_if = "Option::is_none")]
41 + pub onde_app_id: Option<String>,
42 +}
43 +
44 +#[cfg(test)]
45 +mod tests {
46 + use super::*;
47 +
48 + #[test]
49 + fn model_loaded_serializes_onde_app_id_when_present() {
50 + let event = ModelLoadedEvent {
51 + edge_id: "edge-1".into(),
52 + model_id: "org/repo".into(),
53 + model_name: "Repo 3B".into(),
54 + load_duration_ms: 1234,
55 + onde_app_id: Some("app-uuid".into()),
56 + };
57 + let value = serde_json::to_value(&event).expect("serialize");
58 + assert_eq!(value["onde_app_id"], "app-uuid");
59 + assert_eq!(value["load_duration_ms"], 1234);
60 + }
61 +
62 + #[test]
63 + fn onde_app_id_is_omitted_when_absent() {
64 + let event = InferenceEvent {
65 + edge_id: "edge-1".into(),
66 + model_id: "org/repo".into(),
67 + request_id: "onde-1-0".into(),
68 + duration_ms: 42,
69 + status: "success".into(),
70 + onde_app_id: None,
71 + };
72 + let value = serde_json::to_value(&event).expect("serialize");
73 + assert!(value.get("onde_app_id").is_none());
74 + }
75 }