@ondeinference / onde / commits / 5973f46

Record a device-declared country for the Pulse globe

The public /pulse globe still reads cities that GresIQ geolocated from device IPs. Spec 0008 replaces that with a country the host reads from the OS locale or time zone, and this is the SDK half. A host passes the country with ChatEngine::with_pulse_country, or sets ONDE_PULSE_COUNTRY when it goes through UniFFI. The SDK keeps it only if it is two letters and a real ISO 3166-1 code. Placeholder codes (AA, QM to QZ, XA to XZ, ZZ) are dropped, except XK, which Apple and Android both use for Kosovo. Locale strings like sv_SE are rejected, not parsed. On model load the dual-write path writes one document to pulse_geography_observations with the app id, edge id, country and a timestamp. Like the other mirrored collections it needs ONDE_PULSE_DUAL_WRITE and an app id. ONDE_DISABLE_PULSE_GEOGRAPHY turns off this write alone. The document key ends in the UTC date, so repeat loads on the same day overwrite one document, across restarts too. The spec describes an append-only collection with an SDK-side rate limit. A keyed upsert gets the same one-per-day result without in-memory state, and the gateway keeps created_at on upsert, which is what the 30-day sweep ages on. Country stays off pulse_edges because that document never expires. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Seto Elkahfi committed Sep 25, 2026 at 07:40 UTC 5973f46147f92c1c11b147f997207531f0c9c59b

3 files changed +203 -16

src/inference/engine.rs

+28 -1

@@ -265,6 +265,9 @@ pub struct ChatEngine {

265 /// pulse event so the document store associates rows with the right app.

266 /// `None` for open-ended SDK consumers (UniFFI `OndeChatEngine`).

267 onde_app_id: Option<String>,

268 + /// Device-declared ISO 3166-1 alpha-2 country for the public Pulse globe

269 + /// (spec 0008). `None` writes no geography.

270 + pulse_country: Option<String>,

271 }

272

273 #[cfg(any(

@@ -297,9 +300,22 @@ impl ChatEngine {

300 inner: Mutex::new(None),

301 pulse: std::sync::OnceLock::new(),

302 onde_app_id,

303 + pulse_country: None,

304 }

305 }

306

307 + /// Declare the device's country for Pulse geography. Pass the ISO 3166-1

308 + /// alpha-2 region from the OS locale or time zone (for example `"SE"`),

309 + /// never a full locale string or anything derived from the network.

310 + ///

311 + /// Only the country code is sent, as one observation per edge per UTC

312 + /// day, and only when dual-write and an Onde app id are both enabled.

313 + /// UniFFI hosts can set `ONDE_PULSE_COUNTRY` instead.

314 + pub fn with_pulse_country(mut self, country_code: Option<String>) -> Self {

315 + self.pulse_country = country_code;

316 + self

317 + }

318 +

319 /// Lazily initialise the pulse telemetry client.

320 ///

321 /// The underlying GresIQ SDK creates a `reqwest::Client`, which needs

@@ -320,7 +336,13 @@ impl ChatEngine {

336 .ok()

337 .filter(|id| !id.is_empty())

338 });

323 - let client = crate::pulse::PulseClient::new(environment, edge_id, onde_app_id);

339 + let country_code = self.pulse_country.clone().or_else(|| {

340 + std::env::var("ONDE_PULSE_COUNTRY")

341 + .ok()

342 + .filter(|code| !code.is_empty())

343 + });

344 + let client =

345 + crate::pulse::PulseClient::new(environment, edge_id, onde_app_id, country_code);

346

347 match &client {

348 Some(_) => log::info!(

@@ -2121,6 +2143,11 @@ impl ChatEngine {

2143 Self

2144 }

2145

2146 + /// Stub mirror of the real `with_pulse_country`; no telemetry is sent.

2147 + pub fn with_pulse_country(self, _country_code: Option<String>) -> Self {

2148 + self

2149 + }

2150 +

2151 pub async fn load_gguf_model(

2152 &self,

2153 _config: GgufModelConfig,

src/pulse/client.rs

+169 -15

@@ -24,20 +24,15 @@ 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 }

32 impl PulseClient {

33 /// Returns true when pulse telemetry is disabled explicitly by the host app.

34 pub fn disabled_by_env() -> bool {

32 - matches!(

33 - std::env::var("ONDE_DISABLE_PULSE")

34 - .ok()

35 - .as_deref()

36 - .map(str::trim)

37 - .map(str::to_ascii_lowercase)

38 - .as_deref(),

39 - Some("1") | Some("true") | Some("yes") | Some("on")

40 - )

35 + flag_set(std::env::var("ONDE_DISABLE_PULSE").ok().as_deref())

36 }

38 /// ADR-0003 M2 dual-write toggle. The mirror is deliberately opt-in while

@@ -47,7 +42,18 @@ impl PulseClient {

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 {

50 - dual_write_enabled_value(std::env::var("ONDE_PULSE_DUAL_WRITE").ok().as_deref())

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 }

59 /// Build a pulse client using the GresIQ credentials embedded in the SDK.

@@ -62,10 +68,15 @@ impl PulseClient {

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;

@@ -115,10 +126,19 @@ impl PulseClient {

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

@@ -246,7 +266,7 @@ impl PulseClient {

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

249 - /// Move the three model-load mirror requests off the model-loading critical

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.

@@ -306,9 +326,34 @@ impl PulseClient {

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

@@ -336,7 +381,8 @@ impl PulseClient {

381 }

382 }

383

339 -fn dual_write_enabled_value(value: Option<&str>) -> bool {

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")

@@ -387,6 +433,61 @@ fn edge_document(event: &ModelLoadedEvent, onde_app_id: &str) -> serde_json::Val

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.

@@ -402,7 +503,7 @@ mod tests {

503 use super::*;

504

505 #[test]

405 - fn dual_write_requires_an_explicit_positive_flag() {

506 + fn env_flags_require_an_explicit_positive_value() {

507 for disabled in [

508 None,

509 Some(""),

@@ -411,10 +512,10 @@ mod tests {

512 Some("no"),

513 Some("off"),

514 ] {

414 - assert!(!dual_write_enabled_value(disabled));

515 + assert!(!flag_set(disabled));

516 }

517 for enabled in [Some("1"), Some("true"), Some("YES"), Some(" on ")] {

417 - assert!(dual_write_enabled_value(enabled));

518 + assert!(flag_set(enabled));

519 }

520 }

521

@@ -454,6 +555,59 @@ mod tests {

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 {

src/pulse/mod.rs

@@ -8,6 +8,12 @@

8 //! machine gets called in the dashboard. Anything works — just keep it stable

9 //! across restarts or you'll end up with duplicate edges. Defaults to

10 //! `"onde-unknown"`.

11 +//!

12 +//! `ONDE_PULSE_COUNTRY` (or `ChatEngine::with_pulse_country`) declares the

13 +//! device's ISO 3166-1 alpha-2 country for the public Pulse globe. It is only

14 +//! written with `ONDE_PULSE_DUAL_WRITE` and an Onde app id, and

15 +//! `ONDE_DISABLE_PULSE_GEOGRAPHY=1` turns it off on its own. Leave it unset

16 +//! and no location is ever sent.

18 mod client;

19 mod events;