From c8a274a42c04930b32b12616d6493498f4646b88 Mon Sep 17 00:00:00 2001 From: prem315 Date: Mon, 19 Oct 2026 19:56:16 +0530 Subject: [PATCH] Add TUI reconnect and durable session log recovery - Resume session streams from durable log cursors - Reconcile queued prompts and merge live drafts after reconnects - Ignore stale history snapshots across sessions and requests --- packages/tui/src/app.rs | 560 ++++++++++- packages/tui/src/event.rs | 10 +- packages/tui/src/rpc/client.rs | 1548 +++++++++++++++++++++++++++++- packages/tui/src/rpc/protocol.rs | 77 +- 4 files changed, 2117 insertions(+), 78 deletions(-) diff --git a/packages/tui/src/app.rs b/packages/tui/src/app.rs index 0a164f92..5575f899 100644 --- a/packages/tui/src/app.rs +++ b/packages/tui/src/app.rs @@ -134,9 +134,7 @@ impl Turn { pub fn start_llm(&mut self, message_id: &str) { self.retry = None; - let draft = self.draft_mut(message_id); - draft.thinking_parts.clear(); - draft.text_parts.clear(); + let _ = self.draft_mut(message_id); self.sync_thinking(); self.sync_text(); } @@ -349,6 +347,8 @@ pub struct App { pub busy: bool, /// Whether the stream socket dropped and is reconnecting. pub reconnecting: bool, + /// Durable `session.log` cursor (`max(last durable.seq, last log.synced.seq)`). + pub log_cursor: Option, /// Scroll offset in lines from the bottom of the session transcript. pub scroll: usize, /// Maximum scroll offset computed by the last transcript render pass. @@ -357,6 +357,8 @@ pub struct App { next_prompt_seq: u64, prompt_stream: Option, initial_prompt_id: Option, + history_request: Cell, + history_revision: u64, keymap: Keymap, tx: mpsc::UnboundedSender, } @@ -392,12 +394,15 @@ impl App { queued_prompts: Vec::new(), busy: false, reconnecting: false, + log_cursor: None, scroll: 0, max_scroll: Cell::new(0), should_quit: false, next_prompt_seq: 1, prompt_stream: None, initial_prompt_id: None, + history_request: Cell::new(0), + history_revision: 0, keymap: Keymap::new(), tx, }; @@ -436,6 +441,10 @@ impl App { fn apply_session(&mut self, info: SessionInfo) { let same_session = self.session.as_ref().map(|s| &s.id) == Some(&info.id); let host_changed = self.session.as_ref().map(|s| &s.host_dir) != Some(&info.host_dir); + if !same_session { + self.history_request.set(self.history_request.get() + 1); + self.log_cursor = None; + } self.model = info .model .as_ref() @@ -753,17 +762,56 @@ impl App { self.notice = Some(err); true } - RpcEvent::SessionHistoryLoaded(Ok(entries)) => { - self.turns = turns_from_history(&entries); + RpcEvent::SessionHistoryRefresh { session_id } => { + if self.session.as_ref().is_some_and(|s| s.id == session_id) { + self.spawn_session_history(&session_id); + } + false + } + RpcEvent::SessionHistoryLoaded { + session_id, + request_id, + revision, + result, + } => { + if !self.session.as_ref().is_some_and(|s| s.id == session_id) + || request_id != self.history_request.get() + { + return false; + } + if revision != self.history_revision { + self.spawn_session_history(&session_id); + return false; + } + let entries = match result { + Ok(entries) => entries, + Err(err) => { + self.notice = Some(err); + return true; + } + }; + let was_empty = self.turns.is_empty(); + crate::rpc::client::advance_log_cursor( + &mut self.log_cursor, + crate::rpc::client::history_cursor(&entries), + ); + self.sync_turns_from_history(&entries); self.busy = self .turns .last() .is_some_and(|turn| turn.state == TurnState::Running); - self.scroll = 0; - true - } - RpcEvent::SessionHistoryLoaded(Err(err)) => { - self.notice = Some(err); + if was_empty { + self.scroll = 0; + } + let stream_running = self + .prompt_stream + .as_ref() + .is_some_and(|task| !task.is_finished()); + if !stream_running { + if let Some(session_id) = self.session.as_ref().map(|s| s.id.clone()) { + self.spawn_follow_session(session_id, self.log_cursor, self.busy); + } + } true } RpcEvent::SessionConfigured(Ok(info)) => { @@ -851,6 +899,8 @@ impl App { } fn handle_stream(&mut self, event: StreamEvent) -> bool { + // A snapshot requested before this event cannot replace the newer UI state. + self.history_revision += 1; match event { StreamEvent::SessionCreated(info) => { self.apply_session(info); @@ -1004,7 +1054,9 @@ impl App { }; self.notice = Some(msg.clone()); if let Some(turn) = self.turns.last_mut() { - turn.notices.push(msg); + if !turn.notices.contains(&msg) { + turn.notices.push(msg); + } } } StreamEvent::TurnStarted => { @@ -1213,6 +1265,15 @@ impl App { self.usage.model = usage.model; } } + StreamEvent::LogSynced { seq } => { + let prev = self.log_cursor; + crate::rpc::client::advance_log_cursor(&mut self.log_cursor, seq); + if self.log_cursor != prev { + if let Some(session_id) = self.session.as_ref().map(|s| s.id.clone()) { + self.spawn_session_history(&session_id); + } + } + } StreamEvent::Disconnected => { self.reconnecting = true; } @@ -1220,14 +1281,17 @@ impl App { self.reconnecting = false; if self.initial_prompt_id.is_none() { if let Some(session_id) = self.session.as_ref().map(|s| s.id.clone()) { - self.dispatch_pending_queued_prompts(&session_id); + self.reconcile_queued_prompts(&session_id); + self.spawn_session_history(&session_id); } } } StreamEvent::Succeeded => { if let Some(turn) = self.turns.last_mut() { turn.retry = None; - turn.state = TurnState::Succeeded; + if turn.state == TurnState::Running { + turn.state = TurnState::Succeeded; + } } self.busy = false; self.reconnecting = false; @@ -1260,6 +1324,7 @@ impl App { return false; }; + self.history_revision += 1; let message_id = self.next_message_id(); self.prompt.clear(); self.screen = Screen::Session; @@ -1310,6 +1375,7 @@ impl App { let tx = self.tx.clone(); let existing_session_id = self.session.as_ref().map(|s| s.id.clone()); + let after = self.log_cursor; let model = self.model.clone(); let thinking = self.thinking.clone(); @@ -1335,11 +1401,11 @@ impl App { }; let (stream_tx, mut stream_rx) = mpsc::unbounded_channel(); - let stream = client.watch_session_with( + let stream = client.watch_session_from( session_id.clone(), - text, - delivery, - message_id, + after, + Some((text, delivery, message_id)), + true, stream_tx, ); tokio::pin!(stream); @@ -1370,6 +1436,44 @@ impl App { true } + fn spawn_follow_session( + &mut self, + session_id: String, + after: Option, + initially_busy: bool, + ) { + let tx = self.tx.clone(); + if let Ok(handle) = tokio::runtime::Handle::try_current() { + let task = handle.spawn(async move { + let client = RpcClient::default(); + let (stream_tx, mut stream_rx) = mpsc::unbounded_channel(); + let stream = + client.follow_session(session_id.clone(), after, initially_busy, stream_tx); + tokio::pin!(stream); + loop { + tokio::select! { + biased; + Some(event) = stream_rx.recv() => { + if tx.send(AppEvent::Rpc(RpcEvent::SessionStream { + session_id: session_id.clone(), event, + })).is_err() { + break; + } + } + _ = &mut stream => break, + } + } + while let Ok(event) = stream_rx.try_recv() { + let _ = tx.send(AppEvent::Rpc(RpcEvent::SessionStream { + session_id: session_id.clone(), + event, + })); + } + }); + self.prompt_stream = Some(task.abort_handle()); + } + } + fn dispatch_pending_queued_prompts(&mut self, session_id: &str) { let mut to_send = Vec::new(); for queued in &mut self.queued_prompts { @@ -1387,6 +1491,66 @@ impl App { } } + fn reconcile_queued_prompts(&mut self, session_id: &str) { + let mut to_send = Vec::new(); + let mut to_reconcile = Vec::new(); + for queued in &mut self.queued_prompts { + if !queued.dispatched && !queued.admitted { + queued.dispatched = true; + to_send.push(( + queued.prompt.clone(), + queued.delivery, + queued.message_id.clone(), + )); + } else if queued.dispatched { + to_reconcile.push(( + queued.prompt.clone(), + queued.delivery, + queued.message_id.clone(), + )); + } + } + for (text, delivery, message_id) in to_send { + self.spawn_send_prompt(session_id.to_string(), text, delivery, message_id); + } + for (text, delivery, message_id) in to_reconcile { + self.spawn_reconcile_prompt(session_id.to_string(), text, delivery, message_id); + } + } + + fn spawn_reconcile_prompt( + &self, + session_id: String, + text: String, + delivery: PromptDelivery, + message_id: String, + ) { + let tx = self.tx.clone(); + if let Ok(handle) = tokio::runtime::Handle::try_current() { + handle.spawn(async move { + let client = RpcClient::default(); + match client.session_message(&session_id, &message_id).await { + Ok(true) => { + let _ = tx.send(AppEvent::Rpc(RpcEvent::SessionHistoryRefresh { + session_id, + })); + } + Ok(false) => { + let result = client + .send_prompt(&session_id, &text, delivery, &message_id) + .await; + let _ = tx.send(AppEvent::Rpc(RpcEvent::PromptDispatched { + session_id, + message_id, + result, + })); + } + Err(_) => {} + } + }); + } + } + fn stop_prompt_stream(&mut self) { if let Some(task) = self.prompt_stream.take() { task.abort(); @@ -1525,6 +1689,7 @@ impl App { self.usage = UsageReport::default(); self.busy = false; self.reconnecting = false; + self.log_cursor = None; self.scroll = 0; self.notice = Some("started a new session".to_string()); self.spawn_list_models(); @@ -1567,6 +1732,7 @@ impl App { self.stop_prompt_stream(); let session_id = info.id.clone(); let title = info.title.clone(); + self.log_cursor = None; self.apply_session(info); self.turns.clear(); self.queued_prompts.clear(); @@ -1576,6 +1742,7 @@ impl App { self.scroll = 0; self.screen = Screen::Session; self.notice = Some(format!("switched to session {title}")); + self.spawn_session_refresh(); self.spawn_session_history(&session_id); } } @@ -1644,11 +1811,19 @@ impl App { fn spawn_session_history(&self, session_id: &str) { let session_id = session_id.to_string(); + let request_id = self.history_request.get() + 1; + self.history_request.set(request_id); + let revision = self.history_revision; let tx = self.tx.clone(); if let Ok(handle) = tokio::runtime::Handle::try_current() { handle.spawn(async move { - let res = RpcClient::default().session_history(&session_id).await; - let _ = tx.send(AppEvent::Rpc(RpcEvent::SessionHistoryLoaded(res))); + let result = RpcClient::default().session_history(&session_id).await; + let _ = tx.send(AppEvent::Rpc(RpcEvent::SessionHistoryLoaded { + session_id, + request_id, + revision, + result, + })); }); } } @@ -1692,6 +1867,111 @@ impl App { }); } } + + fn sync_turns_from_history(&mut self, entries: &[HistoryEntry]) { + self.queued_prompts + .retain(|q| !entries.iter().any(|e| e.id == q.message_id)); + + let mut next_turns = turns_from_history(entries); + let last_entry_is_user = entries.last().is_some_and(|e| e.role == "user"); + let total_next = next_turns.len(); + + for (idx, turn) in next_turns.iter_mut().enumerate() { + let Some(prompt_id) = turn.prompt_id.as_deref() else { + continue; + }; + let Some(existing) = self + .turns + .iter() + .find(|t| t.prompt_id.as_deref() == Some(prompt_id)) + else { + continue; + }; + + for notice in &existing.notices { + if !turn.notices.contains(notice) { + turn.notices.push(notice.clone()); + } + } + + if idx + 1 == total_next + && last_entry_is_user + && existing.state == TurnState::Running + && self.busy + { + turn.state = TurnState::Running; + } + + if turn.state == TurnState::Running { + if turn.retry.is_none() { + turn.retry = existing.retry.clone(); + } + for draft in &existing.drafts { + let target = turn.draft_mut(&draft.message_id); + for (&part_idx, thinking) in &draft.thinking_parts { + let entry = target.thinking_parts.entry(part_idx).or_default(); + if entry.len() < thinking.len() { + *entry = thinking.clone(); + } + } + for (&part_idx, text) in &draft.text_parts { + let entry = target.text_parts.entry(part_idx).or_default(); + if entry.len() < text.len() { + *entry = text.clone(); + } + } + } + turn.sync_thinking(); + turn.sync_text(); + } + + if matches!( + existing.state, + TurnState::Failed(_) | TurnState::Interrupted(_) + ) || (existing.state == TurnState::Succeeded && turn.state == TurnState::Running) + { + turn.state = existing.state.clone(); + turn.retry = None; + } + + for existing_tool in &existing.tools { + if let Some(tool) = turn + .tools + .iter_mut() + .find(|t| t.call_id == existing_tool.call_id) + { + if tool.patch.is_none() && existing_tool.patch.is_some() { + tool.patch = existing_tool.patch.clone(); + } + if tool.output.is_none() && existing_tool.output.is_some() { + tool.output = existing_tool.output.clone(); + } + if tool.summary.is_none() && existing_tool.summary.is_some() { + tool.summary = existing_tool.summary.clone(); + } + } else if turn.state == TurnState::Running { + turn.tools.push(existing_tool.clone()); + } + } + } + + if self.busy { + if let Some(last_local) = self.turns.last() { + if last_local.state == TurnState::Running { + let in_history = last_local.prompt_id.as_deref().is_some_and(|id| { + next_turns + .iter() + .any(|t| t.prompt_id.as_deref() == Some(id)) + }); + if !in_history { + next_turns.push(last_local.clone()); + } + } + } + } + + self.turns = next_turns; + } } fn theme_named(name: &str) -> Option<&'static Palette> { @@ -2054,6 +2334,71 @@ mod tests { }; use serde_json::json; + #[test] + fn history_cannot_resurrect_completed_turns_or_cross_sessions() { + let mut app = App::default(); + app.session = Some( + serde_json::from_value(json!({ + "id": "ses", "title": "Session", "directory": "/workspace", "updatedAt": 0 + })) + .unwrap(), + ); + app.turns.push(Turn::with_id("prompt", "Question")); + app.busy = true; + let entries = vec![ + HistoryEntry { + id: "prompt".into(), + seq: 1, + role: "user".into(), + state: "committed".into(), + created_at: 0, + parts: vec![], + }, + HistoryEntry { + id: "answer".into(), + seq: 2, + role: "assistant".into(), + state: "draft".into(), + created_at: 0, + parts: vec![], + }, + ]; + let stale = history_event(&app, entries.clone()); + app.handle_stream(StreamEvent::TextEnd { + message_id: "answer".into(), + part_index: 0, + content: "Finished".into(), + }); + app.handle_stream(StreamEvent::Succeeded); + app.update(stale); + assert!(!app.busy); + assert_eq!(app.turns[0].state, TurnState::Succeeded); + assert_eq!(app.turns[0].text, "Finished"); + + let wrong_session = history_event(&app, entries.clone()); + app.session.as_mut().unwrap().id = "other".into(); + app.turns.clear(); + app.log_cursor = None; + app.update(wrong_session); + assert!(app.turns.is_empty()); + assert_eq!(app.log_cursor, None); + + // A newer request wins even when there were no intervening stream events. + let outdated = history_event(&app, entries); + app.history_request.set(app.history_request.get() + 1); + app.update(outdated); + assert!(app.turns.is_empty()); + } + + fn history_event(app: &App, entries: Vec) -> AppEvent { + AppEvent::Rpc(RpcEvent::SessionHistoryLoaded { + session_id: app.session.as_ref().unwrap().id.clone(), + request_id: app.history_request.get(), + revision: app.history_revision, + result: Ok(entries), + }) + } + #[test] fn waits_for_initial_admission_before_dispatching_pending_prompts() { let mut app = App::default(); @@ -2398,28 +2743,31 @@ mod tests { assert_eq!(app.screen, Screen::Session); // Delivering `SessionHistoryLoaded` populates `app.turns`. - app.update(AppEvent::Rpc(RpcEvent::SessionHistoryLoaded(Ok(vec![ - HistoryEntry { - id: "m1".to_string(), - seq: 1, - role: "user".to_string(), - state: "committed".to_string(), - created_at: 2_000, - parts: vec![Part::Text { - text: "Resume me".to_string(), - }], - }, - HistoryEntry { - id: "m2".to_string(), - seq: 2, - role: "assistant".to_string(), - state: "committed".to_string(), - created_at: 2_100, - parts: vec![Part::Text { - text: "Resumed!".to_string(), - }], - }, - ])))); + app.update(history_event( + &app, + vec![ + HistoryEntry { + id: "m1".to_string(), + seq: 1, + role: "user".to_string(), + state: "committed".to_string(), + created_at: 2_000, + parts: vec![Part::Text { + text: "Resume me".to_string(), + }], + }, + HistoryEntry { + id: "m2".to_string(), + seq: 2, + role: "assistant".to_string(), + state: "committed".to_string(), + created_at: 2_100, + parts: vec![Part::Text { + text: "Resumed!".to_string(), + }], + }, + ], + )); assert_eq!(app.turns.len(), 1); assert_eq!(app.turns[0].prompt, "Resume me"); assert_eq!(app.turns[0].text, "Resumed!"); @@ -2894,4 +3242,134 @@ mod tests { .notices .contains(&"forked session ses_2 from entry_4".to_string())); } + + #[test] + fn tracks_log_cursor_and_merges_history_snapshots_with_live_drafts() { + let mut app = App::default(); + app.session = Some(SessionInfo { + id: "ses_sync".to_string(), + title: "Sync".to_string(), + directory: "/workspace".to_string(), + host_dir: None, + sandbox: None, + project: None, + updated_at: 1_000, + model: None, + thinking_level: None, + }); + app.busy = true; + + // Turn 1 is actively streaming live deltas in memory; a queued prompt is waiting. + let mut live_turn = Turn::with_id("u1", "Explain the bug"); + live_turn.append_thinking_delta(Some("a1"), Some(0), "Inspecting client.rs..."); + live_turn.append_text_delta(Some("a1"), Some(1), "The socket drops before "); + app.turns.push(live_turn); + app.queued_prompts.push(QueuedPrompt { + message_id: "u2".to_string(), + prompt: "Also add a test".to_string(), + delivery: PromptDelivery::FollowUp, + admitted: true, + dispatched: true, + }); + + // 1. Mid-turn history snapshot arrives while `a1` is still `state: "draft"` (empty in DB): + // live draft deltas are preserved and `log_cursor` advances to 2. + app.update(history_event( + &app, + vec![ + HistoryEntry { + id: "u1".to_string(), + seq: 1, + role: "user".to_string(), + state: "committed".to_string(), + created_at: 1_000, + parts: vec![Part::Text { + text: "Explain the bug".to_string(), + }], + }, + HistoryEntry { + id: "a1".to_string(), + seq: 2, + role: "assistant".to_string(), + state: "draft".to_string(), + created_at: 1_050, + parts: vec![], + }, + ], + )); + assert_eq!(app.log_cursor, Some(2)); + assert!(app.busy); + assert_eq!(app.turns.len(), 1); + assert_eq!(app.turns[0].thinking, "Inspecting client.rs..."); + assert_eq!(app.turns[0].text, "The socket drops before "); + assert_eq!(app.queued_prompts.len(), 1); + + // 2. `LogSynced` with `seq: 5` advances `log_cursor` to `max(2, 5) = 5`; a smaller seq + // never moves it backwards. + app.update(AppEvent::Rpc(RpcEvent::Stream(StreamEvent::LogSynced { + seq: Some(5), + }))); + assert_eq!(app.log_cursor, Some(5)); + app.update(AppEvent::Rpc(RpcEvent::Stream(StreamEvent::LogSynced { + seq: Some(3), + }))); + assert_eq!(app.log_cursor, Some(5)); + + // 3. Reconnect history snapshot arrives after both turns settled during a disconnect: + // `u2` is removed from `queued_prompts`, both turns are `Succeeded`, and `busy` is false. + app.update(history_event( + &app, + vec![ + HistoryEntry { + id: "u1".to_string(), + seq: 1, + role: "user".to_string(), + state: "committed".to_string(), + created_at: 1_000, + parts: vec![Part::Text { + text: "Explain the bug".to_string(), + }], + }, + HistoryEntry { + id: "a1".to_string(), + seq: 2, + role: "assistant".to_string(), + state: "committed".to_string(), + created_at: 1_100, + parts: vec![Part::Text { + text: "The socket drops before reconciliation.".to_string(), + }], + }, + HistoryEntry { + id: "u2".to_string(), + seq: 6, + role: "user".to_string(), + state: "committed".to_string(), + created_at: 1_200, + parts: vec![Part::Text { + text: "Also add a test".to_string(), + }], + }, + HistoryEntry { + id: "a2".to_string(), + seq: 7, + role: "assistant".to_string(), + state: "committed".to_string(), + created_at: 1_300, + parts: vec![Part::Text { + text: "Added reconnect tests.".to_string(), + }], + }, + ], + )); + assert_eq!(app.log_cursor, Some(7)); + assert!(!app.busy); + assert!(app.queued_prompts.is_empty()); + assert_eq!(app.turns.len(), 2); + assert_eq!(app.turns[0].state, TurnState::Succeeded); + assert_eq!(app.turns[0].text, "The socket drops before reconciliation."); + assert_eq!(app.turns[1].prompt, "Also add a test"); + assert_eq!(app.turns[1].state, TurnState::Succeeded); + assert_eq!(app.turns[1].text, "Added reconnect tests."); + } } diff --git a/packages/tui/src/event.rs b/packages/tui/src/event.rs index fc0f7e65..e60f00ea 100644 --- a/packages/tui/src/event.rs +++ b/packages/tui/src/event.rs @@ -23,7 +23,15 @@ pub enum RpcEvent { ServerInfoLoaded(Result), ModelsLoaded(Result), SessionsLoaded(Result, String>), - SessionHistoryLoaded(Result, String>), + SessionHistoryRefresh { + session_id: String, + }, + SessionHistoryLoaded { + session_id: String, + request_id: u64, + revision: u64, + result: Result, String>, + }, SessionConfigured(Result), PluginsReloaded(Result<(u64, Option), String>), PromptDispatched { diff --git a/packages/tui/src/rpc/client.rs b/packages/tui/src/rpc/client.rs index c4eb860d..024d624e 100644 --- a/packages/tui/src/rpc/client.rs +++ b/packages/tui/src/rpc/client.rs @@ -13,7 +13,8 @@ use tokio_tungstenite::tungstenite::Message; use crate::rpc::protocol::{ ConfigureSessionPayload, CreateSessionPayload, ExitPayload, ListModelsPayload, ModelRefPayload, PromptSessionPayload, RpcAck, RpcRequest, RuntimePayload, ServerMessage, - SessionHistoryResponse, SessionIdPayload, + SessionHistoryResponse, SessionIdPayload, SessionLogPayload, SessionMessagePayload, + SessionMessageResponse, }; use crate::types::{ HistoryEntry, ModelList, ModelRef, PromptDelivery, ServerInfo, ServiceRegistration, @@ -244,6 +245,10 @@ pub enum StreamEvent { /// A plugin or runtime warning surfaced during the turn (as ACP's `Feed.notice` does). Notice(String), Usage(UsageReport), + /// `session.log` replay reached the head sequence read when the call began. + LogSynced { + seq: Option, + }, ExecutionStarted, Succeeded, Failed(String), @@ -256,6 +261,19 @@ pub enum StreamEvent { Reconnected, } +/// Advance a `session.log` cursor (`after`) to `max(cursor, next_seq)`. +pub fn advance_log_cursor(cursor: &mut Option, next_seq: Option) { + if let Some(seq) = next_seq { + *cursor = Some(cursor.map_or(seq, |prev| prev.max(seq))); + } +} + +/// Highest durable sequence in a `session.history` snapshot, used as the initial `after` +/// cursor when opening or resuming an existing session. +pub fn history_cursor(entries: &[HistoryEntry]) -> Option { + entries.iter().map(|entry| entry.seq).max() +} + pub struct RpcClient { url: String, expected: Option, @@ -543,6 +561,229 @@ impl RpcClient { .map(|_| ()) } + /// Check whether a prompt materialized into the conversation (`session.message`). + pub async fn session_message( + &self, + session_id: &str, + message_id: &str, + ) -> Result { + let value = self + .call( + "session.message", + SessionMessagePayload { + session_id: session_id.to_string(), + message_id: message_id.to_string(), + }, + ) + .await?; + parse_session_message(&value) + } + + /// Wait until a session has no work running or queued (`session.wait`). + #[allow(dead_code)] + pub async fn wait_session(&self, session_id: &str) -> Result<(), String> { + self.call( + "session.wait", + SessionIdPayload { + session_id: session_id.to_string(), + }, + ) + .await + .map(|_| ()) + } + + /// Stream durable events for `session_id` via `session.log` (`{ sessionId, after, follow }`), + /// advancing `cursor` to `max(last durable.seq, last log.synced.seq)` and resuming from + /// `cursor` on reconnect when `follow` is `true`. + #[allow(dead_code)] + pub async fn session_log( + &self, + session_id: &str, + after: Option, + follow: bool, + cursor: &mut Option, + event_tx: &mpsc::UnboundedSender, + ) -> Result<(), String> { + advance_log_cursor(cursor, after); + let mut attempt = 0u32; + let mut seen_durable_seqs = std::collections::HashSet::new(); + + loop { + match self + .stream_session_log_once( + session_id, + follow, + cursor, + &mut seen_durable_seqs, + attempt > 0, + event_tx, + ) + .await + { + Ok(true) => return Ok(()), + Ok(false) => { + attempt += 1; + if attempt > RECONNECT_ATTEMPTS { + let _ = event_tx.send(StreamEvent::Failed(format!( + "Lost the connection to {} and could not reconnect", + self.url + ))); + return Ok(()); + } + let _ = event_tx.send(StreamEvent::Disconnected); + sleep(RECONNECT_DELAY).await; + } + Err(message) => { + let _ = event_tx.send(StreamEvent::Failed(message)); + return Ok(()); + } + } + } + } + + /// One connection's worth of `session.log`. `Ok(true)` means a non-followed replay completed; + /// `Ok(false)` means the socket dropped and a follower should reconnect from `cursor`. + pub async fn stream_session_log_once( + &self, + session_id: &str, + follow: bool, + cursor: &mut Option, + seen_durable_seqs: &mut std::collections::HashSet, + is_retry: bool, + event_tx: &mpsc::UnboundedSender, + ) -> Result { + let (ws_stream, _) = match connect_async(&self.url).await { + Ok(stream) => stream, + Err(_) => return Ok(false), + }; + + let (mut write, mut read) = ws_stream.split(); + let log_id = self.next_request_id(); + let log_req = RpcRequest::new( + log_id, + "session.log", + SessionLogPayload { + session_id: session_id.to_string(), + after: *cursor, + follow: Some(follow), + }, + ); + let mut log_text = serde_json::to_string(&log_req).map_err(|e| e.to_string())?; + log_text.push('\n'); + write + .send(Message::Text(log_text)) + .await + .map_err(|e| format!("Failed to subscribe to session.log: {}", e))?; + + let mut reconnected = is_retry; + let mut saw_synced = false; + + while let Some(msg_res) = read.next().await { + let msg = match msg_res { + Ok(m) => m, + Err(_) => return Ok(!follow && saw_synced), + }; + + if let Message::Text(txt) = msg { + for line in txt.lines() { + let line = line.trim(); + if line.is_empty() { + continue; + } + + let parsed: ServerMessage = match serde_json::from_str(line) { + Ok(m) => m, + Err(_) => continue, + }; + + match parsed { + ServerMessage::Chunk { request_id, values } => { + if let Some(id_num) = as_request_id(&request_id) { + let ack = RpcAck::new(id_num); + if let Ok(mut ack_json) = serde_json::to_string(&ack) { + ack_json.push('\n'); + let _ = write.send(Message::Text(ack_json)).await; + } + } + + for env in values { + if env.event_type == "log.synced" { + if env + .aggregate_id + .as_deref() + .is_none_or(|agg| agg == session_id) + { + if reconnected { + reconnected = false; + let _ = event_tx.send(StreamEvent::Reconnected); + } + advance_log_cursor(cursor, env.seq); + saw_synced = true; + let _ = + event_tx.send(StreamEvent::LogSynced { seq: env.seq }); + } + continue; + } + + let data = match &env.data { + Some(d) => d, + None => continue, + }; + + let s_id = data.get("sessionId").and_then(|v| v.as_str()); + let source_id = + data.get("sourceSessionId").and_then(|v| v.as_str()); + let durable_agg = + env.durable.as_ref().map(|d| d.aggregate_id.as_str()); + let matches_session = s_id == Some(session_id) + || durable_agg == Some(session_id) + || (env.event_type == "session.next.forked" + && source_id == Some(session_id)); + if !matches_session { + continue; + } + + if reconnected { + reconnected = false; + let _ = event_tx.send(StreamEvent::Reconnected); + } + + if let Some(seq) = env.durable_seq() { + if durable_agg.is_none_or(|agg| agg == session_id) { + advance_log_cursor(cursor, Some(seq)); + if !seen_durable_seqs.insert(seq) { + continue; + } + } + } + + for stream_event in + decode_durable_session_events(env.event_type.as_str(), data) + { + let _ = event_tx.send(stream_event); + } + } + } + ServerMessage::Exit { request_id, exit } + if as_request_id(&request_id) == Some(log_id) => + { + return match exit { + ExitPayload::Success { .. } => Ok(true), + ExitPayload::Failure { cause } => Err(format_cause(cause.as_ref())), + }; + } + ServerMessage::Defect { defect } => { + return Err(format_cause(defect.as_ref())); + } + _ => {} + } + } + } + } + + Ok(!follow && saw_synced) + } + /// Stream one prompt, reconnecting the socket when the transport drops. /// /// The session lives on the server, not this socket, so a dropped connection @@ -584,9 +825,11 @@ impl RpcClient { ) -> Result<(), String> { self.stream_prompt( session_id, + None, prompt_text, delivery, prompt_message_id, + true, event_tx, false, ) @@ -594,6 +837,7 @@ impl RpcClient { } /// Keep observing the session across busy periods, until the caller cancels the task. + #[allow(dead_code)] pub async fn watch_session_with( &self, session_id: String, @@ -604,36 +848,87 @@ impl RpcClient { ) -> Result<(), String> { self.stream_prompt( session_id, + None, + prompt_text, + delivery, + prompt_message_id, + true, + event_tx, + true, + ) + .await + } + + /// Keep observing a session starting from `after` in `session.log` and optionally dispatching + /// an initial prompt. + pub async fn watch_session_from( + &self, + session_id: String, + after: Option, + initial_prompt: Option<(String, PromptDelivery, String)>, + initially_busy: bool, + event_tx: mpsc::UnboundedSender, + ) -> Result<(), String> { + let (prompt_text, delivery, prompt_message_id) = initial_prompt + .unwrap_or_else(|| (String::new(), PromptDelivery::FollowUp, String::new())); + let turn_active = !prompt_message_id.is_empty() || initially_busy; + self.stream_prompt( + session_id, + after, prompt_text, delivery, prompt_message_id, + turn_active, event_tx, true, ) .await } + /// Follow an existing session without dispatching a prompt, combining `session.log` (`after`, + /// `follow: true`) and `event.subscribe`. + pub async fn follow_session( + &self, + session_id: String, + after: Option, + initially_busy: bool, + event_tx: mpsc::UnboundedSender, + ) -> Result<(), String> { + self.watch_session_from(session_id, after, None, initially_busy, event_tx) + .await + } + #[allow(clippy::too_many_arguments)] async fn stream_prompt( &self, session_id: String, + after: Option, prompt_text: String, delivery: PromptDelivery, prompt_message_id: String, + initially_busy: bool, event_tx: mpsc::UnboundedSender, keep_alive: bool, ) -> Result<(), String> { - let mut prompt_dispatched = false; + let mut prompt_dispatched = prompt_message_id.is_empty(); + let mut log_cursor = after; + let mut turn_active = initially_busy || !prompt_message_id.is_empty(); + let mut initial_prompt_settled = prompt_message_id.is_empty(); + let mut seen_durable_seqs = std::collections::HashSet::new(); let mut attempt = 0u32; loop { match self - .stream_session_once( + .stream_session_once_with_cursor( &session_id, &prompt_text, delivery, &prompt_message_id, &mut prompt_dispatched, + &mut log_cursor, + &mut turn_active, + &mut initial_prompt_settled, + &mut seen_durable_seqs, attempt > 0, keep_alive, &event_tx, @@ -671,7 +966,7 @@ impl RpcClient { /// One connection's worth of a prompt. `Ok(true)` means the turn ended; /// `Ok(false)` means the socket dropped and the caller should retry. - #[allow(clippy::too_many_arguments)] + #[allow(clippy::too_many_arguments, dead_code)] async fn stream_session_once( &self, session_id: &str, @@ -682,6 +977,46 @@ impl RpcClient { is_retry: bool, keep_alive: bool, event_tx: &mpsc::UnboundedSender, + ) -> Result { + let mut log_cursor = None; + let mut turn_active = !prompt_message_id.is_empty(); + let mut initial_prompt_settled = prompt_message_id.is_empty(); + let mut seen_durable_seqs = std::collections::HashSet::new(); + self.stream_session_once_with_cursor( + session_id, + prompt_text, + delivery, + prompt_message_id, + prompt_dispatched, + &mut log_cursor, + &mut turn_active, + &mut initial_prompt_settled, + &mut seen_durable_seqs, + is_retry, + keep_alive, + event_tx, + ) + .await + } + + /// One connection's worth of a session stream, combining `event.subscribe`, `session.log` + /// (`follow: true`, `after: *log_cursor`), and on reconnect `session.message` + `session.wait` + /// reconciliation. + #[allow(clippy::too_many_arguments)] + async fn stream_session_once_with_cursor( + &self, + session_id: &str, + prompt_text: &str, + delivery: PromptDelivery, + prompt_message_id: &str, + prompt_dispatched: &mut bool, + log_cursor: &mut Option, + turn_active: &mut bool, + initial_prompt_settled: &mut bool, + seen_durable_seqs: &mut std::collections::HashSet, + is_retry: bool, + keep_alive: bool, + event_tx: &mpsc::UnboundedSender, ) -> Result { let (ws_stream, _) = match connect_async(&self.url).await { Ok(stream) => stream, @@ -702,9 +1037,16 @@ impl RpcClient { .map_err(|e| format!("Failed to subscribe to events: {}", e))?; let mut prompt_req_id: Option = None; + let mut log_req_id: Option = None; + let mut message_check_req_id: Option = None; + let mut wait_req_id: Option = None; + let mut completion_log_id: Option = None; + let mut pending_terminal: Option = None; + let mut log_synced = false; // On a retry, the first matching event is proof the socket is back. let mut reconnected = is_retry; let mut notices = std::collections::HashSet::new(); + let mut seen_event_ids = std::collections::HashSet::new(); while let Some(msg_res) = read.next().await { let msg = match msg_res { @@ -727,7 +1069,8 @@ impl RpcClient { match parsed { ServerMessage::Chunk { request_id, values } => { - if let Some(id_num) = as_request_id(&request_id) { + let chunk_req_id = as_request_id(&request_id); + if let Some(id_num) = chunk_req_id { let ack = RpcAck::new(id_num); if let Ok(mut ack_json) = serde_json::to_string(&ack) { ack_json.push('\n'); @@ -735,29 +1078,135 @@ impl RpcClient { } } + let from_log = log_req_id.is_some() && chunk_req_id == log_req_id; + let from_completion_log = + completion_log_id.is_some() && chunk_req_id == completion_log_id; + if chunk_req_id != Some(sub_id) && !from_log && !from_completion_log { + continue; + } for env in values { - if env.event_type == "server.connected" && reconnected { - reconnected = false; - let _ = event_tx.send(StreamEvent::Reconnected); - } - // On server.connected, dispatch prompt if not yet dispatched - if env.event_type == "server.connected" && !*prompt_dispatched { - *prompt_dispatched = true; - let prompt_id = self.next_request_id(); - prompt_req_id = Some(prompt_id); - let prompt_req = RpcRequest::new( - prompt_id, - "session.prompt", - PromptSessionPayload { + if env.event_type == "server.connected" { + if reconnected { + reconnected = false; + let _ = event_tx.send(StreamEvent::Reconnected); + } + if !prompt_message_id.is_empty() && !*prompt_dispatched { + *prompt_dispatched = true; + *turn_active = true; + let prompt_id = self.next_request_id(); + prompt_req_id = Some(prompt_id); + let prompt_req = RpcRequest::new( + prompt_id, + "session.prompt", + PromptSessionPayload { + session_id: session_id.to_string(), + text: prompt_text.to_string(), + delivery: delivery.as_str(), + id: prompt_message_id.to_string(), + }, + ); + if let Ok(mut p_json) = serde_json::to_string(&prompt_req) { + p_json.push('\n'); + let _ = write.send(Message::Text(p_json)).await; + } + } else if is_retry { + if !prompt_message_id.is_empty() && !*initial_prompt_settled + { + // Reconcile whether `prompt_message_id` materialized + // into the conversation before the disconnect. + let msg_id = self.next_request_id(); + message_check_req_id = Some(msg_id); + let msg_req = RpcRequest::new( + msg_id, + "session.message", + SessionMessagePayload { + session_id: session_id.to_string(), + message_id: prompt_message_id.to_string(), + }, + ); + if let Ok(mut m_json) = serde_json::to_string(&msg_req) + { + m_json.push('\n'); + let _ = write.send(Message::Text(m_json)).await; + } + } else if *turn_active { + let wait_id = self.next_request_id(); + wait_req_id = Some(wait_id); + let wait_req = RpcRequest::new( + wait_id, + "session.wait", + SessionIdPayload { + session_id: session_id.to_string(), + }, + ); + if let Ok(mut w_json) = serde_json::to_string(&wait_req) + { + w_json.push('\n'); + let _ = write.send(Message::Text(w_json)).await; + } + } + } else if prompt_message_id.is_empty() && *turn_active { + let wait_id = self.next_request_id(); + wait_req_id = Some(wait_id); + let wait_req = RpcRequest::new( + wait_id, + "session.wait", + SessionIdPayload { + session_id: session_id.to_string(), + }, + ); + if let Ok(mut w_json) = serde_json::to_string(&wait_req) { + w_json.push('\n'); + let _ = write.send(Message::Text(w_json)).await; + } + } + + // Subscribe to durable `session.log` (replay after cursor + follow) + let log_id = self.next_request_id(); + log_req_id = Some(log_id); + let log_req = RpcRequest::new( + log_id, + "session.log", + SessionLogPayload { session_id: session_id.to_string(), - text: prompt_text.to_string(), - delivery: delivery.as_str(), - id: prompt_message_id.to_string(), + after: *log_cursor, + follow: Some(true), }, ); - if let Ok(mut p_json) = serde_json::to_string(&prompt_req) { - p_json.push('\n'); - let _ = write.send(Message::Text(p_json.into())).await; + if let Ok(mut l_json) = serde_json::to_string(&log_req) { + l_json.push('\n'); + let _ = write.send(Message::Text(l_json)).await; + } + continue; + } + + if env.event_type == "log.synced" { + if (from_log || from_completion_log) + && env + .aggregate_id + .as_deref() + .is_none_or(|agg| agg == session_id) + { + if reconnected { + reconnected = false; + let _ = event_tx.send(StreamEvent::Reconnected); + } + advance_log_cursor(log_cursor, env.seq); + let _ = + event_tx.send(StreamEvent::LogSynced { seq: env.seq }); + if from_log { + log_synced = true; + if completion_log_id.is_none() { + if let Some(terminal) = pending_terminal.take() { + *turn_active = false; + *initial_prompt_settled = true; + let _ = event_tx.send(terminal); + if !keep_alive { + return Ok(true); + } + } + } + } } continue; } @@ -794,7 +1243,10 @@ impl RpcClient { let s_id = data.get("sessionId").and_then(|v| v.as_str()); let source_id = data.get("sourceSessionId").and_then(|v| v.as_str()); + let durable_agg = + env.durable.as_ref().map(|d| d.aggregate_id.as_str()); let matches_session = s_id == Some(session_id) + || durable_agg == Some(session_id) || (env.event_type == "session.next.forked" && source_id == Some(session_id)); if !matches_session { @@ -808,15 +1260,55 @@ impl RpcClient { let _ = event_tx.send(StreamEvent::Reconnected); } - if let Some(stream_event) = - decode_session_event(env.event_type.as_str(), data) - { + if let Some(seq) = env.durable_seq() { + if durable_agg.is_none_or(|agg| agg == session_id) { + // Only ordered replay/follow may advance `after`. A live + // event can arrive ahead of unread replay on this socket. + if from_log || from_completion_log { + advance_log_cursor(log_cursor, Some(seq)); + } + if !seen_durable_seqs.insert(seq) { + continue; + } + } + } + if let Some(id) = env.id.as_ref().filter(|s| !s.is_empty()) { + if !seen_event_ids.insert(id.clone()) { + continue; + } + } + + // Both sources use the same decoder, so either copy recovers + // completed parts when deduplication chooses its winner. + let decoded = + decode_durable_session_events(env.event_type.as_str(), data); + + for stream_event in decoded { + if matches!( + stream_event, + StreamEvent::ExecutionStarted + | StreamEvent::PromptPromoted { .. } + | StreamEvent::TurnStarted + ) { + *turn_active = true; + } let is_terminal = matches!( stream_event, StreamEvent::Succeeded | StreamEvent::Failed(_) | StreamEvent::Interrupted(_) ); + if is_terminal + && (completion_log_id.is_some() + || (is_retry && !log_synced)) + { + pending_terminal = Some(stream_event); + continue; + } + if is_terminal { + *turn_active = false; + *initial_prompt_settled = true; + } let _ = event_tx.send(stream_event); if is_terminal && !keep_alive { return Ok(true); @@ -824,15 +1316,130 @@ impl RpcClient { } } } - ServerMessage::Exit { - request_id, - exit: ExitPayload::Failure { cause }, - } => { + ServerMessage::Exit { request_id, exit } => { let id = as_request_id(&request_id); - if id == Some(sub_id) - || (prompt_req_id.is_some() && id == prompt_req_id) + if message_check_req_id.is_some() && id == message_check_req_id { + message_check_req_id = None; + match exit { + ExitPayload::Success { value } => { + let found = parse_session_message(&value).unwrap_or(false); + if found { + let wait_id = self.next_request_id(); + wait_req_id = Some(wait_id); + let wait_req = RpcRequest::new( + wait_id, + "session.wait", + SessionIdPayload { + session_id: session_id.to_string(), + }, + ); + if let Ok(mut w_json) = serde_json::to_string(&wait_req) + { + w_json.push('\n'); + let _ = write.send(Message::Text(w_json)).await; + } + } else { + *turn_active = true; + let prompt_id = self.next_request_id(); + prompt_req_id = Some(prompt_id); + let prompt_req = RpcRequest::new( + prompt_id, + "session.prompt", + PromptSessionPayload { + session_id: session_id.to_string(), + text: prompt_text.to_string(), + delivery: delivery.as_str(), + id: prompt_message_id.to_string(), + }, + ); + if let Ok(mut p_json) = + serde_json::to_string(&prompt_req) + { + p_json.push('\n'); + let _ = write.send(Message::Text(p_json)).await; + } + } + } + ExitPayload::Failure { cause } => { + return Err(format_cause(cause.as_ref())); + } + } + } else if prompt_req_id.is_some() && id == prompt_req_id { + prompt_req_id = None; + if let ExitPayload::Failure { cause } = exit { + if is_retry && is_prompt_conflict(cause.as_ref()) { + let wait_id = self.next_request_id(); + wait_req_id = Some(wait_id); + let wait_req = RpcRequest::new( + wait_id, + "session.wait", + SessionIdPayload { + session_id: session_id.to_string(), + }, + ); + if let Ok(mut w_json) = serde_json::to_string(&wait_req) { + w_json.push('\n'); + let _ = write.send(Message::Text(w_json)).await; + } + } else { + return Err(format_cause(cause.as_ref())); + } + } + } else if wait_req_id.is_some() && id == wait_req_id { + wait_req_id = None; + match exit { + ExitPayload::Success { .. } => { + if *turn_active || pending_terminal.is_some() { + // `wait` and the follower run independently. Pin a + // replay AFTER idle so all committed output is drained + // before reporting completion, even if follow lags. + pending_terminal.get_or_insert(StreamEvent::Succeeded); + let drain_id = self.next_request_id(); + completion_log_id = Some(drain_id); + let req = RpcRequest::new( + drain_id, + "session.log", + SessionLogPayload { + session_id: session_id.to_string(), + after: *log_cursor, + follow: Some(false), + }, + ); + let mut text = serde_json::to_string(&req) + .map_err(|e| e.to_string())?; + text.push('\n'); + if write.send(Message::Text(text)).await.is_err() { + return Ok(false); + } + } + } + ExitPayload::Failure { cause } => { + return Err(format_cause(cause.as_ref())); + } + } + } else if completion_log_id.is_some() && id == completion_log_id { + match exit { + ExitPayload::Success { .. } => { + completion_log_id = None; + if let Some(terminal) = pending_terminal.take() { + *turn_active = false; + *initial_prompt_settled = true; + let _ = event_tx.send(terminal); + if !keep_alive { + return Ok(true); + } + } + } + ExitPayload::Failure { cause } => { + return Err(format_cause(cause.as_ref())) + } + } + } else if id == Some(sub_id) + || (log_req_id.is_some() && id == log_req_id) { - return Err(format_cause(cause.as_ref())); + if let ExitPayload::Failure { cause } = exit { + return Err(format_cause(cause.as_ref())); + } } } ServerMessage::Defect { defect } => { @@ -878,6 +1485,31 @@ fn parse_model_list(value: &Value) -> Result { .map_err(|e| format!("Invalid model list response: {e}")) } +fn parse_session_message(value: &Value) -> Result { + serde_json::from_value::(value.clone()) + .map(|res| res.found) + .map_err(|e| format!("Invalid session.message response: {e}")) +} + +fn is_prompt_conflict(cause: Option<&Value>) -> bool { + let Some(cause) = cause else { + return false; + }; + match cause { + Value::Array(items) => items.iter().any(|item| is_prompt_conflict(Some(item))), + Value::Object(_) => { + if str_field(cause, "_tag") == Some("PromptConflictError") { + return true; + } + cause + .get("error") + .or_else(|| cause.get("failure")) + .is_some_and(|inner| is_prompt_conflict(Some(inner))) + } + _ => false, + } +} + fn str_field<'a>(value: &'a Value, key: &str) -> Option<&'a str> { value .get(key) @@ -1208,6 +1840,70 @@ fn format_tagged_error(tag: &str, err: &Value) -> String { } } +/// Decode one durable `session.log` event into its `StreamEvent`(s), including completed thinking +/// and text parts carried on `session.llm.ended` when live deltas were not observed. +pub fn decode_durable_session_events(event_type: &str, data: &Value) -> Vec { + if event_type == "session.llm.ended" { + let mut events = llm_ended_part_events(data); + if let Some(usage) = llm_ended_usage_event(data) { + events.push(usage); + } + return events; + } + decode_session_event(event_type, data).into_iter().collect() +} + +fn llm_ended_part_events(data: &Value) -> Vec { + let mut out = Vec::new(); + let Some(msg) = data.get("message") else { + return out; + }; + let message_id = data + .get("messageId") + .or_else(|| msg.get("messageId")) + .and_then(Value::as_str) + .unwrap_or_default() + .to_string(); + let Some(parts) = msg.get("parts").and_then(Value::as_array) else { + return out; + }; + for (part_index, part) in parts.iter().enumerate() { + match part.get("type").and_then(Value::as_str) { + Some("thinking") => { + if part.get("redacted").and_then(Value::as_bool) == Some(true) { + continue; + } + if let Some(thinking) = part + .get("thinking") + .and_then(Value::as_str) + .filter(|s| !s.is_empty()) + { + out.push(StreamEvent::ThinkingEnd { + message_id: message_id.clone(), + part_index, + content: thinking.to_string(), + }); + } + } + Some("text") => { + if let Some(text) = part + .get("text") + .and_then(Value::as_str) + .filter(|s| !s.is_empty()) + { + out.push(StreamEvent::TextEnd { + message_id: message_id.clone(), + part_index, + content: text.to_string(), + }); + } + } + _ => {} + } + } + out +} + /// Decode one session-scoped `EventEnvelope` payload into a `StreamEvent`. pub fn decode_session_event(event_type: &str, data: &Value) -> Option { match event_type { @@ -1685,6 +2381,216 @@ mod tests { use super::*; use serde_json::json; + type TestSocket = tokio_tungstenite::WebSocketStream; + + async fn request(ws: &mut TestSocket, tag: &str) -> Value { + while let Some(message) = ws.next().await { + if let Message::Text(text) = message.unwrap() { + let value: Value = serde_json::from_str(text.trim()).unwrap(); + if value["tag"] == tag { + return value; + } + } + } + panic!("expected {tag}"); + } + + async fn chunk(ws: &mut TestSocket, id: &Value, values: Value) { + let value = json!({"_tag": "Chunk", "requestId": id, "values": values}); + ws.send(Message::Text(format!("{value}\n"))).await.unwrap(); + } + + async fn success(ws: &mut TestSocket, id: &Value, value: Value) { + let reply = json!({"_tag": "Exit", "requestId": id, + "exit": {"_tag": "Success", "value": value}}); + ws.send(Message::Text(format!("{reply}\n"))).await.unwrap(); + } + + #[tokio::test] + async fn live_events_do_not_skip_unread_log_or_suppress_completed_parts() { + tokio::time::timeout(Duration::from_secs(5), async { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let client = RpcClient::new(format!("ws://{}", listener.local_addr().unwrap())); + let server = tokio::spawn(async move { + let ended = json!({"id": "end", "type": "session.llm.ended", + "durable": {"aggregateId": "ses", "seq": 20, "version": 1}, + "data": {"sessionId": "ses", "messageId": "answer", "message": { + "parts": [{"type": "text", "text": "Recovered answer"}], + "usage": {"totalTokens": 3, "input": 1, "output": 2} + }}}); + for after in [10, 11] { + let (socket, _) = listener.accept().await.unwrap(); + let mut ws = tokio_tungstenite::accept_async(socket).await.unwrap(); + let sub = request(&mut ws, "event.subscribe").await; + chunk(&mut ws, &sub["id"], json!([{ "type": "server.connected" }])).await; + let log = request(&mut ws, "session.log").await; + assert_eq!(log["payload"]["after"], after); + if after == 10 { + // The live head arrives before the ordered replay catches up. + chunk(&mut ws, &sub["id"], json!([ended])).await; + chunk( + &mut ws, + &log["id"], + json!([{ + "type": "session.config.changed", + "durable": {"aggregateId": "ses", "seq": 11, "version": 1}, + "data": {"sessionId": "ses", "thinkingLevel": "low"} + }]), + ) + .await; + } else { + chunk( + &mut ws, + &log["id"], + json!([{ + "type": "session.config.changed", + "durable": {"aggregateId": "ses", "seq": 12, "version": 1}, + "data": {"sessionId": "ses", "thinkingLevel": "high"} + }, ended, {"type": "log.synced", "aggregateId": "ses", "seq": 20}]), + ) + .await; + } + ws.close(None).await.unwrap(); + } + }); + let (tx, mut rx) = mpsc::unbounded_channel(); + let mut cursor = Some(10); + let mut seen = std::collections::HashSet::new(); + for retry in [false, true] { + assert!(!client + .stream_session_once_with_cursor( + "ses", + "", + PromptDelivery::FollowUp, + "", + &mut true, + &mut cursor, + &mut false, + &mut true, + &mut seen, + retry, + true, + &tx + ) + .await + .unwrap()); + if !retry { + assert_eq!(cursor, Some(11)); + } + } + assert_eq!(cursor, Some(20)); + let mut events = Vec::new(); + while let Ok(event) = rx.try_recv() { + events.push(event); + } + assert_eq!( + events + .iter() + .filter(|e| matches!(e, StreamEvent::TextEnd { content, .. } + if content == "Recovered answer")) + .count(), + 1 + ); + assert_eq!( + events + .iter() + .filter(|e| matches!(e, StreamEvent::Usage(_))) + .count(), + 1 + ); + assert!(events.contains(&StreamEvent::ConfigChanged { + model: None, + thinking_level: Some("high".into()) + })); + server.await.unwrap(); + }) + .await + .expect("ordered replay recovery"); + } + + #[tokio::test] + async fn idle_reconciliation_drains_output_before_completing() { + // Exercise wait before initial replay, and wait after the initial replay marker. + for synced_first in [false, true] { + tokio::time::timeout(Duration::from_secs(5), async { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let client = RpcClient::new(format!("ws://{}", listener.local_addr().unwrap())); + let server = tokio::spawn(async move { + let (socket, _) = listener.accept().await.unwrap(); + let mut ws = tokio_tungstenite::accept_async(socket).await.unwrap(); + let sub = request(&mut ws, "event.subscribe").await; + chunk(&mut ws, &sub["id"], json!([{ "type": "server.connected" }])).await; + let check = request(&mut ws, "session.message").await; + let log = request(&mut ws, "session.log").await; + if synced_first { + chunk( + &mut ws, + &log["id"], + json!([{"type": "log.synced", "aggregateId": "ses", "seq": 1}]), + ) + .await; + } + success(&mut ws, &check["id"], json!({"found": true})).await; + let wait = request(&mut ws, "session.wait").await; + success(&mut ws, &wait["id"], Value::Null).await; + // Idle doesn't mean the independent log stream has delivered its tail. + let drain = request(&mut ws, "session.log").await; + assert_eq!(drain["payload"]["follow"], false); + assert_eq!(drain["payload"]["after"], 1); + chunk( + &mut ws, + &drain["id"], + json!([{ + "id": "end", "type": "session.llm.ended", + "durable": {"aggregateId": "ses", "seq": 2, "version": 1}, + "data": {"sessionId": "ses", "messageId": "answer", "message": { + "parts": [{"type": "text", "text": "Offline output"}]}} + }, {"type": "log.synced", "aggregateId": "ses", "seq": 2}]), + ) + .await; + success(&mut ws, &drain["id"], Value::Null).await; + }); + let (tx, mut rx) = mpsc::unbounded_channel(); + assert!(client + .stream_session_once_with_cursor( + "ses", + "prompt", + PromptDelivery::FollowUp, + "prompt", + &mut true, + &mut Some(1), + &mut true, + &mut false, + &mut std::collections::HashSet::new(), + true, + false, + &tx + ) + .await + .unwrap()); + let mut events = Vec::new(); + while let Ok(event) = rx.try_recv() { + events.push(event); + } + let output = events + .iter() + .position(|e| { + matches!(e, StreamEvent::TextEnd {content, ..} + if content == "Offline output") + }) + .expect("recovered output"); + let terminal = events + .iter() + .position(|e| *e == StreamEvent::Succeeded) + .expect("completion"); + assert!(output < terminal); + server.await.unwrap(); + }) + .await + .expect("completion replay barrier"); + } + } + #[test] fn parses_a_started_call_with_its_arguments() { let data = json!({ @@ -3113,6 +4019,580 @@ mod tests { server.await.expect("server task"); } + #[tokio::test] + async fn reconnect_redispatches_prompt_when_session_message_reports_not_found() { + use tokio::net::TcpListener; + use tokio_tungstenite::accept_async; + + tokio::time::timeout(Duration::from_secs(5), async { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let addr = listener.local_addr().expect("addr"); + let url = format!("ws://{addr}/rpc/v1"); + + let server = tokio::spawn(async move { + // 1. First connection: drops right after the client dispatches `session.prompt`. + let (socket, _) = listener.accept().await.expect("accept 1"); + let mut ws = accept_async(socket).await.expect("ws 1"); + let Some(Ok(Message::Text(sub_text))) = ws.next().await else { + panic!("expected event.subscribe"); + }; + let sub_req: Value = serde_json::from_str(sub_text.trim()).expect("sub json"); + let sub_id = sub_req.get("id").cloned().unwrap_or(json!(1)); + let connected = json!({ + "_tag": "Chunk", + "requestId": sub_id, + "values": [{ "type": "server.connected" }] + }); + ws.send(Message::Text(format!("{connected}\n"))) + .await + .expect("send connected 1"); + + let mut saw_first_prompt = false; + while !saw_first_prompt { + let Some(Ok(Message::Text(txt))) = ws.next().await else { + panic!("expected first session.prompt"); + }; + for line in txt.lines() { + let msg: Value = serde_json::from_str(line.trim()).expect("json"); + if msg.get("tag").and_then(Value::as_str) == Some("session.prompt") { + saw_first_prompt = true; + } + } + } + // Drop connection before admitting the prompt. + drop(ws); + + // 2. Reconnect: client checks `session.message`, receives `{ found: false }`, and + // re-dispatches `session.prompt` with the same `messageId`. + let (socket, _) = listener.accept().await.expect("accept 2"); + let mut ws = accept_async(socket).await.expect("ws 2"); + let Some(Ok(Message::Text(sub_text))) = ws.next().await else { + panic!("expected event.subscribe on retry"); + }; + let sub_req: Value = serde_json::from_str(sub_text.trim()).expect("sub json 2"); + let sub_id = sub_req.get("id").cloned().unwrap_or(json!(1)); + let connected = json!({ + "_tag": "Chunk", + "requestId": sub_id, + "values": [{ "type": "server.connected" }] + }); + ws.send(Message::Text(format!("{connected}\n"))) + .await + .expect("send connected 2"); + + let mut msg_check_id = None; + while msg_check_id.is_none() { + let Some(Ok(Message::Text(txt))) = ws.next().await else { + panic!("expected session.message request"); + }; + for line in txt.lines() { + let msg: Value = serde_json::from_str(line.trim()).expect("json"); + if msg.get("tag").and_then(Value::as_str) == Some("session.message") { + assert_eq!(msg["payload"]["sessionId"], "ses_retry"); + assert_eq!(msg["payload"]["messageId"], "msg_retry_1"); + msg_check_id = msg.get("id").cloned(); + } + } + } + + let log = request(&mut ws, "session.log").await; + chunk( + &mut ws, + &log["id"], + json!([{ + "type": "log.synced", "aggregateId": "ses_retry" + }]), + ) + .await; + + let not_found = json!({ + "_tag": "Exit", + "requestId": msg_check_id.unwrap(), + "exit": { "_tag": "Success", "value": { "found": false } } + }); + ws.send(Message::Text(format!("{not_found}\n"))) + .await + .expect("send found: false"); + + let mut redispatched_prompt_id = None; + while redispatched_prompt_id.is_none() { + let Some(Ok(Message::Text(txt))) = ws.next().await else { + panic!("expected re-dispatched session.prompt"); + }; + for line in txt.lines() { + let msg: Value = serde_json::from_str(line.trim()).expect("json"); + if msg.get("tag").and_then(Value::as_str) == Some("session.prompt") { + assert_eq!(msg["payload"]["sessionId"], "ses_retry"); + assert_eq!(msg["payload"]["id"], "msg_retry_1"); + assert_eq!(msg["payload"]["text"], "Retry me"); + redispatched_prompt_id = msg.get("id").cloned(); + } + } + } + + let prompt_ok = json!({ + "_tag": "Exit", + "requestId": redispatched_prompt_id.unwrap(), + "exit": { "_tag": "Success", "value": Value::Null } + }); + ws.send(Message::Text(format!("{prompt_ok}\n"))) + .await + .expect("send prompt ok"); + + let chunk = json!({ + "_tag": "Chunk", + "requestId": sub_id, + "values": [ + { + "type": "session.llm.text.delta", + "data": { + "sessionId": "ses_retry", + "messageId": "msg_a1", + "partIndex": 0, + "delta": "Recovered!" + } + }, + { + "type": "session.execution.succeeded", + "data": { "sessionId": "ses_retry" } + } + ] + }); + ws.send(Message::Text(format!("{chunk}\n"))) + .await + .expect("send completion chunk"); + }); + + let client = RpcClient::new(url); + let (tx, mut rx) = mpsc::unbounded_channel(); + let mut prompt_dispatched = false; + let mut log_cursor = None; + let mut turn_active = true; + let mut initial_settled = false; + let mut seen_seqs = std::collections::HashSet::new(); + + // Attempt 1 drops after dispatching `session.prompt`. + let first = client + .stream_session_once_with_cursor( + "ses_retry", + "Retry me", + PromptDelivery::FollowUp, + "msg_retry_1", + &mut prompt_dispatched, + &mut log_cursor, + &mut turn_active, + &mut initial_settled, + &mut seen_seqs, + false, + false, + &tx, + ) + .await + .expect("attempt 1"); + assert!(!first); + assert!(prompt_dispatched); + + // Attempt 2 reconciles via `session.message` (`found: false`) and re-dispatches. + let second = client + .stream_session_once_with_cursor( + "ses_retry", + "Retry me", + PromptDelivery::FollowUp, + "msg_retry_1", + &mut prompt_dispatched, + &mut log_cursor, + &mut turn_active, + &mut initial_settled, + &mut seen_seqs, + true, + false, + &tx, + ) + .await + .expect("attempt 2"); + assert!(second); + + let mut received = Vec::new(); + while let Ok(evt) = rx.try_recv() { + received.push(evt); + } + assert!(received.contains(&StreamEvent::Reconnected)); + assert!(received.contains(&StreamEvent::TextDelta { + message_id: Some("msg_a1".to_string()), + part_index: Some(0), + delta: "Recovered!".to_string(), + })); + assert!(received.contains(&StreamEvent::Succeeded)); + + server.await.expect("server task"); + }) + .await + .expect("reconnect redispatch test timed out"); + } + + #[tokio::test] + async fn reconnect_resolves_settled_turn_via_session_message_session_wait_and_session_log() { + use tokio::net::TcpListener; + use tokio_tungstenite::accept_async; + + tokio::time::timeout(Duration::from_secs(5), async { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let addr = listener.local_addr().expect("addr"); + let url = format!("ws://{addr}/rpc/v1"); + + let server = tokio::spawn(async move { + let (socket, _) = listener.accept().await.expect("accept reconnect"); + let mut ws = accept_async(socket).await.expect("ws handshake"); + + let Some(Ok(Message::Text(sub_text))) = ws.next().await else { + panic!("expected event.subscribe"); + }; + let sub_req: Value = serde_json::from_str(sub_text.trim()).expect("sub json"); + let sub_id = sub_req.get("id").cloned().unwrap_or(json!(1)); + let connected = json!({ + "_tag": "Chunk", + "requestId": sub_id, + "values": [{ "type": "server.connected" }] + }); + ws.send(Message::Text(format!("{connected}\n"))) + .await + .expect("send connected"); + + let mut msg_check_id = None; + let mut log_id = None; + while msg_check_id.is_none() || log_id.is_none() { + let Some(Ok(Message::Text(txt))) = ws.next().await else { + panic!("expected session.message and session.log"); + }; + for line in txt.lines() { + let msg: Value = serde_json::from_str(line.trim()).expect("json"); + match msg.get("tag").and_then(Value::as_str) { + Some("session.message") => { + assert_eq!(msg["payload"]["sessionId"], "ses_settled"); + assert_eq!(msg["payload"]["messageId"], "msg_settled_1"); + msg_check_id = msg.get("id").cloned(); + } + Some("session.log") => { + assert_eq!(msg["payload"]["sessionId"], "ses_settled"); + assert_eq!(msg["payload"]["after"], 1); + assert_eq!(msg["payload"]["follow"], true); + log_id = msg.get("id").cloned(); + } + _ => {} + } + } + } + + // Replay the durable events that committed during the disconnect window + log.synced. + let log_chunk = json!({ + "_tag": "Chunk", + "requestId": log_id.unwrap(), + "values": [ + { + "id": "evt_2", + "type": "session.llm.started", + "durable": { "aggregateId": "ses_settled", "seq": 2, "version": 1 }, + "data": { + "sessionId": "ses_settled", + "messageId": "msg_a1", + "message": { "model": "gpt-5.6-luna" } + } + }, + { + "id": "evt_3", + "type": "session.llm.ended", + "durable": { "aggregateId": "ses_settled", "seq": 3, "version": 1 }, + "data": { + "sessionId": "ses_settled", + "messageId": "msg_a1", + "reason": "stop", + "message": { + "messageId": "msg_a1", + "role": "assistant", + "model": "gpt-5.6-luna", + "parts": [ + { "type": "thinking", "thinking": "Settled while offline." }, + { "type": "text", "text": "Completed during disconnect." } + ], + "usage": { + "totalTokens": 90, + "input": 60, + "output": 30, + "cost": { "total": 0.0015 } + } + } + } + }, + { + "id": "evt_4", + "type": "session.turn.ended", + "durable": { "aggregateId": "ses_settled", "seq": 4, "version": 1 }, + "data": { + "sessionId": "ses_settled", + "messageId": "msg_a1" + } + }, + { + "type": "log.synced", + "aggregateId": "ses_settled", + "seq": 6 + } + ] + }); + ws.send(Message::Text(format!("{log_chunk}\n"))) + .await + .expect("send log chunk"); + + // Report that `msg_settled_1` materialized into the conversation. + let found_reply = json!({ + "_tag": "Exit", + "requestId": msg_check_id.unwrap(), + "exit": { "_tag": "Success", "value": { "found": true } } + }); + ws.send(Message::Text(format!("{found_reply}\n"))) + .await + .expect("send found: true"); + + // Client must now send `session.wait` (and NOT re-send `session.prompt`). + let mut wait_id = None; + while wait_id.is_none() { + let Some(Ok(Message::Text(txt))) = ws.next().await else { + panic!("expected session.wait request"); + }; + for line in txt.lines() { + let msg: Value = serde_json::from_str(line.trim()).expect("json"); + assert_ne!( + msg.get("tag").and_then(Value::as_str), + Some("session.prompt"), + "must not re-dispatch already promoted prompt" + ); + if msg.get("tag").and_then(Value::as_str) == Some("session.wait") { + assert_eq!(msg["payload"]["sessionId"], "ses_settled"); + wait_id = msg.get("id").cloned(); + } + } + } + + // Session is already idle because the turn finished while disconnected. + let wait_reply = json!({ + "_tag": "Exit", + "requestId": wait_id.unwrap(), + "exit": { "_tag": "Success", "value": Value::Null } + }); + ws.send(Message::Text(format!("{wait_reply}\n"))) + .await + .expect("send wait reply"); + + let drain = request(&mut ws, "session.log").await; + assert_eq!(drain["payload"]["after"], 6); + assert_eq!(drain["payload"]["follow"], false); + chunk(&mut ws, &drain["id"], json!([{ + "type": "log.synced", "aggregateId": "ses_settled", "seq": 6 + }])).await; + success(&mut ws, &drain["id"], Value::Null).await; + }); + + let client = RpcClient::new(url); + let (tx, mut rx) = mpsc::unbounded_channel(); + let mut prompt_dispatched = true; + let mut log_cursor = Some(1); + let mut turn_active = true; + let mut initial_settled = false; + let mut seen_seqs = std::collections::HashSet::new(); + + let done = client + .stream_session_once_with_cursor( + "ses_settled", + "Hello", + PromptDelivery::FollowUp, + "msg_settled_1", + &mut prompt_dispatched, + &mut log_cursor, + &mut turn_active, + &mut initial_settled, + &mut seen_seqs, + true, + false, + &tx, + ) + .await + .expect("stream_session_once_with_cursor"); + assert!(done); + assert!(!turn_active); + assert!(initial_settled); + // Cursor advances to `max(last durable.seq (4), last log.synced.seq (6))` = 6. + assert_eq!(log_cursor, Some(6)); + + let mut received = Vec::new(); + while let Ok(evt) = rx.try_recv() { + received.push(evt); + } + assert!(received.contains(&StreamEvent::Reconnected)); + assert!(received.contains(&StreamEvent::ThinkingEnd { + message_id: "msg_a1".to_string(), + part_index: 0, + content: "Settled while offline.".to_string(), + })); + assert!(received.contains(&StreamEvent::TextEnd { + message_id: "msg_a1".to_string(), + part_index: 1, + content: "Completed during disconnect.".to_string(), + })); + assert!(received.contains(&StreamEvent::LogSynced { seq: Some(6) })); + assert!(received.contains(&StreamEvent::Succeeded)); + + server.await.expect("server task"); + }) + .await + .expect("reconnect settled turn test timed out"); + } + + #[tokio::test] + async fn session_log_tracks_max_cursor_across_replay_follow_and_reconnect() { + use tokio::net::TcpListener; + use tokio_tungstenite::accept_async; + + tokio::time::timeout(Duration::from_secs(5), async { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let addr = listener.local_addr().expect("addr"); + let url = format!("ws://{addr}/rpc/v1"); + + let server = tokio::spawn(async move { + // 1. First connection: `after` is None; replays seq 0, sends `log.synced` at seq 3 + // (head past last event), then follows seq 4 before dropping. + let (socket, _) = listener.accept().await.expect("accept 1"); + let mut ws = accept_async(socket).await.expect("ws 1"); + let Some(Ok(Message::Text(req_text))) = ws.next().await else { + panic!("expected session.log 1"); + }; + let req: Value = serde_json::from_str(req_text.trim()).expect("json 1"); + assert_eq!(req["tag"], "session.log"); + assert_eq!(req["payload"]["sessionId"], "ses_log"); + assert!(req["payload"].get("after").is_none()); + assert_eq!(req["payload"]["follow"], true); + let log_id = req["id"].clone(); + + let chunk = json!({ + "_tag": "Chunk", + "requestId": log_id, + "values": [ + { + "id": "evt_0", + "type": "session.prompt.promoted", + "durable": { "aggregateId": "ses_log", "seq": 0, "version": 1 }, + "data": { + "sessionId": "ses_log", + "messageId": "msg_u0", + "delivery": "followUp", + "prompt": { "parts": [{ "type": "text", "text": "First prompt" }] } + } + }, + { + "type": "log.synced", + "aggregateId": "ses_log", + "seq": 3 + }, + { + "id": "evt_4", + "type": "session.turn.ended", + "durable": { "aggregateId": "ses_log", "seq": 4, "version": 1 }, + "data": { + "sessionId": "ses_log", + "messageId": "msg_a0" + } + } + ] + }); + ws.send(Message::Text(format!("{chunk}\n"))) + .await + .expect("send chunk 1"); + let _ = ws.next().await; + drop(ws); + + // 2. Second connection: resumes with `after: 4` (`max(durable.seq 4, synced.seq 3)`), + // receives `log.synced` with `seq: 7` (forked head above last event), then ends. + let (socket, _) = listener.accept().await.expect("accept 2"); + let mut ws = accept_async(socket).await.expect("ws 2"); + let Some(Ok(Message::Text(req_text))) = ws.next().await else { + panic!("expected session.log 2"); + }; + let req: Value = serde_json::from_str(req_text.trim()).expect("json 2"); + assert_eq!(req["tag"], "session.log"); + assert_eq!(req["payload"]["sessionId"], "ses_log"); + assert_eq!(req["payload"]["after"], 4); + assert_eq!(req["payload"]["follow"], false); + let log_id = req["id"].clone(); + + let chunk = json!({ + "_tag": "Chunk", + "requestId": log_id, + "values": [ + { + "type": "log.synced", + "aggregateId": "ses_log", + "seq": 7 + } + ] + }); + ws.send(Message::Text(format!("{chunk}\n"))) + .await + .expect("send chunk 2"); + let exit = json!({ + "_tag": "Exit", + "requestId": log_id, + "exit": { "_tag": "Success", "value": Value::Null } + }); + ws.send(Message::Text(format!("{exit}\n"))) + .await + .expect("send exit 2"); + }); + + let client = RpcClient::new(url); + let (tx, mut rx) = mpsc::unbounded_channel(); + let mut cursor = None; + let mut seen_seqs = std::collections::HashSet::new(); + + let finished_first = client + .stream_session_log_once("ses_log", true, &mut cursor, &mut seen_seqs, false, &tx) + .await + .expect("log 1"); + assert!(!finished_first); + assert_eq!(cursor, Some(4)); + + let finished_second = client + .stream_session_log_once("ses_log", false, &mut cursor, &mut seen_seqs, true, &tx) + .await + .expect("log 2"); + assert!(finished_second); + assert_eq!(cursor, Some(7)); + + let mut received = Vec::new(); + while let Ok(evt) = rx.try_recv() { + received.push(evt); + } + assert_eq!( + received, + vec![ + StreamEvent::PromptPromoted { + message_id: "msg_u0".to_string(), + prompt: "First prompt".to_string(), + delivery: PromptDelivery::FollowUp, + }, + StreamEvent::LogSynced { seq: Some(3) }, + StreamEvent::TurnEnded { + message_id: "msg_a0".to_string(), + }, + StreamEvent::Reconnected, + StreamEvent::LogSynced { seq: Some(7) }, + ] + ); + + server.await.expect("server task"); + }) + .await + .expect("session_log cursor test timed out"); + } + #[tokio::test] #[ignore] async fn test_rpc_flow() { diff --git a/packages/tui/src/rpc/protocol.rs b/packages/tui/src/rpc/protocol.rs index 447c6128..b5c94c87 100644 --- a/packages/tui/src/rpc/protocol.rs +++ b/packages/tui/src/rpc/protocol.rs @@ -86,15 +86,64 @@ pub enum ExitPayload { Failure { cause: Option }, } +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct DurableMeta { + #[serde(rename = "aggregateId")] + pub aggregate_id: String, + pub seq: u64, + #[serde(default)] + pub version: u64, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct LogSynced { + #[serde(rename = "aggregateId")] + pub aggregate_id: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub seq: Option, +} + #[derive(Debug, Clone, Deserialize)] pub struct EventEnvelope { - #[allow(dead_code)] pub id: Option, #[serde(rename = "type")] pub event_type: String, + #[serde(default)] + pub durable: Option, + #[serde(rename = "aggregateId", default)] + pub aggregate_id: Option, + #[serde(default)] + pub seq: Option, pub data: Option, } +impl EventEnvelope { + #[allow(dead_code)] + pub fn as_log_synced(&self) -> Option { + if self.event_type == "log.synced" { + Some(LogSynced { + aggregate_id: self.aggregate_id.clone().unwrap_or_default(), + seq: self.seq, + }) + } else { + None + } + } + + pub fn durable_seq(&self) -> Option { + self.durable.as_ref().map(|d| d.seq) + } + + #[allow(dead_code)] + pub fn cursor_seq(&self) -> Option { + if self.event_type == "log.synced" { + self.seq + } else { + self.durable_seq() + } + } +} + #[derive(Debug, Serialize)] pub struct CreateSessionPayload { #[serde(rename = "hostDir", skip_serializing_if = "Option::is_none")] @@ -133,12 +182,30 @@ pub struct PromptSessionPayload { pub id: String, } -#[derive(Debug, Serialize)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] pub struct SessionIdPayload { #[serde(rename = "sessionId")] pub session_id: String, } +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub struct SessionMessagePayload { + #[serde(rename = "sessionId")] + pub session_id: String, + #[serde(rename = "messageId")] + pub message_id: String, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub struct SessionLogPayload { + #[serde(rename = "sessionId")] + pub session_id: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub after: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub follow: Option, +} + #[derive(Debug, Serialize)] pub struct ListModelsPayload<'a> { #[serde(rename = "hostDir", skip_serializing_if = "Option::is_none")] @@ -150,3 +217,9 @@ pub struct SessionHistoryResponse { #[serde(default)] pub entries: Vec, } + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct SessionMessageResponse { + #[serde(default)] + pub found: bool, +}