Module conversation 

Conversation handles, backend contracts, and conversation requests and responses. Conversation metadata, accepted prompts, exact file reads and durable interactions.

Creation inputs are passive. Live operations use ConversationBackend; its consuming close permanently terminates the conversation. Prompt streams observe runtime-owned accepted work, so dropping an observer is not cancellation.

Streaming, durable approval and explicit cancellation

This complete Custom journey requires a running Ollama server and the exact configured model with function-calling support. It is therefore no_run. Replace only the caller configuration to use your intended model; selection does not substitute a fallback. The caller supplies the executable tool. Caller-built actions run in process without Kit sandbox enforcement, so Custom requires FullAccess, Enabled and read-write roots for this configuration; roots do not confine the action. This example’s inspect_status action performs no filesystem or network I/O.

Notifications are an exclusive, durable, at-least-once delivery path. Inspect an Interaction before resolving its exact ID; advancing passive delivery is acknowledgment, and dropping the receiver is not acknowledgment. The example deliberately denies every requested action: a production UI can instead collect a caller decision and send Allow or schema-valid Submit content. Denial is a real interaction resolution, not a fabricated tool result.

The output observer requires the authoritative Completed value. Deltas can include earlier assistant turns around tool execution, so concatenating them is not an alternative final answer. Observer loss does not abort accepted work; the separate cancellation portion uses abort, then observes settlement.

use std::{num::NonZeroUsize, time::Duration};
use agents_kit::{AgentBackend, AgentBuilder, AgentStorage, ConversationBackend, WorkspaceBackend};
use agents_kit::config::{AgentConfig, CustomAgentConfig, CustomLoopLimits, CustomToolApproval};
use agents_kit::conversation::NewConversation;
use agents_kit::notification::{ConversationNotificationKind, NotificationResolution};
use agents_kit::prompt::{PromptEvent, PromptRequest};
use agents_kit::{model::ModelSelector, tool::ToolSelection};
use agents_kit::workspace::{NewWorkspace, WorkspaceContext, WorkspaceDefaults,
    WorkspaceExecutionPolicy, WorkspaceNetworkAccess, WorkspaceRoot,
    WorkspaceRootAccess, WorkspaceSandboxMode};
use providers_kit::{Provider, ProviderConfig,
    config::{OllamaConfig, ProviderKind, TransportConfig}};
use storage_kit::{StateStore,
    state::config::{LanceStateStoreConfig, StateStoreConfig}};
use tools_kit::{Toolset, declaration::{ToolInfo, ToolName}, execution::ToolResult,
    local::ToolRegistration, schema::JsonSchema};
let store = StateStore::connect(StateStoreConfig::Lance(LanceStateStoreConfig::memory())).await?;
let tools = Toolset::local(vec![ToolRegistration::function(
    ToolInfo::new(ToolName::new("inspect_status")?,
        JsonSchema::strict(serde_json::json!({"type": "object", "additionalProperties": false}))),
    |_| async { Ok(ToolResult::json(serde_json::json!({"status": "ready"}))) },
)])?;
let custom = CustomAgentConfig::new(ModelSelector::new("exact-tool-capable-model")?,
    ToolSelection::Auto, CustomLoopLimits::new(4, 4, 1)?, CustomToolApproval::Interaction)?;
let config = AgentConfig::new(custom.into(), Duration::from_secs(60), NonZeroUsize::MIN)?;
let provider = Provider::from_config(ProviderConfig::new(
    ProviderKind::Ollama(OllamaConfig::local()), TransportConfig::default()))?;
let agent = AgentBuilder::create(config, AgentStorage::new(store))
    .with_provider(provider).with_toolset(tools).build().await?;
let outcome = async {
    let cwd = std::env::current_dir()?;
    let context = WorkspaceContext::new(cwd.clone(),
        vec![WorkspaceRoot::new(cwd, WorkspaceRootAccess::ReadWrite)?],
        WorkspaceExecutionPolicy::new(WorkspaceSandboxMode::FullAccess, WorkspaceNetworkAccess::Enabled))?;
    let workspace = agent.new_workspace(NewWorkspace::new("review", context, WorkspaceDefaults::empty())?).await?;
    let conversation = workspace.new_conversation(NewConversation::new(None, None, vec![])?).await?;
    let work = async {
        let mut notifications = conversation.notification_stream()?;
        let request = PromptRequest::new("Call inspect_status, then describe its outcome.")?
            .with_tool_selection(ToolSelection::Required)?
            .with_timeout(Duration::from_secs(90))?;
        let mut output = conversation.prompt_stream(request).await?;
        let completion = tokio::time::timeout(Duration::from_secs(90), async {
            loop {
                tokio::select! {
                    event = output.recv() => match event {
                        Some(Ok(PromptEvent::OutputDelta { text })) => print!("{text}"),
                        Some(Ok(PromptEvent::Completed(response))) => break Ok(response),
                        Some(Err(error)) => break Err(error),
                        None => break Err(agents_kit::KitError::protocol("completion is missing")),
                    },
                    event = notifications.recv() => {
                        let event = event.ok_or_else(|| agents_kit::KitError::closed("notification stream ended"))??;
                        match event.kind() {
                            ConversationNotificationKind::Interaction { title, body, .. } => {
                                println!("Decision requested: {title}: {body}");
                                conversation.resolve_notification(event.id().clone(),
                                    NotificationResolution::Deny { reason: "No action approved by this example".into() }).await?;
                            }
                            _ => println!("Observed passive notification {}", event.id()),
                        }
                    }
                }
            }
        }).await.map_err(|_| agents_kit::KitError::timeout("caller observation deadline"))??;
        assert!(output.recv().await.is_none());
        assert!(!completion.output().trim().is_empty());
        println!("Authoritative answer: {}", completion.output());

        // Admission succeeds before cancellation. Dropping `output` would not cancel.
        let mut cancelled = conversation.prompt_stream(PromptRequest::new("Explain the task in detail.")?
            .with_tool_selection(ToolSelection::None)?
            .with_timeout(Duration::from_secs(90))?).await?;
        conversation.abort().await?;
        let terminal = tokio::time::timeout(Duration::from_secs(90), async {
            loop {
                match cancelled.recv().await {
                    Some(Ok(PromptEvent::OutputDelta { .. })) => {}
                    Some(Ok(PromptEvent::Completed(_))) => break Ok(()), // completion may win the race
                    Some(Err(error)) if error.kind() == agents_kit::error::ErrorKind::Cancelled => break Ok(()),
                    Some(Err(error)) => break Err(error),
                    None => break Err(agents_kit::KitError::protocol("cancellation terminal is missing")),
                }
            }
        }).await.map_err(|_| agents_kit::KitError::timeout("cancellation observation deadline"))?;
        terminal?;
        Ok::<(), agents_kit::KitError>(())
    }.await;
    let close = conversation.close().await; // also settles work after observation failure
    let workspace_close = workspace.close().await;
    work?;
    close?;
    workspace_close?;
    Ok::<(), Box<dyn std::error::Error>>(())
}.await;
let shutdown = agent.shutdown().await;
outcome?;
shutdown?;

Structs

Conversation

Opaque live handle for one portable conversation.

ConversationInfo

Descriptive state for one portable conversation.

FileRead

Exact content returned by a portable workspace-relative file read.

NewConversation

Validated inputs for creating a conversation.

Traits

ConversationBackend

Public implementation contract for one live portable conversation.