//! The common event model and the `Driver` trait -- the one abstraction //! everything hangs off (see PLAN.md). //! //! A driver translates its child process's JSONL dialect into [`Event`]s //! and accepts the small inbound vocabulary below. The transcript, the SSE //! stream, and the phone UI work purely in this model; nothing downstream //! of a driver may branch on the session kind. use serde::{Deserialize, Serialize}; use tokio::sync::mpsc; /// The name a session's image is stored and served under -- returned by /// `POST /attachments` for an upload, minted by a driver for one a tool /// produced, and fetched back from `/sessions/{id}/files/{ref}`. Both /// directions use the one id so the transcript renders them identically. pub type ImageRef = String; /// One choice offered in answer to a [`Event::Question`]. /// /// More than a label because the reader is deciding, not confirming: what /// an option means, and what picking it would produce, are the things that /// decide it. Both are optional -- a permission's Allow and Deny mean /// exactly what they say. #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct QuestionOption { pub label: String, /// A sentence about what this option means. #[serde(default, skip_serializing_if = "Option::is_none")] pub description: Option, /// A block to show as written -- a mockup, a diff, a config file. #[serde(default, skip_serializing_if = "Option::is_none")] pub preview: Option, } impl QuestionOption { /// An option that is only its label, which is most of them. pub fn plain(label: impl Into) -> Self { Self { label: label.into(), description: None, preview: None, } } } /// Everything a session can tell the outside world. Every event is /// appended to the session's transcript with a sequence number, then fanned /// out to SSE subscribers; the phone renders purely from this stream, so /// reconnecting is just "events after seq N" -- no separate history path /// to drift from the live one. #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] // `rename_all` renames the variants; `rename_all_fields` renames what is // inside them. Both are needed and only the first is obvious: every field // here was a single lowercase word until `pre_tokens` arrived, so a // multi-word field went out as snake_case, the app looked for camelCase and // found nothing, and the event still rendered -- as the "no counts were // reported" case, which is a state it is allowed to be in. A wire mismatch // that lands on a plausible state is invisible; anything added below with a // two-word field would have hit the same thing. #[serde( tag = "type", rename_all = "camelCase", rename_all_fields = "camelCase" )] pub enum Event { /// What the user sent, written into the transcript by the manager (not /// by drivers) so every device renders the full conversation from the /// one stream. Recorded when the session reads the message, which is /// what `MessageTaken` reports. UserMessage { /// The [`Event::MessageQueued`] this resolves, when it waited. /// /// A message sent between turns is read at once and never queued, /// so this is `None` for most of them. It is the pair to the id on /// `MessageQueued` and exists for the same reason `CommandSent` /// carries one: the phone has a bubble on screen for the waiting /// message and needs to know *which* one this is, rather than /// matching on the text and clearing the wrong one when the same /// thing was sent twice. #[serde(default, skip_serializing_if = "Option::is_none")] id: Option, text: String, /// What was attached to it, by the ref the files route serves. /// /// On the message rather than beside it. These used to be their own /// `Image` events emitted just before, which drew a person's /// screenshot as a row of its own floating above the bubble that /// sent it -- and left the phone to decide, from nothing but /// adjacency, which message an image belonged to. Belonging is not /// something to infer when the sender knew. #[serde(default, skip_serializing_if = "Vec::is_empty")] images: Vec, }, /// A message accepted from the phone that the session cannot read yet. /// /// Recorded, unlike the message itself, and that difference is the /// point. The *message* belongs in the transcript where the session /// read it -- see `MessageTaken` -- but something has to say it is /// waiting, and it has to be the server that says it: the phone used /// to remember its own outgoing messages, so leaving the session /// screen or restarting the app showed nothing pending when something /// was, which reads as "nothing queued" rather than "I have forgotten". /// /// Carries no row of its own. It is resolved by the `UserMessage` /// bearing the same id, exactly as `CommandQueued` is resolved by /// `CommandSent`. MessageQueued { id: String, text: String, /// Carried for the same reason [`Event::UserMessage`] carries it, /// and it matters more here: a waiting message is on screen for as /// long as the turn runs, so its attachment has nowhere else to be. #[serde(default, skip_serializing_if = "Vec::is_empty")] images: Vec, }, /// A message taken out of the queue before the session read it, by /// somebody tapping the bubble that was waiting for it. /// /// Recorded for the same reason `MessageQueued` is: the queue is the /// server's, so what is waiting has to be answerable from the /// transcript alone. Without it a phone that reconnects replays the /// `MessageQueued` and puts back a bubble for a message that will /// never arrive -- and nothing later would ever resolve it, since the /// `UserMessage` that normally does is exactly what is not coming. /// /// Only ever sent for a message that had not been handed over. One /// that has is not droppable and says so instead; see /// [`Unqueued::AlreadySent`]. MessageDropped { id: String, }, /// A driver has taken one of the user's messages and started reading /// it. The manager turns this into the `UserMessage` above, so it /// never reaches a phone itself. /// /// It exists because sending and being read are not the same moment. A /// message sent into a running turn waits for that turn to finish, and /// until then the session has not seen it -- so recording it among /// things already read puts it in the transcript above output that /// predates it, and leaves a phone drawing it as still waiting with /// nothing coming to say otherwise. MessageTaken { /// The `MessageQueued` this answers, or `None` when it never /// waited. Carried through onto the `UserMessage`. id: Option, text: String, /// Carried through onto the `UserMessage` with everything else. #[serde(default, skip_serializing_if = "Vec::is_empty")] images: Vec, }, /// Streaming assistant text; the phone renders the concatenation as /// markdown. AssistantText { delta: String, }, ToolStart { id: String, tool: String, input: serde_json::Value, }, ToolUpdate { id: String, output: String, }, ToolEnd { id: String, output: String, }, /// An image the session produced or was sent, saved under the session /// dir and referenced by id; the phone fetches it by URL. Image { #[serde(rename = "ref")] image: ImageRef, /// The tool call whose result carried it, when one did. /// /// A screenshot belongs under the call that took it, not floating /// beside it -- the reader has to pair them by position otherwise, /// and position is exactly what a page boundary breaks. `None` for /// an image a person attached to their own message, which belongs /// to no call. #[serde(default, skip_serializing_if = "Option::is_none")] about: Option, }, /// Anything the session needs a human for: AskUserQuestion, and /// permission requests, are the same shape with different options. Question { id: String, prompt: String, /// A few words naming what the question is about, when the asker /// offered one -- a tag beside the question rather than part of /// it. `None` for a permission, which is about the call above it. #[serde(default, skip_serializing_if = "Option::is_none")] header: Option, options: Vec, /// Whether several options may be chosen at once. /// /// Here rather than left for a phone to work out from the dialect /// underneath: how many answers a question takes is a fact about /// the question, and the alternative was the app parsing Claude /// Code's tool input to find out -- one dialect's schema, written /// out a second time in Kotlin, where no other dialect could /// reach it. #[serde(default, skip_serializing_if = "std::ops::Not::not")] multi_select: bool, /// The tool call this is permission for, when it is one. /// /// The CLI's `can_use_tool` request carries the `tool_use_id` of /// the call it is asking about, so a phone can draw the ask on the /// tool's own row rather than as a second card repeating its /// input. `None` for anything that is not about a tool -- /// AskUserQuestion, and an echo session's question. #[serde(default, skip_serializing_if = "Option::is_none")] about: Option, }, /// A message another agent sent this session. /// /// Its own kind rather than a `UserMessage`, because it is not /// something the reader said and a transcript that renders it in their /// voice is claiming they did. It also explains what would otherwise /// be inexplicable: a session that starts working on something nobody /// on this phone asked for. PeerMessage { /// The sending session's own name, which is what the reader /// recognises it by -- the socket path it came from is not. from: String, text: String, /// The seq of the `Status::Running` that opened the turn this /// message started, so a reader can draw it above that turn. /// /// It exists because the live Claude Code path cannot record the /// message where it belongs. The CLI says nothing about a peer /// message until the turn's `result` -- see /// `claude::translate` -- so the event is appended after /// everything it caused, and an append-only transcript cannot go /// back and insert it. Carrying the position instead keeps one /// order on the wire and one order on screen without a second /// source for either. /// /// Filled in by the pump, which is the only place that knows a /// seq, and only where a turn was open: `None` for a message read /// out of a session file by `import`, which already has it in the /// right place, and for one that started no turn. #[serde(default, skip_serializing_if = "Option::is_none")] turn_start: Option, }, /// The manager's record of a question being answered, so a rendered /// question card resolves on every device, not just the one that /// answered it. /// /// A list because a question can take several answers, and one that /// took one is the list of length one rather than a different shape. /// What a dialect makes of that -- Claude Code's answers map holds a /// string, so several become one line -- is that dialect's business /// and is done where it talks to it. Answered { id: String, answers: Vec, }, Status { state: SessionStatus, }, /// What the session is set to, as the session itself reports it. /// /// Asking for a change and having one are different things, and only /// this one is a measurement: a model name the dialect does not know, /// a mode it refuses, or a driver whose model is fixed at startup all /// leave a request that was sent and nothing that changed. Reporting /// from the request instead put the answer on the phone before the /// question had been answered, and left it there when the answer was /// no. /// /// Either field alone, because the two are confirmed separately and /// by different things -- the CLI echoes a mode change, and names the /// model it resolved an alias to when a session starts. Settings { #[serde(default, skip_serializing_if = "Option::is_none")] model: Option, #[serde(default, skip_serializing_if = "Option::is_none")] permission_mode: Option, }, /// Per-turn token counts, where the dialect reports them. UsageDelta { /// What this turn cost: the tokens it was charged for. tokens: u64, /// What the model was holding when the turn ended -- see /// [`context_tokens`] for what goes into it. /// /// Carried on the event rather than summed by whoever is reading, /// because it is not a sum: a conversation's context goes *down* /// at a compaction and a clear, so adding turns up would report a /// figure the session stopped being true of long ago. It is also /// the number a reader is asking about -- how much room is left /// before the next compaction -- rather than what has been spent /// getting here. /// /// `None` where the dialect did not say, which every reader has to /// be able to draw: a turn whose usage the CLI omitted leaves the /// context unmeasured rather than unchanged, and entries written /// before this existed have no answer at all. #[serde(default, skip_serializing_if = "Option::is_none")] context: Option, }, /// A compaction that finished, and how much context it recovered. /// /// The counts are the point, and a spinner is not: what a reader wants /// afterwards is that the session went from a million tokens to ten /// thousand, which is measured rather than estimated. They are /// optional because the record has shipped without them, and "the /// compaction happened, we don't know by how much" is a state this /// has to be able to say -- filling in a plausible number would make /// it indistinguishable from one that was counted. Compacted { #[serde(default, skip_serializing_if = "Option::is_none")] pre_tokens: Option, #[serde(default, skip_serializing_if = "Option::is_none")] post_tokens: Option, /// What asked for it, in the dialect's own word -- `auto` when the /// session compacted on its own. Carried rather than reduced to a /// bool so an unrecognised trigger stays unrecognised: an /// automatic compaction is the one worth naming, because it /// explains a wait nobody asked for, and defaulting the unknown /// case to "you asked for this" would explain it away. #[serde(default, skip_serializing_if = "Option::is_none")] trigger: Option, }, /// A command the session was asked to run on itself, held because it /// cannot run yet. /// /// These are not messages: `/compact` and `/rename` are instructions /// to the session about itself, and a session in the middle of a turn /// reads a line written to it as something the model should see. So /// they wait for the turn to end, and this is what a phone draws /// while they do -- otherwise pressing Compact during a long turn /// does nothing visible for minutes and looks like it was missed. CommandQueued { id: String, /// What to show for it: the command as a person would type it. text: String, }, /// The same command, now handed to the session. Its [`CommandQueued`] /// stops being pending when this arrives, matched by `id`; a command /// that ran immediately has only this. CommandSent { id: String, text: String, }, /// The conversation was cleared: everything above this is still in /// the record but is no longer in the session's context. /// /// Nothing is deleted. A transcript is the thing a person scrolls /// back through, and a session that dropped its history from the /// screen as well as from the model would lose the only copy the /// phone has -- so this is a divider, not a truncation, and the /// events before it stay exactly where they were. /// /// It is also what makes clearing mean the same thing for every /// driver, which is why the marker lives here rather than in one /// dialect: `llama` folds its conversation out of the transcript and /// simply folds from the last one of these, and `claude` starts a new /// CLI conversation behind it. /// /// **Load-bearing, not decorative.** For any driver that rebuilds its /// conversation from the transcript, this marker decides what the /// model is given -- dropping it, or treating it as something only /// the phone draws, silently puts a cleared conversation back in /// front of the model at full cost. Today `llama::conversation` is /// the only fold that reads it, which is the reason to write this /// down rather than leave it to be inferred from a second example /// that does not exist yet. Cleared, Error { message: String, }, } /// How much the model was holding, from the three figures a turn reports. /// /// The input side only -- prompt plus both cache figures. A cached token /// is cheaper but it is still one the model was given, so all three count; /// output is left out because it is what the turn produced rather than /// what continuing from here has to carry. /// /// One function so the definition cannot drift, because it is extracted in /// two quite different ways: the live translators have the usage object /// parsed, and `import::context_tokens` scans it out of a raw line without /// parsing, since those files reach tens of megabytes. pub fn context_tokens(input: u64, cache_creation: u64, cache_read: u64) -> u64 { input + cache_creation + cache_read } /// The context after `event`, given what it was before. /// /// The whole rule in one place, because three readers need the same /// answer: the pump keeping a live session's figure, the transcript /// seeding it at startup, and the phone folding the same events into what /// it draws. Written here beside the events it reads so a fourth reader /// finds it. /// /// The two that *lower* it are the point. A clear takes the conversation /// away and a compaction replaces it with a summary, so a figure measured /// before either stopped being true at that moment -- and carrying it /// forward is how a session that had just been cleared went on reporting /// the context it no longer had. /// /// `None` is "we don't know", which is a state each of them can reach: /// nothing has been measured yet, a compaction finished without saying /// how much it recovered, or a clear left a conversation nobody has /// counted since. pub fn context_after(current: Option, event: &Event) -> Option { match event { // `or`, so a turn the dialect reported no usage for leaves the last // measurement standing: it is stale by a turn, which every context // figure is, rather than wrong. Event::UsageDelta { context, .. } => context.or(current), Event::Compacted { post_tokens, .. } => *post_tokens, Event::Cleared => None, _ => current, } } /// Something a session can be asked to do to itself. /// /// A closed set rather than a string, because the two that are not /// dialect-specific have to reach every provider: compaction is a /// capability an llama session may one day have, and a name is this /// server's own. `Raw` is the escape for a dialect's own commands -- /// `/context`, `/usage` -- which only the thing running the session can /// interpret. #[derive(Debug, Clone, PartialEq)] pub enum SessionCommand { Compact, Clear, SetTitle(String), Raw(String), } impl SessionCommand { /// What a person would have typed to ask for this, which is what a /// phone shows while it waits. pub fn label(&self) -> String { match self { Self::Compact => "/compact".to_string(), Self::Clear => "/clear".to_string(), Self::SetTitle(title) => format!("/rename {title}"), Self::Raw(text) => text.clone(), } } /// Runs it. Called only at a boundary -- see [`Event::CommandQueued`]. pub fn apply(&self, driver: &dyn Driver) { match self { Self::Compact => driver.compact(), Self::Clear => driver.clear(), Self::SetTitle(title) => driver.set_title(title), Self::Raw(text) => driver.run_command(text), } } } #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub enum SessionStatus { Idle, Running, AwaitingInput, Compacting, Exited, /// There is a process recorded for this session and the machine will /// not say whether it is still running. /// /// Its own state rather than the nearest of the others, because both /// neighbours are lies with consequences: `Exited` invites starting a /// second process against a conversation that may already have one, /// and `Idle` claims a session is waiting for you when nobody has /// checked. It resolves itself -- the driver keeps asking -- so what /// it means to a reader is "wait", not "act". Unknown, } /// What became of a request to take a queued message back. /// /// Three states rather than a bool because the two failures are not the /// same fact. A driver that writes into its session the moment a message /// arrives -- which is what `ClaudeDriver` does, so that a steer reaches /// the model at the next tool boundary rather than at the end of the turn /// -- can never take one back, and a phone that was told only "no" would /// have to guess whether it had asked too late or asked about nothing. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum Unqueued { /// Out of the queue; the session will never read it. Dropped, /// Already handed to the session, so there is nothing left to take /// back. The message is on its way into the conversation. AlreadySent, /// Nothing is waiting under that id. Unknown, } /// Where a driver reports events. Unbounded because producers are child /// processes a slow phone must never be able to stall; the transcript file /// is the backpressure-free buffer of record. pub type EventSink = mpsc::UnboundedSender; /// The inbound half of a session. Deliberately small; see PLAN.md for the /// per-driver mapping of each method onto its dialect. /// /// `send_user_message` during a run is the point of the whole app: both /// real dialects queue it for injection at the next tool boundary rather /// than the end of the turn. pub trait Driver: Send + Sync { /// Takes a message, now or once the session is free for it. /// /// Every driver owes exactly one `MessageTaken` per message, at the /// moment it actually starts reading it: that event is what puts the /// message in the transcript, so a driver that never sends it drops /// the message from the conversation entirely. fn send_user_message(&self, text: String, images: Vec); /// Takes back a message that is still waiting, named by the id its /// [`Event::MessageQueued`] carried. /// /// Answering is the whole of the contract: a driver that drops the /// message owes an [`Event::MessageDropped`], and one that cannot must /// say which of the two reasons it is, because they are different /// things to a reader -- "the session has already been told" is worth /// knowing, and "there is nothing under that id" means the bubble on /// screen is stale. The default is the honest answer for a driver with /// no queue at all: nothing of yours is waiting. fn unqueue(&self, _id: &str) -> Unqueued { Unqueued::Unknown } /// Answers one question with everything that was chosen, in the order /// it was offered. One answer is a list of one; a driver whose dialect /// takes a single value joins them where it writes it. fn answer_question(&self, id: &str, answers: &[String]); /// Stop mid-run; the session survives. fn interrupt(&self); fn set_model(&self, model: &str); /// How much the session asks about before acting. Live rather than /// spawn-only: the answer changes with what is being done, and a phone /// is the worst place to answer "may I run this?" forty times. fn set_permission_mode(&self, mode: &str); // Both of the above are requests, and neither reports the outcome by // returning. A driver that actually changes the setting owes an // [`Event::Settings`] once it has -- that event, and not the request, // is what the manager and the phone read. One that cannot change it // owes an [`Event::Error`] saying why; saying nothing leaves a phone // showing a setting nobody applied. /// Tells the process what this conversation is called, when it has /// somewhere to put it. /// /// Unlike the two above, this is not a request that can fail: the /// rename has already happened in this server's own config, which is /// what a phone lists and the only place the name has to be. So a /// driver whose process has no notion of a name does nothing here and /// says nothing -- there is no failure to report, and an error beside /// a rename that plainly worked would be a puzzle rather than a /// warning. /// /// Claude Code has one: `--name` when a session is created and /// `/rename` afterwards, which is what puts the same name in its own /// session picker and in what other agents see. fn set_title(&self, title: &str); /// Runs a command this session's own dialect understands, verbatim. /// /// For the ones this app has no opinion about -- `/context`, `/usage`, /// anything a CLI adds next month. A driver whose process has no such /// vocabulary says so with an [`Event::Error`] rather than sending it /// as a message, which would put a line meant for the session in front /// of the model instead. /// /// Like [`Driver::compact`] and [`Driver::set_title`], this is called /// only when the session is between turns; the waiting is done above, /// once, for every driver. fn run_command(&self, text: &str); /// pi: native compaction; claude: `/compact`. fn compact(&self); /// Drops the conversation so far without ending the session. /// /// The cheap half of managing a long session, and the reason it is a /// driver operation rather than a manager one: compaction *reads* the /// whole conversation in order to summarise it, so on a large context /// it is itself one of the most expensive requests the session will /// make -- measured at 1.7 million tokens for a single automatic /// compaction on 2026-08-29. Clearing costs nothing, because nothing /// is sent. /// /// Every implementation emits [`Event::Cleared`] so the transcript /// carries the divider whatever the dialect did behind it. fn clear(&self); /// Stop attending to the process but leave it running, because this /// server is going away and means to adopt it again when it comes /// back. /// /// This is deliberately not a shutdown. A backend restart -- a /// rebuild, a service restart, a crash -- must not end a turn that is /// in flight, so a session's process outlives the server that started /// it and is found again through `session::process`. A driver with no /// process of its own has nothing to do here. /// /// Its counterpart is [`Driver::stop`]. Every driver owes exactly one /// of the two on the way out, and which one is the difference between /// "back shortly" and "this conversation is over". /// Whether a line written *now* would start a turn of its own, rather /// than landing inside one already in flight. /// /// Asked of the driver because the driver is the only thing that knows: /// it sees every line it wrote and every line that came back, and it /// updates this the instant it writes rather than when output returns. /// The manager's `SessionStatus` cannot answer it -- that is built from /// what has been *recorded*, so between writing a line and the CLI's /// first output it still reads idle, and a second line sent in that gap /// lands inside the turn the first one started. For a command that is /// the difference between being executed and being read to the model as /// text, which is silent both ways. /// /// Defaults to true for a driver with no turn of its own to be inside. fn between_turns(&self) -> bool { true } fn detach(&self); /// End the process for good, because it must not survive this. The /// path out for everything [`detach`] preserves. /// /// Two callers, and the difference between them is only what is being /// ended: a session being deleted, whose conversation goes with it, and /// a throwaway session at a server's exit, whose transcript stays and /// whose process does not (see [`SessionConfig::throwaway`]). /// /// [`detach`]: Driver::detach /// [`SessionConfig::throwaway`]: crate::config::SessionConfig::throwaway fn stop(&self); } #[cfg(test)] mod tests { use super::*; /// A tripwire for the wire format, not for serde. /// /// The app reads these names, and getting one wrong does not fail /// loudly: a field the app cannot find reads as a field the server /// chose not to send, which several of them are allowed to be. #[test] fn multi_word_fields_go_out_in_camel_case() { let json = serde_json::to_value(Event::Compacted { pre_tokens: Some(28719), post_tokens: Some(1125), trigger: Some("manual".to_string()), }) .expect("serialize"); assert_eq!( json, serde_json::json!({ "type": "compacted", "preTokens": 28719, "postTokens": 1125, "trigger": "manual", }) ); } /// The two events that take the context *down* are the point of the /// fold: a figure measured before a compaction or a clear stopped being /// true at that moment, and carrying it forward is how a session that /// had just been cleared went on reporting the context it no longer /// had. #[test] fn a_compaction_and_a_clear_move_the_context_a_turn_cannot() { let after = |current, event| context_after(current, &event); assert_eq!( after( Some(500), Event::UsageDelta { tokens: 12, context: Some(30_100), } ), Some(30_100) ); assert_eq!( after( Some(128_402), Event::Compacted { pre_tokens: Some(128_402), post_tokens: Some(9_617), trigger: Some("auto".to_string()), } ), Some(9_617) ); assert_eq!(after(Some(9_617), Event::Cleared), None); // A compaction that did not say how much it recovered leaves the // context unknown rather than stale: it definitely moved, and the // one thing that is certainly wrong is the figure from before it. assert_eq!( after( Some(128_402), Event::Compacted { pre_tokens: None, post_tokens: None, trigger: None, } ), None ); // A turn the dialect reported no context for is stale by a turn, // which every context figure is, rather than unknown. assert_eq!( after( Some(30_100), Event::UsageDelta { tokens: 12, context: None, } ), Some(30_100) ); // Everything else leaves it alone. assert_eq!( after( Some(30_100), Event::Status { state: SessionStatus::Idle, } ), Some(30_100) ); } }