bamboo_agent_core/agent/

events.rs

//! Agent event system for real-time streaming.
//!
//! This module defines the event types emitted during agent execution,
//! which are streamed to clients via Server-Sent Events (SSE).
//!
//! # Event Types
//!
//! - [`AgentEvent`] - All possible agent execution events
//! - [`TokenUsage`] - Token consumption statistics
//! - [`TokenBudgetUsage`] - Detailed token budget information
//!
//! # Event Flow
//!
//! 1. **Token** events stream generated text
//! 2. **ToolStart/ToolComplete** track tool execution
//! 3. **TaskListUpdated** tracks progress
//! 4. **TokenBudgetUpdated** reports context management
//! 5. **Complete**, **Cancelled**, or **Error** ends the stream
//!
//! # Example
//!
//! ```javascript
//! const eventSource = new EventSource('/api/v1/events/session-id');
//! eventSource.onmessage = (event) => {
//!   const data = JSON.parse(event.data);
//!   switch (data.type) {
//!     case 'token':
//!       console.log('Token:', data.content);
//!       break;
//!     case 'complete':
//!       console.log('Done!');
//!       eventSource.close();
//!       break;
//!   }
//! };
//! ```

use crate::tools::ToolResult;
use bamboo_domain::{TaskItemStatus, TaskList};
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};

/// Represents events emitted during agent execution.
///
/// These events are streamed to clients via SSE to provide real-time
/// feedback on agent progress, tool execution, and completion.
///
/// # Variants
///
/// ## Text Generation
/// - `Token` - Streaming text token
/// - `ReasoningToken` - Streaming reasoning/thinking token (separate channel)
///
/// ## Tool Execution
/// - `ToolStart` - Tool execution started
/// - `ToolComplete` - Tool finished successfully
/// - `ToolError` - Tool execution failed
///
/// ## User Interaction
/// - `NeedClarification` - Agent needs user input
///
/// ## Progress Tracking
/// - `TaskListUpdated` - Task list created or modified
/// - `TaskListItemProgress` - Individual item progress
/// - `TaskListCompleted` - All items completed
/// - `TaskEvaluationStarted` - Task evaluation began
/// - `TaskEvaluationCompleted` - Task evaluation finished
///
/// ## Context Management
/// - `TokenBudgetUpdated` - Context budget changed
/// - `ContextCompressionStatus` - Context compression lifecycle progress
/// - `ContextSummarized` - Old messages summarized
///
/// ## Sub-agents (Async Spawn)
/// - `SubAgentStarted` - A child session is created and scheduled to run
/// - `SubAgentEvent` - Forwarded raw child event (full fidelity)
/// - `SubAgentHeartbeat` - Periodic heartbeat while the child is running
/// - `SubAgentCompleted` - Child session finished (completed/cancelled/error)
///
/// ## Terminal Events
/// - `Complete` - Execution finished successfully
/// - `Cancelled` - Execution was cancelled by the user
/// - `Error` - Execution failed
///
/// # Serialization
///
/// Events are serialized as JSON with a `type` field for discrimination:
/// ```json
/// {"type": "token", "content": "Hello"}
/// {"type": "complete", "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15}}
/// {"type": "cancelled", "message": "Agent execution cancelled by user"}
/// ```
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum AgentEvent {
    /// Text token generated by the LLM.
    Token {
        /// Generated text content
        content: String,
    },

    /// Reasoning/thinking token generated by the LLM.
    ///
    /// This is streamed separately from assistant answer tokens so the UI can
    /// choose whether and how to display model reasoning traces.
    ReasoningToken {
        /// Generated reasoning content
        content: String,
    },

    /// Streaming output emitted while a specific tool call is running.
    ///
    /// This is used to render "live output" inside a tool-call card in the UI
    /// without mixing tool output into the assistant's main token stream.
    ToolToken {
        /// Tool call identifier that this output belongs to.
        tool_call_id: String,
        /// Output chunk.
        content: String,
    },

    /// Tool execution started.
    ToolStart {
        /// Unique tool call identifier
        tool_call_id: String,
        /// Name of the tool being executed
        tool_name: String,
        /// Tool arguments (JSON)
        arguments: serde_json::Value,
    },

    /// Tool execution completed successfully.
    ToolComplete {
        /// Tool call identifier
        tool_call_id: String,
        /// Tool execution result
        result: ToolResult,
    },

    /// Tool execution failed.
    ToolError {
        /// Tool call identifier
        tool_call_id: String,
        /// Error message
        error: String,
    },

    /// Structured lifecycle event for tool execution tracking.
    ///
    /// These events complement `ToolStart`/`ToolComplete`/`ToolError` with
    /// richer metadata (mutability, auto-approval, wall-clock timing) and
    /// are emitted by `ToolEmitter` (in `bamboo-agent-tools`).
    ToolLifecycle {
        /// Tool call identifier
        tool_call_id: String,
        /// Canonical tool name
        tool_name: String,
        /// Lifecycle phase: "begin", "finished", "error", "cancelled"
        phase: String,
        /// Wall-clock milliseconds since the call began (None for begin)
        #[serde(skip_serializing_if = "Option::is_none")]
        elapsed_ms: Option<u64>,
        /// Whether the tool mutates state (writes files, runs commands)
        is_mutating: bool,
        /// Whether execution was auto-approved (no user prompt needed)
        auto_approved: bool,
        /// Human-readable summary
        #[serde(skip_serializing_if = "Option::is_none")]
        summary: Option<String>,
        /// Error message (if phase == "error")
        #[serde(skip_serializing_if = "Option::is_none")]
        error: Option<String>,
    },

    /// Agent needs clarification from the user.
    NeedClarification {
        /// Question to ask the user
        question: String,
        /// Optional predefined options
        options: Option<Vec<String>>,
        /// Tool call identifier that triggered this clarification
        #[serde(default, skip_serializing_if = "Option::is_none")]
        tool_call_id: Option<String>,
        /// Whether the user can provide a free-text response
        #[serde(default = "default_allow_custom")]
        allow_custom: bool,
    },

    /// Emitted when task list is created or updated.
    TaskListUpdated {
        /// Current task list state.
        task_list: TaskList,
    },

    /// Emitted when a task item makes progress (delta update).
    TaskListItemProgress {
        /// Session identifier
        session_id: String,
        /// Item identifier
        item_id: String,
        /// New item status
        status: TaskItemStatus,
        /// Number of tool calls made
        tool_calls_count: usize,
        /// Item version (for optimistic concurrency)
        version: u64,
    },

    /// Emitted when all task items are completed.
    TaskListCompleted {
        /// Session identifier
        session_id: String,
        /// Completion timestamp
        completed_at: DateTime<Utc>,
        /// Total agent rounds executed
        total_rounds: u32,
        /// Total tool calls made
        total_tool_calls: usize,
    },

    /// Emitted when task evaluation starts.
    TaskEvaluationStarted {
        /// Session identifier
        session_id: String,
        /// Number of items to evaluate
        items_count: usize,
    },

    /// Emitted when task evaluation completes.
    TaskEvaluationCompleted {
        /// Session identifier
        session_id: String,
        /// Number of items updated
        updates_count: usize,
        /// Evaluation reasoning
        reasoning: String,
    },

    /// Emitted when token budget is prepared (after context truncation)
    TokenBudgetUpdated {
        /// Token budget details
        usage: TokenBudgetUsage,
    },

    /// Emitted when host-side context compression lifecycle changes.
    ContextCompressionStatus {
        /// Compression phase label (for example: pre-turn, mid-turn).
        phase: String,
        /// Compression status: started | completed | failed | skipped
        status: String,
    },

    /// Emitted when conversation context is summarized
    ContextSummarized {
        /// Generated summary text
        summary: String,
        /// Number of old messages summarized
        messages_summarized: usize,
        /// Tokens saved by summarization
        tokens_saved: u32,
        /// Context usage percentage before compression
        #[serde(default)]
        usage_before_percent: f64,
        /// Context usage percentage after compression
        #[serde(default)]
        usage_after_percent: f64,
        /// What triggered the compression: "auto" | "manual" | "critical"
        #[serde(default)]
        trigger_type: String,
    },

    /// Emitted when context pressure reaches warning or critical levels.
    /// Frontend should display this to the user as a proactive notification.
    ContextPressureNotification {
        /// Context usage as a percentage of the context window.
        percent: f64,
        /// Severity level: "warning" (70%) or "critical" (90%).
        level: String,
        /// Human-readable message describing the pressure state.
        message: String,
    },

    /// A child session was spawned from a parent session (async background job).
    SubAgentStarted {
        parent_session_id: String,
        child_session_id: String,
        /// Optional title (useful for UI lists).
        #[serde(default, skip_serializing_if = "Option::is_none")]
        title: Option<String>,
    },

    /// Forwarded raw child event to the parent session stream.
    ///
    /// Child sessions are not allowed to spawn further sessions, so this should not nest.
    SubAgentEvent {
        parent_session_id: String,
        child_session_id: String,
        event: Box<AgentEvent>,
    },

    /// Heartbeat emitted while a child session is running.
    SubAgentHeartbeat {
        parent_session_id: String,
        child_session_id: String,
        timestamp: DateTime<Utc>,
    },

    /// Child session finished (completed/cancelled/error).
    SubAgentCompleted {
        parent_session_id: String,
        child_session_id: String,
        /// One of: "completed" | "cancelled" | "error" | "skipped"
        status: String,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        error: Option<String>,
    },

    /// Plan mode was entered.
    PlanModeEntered {
        /// Session identifier
        session_id: String,
        /// Optional reason for entering plan mode
        #[serde(default, skip_serializing_if = "Option::is_none")]
        reason: Option<String>,
        /// Previous permission mode before entering plan mode
        pre_permission_mode: String,
        /// RFC3339 timestamp when plan mode was entered.
        entered_at: chrono::DateTime<chrono::Utc>,
        /// Current plan mode phase/status.
        status: bamboo_domain::PlanModeStatus,
        /// Path to the persisted plan file, if already available.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        plan_file_path: Option<String>,
    },

    /// Plan mode was exited.
    PlanModeExited {
        /// Session identifier
        session_id: String,
        /// Whether the exit was approved by the user
        approved: bool,
        /// The permission mode restored after exiting
        restored_mode: String,
        /// Plan content that was reviewed, if any
        #[serde(default, skip_serializing_if = "Option::is_none")]
        plan: Option<String>,
    },

    /// Plan file was updated.
    PlanFileUpdated {
        /// Session identifier
        session_id: String,
        /// Path to the plan file
        file_path: String,
        /// Summary of the plan content (truncated)
        content_summary: String,
    },

    /// Runner progress update emitted at the start of each agent turn.
    ///
    /// Used to track live execution progress (round count, current activity)
    /// for diagnostic visibility, especially for child sessions.
    RunnerProgress {
        /// Session identifier
        session_id: String,
        /// Current turn/round count
        round_count: u32,
    },

    /// Session title was updated (auto-generated by backend or manually renamed via PATCH).
    SessionTitleUpdated {
        session_id: String,
        title: String,
        title_version: u64,
        source: TitleSource,
        updated_at: chrono::DateTime<chrono::Utc>,
    },

    /// Session pinned flag was toggled via PATCH.
    ///
    /// Replayable metadata event. `pinned` is an idempotent boolean so the
    /// latest event wins; `updated_at` is used by the frontend to suppress
    /// stale replays.
    SessionPinnedUpdated {
        session_id: String,
        pinned: bool,
        updated_at: chrono::DateTime<chrono::Utc>,
    },

    /// Execution run has started and the runner is now active.
    ///
    /// Emitted as the first event after a runner reservation succeeds,
    /// before any token or tool events. Carries the `run_id` so the
    /// frontend can correlate subsequent SSE events across reconnects.
    ExecutionStarted {
        /// Unique identifier for this execution run.
        run_id: String,
        /// Session identifier.
        session_id: String,
        /// ISO 8601 timestamp when the run started.
        started_at: String,
    },

    /// Tool execution requires user approval before proceeding.
    ///
    /// Emitted when a permission checker determines that a tool call needs
    /// explicit user confirmation (e.g., mutating operations in restricted
    /// permission mode). The frontend should present the approval request and
    /// either grant or deny it.
    ToolApprovalRequested {
        /// Unique identifier for the tool call awaiting approval.
        tool_call_id: String,
        /// Name of the tool being executed.
        tool_name: String,
        /// Parameters that were passed to the tool.
        parameters: serde_json::Value,
    },

    /// Agent execution completed successfully.
    Complete {
        /// Final token usage statistics
        usage: TokenUsage,
    },

    /// Agent execution was cancelled.
    Cancelled {
        /// Optional human-readable message explaining the cancellation.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        message: Option<String>,
    },

    /// Agent execution failed.
    Error {
        /// Error message
        message: String,
    },
}

fn default_allow_custom() -> bool {
    true
}

/// Source that triggered a session title update.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum TitleSource {
    Auto,
    Manual,
    Fallback,
}

/// Re-exported shared token usage type.
///
/// See [`bamboo_domain::TokenUsage`] for the canonical definition.
pub use bamboo_domain::TokenUsage;

pub use bamboo_domain::budget_types::TokenBudgetUsage;

#[cfg(test)]
mod tests {
    use super::*;
    use bamboo_domain::{TaskItem, TaskItemStatus, TaskList};

    fn sample_task_list() -> TaskList {
        TaskList {
            session_id: "session-1".to_string(),
            title: "Task List".to_string(),
            items: vec![TaskItem {
                id: "task_1".to_string(),
                description: "Implement event rename".to_string(),
                status: TaskItemStatus::InProgress,
                depends_on: Vec::new(),
                notes: "Implementing".to_string(),
                ..TaskItem::default()
            }],
            created_at: Utc::now(),
            updated_at: Utc::now(),
        }
    }

    #[test]
    fn task_list_updated_serializes_with_task_names() {
        let event = AgentEvent::TaskListUpdated {
            task_list: sample_task_list(),
        };

        let value = serde_json::to_value(event).expect("event should serialize");
        assert_eq!(value["type"], "task_list_updated");
        assert!(value.get("task_list").is_some());
        assert!(value.get("todo_list").is_none());
    }

    #[test]
    fn cancelled_serializes_with_snake_case_type() {
        let event = AgentEvent::Cancelled {
            message: Some("Agent execution cancelled by user".to_string()),
        };

        let value = serde_json::to_value(event).expect("event should serialize");
        assert_eq!(value["type"], "cancelled");
        assert_eq!(
            value["message"],
            serde_json::Value::String("Agent execution cancelled by user".to_string())
        );
    }

    #[test]
    fn task_evaluation_completed_serializes_with_task_type() {
        let event = AgentEvent::TaskEvaluationCompleted {
            session_id: "session-1".to_string(),
            updates_count: 2,
            reasoning: "Updated statuses".to_string(),
        };

        let value = serde_json::to_value(event).expect("event should serialize");
        assert_eq!(value["type"], "task_evaluation_completed");
    }

    #[test]
    fn context_compression_status_serializes_with_phase_and_status() {
        let event = AgentEvent::ContextCompressionStatus {
            phase: "mid-turn".to_string(),
            status: "started".to_string(),
        };

        let value = serde_json::to_value(event).expect("event should serialize");
        assert_eq!(value["type"], "context_compression_status");
        assert_eq!(value["phase"], "mid-turn");
        assert_eq!(value["status"], "started");
    }

    #[test]
    fn need_clarification_serializes_with_new_fields() {
        let event = AgentEvent::NeedClarification {
            question: "Continue?".to_string(),
            options: Some(vec!["Yes".to_string(), "No".to_string()]),
            tool_call_id: Some("tool-1".to_string()),
            allow_custom: false,
        };

        let value = serde_json::to_value(event).expect("event should serialize");
        assert_eq!(value["type"], "need_clarification");
        assert_eq!(value["question"], "Continue?");
        assert_eq!(value["options"], serde_json::json!(["Yes", "No"]));
        assert_eq!(value["tool_call_id"], "tool-1");
        assert_eq!(value["allow_custom"], false);
    }

    #[test]
    fn need_clarification_deserializes_from_old_format_without_new_fields() {
        let json = serde_json::json!({
            "type": "need_clarification",
            "question": "Continue?",
            "options": ["Yes", "No"]
        });

        let event: AgentEvent =
            serde_json::from_value(json).expect("should deserialize old format");
        match event {
            AgentEvent::NeedClarification {
                question,
                options,
                tool_call_id,
                allow_custom,
            } => {
                assert_eq!(question, "Continue?");
                assert_eq!(options, Some(vec!["Yes".to_string(), "No".to_string()]));
                assert_eq!(tool_call_id, None);
                assert!(allow_custom); // default_allow_custom returns true
            }
            other => panic!("unexpected event: {other:?}"),
        }
    }

    #[test]
    fn need_clarification_deserializes_with_allow_custom_false() {
        let json = serde_json::json!({
            "type": "need_clarification",
            "question": "Pick one",
            "allow_custom": false
        });

        let event: AgentEvent = serde_json::from_value(json).expect("should deserialize");
        match event {
            AgentEvent::NeedClarification {
                question,
                options,
                tool_call_id,
                allow_custom,
            } => {
                assert_eq!(question, "Pick one");
                assert_eq!(options, None);
                assert_eq!(tool_call_id, None);
                assert!(!allow_custom);
            }
            other => panic!("unexpected event: {other:?}"),
        }
    }

    #[test]
    fn plan_mode_entered_serializes_correctly() {
        let entered_at = Utc::now();
        let event = AgentEvent::PlanModeEntered {
            session_id: "sess-1".to_string(),
            reason: Some("Complex refactor".to_string()),
            pre_permission_mode: "default".to_string(),
            entered_at,
            status: bamboo_domain::PlanModeStatus::Exploring,
            plan_file_path: None,
        };

        let value = serde_json::to_value(event).expect("event should serialize");
        assert_eq!(value["type"], "plan_mode_entered");
        assert_eq!(value["session_id"], "sess-1");
        assert_eq!(value["reason"], "Complex refactor");
        assert_eq!(value["pre_permission_mode"], "default");
        assert_eq!(value["status"], "exploring");
        assert_eq!(value["entered_at"], entered_at.to_rfc3339());
    }

    #[test]
    fn plan_mode_exited_serializes_correctly() {
        let event = AgentEvent::PlanModeExited {
            session_id: "sess-1".to_string(),
            approved: true,
            restored_mode: "accept_edits".to_string(),
            plan: Some("# Plan\n1. Step one".to_string()),
        };

        let value = serde_json::to_value(event).expect("event should serialize");
        assert_eq!(value["type"], "plan_mode_exited");
        assert_eq!(value["session_id"], "sess-1");
        assert_eq!(value["approved"], true);
        assert_eq!(value["restored_mode"], "accept_edits");
        assert_eq!(value["plan"], "# Plan\n1. Step one");
    }

    #[test]
    fn plan_file_updated_serializes_correctly() {
        let event = AgentEvent::PlanFileUpdated {
            session_id: "sess-1".to_string(),
            file_path: "/tmp/plans/sess-1.md".to_string(),
            content_summary: "Implementation plan for feature X".to_string(),
        };

        let value = serde_json::to_value(event).expect("event should serialize");
        assert_eq!(value["type"], "plan_file_updated");
        assert_eq!(value["session_id"], "sess-1");
        assert_eq!(value["file_path"], "/tmp/plans/sess-1.md");
        assert_eq!(
            value["content_summary"],
            "Implementation plan for feature X"
        );
    }

    #[test]
    fn tool_approval_requested_serializes_correctly() {
        let event = AgentEvent::ToolApprovalRequested {
            tool_call_id: "call-abc".to_string(),
            tool_name: "Write".to_string(),
            parameters: serde_json::json!({"file_path": "/tmp/test.txt"}),
        };

        let value = serde_json::to_value(event).expect("event should serialize");
        assert_eq!(value["type"], "tool_approval_requested");
        assert_eq!(value["tool_call_id"], "call-abc");
        assert_eq!(value["tool_name"], "Write");
        assert_eq!(
            value["parameters"],
            serde_json::json!({"file_path": "/tmp/test.txt"})
        );
    }

    #[test]
    fn tool_approval_requested_deserializes_correctly() {
        let json = serde_json::json!({
            "type": "tool_approval_requested",
            "tool_call_id": "call-xyz",
            "tool_name": "Bash",
            "parameters": {"command": "ls -la"}
        });

        let event: AgentEvent = serde_json::from_value(json).expect("should deserialize");
        match event {
            AgentEvent::ToolApprovalRequested {
                tool_call_id,
                tool_name,
                parameters,
            } => {
                assert_eq!(tool_call_id, "call-xyz");
                assert_eq!(tool_name, "Bash");
                assert_eq!(parameters, serde_json::json!({"command": "ls -la"}));
            }
            other => panic!("unexpected event: {other:?}"),
        }
    }

    #[test]
    fn session_title_updated_round_trips_with_source_variants() {
        use chrono::Utc;
        let event = AgentEvent::SessionTitleUpdated {
            session_id: "sess-1".to_string(),
            title: "My title".to_string(),
            title_version: 3,
            source: TitleSource::Auto,
            updated_at: Utc::now(),
        };
        let json = serde_json::to_string(&event).unwrap();
        assert!(
            json.contains("\"type\":\"session_title_updated\""),
            "json: {json}"
        );
        assert!(json.contains("\"source\":\"auto\""), "json: {json}");
        let _decoded: AgentEvent = serde_json::from_str(&json).unwrap();
    }

    #[test]
    fn plan_mode_events_deserialize_without_optional_fields() {
        let json = serde_json::json!({
            "type": "plan_mode_entered",
            "session_id": "sess-1",
            "pre_permission_mode": "default",
            "entered_at": "2025-01-01T00:00:00Z",
            "status": "exploring"
        });

        let event: AgentEvent = serde_json::from_value(json).expect("should deserialize");
        match event {
            AgentEvent::PlanModeEntered {
                session_id,
                reason,
                pre_permission_mode,
                entered_at,
                status,
                plan_file_path,
            } => {
                assert_eq!(session_id, "sess-1");
                assert_eq!(reason, None);
                assert_eq!(pre_permission_mode, "default");
                assert_eq!(entered_at.to_rfc3339(), "2025-01-01T00:00:00+00:00");
                assert_eq!(status, bamboo_domain::PlanModeStatus::Exploring);
                assert_eq!(plan_file_path, None);
            }
            other => panic!("unexpected event: {other:?}"),
        }
    }
}