//! L1 tests for [`ChatBridge`] — the structured-reply ↔ Channel registry, the //! twin of [`PtyBridge`] (ARCHITECTURE §17.7) — and for [`chunk_from_event`], the //! single `ReplyEvent → ReplyChunk` translation point. //! //! These exercise the load-bearing transport logic the D4 commands are thin //! shells over: the `agent_send` pump's delivery + scrollback recording, the //! `reattach_agent_chat` repaint-without-respawn, the `close_agent_session` //! teardown, and — the most critical invariant — generation supersede (no double //! emission after a re-attach). A real [`tauri::ipc::Channel`] built from a //! capturing closure is used (no Tauri runtime, no real CLI/PTY/session). use std::sync::{Arc, Mutex}; use tauri::ipc::{Channel, InvokeResponseBody}; use app_tauri_lib::chat::{chunk_from_event, ChatBridge}; use app_tauri_lib::dto::ReplyChunk; use domain::ids::SessionId; use domain::ports::ReplyEvent; use uuid::Uuid; /// Builds a `Channel` whose sent chunks are recorded into `sink`. /// /// `ReplyChunk` is `Serialize`/`Deserialize`, so chunks arrive as a JSON string /// in an `InvokeResponseBody::Json`; we parse them back to `ReplyChunk` for /// assertions (round-tripping the exact camelCase tagged shape on the way). fn capturing_channel(sink: Arc>>) -> Channel { Channel::new(move |body: InvokeResponseBody| { let chunk: ReplyChunk = match body { InvokeResponseBody::Json(s) => serde_json::from_str(&s).unwrap(), InvokeResponseBody::Raw(b) => serde_json::from_slice(&b).unwrap(), }; sink.lock().unwrap().push(chunk); Ok(()) }) } fn sid() -> SessionId { SessionId::from_uuid(Uuid::new_v4()) } fn delta(s: &str) -> ReplyChunk { ReplyChunk::TextDelta { text: s.to_owned() } } fn final_chunk(s: &str) -> ReplyChunk { ReplyChunk::Final { content: s.to_owned(), } } // --------------------------------------------------------------------------- // chunk_from_event — exhaustive mapping (zone 6) // --------------------------------------------------------------------------- #[test] fn chunk_from_event_maps_text_delta() { assert_eq!( chunk_from_event(ReplyEvent::TextDelta { text: "hi".into() }), ReplyChunk::TextDelta { text: "hi".into() } ); } #[test] fn chunk_from_event_maps_tool_activity() { assert_eq!( chunk_from_event(ReplyEvent::ToolActivity { label: "reads file".into() }), ReplyChunk::ToolActivity { label: "reads file".into() } ); } #[test] fn chunk_from_event_maps_final() { assert_eq!( chunk_from_event(ReplyEvent::Final { content: "done".into() }), ReplyChunk::Final { content: "done".into() } ); } // --------------------------------------------------------------------------- // agent_send pump behaviour: deltas* then exactly one Final (zone 2) // --------------------------------------------------------------------------- /// Replays a turn's events through the bridge exactly as the `agent_send` pump /// does (`chunk_from_event` + `send_output`), then `detach_if(gen)` at stream /// end. This is the body of the spawned pump thread, run inline. fn pump_turn(bridge: &ChatBridge, session: &SessionId, gen: u64, events: Vec) { for event in events { let _ = bridge.send_output(session, chunk_from_event(event)); } bridge.detach_if(session, gen); } #[test] fn pump_delivers_deltas_then_exactly_one_final_in_order() { let bridge = ChatBridge::new(); let session = sid(); let sink = Arc::new(Mutex::new(Vec::new())); let gen = bridge.register(session, capturing_channel(Arc::clone(&sink))); pump_turn( &bridge, &session, gen, vec![ ReplyEvent::TextDelta { text: "Hel".into() }, ReplyEvent::TextDelta { text: "lo".into() }, ReplyEvent::Final { content: "Hello".into(), }, ], ); let got = sink.lock().unwrap(); assert_eq!( got.as_slice(), &[delta("Hel"), delta("lo"), final_chunk("Hello")], "deltas in order, then the single Final" ); // Exactly one Final, and it is last (deterministic terminal chunk). let finals = got.iter().filter(|c| matches!(c, ReplyChunk::Final { .. })).count(); assert_eq!(finals, 1, "exactly one Final per turn"); assert!(matches!(got.last(), Some(ReplyChunk::Final { .. }))); } #[test] fn pump_with_only_a_final_delivers_just_the_final() { // Zero deltas is valid (deltas*); the turn still ends on exactly one Final. let bridge = ChatBridge::new(); let session = sid(); let sink = Arc::new(Mutex::new(Vec::new())); let gen = bridge.register(session, capturing_channel(Arc::clone(&sink))); pump_turn( &bridge, &session, gen, vec![ReplyEvent::Final { content: "ok".into(), }], ); assert_eq!(sink.lock().unwrap().as_slice(), &[final_chunk("ok")]); } // --------------------------------------------------------------------------- // Generation supersede — the critical invariant (zone 1) // --------------------------------------------------------------------------- /// A `reattach_agent_chat` mid-turn: a first pump is in flight (old generation) /// when the view re-attaches (new channel, bumped generation). The old pump's /// remaining chunks must NOT reach the *new* channel, and when the old pump ends /// its `detach_if(old_gen)` must be a no-op (never tearing down the live one). #[test] fn reattach_supersedes_old_pump_no_double_emission() { let bridge = ChatBridge::new(); let session = sid(); let old = Arc::new(Mutex::new(Vec::new())); let new = Arc::new(Mutex::new(Vec::new())); // Turn starts: old channel registered, pump emits a first delta. let old_gen = bridge.register(session, capturing_channel(Arc::clone(&old))); bridge.send_output(&session, delta("a")); // delivered to old // View navigates away & back → reattach registers a NEW channel (bumps gen). let _new_gen = bridge.register(session, capturing_channel(Arc::clone(&new))); // Old pump keeps draining its (now stale) stream, then ends with its own gen. bridge.send_output(&session, delta("b")); // recorded to scrollback, routed to NEW channel bridge.detach_if(&session, old_gen); // MUST be a no-op (newer gen present) // The new (live) attach must NOT have been detached: it is still deliverable. assert!( bridge.send_output(&session, final_chunk("done")), "live re-attach channel must survive a superseded pump's detach_if" ); // The OLD channel only ever saw the single pre-reattach delta — no double emit. assert_eq!( old.lock().unwrap().as_slice(), &[delta("a")], "superseded channel receives nothing after the reattach" ); } #[test] fn detach_if_current_generation_stops_delivery() { // When the pump that owns the current generation ends (e.g. Final reached and // no reattach happened), detach_if drops the channel: further output is not // delivered (turn over) but the session stays known until close. let bridge = ChatBridge::new(); let session = sid(); let sink = Arc::new(Mutex::new(Vec::new())); let gen = bridge.register(session, capturing_channel(Arc::clone(&sink))); bridge.send_output(&session, final_chunk("end")); bridge.detach_if(&session, gen); // Channel detached: a stray chunk is recorded to scrollback but not delivered. assert!(!bridge.send_output(&session, delta("late"))); assert_eq!(sink.lock().unwrap().as_slice(), &[final_chunk("end")]); // Session is still tracked (scrollback survives until close). assert_eq!(bridge.active_sessions(), 1); } // --------------------------------------------------------------------------- // reattach_agent_chat: scrollback repaint without re-spawn (zone 3) // --------------------------------------------------------------------------- #[test] fn scrollback_accumulates_every_routed_chunk_in_order() { let bridge = ChatBridge::new(); let session = sid(); let gen = bridge.register(session, capturing_channel(Arc::new(Mutex::new(Vec::new())))); pump_turn( &bridge, &session, gen, vec![ ReplyEvent::TextDelta { text: "x".into() }, ReplyEvent::ToolActivity { label: "runs".into() }, ReplyEvent::Final { content: "x".into() }, ], ); assert_eq!( bridge.scrollback(&session), vec![ delta("x"), ReplyChunk::ToolActivity { label: "runs".into() }, final_chunk("x"), ], "scrollback retains the full conversation, in order" ); } #[test] fn reattach_repaints_scrollback_and_preserves_it_without_resend() { // Models reattach_agent_chat: a first turn streamed, the view detached, then a // new channel registers. The scrollback survives the re-register (no session // method is called — no re-spawn), and is what the command returns. let bridge = ChatBridge::new(); let session = sid(); let g0 = bridge.register(session, capturing_channel(Arc::new(Mutex::new(Vec::new())))); pump_turn( &bridge, &session, g0, vec![ ReplyEvent::TextDelta { text: "Hi".into() }, ReplyEvent::Final { content: "Hi".into(), }, ], ); // Snapshot exactly what reattach_agent_chat returns BEFORE swapping channel. let repainted = bridge.scrollback(&session); assert_eq!(repainted, vec![delta("Hi"), final_chunk("Hi")]); // Re-attach: register a fresh channel. Scrollback is preserved (not cleared, // not duplicated), generation bumped. let new_sink = Arc::new(Mutex::new(Vec::new())); let g1 = bridge.register(session, capturing_channel(Arc::clone(&new_sink))); assert_eq!(g1, g0 + 1, "re-attach bumps the generation"); assert_eq!( bridge.scrollback(&session), repainted, "scrollback unchanged by re-attach (no re-send, no re-spawn)" ); // The fresh channel got nothing yet: reattach does NOT replay over the channel // (the command returns the scrollback for the front to replay itself). assert!( new_sink.lock().unwrap().is_empty(), "re-attach must not push the old turns onto the new channel" ); } #[test] fn scrollback_of_unknown_session_is_empty() { let bridge = ChatBridge::new(); assert!(bridge.scrollback(&sid()).is_empty()); } // --------------------------------------------------------------------------- // close_agent_session: unregister purges channel AND scrollback (zone 4) // --------------------------------------------------------------------------- #[test] fn unregister_purges_channel_and_scrollback() { let bridge = ChatBridge::new(); let session = sid(); let sink = Arc::new(Mutex::new(Vec::new())); let gen = bridge.register(session, capturing_channel(Arc::clone(&sink))); pump_turn( &bridge, &session, gen, vec![ReplyEvent::Final { content: "bye".into(), }], ); assert_eq!(bridge.active_sessions(), 1); assert!(!bridge.scrollback(&session).is_empty()); bridge.unregister(&session); assert_eq!(bridge.active_sessions(), 0, "session removed"); assert!( bridge.scrollback(&session).is_empty(), "scrollback purged on close" ); assert!( !bridge.send_output(&session, delta("z")), "no delivery after close" ); // The closed channel must not receive the post-close chunk. assert_eq!(sink.lock().unwrap().as_slice(), &[final_chunk("bye")]); } // --------------------------------------------------------------------------- // Registry basics, mirrored from pty_bridge.rs for parity // --------------------------------------------------------------------------- #[test] fn register_returns_monotonic_generation_per_session() { let bridge = ChatBridge::new(); let session = sid(); let g0 = bridge.register(session, capturing_channel(Arc::new(Mutex::new(Vec::new())))); let g1 = bridge.register(session, capturing_channel(Arc::new(Mutex::new(Vec::new())))); assert_eq!(g0, 0); assert_eq!(g1, 1); assert_eq!(bridge.active_sessions(), 1, "same id replaced, not added"); } #[test] fn send_output_to_unknown_session_returns_false() { let bridge = ChatBridge::new(); assert!(!bridge.send_output(&sid(), delta("x"))); } #[test] fn register_same_session_replaces_channel() { let bridge = ChatBridge::new(); let session = sid(); let first = Arc::new(Mutex::new(Vec::new())); let second = Arc::new(Mutex::new(Vec::new())); bridge.register(session, capturing_channel(Arc::clone(&first))); bridge.register(session, capturing_channel(Arc::clone(&second))); bridge.send_output(&session, delta("9")); assert!(first.lock().unwrap().is_empty(), "old channel unused"); assert_eq!(second.lock().unwrap().as_slice(), &[delta("9")]); }