@ondeinference / onde
Code Commits Issues Pulls Cloud Agent CI/CD
development
copilot/assess-inference-engine-complexity copilot/fix-onde-npm-sdk-v1-1-2-release development feature/diffusion-vision main
rs 627 lines 23.5 KB
Raw
1 use smbcloud_gresiq_sdk::{Environment, GresiqClient, GresiqCredentials};
2
3 use super::events::{InferenceEvent, ModelLoadedEvent};
4
5 /// GresIQ credentials embedded at SDK build time — one pair per environment.
6 /// Consumer apps never set these — they're Onde Inference's own credentials.
7 const EMBEDDED_API_KEY_DEV: Option<&str> = option_env!("GRESIQ_API_KEY_DEV");
8 const EMBEDDED_API_SECRET_DEV: Option<&str> = option_env!("GRESIQ_API_SECRET_DEV");
9 const EMBEDDED_API_KEY_PRODUCTION: Option<&str> = option_env!("GRESIQ_API_KEY_PRODUCTION");
10 const EMBEDDED_API_SECRET_SECRET_PRODUCTION: Option<&str> =
11 option_env!("GRESIQ_API_SECRET_PRODUCTION");
12
13 /// Onde telemetry client. Wraps GresiqClient so pulse events land in the
14 /// right GresIQ-managed tables without consumer apps knowing anything about
15 /// the GresIQ auth layer underneath.
16 ///
17 /// GresIQ credentials are embedded in the SDK at build time.
18 /// Consumer apps only provide an `edge_id` (stable device identifier).
19 ///
20 /// Cheap to clone: the inner GresiqClient holds an Arc-backed reqwest::Client,
21 /// so cloning is a pointer bump, not a new TCP connection.
22 #[derive(Debug, Clone)]
23 pub struct PulseClient {
24 inner: GresiqClient,
25 edge_id: String,
26 onde_app_id: Option<String>,
27 /// Device-declared ISO 3166-1 alpha-2 country (spec 0008), already
28 /// validated. `None` means no geography is ever written.
29 country_code: Option<String>,
30 }
31
32 impl PulseClient {
33 /// Returns true when pulse telemetry is disabled explicitly by the host app.
34 pub fn disabled_by_env() -> bool {
35 flag_set(std::env::var("ONDE_DISABLE_PULSE").ok().as_deref())
36 }
37
38 /// ADR-0003 M2 dual-write toggle. The mirror is deliberately opt-in while
39 /// document parity is being proven. When enabled, every attributed pulse
40 /// event is *also* written to the GresIQ document gateway
41 /// (`/gresiq/v1/collections/:collection`) in addition to the semantic
42 /// `pulse/*` routes, which still serve reads until the M5 cut-over. Set
43 /// `ONDE_PULSE_DUAL_WRITE=1` (or `true`/`yes`/`on`) to enable it.
44 fn dual_write_enabled() -> bool {
45 flag_set(std::env::var("ONDE_PULSE_DUAL_WRITE").ok().as_deref())
46 }
47
48 /// Kill switch for country observations alone (spec 0008). Setting
49 /// `ONDE_DISABLE_PULSE_GEOGRAPHY=1` stops geography writes while every
50 /// other pulse event keeps flowing.
51 fn geography_disabled() -> bool {
52 flag_set(
53 std::env::var("ONDE_DISABLE_PULSE_GEOGRAPHY")
54 .ok()
55 .as_deref(),
56 )
57 }
58
59 /// Build a pulse client using the GresIQ credentials embedded in the SDK.
60 ///
61 /// Returns `None` if the SDK was compiled without `GRESIQ_API_KEY` /
62 /// `GRESIQ_API_SECRET` (e.g. a local dev build of onde without `.env`).
63 /// In that case telemetry is silently disabled — no panic, no partial state.
64 ///
65 /// `edge_id` is a stable device identifier (installation UUID).
66 /// Pass an empty string to default to `"onde-unknown"`.
67 ///
68 /// `onde_app_id` is the Onde app UUID (from the ondeinference.com
69 /// dashboard) that owns this telemetry. `None` is fine — events then carry
70 /// no app association (open-source/direct Rust consumers).
71 ///
72 /// `country_code` is the ISO 3166-1 alpha-2 region the host read from the
73 /// OS locale or time zone. A malformed or user-assigned code is dropped
74 /// rather than guessed at, so the observation is simply never written.
75 pub fn new(
76 environment: Environment,
77 edge_id: String,
78 onde_app_id: Option<String>,
79 country_code: Option<String>,
80 ) -> Option<Self> {
81 if Self::disabled_by_env() {
82 return None;
83 }
84
85 let (api_key, api_secret) = match environment {
86 Environment::Dev => (EMBEDDED_API_KEY_DEV?, EMBEDDED_API_SECRET_DEV?),
87 Environment::Production => (
88 EMBEDDED_API_KEY_PRODUCTION?,
89 EMBEDDED_API_SECRET_SECRET_PRODUCTION?,
90 ),
91 };
92
93 let edge_id = if edge_id.is_empty() {
94 "onde-unknown".to_string()
95 } else {
96 edge_id
97 };
98
99 // reqwest 0.12.x requires a live Tokio runtime (with I/O reactor)
100 // to construct a Client. GresiqClient::from_credentials calls
101 // reqwest::Client::new() internally. Guard against panics when
102 // called from a thread/context that lacks a Tokio reactor.
103 if tokio::runtime::Handle::try_current().is_err() {
104 log::warn!(
105 "pulse: no Tokio runtime available — \
106 deferring PulseClient creation"
107 );
108 return None;
109 }
110
111 let credentials = GresiqCredentials {
112 api_key,
113 api_secret,
114 };
115
116 let inner = match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
117 GresiqClient::from_credentials(environment, credentials)
118 })) {
119 Ok(client) => client,
120 Err(_) => {
121 log::warn!(
122 "pulse: GresiqClient::from_credentials panicked \
123 (likely missing Tokio reactor) — telemetry disabled"
124 );
125 return None;
126 }
127 };
128
129 let country_code = country_code.and_then(|raw| {
130 let normalized = normalize_country_code(&raw);
131 if normalized.is_none() {
132 log::debug!("pulse: ignoring unsupported country code");
133 }
134 normalized
135 });
136
137 Some(PulseClient {
138 inner,
139 edge_id,
140 onde_app_id,
141 country_code,
142 })
143 }
144
145 /// Writes the model-load event to the pulse/model_loaded table and
146 /// **awaits** the result before returning.
147 ///
148 /// This must complete before any `record_inference` call for the same
149 /// edge + model is made, because the API enforces a foreign-key
150 /// constraint: an `inference_event` row is rejected with HTTP 404
151 /// ("Call model_loaded first") if no matching `model_loaded` row exists.
152 ///
153 /// Keeping this synchronous with respect to the model-load path means
154 /// the row is always present by the time the first inference event fires.
155 /// A failed write emits a warn! log line — no retry, no queue, and no
156 /// effect on the caller.
157 pub async fn record_model_loaded(
158 &self,
159 model_id: String,
160 model_name: String,
161 load_duration_ms: u64,
162 ) {
163 let client = self.clone();
164 let send_event = async move {
165 let event = ModelLoadedEvent {
166 edge_id: client.edge_id.clone(),
167 model_id,
168 model_name,
169 load_duration_ms,
170 onde_app_id: client.onde_app_id.clone(),
171 };
172 if let Err(error) = client.inner.insert("pulse/model_loaded", &event).await {
173 log::warn!("pulse: model_loaded failed: {}", error);
174 }
175 client.spawn_model_loaded_dual_write(event);
176 };
177
178 if tokio::runtime::Handle::try_current().is_ok() {
179 send_event.await;
180 return;
181 }
182
183 let join = std::thread::spawn(move || {
184 let runtime = tokio::runtime::Builder::new_current_thread()
185 .enable_all()
186 .build();
187
188 match runtime {
189 Ok(runtime) => {
190 runtime.block_on(send_event);
191 }
192 Err(error) => {
193 log::warn!(
194 "pulse: could not create fallback runtime for model_loaded: {}",
195 error
196 );
197 }
198 }
199 });
200
201 if join.join().is_err() {
202 log::warn!("pulse: fallback model_loaded thread panicked");
203 }
204 }
205
206 /// Same fire-and-forget pattern as record_model_loaded but for inference
207 /// completions. Writes to pulse/inference_event. Logs on failure, no retry.
208 ///
209 /// Swift / UniFFI consumers may call into the SDK from contexts where
210 /// `tokio::spawn` does not have a current runtime handle even though the
211 /// outer API is async. To avoid panicking on Apple platforms, we prefer
212 /// the current Tokio runtime when available and otherwise fall back to a
213 /// short-lived current-thread runtime on a native background thread.
214 pub fn record_inference(
215 &self,
216 model_id: String,
217 request_id: String,
218 duration_ms: u64,
219 ttft_ms: Option<u64>,
220 status: String,
221 ) {
222 let client = self.clone();
223 let send_event = async move {
224 let event = InferenceEvent {
225 edge_id: client.edge_id.clone(),
226 model_id,
227 request_id,
228 duration_ms,
229 ttft_ms,
230 status,
231 onde_app_id: client.onde_app_id.clone(),
232 };
233 if let Err(error) = client.inner.insert("pulse/inference_event", &event).await {
234 log::warn!("pulse: inference_event failed: {}", error);
235 }
236 client.dual_write_inference(&event).await;
237 };
238
239 if let Ok(handle) = tokio::runtime::Handle::try_current() {
240 handle.spawn(send_event);
241 return;
242 }
243
244 std::thread::spawn(move || {
245 let runtime = tokio::runtime::Builder::new_current_thread()
246 .enable_all()
247 .build();
248
249 match runtime {
250 Ok(runtime) => {
251 runtime.block_on(send_event);
252 }
253 Err(error) => {
254 log::warn!(
255 "pulse: could not create fallback runtime for inference_event: {}",
256 error
257 );
258 }
259 }
260 });
261 }
262
263 // ── ADR-0003 M2 dual-write to the document gateway ───────────────────────
264 //
265 // These mirror each semantic pulse event into JSONB collections under the
266 // Onde gresiq_app. The relational `pulse/*` routes still own reads until the
267 // M5 cut-over, so failures are logged and never affect inference.
268
269 /// Move the model-load mirror requests off the model-loading critical
270 /// path. A dedicated thread is acceptable here because this runs once per
271 /// model activation, not once per inference, and also works for callers
272 /// whose original async runtime is short-lived.
273 fn spawn_model_loaded_dual_write(&self, event: ModelLoadedEvent) {
274 if !Self::dual_write_enabled() || event.onde_app_id.is_none() {
275 return;
276 }
277
278 let client = self.clone();
279 std::thread::spawn(move || {
280 let runtime = tokio::runtime::Builder::new_current_thread()
281 .enable_all()
282 .build();
283 match runtime {
284 Ok(runtime) => runtime.block_on(client.dual_write_model_loaded(&event)),
285 Err(error) => log::warn!(
286 "pulse: could not create runtime for model-loaded dual-write: {}",
287 error
288 ),
289 }
290 });
291 }
292
293 /// Mirror a model-load into `pulse_edges`, `pulse_models`, and
294 /// `pulse_deployments`. Each collection is a separate upsert (the
295 /// FK-ordered chain of the semantic route becomes client-orchestrated
296 /// multi-insert; the document store has no relationships to order).
297 async fn dual_write_model_loaded(&self, event: &ModelLoadedEvent) {
298 let Some(onde_app_id) = event.onde_app_id.as_deref() else {
299 return;
300 };
301
302 let edge_key = scoped_document_key(onde_app_id, &[&event.edge_id]);
303
304 let model_key = scoped_document_key(onde_app_id, &[&event.model_id]);
305 let deployment_key = scoped_document_key(onde_app_id, &[&event.edge_id, &event.model_id]);
306 let edge_document = edge_document(event, onde_app_id);
307 let model_document = serde_json::json!({
308 "slug": event.model_id,
309 "model_id": event.model_id,
310 "model_name": event.model_name,
311 "onde_app_id": event.onde_app_id,
312 });
313 let deployment_document = serde_json::json!({
314 "edge_id": event.edge_id,
315 "model_id": event.model_id,
316 "load_duration_ms": event.load_duration_ms,
317 "onde_app_id": event.onde_app_id,
318 "residency_state": "loaded",
319 });
320
321 tokio::join!(
322 self.write_doc("pulse_edges", Some(&edge_key), &edge_document),
323 self.write_doc("pulse_models", Some(&model_key), &model_document),
324 self.write_doc(
325 "pulse_deployments",
326 Some(&deployment_key),
327 &deployment_document,
328 ),
329 self.write_geography_observation(onde_app_id, &event.edge_id),
330 );
331 }
332
333 /// Record the device-declared country in `pulse_geography_observations`
334 /// (spec 0008). Country is kept off `pulse_edges` on purpose: that upsert
335 /// is durable, and edge-linked geography must age out after 30 days.
336 ///
337 /// The key carries the UTC date, so repeated model loads within a day
338 /// overwrite one document instead of building a location trail. The
339 /// gateway keeps `created_at` on upsert, which is what the retention sweep
340 /// ages on.
341 async fn write_geography_observation(&self, onde_app_id: &str, edge_id: &str) {
342 let Some(country_code) = self.country_code.as_deref() else {
343 return;
344 };
345 if Self::geography_disabled() {
346 return;
347 }
348
349 let observed_at_ms = epoch_millis();
350 let key = scoped_document_key(onde_app_id, &[edge_id, &utc_date(observed_at_ms)]);
351 let document = geography_document(onde_app_id, edge_id, country_code, observed_at_ms);
352
353 self.write_doc("pulse_geography_observations", Some(&key), &document)
354 .await;
355 }
356
357 /// Mirror an inference event idempotently, so a retry can't double-count
358 /// the same inference during cutover validation. Request IDs are only
359 /// unique within one process (millisecond plus a per-process counter), so
360 /// two edges of the same app can mint the same ID; the edge is part of the
361 /// key for that reason.
362 async fn dual_write_inference(&self, event: &InferenceEvent) {
363 if !Self::dual_write_enabled() {
364 return;
365 }
366 let Some(onde_app_id) = event.onde_app_id.as_deref() else {
367 return;
368 };
369 let event_key = scoped_document_key(onde_app_id, &[&event.edge_id, &event.request_id]);
370 let document = inference_document(event);
371
372 self.write_doc("pulse_inference_events", Some(&event_key), &document)
373 .await;
374 }
375
376 /// Upsert (or append, when `key` is `None`) one document, logging failures.
377 async fn write_doc(&self, collection: &str, key: Option<&str>, doc: &serde_json::Value) {
378 if let Err(error) = self.inner.upsert_document(collection, key, doc).await {
379 log::warn!("pulse: dual-write to {} failed: {}", collection, error);
380 }
381 }
382 }
383
384 /// Parse an opt-in env flag: `1`, `true`, `yes` or `on`, case-insensitive.
385 fn flag_set(value: Option<&str>) -> bool {
386 matches!(
387 value.map(str::trim).map(str::to_ascii_lowercase).as_deref(),
388 Some("1") | Some("true") | Some("yes") | Some("on")
389 )
390 }
391
392 /// Keys are scoped inside Onde's shared GresIQ app so identical installation
393 /// ids belonging to different Onde customer apps cannot overwrite each other.
394 fn scoped_document_key(onde_app_id: &str, parts: &[&str]) -> String {
395 std::iter::once(onde_app_id)
396 .chain(parts.iter().copied())
397 .collect::<Vec<_>>()
398 .join(":")
399 }
400
401 /// Build the inference document. `ttft_ms` is omitted entirely when the path
402 /// could not measure it, so a reader can tell "not measured" from a real zero
403 /// — the relational route stores a hardcoded `0` and cannot make that
404 /// distinction.
405 fn inference_document(event: &InferenceEvent) -> serde_json::Value {
406 let mut document = serde_json::json!({
407 "edge_id": event.edge_id,
408 "model_id": event.model_id,
409 "request_id": event.request_id,
410 "duration_ms": event.duration_ms,
411 "status": event.status,
412 "onde_app_id": event.onde_app_id,
413 });
414
415 if let (Some(map), Some(ttft_ms)) = (document.as_object_mut(), event.ttft_ms) {
416 map.insert("ttft_ms".to_string(), serde_json::json!(ttft_ms));
417 }
418
419 document
420 }
421
422 /// `status` is always `"online"`: this document is only written when the edge
423 /// just loaded a model, and nothing reports a shutdown. `last_seen_ms` is what
424 /// makes that honest — a reader counts an edge as active by recency, the way
425 /// the relational dashboard windows on `last_seen_at`, rather than trusting a
426 /// status that is never revoked.
427 fn edge_document(event: &ModelLoadedEvent, onde_app_id: &str) -> serde_json::Value {
428 serde_json::json!({
429 "edge_id": event.edge_id,
430 "onde_app_id": onde_app_id,
431 "status": "online",
432 "last_seen_ms": epoch_millis(),
433 })
434 }
435
436 /// The spec 0008 raw observation. It holds nothing except the country: no
437 /// locale string, time zone, IP, or coordinates.
438 fn geography_document(
439 onde_app_id: &str,
440 edge_id: &str,
441 country_code: &str,
442 observed_at_ms: u64,
443 ) -> serde_json::Value {
444 serde_json::json!({
445 "onde_app_id": onde_app_id,
446 "edge_id": edge_id,
447 "country_code": country_code,
448 "observed_at_ms": observed_at_ms,
449 })
450 }
451
452 /// Accept exactly two ASCII letters and uppercase them. ISO 3166-1
453 /// user-assigned codes (`AA`, `QM`–`QZ`, `XA`–`XZ`, `ZZ`) name no country and
454 /// are rejected, except `XK`, which Apple and Android both report for Kosovo.
455 fn normalize_country_code(raw: &str) -> Option<String> {
456 let code = raw.trim().to_ascii_uppercase();
457 let [first, second] = code.as_bytes() else {
458 return None;
459 };
460 if !first.is_ascii_uppercase() || !second.is_ascii_uppercase() {
461 return None;
462 }
463 let user_assigned = code == "AA"
464 || code == "ZZ"
465 || (*first == b'Q' && *second >= b'M')
466 || (*first == b'X' && code != "XK");
467 (!user_assigned).then_some(code)
468 }
469
470 /// `YYYY-MM-DD` for a Unix-epoch millisecond timestamp, in UTC. This is
471 /// Howard Hinnant's `civil_from_days`, which avoids a date dependency.
472 fn utc_date(epoch_ms: u64) -> String {
473 let days = (epoch_ms / 86_400_000) as i64;
474 let z = days + 719_468;
475 let era = z.div_euclid(146_097);
476 let day_of_era = z.rem_euclid(146_097);
477 let year_of_era =
478 (day_of_era - day_of_era / 1_460 + day_of_era / 36_524 - day_of_era / 146_096) / 365;
479 let day_of_year = day_of_era - (365 * year_of_era + year_of_era / 4 - year_of_era / 100);
480 let shifted_month = (5 * day_of_year + 2) / 153;
481 let day = day_of_year - (153 * shifted_month + 2) / 5 + 1;
482 let month = if shifted_month < 10 {
483 shifted_month + 3
484 } else {
485 shifted_month - 9
486 };
487 let year = year_of_era + era * 400 + i64::from(month <= 2);
488 format!("{year:04}-{month:02}-{day:02}")
489 }
490
491 /// Milliseconds since the Unix epoch. The crate has no date dependency and
492 /// this is only ever read as a recency window, so an integer beats pulling in
493 /// a formatting library for an RFC 3339 string.
494 fn epoch_millis() -> u64 {
495 std::time::SystemTime::now()
496 .duration_since(std::time::UNIX_EPOCH)
497 .map(|elapsed| elapsed.as_millis() as u64)
498 .unwrap_or(0)
499 }
500
501 #[cfg(test)]
502 mod tests {
503 use super::*;
504
505 #[test]
506 fn env_flags_require_an_explicit_positive_value() {
507 for disabled in [
508 None,
509 Some(""),
510 Some("0"),
511 Some("false"),
512 Some("no"),
513 Some("off"),
514 ] {
515 assert!(!flag_set(disabled));
516 }
517 for enabled in [Some("1"), Some("true"), Some("YES"), Some(" on ")] {
518 assert!(flag_set(enabled));
519 }
520 }
521
522 #[test]
523 fn document_keys_include_the_onde_app_boundary() {
524 assert_eq!(
525 scoped_document_key("app-a", &["edge-1", "model-1"]),
526 "app-a:edge-1:model-1"
527 );
528 assert_ne!(
529 scoped_document_key("app-a", &["onde-unknown"]),
530 scoped_document_key("app-b", &["onde-unknown"])
531 );
532 // Request IDs collide across edges (each process counts from 0), so
533 // the inference key carries the edge.
534 assert_ne!(
535 scoped_document_key("app-a", &["edge-1", "onde-1720000000000-0"]),
536 scoped_document_key("app-a", &["edge-2", "onde-1720000000000-0"])
537 );
538 }
539
540 #[test]
541 fn edge_document_carries_cutover_state_and_tenant_id() {
542 let event = ModelLoadedEvent {
543 edge_id: "edge-1".to_string(),
544 model_id: "model-1".to_string(),
545 model_name: "Model 1".to_string(),
546 load_duration_ms: 42,
547 onde_app_id: Some("app-a".to_string()),
548 };
549 let document = edge_document(&event, "app-a");
550 assert_eq!(document["edge_id"], "edge-1");
551 assert_eq!(document["onde_app_id"], "app-a");
552 assert_eq!(document["status"], "online");
553 // Liveness has to be derivable from the document itself: nothing ever
554 // writes an offline status, so a reader windows on this instead.
555 assert!(document["last_seen_ms"].as_u64().unwrap_or(0) > 0);
556 }
557
558 #[test]
559 fn country_codes_are_two_letters_and_never_user_assigned() {
560 assert_eq!(normalize_country_code("SE").as_deref(), Some("SE"));
561 assert_eq!(normalize_country_code(" id ").as_deref(), Some("ID"));
562 assert_eq!(normalize_country_code("XK").as_deref(), Some("XK"));
563 // Full locale strings, numeric UN M.49 regions and time zones must be
564 // rejected rather than parsed: the host extracts the country.
565 for rejected in [
566 "",
567 "S",
568 "SWE",
569 "sv_SE",
570 "sv-SE",
571 "001",
572 "Europe/Stockholm",
573 "S1",
574 "ÅÄ",
575 ] {
576 assert_eq!(normalize_country_code(rejected), None, "{rejected}");
577 }
578 for user_assigned in ["AA", "QM", "QZ", "XA", "XX", "ZZ"] {
579 assert_eq!(
580 normalize_country_code(user_assigned),
581 None,
582 "{user_assigned}"
583 );
584 }
585 }
586
587 #[test]
588 fn utc_date_formats_civil_days() {
589 assert_eq!(utc_date(0), "1970-01-01");
590 assert_eq!(utc_date(1_709_164_800_000), "2024-02-29");
591 assert_eq!(utc_date(1_788_525_296_000), "2026-09-04");
592 // The last millisecond of a day still belongs to it.
593 assert_eq!(utc_date(86_400_000 - 1), "1970-01-01");
594 }
595
596 #[test]
597 fn geography_document_carries_only_the_country() {
598 let document = geography_document("app-a", "edge-1", "SE", 1_788_525_296_000);
599 let mut fields: Vec<_> = document
600 .as_object()
601 .map(|map| map.keys().cloned().collect())
602 .unwrap_or_default();
603 fields.sort();
604 assert_eq!(
605 fields,
606 ["country_code", "edge_id", "observed_at_ms", "onde_app_id"]
607 );
608 assert_eq!(document["country_code"], "SE");
609 }
610
611 #[test]
612 fn inference_document_omits_ttft_when_it_was_not_measured() {
613 let mut event = InferenceEvent {
614 edge_id: "edge-1".to_string(),
615 model_id: "model-1".to_string(),
616 request_id: "req-1".to_string(),
617 duration_ms: 400,
618 ttft_ms: None,
619 status: "success".to_string(),
620 onde_app_id: Some("app-a".to_string()),
621 };
622 assert!(inference_document(&event).get("ttft_ms").is_none());
623
624 event.ttft_ms = Some(90);
625 assert_eq!(inference_document(&event)["ttft_ms"], 90);
626 }
627 }