feat(agent): commandes Tauri + ChatBridge streaming (D4) — §17

Expose l'exécution structurée à l'UI (jumeau du chemin PTY) :
- ChatBridge (chat.rs) : pompe ReplyStream -> Channel<ReplyChunk>,
  generation-tracked (anti-double-pompe) ; scrollback de conversation
  côté transport (le port AgentSession est sans mémoire), purgé à la
  fermeture, préservé à la ré-attache.
- Commandes agent_send / reattach_agent_chat / close_agent_session.
- DTO : ReplyChunk (textDelta/toolActivity/final, camelCase tagué kind),
  ReattachChatDto, CellKind {pty,chat} + champ cellKind dérivé sur
  TerminalSessionDto (chat ssi LaunchAgentOutput.structured = Some).
- Wiring composition root : StructuredSessions + ChatBridge dans AppState,
  via la factory déjà injectée (aucun new d'adapter — règle D §17.8).
- From<AgentSessionError> for AppError (mappe sur PROCESS).

Tests (QA) : 35 unitaires verts — generation supersede validé par mutation
test, séquence deltas*+Final, reattach sans re-spawn, teardown shutdown+
unregister, DTO cellKind + round-trip ReplyChunk, non-régression PTY.
cargo test --workspace : 803 passed, 0 failed.

Reste D5 : AgentChatView (frontend) + routage cellKind dans LayoutGrid.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-06-09 23:15:08 +02:00
parent 050afa7d24
commit f4d5727a69
12 changed files with 1153 additions and 9 deletions

View File

@ -0,0 +1,364 @@
//! 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<ReplyChunk>` 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<Mutex<Vec<ReplyChunk>>>) -> Channel<ReplyChunk> {
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<ReplyEvent>) {
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")]);
}

View File

@ -0,0 +1,211 @@
//! L1 tests for the D4 structured-chat DTO contract (ARCHITECTURE §17.7):
//! - `ReplyChunk` tagged camelCase round-trip (`kind` + payload),
//! - `ReattachChatDto` camelCase wire shape (typed scrollback),
//! - the **derived** `cellKind` on `TerminalSessionDto` (`chat` ⇔
//! `structured: Some(..)`, `pty` otherwise),
//! - non-regression: every `TerminalSessionDto` construction path now serialises a
//! `cellKind` and PTY paths keep `"pty"`.
use app_tauri_lib::dto::{CellKind, ReattachChatDto, ReplyChunk, TerminalSessionDto};
use application::{LaunchAgentOutput, StructuredSessionDescriptor};
use domain::{AgentId, NodeId, PtySize, SessionKind, SessionStatus, TerminalSession};
use domain::project::ProjectPath;
use domain::SessionId;
use serde_json::json;
use uuid::Uuid;
// ---------------------------------------------------------------------------
// ReplyChunk — tagged camelCase, exact wire shape + round-trip (zone 5/6)
// ---------------------------------------------------------------------------
#[test]
fn reply_chunk_text_delta_serialises_exact_camel_case() {
let v = serde_json::to_value(ReplyChunk::TextDelta {
text: "hello".into(),
})
.unwrap();
assert_eq!(v, json!({ "kind": "textDelta", "text": "hello" }));
}
#[test]
fn reply_chunk_tool_activity_serialises_exact_camel_case() {
let v = serde_json::to_value(ReplyChunk::ToolActivity {
label: "reads file".into(),
})
.unwrap();
assert_eq!(v, json!({ "kind": "toolActivity", "label": "reads file" }));
}
#[test]
fn reply_chunk_final_serialises_exact_camel_case() {
let v = serde_json::to_value(ReplyChunk::Final {
content: "done".into(),
})
.unwrap();
assert_eq!(v, json!({ "kind": "final", "content": "done" }));
}
#[test]
fn reply_chunk_round_trips_through_json_for_every_variant() {
for chunk in [
ReplyChunk::TextDelta { text: "x".into() },
ReplyChunk::ToolActivity { label: "runs".into() },
ReplyChunk::Final { content: "y".into() },
] {
let v = serde_json::to_value(&chunk).unwrap();
let back: ReplyChunk = serde_json::from_value(v).unwrap();
assert_eq!(back, chunk, "round-trip preserves the variant + payload");
}
}
#[test]
fn reply_chunk_deserialises_from_camel_case_wire_payload() {
// The shape the frontend (or a mock gateway) emits.
let back: ReplyChunk =
serde_json::from_value(json!({ "kind": "textDelta", "text": "hi" })).unwrap();
assert_eq!(back, ReplyChunk::TextDelta { text: "hi".into() });
}
#[test]
fn reply_chunk_rejects_snake_case_tag() {
// Guard: a snake_case `text_delta` is NOT a valid wire tag (contract is camelCase).
let r: Result<ReplyChunk, _> =
serde_json::from_value(json!({ "kind": "text_delta", "text": "x" }));
assert!(r.is_err(), "snake_case kind must not deserialise");
}
// ---------------------------------------------------------------------------
// ReattachChatDto — typed scrollback, camelCase (zone 5)
// ---------------------------------------------------------------------------
#[test]
fn reattach_chat_dto_serialises_camel_case_with_typed_scrollback() {
let dto = ReattachChatDto {
session_id: "sess-1".into(),
scrollback: vec![
ReplyChunk::TextDelta { text: "Hi".into() },
ReplyChunk::Final { content: "Hi".into() },
],
};
let v = serde_json::to_value(&dto).unwrap();
assert_eq!(
v,
json!({
"sessionId": "sess-1",
"scrollback": [
{ "kind": "textDelta", "text": "Hi" },
{ "kind": "final", "content": "Hi" },
],
})
);
assert!(v.get("session_id").is_none(), "no snake_case leak");
}
#[test]
fn reattach_chat_dto_empty_scrollback_is_empty_array() {
let dto = ReattachChatDto {
session_id: "s".into(),
scrollback: vec![],
};
let v = serde_json::to_value(&dto).unwrap();
assert_eq!(v["scrollback"], json!([]));
}
// ---------------------------------------------------------------------------
// CellKind enum wire shape
// ---------------------------------------------------------------------------
#[test]
fn cell_kind_serialises_lowercase_pty_and_chat() {
assert_eq!(serde_json::to_value(CellKind::Pty).unwrap(), json!("pty"));
assert_eq!(serde_json::to_value(CellKind::Chat).unwrap(), json!("chat"));
}
// ---------------------------------------------------------------------------
// cellKind derivation on TerminalSessionDto (zone 5)
// ---------------------------------------------------------------------------
fn agent_session(session_id: u128) -> (SessionId, TerminalSession) {
let sid = SessionId::from_uuid(Uuid::from_u128(session_id));
let node_id = NodeId::from_uuid(Uuid::from_u128(8));
let agent_id = AgentId::from_uuid(Uuid::from_u128(9));
let cwd = ProjectPath::new("/tmp/project".to_owned()).expect("valid path");
let size = PtySize::new(24, 80).unwrap();
let mut session =
TerminalSession::starting(sid, node_id, cwd, SessionKind::Agent { agent_id }, size);
session.status = SessionStatus::Running;
(sid, session)
}
#[test]
fn launch_output_with_structured_descriptor_derives_chat_cell_kind() {
let (sid, session) = agent_session(7);
let descriptor = StructuredSessionDescriptor {
session_id: sid,
agent_id: AgentId::from_uuid(Uuid::from_u128(9)),
node_id: NodeId::from_uuid(Uuid::from_u128(8)),
conversation_id: None,
};
let out = LaunchAgentOutput {
session,
assigned_conversation_id: None,
structured: Some(descriptor),
};
let dto = TerminalSessionDto::from(out);
assert_eq!(dto.cell_kind, CellKind::Chat);
let v = serde_json::to_value(&dto).unwrap();
assert_eq!(v["cellKind"], "chat", "structured ⇒ chat on the wire");
}
#[test]
fn launch_output_without_structured_descriptor_derives_pty_cell_kind() {
let (_sid, session) = agent_session(7);
let out = LaunchAgentOutput {
session,
assigned_conversation_id: None,
structured: None,
};
let dto = TerminalSessionDto::from(out);
assert_eq!(dto.cell_kind, CellKind::Pty, "no descriptor ⇒ pty");
let v = serde_json::to_value(&dto).unwrap();
assert_eq!(v["cellKind"], "pty");
}
// ---------------------------------------------------------------------------
// Non-regression: cellKind is always present & "pty" on the historical paths
// ---------------------------------------------------------------------------
#[test]
fn terminal_session_dto_from_domain_session_is_always_pty() {
// From<TerminalSession> (e.g. open_terminal / change_agent_profile relaunch).
let (_sid, session) = agent_session(11);
let dto = TerminalSessionDto::from(session);
assert_eq!(dto.cell_kind, CellKind::Pty);
let v = serde_json::to_value(&dto).unwrap();
assert_eq!(
v["cellKind"], "pty",
"the new field is always present on the PTY path (non-breaking shape)"
);
}
#[test]
fn pty_launch_output_serialises_cellkind_pty_without_breaking_existing_keys() {
// Guard the exact historical key set + the new derived field for a PTY launch.
let (sid, session) = agent_session(12);
let out = LaunchAgentOutput {
session,
assigned_conversation_id: None,
structured: None,
};
let v = serde_json::to_value(TerminalSessionDto::from(out)).unwrap();
assert_eq!(v["sessionId"], sid.to_string());
assert_eq!(v["cwd"], "/tmp/project");
assert_eq!(v["rows"], 24);
assert_eq!(v["cols"], 80);
assert_eq!(v["cellKind"], "pty");
// assignedConversationId omitted when None (skip_serializing_if) — unchanged.
assert!(v.get("assignedConversationId").is_none());
assert!(v.get("session_id").is_none(), "no snake_case leak");
}