@ondeinference / ed / commits / 97a85db

ed-agent-tauri: multi-session state, webview approvals, text deltas

EdSessions keeps one Ed over many stored conversations, with a cache and a generation fence so a reply that finishes after logout is not saved. TauriApprovals lets the webview answer approval requests. Adds the chat_text_delta event and the session, submit and approval commands. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-Authored-By: siGit Code v1.6.3 <noreply@sigit.si>

Seto Elkahfi committed Oct 9, 2026 at 00:12 UTC 97a85db674583a2ada920af72499e448ad7e52d7
11 files changed +867 -29
Cargo.lock
+4
@@ -1759,10 +1759,14 @@ dependencies = [
1759 name = "ed-agent-tauri"
1760 version = "1.0.1"
1761 dependencies = [
1762 + "async-trait",
1763 "ed-agent",
1764 "log",
1765 "serde",
1766 + "serde_json",
1767 "tauri",
1768 + "tokio",
1769 + "uuid 1.27.0",
1770 ]
1771
1772 [[package]]
Cargo.toml
+1
@@ -30,6 +30,7 @@ tauri = { version = "2", features = [] }
30 # versions break extraction with "Invalid string data" errors.
31 uniffi = { version = "=0.31.0", features = ["tokio"] }
32 smbcloud-gresiq-sdk = "0.4.7"
33 +uuid = { version = "1", features = ["v4"] }
34
35 # Internal
36 ed-agent = { version = "1.0.1", path = "crates/ed-agent" }
crates/ed-agent-tauri/Cargo.toml
+7
@@ -16,3 +16,10 @@ ed-agent.workspace = true
16 tauri.workspace = true
17 serde.workspace = true
18 log.workspace = true
19 +async-trait.workspace = true
20 +serde_json.workspace = true
21 +uuid.workspace = true
22 +tokio = { workspace = true, features = ["time"] }
23 +
24 +[dev-dependencies]
25 +tokio = { workspace = true, features = ["time", "macros", "rt"] }
crates/ed-agent-tauri/README.md
+16 -6
@@ -23,16 +23,26 @@ These events reach the webview: `chat_status_changed`, `chat_reply`,
23 `chat_tool_requested`, `chat_approval_requested`, `chat_tool_started`,
24 `chat_tool_finished`, `chat_agent_reply`, and `chat_warning`.
25
26 +## Several conversations
27 +
28 +`EdSessions` is the multi-session form. Give it a `SessionStore` and it handles
29 +switching, the in-memory cache, logout (`reset`) and approvals. Register
30 +`chat_list_sessions`, `chat_new_session`, `chat_switch_session`,
31 +`chat_delete_session`, `chat_get_session_history`, `chat_submit` and
32 +`chat_respond_approval`.
33 +
34 +`chat_submit` returns immediately; the answer is a `chat_reply` event
35 +`{id, session, reply, duration, error}`, preceded by `chat_text_delta
36 +{session, id, delta}` events. Approvals use `chat_approval_requested
37 +{request_id, session, call, risk}` and `chat_approval_resolved {request_id}`;
38 +answer with `chat_respond_approval`. To rewrite the message first, wrap
39 +`EdSessions::submit_with` in your own command.
40 +
41 Loading a model is deliberately not a command here — which model, and from
42 where, is an application decision. Call `Ed::load` or `Ed::load_with` from your
43 own command.
44
30 -Tool calling needs both halves from you. Build the agent with `Ed::with_agent`
31 -to pass a `ToolExecutor` and an `ApprovalHandler`; `Ed::with_sink` rejects every
32 -tool and denies every approval. Approvals are not a command either: answering
33 -one means a webview round trip inside an awaited Rust call, and hosts differ
34 -enough about how to wire that up that this crate emits
35 -`chat_approval_requested` and leaves the answering to you.
45 +Without `EdSessions`, tool calling needs both halves from you: build the agent with `Ed::with_agent` to pass a `ToolExecutor` and an `ApprovalHandler`.
46
47 ## Copyright
48
crates/ed-agent-tauri/src/approvals.rs new
+208
@@ -0,0 +1,208 @@
1 +//! Approvals answered by the webview.
2 +
3 +use std::collections::HashMap;
4 +use std::time::Duration;
5 +
6 +use async_trait::async_trait;
7 +use ed_agent::{ApprovalDecision, ApprovalHandler, ApprovalRequest};
8 +use log::{error, warn};
9 +use serde::Serialize;
10 +use tauri::{AppHandle, Emitter};
11 +use tokio::sync::{oneshot, Mutex};
12 +
13 +use crate::events::{
14 + ApprovalRequestedPayload, ApprovalResolvedPayload, EVENT_CHAT_APPROVAL_REQUESTED,
15 + EVENT_CHAT_APPROVAL_RESOLVED,
16 +};
17 +use crate::scope::TurnScope;
18 +
19 +/// How long to wait for the user before denying.
20 +pub const DEFAULT_APPROVAL_TIMEOUT: Duration = Duration::from_secs(120);
21 +
22 +/// An [`ApprovalHandler`] that asks the webview.
23 +///
24 +/// For each request it emits `chat_approval_requested`, waits for the webview
25 +/// to call `chat_respond_approval` (or [`respond`](TauriApprovals::respond)),
26 +/// and emits `chat_approval_resolved` when the wait ends, however it ends.
27 +/// A timeout, a closed window or [`cancel_all`](TauriApprovals::cancel_all)
28 +/// all mean *deny*.
29 +type Emit = Box<dyn Fn(&str, serde_json::Value) -> Result<(), String> + Send + Sync>;
30 +
31 +pub struct TauriApprovals {
32 + emit: Emit,
33 + scope: TurnScope,
34 + timeout: Duration,
35 + pending: Mutex<HashMap<String, oneshot::Sender<ApprovalDecision>>>,
36 +}
37 +
38 +impl TauriApprovals {
39 + pub fn new(app: AppHandle, timeout: Duration) -> Self {
40 + Self::scoped(app, TurnScope::default(), timeout)
41 + }
42 +
43 + pub(crate) fn scoped(app: AppHandle, scope: TurnScope, timeout: Duration) -> Self {
44 + let emit: Emit =
45 + Box::new(move |event, payload| app.emit(event, payload).map_err(|err| err.to_string()));
46 + Self::with_emitter(emit, scope, timeout)
47 + }
48 +
49 + fn with_emitter(emit: Emit, scope: TurnScope, timeout: Duration) -> Self {
50 + Self {
51 + emit,
52 + scope,
53 + timeout,
54 + pending: Mutex::new(HashMap::new()),
55 + }
56 + }
57 +
58 + fn emit(&self, event: &str, payload: impl Serialize) -> Result<(), String> {
59 + let value = serde_json::to_value(payload).map_err(|err| err.to_string())?;
60 + (self.emit)(event, value)
61 + }
62 +
63 + /// Answer a pending request. Errors if it already timed out or was
64 + /// answered.
65 + pub async fn respond(
66 + &self,
67 + request_id: &str,
68 + decision: ApprovalDecision,
69 + ) -> Result<(), String> {
70 + let gone = || "This approval request is no longer pending.".to_string();
71 + let sender = self
72 + .pending
73 + .lock()
74 + .await
75 + .remove(request_id)
76 + .ok_or_else(gone)?;
77 + sender.send(decision).map_err(|_| gone())
78 + }
79 +
80 + /// Deny everything that is waiting. Use it when the conversation the
81 + /// requests belong to goes away.
82 + pub async fn cancel_all(&self) {
83 + let pending: Vec<_> = self.pending.lock().await.drain().collect();
84 + for (_, sender) in pending {
85 + // Each waiting `approve` call emits its own resolved event once it
86 + // sees the denial, so every request closes exactly once.
87 + let _ = sender.send(ApprovalDecision::Deny);
88 + }
89 + }
90 +}
91 +
92 +#[async_trait]
93 +impl ApprovalHandler for TauriApprovals {
94 + async fn approve(&self, request: ApprovalRequest) -> ApprovalDecision {
95 + let request_id = uuid::Uuid::new_v4().to_string();
96 + let (sender, receiver) = oneshot::channel();
97 + self.pending.lock().await.insert(request_id.clone(), sender);
98 +
99 + let (session, _) = self.scope.get();
100 + let payload = ApprovalRequestedPayload {
101 + request_id: request_id.clone(),
102 + session,
103 + call: request.call,
104 + risk: request.risk,
105 + };
106 + if let Err(err) = self.emit(EVENT_CHAT_APPROVAL_REQUESTED, payload) {
107 + self.pending.lock().await.remove(&request_id);
108 + error!("ed-tauri: could not request approval: {err}");
109 + return ApprovalDecision::Deny;
110 + }
111 +
112 + let decision = match tokio::time::timeout(self.timeout, receiver).await {
113 + Ok(Ok(decision)) => decision,
114 + Ok(Err(_)) => ApprovalDecision::Deny,
115 + Err(_) => {
116 + warn!("ed-tauri: approval {request_id} timed out; denying");
117 + ApprovalDecision::Deny
118 + }
119 + };
120 + self.pending.lock().await.remove(&request_id);
121 + if let Err(err) = self.emit(
122 + EVENT_CHAT_APPROVAL_RESOLVED,
123 + ApprovalResolvedPayload { request_id },
124 + ) {
125 + error!("ed-tauri: could not close approval: {err}");
126 + }
127 + decision
128 + }
129 +}
130 +
131 +#[cfg(test)]
132 +mod tests {
133 + use std::sync::Arc;
134 +
135 + use ed_agent::{ToolCall, ToolRisk};
136 +
137 + use super::*;
138 +
139 + fn request() -> ApprovalRequest {
140 + ApprovalRequest {
141 + call: ToolCall {
142 + id: "1".into(),
143 + name: "send_message".into(),
144 + arguments: "{}".into(),
145 + },
146 + risk: ToolRisk::Mutating,
147 + }
148 + }
149 +
150 + fn approvals(timeout: Duration) -> Arc<TauriApprovals> {
151 + Arc::new(TauriApprovals::with_emitter(
152 + Box::new(|_, _| Ok(())),
153 + TurnScope::default(),
154 + timeout,
155 + ))
156 + }
157 +
158 + async fn pending_id(approvals: &TauriApprovals) -> String {
159 + loop {
160 + if let Some(id) = approvals.pending.lock().await.keys().next().cloned() {
161 + return id;
162 + }
163 + tokio::task::yield_now().await;
164 + }
165 + }
166 +
167 + #[tokio::test]
168 + async fn the_webview_can_allow() {
169 + let approvals = approvals(Duration::from_secs(5));
170 + let waiting = tokio::spawn({
171 + let approvals = approvals.clone();
172 + async move { approvals.approve(request()).await }
173 + });
174 + let id = pending_id(&approvals).await;
175 +
176 + approvals
177 + .respond(&id, ApprovalDecision::AllowForSession)
178 + .await
179 + .unwrap();
180 +
181 + assert_eq!(waiting.await.unwrap(), ApprovalDecision::AllowForSession);
182 + assert!(approvals
183 + .respond(&id, ApprovalDecision::Deny)
184 + .await
185 + .is_err());
186 + }
187 +
188 + #[tokio::test]
189 + async fn cancel_all_denies_everything_pending() {
190 + let approvals = approvals(Duration::from_secs(5));
191 + let waiting = tokio::spawn({
192 + let approvals = approvals.clone();
193 + async move { approvals.approve(request()).await }
194 + });
195 + pending_id(&approvals).await;
196 +
197 + approvals.cancel_all().await;
198 +
199 + assert_eq!(waiting.await.unwrap(), ApprovalDecision::Deny);
200 + }
201 +
202 + #[tokio::test]
203 + async fn an_unanswered_request_times_out_as_a_denial() {
204 + let approvals = approvals(Duration::from_millis(20));
205 + assert_eq!(approvals.approve(request()).await, ApprovalDecision::Deny);
206 + assert!(approvals.pending.lock().await.is_empty());
207 + }
208 +}
crates/ed-agent-tauri/src/commands.rs
+105 -3
@@ -1,8 +1,11 @@
1 //! The commands a webview invokes.
2
3 -use ed_agent::{AgentReply, ChatMessage, ChatReply, Ed, EngineInfo};
3 +use std::sync::Arc;
4 +
5 +use ed_agent::{AgentReply, ApprovalDecision, ChatMessage, ChatReply, Ed, EngineInfo, SessionMeta};
6 use tauri::State;
7
8 +use crate::sessions::EdSessions;
9 use crate::sink::TauriSink;
10
11 /// Managed state holding the agent.
@@ -12,12 +15,12 @@ use crate::sink::TauriSink;
15 /// runs more than one window against separate engines, needs to control the
16 /// lifetime, and a `static` takes that away.
17 pub struct EdState {
15 - ed: Ed<TauriSink>,
18 + ed: Arc<Ed<TauriSink>>,
19 }
20
21 impl EdState {
22 pub fn new(ed: Ed<TauriSink>) -> Self {
20 - Self { ed }
23 + Self { ed: Arc::new(ed) }
24 }
25
26 /// The agent, for commands the host defines itself — loading a model,
@@ -26,6 +29,44 @@ impl EdState {
29 pub fn ed(&self) -> &Ed<TauriSink> {
30 &self.ed
31 }
32 +
33 + /// A shared handle to the agent, for work moved onto a spawned task.
34 + pub fn shared(&self) -> Arc<Ed<TauriSink>> {
35 + self.ed.clone()
36 + }
37 +
38 + /// Run an agent turn on a spawned task and deliver the result as a
39 + /// `chat_reply` event `{id, session, reply, duration, error}`, with
40 + /// `session` empty. Returns immediately.
41 + ///
42 + /// For single-conversation apps. A long-pending invoke can be dropped by a
43 + /// mobile webview, so the answer travels by event. Text deltas arrive as
44 + /// `chat_text_delta` while it runs.
45 + pub fn submit(&self, app: tauri::AppHandle, id: String, message: String) {
46 + use tauri::Emitter;
47 + let ed = self.ed.clone();
48 + tokio::spawn(async move {
49 + let payload = match ed.run(message).await {
50 + Ok(reply) => crate::SubmitReplyPayload {
51 + id,
52 + session: String::new(),
53 + reply: Some(reply.text),
54 + duration: Some(reply.duration),
55 + error: None,
56 + },
57 + Err(err) => crate::SubmitReplyPayload {
58 + id,
59 + session: String::new(),
60 + reply: None,
61 + duration: None,
62 + error: Some(err.to_string()),
63 + },
64 + };
65 + if let Err(err) = app.emit(crate::EVENT_CHAT_REPLY, payload) {
66 + log::warn!("ed-tauri: could not emit chat_reply: {err}");
67 + }
68 + });
69 + }
70 }
71
72 /// Current status, loaded model, footprint, and history length.
@@ -88,3 +129,64 @@ pub async fn chat_cancel(state: State<'_, EdState>) -> Result<(), String> {
129 state.ed.cancel();
130 Ok(())
131 }
132 +
133 +/// Stored sessions, most recently updated first.
134 +#[tauri::command]
135 +pub async fn chat_list_sessions(state: State<'_, EdSessions>) -> Result<Vec<SessionMeta>, String> {
136 + state.list().await.map_err(|e| e.to_string())
137 +}
138 +
139 +/// Start an empty session and make it live. Returns its id.
140 +#[tauri::command]
141 +pub async fn chat_new_session(state: State<'_, EdSessions>) -> Result<String, String> {
142 + Ok(state.new_session().await)
143 +}
144 +
145 +/// Make `session` live and return its messages.
146 +#[tauri::command]
147 +pub async fn chat_switch_session(
148 + state: State<'_, EdSessions>,
149 + session: String,
150 +) -> Result<Vec<ChatMessage>, String> {
151 + Ok(state.history(&session).await)
152 +}
153 +
154 +#[tauri::command]
155 +pub async fn chat_delete_session(
156 + state: State<'_, EdSessions>,
157 + session: String,
158 +) -> Result<(), String> {
159 + state.delete(&session).await.map_err(|e| e.to_string())
160 +}
161 +
162 +/// The messages of `session`, without switching to it.
163 +#[tauri::command]
164 +pub async fn chat_get_session_history(
165 + state: State<'_, EdSessions>,
166 + session: String,
167 +) -> Result<Vec<ChatMessage>, String> {
168 + Ok(state.peek(&session).await)
169 +}
170 +
171 +/// Run a turn in `session`. Returns at once; the result is a `chat_reply`
172 +/// event carrying `id`.
173 +#[tauri::command]
174 +pub async fn chat_submit(
175 + state: State<'_, EdSessions>,
176 + session: String,
177 + id: String,
178 + message: String,
179 +) -> Result<(), String> {
180 + state.submit(&session, &id, message).await;
181 + Ok(())
182 +}
183 +
184 +/// Answer a `chat_approval_requested` event.
185 +#[tauri::command]
186 +pub async fn chat_respond_approval(
187 + state: State<'_, EdSessions>,
188 + request_id: String,
189 + decision: ApprovalDecision,
190 +) -> Result<(), String> {
191 + state.respond_approval(&request_id, decision).await
192 +}
crates/ed-agent-tauri/src/events.rs
+50
@@ -45,6 +45,56 @@ pub const EVENT_CHAT_AGENT_REPLY: &str = "chat_agent_reply";
45 /// is still viable, such as an unverified model or a spent tool-round budget.
46 pub const EVENT_CHAT_WARNING: &str = "chat_warning";
47
48 +/// Emitted with a [`TextDeltaPayload`] for each piece of a reply as it is
49 +/// generated. Concatenate the deltas of one `id` to grow the assistant bubble;
50 +/// [`EVENT_CHAT_REPLY`] still arrives at the end with the whole text.
51 +pub const EVENT_CHAT_TEXT_DELTA: &str = "chat_text_delta";
52 +
53 +/// Emitted with an [`ApprovalResolvedPayload`] once an approval request has
54 +/// been answered, timed out, or cancelled, so the webview can close its sheet.
55 +pub const EVENT_CHAT_APPROVAL_RESOLVED: &str = "chat_approval_resolved";
56 +
57 +/// Payload for [`EVENT_CHAT_TEXT_DELTA`].
58 +///
59 +/// `session` and `id` are empty when the turn didn't come through a `submit`
60 +/// call, which is the only thing that knows them.
61 +#[derive(Debug, Clone, Serialize, Deserialize)]
62 +pub struct TextDeltaPayload {
63 + pub session: String,
64 + pub id: String,
65 + pub delta: String,
66 +}
67 +
68 +/// Payload for [`EVENT_CHAT_REPLY`] when the turn came through `submit`.
69 +///
70 +/// A superset of [`ChatReply`](ed_agent::ChatReply), so a listener written for
71 +/// the single-session event keeps working.
72 +#[derive(Debug, Clone, Serialize, Deserialize)]
73 +pub struct SubmitReplyPayload {
74 + pub id: String,
75 + pub session: String,
76 + pub reply: Option<String>,
77 + pub duration: Option<String>,
78 + pub error: Option<String>,
79 +}
80 +
81 +/// Payload for [`EVENT_CHAT_APPROVAL_REQUESTED`] when the approval handler is
82 +/// [`TauriApprovals`](crate::TauriApprovals). Answer it with
83 +/// `chat_respond_approval { request_id, decision }`.
84 +#[derive(Debug, Clone, Serialize, Deserialize)]
85 +pub struct ApprovalRequestedPayload {
86 + pub request_id: String,
87 + pub session: String,
88 + pub call: ed_agent::ToolCall,
89 + pub risk: ed_agent::ToolRisk,
90 +}
91 +
92 +/// Payload for [`EVENT_CHAT_APPROVAL_RESOLVED`].
93 +#[derive(Debug, Clone, Serialize, Deserialize)]
94 +pub struct ApprovalResolvedPayload {
95 + pub request_id: String,
96 +}
97 +
98 /// Payload for [`EVENT_CHAT_STATUS_CHANGED`].
99 ///
100 /// `status` serialises to `"unloaded"`, `"loading"`, `"ready"`, `"generating"`,
crates/ed-agent-tauri/src/lib.rs
+46 -16
@@ -43,21 +43,41 @@
43 //! where, is an application decision. Call [`Ed::load`](ed_agent::Ed::load)
44 //! from your own command or during setup.
45 //!
46 +//! # Several conversations
47 +//!
48 +//! For an app with a session list, build an [`EdSessions`] instead of an
49 +//! [`EdState`]. You supply a [`SessionStore`](ed_agent::SessionStore); it owns
50 +//! the live session, the cache, the logout fence and approvals:
51 +//!
52 +//! ```text
53 +//! let sessions = EdSessions::builder(app.handle().clone(), MyStore::new())
54 +//! .executor(my_executor)
55 +//! .build();
56 +//! app.manage(sessions);
57 +//! ```
58 +//!
59 +//! Register `chat_list_sessions`, `chat_new_session`, `chat_switch_session`,
60 +//! `chat_delete_session`, `chat_get_session_history`, `chat_submit` and
61 +//! `chat_respond_approval`. `chat_submit` returns at once and the answer comes
62 +//! as a `chat_reply` event `{id, session, reply, duration, error}`, with
63 +//! `chat_text_delta {session, id, delta}` events while it generates.
64 +//!
65 +//! A host that has to rewrite the message first (to add context) or recover
66 +//! from an engine failure wraps [`EdSessions::submit_with`] in a command of its
67 +//! own. Call [`EdSessions::reset`] on logout or account switch.
68 +//!
69 //! # Tools
70 //!
48 -//! Tool calling works through this crate, but the host owns both halves of it.
49 -//! Build the agent with [`Ed::with_agent`](ed_agent::Ed::with_agent) to supply
50 -//! a [`ToolExecutor`](ed_agent::ToolExecutor) and an
51 -//! [`ApprovalHandler`](ed_agent::ApprovalHandler); the default `Ed::with_sink`
52 -//! rejects every tool and denies every approval, which is the right default but
53 -//! not a working agent.
71 +//! Tool calling works through this crate, but the host supplies the executor.
72 +//! With [`EdSessions`] approvals are handled for you: [`TauriApprovals`] emits
73 +//! `chat_approval_requested {request_id, session, call, risk}`, waits (120 s by
74 +//! default) for `chat_respond_approval {request_id, decision}`, then emits
75 +//! `chat_approval_resolved {request_id}`. A timeout or a reset denies.
76 //!
55 -//! Approvals are deliberately not a command. Answering one means a webview
56 -//! round trip in the middle of an awaited Rust call, and how a host wants to
57 -//! wire that up (a channel, a shared map, a modal it blocks on) differs enough
58 -//! that guessing here would be worse than leaving it. What this crate does give
59 -//! you is [`EVENT_CHAT_APPROVAL_REQUESTED`], so the sheet can go up at the
60 -//! right moment.
77 +//! With [`EdState`], build the agent with
78 +//! [`Ed::with_agent`](ed_agent::Ed::with_agent) and your own
79 +//! [`ApprovalHandler`](ed_agent::ApprovalHandler); the default `Ed::with_sink`
80 +//! rejects every tool and denies every approval.
81 //!
82 //! # Events
83 //!
@@ -74,20 +94,30 @@
94 //! [`ToolExecutionResult`](ed_agent::ToolExecutionResult)
95 //! - [`EVENT_CHAT_AGENT_REPLY`] with an [`AgentReply`](ed_agent::AgentReply)
96 //! - [`EVENT_CHAT_WARNING`] with a string
97 +//! - [`EVENT_CHAT_TEXT_DELTA`] with a [`TextDeltaPayload`]
98 +//! - [`EVENT_CHAT_APPROVAL_RESOLVED`] with an [`ApprovalResolvedPayload`]
99
100 #![forbid(unsafe_code)]
101
102 +mod approvals;
103 mod commands;
104 mod events;
105 +mod scope;
106 +mod sessions;
107 mod sink;
108
109 +pub use approvals::{TauriApprovals, DEFAULT_APPROVAL_TIMEOUT};
110 pub use commands::{
85 - chat_cancel, chat_clear_history, chat_get_history, chat_get_status, chat_run,
86 - chat_send_message, EdState,
111 + chat_cancel, chat_clear_history, chat_delete_session, chat_get_history,
112 + chat_get_session_history, chat_get_status, chat_list_sessions, chat_new_session,
113 + chat_respond_approval, chat_run, chat_send_message, chat_submit, chat_switch_session, EdState,
114 };
115 pub use events::{
89 - ChatStatusPayload, EVENT_CHAT_AGENT_REPLY, EVENT_CHAT_APPROVAL_REQUESTED, EVENT_CHAT_REPLY,
90 - EVENT_CHAT_STATUS_CHANGED, EVENT_CHAT_TOOL_FINISHED, EVENT_CHAT_TOOL_REQUESTED,
116 + ApprovalRequestedPayload, ApprovalResolvedPayload, ChatStatusPayload, SubmitReplyPayload,
117 + TextDeltaPayload, EVENT_CHAT_AGENT_REPLY, EVENT_CHAT_APPROVAL_REQUESTED,
118 + EVENT_CHAT_APPROVAL_RESOLVED, EVENT_CHAT_REPLY, EVENT_CHAT_STATUS_CHANGED,
119 + EVENT_CHAT_TEXT_DELTA, EVENT_CHAT_TOOL_FINISHED, EVENT_CHAT_TOOL_REQUESTED,
120 EVENT_CHAT_TOOL_STARTED, EVENT_CHAT_WARNING,
121 };
122 +pub use sessions::{EdSessions, EdSessionsBuilder};
123 pub use sink::TauriSink;
crates/ed-agent-tauri/src/scope.rs new
+31
@@ -0,0 +1,31 @@
1 +//! Which session and message a running turn belongs to.
2 +
3 +use std::sync::{Arc, Mutex};
4 +
5 +/// Shared between the sink and the approval handler so that events emitted
6 +/// from inside Ed (which knows nothing about sessions) can still say which
7 +/// session they belong to.
8 +///
9 +/// Ed runs one turn at a time, so one slot is enough.
10 +#[derive(Clone, Default)]
11 +pub(crate) struct TurnScope(Arc<Mutex<Option<(String, String)>>>);
12 +
13 +impl TurnScope {
14 + pub(crate) fn set(&self, session: &str, id: &str) {
15 + *self.0.lock().unwrap_or_else(|e| e.into_inner()) =
16 + Some((session.to_owned(), id.to_owned()));
17 + }
18 +
19 + pub(crate) fn clear(&self) {
20 + *self.0.lock().unwrap_or_else(|e| e.into_inner()) = None;
21 + }
22 +
23 + /// `(session, id)`, both empty outside a `submit`.
24 + pub(crate) fn get(&self) -> (String, String) {
25 + self.0
26 + .lock()
27 + .unwrap_or_else(|e| e.into_inner())
28 + .clone()
29 + .unwrap_or_default()
30 + }
31 +}
crates/ed-agent-tauri/src/sessions.rs new
+368
@@ -0,0 +1,368 @@
1 +//! Several conversations over one Ed.
2 +
3 +use std::collections::HashMap;
4 +use std::future::Future;
5 +use std::sync::atomic::{AtomicU64, Ordering};
6 +use std::sync::Arc;
7 +use std::time::Duration;
8 +
9 +use ed_agent::{
10 + AgentConfig, AgentError, AgentReply, ApprovalDecision, ChatMessage, Ed, RejectingExecutor,
11 + SessionError, SessionMeta, SessionStore, ToolExecutor,
12 +};
13 +use log::{error, info, warn};
14 +use tauri::{AppHandle, Emitter};
15 +use tokio::sync::Mutex;
16 +
17 +use crate::approvals::{TauriApprovals, DEFAULT_APPROVAL_TIMEOUT};
18 +use crate::events::{SubmitReplyPayload, EVENT_CHAT_REPLY};
19 +use crate::scope::TurnScope;
20 +use crate::sink::TauriSink;
21 +
22 +#[derive(Default)]
23 +struct State {
24 + active: Option<String>,
25 + /// Histories of sessions that were switched away from, so switching back
26 + /// doesn't hit the store.
27 + cache: HashMap<String, Vec<ChatMessage>>,
28 +}
29 +
30 +struct Shared {
31 + app: AppHandle,
32 + ed: Arc<Ed<TauriSink>>,
33 + store: Arc<dyn SessionStore>,
34 + approvals: Arc<TauriApprovals>,
35 + scope: TurnScope,
36 + /// Bumped by [`EdSessions::reset`]. A turn captures it at dispatch and
37 + /// refuses to save or emit if it changed by completion, so a reply that
38 + /// finishes after logout or an account switch can't land in the next
39 + /// account's storage.
40 + generation: AtomicU64,
41 + state: Mutex<State>,
42 + error_message: Option<String>,
43 + new_id: Box<dyn Fn() -> String + Send + Sync>,
44 +}
45 +
46 +/// Multi-session chat state: one [`Ed`], many conversations.
47 +///
48 +/// Register it with `app.manage(sessions)` and the `chat_*_session` commands.
49 +/// It owns which session is live in the engine, an in-memory cache, the
50 +/// generation fence, and approvals; persistence stays with the host's
51 +/// [`SessionStore`].
52 +///
53 +/// Cheap to clone.
54 +#[derive(Clone)]
55 +pub struct EdSessions {
56 + shared: Arc<Shared>,
57 +}
58 +
59 +/// Configuration for [`EdSessions`].
60 +pub struct EdSessionsBuilder {
61 + app: AppHandle,
62 + store: Arc<dyn SessionStore>,
63 + executor: Arc<dyn ToolExecutor>,
64 + config: AgentConfig,
65 + onde_app_id: Option<String>,
66 + approval_timeout: Duration,
67 + error_message: Option<String>,
68 + new_id: Box<dyn Fn() -> String + Send + Sync>,
69 +}
70 +
71 +impl EdSessionsBuilder {
72 + /// Run tools through `executor`. Defaults to rejecting every tool.
73 + pub fn executor(mut self, executor: Arc<dyn ToolExecutor>) -> Self {
74 + self.executor = executor;
75 + self
76 + }
77 +
78 + pub fn config(mut self, config: AgentConfig) -> Self {
79 + self.config = config;
80 + self
81 + }
82 +
83 + /// Associate telemetry with an Onde app id.
84 + pub fn onde_app_id(mut self, id: impl Into<String>) -> Self {
85 + self.onde_app_id = Some(id.into());
86 + self
87 + }
88 +
89 + /// How long an approval waits for the user. Default 120 seconds.
90 + pub fn approval_timeout(mut self, timeout: Duration) -> Self {
91 + self.approval_timeout = timeout;
92 + self
93 + }
94 +
95 + /// Show this to the user instead of the error text when a turn fails. The
96 + /// real error is still logged.
97 + pub fn error_message(mut self, message: impl Into<String>) -> Self {
98 + self.error_message = Some(message.into());
99 + self
100 + }
101 +
102 + /// How [`EdSessions::new_session`] names a session. Defaults to a random UUID. A store that
103 + /// treats some ids as ephemeral (by prefix, say) can mint matching ones here.
104 + pub fn id_generator(mut self, generator: impl Fn() -> String + Send + Sync + 'static) -> Self {
105 + self.new_id = Box::new(generator);
106 + self
107 + }
108 +
109 + pub fn build(self) -> EdSessions {
110 + let scope = TurnScope::default();
111 + let approvals = Arc::new(TauriApprovals::scoped(
112 + self.app.clone(),
113 + scope.clone(),
114 + self.approval_timeout,
115 + ));
116 + let sink = TauriSink::scoped(self.app.clone(), scope.clone(), false);
117 + let ed = Ed::with_agent_and_app_id(
118 + sink,
119 + self.executor,
120 + approvals.clone(),
121 + self.config,
122 + self.onde_app_id,
123 + );
124 + EdSessions {
125 + shared: Arc::new(Shared {
126 + app: self.app,
127 + ed: Arc::new(ed),
128 + store: self.store,
129 + approvals,
130 + scope,
131 + generation: AtomicU64::new(0),
132 + state: Mutex::new(State::default()),
133 + error_message: self.error_message,
134 + new_id: self.new_id,
135 + }),
136 + }
137 + }
138 +}
139 +
140 +impl EdSessions {
141 + pub fn builder(app: AppHandle, store: impl SessionStore) -> EdSessionsBuilder {
142 + EdSessionsBuilder {
143 + app,
144 + store: Arc::new(store),
145 + executor: Arc::new(RejectingExecutor),
146 + config: AgentConfig::default(),
147 + onde_app_id: None,
148 + approval_timeout: DEFAULT_APPROVAL_TIMEOUT,
149 + error_message: None,
150 + new_id: Box::new(|| uuid::Uuid::new_v4().to_string()),
151 + }
152 + }
153 +
154 + /// The agent, for loading a model and anything else Ed exposes.
155 + pub fn ed(&self) -> &Arc<Ed<TauriSink>> {
156 + &self.shared.ed
157 + }
158 +
159 + pub fn approvals(&self) -> &Arc<TauriApprovals> {
160 + &self.shared.approvals
161 + }
162 +
163 + /// The current generation. Capture it before slow work and compare with
164 + /// [`is_current`](EdSessions::is_current) afterwards.
165 + pub fn generation(&self) -> u64 {
166 + self.shared.generation.load(Ordering::SeqCst)
167 + }
168 +
169 + pub fn is_current(&self, generation: u64) -> bool {
170 + self.generation() == generation
171 + }
172 +
173 + /// The id of the session live in the engine, if any.
174 + pub async fn active(&self) -> Option<String> {
175 + self.shared.state.lock().await.active.clone()
176 + }
177 +
178 + /// Stored sessions, most recently updated first.
179 + pub async fn list(&self) -> Result<Vec<SessionMeta>, SessionError> {
180 + self.shared.store.list().await
181 + }
182 +
183 + /// Make `session` the live conversation.
184 + ///
185 + /// The outgoing session's history is kept in memory; the incoming one
186 + /// comes from memory, then the store. An unknown id starts empty.
187 + pub async fn activate(&self, session: &str) {
188 + let shared = &self.shared;
189 + let mut state = shared.state.lock().await;
190 + if state.active.as_deref() == Some(session) {
191 + return;
192 + }
193 + if let Some(previous) = state.active.take() {
194 + let live = shared.ed.history().await;
195 + state.cache.insert(previous, live);
196 + }
197 + let stored = match state.cache.remove(session) {
198 + Some(messages) => messages,
199 + None => match shared.store.load(session).await {
200 + Ok(messages) => messages.unwrap_or_default(),
201 + Err(err) => {
202 + error!("ed-tauri: could not load session {session}: {err}");
203 + Vec::new()
204 + }
205 + },
206 + };
207 + shared.ed.restore_history(stored).await;
208 + state.active = Some(session.to_owned());
209 + }
210 +
211 + /// Start an empty session and make it live. Returns its id. The store
212 + /// hears about it when the first turn completes.
213 + pub async fn new_session(&self) -> String {
214 + let id = (self.shared.new_id)();
215 + self.activate(&id).await;
216 + id
217 + }
218 +
219 + /// The messages of `session`, activating it.
220 + pub async fn history(&self, session: &str) -> Vec<ChatMessage> {
221 + self.activate(session).await;
222 + self.shared.ed.history().await
223 + }
224 +
225 + /// The messages of `session` without switching to it: the live history if it is the active
226 + /// session, else the cache, else the store. For rendering a session that is not on screen,
227 + /// without disturbing a turn that is running in another.
228 + pub async fn peek(&self, session: &str) -> Vec<ChatMessage> {
229 + let shared = &self.shared;
230 + let state = shared.state.lock().await;
231 + if state.active.as_deref() == Some(session) {
232 + return shared.ed.history().await;
233 + }
234 + if let Some(cached) = state.cache.get(session) {
235 + return cached.clone();
236 + }
237 + drop(state);
238 + match shared.store.load(session).await {
239 + Ok(messages) => messages.unwrap_or_default(),
240 + Err(err) => {
241 + error!("ed-tauri: could not load session {session}: {err}");
242 + Vec::new()
243 + }
244 + }
245 + }
246 +
247 + /// Delete a session from the store and memory. If it was live, the engine
248 + /// is cleared and no session is live until the next activate.
249 + pub async fn delete(&self, session: &str) -> Result<(), SessionError> {
250 + let shared = &self.shared;
251 + let result = shared.store.delete(session).await;
252 + let mut state = shared.state.lock().await;
253 + state.cache.remove(session);
254 + if state.active.as_deref() == Some(session) {
255 + state.active = None;
256 + shared.ed.clear_history().await;
257 + }
258 + result
259 + }
260 +
261 + /// Drop all in-memory state and fence off in-flight turns. For logout and
262 + /// account switches: the next account must never see this one's chats.
263 + /// Stored sessions are left alone.
264 + pub async fn reset(&self) {
265 + let shared = &self.shared;
266 + shared.generation.fetch_add(1, Ordering::SeqCst);
267 + shared.ed.cancel();
268 + shared.approvals.cancel_all().await;
269 + let mut state = shared.state.lock().await;
270 + state.cache.clear();
271 + state.active = None;
272 + shared.ed.clear_history().await;
273 + }
274 +
275 + /// Save the live history under `session`, unless `generation` is stale.
276 + pub async fn save(&self, session: &str, generation: u64) {
277 + let shared = &self.shared;
278 + let live = shared.ed.history().await;
279 + let mut state = shared.state.lock().await;
280 + if !self.is_current(generation) {
281 + return;
282 + }
283 + if let Err(err) = shared.store.save(session, &live).await {
284 + error!("ed-tauri: could not save session {session}: {err}");
285 + }
286 + state.cache.insert(session.to_owned(), live);
287 + }
288 +
289 + /// Run one agent turn in `session` and deliver the result as a
290 + /// `chat_reply` event.
291 + ///
292 + /// Returns as soon as the session is active and the turn is spawned. A
293 + /// long-pending invoke can be dropped by a mobile webview, so the answer
294 + /// travels by event: `{id, session, reply, duration, error}`.
295 + pub async fn submit(&self, session: &str, id: &str, message: impl Into<String>) {
296 + let message = message.into();
297 + self.submit_with(session, id, move |ed| async move { ed.run(message).await })
298 + .await;
299 + }
300 +
301 + /// Like [`submit`](EdSessions::submit), with the turn supplied by the
302 + /// host: rewrite the message, or recover and retry on a known engine
303 + /// failure. The closure runs on the spawned task.
304 + pub async fn submit_with<F, Fut>(&self, session: &str, id: &str, turn: F)
305 + where
306 + F: FnOnce(Arc<Ed<TauriSink>>) -> Fut + Send + 'static,
307 + Fut: Future<Output = Result<AgentReply, AgentError>> + Send + 'static,
308 + {
309 + self.activate(session).await;
310 + let generation = self.generation();
311 + let this = self.clone();
312 + let session = session.to_owned();
313 + let id = id.to_owned();
314 + info!("ed-tauri: dispatch session={session} id={id}");
315 +
316 + tokio::spawn(async move {
317 + this.shared.scope.set(&session, &id);
318 + let result = turn(this.shared.ed.clone()).await;
319 + this.shared.scope.clear();
320 +
321 + if !this.is_current(generation) {
322 + info!("ed-tauri: reply dropped, the account context changed (session={session})");
323 + return;
324 + }
325 + this.save(&session, generation).await;
326 + if !this.is_current(generation) {
327 + return;
328 + }
329 +
330 + let payload = match result {
331 + Ok(reply) => SubmitReplyPayload {
332 + id,
333 + session,
334 + reply: Some(reply.text),
335 + duration: Some(reply.duration),
336 + error: None,
337 + },
338 + Err(err) => {
339 + error!("ed-tauri: turn failed: {err}");
340 + SubmitReplyPayload {
341 + id,
342 + session,
343 + reply: None,
344 + duration: None,
345 + error: Some(
346 + this.shared
347 + .error_message
348 + .clone()
349 + .unwrap_or_else(|| err.to_string()),
350 + ),
351 + }
352 + }
353 + };
354 + if let Err(err) = this.shared.app.emit(EVENT_CHAT_REPLY, payload) {
355 + warn!("ed-tauri: could not emit chat_reply: {err}");
356 + }
357 + });
358 + }
359 +
360 + /// Answer a pending approval.
361 + pub async fn respond_approval(
362 + &self,
363 + request_id: &str,
364 + decision: ApprovalDecision,
365 + ) -> Result<(), String> {
366 + self.shared.approvals.respond(request_id, decision).await
367 + }
368 +}
crates/ed-agent-tauri/src/sink.rs
+31 -4
@@ -8,10 +8,11 @@ use serde::Serialize;
8 use tauri::{AppHandle, Emitter};
9
10 use crate::events::{
11 - ChatStatusPayload, EVENT_CHAT_AGENT_REPLY, EVENT_CHAT_APPROVAL_REQUESTED, EVENT_CHAT_REPLY,
12 - EVENT_CHAT_STATUS_CHANGED, EVENT_CHAT_TOOL_FINISHED, EVENT_CHAT_TOOL_REQUESTED,
13 - EVENT_CHAT_TOOL_STARTED, EVENT_CHAT_WARNING,
11 + ChatStatusPayload, TextDeltaPayload, EVENT_CHAT_AGENT_REPLY, EVENT_CHAT_APPROVAL_REQUESTED,
12 + EVENT_CHAT_REPLY, EVENT_CHAT_STATUS_CHANGED, EVENT_CHAT_TEXT_DELTA, EVENT_CHAT_TOOL_FINISHED,
13 + EVENT_CHAT_TOOL_REQUESTED, EVENT_CHAT_TOOL_STARTED, EVENT_CHAT_WARNING,
14 };
15 +use crate::scope::TurnScope;
16
17 /// An [`EventSink`] that emits to the webview.
18 ///
@@ -21,11 +22,21 @@ use crate::events::{
22 /// listening.
23 pub struct TauriSink {
24 app: AppHandle,
25 + scope: TurnScope,
26 + emit_approvals: bool,
27 }
28
29 impl TauriSink {
30 pub fn new(app: AppHandle) -> Self {
28 - Self { app }
31 + Self::scoped(app, TurnScope::default(), true)
32 + }
33 +
34 + pub(crate) fn scoped(app: AppHandle, scope: TurnScope, emit_approvals: bool) -> Self {
35 + Self {
36 + app,
37 + scope,
38 + emit_approvals,
39 + }
40 }
41
42 fn emit(&self, event: &str, payload: impl Serialize + Clone) {
@@ -55,7 +66,23 @@ impl EventSink for TauriSink {
66 self.emit(EVENT_CHAT_TOOL_REQUESTED, call);
67 }
68
69 + fn text_delta(&self, delta: &str) {
70 + let (session, id) = self.scope.get();
71 + self.emit(
72 + EVENT_CHAT_TEXT_DELTA,
73 + TextDeltaPayload {
74 + session,
75 + id,
76 + delta: delta.to_owned(),
77 + },
78 + );
79 + }
80 +
81 fn approval_requested(&self, request: &ApprovalRequest) {
82 + // `TauriApprovals` emits its own, richer event for the same name.
83 + if !self.emit_approvals {
84 + return;
85 + }
86 self.emit(EVENT_CHAT_APPROVAL_REQUESTED, request);
87 }
88