agents_kit/conversation.rs
1//! Conversation metadata, accepted prompts, exact file reads and durable interactions.
2//!
3//! Creation inputs are passive. Live operations use [`ConversationBackend`]; its
4//! consuming close permanently terminates the conversation. Prompt streams observe
5//! runtime-owned accepted work, so dropping an observer is not cancellation.
6//!
7//! # Streaming, durable approval and explicit cancellation
8//!
9//! This complete Custom journey requires a running Ollama server and the exact
10//! configured model with function-calling support. It is therefore `no_run`.
11//! Replace only the caller configuration to use your intended model; selection
12//! does not substitute a fallback. The caller supplies the executable tool.
13//! Caller-built actions run in process without Kit sandbox enforcement, so Custom
14//! requires `FullAccess`, `Enabled` and read-write roots for this configuration; roots do
15//! not confine the action. This example's `inspect_status` action performs no
16//! filesystem or network I/O.
17//!
18//! Notifications are an exclusive, durable, at-least-once delivery path. Inspect
19//! an Interaction before resolving its exact ID; advancing passive delivery is
20//! acknowledgment, and dropping the receiver is not acknowledgment. The example
21//! deliberately denies every requested action: a production UI can instead
22//! collect a caller decision and send `Allow` or schema-valid `Submit` content.
23//! Denial is a real interaction resolution, not a fabricated tool result.
24//!
25//! The output observer requires the authoritative `Completed` value. Deltas can
26//! include earlier assistant turns around tool execution, so concatenating them
27//! is not an alternative final answer. Observer loss does not abort accepted work;
28//! the separate cancellation portion uses `abort`, then observes settlement.
29//!
30//! ```no_run,standalone_crate
31//! #![recursion_limit = "256"] // Nested concrete async futures need this on Linux.
32//! use std::{num::NonZeroUsize, time::Duration};
33//! use agents_kit::{AgentBackend, AgentBuilder, AgentStorage, ConversationBackend, WorkspaceBackend};
34//! use agents_kit::config::{AgentConfig, CustomAgentConfig, CustomLoopLimits, CustomToolApproval};
35//! use agents_kit::conversation::NewConversation;
36//! use agents_kit::notification::{ConversationNotificationKind, NotificationResolution};
37//! use agents_kit::prompt::{PromptEvent, PromptRequest};
38//! use agents_kit::{model::ModelSelector, tool::ToolSelection};
39//! use agents_kit::workspace::{NewWorkspace, WorkspaceContext, WorkspaceDefaults,
40//! WorkspaceExecutionPolicy, WorkspaceNetworkAccess, WorkspaceRoot,
41//! WorkspaceRootAccess, WorkspaceSandboxMode};
42//! use providers_kit::{Provider, ProviderConfig,
43//! config::{OllamaConfig, ProviderKind, TransportConfig}};
44//! use storage_kit::{StateStore,
45//! state::config::{LanceStateStoreConfig, StateStoreConfig}};
46//! use tools_kit::{Toolset, declaration::{ToolInfo, ToolName}, execution::ToolResult,
47//! local::ToolRegistration, schema::JsonSchema};
48//! # let runtime = tokio::runtime::Builder::new_current_thread().enable_all().build()?;
49//! # runtime.block_on(async {
50//! let store = StateStore::connect(StateStoreConfig::Lance(LanceStateStoreConfig::memory())).await?;
51//! let tools = Toolset::local(vec![ToolRegistration::function(
52//! ToolInfo::new(ToolName::new("inspect_status")?,
53//! JsonSchema::strict(serde_json::json!({"type": "object", "additionalProperties": false}))),
54//! |_| async { Ok(ToolResult::json(serde_json::json!({"status": "ready"}))) },
55//! )])?;
56//! let custom = CustomAgentConfig::new(ModelSelector::new("exact-tool-capable-model")?,
57//! ToolSelection::Auto, CustomLoopLimits::new(4, 4, 1)?, CustomToolApproval::Interaction)?;
58//! let config = AgentConfig::new(custom.into(), Duration::from_secs(60), NonZeroUsize::MIN)?;
59//! let provider = Provider::from_config(ProviderConfig::new(
60//! ProviderKind::Ollama(OllamaConfig::local()), TransportConfig::default()))?;
61//! let agent = AgentBuilder::create(config, AgentStorage::new(store))
62//! .with_provider(provider).with_toolset(tools).build().await?;
63//! let outcome = async {
64//! let cwd = std::env::current_dir()?;
65//! let context = WorkspaceContext::new(cwd.clone(),
66//! vec![WorkspaceRoot::new(cwd, WorkspaceRootAccess::ReadWrite)?],
67//! WorkspaceExecutionPolicy::new(WorkspaceSandboxMode::FullAccess, WorkspaceNetworkAccess::Enabled))?;
68//! let workspace = agent.new_workspace(NewWorkspace::new("review", context, WorkspaceDefaults::empty())?).await?;
69//! let work = async {
70//! let conversation = workspace.new_conversation(NewConversation::new(None, None, vec![])?).await?;
71//! let work = async {
72//! let mut notifications = conversation.notification_stream()?;
73//! let request = PromptRequest::new("Call inspect_status, then describe its outcome.")?
74//! .with_tool_selection(ToolSelection::Required)?
75//! .with_timeout(Duration::from_secs(90))?;
76//! let mut output = conversation.prompt_stream(request).await?;
77//! let completion = tokio::time::timeout(Duration::from_secs(90), async {
78//! loop {
79//! tokio::select! {
80//! event = output.recv() => match event {
81//! Some(Ok(PromptEvent::OutputDelta { text })) => print!("{text}"),
82//! Some(Ok(PromptEvent::Completed(response))) => break Ok(response),
83//! Some(Err(error)) => break Err(error),
84//! None => break Err(agents_kit::KitError::protocol("completion is missing")),
85//! },
86//! event = notifications.recv() => {
87//! let event = event.ok_or_else(|| agents_kit::KitError::closed("notification stream ended"))??;
88//! match event.kind() {
89//! ConversationNotificationKind::Interaction { title, body, .. } => {
90//! println!("Decision requested: {title}: {body}");
91//! conversation.resolve_notification(event.id().clone(),
92//! NotificationResolution::Deny { reason: "No action approved by this example".into() }).await?;
93//! }
94//! _ => println!("Observed passive notification {}", event.id()),
95//! }
96//! }
97//! }
98//! }
99//! }).await.map_err(|_| agents_kit::KitError::timeout("caller observation deadline"))??;
100//! assert!(output.recv().await.is_none());
101//! assert!(!completion.output().trim().is_empty());
102//! println!("Authoritative answer: {}", completion.output());
103//!
104//! // Admission succeeds before cancellation. Dropping `output` would not cancel.
105//! let mut cancelled = conversation.prompt_stream(PromptRequest::new("Explain the task in detail.")?
106//! .with_tool_selection(ToolSelection::None)?
107//! .with_timeout(Duration::from_secs(90))?).await?;
108//! conversation.abort().await?;
109//! let terminal = tokio::time::timeout(Duration::from_secs(90), async {
110//! loop {
111//! match cancelled.recv().await {
112//! Some(Ok(PromptEvent::OutputDelta { .. })) => {}
113//! Some(Ok(PromptEvent::Completed(_))) => break Ok(()), // completion may win the race
114//! Some(Err(error)) if error.kind() == agents_kit::error::ErrorKind::Cancelled => break Ok(()),
115//! Some(Err(error)) => break Err(error),
116//! None => break Err(agents_kit::KitError::protocol("cancellation terminal is missing")),
117//! }
118//! }
119//! }).await.map_err(|_| agents_kit::KitError::timeout("cancellation observation deadline"))?;
120//! terminal?;
121//! Ok::<(), agents_kit::KitError>(())
122//! }.await;
123//! let close = conversation.close().await; // also settles work after observation failure
124//! match (work, close) {
125//! (Err(primary), Err(cleanup)) => Err(primary.with_supplementary(cleanup)),
126//! (Err(error), _) | (_, Err(error)) => Err(error),
127//! (Ok(()), Ok(())) => Ok(()),
128//! }
129//! }.await;
130//! let close = workspace.close().await; // also runs if conversation creation failed
131//! match (work, close) {
132//! (Err(primary), Err(cleanup)) => Err(primary.with_supplementary(cleanup)),
133//! (Err(error), _) | (_, Err(error)) => Err(error),
134//! (Ok(()), Ok(())) => Ok(()),
135//! }
136//! }.await;
137//! let shutdown = agent.shutdown().await;
138//! match (outcome, shutdown) {
139//! (Err(primary), Err(cleanup)) => Err(primary.with_supplementary(cleanup)),
140//! (Err(error), _) | (_, Err(error)) => Err(error),
141//! (Ok(()), Ok(())) => Ok(()),
142//! }?;
143//! # Ok::<(), Box<dyn std::error::Error>>(())
144//! # })?;
145//! # Ok::<(), Box<dyn std::error::Error>>(())
146//! ```
147//!
148use std::future::Future;
149use std::path::PathBuf;
150use tools_kit::ToolsetBackend;
151
152use providers_kit::ProviderBackend;
153use storage_kit::{DatabaseBackend, StateStoreBackend};
154
155pub use crate::file::FileRead;
156
157use crate::backends::{codex, custom, opencode};
158use crate::prompt::PromptStream;
159use crate::{
160 ConversationId, ConversationNotificationReceiver, KitError, KnowledgeSource, NotificationId,
161 NotificationResolution, PromptRequest, PromptResponse, WorkspaceDefaults, WorkspaceId,
162};
163
164/// Descriptive state for one portable conversation.
165///
166/// # Examples
167///
168/// ```
169/// use agents_kit::conversation::ConversationInfo;
170/// let id = "conversation_00000000000000000000000000000000".parse()?;
171/// let workspace = "workspace_00000000000000000000000000000000".parse()?;
172/// let info = ConversationInfo::new(id, workspace, Some("Review".into()))?;
173/// assert_eq!(info.title(), Some("Review"));
174/// # Ok::<(), agents_kit::KitError>(())
175/// ```
176#[derive(Clone, Debug, Eq, PartialEq)]
177pub struct ConversationInfo {
178 id: ConversationId,
179 workspace_id: WorkspaceId,
180 title: Option<String>,
181}
182
183impl ConversationInfo {
184 /// Validates and constructs descriptive conversation information.
185 ///
186 /// Titles, when present, must be nonempty and must not contain leading or
187 /// trailing whitespace.
188 ///
189 /// # Errors
190 ///
191 /// Returns [`crate::error::ErrorKind::InvalidInput`] when `title` violates the portable
192 /// naming contract.
193 ///
194 /// # Examples
195 ///
196 /// ```
197 /// use agents_kit::{
198 /// conversation::ConversationInfo,
199 /// identity::{ConversationId, WorkspaceId},
200 /// };
201 ///
202 /// let id: ConversationId =
203 /// "conversation_00000000000000000000000000000000".parse()?;
204 /// let workspace_id: WorkspaceId =
205 /// "workspace_00000000000000000000000000000000".parse()?;
206 /// let info =
207 /// ConversationInfo::new(id, workspace_id, Some("release".to_owned()))?;
208 /// assert_eq!(info.title(), Some("release"));
209 /// # Ok::<(), agents_kit::KitError>(())
210 /// ```
211 pub fn new(
212 id: ConversationId,
213 workspace_id: WorkspaceId,
214 title: Option<String>,
215 ) -> Result<Self, KitError> {
216 Self::validate_title(title.as_deref())?;
217 Ok(Self {
218 id,
219 workspace_id,
220 title,
221 })
222 }
223
224 /// Returns the portable conversation identifier.
225 ///
226 /// See [`ConversationInfo`] for a complete example.
227 pub fn id(&self) -> &ConversationId {
228 &self.id
229 }
230
231 /// Returns the portable identifier of the owning workspace.
232 ///
233 /// See [`ConversationInfo`] for a complete example.
234 pub fn workspace_id(&self) -> &WorkspaceId {
235 &self.workspace_id
236 }
237
238 /// Returns the configured title, when present.
239 ///
240 /// See [`ConversationInfo`] for a complete example.
241 pub fn title(&self) -> Option<&str> {
242 self.title.as_deref()
243 }
244
245 fn validate_title(title: Option<&str>) -> Result<(), KitError> {
246 if title.is_some_and(|value| value.is_empty() || value.trim() != value) {
247 return Err(KitError::invalid_input(
248 "conversation title must be nonempty and unpadded when present",
249 ));
250 }
251 Ok(())
252 }
253}
254
255/// Validated inputs for creating a conversation.
256///
257/// # Examples
258///
259/// ```
260/// use agents_kit::conversation::NewConversation;
261///
262/// let request = NewConversation::new(
263/// Some("incident review".to_owned()),
264/// None,
265/// Vec::new(),
266/// )?;
267/// assert_eq!(request.title(), Some("incident review"));
268/// # Ok::<(), agents_kit::KitError>(())
269/// ```
270#[derive(Clone, Debug, Eq, PartialEq)]
271pub struct NewConversation {
272 title: Option<String>,
273 defaults: Option<WorkspaceDefaults>,
274 knowledge_sources: Vec<KnowledgeSource>,
275}
276
277impl NewConversation {
278 /// Validates and constructs conversation-creation inputs.
279 ///
280 /// # Errors
281 ///
282 /// Returns `InvalidInput` for an empty or padded title, an invalid knowledge policy/source, or duplicate source IDs. Construction performs no native I/O.
283 ///
284 /// # Examples
285 ///
286 /// ```
287 /// use agents_kit::conversation::NewConversation;
288 /// let request = NewConversation::new(Some("Review".into()), None, Vec::new())?;
289 /// assert_eq!(request.title(), Some("Review"));
290 /// assert!(request.defaults().is_none());
291 /// assert!(request.knowledge_sources().is_empty());
292 /// # Ok::<(), agents_kit::KitError>(())
293 /// ```
294 pub fn new(
295 title: Option<String>,
296 defaults: Option<WorkspaceDefaults>,
297 knowledge_sources: Vec<KnowledgeSource>,
298 ) -> Result<Self, KitError> {
299 ConversationInfo::validate_title(title.as_deref())?;
300 if let Some(defaults) = &defaults
301 && let Some(policy) = defaults.knowledge_policy()
302 {
303 policy.validate()?;
304 }
305 let mut ids = std::collections::BTreeSet::new();
306 for source in &knowledge_sources {
307 source.validate()?;
308 if !ids.insert(source.source_id()) {
309 return Err(KitError::invalid_input(
310 "conversation knowledge source ids must be unique",
311 ));
312 }
313 }
314 Ok(Self {
315 title,
316 defaults,
317 knowledge_sources,
318 })
319 }
320
321 /// Returns the requested title, when present.
322 ///
323 /// See [`NewConversation`] for a complete example.
324 pub fn title(&self) -> Option<&str> {
325 self.title.as_deref()
326 }
327
328 /// Returns conversation-level default overrides, when present.
329 ///
330 /// See [`NewConversation`] for a complete example.
331 pub fn defaults(&self) -> Option<&WorkspaceDefaults> {
332 self.defaults.as_ref()
333 }
334
335 /// Returns the initial managed-knowledge sources.
336 ///
337 /// See [`NewConversation`] for a complete example.
338 pub fn knowledge_sources(&self) -> &[KnowledgeSource] {
339 &self.knowledge_sources
340 }
341}
342
343/// Public implementation contract for one live portable conversation.
344///
345/// A caller workspace returns its own concrete conversation through
346/// [`WorkspaceBackend::Conversation`](crate::WorkspaceBackend::Conversation).
347/// Prompting reaches a terminal portable result, abort joins active work, and close
348/// permanently closes the durable conversation. Runtime composition may use an
349/// exhaustive caller-owned enum hierarchy.
350///
351/// # Runnable caller example
352///
353/// This small test double accepts one prompt and one file path. It demonstrates
354/// portable consumption without a service or database; it is not a durable agent.
355/// Notification and interaction methods return explicit Unsupported errors;
356/// request assertions make the fixture's expected inputs visible. The streaming
357/// response is caller-authored, not fabricated deltas from a supplied backend.
358///
359/// ```
360/// use std::path::{Path, PathBuf};
361/// use agents_kit::{ConversationBackend, KitError};
362/// use agents_kit::conversation::{ConversationInfo, FileRead};
363/// use agents_kit::identity::{ConversationId, NotificationId, WorkspaceId};
364/// use agents_kit::notification::{ConversationNotificationReceiver, NotificationResolution};
365/// use agents_kit::prompt::{PromptEvent, PromptRequest, PromptResponse, PromptStream};
366///
367/// struct ScriptedConversation {
368/// id: ConversationId,
369/// workspace: WorkspaceId,
370/// }
371/// impl ConversationBackend for ScriptedConversation {
372/// fn id(&self) -> ConversationId { self.id.clone() }
373/// async fn info(&self) -> Result<ConversationInfo, KitError> {
374/// ConversationInfo::new(self.id(), self.workspace.clone(), Some("Example".into()))
375/// }
376/// async fn prompt(&self, request: PromptRequest) -> Result<PromptResponse, KitError> {
377/// assert_eq!(request.text(), "Status?");
378/// Ok(PromptResponse::new("ready".into(), None))
379/// }
380/// async fn prompt_stream(&self, request: PromptRequest) -> Result<PromptStream, KitError> {
381/// assert_eq!(request.text(), "Status?");
382/// Ok(PromptStream::from_stream(futures_util::stream::iter([
383/// Ok(PromptEvent::OutputDelta { text: "ready".into() }),
384/// Ok(PromptEvent::Completed(PromptResponse::new("ready".into(), None))),
385/// ])))
386/// }
387/// async fn read_file(&self, path: PathBuf) -> Result<FileRead, KitError> {
388/// assert_eq!(path, Path::new("status.txt"));
389/// FileRead::from_bytes(path, b"ready\n".to_vec())
390/// }
391/// // No background work or durable interaction exists in this fixture.
392/// async fn abort(&self) -> Result<(), KitError> { Ok(()) }
393/// fn notification_stream(&self) -> Result<ConversationNotificationReceiver, KitError> {
394/// Err(KitError::unsupported("this fixture has no notifications"))
395/// }
396/// async fn resolve_notification(&self, _: NotificationId, _: NotificationResolution)
397/// -> Result<(), KitError> {
398/// Err(KitError::unsupported("this fixture has no interactions"))
399/// }
400/// async fn close(self) -> Result<(), KitError> { Ok(()) }
401/// }
402/// # tokio::runtime::Builder::new_current_thread().build()?.block_on(async {
403/// let conversation = ScriptedConversation {
404/// id: "conversation_00000000000000000000000000000001".parse()?,
405/// workspace: "workspace_00000000000000000000000000000001".parse()?,
406/// };
407/// let outcome = async {
408/// assert_eq!(conversation.info().await?.id(), &conversation.id());
409/// assert_eq!(conversation.prompt(PromptRequest::new("Status?")?).await?.output(), "ready");
410/// let mut output = conversation.prompt_stream(PromptRequest::new("Status?")?).await?;
411/// assert!(matches!(output.recv().await.transpose()?,
412/// Some(PromptEvent::OutputDelta { text }) if text == "ready"));
413/// assert!(matches!(output.recv().await.transpose()?,
414/// Some(PromptEvent::Completed(response)) if response.output() == "ready"));
415/// assert!(output.recv().await.is_none());
416/// assert!(output.recv().await.is_none());
417/// assert_eq!(conversation.read_file("status.txt".into()).await?.text(), Some("ready\n"));
418/// conversation.abort().await?;
419/// Ok::<(), KitError>(())
420/// }.await;
421/// let cleanup = conversation.close().await;
422/// match (outcome, cleanup) {
423/// (Err(primary), Err(cleanup)) => Err(primary.with_supplementary(cleanup)),
424/// (Err(error), _) | (_, Err(error)) => Err(error),
425/// (Ok(()), Ok(())) => Ok(()),
426/// }
427/// # })?;
428/// # Ok::<(), Box<dyn std::error::Error>>(())
429/// ```
430pub trait ConversationBackend: Send + Sync + 'static {
431 /// Returns the durable portable conversation identity.
432 ///
433 /// See the runnable caller implementation on [`ConversationBackend`]. Supplied backends retain their own runtime and storage prerequisites.
434 fn id(&self) -> ConversationId;
435
436 /// Reads the current descriptive conversation state.
437 ///
438 /// # Errors
439 ///
440 /// Returns `Closed` for a terminal handle, `Conflict` for stale ownership/revision, or a typed storage/backend failure.
441 ///
442 /// See the runnable caller implementation on [`ConversationBackend`]. Supplied backends retain their own runtime and storage prerequisites.
443 fn info(&self) -> impl Future<Output = Result<ConversationInfo, KitError>> + Send + '_;
444
445 /// Runs one foreground prompt through its terminal portable result.
446 ///
447 /// Unary and incremental calls share one accepted prompt owner. Dropping the waiter does not cancel accepted work or free its slot. Success retains exact authored text; structured output is separately validated and usage is present only when sourceable.
448 ///
449 /// # Errors
450 ///
451 /// Returns `InvalidInput`/`Unsupported` before acceptance for invalid or unsupported requests, `Busy` when admission is occupied, or typed cancellation, deadline, conflict, storage, and backend failures. After ambiguous acceptance, do not blindly retry: cancellation/reconciliation retains the generation fence until settlement.
452 ///
453 /// See the runnable caller implementation on [`ConversationBackend`]. Supplied backends retain their own runtime and storage prerequisites.
454 fn prompt(
455 &self,
456 request: PromptRequest,
457 ) -> impl Future<Output = Result<PromptResponse, KitError>> + Send + '_;
458
459 /// Starts real incremental observation of one foreground prompt.
460 ///
461 /// The default rejects unsupported streaming before accepting work. Override
462 /// using public [`PromptStream`] and [`PromptEvent`](crate::prompt::PromptEvent)
463 /// values; do not fabricate deltas from a terminal response. Dropping an
464 /// observer does not cancel accepted work or release the conversation slot.
465 ///
466 /// # Errors
467 ///
468 /// Returns `Unsupported` when the backend has no genuine incremental route, or the same admission failures as `prompt`. After admission, the stream carries terminal failures; premature source EOF is a protocol failure.
469 ///
470 /// The runnable [`ConversationBackend`] example consumes real caller-stream events. [`PromptStream`] also demonstrates terminal validation; the [native interaction recipe](crate::conversation) requires a configured service.
471 fn prompt_stream(
472 &self,
473 _request: PromptRequest,
474 ) -> impl Future<Output = Result<PromptStream, KitError>> + Send + '_ {
475 async {
476 Err(KitError::unsupported(
477 "incremental prompt output is unavailable",
478 ))
479 }
480 }
481
482 /// Aborts the active prompt, or succeeds when no prompt is active.
483 ///
484 /// Cancellation targets only the active generation and joins its settlement. Natural completion may win; abort does not replace the prompt waiter or return its answer.
485 ///
486 /// # Errors
487 ///
488 /// Returns a typed cancellation, timeout, conflict, lifecycle, or backend failure. Ambiguous cancellation retains recovery ownership; an error does not prove remote work stopped.
489 ///
490 /// See the runnable caller implementation on [`ConversationBackend`]. Supplied backends retain their own runtime and storage prerequisites.
491 fn abort(&self) -> impl Future<Output = Result<(), KitError>> + Send + '_;
492
493 /// Reads a validated workspace-relative file through the implementation.
494 ///
495 /// The path is interpreted against the admitted workspace cwd on the backend host. Absolute paths, dot/parent components and repeated separators are rejected rather than normalized.
496 ///
497 /// # Errors
498 ///
499 /// Returns `InvalidInput` for a nonportable relative path, `Unsupported` for a route unable to preserve exact bytes/policy, or a typed revision, lifecycle, storage, or native read failure. An untyped native rejection is not evidence of `NotFound`.
500 ///
501 /// See the runnable caller implementation on [`ConversationBackend`]. Supplied backends retain their own runtime and storage prerequisites.
502 fn read_file(
503 &self,
504 path: PathBuf,
505 ) -> impl Future<Output = Result<FileRead, KitError>> + Send + '_;
506
507 /// Claims the conversation's exclusive durable notification receiver.
508 ///
509 /// # Errors
510 ///
511 /// Returns `Busy` while another receiver owns the lease, `Closed` for a closed handle, or a typed admission failure. Later durable delivery failures are yielded by `recv`.
512 ///
513 /// See [`ConversationNotificationReceiver`] for executable caller-stream delivery, and the [native interaction recipe](crate::conversation) for durable delivery.
514 fn notification_stream(&self) -> Result<ConversationNotificationReceiver, KitError>;
515
516 /// Submits a resolution for an unresolved portable notification.
517 ///
518 /// # Errors
519 ///
520 /// Returns `InvalidInput` for an invalid resolution, `NotFound` for an unknown notification, or typed lifecycle, conflict, storage, and native resolution failures. Intent is durable before dispatch; known or ambiguous acceptance is never blindly resent.
521 ///
522 /// See [`NotificationResolution`] for resolution values and the [native interaction recipe](crate::conversation) for resolving an admitted interaction. Caller implementations are exercised by the [direct hierarchy tests](https://github.com/jecsand838/ai-kit/blob/main/agents-kit/tests/mock_hierarchy/mod.rs) without a service.
523 fn resolve_notification(
524 &self,
525 id: NotificationId,
526 resolution: NotificationResolution,
527 ) -> impl Future<Output = Result<(), KitError>> + Send + '_;
528
529 /// Permanently closes the conversation and consumes the implementation.
530 ///
531 /// Unlike parent workspace close or Agent shutdown, this permanently closes the durable conversation. Admission is sealed and active work, interactions, knowledge and owned resources are settled before completion.
532 ///
533 /// # Errors
534 ///
535 /// Returns a typed cancellation, reconciliation, storage, or cleanup failure. The handle remains consumed on failure; durable recovery obligations remain authoritative.
536 ///
537 /// See the runnable caller implementation on [`ConversationBackend`]. Supplied backends retain their own runtime and storage prerequisites.
538 fn close(self) -> impl Future<Output = Result<(), KitError>> + Send
539 where
540 Self: Sized;
541}
542
543/// Opaque live handle for one portable conversation.
544///
545/// See [`ConversationBackend`] and the [native interaction recipe](crate::conversation) for usage.
546pub struct Conversation<
547 S: StateStoreBackend = storage_kit::StateStore,
548 D: DatabaseBackend = storage_kit::Database,
549 P: ProviderBackend = providers_kit::Provider,
550 T: ToolsetBackend = tools_kit::Toolset,
551> {
552 inner: ConversationInner<S, D, P, T>,
553}
554
555enum ConversationInner<
556 S: StateStoreBackend,
557 D: DatabaseBackend,
558 P: ProviderBackend,
559 T: ToolsetBackend,
560> {
561 Codex(codex::conversations::CodexConversation<S, D, T>),
562 OpenCode(opencode::conversations::OpenCodeConversation<S, D>),
563 Custom(custom::CustomConversation<S, D, P, T>),
564}
565
566impl<S, D, P, T> Conversation<S, D, P, T>
567where
568 S: StateStoreBackend,
569 D: DatabaseBackend,
570 P: ProviderBackend,
571 T: ToolsetBackend,
572{
573 pub(crate) fn codex(backend: codex::conversations::CodexConversation<S, D, T>) -> Self {
574 Self {
575 inner: ConversationInner::Codex(backend),
576 }
577 }
578
579 pub(crate) fn opencode(backend: opencode::conversations::OpenCodeConversation<S, D>) -> Self {
580 Self {
581 inner: ConversationInner::OpenCode(backend),
582 }
583 }
584
585 pub(crate) fn custom(backend: custom::CustomConversation<S, D, P, T>) -> Self {
586 Self {
587 inner: ConversationInner::Custom(backend),
588 }
589 }
590}
591
592impl<S, D, P, T> ConversationBackend for ConversationInner<S, D, P, T>
593where
594 S: StateStoreBackend,
595 D: DatabaseBackend,
596 P: ProviderBackend,
597 T: ToolsetBackend,
598{
599 fn id(&self) -> ConversationId {
600 match self {
601 ConversationInner::Codex(backend) => backend.id(),
602 ConversationInner::OpenCode(backend) => backend.id(),
603 ConversationInner::Custom(backend) => backend.id(),
604 }
605 }
606
607 async fn info(&self) -> Result<ConversationInfo, KitError> {
608 match self {
609 ConversationInner::Codex(backend) => backend.info().await,
610 ConversationInner::OpenCode(backend) => backend.info().await,
611 ConversationInner::Custom(backend) => backend.info().await,
612 }
613 }
614
615 async fn prompt(&self, request: PromptRequest) -> Result<PromptResponse, KitError> {
616 match self {
617 ConversationInner::Codex(backend) => backend.prompt(request).await,
618 ConversationInner::OpenCode(backend) => backend.prompt(request).await,
619 ConversationInner::Custom(backend) => backend.prompt(request).await,
620 }
621 }
622
623 async fn prompt_stream(&self, request: PromptRequest) -> Result<PromptStream, KitError> {
624 match self {
625 ConversationInner::Codex(backend) => backend.prompt_stream(request).await,
626 ConversationInner::OpenCode(backend) => backend.prompt_stream(request).await,
627 ConversationInner::Custom(backend) => backend.prompt_stream(request).await,
628 }
629 }
630
631 async fn abort(&self) -> Result<(), KitError> {
632 match self {
633 ConversationInner::Codex(backend) => backend.abort().await,
634 ConversationInner::OpenCode(backend) => backend.abort().await,
635 ConversationInner::Custom(backend) => backend.abort().await,
636 }
637 }
638
639 async fn read_file(&self, path: PathBuf) -> Result<FileRead, KitError> {
640 match self {
641 ConversationInner::Codex(backend) => backend.read_file(path).await,
642 ConversationInner::OpenCode(backend) => backend.read_file(path).await,
643 ConversationInner::Custom(backend) => backend.read_file(path).await,
644 }
645 }
646
647 fn notification_stream(&self) -> Result<ConversationNotificationReceiver, KitError> {
648 match self {
649 ConversationInner::Codex(backend) => backend.notification_stream(),
650 ConversationInner::OpenCode(backend) => backend.notification_stream(),
651 ConversationInner::Custom(backend) => backend.notification_stream(),
652 }
653 }
654
655 async fn resolve_notification(
656 &self,
657 id: NotificationId,
658 resolution: NotificationResolution,
659 ) -> Result<(), KitError> {
660 match self {
661 ConversationInner::Codex(backend) => backend.resolve_notification(id, resolution).await,
662 ConversationInner::OpenCode(backend) => {
663 backend.resolve_notification(id, resolution).await
664 }
665 ConversationInner::Custom(backend) => {
666 backend.resolve_notification(id, resolution).await
667 }
668 }
669 }
670
671 async fn close(self) -> Result<(), KitError> {
672 match self {
673 ConversationInner::Codex(backend) => backend.close().await,
674 ConversationInner::OpenCode(backend) => backend.close().await,
675 ConversationInner::Custom(backend) => backend.close().await,
676 }
677 }
678}
679
680impl<S, D, P, T> ConversationBackend for Conversation<S, D, P, T>
681where
682 S: StateStoreBackend,
683 D: DatabaseBackend,
684 P: ProviderBackend,
685 T: ToolsetBackend,
686{
687 fn id(&self) -> ConversationId {
688 self.inner.id()
689 }
690
691 async fn info(&self) -> Result<ConversationInfo, KitError> {
692 self.inner.info().await.map_err(|error| {
693 error
694 .at_operation(crate::error::Operation::ConversationInfo)
695 .in_phase(crate::error::ErrorPhase::StateRead)
696 })
697 }
698
699 async fn prompt(&self, request: PromptRequest) -> Result<PromptResponse, KitError> {
700 self.inner.prompt(request).await.map_err(|error| {
701 error
702 .at_operation(crate::error::Operation::Prompt)
703 .in_phase(crate::error::ErrorPhase::ModelGeneration)
704 })
705 }
706
707 async fn prompt_stream(&self, request: PromptRequest) -> Result<PromptStream, KitError> {
708 self.inner.prompt_stream(request).await.map_err(|error| {
709 error
710 .at_operation(crate::error::Operation::Prompt)
711 .in_phase(crate::error::ErrorPhase::ModelGeneration)
712 })
713 }
714
715 async fn abort(&self) -> Result<(), KitError> {
716 self.inner.abort().await.map_err(|error| {
717 error
718 .at_operation(crate::error::Operation::Abort)
719 .in_phase(crate::error::ErrorPhase::Cancellation)
720 })
721 }
722
723 async fn read_file(&self, path: PathBuf) -> Result<FileRead, KitError> {
724 self.inner.read_file(path).await.map_err(|error| {
725 error
726 .at_operation(crate::error::Operation::ReadFile)
727 .in_phase(crate::error::ErrorPhase::Admission)
728 })
729 }
730
731 fn notification_stream(&self) -> Result<ConversationNotificationReceiver, KitError> {
732 self.inner.notification_stream().map_err(|error| {
733 error
734 .at_operation(crate::error::Operation::NotificationStream)
735 .in_phase(crate::error::ErrorPhase::Admission)
736 })
737 }
738
739 async fn resolve_notification(
740 &self,
741 id: NotificationId,
742 resolution: NotificationResolution,
743 ) -> Result<(), KitError> {
744 self.inner
745 .resolve_notification(id, resolution)
746 .await
747 .map_err(|error| {
748 error
749 .at_operation(crate::error::Operation::ResolveNotification)
750 .in_phase(crate::error::ErrorPhase::InteractionResolution)
751 })
752 }
753
754 async fn close(self) -> Result<(), KitError> {
755 self.inner.close().await.map_err(|error| {
756 error
757 .at_operation(crate::error::Operation::CloseConversation)
758 .in_phase(crate::error::ErrorPhase::Lifecycle)
759 })
760 }
761}