diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/Api.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/Api.kt index 58b9fee..d353440 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/Api.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/Api.kt @@ -149,6 +149,10 @@ data class SessionSummary( * on when a backend is too old to say, which matches what that backend actually does. */ val notify: Boolean, + /** + * Every token this session has spent, as the server counts it -- see `SessionEvent.UsageDelta`. + */ + val totalTokens: Long, val status: String, val lastActivity: Double, ) @@ -165,6 +169,7 @@ private fun parseSession(session: JSONObject) = permissionMode = session.optString("permissionMode").ifEmpty { null }, imported = session.optBoolean("imported", false), notify = session.optBoolean("notify", true), + totalTokens = session.optLong("totalTokens", 0), status = session.getString("status"), lastActivity = session.getDouble("lastActivity"), ) diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/Events.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/Events.kt index 11a6df7..ee245db 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/Events.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/Events.kt @@ -102,7 +102,15 @@ sealed class SessionEvent { */ data class Settings(val model: String?, val permissionMode: String?) : SessionEvent() - data class UsageDelta(val tokens: Long) : SessionEvent() + /** + * What a turn cost, and what the session has cost in total. + * + * [total] is the server's running figure, carried on the event so a reader never adds up its + * own: a phone opens a session on the newest page of the transcript, so a sum it computed would + * be that page's share of the conversation wearing the whole conversation's label. Zero on + * entries recorded before the backend sent it. + */ + data class UsageDelta(val tokens: Long, val total: Long) : SessionEvent() /** * A compaction that finished, and how much context it recovered. @@ -200,7 +208,8 @@ fun parseSeqEvent(json: String): SeqEvent { model = body.optString("model").ifEmpty { null }, permissionMode = body.optString("permissionMode").ifEmpty { null }, ) - "usageDelta" -> SessionEvent.UsageDelta(body.getLong("tokens")) + "usageDelta" -> + SessionEvent.UsageDelta(body.getLong("tokens"), body.optLong("total", 0)) "compacted" -> SessionEvent.Compacted( preTokens = if (body.has("preTokens")) body.getLong("preTokens") else null, diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/SessionListScreen.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/SessionListScreen.kt index 8b1c383..87c8c72 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/SessionListScreen.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/SessionListScreen.kt @@ -263,11 +263,12 @@ private fun SessionCard( Spacer(Modifier.height(4.dp)) Row(modifier = Modifier.fillMaxWidth()) { Text( - // Provider, then where it runs -- "on " rather - // than a bare name, so a host isn't mistaken for a model. + // Machine, then what runs on it, then what it is set to: the same order + // and separator as the session screen's header and the usage dialog, so + // one pair of facts is not written three ways. listOfNotNull( + session.setupName, session.provider, - "on ${session.setupName}", session.model?.let { modelLabel(it) }, ) .joinToString(" · "), diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/SessionScreen.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/SessionScreen.kt index 5225d0d..893f801 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/SessionScreen.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/SessionScreen.kt @@ -366,7 +366,10 @@ fun SessionScreen( val scope = rememberCoroutineScope() var items by remember { mutableStateOf(listOf()) } var status by remember { mutableStateOf(summary.status) } - var totalTokens by remember { mutableLongStateOf(0L) } + // Seeded from the row this screen was opened from, so a conversation that has spent + // something says so before any turn happens here. Zero used to mean "nothing yet" and + // "nothing in the page I loaded" at once, and the second one is most of the long sessions. + var totalTokens by remember(summary.id) { mutableLongStateOf(summary.totalTokens) } // When this screen saw the current compaction start, on this device's own clock, and how long // ago that is. See `compactingLabel`: null is the honest answer whenever the start was not // witnessed here, which is what opening a session that is already compacting looks like. @@ -451,7 +454,13 @@ fun SessionScreen( moreHistory = entry.seq > 1L } when (val event = entry.event) { - is SessionEvent.UsageDelta -> totalTokens += event.tokens + // Taken, not accumulated: the server's running total is on the event, and adding + // up the deltas this screen happened to receive counted one page of a conversation + // and called it the whole. `max` because pages arrive in no guaranteed order and an + // older event's total is a smaller true answer, never a correction downwards; it + // also leaves the seeded figure alone for transcripts recorded before the backend + // sent a total at all. + is SessionEvent.UsageDelta -> totalTokens = maxOf(totalTokens, event.total) else -> { // What the session says it is set to now, which is the only thing that // says it: picking from either menu asks, and the answer comes back here. @@ -795,15 +804,18 @@ fun SessionScreen( Spacer(Modifier.width(8.dp)) Column(Modifier.weight(1f)) { Text(title, style = MaterialTheme.typography.titleMedium) + // Machine first, then what runs on it -- the same order and the same wording + // everywhere this pair appears, so it reads as one fact rather than as two + // sentences with different grammar. The "on" that used to sit in the middle + // made it a phrase, which only works in one order and stops working the moment + // the pair is shown anywhere else. + // + // No model. The picker in the footer already shows what this session is set to, + // and showing it twice means two things to keep in step -- they disagreed for a + // moment on every model change, since one follows the request and the other the + // session's own answer. Text( - listOfNotNull( - summary.provider, - "on ${summary.setupName}", - // What the session says it is set to now, which is the same fact - // the picker below shows and has to be the same answer. - model?.let { modelLabel(it) }, - ) - .joinToString(" · "), + "${summary.setupName} · ${summary.provider}", style = MaterialTheme.typography.bodySmall, color = MaterialTheme.colorScheme.onSurfaceVariant, ) diff --git a/server/src/session/claude/translate.rs b/server/src/session/claude/translate.rs index 11c83ec..9f6d258 100644 --- a/server/src/session/claude/translate.rs +++ b/server/src/session/claude/translate.rs @@ -222,7 +222,7 @@ impl Translator { }); } if tokens > 0 { - events.push(Event::UsageDelta { tokens }); + events.push(Event::UsageDelta { tokens, total: 0 }); } events.push(Event::Status { state: SessionStatus::Idle, @@ -1052,7 +1052,10 @@ mod tests { assert_eq!( events, vec![ - Event::UsageDelta { tokens: 182 }, + Event::UsageDelta { + tokens: 182, + total: 0 + }, Event::Status { state: SessionStatus::Idle }, diff --git a/server/src/session/driver.rs b/server/src/session/driver.rs index 553088a..e7695b4 100644 --- a/server/src/session/driver.rs +++ b/server/src/session/driver.rs @@ -230,6 +230,22 @@ pub enum Event { /// Per-turn token counts, where the dialect reports them. UsageDelta { tokens: u64, + /// Every token this session has spent, this turn included. + /// + /// Filled in by the pump, not by drivers: a driver reports what its + /// turn cost, and only the pump sees all of them. Carried on the + /// event rather than left to be added up by whoever is reading, + /// because a reader has only *part* of the transcript -- a phone + /// opens a session on the newest page -- so a total it summed + /// itself would be the newest page's total wearing the whole + /// conversation's label. Worse when the page has no turn in it at + /// all: the count then reads zero, and a zero is drawn as nothing. + /// + /// Zero on entries written before this existed, which is why the + /// pump seeds its running total by adding up `tokens` at startup + /// rather than reading the last of these. + #[serde(default)] + total: u64, }, /// A compaction that finished, and how much context it recovered. /// diff --git a/server/src/session/echo.rs b/server/src/session/echo.rs index a64d60d..23fbd1e 100644 --- a/server/src/session/echo.rs +++ b/server/src/session/echo.rs @@ -446,6 +446,7 @@ impl EchoDriver { } send(Event::UsageDelta { tokens: text.split_whitespace().count() as u64, + total: 0, }); finish(); }); diff --git a/server/src/session/llama.rs b/server/src/session/llama.rs index 0378727..39c9dda 100644 --- a/server/src/session/llama.rs +++ b/server/src/session/llama.rs @@ -608,7 +608,7 @@ fn generate( } } if tokens > 0 { - let _ = sink.send(Event::UsageDelta { tokens }); + let _ = sink.send(Event::UsageDelta { tokens, total: 0 }); } Ok(()) } @@ -712,7 +712,10 @@ mod tests { Event::AssistantText { delta: "still here".into(), }, - Event::UsageDelta { tokens: 12 }, + Event::UsageDelta { + tokens: 12, + total: 12, + }, ]); let messages = conversation(&path); assert_eq!(messages.len(), 2); diff --git a/server/src/session/mod.rs b/server/src/session/mod.rs index 337d47d..98b9217 100644 --- a/server/src/session/mod.rs +++ b/server/src/session/mod.rs @@ -143,6 +143,9 @@ pub struct SessionInfo { pub imported: bool, #[serde(skip_serializing_if = "Option::is_none")] pub cwd: Option, + /// Every token this session has spent, so a phone showing a total does + /// not have to add up a transcript it only holds part of. + pub total_tokens: u64, /// Whether this session announces itself -- reported for the same /// reason `permission_mode` is: a switch that guesses its own position /// is how you turn something off while believing you are reading it. @@ -253,6 +256,12 @@ struct Shared { /// session was *launched* with, so reporting from it would show the /// mode a change had already replaced. permission_mode: Mutex>, + /// Every token this session has spent. + /// + /// Kept here because only the pump sees every turn, and reported on the + /// session row so a phone opening a long conversation has the real + /// figure rather than the newest page's share of it. + total_tokens: Mutex, /// Whether this session's attention-wanting moments are announced. /// /// Mirrored out of the config so the pump can read it without taking @@ -370,6 +379,7 @@ impl LiveSession { title: self.shared.title.lock().unwrap().clone(), model: self.shared.model.lock().unwrap().clone(), permission_mode: self.shared.permission_mode.lock().unwrap().clone(), + total_tokens: *self.shared.total_tokens.lock().unwrap(), notify: *self.shared.notify.lock().unwrap(), imported, keeps_own_transcript, @@ -706,6 +716,7 @@ impl SessionManager { title: meta.title.clone(), model: meta.model.clone(), permission_mode: meta.permission_mode.clone(), + total_tokens: 0, notify: meta.notify, imported: import::read_cursor(&self.data_dir.join(&meta.id)).is_some(), keeps_own_transcript: keeps_own_transcript( @@ -1213,9 +1224,14 @@ fn launch( // this is then the only true answer available. status: Mutex::new(transcript.last_status().unwrap_or(SessionStatus::Idle)), title: Mutex::new(meta.title.clone()), - last_activity: Mutex::new(now()), + // What the transcript last recorded, not the clock: this server has + // just been told nothing, and `now()` claimed every relaunched + // session had been active this instant -- see + // `Transcript::last_activity`. + last_activity: Mutex::new(transcript.last_activity().unwrap_or_else(now)), model: Mutex::new(meta.model.clone()), permission_mode: Mutex::new(meta.permission_mode.clone()), + total_tokens: Mutex::new(transcript.total_tokens()), notify: Mutex::new(meta.notify), written: Mutex::new(0), }); @@ -1348,6 +1364,18 @@ async fn pump( // for where a user's message sits: where the session read it. let event = match event { Event::MessageTaken { id, text } => Event::UserMessage { id, text }, + // The running total is the pump's to keep, for the reason the + // field gives: a driver knows what its own turn cost and + // nothing else does. Added here rather than at each driver so + // a new one cannot get it wrong by leaving it out. + Event::UsageDelta { tokens, .. } => { + let mut total = shared.total_tokens.lock().unwrap(); + *total += tokens; + Event::UsageDelta { + tokens, + total: *total, + } + } other => other, }; // Nothing changed, so there is nothing to record. Both of these @@ -1893,6 +1921,115 @@ mod tests { ))); } + #[tokio::test] + async fn a_restart_reports_when_a_session_last_did_something() { + let dir = tempfile::tempdir().expect("tempdir"); + let config_path = dir.path().join("config.ron"); + let data_dir = dir.path().join("sessions"); + seed_echo_only(&config_path); + + let manager = SessionManager::new( + config_path.clone(), + data_dir.clone(), + data_dir.join("models"), + ) + .expect("manager"); + let info = manager.spawn_session(echo_spec()).expect("spawn"); + let session = manager.session(&info.id).expect("live session"); + let mut rx = session.subscribe(); + session.send_message("something".to_string(), Vec::new()); + collect_turn(&mut rx).await; + let before_restart = manager.sessions()[0].last_activity; + drop(rx); + drop(session); + drop(manager); + + // Far enough back that a restart taking the clock cannot pass by + // being fast: the assertion is about which source was used, not + // about how long the test took. + let long_ago = before_restart - 86_400.0; + rewrite_transcript_times(&data_dir.join(&info.id).join("transcript.jsonl"), long_ago); + + let manager = SessionManager::new(config_path, data_dir.clone(), data_dir.join("models")) + .expect("manager restart"); + let listed = manager.sessions(); + assert_eq!(listed.len(), 1); + assert!( + (listed[0].last_activity - long_ago).abs() < 1.0, + "a relaunched session reported {} instead of the {long_ago} its transcript records \ + -- every row would read \"just now\" and the list would sort by nothing", + listed[0].last_activity, + ); + } + + /// The total covers the whole conversation, not the part a reader holds. + /// + /// The bug this fixes was invisible in exactly the way that matters: a + /// phone opens a session on its newest page and used to add up the + /// `UsageDelta`s it found there, so a long conversation reported its + /// last few turns as the total -- and a page with no turn in it at all + /// reported nothing, since zero is drawn as blank. Both readings looked + /// like an answer. + #[tokio::test] + async fn the_token_total_covers_the_whole_conversation() { + let dir = tempfile::tempdir().expect("tempdir"); + let config_path = dir.path().join("config.ron"); + let data_dir = dir.path().join("sessions"); + seed_echo_only(&config_path); + let manager = SessionManager::new( + config_path.clone(), + data_dir.clone(), + data_dir.join("models"), + ) + .expect("manager"); + let info = manager.spawn_session(echo_spec()).expect("spawn"); + let session = manager.session(&info.id).expect("live"); + + // Echo charges a token a word, so the arithmetic is checkable. + let mut rx = session.subscribe(); + session.send_message("one two three".to_string(), Vec::new()); + collect_turn(&mut rx).await; + session.send_message("four five".to_string(), Vec::new()); + collect_turn(&mut rx).await; + let running = manager.sessions()[0].total_tokens; + assert_eq!(running, 5, "two turns of three and two words"); + + // The event carries it too, so a phone never has to add up its own. + let last_total = transcript::read_after(session.transcript_path(), 0) + .expect("transcript") + .iter() + .rev() + .find_map(|entry| match entry.event { + Event::UsageDelta { total, .. } => Some(total), + _ => None, + }) + .expect("a usage event"); + assert_eq!(last_total, running); + + // And a restart picks it up from the file rather than starting over. + drop(rx); + drop(session); + drop(manager); + let manager = SessionManager::new(config_path, data_dir.clone(), data_dir.join("models")) + .expect("manager restart"); + assert_eq!(manager.sessions()[0].total_tokens, running); + } + + /// Backdates every line in a transcript, so a restart has something to + /// report that the clock could not have produced. + fn rewrite_transcript_times(path: &Path, ts: f64) { + let text = std::fs::read_to_string(path).expect("read transcript"); + let rewritten: String = text + .lines() + .map(|line| { + let mut entry: serde_json::Value = serde_json::from_str(line).expect("line"); + entry["ts"] = serde_json::json!(ts); + format!("{entry}\n") + }) + .collect(); + std::fs::write(path, rewritten).expect("write transcript"); + } + #[tokio::test] async fn a_restart_relaunches_sessions_and_continues_the_numbering() { let dir = tempfile::tempdir().expect("tempdir"); diff --git a/server/src/session/transcript.rs b/server/src/session/transcript.rs index ddc7cd7..7e022ad 100644 --- a/server/src/session/transcript.rs +++ b/server/src/session/transcript.rs @@ -31,14 +31,16 @@ pub struct Transcript { file: File, next_seq: u64, last_status: Option, + last_activity: Option, + total_tokens: u64, } impl Transcript { /// Opens (or creates) the log at `path`, continuing the sequence from /// the last line if one exists. pub fn open(path: &Path) -> Result { - // One pass for both answers. They are wanted at the same moment by - // the same caller, and reading the file twice to get them doubled + // One pass for all three answers. They are wanted at the same moment + // by the same caller, and reading the file again for each doubled // the cost of starting every session -- which is paid per session, // at the point a restart is trying to be quick. let existing = read_after(path, 0)?; @@ -59,6 +61,17 @@ impl Transcript { file, next_seq: last_seq + 1, last_status, + last_activity: existing.last().map(|entry| entry.ts), + // Added up rather than read off the newest entry: `total` is a + // later addition, so a transcript written before it has zero on + // every line while `tokens` was always there. + total_tokens: existing + .iter() + .map(|entry| match entry.event { + Event::UsageDelta { tokens, .. } => tokens, + _ => 0, + }) + .sum(), }) } @@ -76,6 +89,33 @@ impl Transcript { self.last_status } + /// When this session last did anything, as of opening. + /// + /// Read from the file for the same reason [`Transcript::last_status`] + /// is, and it is the same mistake in the other direction: a restarting + /// server has been told nothing, and taking the clock instead said every + /// session it relaunched had been active this second. On the phone that + /// is every row reading "just now" and the list -- which is sorted by + /// this -- coming back in an order that means nothing, with the + /// conversation somebody was in the middle of buried among sessions + /// untouched for days. + /// + /// `None` for a transcript with no lines in it, which is a session that + /// genuinely has not done anything yet; its caller uses the clock, which + /// is right there and only there. + pub fn last_activity(&self) -> Option { + self.last_activity + } + + /// Everything this session has spent, as of opening. + /// + /// Zero for an empty transcript, which is a session that has spent + /// nothing -- the one case where zero is the answer rather than the + /// absence of one. + pub fn total_tokens(&self) -> u64 { + self.total_tokens + } + /// Appends `event`, assigning it the next sequence number. Flushed per /// event: each line is tiny, and the transcript is the source of truth /// a crash must not lose the tail of. @@ -352,7 +392,10 @@ mod tests { Event::Status { state: SessionStatus::Idle, }, - Event::UsageDelta { tokens: 42 }, + Event::UsageDelta { + tokens: 42, + total: 0, + }, Event::Error { message: "boom".into(), },