Files
optimclaw/src/agent/session.rs
T
d144484b06 feat: WASM channel attachments with LLM pipeline integration (#596)
* feat: add inbound attachment support to WASM channel system

Add attachment record to WIT interface and implement inbound media
parsing across all four channel implementations (Telegram, Slack,
WhatsApp, Discord). Attachments flow from WASM channels through
EmittedMessage to IncomingMessage with validation (size limits,
MIME allowlist, count caps) at the host boundary.

- Add `attachment` record to `emitted-message` in wit/channel.wit
- Add `IncomingAttachment` struct to channel.rs and re-export
- Add host-side validation (20MB total, 10 max, MIME allowlist)
- Telegram: parse photo, document, audio, video, voice, sticker
- Slack: parse file attachments with url_private
- WhatsApp: parse image, audio, video, document with captions
- Discord: backward-compatible empty attachments
- Update FEATURE_PARITY.md section 7
- Add fixture-based tests per channel and host integration tests

[skip-regression-check]

Co-Authored-By: Claude Opus 4.6 <[email protected]>

* feat: integrate outbound attachment support and reconcile WIT types (#409)

Reconcile PR #409's outbound attachment work with our inbound attachment
support into a unified design:

WIT type split:
- `inbound-attachment` in channel-host: metadata-only (id, mime_type,
  filename, size_bytes, source_url, storage_key, extracted_text)
- `attachment` in channel: raw bytes (filename, mime_type, data) on
  agent-response for outbound sending

Outbound features (from PR #409):
- `on-broadcast` WIT export for proactive messages without prior inbound
- Telegram: multipart sendPhoto/sendDocument with auto photo→document
  fallback for files >10MB
- wrapper.rs: `call_on_broadcast`, `read_attachments` from disk,
  attachment params threaded through `call_on_respond`
- HTTP tool: `save_to` param for binary downloads to /tmp/ (50MB limit,
  path traversal protection, SSRF-safe redirect following)
- Message tool: allow /tmp/ paths for attachments alongside base_dir
- Credential env var fallback in inject_channel_credentials

Channel updates:
- All 4 channels implement on_broadcast (Telegram full, others stub)
- Telegram: polling_enabled config, adjusted poll timeout
- Inbound attachment types renamed to InboundAttachment in all channels

Tests: 1965 passing (9 new), 0 clippy warnings

[skip-regression-check]

Co-Authored-By: Claude Opus 4.6 <[email protected]>

* feat: add audio transcription pipeline and extensible WIT attachment design

Add host-side transcription middleware (OpenAI Whisper) that detects audio
attachments with inline data on incoming messages and transcribes them
automatically. Refactor WIT inbound-attachment to use extras-json and a
store-attachment-data host function instead of typed fields, so future
attachment properties (dimensions, codec, etc.) don't require WIT changes
that invalidate all channel plugins.

- Add src/transcription/ module: TranscriptionProvider trait,
  TranscriptionMiddleware, AudioFormat enum, OpenAI Whisper provider
- Add src/config/transcription.rs: TRANSCRIPTION_ENABLED/MODEL/BASE_URL
- Wire middleware into agent message loop via AgentDeps
- WIT: replace data + duration-secs with extras-json + store-attachment-data
- Host: parse extras-json for well-known keys, merge stored binary data
- Telegram: download voice files via store-attachment-data, add duration
  to extras-json, add /file/bot to HTTP allowlist, voice-only placeholder
- Add reqwest multipart feature for Whisper API uploads
- 5 regression tests for transcription middleware

Co-Authored-By: Claude Opus 4.6 <[email protected]>

* feat: wire attachment processing into LLM pipeline with multimodal image support

Attachments on incoming messages are now augmented into user text via XML tags
before entering the turn system, and images with data are passed as multimodal
content parts (base64 data URIs) to LLM providers. This enables audio transcripts,
document text, and image content to reach the LLM without changes to ChatMessage
serialization or provider interfaces.

- Add src/agent/attachments.rs with augment_with_attachments() and 9 unit tests
- Add ContentPart/ImageUrl types to llm::provider with OpenAI-compatible serde
- Carry image_content_parts transiently on Turn (skipped in serialization)
- Update nearai_chat and rig_adapter to serialize multimodal content
- Add 3 e2e tests verifying attachments flow through the full agent loop

Co-Authored-By: Claude Opus 4.6 <[email protected]>

* fix: CI failures — formatting, version bumps, and Telegram voice test

- Fix cargo fmt formatting in attachments.rs, nearai_chat.rs, rig_adapter.rs,
  e2e_attachments.rs
- Bump channel registry versions 0.1.0 → 0.2.0 (discord, slack, telegram,
  whatsapp) to satisfy version-bump CI check
- Fix Telegram test_extract_attachments_voice: add missing required `duration`
  field to voice fixture JSON

Co-Authored-By: Claude Opus 4.6 <[email protected]>

* fix: bump WIT channel version to 0.3.0, fix Telegram voice test, add pre-commit hook

- Bump wit/channel.wit package version 0.2.0 → 0.3.0 (interface changed with
  store-attachment-data)
- Update WIT_CHANNEL_VERSION constant and registry wit_version fields to match
- Fix Telegram test_extract_attachments_voice: gate voice download behind
  #[cfg(target_arch = "wasm32")] so host functions aren't called in native tests,
  update assertions for generated filename and extras_json duration
- Add @0.3.0 linker stubs in wit_compat.rs
- Add .githooks/pre-commit hook that runs scripts/check-version-bumps.sh when
  WIT or extension sources are staged
- Symlink commit-msg regression hook into .githooks/

[skip-regression-check]

Co-Authored-By: Claude Opus 4.6 <[email protected]>

* refactor: extract voice download from extract_attachments into handle_message

Move download_voice_file + store_attachment_data calls out of
extract_attachments into a separate download_and_store_voice function
called from handle_message. This keeps extract_attachments as a pure
data-mapping function with no host calls, making it fully testable
in native unit tests without #[cfg(target_arch)] gates.

[skip-regression-check]

Co-Authored-By: Claude Opus 4.6 <[email protected]>

* fix: address PR review comments — security, correctness, and code quality

Security fixes:
- Add path validation to read_attachments (restrict to /tmp/) preventing
  arbitrary file reads from compromised tools
- Escape XML special characters in attachment filenames, MIME types, and
  extracted text to prevent prompt injection via tag spoofing
- Percent-encode file_id in Telegram getFile URL to prevent query injection
- Clone SecretString directly instead of expose_secret().to_string()

Correctness fixes:
- Fix store_attachment_data overwrite accounting: subtract old entry size
  before adding new to prevent inflated totals and false rejections
- Use max(reported, stored_size) for attachment size accounting to prevent
  WASM channels from under-reporting size_bytes to bypass limits
- Add application/octet-stream to MIME allowlist (channels default unknown
  types to this)

Code quality:
- Extract send_response helper in Telegram, deduplicating on_respond and
  on_broadcast
- Rename misleading Discord test to test_parse_slash_command_interaction
- Fix .githooks/commit-msg to use relative symlink (portable across machines)

[skip-regression-check]

Co-Authored-By: Claude Opus 4.6 <[email protected]>

* feat: add tool_upgrade command + fix TOCTOU in save_to path validation

Add `tool_upgrade` — a new extension management tool that automatically
detects and reinstalls WASM extensions with outdated WIT versions.
Preserves authentication secrets during upgrade. Supports upgrading a
single extension by name or all installed WASM tools/channels at once.

Fix TOCTOU in `validate_save_to_path`: validate the path *before*
creating parent directories, so traversal paths like `/tmp/../../etc/`
cannot cause filesystem mutations outside /tmp before being rejected.

[skip-regression-check]

Co-Authored-By: Claude Opus 4.6 <[email protected]>

* fix: unify WIT package version to 0.3.0 across tool.wit and all capabilities

tool.wit and channel.wit share the `near:agent` package namespace, so they
must declare the same version. Bumps tool.wit from 0.2.0 to 0.3.0 and
updates all capabilities files and registry entries to match.

Fixes `cargo component build` failure: "package identifier near:[email protected]
does not match previous package name of near:[email protected]"

[skip-regression-check]

Co-Authored-By: Claude Opus 4.6 <[email protected]>

* fix: move WIT file comments after package declaration

WIT treats `//` comments before `package` as doc comments. When both
tool.wit and channel.wit had header comments, the parser rejected them
as "doc comments on multiple 'package' items". Move comments after the
package declaration in both files.

Also bumps tool registry versions to 0.2.0 to match the WIT 0.3.0 bump.

[skip-regression-check]

Co-Authored-By: Claude Opus 4.6 <[email protected]>

* feat: display extension versions in gateway Extensions tab

Add version field to InstalledExtension and RegistryEntry types, pipe
through the web API (ExtensionInfo, RegistryEntryInfo), and render as
a badge in the gateway UI for both installed and available extensions.

For installed WASM extensions, version is read from the capabilities
file with a fallback to the registry entry when the local file has no
version (old installations). Bump all extension Cargo.toml and registry
JSON versions from 0.1.0 to 0.2.0 to keep them in sync.

[skip-regression-check]

Co-Authored-By: Claude Opus 4.6 <[email protected]>

* feat: add document text extraction middleware for PDF, Office, and text files

Extract text from document attachments (PDF, DOCX, PPTX, XLSX, RTF, plain text,
code files) so the LLM can reason about uploaded documents. Uses pdf-extract for
PDFs, zip+XML parsing for Office XML formats, and UTF-8 decode for text files.
Wired into the agent loop after transcription middleware.

Co-Authored-By: Claude Opus 4.6 <[email protected]>

* fix: download document files in Telegram channel for text extraction

The DocumentExtractionMiddleware needs file bytes in the attachment `data`
field, but only voice files were being downloaded. Document attachments
(PDFs, DOCX, etc.) had empty `data` and a source_url with a credential
placeholder that only works inside the WASM host's http_request.

Add `download_and_store_documents()` that downloads non-voice, non-image,
non-audio attachments via the existing two-step getFile→download flow and
stores bytes via `store_attachment_data` for host-side extraction.

Also rename `download_voice_file` → `download_telegram_file` since it's
generic for any file_id.

Co-Authored-By: Claude Opus 4.6 <[email protected]>

* fix: allow Office MIME types and increase file download limit for Telegram

Two issues preventing document extraction from Telegram:

1. PPTX/DOCX/XLSX MIME types (application/vnd.*) were dropped by the
   WASM host attachment allowlist — add application/vnd., application/msword,
   and application/rtf prefixes.

2. Telegram file downloads over 10 MB failed with "Response body too large" —
   set max_response_bytes to 20 MB in Telegram capabilities.

[skip-regression-check]

Co-Authored-By: Claude Opus 4.6 <[email protected]>

* fix: report document extraction errors back to user instead of silently skipping

- Bump max_response_bytes to 50 MB for Telegram file downloads
- When document extraction fails (too large, download error, parse error),
  set extracted_text to a user-friendly error message instead of leaving it
  None. This ensures the LLM tells the user what went wrong.
- On Telegram download failure, set extracted_text with the error so the
  user sees feedback even when the file never reaches the extraction middleware.

Co-Authored-By: Claude Opus 4.6 <[email protected]>

* feat: store extracted document text in workspace memory for search/recall

After document extraction succeeds, write the extracted text to workspace
memory at `documents/{date}/{filename}`. This enables:
- Full-text and semantic search over past uploaded documents
- Cross-conversation recall ("what did that PDF say?")
- Automatic chunking and embedding via the workspace pipeline

Documents are stored with metadata header (uploader, channel, date, MIME type).
Error messages (extraction failures) are not stored — only successful extractions.

Co-Authored-By: Claude Opus 4.6 <[email protected]>

* fix: CI failures — formatting, unused assignment warning

- Run cargo fmt on document_extraction and agent_loop modules
- Suppress unused_assignments warning on trace_llm_ref (used only
  behind #[cfg(feature = "libsql")])

[skip-regression-check]

Co-Authored-By: Claude Opus 4.6 <[email protected]>

* fix: address PR review comments — security, correctness, and code quality

Security fixes:
- Remove SSRF-prone download() from DocumentExtractionMiddleware (#13)
- Sanitize filenames in workspace path to prevent directory traversal (#11)
- Pre-check file size before reading in WASM wrapper to prevent OOM (#2)
- Percent-encode file_id in Telegram source URLs (#7)

Correctness fixes:
- Clear image_content_parts on turn end to prevent memory leak (#1)
- Find first *successful* transcription instead of first overall (#3)
- Enforce data.len() size limit in document extraction (#10)
- Use UTF-8 safe truncation with char_indices() (#12)

Robustness & code quality:
- Add 120s timeout to OpenAI Whisper HTTP client (#5)
- Trim trailing slash from Whisper base_url (#6)
- Allow ~/.ironclaw/ paths in WASM wrapper (#8)
- Return error from on_broadcast in Slack/Discord/WhatsApp (#9)
- Fix doc comment in HTTP tool (#4)

Co-Authored-By: Claude Opus 4.6 <[email protected]>

* fix: formatting — cargo fmt

Co-Authored-By: Claude Opus 4.6 <[email protected]>

* fix: address latest PR review — doc comments, error messages, version bumps

- Fix DocumentExtractionMiddleware doc comment (no longer downloads from source_url)
- Fix error message: "no inline data" instead of "no download URL"
- Log error + fallback instead of silent unwrap_or_default on Whisper HTTP client
- Bump all capabilities.json versions from 0.1.0 to 0.2.0 to match Cargo.toml

Co-Authored-By: Claude Opus 4.6 <[email protected]>

* fix: remove unsupported profile: minimal from CI workflows [skip-regression-check]

dtolnay/rust-toolchain@stable does not accept the 'profile' input
(it was a parameter for the deprecated actions-rs/toolchain action).

Co-Authored-By: Claude Opus 4.6 <[email protected]>

* fix: merge with latest main — resolve compilation errors and PR review nits

- Add version: None to RegistryEntry/InstalledExtension test constructors
- Fix MessageContent type mismatches in nearai_chat tests (String → MessageContent::Text)
- Fix .contains() calls on MessageContent — use .as_text().unwrap()
- Remove redundant trace_llm_ref = None assignment in test_rig
- Check data size before clone in document extraction to avoid unnecessary allocation

[skip-regression-check]

Co-Authored-By: Claude Opus 4.6 <[email protected]>

---------

Co-authored-by: Claude Opus 4.6 <[email protected]>
2026-03-07 18:01:40 +00:00

1033 lines
33 KiB
Rust

//! Session and thread model for turn-based agent interactions.
//!
//! A Session contains one or more Threads. Each Thread represents a
//! conversation/interaction sequence with the agent. Threads contain
//! Turns, which are request/response pairs.
//!
//! This model supports:
//! - Undo: Roll back to a previous turn
//! - Interrupt: Cancel the current turn mid-execution
//! - Compaction: Summarize old turns to save context
//! - Resume: Continue from a saved checkpoint
use std::collections::{HashMap, HashSet};
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use uuid::Uuid;
use crate::llm::{ChatMessage, ToolCall};
/// A session containing one or more threads.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Session {
/// Unique session ID.
pub id: Uuid,
/// User ID that owns this session.
pub user_id: String,
/// Active thread ID.
pub active_thread: Option<Uuid>,
/// All threads in this session.
pub threads: HashMap<Uuid, Thread>,
/// When the session was created.
pub created_at: DateTime<Utc>,
/// When the session was last active.
pub last_active_at: DateTime<Utc>,
/// Session metadata.
pub metadata: serde_json::Value,
/// Tools that have been auto-approved for this session ("always approve").
#[serde(default)]
pub auto_approved_tools: HashSet<String>,
}
impl Session {
/// Create a new session.
pub fn new(user_id: impl Into<String>) -> Self {
let now = Utc::now();
Self {
id: Uuid::new_v4(),
user_id: user_id.into(),
active_thread: None,
threads: HashMap::new(),
created_at: now,
last_active_at: now,
metadata: serde_json::Value::Null,
auto_approved_tools: HashSet::new(),
}
}
/// Check if a tool has been auto-approved for this session.
pub fn is_tool_auto_approved(&self, tool_name: &str) -> bool {
self.auto_approved_tools.contains(tool_name)
}
/// Add a tool to the auto-approved set.
pub fn auto_approve_tool(&mut self, tool_name: impl Into<String>) {
self.auto_approved_tools.insert(tool_name.into());
}
/// Create a new thread in this session.
pub fn create_thread(&mut self) -> &mut Thread {
let thread = Thread::new(self.id);
let thread_id = thread.id;
self.active_thread = Some(thread_id);
self.last_active_at = Utc::now();
self.threads.entry(thread_id).or_insert(thread)
}
/// Get the active thread.
pub fn active_thread(&self) -> Option<&Thread> {
self.active_thread.and_then(|id| self.threads.get(&id))
}
/// Get the active thread mutably.
pub fn active_thread_mut(&mut self) -> Option<&mut Thread> {
self.active_thread.and_then(|id| self.threads.get_mut(&id))
}
/// Get or create the active thread.
pub fn get_or_create_thread(&mut self) -> &mut Thread {
match self.active_thread {
None => self.create_thread(),
Some(id) => {
if self.threads.contains_key(&id) {
// Safe: contains_key confirmed the entry exists.
self.threads.get_mut(&id).unwrap()
} else {
// Stale active_thread ID: create a new thread, which
// updates self.active_thread to the new thread's ID.
self.create_thread()
}
}
}
}
/// Switch to a different thread.
pub fn switch_thread(&mut self, thread_id: Uuid) -> bool {
if self.threads.contains_key(&thread_id) {
self.active_thread = Some(thread_id);
self.last_active_at = Utc::now();
true
} else {
false
}
}
}
/// State of a thread.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum ThreadState {
/// Thread is idle, waiting for input.
Idle,
/// Thread is processing a turn.
Processing,
/// Thread is waiting for user approval.
AwaitingApproval,
/// Thread has completed (no more turns expected).
Completed,
/// Thread was interrupted.
Interrupted,
}
/// Pending auth token request.
///
/// When `tool_auth` returns `awaiting_token`, the thread enters auth mode.
/// The next user message is intercepted before entering the normal pipeline
/// (no logging, no turn creation, no history) and routed directly to the
/// credential store.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PendingAuth {
/// Extension name to authenticate.
pub extension_name: String,
}
/// Pending tool approval request stored on a thread.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PendingApproval {
/// Unique request ID.
pub request_id: Uuid,
/// Tool name requiring approval.
pub tool_name: String,
/// Tool parameters (original values, used for execution).
pub parameters: serde_json::Value,
/// Redacted tool parameters (sensitive values replaced with `[REDACTED]`).
/// Used for display in approval UI, logs, and SSE broadcasts.
#[serde(default)]
pub display_parameters: serde_json::Value,
/// Description of what the tool will do.
pub description: String,
/// Tool call ID from LLM (for proper context continuation).
pub tool_call_id: String,
/// Context messages at the time of the request (to resume from).
pub context_messages: Vec<ChatMessage>,
/// Remaining tool calls from the same assistant message that were not
/// executed yet when approval was requested.
#[serde(default)]
pub deferred_tool_calls: Vec<ToolCall>,
}
/// A conversation thread within a session.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Thread {
/// Unique thread ID.
pub id: Uuid,
/// Parent session ID.
pub session_id: Uuid,
/// Current state.
pub state: ThreadState,
/// Turns in this thread.
pub turns: Vec<Turn>,
/// When the thread was created.
pub created_at: DateTime<Utc>,
/// When the thread was last updated.
pub updated_at: DateTime<Utc>,
/// Thread metadata (e.g., title, tags).
pub metadata: serde_json::Value,
/// Pending approval request (when state is AwaitingApproval).
#[serde(default)]
pub pending_approval: Option<PendingApproval>,
/// Pending auth token request (thread is in auth mode).
#[serde(default)]
pub pending_auth: Option<PendingAuth>,
}
impl Thread {
/// Create a new thread.
pub fn new(session_id: Uuid) -> Self {
let now = Utc::now();
Self {
id: Uuid::new_v4(),
session_id,
state: ThreadState::Idle,
turns: Vec::new(),
created_at: now,
updated_at: now,
metadata: serde_json::Value::Null,
pending_approval: None,
pending_auth: None,
}
}
/// Create a thread with a specific ID (for DB hydration).
pub fn with_id(id: Uuid, session_id: Uuid) -> Self {
let now = Utc::now();
Self {
id,
session_id,
state: ThreadState::Idle,
turns: Vec::new(),
created_at: now,
updated_at: now,
metadata: serde_json::Value::Null,
pending_approval: None,
pending_auth: None,
}
}
/// Get the current turn number (1-indexed for display).
pub fn turn_number(&self) -> usize {
self.turns.len() + 1
}
/// Get the last turn.
pub fn last_turn(&self) -> Option<&Turn> {
self.turns.last()
}
/// Get the last turn mutably.
pub fn last_turn_mut(&mut self) -> Option<&mut Turn> {
self.turns.last_mut()
}
/// Start a new turn with user input.
pub fn start_turn(&mut self, user_input: impl Into<String>) -> &mut Turn {
let turn_number = self.turns.len();
let turn = Turn::new(turn_number, user_input);
self.turns.push(turn);
self.state = ThreadState::Processing;
self.updated_at = Utc::now();
// turn_number was len() before push, so it's a valid index after push
&mut self.turns[turn_number]
}
/// Complete the current turn with a response.
pub fn complete_turn(&mut self, response: impl Into<String>) {
if let Some(turn) = self.turns.last_mut() {
turn.complete(response);
}
self.state = ThreadState::Idle;
self.updated_at = Utc::now();
}
/// Fail the current turn with an error.
pub fn fail_turn(&mut self, error: impl Into<String>) {
if let Some(turn) = self.turns.last_mut() {
turn.fail(error);
}
self.state = ThreadState::Idle;
self.updated_at = Utc::now();
}
/// Mark the thread as awaiting approval with pending request details.
pub fn await_approval(&mut self, pending: PendingApproval) {
self.state = ThreadState::AwaitingApproval;
self.pending_approval = Some(pending);
self.updated_at = Utc::now();
}
/// Take the pending approval (clearing it from the thread).
pub fn take_pending_approval(&mut self) -> Option<PendingApproval> {
self.pending_approval.take()
}
/// Clear pending approval and return to idle state.
pub fn clear_pending_approval(&mut self) {
self.pending_approval = None;
self.state = ThreadState::Idle;
self.updated_at = Utc::now();
}
/// Enter auth mode: next user message will be routed directly to
/// the credential store, bypassing the normal pipeline entirely.
pub fn enter_auth_mode(&mut self, extension_name: String) {
self.pending_auth = Some(PendingAuth { extension_name });
self.updated_at = Utc::now();
}
/// Take the pending auth (clearing auth mode).
pub fn take_pending_auth(&mut self) -> Option<PendingAuth> {
self.pending_auth.take()
}
/// Interrupt the current turn.
pub fn interrupt(&mut self) {
if let Some(turn) = self.turns.last_mut() {
turn.interrupt();
}
self.state = ThreadState::Interrupted;
self.updated_at = Utc::now();
}
/// Resume after interruption.
pub fn resume(&mut self) {
if self.state == ThreadState::Interrupted {
self.state = ThreadState::Idle;
self.updated_at = Utc::now();
}
}
/// Get all messages for context building.
pub fn messages(&self) -> Vec<ChatMessage> {
let mut messages = Vec::new();
for turn in &self.turns {
if turn.image_content_parts.is_empty() {
messages.push(ChatMessage::user(&turn.user_input));
} else {
messages.push(ChatMessage::user_with_parts(
&turn.user_input,
turn.image_content_parts.clone(),
));
}
if let Some(ref response) = turn.response {
messages.push(ChatMessage::assistant(response));
}
}
messages
}
/// Truncate turns to a specific count (keeping most recent).
pub fn truncate_turns(&mut self, keep: usize) {
if self.turns.len() > keep {
let drain_count = self.turns.len() - keep;
self.turns.drain(0..drain_count);
// Re-number remaining turns
for (i, turn) in self.turns.iter_mut().enumerate() {
turn.turn_number = i;
}
}
}
/// Restore thread state from a checkpoint's messages.
///
/// Clears existing turns and rebuilds from message pairs.
/// Messages should alternate: user, assistant, user, assistant...
pub fn restore_from_messages(&mut self, messages: Vec<ChatMessage>) {
self.turns.clear();
self.state = ThreadState::Idle;
// Messages alternate: user, assistant, user, assistant...
let mut iter = messages.into_iter().peekable();
let mut turn_number = 0;
while let Some(msg) = iter.next() {
if msg.role == crate::llm::Role::User {
let mut turn = Turn::new(turn_number, &msg.content);
// Check if next is assistant response
if let Some(next) = iter.peek()
&& next.role == crate::llm::Role::Assistant
{
// iter.next() is guaranteed Some after a successful peek()
if let Some(response) = iter.next() {
turn.complete(&response.content);
}
}
self.turns.push(turn);
turn_number += 1;
}
}
self.updated_at = Utc::now();
}
}
/// State of a turn.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum TurnState {
/// Turn is being processed.
Processing,
/// Turn completed successfully.
Completed,
/// Turn failed with an error.
Failed,
/// Turn was interrupted.
Interrupted,
}
/// A single turn (request/response pair) in a thread.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Turn {
/// Turn number (0-indexed).
pub turn_number: usize,
/// User input that started this turn.
pub user_input: String,
/// Agent response (if completed).
pub response: Option<String>,
/// Tool calls made during this turn.
pub tool_calls: Vec<TurnToolCall>,
/// Turn state.
pub state: TurnState,
/// When the turn started.
pub started_at: DateTime<Utc>,
/// When the turn completed.
pub completed_at: Option<DateTime<Utc>>,
/// Error message (if failed).
pub error: Option<String>,
/// Transient image content parts for multimodal LLM input.
/// Not serialized — images are only needed for the current LLM call.
/// The text description in `user_input` persists for compaction/context.
#[serde(skip)]
pub image_content_parts: Vec<crate::llm::ContentPart>,
}
impl Turn {
/// Create a new turn.
pub fn new(turn_number: usize, user_input: impl Into<String>) -> Self {
Self {
turn_number,
user_input: user_input.into(),
response: None,
tool_calls: Vec::new(),
state: TurnState::Processing,
started_at: Utc::now(),
completed_at: None,
error: None,
image_content_parts: Vec::new(),
}
}
/// Complete this turn.
pub fn complete(&mut self, response: impl Into<String>) {
self.response = Some(response.into());
self.state = TurnState::Completed;
self.completed_at = Some(Utc::now());
// Free image data — only needed for the initial LLM call, not subsequent turns
self.image_content_parts.clear();
}
/// Fail this turn.
pub fn fail(&mut self, error: impl Into<String>) {
self.error = Some(error.into());
self.state = TurnState::Failed;
self.completed_at = Some(Utc::now());
self.image_content_parts.clear();
}
/// Interrupt this turn.
pub fn interrupt(&mut self) {
self.state = TurnState::Interrupted;
self.completed_at = Some(Utc::now());
self.image_content_parts.clear();
}
/// Record a tool call.
pub fn record_tool_call(&mut self, name: impl Into<String>, params: serde_json::Value) {
self.tool_calls.push(TurnToolCall {
name: name.into(),
parameters: params,
result: None,
error: None,
});
}
/// Record tool call result.
pub fn record_tool_result(&mut self, result: serde_json::Value) {
if let Some(call) = self.tool_calls.last_mut() {
call.result = Some(result);
}
}
/// Record tool call error.
pub fn record_tool_error(&mut self, error: impl Into<String>) {
if let Some(call) = self.tool_calls.last_mut() {
call.error = Some(error.into());
}
}
}
/// Record of a tool call made during a turn.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TurnToolCall {
/// Tool name.
pub name: String,
/// Parameters passed to the tool.
pub parameters: serde_json::Value,
/// Result from the tool (if successful).
pub result: Option<serde_json::Value>,
/// Error from the tool (if failed).
pub error: Option<String>,
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_session_creation() {
let mut session = Session::new("user-123");
assert!(session.active_thread.is_none());
session.create_thread();
assert!(session.active_thread.is_some());
}
#[test]
fn test_thread_turns() {
let mut thread = Thread::new(Uuid::new_v4());
thread.start_turn("Hello");
assert_eq!(thread.state, ThreadState::Processing);
assert_eq!(thread.turns.len(), 1);
thread.complete_turn("Hi there!");
assert_eq!(thread.state, ThreadState::Idle);
assert_eq!(thread.turns[0].response, Some("Hi there!".to_string()));
}
#[test]
fn test_thread_messages() {
let mut thread = Thread::new(Uuid::new_v4());
thread.start_turn("First message");
thread.complete_turn("First response");
thread.start_turn("Second message");
thread.complete_turn("Second response");
let messages = thread.messages();
assert_eq!(messages.len(), 4);
}
#[test]
fn test_turn_tool_calls() {
let mut turn = Turn::new(0, "Test input");
turn.record_tool_call("echo", serde_json::json!({"message": "test"}));
turn.record_tool_result(serde_json::json!("test"));
assert_eq!(turn.tool_calls.len(), 1);
assert!(turn.tool_calls[0].result.is_some());
}
#[test]
fn test_restore_from_messages() {
let mut thread = Thread::new(Uuid::new_v4());
// First add some turns
thread.start_turn("Original message");
thread.complete_turn("Original response");
// Now restore from different messages
let messages = vec![
ChatMessage::user("Hello"),
ChatMessage::assistant("Hi there!"),
ChatMessage::user("How are you?"),
ChatMessage::assistant("I'm good!"),
];
thread.restore_from_messages(messages);
assert_eq!(thread.turns.len(), 2);
assert_eq!(thread.turns[0].user_input, "Hello");
assert_eq!(thread.turns[0].response, Some("Hi there!".to_string()));
assert_eq!(thread.turns[1].user_input, "How are you?");
assert_eq!(thread.turns[1].response, Some("I'm good!".to_string()));
assert_eq!(thread.state, ThreadState::Idle);
}
#[test]
fn test_restore_from_messages_incomplete_turn() {
let mut thread = Thread::new(Uuid::new_v4());
// Messages with incomplete last turn (no assistant response)
let messages = vec![
ChatMessage::user("Hello"),
ChatMessage::assistant("Hi there!"),
ChatMessage::user("How are you?"),
];
thread.restore_from_messages(messages);
assert_eq!(thread.turns.len(), 2);
assert_eq!(thread.turns[1].user_input, "How are you?");
assert!(thread.turns[1].response.is_none());
}
#[test]
fn test_enter_auth_mode() {
let mut thread = Thread::new(Uuid::new_v4());
assert!(thread.pending_auth.is_none());
thread.enter_auth_mode("telegram".to_string());
assert!(thread.pending_auth.is_some());
assert_eq!(
thread.pending_auth.as_ref().unwrap().extension_name,
"telegram"
);
}
#[test]
fn test_take_pending_auth() {
let mut thread = Thread::new(Uuid::new_v4());
thread.enter_auth_mode("notion".to_string());
let pending = thread.take_pending_auth();
assert!(pending.is_some());
assert_eq!(pending.unwrap().extension_name, "notion");
// Should be cleared after take
assert!(thread.pending_auth.is_none());
assert!(thread.take_pending_auth().is_none());
}
#[test]
fn test_pending_auth_serialization() {
let mut thread = Thread::new(Uuid::new_v4());
thread.enter_auth_mode("openai".to_string());
let json = serde_json::to_string(&thread).expect("should serialize");
assert!(json.contains("pending_auth"));
assert!(json.contains("openai"));
let restored: Thread = serde_json::from_str(&json).expect("should deserialize");
assert!(restored.pending_auth.is_some());
assert_eq!(restored.pending_auth.unwrap().extension_name, "openai");
}
#[test]
fn test_pending_auth_default_none() {
// Deserialization of old data without pending_auth should default to None
let mut thread = Thread::new(Uuid::new_v4());
thread.pending_auth = None;
let json = serde_json::to_string(&thread).expect("serialize");
// Remove the pending_auth field to simulate old data
let json = json.replace(",\"pending_auth\":null", "");
let restored: Thread = serde_json::from_str(&json).expect("should deserialize");
assert!(restored.pending_auth.is_none());
}
#[test]
fn test_thread_with_id() {
let specific_id = Uuid::new_v4();
let session_id = Uuid::new_v4();
let thread = Thread::with_id(specific_id, session_id);
assert_eq!(thread.id, specific_id);
assert_eq!(thread.session_id, session_id);
assert_eq!(thread.state, ThreadState::Idle);
assert!(thread.turns.is_empty());
}
#[test]
fn test_thread_with_id_restore_messages() {
let thread_id = Uuid::new_v4();
let session_id = Uuid::new_v4();
let mut thread = Thread::with_id(thread_id, session_id);
let messages = vec![
ChatMessage::user("Hello from DB"),
ChatMessage::assistant("Restored response"),
];
thread.restore_from_messages(messages);
assert_eq!(thread.id, thread_id);
assert_eq!(thread.turns.len(), 1);
assert_eq!(thread.turns[0].user_input, "Hello from DB");
assert_eq!(
thread.turns[0].response,
Some("Restored response".to_string())
);
}
#[test]
fn test_restore_from_messages_empty() {
let mut thread = Thread::new(Uuid::new_v4());
// Add a turn first, then restore with empty vec
thread.start_turn("hello");
thread.complete_turn("hi");
assert_eq!(thread.turns.len(), 1);
thread.restore_from_messages(Vec::new());
// Should clear all turns and stay idle
assert!(thread.turns.is_empty());
assert_eq!(thread.state, ThreadState::Idle);
}
#[test]
fn test_restore_from_messages_only_assistant_messages() {
let mut thread = Thread::new(Uuid::new_v4());
// Only assistant messages (no user messages to anchor turns)
let messages = vec![
ChatMessage::assistant("I'm here"),
ChatMessage::assistant("Still here"),
];
thread.restore_from_messages(messages);
// Assistant-only messages have no user turn to attach to, so
// they should be skipped entirely.
assert!(thread.turns.is_empty());
}
#[test]
fn test_restore_from_messages_multiple_user_messages_in_a_row() {
let mut thread = Thread::new(Uuid::new_v4());
// Two user messages with no assistant response between them
let messages = vec![
ChatMessage::user("first"),
ChatMessage::user("second"),
ChatMessage::assistant("reply to second"),
];
thread.restore_from_messages(messages);
// First user message becomes a turn with no response,
// second user message pairs with the assistant response.
assert_eq!(thread.turns.len(), 2);
assert_eq!(thread.turns[0].user_input, "first");
assert!(thread.turns[0].response.is_none());
assert_eq!(thread.turns[1].user_input, "second");
assert_eq!(
thread.turns[1].response,
Some("reply to second".to_string())
);
}
#[test]
fn test_thread_switch() {
let mut session = Session::new("user-1");
let t1_id = session.create_thread().id;
let t2_id = session.create_thread().id;
// After creating two threads, active should be the last one
assert_eq!(session.active_thread, Some(t2_id));
// Switch back to the first
assert!(session.switch_thread(t1_id));
assert_eq!(session.active_thread, Some(t1_id));
// Switching to a nonexistent thread should fail
let fake_id = Uuid::new_v4();
assert!(!session.switch_thread(fake_id));
// Active thread should remain unchanged
assert_eq!(session.active_thread, Some(t1_id));
}
#[test]
fn test_get_or_create_thread_idempotent() {
let mut session = Session::new("user-1");
let tid1 = session.get_or_create_thread().id;
let tid2 = session.get_or_create_thread().id;
// Should return the same thread (not create a new one each time)
assert_eq!(tid1, tid2);
assert_eq!(session.threads.len(), 1);
}
#[test]
fn test_truncate_turns() {
let mut thread = Thread::new(Uuid::new_v4());
for i in 0..5 {
thread.start_turn(format!("msg-{}", i));
thread.complete_turn(format!("resp-{}", i));
}
assert_eq!(thread.turns.len(), 5);
thread.truncate_turns(3);
assert_eq!(thread.turns.len(), 3);
// Should keep the most recent turns
assert_eq!(thread.turns[0].user_input, "msg-2");
assert_eq!(thread.turns[1].user_input, "msg-3");
assert_eq!(thread.turns[2].user_input, "msg-4");
// Turn numbers should be re-indexed
assert_eq!(thread.turns[0].turn_number, 0);
assert_eq!(thread.turns[1].turn_number, 1);
assert_eq!(thread.turns[2].turn_number, 2);
}
#[test]
fn test_truncate_turns_noop_when_fewer() {
let mut thread = Thread::new(Uuid::new_v4());
thread.start_turn("only one");
thread.complete_turn("response");
thread.truncate_turns(10);
assert_eq!(thread.turns.len(), 1);
assert_eq!(thread.turns[0].user_input, "only one");
}
#[test]
fn test_thread_interrupt_and_resume() {
let mut thread = Thread::new(Uuid::new_v4());
thread.start_turn("do something");
assert_eq!(thread.state, ThreadState::Processing);
thread.interrupt();
assert_eq!(thread.state, ThreadState::Interrupted);
let last_turn = thread.last_turn().unwrap();
assert_eq!(last_turn.state, TurnState::Interrupted);
assert!(last_turn.completed_at.is_some());
thread.resume();
assert_eq!(thread.state, ThreadState::Idle);
}
#[test]
fn test_resume_only_from_interrupted() {
let mut thread = Thread::new(Uuid::new_v4());
// Idle thread: resume should be a no-op
assert_eq!(thread.state, ThreadState::Idle);
thread.resume();
assert_eq!(thread.state, ThreadState::Idle);
// Processing thread: resume should not change state
thread.start_turn("work");
assert_eq!(thread.state, ThreadState::Processing);
thread.resume();
assert_eq!(thread.state, ThreadState::Processing);
}
#[test]
fn test_turn_fail() {
let mut thread = Thread::new(Uuid::new_v4());
thread.start_turn("risky operation");
thread.fail_turn("connection timed out");
assert_eq!(thread.state, ThreadState::Idle);
let turn = thread.last_turn().unwrap();
assert_eq!(turn.state, TurnState::Failed);
assert_eq!(turn.error, Some("connection timed out".to_string()));
assert!(turn.response.is_none());
assert!(turn.completed_at.is_some());
}
#[test]
fn test_messages_with_incomplete_last_turn() {
let mut thread = Thread::new(Uuid::new_v4());
thread.start_turn("first");
thread.complete_turn("first reply");
thread.start_turn("second (in progress)");
let messages = thread.messages();
// Should have 3 messages: user, assistant, user (no assistant for in-progress)
assert_eq!(messages.len(), 3);
assert_eq!(messages[0].content, "first");
assert_eq!(messages[1].content, "first reply");
assert_eq!(messages[2].content, "second (in progress)");
}
#[test]
fn test_thread_serialization_round_trip() {
let mut thread = Thread::new(Uuid::new_v4());
thread.start_turn("hello");
thread.complete_turn("world");
let json = serde_json::to_string(&thread).unwrap();
let restored: Thread = serde_json::from_str(&json).unwrap();
assert_eq!(restored.id, thread.id);
assert_eq!(restored.session_id, thread.session_id);
assert_eq!(restored.turns.len(), 1);
assert_eq!(restored.turns[0].user_input, "hello");
assert_eq!(restored.turns[0].response, Some("world".to_string()));
}
#[test]
fn test_session_serialization_round_trip() {
let mut session = Session::new("user-ser");
session.create_thread();
session.auto_approve_tool("echo");
let json = serde_json::to_string(&session).unwrap();
let restored: Session = serde_json::from_str(&json).unwrap();
assert_eq!(restored.user_id, "user-ser");
assert_eq!(restored.threads.len(), 1);
assert!(restored.is_tool_auto_approved("echo"));
assert!(!restored.is_tool_auto_approved("shell"));
}
#[test]
fn test_auto_approved_tools() {
let mut session = Session::new("user-1");
assert!(!session.is_tool_auto_approved("shell"));
session.auto_approve_tool("shell");
assert!(session.is_tool_auto_approved("shell"));
// Idempotent
session.auto_approve_tool("shell");
assert_eq!(session.auto_approved_tools.len(), 1);
}
#[test]
fn test_turn_tool_call_error() {
let mut turn = Turn::new(0, "test");
turn.record_tool_call("http", serde_json::json!({"url": "example.com"}));
turn.record_tool_error("timeout");
assert_eq!(turn.tool_calls.len(), 1);
assert_eq!(turn.tool_calls[0].error, Some("timeout".to_string()));
assert!(turn.tool_calls[0].result.is_none());
}
#[test]
fn test_turn_number_increments() {
let mut thread = Thread::new(Uuid::new_v4());
// Before any turns, turn_number() is 1 (1-indexed for display)
assert_eq!(thread.turn_number(), 1);
thread.start_turn("first");
thread.complete_turn("done");
assert_eq!(thread.turn_number(), 2);
thread.start_turn("second");
assert_eq!(thread.turn_number(), 3);
}
#[test]
fn test_complete_turn_on_empty_thread() {
let mut thread = Thread::new(Uuid::new_v4());
// Completing a turn when there are no turns should be a safe no-op
thread.complete_turn("phantom response");
assert_eq!(thread.state, ThreadState::Idle);
assert!(thread.turns.is_empty());
}
#[test]
fn test_fail_turn_on_empty_thread() {
let mut thread = Thread::new(Uuid::new_v4());
// Failing a turn when there are no turns should be a safe no-op
thread.fail_turn("phantom error");
assert_eq!(thread.state, ThreadState::Idle);
assert!(thread.turns.is_empty());
}
#[test]
fn test_pending_approval_flow() {
let mut thread = Thread::new(Uuid::new_v4());
let approval = PendingApproval {
request_id: Uuid::new_v4(),
tool_name: "shell".to_string(),
parameters: serde_json::json!({"command": "rm -rf /"}),
display_parameters: serde_json::json!({"command": "rm -rf /"}),
description: "dangerous command".to_string(),
tool_call_id: "call_123".to_string(),
context_messages: vec![ChatMessage::user("do it")],
deferred_tool_calls: vec![],
};
thread.await_approval(approval);
assert_eq!(thread.state, ThreadState::AwaitingApproval);
assert!(thread.pending_approval.is_some());
let taken = thread.take_pending_approval();
assert!(taken.is_some());
assert_eq!(taken.unwrap().tool_name, "shell");
assert!(thread.pending_approval.is_none());
}
#[test]
fn test_clear_pending_approval() {
let mut thread = Thread::new(Uuid::new_v4());
let approval = PendingApproval {
request_id: Uuid::new_v4(),
tool_name: "http".to_string(),
parameters: serde_json::json!({}),
display_parameters: serde_json::json!({}),
description: "test".to_string(),
tool_call_id: "call_456".to_string(),
context_messages: vec![],
deferred_tool_calls: vec![],
};
thread.await_approval(approval);
thread.clear_pending_approval();
assert_eq!(thread.state, ThreadState::Idle);
assert!(thread.pending_approval.is_none());
}
#[test]
fn test_active_thread_accessors() {
let mut session = Session::new("user-1");
assert!(session.active_thread().is_none());
assert!(session.active_thread_mut().is_none());
let tid = session.create_thread().id;
assert!(session.active_thread().is_some());
assert_eq!(session.active_thread().unwrap().id, tid);
// Mutably modify through accessor
session.active_thread_mut().unwrap().start_turn("test");
assert_eq!(
session.active_thread().unwrap().state,
ThreadState::Processing
);
}
}