@ondeinference / onde / commits / 4b14e20

feat(pulse): measure TTFT and stamp edge liveness in dual-write

Two gaps found by reading the document payloads against the onde-web reader that consumes them. TTFT was never measured anywhere. The relational route stores a hardcoded `ttft_ms: 0` placeholder, so the dashboard's latency tiles have been showing aggregates over zeros. The streaming path is the only one that can observe a first token, so it now times the first content chunk and reports it; the blocking path sends None, and the document omits the field entirely rather than writing a zero that reads as a measurement. Streamed inferences were also invisible to pulse — only the blocking path reported one. stream_message now records its inference event, which raises reported volume but does not skew document-vs-relational parity, since both paths receive the same event. Edge documents carry last_seen_ms. Status is written as "online" at model-load and nothing ever revokes it, so a reader counting on status alone would report every edge that ever ran. The relational dashboard windows on last_seen_at; this gives the document reader the same basis. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Seto Elkahfi committed Sep 4, 2026 at 21:26 UTC 4b14e203cf4f7f5296e754c54c20a18e3848dd9a

3 files changed +119 -15

src/inference/engine.rs

+31 -2

@@ -1096,6 +1096,9 @@ impl ChatEngine {

1096 pulse_model_id,

1097 crate::pulse::next_request_id(),

1098 elapsed.as_millis() as u64,

1099 + // A blocking request returns the whole completion at once,

1100 + // so there is no first-token moment to measure here.

1101 + None,

1102 "success".to_string(),

1103 );

1104 }

@@ -1195,11 +1198,16 @@ impl ChatEngine {

1198 ) -> Result<tokio::sync::mpsc::Receiver<StreamChunk>, InferenceError> {

1199 let user_message = user_message.into();

1200

1198 - let (model, request) = {

1201 + let (model, request, pulse_model_id) = {

1202 let guard = self.inner.lock().await;

1203 let loaded = guard.as_ref().ok_or(InferenceError::NoModelLoaded)?;

1204 let request = self::build_request(loaded, &user_message);

1202 - (loaded.model.clone(), request)

1205 + let pulse_model_id = match &loaded.config {

1206 + LoadedModelConfig::Gguf(c) => c.model_id.clone(),

1207 + #[cfg(target_os = "macos")]

1208 + LoadedModelConfig::Isq(c) => c.model_id.clone(),

1209 + };

1210 + (loaded.model.clone(), request, pulse_model_id)

1211 };

1212

1213 let (tx, rx) = tokio::sync::mpsc::channel::<StreamChunk>(64);

@@ -1219,17 +1227,22 @@ impl ChatEngine {

1227 let inner_ref: &'static Mutex<Option<LoadedModel>> = unsafe { &*inner_ptr };

1228

1229 let user_msg_clone = user_message.clone();

1230 + // Cloning the pulse client is a pointer bump (Arc-backed reqwest

1231 + // client), so the task owns one rather than borrowing the engine.

1232 + let pulse = self.pulse().cloned();

1233

1234 tokio::task::spawn(async move {

1235 // `model` is an `Arc<Model>` moved into this task, so it's owned.

1236 // `stream_chat_request` borrows `&Model` — the borrow is scoped

1237 // to this async block and does NOT need to be `'static`.

1238 + let started = std::time::Instant::now();

1239 let stream_result = model.stream_chat_request(request).await;

1240

1241 match stream_result {

1242 Ok(mut stream) => {

1243 let mut assembled = String::new();

1244 let mut last_finish_reason = None;

1245 + let mut ttft_ms = None;

1246

1247 // `Stream::next()` is an inherent async method on

1248 // mistralrs::Stream that returns `Option<Response>`.

@@ -1239,6 +1252,9 @@ impl ChatEngine {

1252 mistralrs::Response::Chunk(chunk) => {

1253 if let Some(choice) = chunk.choices.first() {

1254 if let Some(ref text) = choice.delta.content {

1255 + ttft_ms.get_or_insert_with(|| {

1256 + started.elapsed().as_millis() as u64

1257 + });

1258 assembled.push_str(text);

1259 let _ = tx

1260 .send(StreamChunk {

@@ -1291,6 +1307,19 @@ impl ChatEngine {

1307 }

1308 }

1309

1310 + // Streamed inferences were invisible to pulse until now;

1311 + // only the blocking path reported. This is also the one

1312 + // path that can supply a real TTFT.

1313 + if let Some(pulse) = pulse {

1314 + pulse.record_inference(

1315 + pulse_model_id,

1316 + crate::pulse::next_request_id(),

1317 + started.elapsed().as_millis() as u64,

1318 + ttft_ms,

1319 + "success".to_string(),

1320 + );

1321 + }

1322 +

1323 // Send the final "done" chunk.

1324 let _ = tx

1325 .send(StreamChunk {

src/pulse/client.rs

+62 -13

@@ -196,6 +196,7 @@ impl PulseClient {

196 model_id: String,

197 request_id: String,

198 duration_ms: u64,

199 + ttft_ms: Option<u64>,

200 status: String,

201 ) {

202 let client = self.clone();

@@ -205,6 +206,7 @@ impl PulseClient {

206 model_id,

207 request_id,

208 duration_ms,

209 + ttft_ms,

210 status,

211 onde_app_id: client.onde_app_id.clone(),

212 };

@@ -318,20 +320,10 @@ impl PulseClient {

320 return;

321 };

322 let event_key = scoped_document_key(onde_app_id, &[&event.request_id]);

323 + let document = inference_document(event);

324

322 - self.write_doc(

323 - "pulse_inference_events",

324 - Some(&event_key),

325 - &serde_json::json!({

326 - "edge_id": event.edge_id,

327 - "model_id": event.model_id,

328 - "request_id": event.request_id,

329 - "duration_ms": event.duration_ms,

330 - "status": event.status,

331 - "onde_app_id": event.onde_app_id,

332 - }),

333 - )

334 - .await;

325 + self.write_doc("pulse_inference_events", Some(&event_key), &document)

326 + .await;

327 }

328

329 /// Upsert (or append, when `key` is `None`) one document, logging failures.

@@ -358,14 +350,51 @@ fn scoped_document_key(onde_app_id: &str, parts: &[&str]) -> String {

350 .join(":")

351 }

352

353 +/// Build the inference document. `ttft_ms` is omitted entirely when the path

354 +/// could not measure it, so a reader can tell "not measured" from a real zero

355 +/// — the relational route stores a hardcoded `0` and cannot make that

356 +/// distinction.

357 +fn inference_document(event: &InferenceEvent) -> serde_json::Value {

358 + let mut document = serde_json::json!({

359 + "edge_id": event.edge_id,

360 + "model_id": event.model_id,

361 + "request_id": event.request_id,

362 + "duration_ms": event.duration_ms,

363 + "status": event.status,

364 + "onde_app_id": event.onde_app_id,

365 + });

366 +

367 + if let (Some(map), Some(ttft_ms)) = (document.as_object_mut(), event.ttft_ms) {

368 + map.insert("ttft_ms".to_string(), serde_json::json!(ttft_ms));

369 + }

370 +

371 + document

372 +}

373 +

374 +/// `status` is always `"online"`: this document is only written when the edge

375 +/// just loaded a model, and nothing reports a shutdown. `last_seen_ms` is what

376 +/// makes that honest — a reader counts an edge as active by recency, the way

377 +/// the relational dashboard windows on `last_seen_at`, rather than trusting a

378 +/// status that is never revoked.

379 fn edge_document(event: &ModelLoadedEvent, onde_app_id: &str) -> serde_json::Value {

380 serde_json::json!({

381 "edge_id": event.edge_id,

382 "onde_app_id": onde_app_id,

383 "status": "online",

384 + "last_seen_ms": epoch_millis(),

385 })

386 }

387

388 +/// Milliseconds since the Unix epoch. The crate has no date dependency and

389 +/// this is only ever read as a recency window, so an integer beats pulling in

390 +/// a formatting library for an RFC 3339 string.

391 +fn epoch_millis() -> u64 {

392 + std::time::SystemTime::now()

393 + .duration_since(std::time::UNIX_EPOCH)

394 + .map(|elapsed| elapsed.as_millis() as u64)

395 + .unwrap_or(0)

396 +}

397 +

398 #[cfg(test)]

399 mod tests {

400 use super::*;

@@ -416,5 +445,25 @@ mod tests {

445 assert_eq!(document["edge_id"], "edge-1");

446 assert_eq!(document["onde_app_id"], "app-a");

447 assert_eq!(document["status"], "online");

448 + // Liveness has to be derivable from the document itself: nothing ever

449 + // writes an offline status, so a reader windows on this instead.

450 + assert!(document["last_seen_ms"].as_u64().unwrap_or(0) > 0);

451 + }

452 +

453 + #[test]

454 + fn inference_document_omits_ttft_when_it_was_not_measured() {

455 + let mut event = InferenceEvent {

456 + edge_id: "edge-1".to_string(),

457 + model_id: "model-1".to_string(),

458 + request_id: "req-1".to_string(),

459 + duration_ms: 400,

460 + ttft_ms: None,

461 + status: "success".to_string(),

462 + onde_app_id: Some("app-a".to_string()),

463 + };

464 + assert!(inference_document(&event).get("ttft_ms").is_none());

465 +

466 + event.ttft_ms = Some(90);

467 + assert_eq!(inference_document(&event)["ttft_ms"], 90);

468 }

469 }

src/pulse/events.rs

+26

@@ -32,6 +32,11 @@ pub struct InferenceEvent {

32 pub request_id: String,

33 /// Wall-clock time for the whole response, in milliseconds.

34 pub duration_ms: u64,

35 + /// Time to first token, in milliseconds. Only the streaming path can

36 + /// observe this; a blocking `send_message` gets the whole response at

37 + /// once and reports `None` rather than a stand-in value.

38 + #[serde(skip_serializing_if = "Option::is_none")]

39 + pub ttft_ms: Option<u64>,

40 /// `"success"`, `"cancelled"`, or `"error"`. Currently always

41 /// `"success"` here — inference errors throw before we reach this point.

42 pub status: String,

@@ -66,10 +71,31 @@ mod tests {

71 model_id: "org/repo".into(),

72 request_id: "onde-1-0".into(),

73 duration_ms: 42,

74 + ttft_ms: None,

75 status: "success".into(),

76 onde_app_id: None,

77 };

78 let value = serde_json::to_value(&event).expect("serialize");

79 assert!(value.get("onde_app_id").is_none());

80 }

81 +

82 + #[test]

83 + fn ttft_is_omitted_when_the_path_cannot_measure_it() {

84 + let mut event = InferenceEvent {

85 + edge_id: "edge-1".into(),

86 + model_id: "org/repo".into(),

87 + request_id: "onde-1-0".into(),

88 + duration_ms: 42,

89 + ttft_ms: None,

90 + status: "success".into(),

91 + onde_app_id: None,

92 + };

93 + assert!(serde_json::to_value(&event)

94 + .expect("serialize")

95 + .get("ttft_ms")

96 + .is_none());

97 +

98 + event.ttft_ms = Some(17);

99 + assert_eq!(serde_json::to_value(&event).expect("serialize")["ttft_ms"], 17);

100 + }

101 }