Conversation

Stateful session management for the Antigravity Rust SDK.

Overview

Conversation is a stateful wrapper around a Connection that tracks step history, accumulates token usage metadata, and processes stream chunks. It is the primary interface for interacting with an active agent session — whether you want streaming chunk-by-chunk output or a simple blocking call-and-response.

Most users access Conversation through Agent<Started>:

use antigravity_sdk_rust::agent::Agent;

#[tokio::main]
async fn main() -> Result<(), anyhow::Error> {
    let agent = Agent::builder()
        .allow_all()
        .build()
        .start().await?;

    // Get the active Conversation (Arc<Conversation>)
    let conversation = agent.conversation();

    // Use it for streaming or blocking chat
    let response = conversation.chat_to_completion("Hello!").await?;
    println!("{}", response.text);

    agent.stop().await
}

Creating a Conversation

Conversation wraps an AnyConnection and an optional history size limit:

use antigravity_sdk_rust::conversation::Conversation;
use antigravity_sdk_rust::connection::AnyConnection;

// Created internally by Agent::start(), but can be constructed directly:
let conversation = Conversation::new(
    any_connection,     // AnyConnection (Local, Wasm, or Mock)
    Some(5000),         // max_history_size (None → default 10,000; Some(0) → unlimited)
);

Parameters

Parameter	Type	Default	Description
conn	AnyConnection	—	The underlying connection to the harness
max_history_size	Option<usize>	Some(10_000)	Max steps retained in memory. None uses 10,000. Some(0) disables trimming.

Key Methods

Chatting

chat(prompt) — Streaming

Sends a prompt and returns a BoxStream<StreamChunk> for real-time processing:

use antigravity_sdk_rust::types::StreamChunk;
use futures_util::StreamExt;

let mut stream = conversation.chat("Explain monads").await?;

while let Some(chunk_res) = stream.next().await {
    match chunk_res? {
        StreamChunk::Thought { text, step_index } => {
            // Model's internal reasoning (thinking tokens)
            eprint!("{}", text);
        }
        StreamChunk::Text { text, step_index } => {
            // Response text fragments
            print!("{}", text);
        }
        StreamChunk::ToolCall(call) => {
            // The model requested a tool execution
            println!("[Tool: {} | Args: {}]", call.name, call.args);
        }
    }
}

Signature:

pub async fn chat(
    &self,
    prompt: &str,
) -> Result<BoxStream<'static, Result<StreamChunk, anyhow::Error>>, anyhow::Error>

chat_to_completion(prompt) — Blocking

Sends a prompt and waits for the complete response, accumulating all chunks internally:

let response = conversation.chat_to_completion("What is 2 + 2?").await?;

println!("Response: {}", response.text);
println!("Thinking: {}", response.thinking);
println!("Steps: {}", response.steps.len());
// This turn only; `conversation.total_usage()` is the session total.
if let Some(usage) = &response.usage_metadata {
    println!("Turn tokens: {}", usage.total_token_count);
}

Signature:

pub async fn chat_to_completion(
    &self,
    prompt: &str,
) -> Result<ChatResponse, anyhow::Error>

Sending & Receiving (Low-Level)

Method	Signature	Description
send(prompt)	async fn send(&self, prompt: &str) -> Result<()>	Sends a raw prompt and registers a turn boundary. Does not return a stream.
receive_steps()	fn receive_steps(&self) -> BoxStream<'static, Result<Step>>	Raw step-level stream. Inserts steps into history, tracks compaction, enforces history limits.
receive_chunks()	fn receive_chunks(&self) -> BoxStream<'static, Result<StreamChunk>>	Filters receive_steps() into high-level StreamChunk events. Only emits model→user content.

receive_chunks() Filtering Logic

The chunk stream applies these rules:

1. Thought chunks — emitted when source == Model, target == User, and thinking_delta is non-empty
2. Text chunks — emitted when source == Model, target == User, and content_delta is non-empty
3. ToolCall chunks — emitted for each ToolCall in the step, deduplicated by call.id (empty IDs are never deduplicated)
4. Environment-targeted steps are silently filtered out (e.g., harness internal tool dispatches)

Connection Access

Method	Return Type	Description
connection()	AnyConnection	Returns the underlying connection (clone of Arc)
conversation_id()	&str	Session ID (delegates to Connection::conversation_id())
is_idle()	bool	Whether the connection is idle
disconnect()	async -> Result<()>	Gracefully closes the connection

StreamChunk

The streaming fragment enum used by chat() and receive_chunks():

use antigravity_sdk_rust::types::ToolCall;

#[derive(Debug, Clone)]
pub enum StreamChunk {
    /// Model's internal reasoning/thinking fragment.
    Thought {
        /// Index of the step that produced this chunk.
        step_index: u32,
        /// Thinking text delta.
        text: String,
    },

    /// Response text fragment directed at the user.
    Text {
        /// Index of the step that produced this chunk.
        step_index: u32,
        /// Text content delta.
        text: String,
    },

    /// A complete tool call requested by the model.
    ToolCall(ToolCall),
}

ToolCall Structure

#[derive(Debug, Clone)]
pub struct ToolCall {
    /// Unique correlation ID for this call.
    pub id: String,
    /// Name of the tool to invoke.
    pub name: String,
    /// Arguments as a JSON value.
    pub args: serde_json::Value,
    /// Optional canonical filesystem path (for file-targeting tools).
    pub canonical_path: Option<String>,
}

Streaming Example

The complete streaming pattern from examples/streaming.rs:

use antigravity_sdk_rust::agent::Agent;
use antigravity_sdk_rust::types::StreamChunk;
use futures_util::StreamExt;

#[tokio::main]
async fn main() -> Result<(), anyhow::Error> {
    dotenvy::dotenv().ok();

    let agent = Agent::builder()
        .default_model("gemini-3.5-flash")
        .allow_all()
        .build()
        .start().await?;

    let conversation = agent.conversation();

    let prompt = "Solve this riddle: I speak without a mouth \
        and hear without ears. What am I?";
    println!("User: {}\n", prompt);

    let mut stream = conversation.chat(prompt).await?;

    while let Some(chunk_res) = stream.next().await {
        match chunk_res? {
            StreamChunk::Thought { text, .. } => {
                // Print reasoning tokens as they arrive
                print!("{}", text);
                std::io::Write::flush(&mut std::io::stdout())?;
            }
            StreamChunk::Text { text, .. } => {
                // Print response text tokens as they arrive
                print!("{}", text);
                std::io::Write::flush(&mut std::io::stdout())?;
            }
            StreamChunk::ToolCall(call) => {
                println!(
                    "\n[Tool Call: {} with args: {}]",
                    call.name, call.args
                );
            }
        }
    }

    println!(); // Final newline
    agent.stop().await?;
    Ok(())
}

State Management

Conversation maintains a ConversationState protected by an async Mutex:

pub struct ConversationState {
    /// All executed steps (prompts, tool calls, responses, compaction markers).
    pub steps: Vec<Step>,
    /// Step indices marking each user prompt turn boundary.
    pub turn_start_indices: Vec<usize>,
    /// Step indices where history compaction occurred.
    pub compaction_indices: Vec<usize>,
    /// Cumulative token usage across all turns.
    pub cumulative_usage: UsageMetadata,
    /// Token usage for the current active turn.
    pub turn_usage: Option<UsageMetadata>,
}

State Query Methods

All state methods are async because they acquire the internal Mutex:

Method	Return Type	Description
history()	Vec<Step>	Clone of the full step history
turn_count()	usize	Number of user-initiated turns
compaction_indices()	Vec<usize>	Where the harness compacted history
last_response()	String	Text content of the last complete model response
total_usage()	UsageMetadata	Cumulative token usage for the entire session
last_turn_usage()	Option<UsageMetadata>	Token usage for the most recent turn
clear_history()	()	Resets all steps, turns, compactions, and usage stats

Usage Example

// After a chat interaction
let history = conversation.history().await;
println!("Total steps: {}", history.len());
println!("Turns completed: {}", conversation.turn_count().await);

let total = conversation.total_usage().await;
println!("Session tokens: {} prompt, {} generated, {} total",
    total.prompt_token_count,
    total.candidates_token_count,
    total.total_token_count,
);

if let Some(turn) = conversation.last_turn_usage().await {
    println!("Last turn: {} thinking tokens", turn.thoughts_token_count);
}

// Check if compaction happened
let compactions = conversation.compaction_indices().await;
if !compactions.is_empty() {
    println!("History was compacted at step indices: {:?}", compactions);
}

History Auto-Trimming

When max_history_size > 0 and the step count exceeds the limit, older steps are drained from the front. Turn and compaction indices are adjusted accordingly:

// Only keep the last 100 steps in memory
let conversation = Conversation::new(any_connection, Some(100));

// Disable auto-trimming entirely
let conversation = Conversation::new(any_connection, Some(0));

// Use the default (10,000 steps)
let conversation = Conversation::new(any_connection, None);

ChatResponse

The complete result returned by chat_to_completion():

#[derive(Debug, Clone)]
pub struct ChatResponse {
    /// Combined final response text.
    pub text: String,
    /// Combined model reasoning/thinking text.
    pub thinking: String,
    /// All steps executed during this turn.
    pub steps: Vec<Step>,
    /// Cumulative token usage metrics.
    pub usage_metadata: Option<UsageMetadata>,
}

UsageMetadata

#[derive(Debug, Clone, Default)]
pub struct UsageMetadata {
    /// Tokens included in the request prompt.
    pub prompt_token_count: i32,
    /// Tokens generated in model candidates.
    pub candidates_token_count: i32,
    /// Total combined tokens.
    pub total_token_count: i32,
    /// Cache-hit content tokens.
    pub cached_content_token_count: i32,
    /// Tokens consumed during reasoning/thinking.
    pub thoughts_token_count: i32,
}

Step Model

Each event in the trajectory is represented as a Step:

#[derive(Debug, Clone)]
pub struct Step {
    pub id: String,                              // Unique step ID
    pub step_index: u32,                         // Position in trajectory
    pub r#type: StepType,                        // TextResponse, ToolCall, Compaction, Finish, etc.
    pub source: StepSource,                      // System, User, Model
    pub target: StepTarget,                      // User, Environment
    pub status: StepStatus,                      // Active, Done, Error, TerminalError, etc.
    pub content: String,                         // Full accumulated text
    pub content_delta: String,                   // Incremental text delta
    pub thinking: String,                        // Full accumulated thinking
    pub thinking_delta: String,                  // Incremental thinking delta
    pub tool_calls: Vec<ToolCall>,               // Tool calls in this step
    pub error: String,                           // Error message (if any)
    pub is_complete_response: Option<bool>,       // True for final model response
    pub structured_output: Option<serde_json::Value>, // Parsed structured output
    pub usage_metadata: Option<UsageMetadata>,   // Per-step token usage
    pub cascade_id: String,                      // Execution cascade group
    pub trajectory_id: String,                   // Sub-agent trajectory group
    pub http_code: u32,                          // HTTP status code (if applicable)
}

Step Enums

Enum	Variants
StepType	TextResponse, ToolCall, SystemMessage, Compaction, Finish, Unknown
StepSource	System, User, Model, Unknown
StepTarget	User, Environment, Unspecified, Unknown
StepStatus	Active, Done, WaitingForUser, Error, Canceled, TerminalError, Unknown

Python SDK Comparison

Concept	Python SDK	Rust SDK
Conversation class	Conversation	Conversation (same name)
Streaming	async for chunk in conversation.chat(prompt)	conversation.chat(prompt).await?.next().await
Blocking	conversation.chat_to_completion(prompt)	conversation.chat_to_completion(prompt).await?
History	conversation.history (property)	conversation.history().await (async method)
Turn count	conversation.turn_count	conversation.turn_count().await
Token usage	conversation.total_usage	conversation.total_usage().await
State mutex	GIL + threading lock	tokio::sync::Mutex<ConversationState>
Stream type	AsyncIterator[StreamChunk]	BoxStream<'static, Result<StreamChunk>>
Max history	max_history_size param	max_history_size: Option<usize> (default 10,000)

Key difference: In the Rust SDK, all state-querying methods are async because the internal state is protected by a tokio::sync::Mutex. In Python, these are synchronous properties protected by the GIL.

send drains the previous turn

Steps still queued from the previous turn are drained into history before a new prompt goes out. A caller who stopped reading mid-turn used to lose those steps entirely, and the next turn's boundary was recorded at the wrong index.

The drain only runs once a turn has actually been sent — a freshly connected session reports not-idle until the harness says otherwise, and draining there would block on a stream with nothing to deliver.

wait_for_idle() resolves when the turn in flight finishes, returning immediately if none is running. It is watch-backed, so it notices the moment the harness reports idle rather than on the next tick of a poll loop.

Multimodal prompts and slash commands

Content carries text, attachments and slash commands, and goes out as the harness's complex_user_input — the plain prompt field is a bare string and can carry none of them. The types existed in this crate and reached nothing: a caller could build a Content and had no way to send it.

# use antigravity_sdk_rust::types::{Content, ContentPrimitive, Media, MimeType, ImageMime};
# async fn demo(agent: &antigravity_sdk_rust::agent::Agent<antigravity_sdk_rust::agent::Started>) -> Result<(), anyhow::Error> {
let prompt = Content::Multi(vec![
    ContentPrimitive::Text("what changed in this screenshot?".to_string()),
    ContentPrimitive::Media(Media {
        data: std::fs::read("before.png")?,
        mime_type: MimeType::Image(ImageMime::Png),
        description: None,
    }),
]);
let response = agent.chat_content(&prompt).await?;
# Ok(()) }

Text parts go through the same control-character strip as a plain prompt — a multimodal path that skipped it would be a way around it. An empty prompt is rejected before it reaches the harness, whichever form it takes.

last_structured_output() returns the most recent FINISH payload, which is what a response_schema produces; reaching it previously meant walking history() backwards looking for the right step type.