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::{
    AgentHookPoint, HookResult, PendingQuestionSource, TaskItem, TaskItemStatus, TaskList,
};
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};

fn default_title_generated() -> bool {
    true
}

/// 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
/// - `GoldEvaluationStarted` - Gold observe-only evaluation began
/// - `GoldEvaluationCompleted` - Gold observe-only evaluation finished
///
/// ## Context Management
/// - `TokenBudgetUpdated` - Context budget changed
/// - `ContextCompressionStatus` - Context compression lifecycle progress
/// - `ContextSummarized` - Old messages summarized
/// - `ContextArchived` - Exact old messages excluded from the active window
///
/// ## Sub-agents (Async Spawn)
/// - `SubAgentStarted` - A child session is created and scheduled to run
/// - `SubAgentEvent` - Legacy parent projection of a raw child event; current
///   runtimes publish full fidelity on the child's own session channel
/// - `SubAgentHeartbeat` - Periodic heartbeat while the child is running
/// - `SubAgentCompleted` - Child session finished (completed/cancelled/error)
///
/// ## Resource Guardrails
/// - `BudgetExceeded` - A per-run token/tool-call/subagent budget tripped and
///   the run was gracefully stopped (issue #221)
///
/// ## 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,
    },

    /// Internal marker that binds subsequent visible [`Token`](Self::Token)
    /// events to one stable assistant message. Event forwarders MUST consume
    /// this variant locally: it is not part of the generic agent/account stream
    /// and must never be serialized to clients as a raw `AgentEvent`.
    #[doc(hidden)]
    VisibleMessageStart {
        /// Stable id reused by the final persisted assistant [`Message`].
        message_id: String,
        /// Timestamp reused by the final persisted assistant [`Message`].
        created_at: DateTime<Utc>,
    },

    /// Internal rollback for a visible message whose provider attempt is being
    /// retried. Like [`VisibleMessageStart`](Self::VisibleMessageStart), this is
    /// consumed by event forwarders and never crosses a public generic stream.
    #[doc(hidden)]
    VisibleMessageDiscard { message_id: 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>,
    },

    /// A registered lifecycle hook completed at an engine seam.
    HookLifecycle {
        /// Stable hook name supplied by the hook implementation.
        hook_name: String,
        /// Engine seam where the hook ran.
        point: AgentHookPoint,
        /// Lifecycle phase. Currently `completed`; kept explicit so future
        /// start/error events remain schema-compatible.
        phase: String,
        /// Wall-clock time spent inside the hook.
        duration_ms: u64,
        /// Decision returned by the hook.
        decision: HookResult,
    },

    /// 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>,
        /// Tool name that triggered this clarification, when known.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        tool_name: Option<String>,
        /// Whether the user can provide a free-text response
        #[serde(default = "default_allow_custom")]
        allow_custom: bool,
        /// Origin of the pending question, when known.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        source: Option<PendingQuestionSource>,
    },

    /// Emitted when task list is created or updated.
    TaskListUpdated {
        /// Current task list state.
        task_list: TaskList,
        /// Monotonic persisted task-list generation.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        version: Option<u64>,
    },

    /// 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,
        /// Rich item projection for blocker/evidence/transition rendering.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        item: Option<TaskItem>,
    },

    /// 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,
        /// Monotonic persisted task-list generation.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        version: Option<u64>,
    },

    /// Emitted when task evaluation starts.
    TaskEvaluationStarted {
        /// Session identifier
        session_id: String,
        /// Number of items to evaluate
        items_count: usize,
        /// Task-list generation evaluated. Optional for older producers/frames.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        generation: Option<u64>,
    },

    /// Emitted when task evaluation completes.
    TaskEvaluationCompleted {
        /// Session identifier
        session_id: String,
        /// Number of items updated
        updates_count: usize,
        /// Evaluation reasoning
        reasoning: String,
        /// Task-list generation evaluated. Optional for backward compatibility.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        generation: Option<u64>,
    },

    /// Emitted when an unfinished task evaluation is stopped because its owning
    /// run completed, suspended, or was cancelled.
    TaskEvaluationCancelled {
        session_id: String,
        reason: String,
        /// Task-list generation whose evaluation was cancelled.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        generation: Option<u64>,
    },

    /// Emitted when gold observe-only evaluation starts.
    GoldEvaluationStarted {
        /// Session identifier
        session_id: String,
        /// Evaluation checkpoint
        checkpoint: GoldCheckpoint,
        /// Current iteration / round number associated with the evaluation
        iteration: u32,
    },

    /// Emitted when gold observe-only evaluation completes.
    GoldEvaluationCompleted {
        /// Session identifier
        session_id: String,
        /// Evaluation checkpoint
        checkpoint: GoldCheckpoint,
        /// Current iteration / round number associated with the evaluation
        iteration: u32,
        /// Gold decision for the current checkpoint
        decision: GoldDecision,
        /// Confidence in the decision
        confidence: GoldConfidence,
        /// Short reasoning summary
        reasoning: String,
    },

    /// Emitted when an unfinished Gold evaluation is stopped with its owning
    /// run or locally rejected before provider dispatch. The `route_rejected`
    /// reason ends only the evaluation; it does not cancel the owning run.
    GoldEvaluationCancelled { session_id: String, reason: String },

    /// Emitted whenever the runtime goal state changes — a new status
    /// (active/complete/blocked/…), an incremented continuation count, or a
    /// freshly recorded side-channel double-check verdict. Lets the UI reflect
    /// live goal progress without re-fetching history. Ephemeral: it rides only
    /// the per-session `/events/{id}` stream; reconnecting clients read the
    /// authoritative `goal_state` from the history endpoint instead.
    GoalStatusChanged {
        /// Session identifier
        session_id: String,
        /// Full serialized goal state — identical shape to the history
        /// response's `goal_state` field (see `bamboo_engine::runtime::goal_state`).
        goal_state: serde_json::Value,
    },

    /// 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 after a retrieval-window boundary is durably checkpointed.
    /// Contains structural evidence only; raw message and memory content must
    /// never enter this event.
    ContextArchived {
        archive_event_id: String,
        trigger_type: String,
        messages_archived: usize,
        groups_archived: usize,
        user_turns_archived: usize,
        active_tokens_before: u32,
        active_tokens_after: u32,
        target_tokens: u32,
        retained_recent_user_turns: usize,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        oldest_retained_message_id: Option<String>,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        oldest_retained_user_message_id: Option<String>,
        model_context_epoch: u64,
        reset_reason: 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>,
    },

    /// Legacy raw child projection on the parent session stream. Retained for
    /// wire compatibility; current runtimes publish raw events only on the
    /// child's own independently subscribable session channel.
    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>,
    },

    /// Background Bash shell finished (completed/killed/error).
    ///
    /// Emitted by the background shell runtime when a `run_in_background`
    /// command exits, so clients can react to (and, in later phases, resume
    /// around) long-running commands. Phase 1 (issue #84): completion signal
    /// only — this does not change the default foreground behavior.
    ///
    /// Delivery scope: a *live* signal. It rides the per-session
    /// `/events/{id}` stream and the in-memory late-subscriber replay cache
    /// (`is_critical_event`), but is intentionally **not** a durable change in
    /// Phase 1 — it is not written to the account change journal. Treat it as
    /// ephemeral: a reconnecting client should not rely on seeing a past
    /// `BashCompleted` via the journaled history.
    BashCompleted {
        /// Background shell session identifier (same value returned as `bash_id`).
        bash_id: String,
        /// The command string that was executed.
        command: String,
        /// Process exit code, when available (`None` for signal/killed termination).
        #[serde(default, skip_serializing_if = "Option::is_none")]
        exit_code: Option<i32>,
        /// One of: "completed" | "killed" | "error".
        status: 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,
        /// Status after persisting the plan file.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        status: Option<bamboo_domain::PlanModeStatus>,
    },

    /// 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,
    },

    /// Typed, bounded permission posture activated by an execution boundary.
    /// External workers send this once before vendor execution so the actor
    /// host can persist executor-specific mapping without relying on unknown
    /// fields attached to a generic progress event.
    PermissionPostureActivated {
        session_id: String,
        policy_revision: u64,
        requested_mode: String,
        effective_mode: String,
        executor_mapping: String,
    },

    /// Session title was updated (auto-generated by backend or manually renamed via PATCH).
    SessionTitleUpdated {
        session_id: String,
        title: String,
        title_version: u64,
        #[serde(default = "default_title_generated")]
        title_generated: bool,
        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>,
    },

    /// A new session was created.
    ///
    /// Change-feed event: durable, journaled, carried on the account `/stream`
    /// feed so other clients can insert the session into their list without a
    /// full `GET /sessions` poll.
    SessionCreated {
        session_id: String,
        /// Stable Project membership at creation time. Explicit `null` means
        /// Unassigned so replay consumers can recover grouping without a
        /// follow-up session fetch.
        #[serde(default)]
        project_id: Option<String>,
        title: String,
        kind: bamboo_domain::SessionKind,
        created_at: chrono::DateTime<chrono::Utc>,
    },

    /// A session was deleted.
    ///
    /// Change-feed event: durable, journaled. Clients remove the session from
    /// their local list on receipt.
    SessionDeleted { session_id: String },

    /// A session's message history was cleared (session kept).
    ///
    /// Change-feed event: durable, journaled. Clients drop cached messages for
    /// the session and refetch lazily.
    SessionCleared { session_id: String },

    /// A message was appended to a session.
    ///
    /// Change-feed event: durable, journaled. The `seq` assigned to this event
    /// on the account feed is the message's feed coordinate (used by
    /// `GET /history/{id}?since={seq}` to compute deltas). `content` is the
    /// plain-text body matching what `/history` returns to the UI.
    MessageAppended {
        session_id: String,
        message_id: String,
        role: bamboo_domain::Role,
        content: String,
        created_at: chrono::DateTime<chrono::Utc>,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        thread_id: Option<String>,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        in_reply_to: Option<String>,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        correlation_id: Option<String>,
    },

    /// The runtime session snapshot for a run has been saved successfully.
    ///
    /// `Complete` ends the low-latency token stream and may be observed before
    /// the final checkpoint. This durable change-feed barrier is emitted only
    /// after `/history/{id}` can read the saved assistant/tool tail, allowing
    /// clients to reconcile without guessing at persistence timing.
    SessionHistoryCommitted { session_id: String },

    /// 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,
    },

    /// A child sub-agent (out-of-process worker) hit a gated tool and proxied
    /// the approval decision to this parent over the actor protocol (Phase 2).
    /// The parent surfaces it to the human; the decision is routed back to the
    /// waiting child via
    /// `external_agents::live::deliver_approval(child_session_id, request_id, approved)`.
    ChildApprovalRequested {
        /// The child session whose gated tool is blocked awaiting approval.
        child_session_id: String,
        /// Correlates the eventual approve/deny reply back to the blocked tool.
        request_id: String,
        /// Name of the gated tool the child wants to run.
        tool_name: String,
        /// Human-readable description of the permission requested.
        permission: String,
        /// The concrete resource the action targets.
        resource: String,
    },

    /// Durable, versioned approval lifecycle delta for a child agent.
    ChildApprovalChanged {
        parent_session_id: String,
        child_session_id: String,
        /// Execution attempt that produced this approval. Older persisted
        /// events predate attempt tracking and deserialize as attempt zero.
        #[serde(default)]
        child_attempt: u32,
        request_id: String,
        version: u64,
        /// `pending` | `approved` | `denied` | `expired` | `delivery_failed`.
        status: String,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        reason: Option<String>,
        tool_name: String,
        permission: String,
        resource: String,
        created_at: String,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        resolved_at: Option<String>,
    },

    /// A per-run resource guardrail (token / tool-call / subagent budget)
    /// tripped and the run was gracefully stopped — issue #221.
    ///
    /// Mirrors the `runtime.completion_reason = "budget_exceeded"` session
    /// metadata stamp (see `bamboo_engine::runtime::config::AgentLoopConfig`)
    /// as a structured, client-observable signal so a caller can display it
    /// and (per the issue's "熔断" ask) react without polling session state.
    /// The run still finalizes exactly like a normal completion — this event
    /// precedes `Complete` in the stream, it does not replace it.
    BudgetExceeded {
        /// Session identifier.
        session_id: String,
        /// Which budget tripped: `"max_total_tokens"` | `"max_tool_calls"` |
        /// `"max_subagents"`.
        kind: String,
        /// The configured limit that was exceeded.
        limit: u64,
        /// The actual cumulative usage observed when the guardrail tripped.
        actual: u64,
    },

    /// 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,
    },

    /// A workflow catalog definition was added, changed, shadowed, or removed.
    WorkflowChanged {
        workflow_id: String,
        revision: u64,
        scope: String,
    },

    /// A workflow bundle became invalid while its last-known-good definition stays active.
    WorkflowInvalid {
        workflow_id: String,
        revision: u64,
        scope: String,
    },

    /// A previously invalid workflow bundle parsed successfully again.
    WorkflowRecovered {
        workflow_id: String,
        revision: u64,
        scope: String,
    },

    /// A first-class Project was created in the account registry.
    ProjectCreated { project_id: String, revision: u64 },

    /// A Project manifest or its shared resource inventory changed.
    ProjectUpdated { project_id: String, revision: u64 },

    /// A Project was archived. Sessions and resources are deliberately retained.
    ProjectArchived { project_id: String, revision: u64 },

    /// A session's stable Project membership and/or mutable Workspace changed.
    ///
    /// The historical variant name is retained for wire compatibility. Clients
    /// should use `metadata_version` to order refreshes and consume both fields.
    SessionProjectUpdated {
        session_id: String,
        /// Explicit `null` means Unassigned. Keep the field present so account
        /// feed replay consumers can recover session grouping without a fetch.
        #[serde(default)]
        project_id: Option<String>,
        /// Explicit `null` means the session has no persisted Workspace.
        /// `default` keeps older journal entries deserializable.
        #[serde(default)]
        workspace_path: Option<String>,
        metadata_version: u64,
    },

    /// A configuration section published a new last-known-good snapshot.
    #[serde(rename = "config.changed")]
    ConfigChanged { section: String, revision: u64 },

    /// A configuration edit was rejected while the prior runtime stayed live.
    #[serde(rename = "config.invalid")]
    ConfigInvalid { section: String, revision: u64 },

    /// A previously invalid configuration section became healthy again.
    #[serde(rename = "config.recovered")]
    ConfigRecovered { section: String, revision: u64 },

    /// An instruction workflow became the session's fixed active revision.
    WorkflowActivated {
        event_id: String,
        session_id: String,
        workflow_id: String,
        revision: u64,
        invoked_by: String,
    },

    /// The previously active instruction workflow was explicitly superseded.
    WorkflowDeactivated {
        event_id: String,
        session_id: String,
        workflow_id: String,
        revision: u64,
    },

    /// A user-facing notification derived from agent activity by the backend
    /// notification policy. Clients render this (e.g. an OS desktop notification)
    /// after applying their own presence checks (window focus). The decision of
    /// *whether* to notify — category, priority, preference gating, dedup — is
    /// made server-side in `bamboo-notification`; the client just delivers it.
    Notification {
        /// Unique id (for client-side dedup / dismissal).
        id: String,
        /// Session this notification is about.
        session_id: String,
        /// Category, e.g. `needs_approval` | `needs_clarification` | `run_completed`
        /// | `run_failed` | `subagent_completed` | `context_critical`.
        category: String,
        /// Priority: `high` | `normal` | `low`.
        priority: String,
        /// Short title line.
        title: String,
        /// Body text.
        body: String,
        /// Stable key for client-side coalescing within a short window.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        dedup_key: Option<String>,
        /// RFC3339 creation timestamp.
        created_at: String,
    },
}

/// Read-only interpretation of one legacy parent-visible child lifecycle.
/// The IDs in a worker frame are observations, not ActorDirectory authority.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct LegacyChildLifecycleObservation<'a> {
    pub parent_session_id: &'a str,
    pub child_session_id: &'a str,
    pub completed: bool,
}

impl AgentEvent {
    /// Project grouping/citation/tracing from a committed transcript message.
    /// These fields are display data, never an authorization capability.
    pub fn message_appended(session_id: &str, message: &bamboo_domain::Message) -> Self {
        let reference = |field: &str| {
            message
                .metadata
                .as_ref()
                .and_then(|m| m.get("session_message"))
                .and_then(|m| m.get(field))
                .and_then(serde_json::Value::as_str)
                .map(str::to_owned)
        };
        Self::MessageAppended {
            session_id: session_id.into(),
            message_id: message.id.clone(),
            role: message.role.clone(),
            content: message.content.clone(),
            created_at: message.created_at,
            thread_id: reference("thread_id"),
            in_reply_to: reference("in_reply_to"),
            correlation_id: reference("correlation_id"),
        }
    }
    /// Interpret direct and one-level parent-wrapped child lifecycle events
    /// identically for the bounded legacy replay cache. A wrapped event is
    /// admitted only when its inner parent equals the wrapper's child. This
    /// never changes canonical actor state or expands a token/event tree.
    pub fn legacy_child_lifecycle_observation(
        &self,
    ) -> Option<LegacyChildLifecycleObservation<'_>> {
        const MAX_ID_BYTES: usize = 256;
        const MAX_DETAIL_BYTES: usize = 4096;
        let leaf = match self {
            Self::SubAgentEvent {
                parent_session_id,
                child_session_id,
                event,
            } => {
                if parent_session_id.is_empty()
                    || parent_session_id.len() > MAX_ID_BYTES
                    || child_session_id.is_empty()
                    || child_session_id.len() > MAX_ID_BYTES
                    || parent_session_id == child_session_id
                {
                    return None;
                }
                let inner = event.as_ref();
                let inner_parent = match inner {
                    Self::SubAgentStarted {
                        parent_session_id, ..
                    }
                    | Self::SubAgentCompleted {
                        parent_session_id, ..
                    } => parent_session_id,
                    _ => return None,
                };
                if inner_parent != child_session_id {
                    return None;
                }
                inner
            }
            other => other,
        };
        let observation = match leaf {
            Self::SubAgentStarted {
                parent_session_id,
                child_session_id,
                title,
            } => {
                if title
                    .as_ref()
                    .is_some_and(|title| title.len() > MAX_DETAIL_BYTES)
                {
                    return None;
                }
                LegacyChildLifecycleObservation {
                    parent_session_id,
                    child_session_id,
                    completed: false,
                }
            }
            Self::SubAgentCompleted {
                parent_session_id,
                child_session_id,
                error,
                ..
            } => {
                if error
                    .as_ref()
                    .is_some_and(|error| error.len() > MAX_DETAIL_BYTES)
                {
                    return None;
                }
                LegacyChildLifecycleObservation {
                    parent_session_id,
                    child_session_id,
                    completed: true,
                }
            }
            _ => return None,
        };
        if observation.parent_session_id.is_empty()
            || observation.parent_session_id.len() > MAX_ID_BYTES
            || observation.child_session_id.is_empty()
            || observation.child_session_id.len() > MAX_ID_BYTES
            || observation.parent_session_id == observation.child_session_id
        {
            return None;
        }
        Some(observation)
    }

    /// Returns the session this event pertains to, when it carries one.
    ///
    /// Used by the account change-feed to route each event to the right
    /// client-side session without a per-session connection. For sub-agent
    /// events the *parent* session id is returned (that is the session a client
    /// observes in its list). Pure streaming/diagnostic variants (`Token`,
    /// `Complete`, …) return `None`; those are routed by their owning
    /// per-session forwarder instead.
    pub fn session_id(&self) -> Option<&str> {
        match self {
            AgentEvent::TaskListUpdated { task_list, .. } => Some(task_list.session_id.as_str()),
            AgentEvent::TaskListItemProgress { session_id, .. }
            | AgentEvent::TaskListCompleted { session_id, .. }
            | AgentEvent::TaskEvaluationStarted { session_id, .. }
            | AgentEvent::TaskEvaluationCompleted { session_id, .. }
            | AgentEvent::TaskEvaluationCancelled { session_id, .. }
            | AgentEvent::GoldEvaluationStarted { session_id, .. }
            | AgentEvent::GoldEvaluationCompleted { session_id, .. }
            | AgentEvent::GoldEvaluationCancelled { session_id, .. }
            | AgentEvent::GoalStatusChanged { session_id, .. }
            | AgentEvent::PlanModeEntered { session_id, .. }
            | AgentEvent::PlanModeExited { session_id, .. }
            | AgentEvent::PlanFileUpdated { session_id, .. }
            | AgentEvent::RunnerProgress { session_id, .. }
            | AgentEvent::PermissionPostureActivated { session_id, .. }
            | AgentEvent::SessionTitleUpdated { session_id, .. }
            | AgentEvent::SessionPinnedUpdated { session_id, .. }
            | AgentEvent::SessionCreated { session_id, .. }
            | AgentEvent::SessionDeleted { session_id, .. }
            | AgentEvent::SessionCleared { session_id, .. }
            | AgentEvent::MessageAppended { session_id, .. }
            | AgentEvent::SessionHistoryCommitted { session_id }
            | AgentEvent::ExecutionStarted { session_id, .. }
            | AgentEvent::BudgetExceeded { session_id, .. }
            | AgentEvent::WorkflowActivated { session_id, .. }
            | AgentEvent::WorkflowDeactivated { session_id, .. }
            | AgentEvent::SessionProjectUpdated { session_id, .. }
            | AgentEvent::Notification { session_id, .. } => Some(session_id.as_str()),
            AgentEvent::SubAgentStarted {
                parent_session_id, ..
            }
            | AgentEvent::SubAgentEvent {
                parent_session_id, ..
            }
            | AgentEvent::SubAgentHeartbeat {
                parent_session_id, ..
            }
            | AgentEvent::SubAgentCompleted {
                parent_session_id, ..
            }
            | AgentEvent::ChildApprovalChanged {
                parent_session_id, ..
            } => Some(parent_session_id.as_str()),
            _ => None,
        }
    }

    /// Whether this event carries live session state that a late per-session
    /// subscriber must replay before consuming the broadcast stream.
    ///
    /// This classification is shared by both server-owned and generic engine
    /// forwarders. Keeping it in core prevents one execution entry point from
    /// broadcasting a clarification (or other critical state) without first
    /// populating the runner replay cache.
    pub fn is_replayable_session_state(&self) -> bool {
        self.legacy_child_lifecycle_observation().is_some()
            || matches!(
                self,
                AgentEvent::TaskListUpdated { .. }
                    | AgentEvent::TaskListCompleted { .. }
                    | AgentEvent::SubAgentStarted { .. }
                    | AgentEvent::SubAgentCompleted { .. }
                    | AgentEvent::ChildApprovalRequested { .. }
                    | AgentEvent::ChildApprovalChanged { .. }
                    | AgentEvent::BashCompleted { .. }
                    | AgentEvent::SessionTitleUpdated { .. }
                    | AgentEvent::SessionPinnedUpdated { .. }
                    | AgentEvent::PlanModeEntered { .. }
                    | AgentEvent::PlanModeExited { .. }
                    | AgentEvent::BudgetExceeded { .. }
                    | AgentEvent::NeedClarification { .. }
                    | AgentEvent::WorkflowActivated { .. }
                    | AgentEvent::WorkflowDeactivated { .. }
            )
    }

    /// Whether this event belongs on the durable account change feed.
    ///
    /// Durable change events are low-volume, journaled to disk, and resumable
    /// via the account `/stream` feed. Ephemeral events — token-by-token
    /// streaming (`Token`/`ReasoningToken`/`ToolToken`), heartbeats, live
    /// budget/pressure gauges, and raw forwarded sub-agent events — return
    /// `false`: they stay exclusively on the per-session `/events/{id}` stream.
    /// Keeping them off the journal and the multiplexed feed is the core
    /// data-transfer win. This method lives in core so both the server and the
    /// engine forwarder can filter before cloning onto the feed.
    pub fn is_durable_change(&self) -> bool {
        matches!(
            self,
            AgentEvent::MessageAppended { .. }
                | AgentEvent::SessionHistoryCommitted { .. }
                | AgentEvent::SessionCreated { .. }
                | AgentEvent::SessionDeleted { .. }
                | AgentEvent::SessionCleared { .. }
                | AgentEvent::SessionTitleUpdated { .. }
                | AgentEvent::SessionPinnedUpdated { .. }
                | AgentEvent::TaskListUpdated { .. }
                | AgentEvent::TaskListItemProgress { .. }
                | AgentEvent::TaskListCompleted { .. }
                | AgentEvent::TaskEvaluationCompleted { .. }
                | AgentEvent::TaskEvaluationCancelled { .. }
                | AgentEvent::GoldEvaluationCancelled { .. }
                | AgentEvent::PlanModeEntered { .. }
                | AgentEvent::PlanModeExited { .. }
                | AgentEvent::PlanFileUpdated { .. }
                | AgentEvent::SubAgentStarted { .. }
                | AgentEvent::SubAgentCompleted { .. }
                | AgentEvent::ChildApprovalChanged { .. }
                | AgentEvent::NeedClarification { .. }
                | AgentEvent::ToolApprovalRequested { .. }
                | AgentEvent::ExecutionStarted { .. }
                | AgentEvent::BudgetExceeded { .. }
                | AgentEvent::Complete { .. }
                | AgentEvent::Cancelled { .. }
                | AgentEvent::Error { .. }
                | AgentEvent::WorkflowChanged { .. }
                | AgentEvent::WorkflowInvalid { .. }
                | AgentEvent::ProjectCreated { .. }
                | AgentEvent::ProjectUpdated { .. }
                | AgentEvent::ProjectArchived { .. }
                | AgentEvent::SessionProjectUpdated { .. }
                | AgentEvent::ConfigChanged { .. }
                | AgentEvent::ConfigInvalid { .. }
                | AgentEvent::ConfigRecovered { .. }
                | AgentEvent::WorkflowRecovered { .. }
                | AgentEvent::WorkflowActivated { .. }
                | AgentEvent::WorkflowDeactivated { .. }
        )
    }
}

fn default_allow_custom() -> bool {
    true
}

/// Gold evaluation checkpoint.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum GoldCheckpoint {
    PostRound,
    Terminal,
}

impl GoldCheckpoint {
    pub fn as_str(self) -> &'static str {
        match self {
            Self::PostRound => "post_round",
            Self::Terminal => "terminal",
        }
    }
}

/// Gold evaluator decision.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum GoldDecision {
    Continue,
    Achieved,
    Blocked,
    NeedInput,
    Exhausted,
}

impl GoldDecision {
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Continue => "continue",
            Self::Achieved => "achieved",
            Self::Blocked => "blocked",
            Self::NeedInput => "need_input",
            Self::Exhausted => "exhausted",
        }
    }
}

/// Confidence level for a Gold evaluation result.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum GoldConfidence {
    Low,
    Medium,
    High,
}

impl GoldConfidence {
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Low => "low",
            Self::Medium => "medium",
            Self::High => "high",
        }
    }

    /// Ordinal rank for threshold comparisons (`Low` < `Medium` < `High`).
    pub fn rank(self) -> u8 {
        match self {
            Self::Low => 0,
            Self::Medium => 1,
            Self::High => 2,
        }
    }

    /// Whether this confidence meets or exceeds the given floor.
    pub fn meets(self, floor: GoldConfidence) -> bool {
        self.rank() >= floor.rank()
    }
}

/// 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::{ProviderPromptUsage, 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(),
            version: Some(7),
        };

        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_eq!(value["version"], 7);
        assert!(value.get("todo_list").is_none());
    }

    #[test]
    fn context_archived_serializes_structural_evidence_without_history_content() {
        let event = AgentEvent::ContextArchived {
            archive_event_id: "compression-event-1".to_string(),
            trigger_type: "auto".to_string(),
            messages_archived: 8,
            groups_archived: 4,
            user_turns_archived: 4,
            active_tokens_before: 9_000,
            active_tokens_after: 5_000,
            target_tokens: 6_000,
            retained_recent_user_turns: 3,
            oldest_retained_message_id: Some("message-9".to_string()),
            oldest_retained_user_message_id: Some("message-9".to_string()),
            model_context_epoch: 2,
            reset_reason: "compression".to_string(),
        };

        let value = serde_json::to_value(&event).expect("archive event should serialize");
        assert_eq!(value["type"], "context_archived");
        assert_eq!(value["messages_archived"], 8);
        assert_eq!(value["active_tokens_after"], 5_000);
        assert_eq!(value["reset_reason"], "compression");
        let wire = serde_json::to_string(&value).unwrap();
        assert!(!wire.contains("summary"));
        assert!(!wire.contains("raw_message"));
        assert!(!wire.contains("tool_result"));
        assert!(matches!(
            serde_json::from_value::<AgentEvent>(value).unwrap(),
            AgentEvent::ContextArchived {
                archive_event_id,
                messages_archived: 8,
                model_context_epoch: 2,
                ..
            } if archive_event_id == "compression-event-1"
        ));
    }

    #[test]
    fn session_project_updated_serializes_unassignment_as_explicit_null() {
        let event = AgentEvent::SessionProjectUpdated {
            session_id: "session-1".to_string(),
            project_id: None,
            workspace_path: Some("/workspaces/current".to_string()),
            metadata_version: 4,
        };

        let value = serde_json::to_value(&event).expect("event should serialize");
        assert_eq!(value["type"], "session_project_updated");
        assert!(
            value
                .get("project_id")
                .is_some_and(serde_json::Value::is_null),
            "unassignment must carry an explicit project_id: null"
        );
        assert_eq!(value["workspace_path"], "/workspaces/current");

        let restored: AgentEvent = serde_json::from_value(value).expect("event should deserialize");
        assert!(matches!(
            restored,
            AgentEvent::SessionProjectUpdated {
                session_id,
                project_id: None,
                workspace_path: Some(workspace_path),
                metadata_version: 4,
            } if session_id == "session-1" && workspace_path == "/workspaces/current"
        ));
    }

    #[test]
    fn session_project_updated_deserializes_legacy_event_without_workspace() {
        let restored: AgentEvent = serde_json::from_value(serde_json::json!({
            "type": "session_project_updated",
            "session_id": "session-1",
            "project_id": "project-1",
            "metadata_version": 2
        }))
        .expect("legacy event should deserialize");

        assert!(matches!(
            restored,
            AgentEvent::SessionProjectUpdated {
                workspace_path: None,
                metadata_version: 2,
                ..
            }
        ));
    }

    #[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 session_history_committed_is_a_routable_durable_barrier() {
        let event = AgentEvent::SessionHistoryCommitted {
            session_id: "session-1".to_string(),
        };

        let value = serde_json::to_value(&event).expect("event should serialize");
        assert_eq!(value["type"], "session_history_committed");
        assert_eq!(value["session_id"], "session-1");
        assert_eq!(event.session_id(), Some("session-1"));
        assert!(event.is_durable_change());

        let restored: AgentEvent = serde_json::from_value(value).expect("event should deserialize");
        assert!(matches!(
            restored,
            AgentEvent::SessionHistoryCommitted { session_id } if session_id == "session-1"
        ));
    }

    #[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(),
            generation: Some(7),
        };

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

    #[test]
    fn task_evaluation_event_without_generation_remains_deserializable() {
        let event: AgentEvent = serde_json::from_value(serde_json::json!({
            "type": "task_evaluation_started",
            "session_id": "session-1",
            "items_count": 2
        }))
        .expect("legacy task evaluation frame should remain compatible");

        assert!(matches!(
            event,
            AgentEvent::TaskEvaluationStarted {
                generation: None,
                ..
            }
        ));
    }

    #[test]
    fn evaluation_cancelled_events_serialize_as_terminal_lifecycle_events() {
        let task = AgentEvent::TaskEvaluationCancelled {
            session_id: "session-1".to_string(),
            reason: "run_suspended".to_string(),
            generation: Some(7),
        };
        let gold = AgentEvent::GoldEvaluationCancelled {
            session_id: "session-1".to_string(),
            reason: "run_completed".to_string(),
        };

        assert!(task.is_durable_change());
        assert!(gold.is_durable_change());
        let task_value = serde_json::to_value(task).unwrap();
        let gold_value = serde_json::to_value(gold).unwrap();
        assert_eq!(task_value["type"], "task_evaluation_cancelled");
        assert_eq!(task_value["reason"], "run_suspended");
        assert_eq!(gold_value["type"], "gold_evaluation_cancelled");
        assert_eq!(gold_value["reason"], "run_completed");
    }

    #[test]
    fn gold_evaluation_completed_serializes_with_gold_type_and_fields() {
        let event = AgentEvent::GoldEvaluationCompleted {
            session_id: "session-1".to_string(),
            checkpoint: GoldCheckpoint::PostRound,
            iteration: 3,
            decision: GoldDecision::Continue,
            confidence: GoldConfidence::Medium,
            reasoning: "Need one more iteration".to_string(),
        };

        let value = serde_json::to_value(event).expect("event should serialize");
        assert_eq!(value["type"], "gold_evaluation_completed");
        assert_eq!(value["checkpoint"], "post_round");
        assert_eq!(value["iteration"], 3);
        assert_eq!(value["decision"], "continue");
        assert_eq!(value["confidence"], "medium");
        assert_eq!(value["reasoning"], "Need one more iteration");
    }

    #[test]
    fn gold_evaluation_started_deserializes() {
        let json = serde_json::json!({
            "type": "gold_evaluation_started",
            "session_id": "session-1",
            "checkpoint": "terminal",
            "iteration": 7
        });

        let event: AgentEvent = serde_json::from_value(json).expect("should deserialize");
        match event {
            AgentEvent::GoldEvaluationStarted {
                session_id,
                checkpoint,
                iteration,
            } => {
                assert_eq!(session_id, "session-1");
                assert_eq!(checkpoint, GoldCheckpoint::Terminal);
                assert_eq!(iteration, 7);
            }
            other => panic!("unexpected event: {other:?}"),
        }
    }

    #[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()),
            tool_name: Some("conclusion_with_options".to_string()),
            allow_custom: false,
            source: Some(PendingQuestionSource::PauseTool),
        };

        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["tool_name"], "conclusion_with_options");
        assert_eq!(value["allow_custom"], false);
        assert_eq!(value["source"], "pause_tool");
    }

    #[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,
                tool_name,
                allow_custom,
                source,
            } => {
                assert_eq!(question, "Continue?");
                assert_eq!(options, Some(vec!["Yes".to_string(), "No".to_string()]));
                assert_eq!(tool_call_id, None);
                assert_eq!(tool_name, None);
                assert!(allow_custom); // default_allow_custom returns true
                assert_eq!(source, None);
            }
            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,
                tool_name,
                allow_custom,
                source,
            } => {
                assert_eq!(question, "Pick one");
                assert_eq!(options, None);
                assert_eq!(tool_call_id, None);
                assert_eq!(tool_name, None);
                assert!(!allow_custom);
                assert_eq!(source, None);
            }
            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");
        // Compare against serde's own serialization (RFC3339 with `Z` for UTC),
        // not `to_rfc3339()` which emits a `+00:00` offset instead.
        assert_eq!(
            value["entered_at"],
            serde_json::to_value(entered_at).unwrap()
        );
    }

    #[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(),
            status: Some(bamboo_domain::PlanModeStatus::AwaitingApproval),
        };

        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 child_approval_changed_routes_to_parent_and_is_durable() {
        let event = AgentEvent::ChildApprovalChanged {
            parent_session_id: "parent-1".into(),
            child_session_id: "child-1".into(),
            child_attempt: 3,
            request_id: "req-1".into(),
            version: 2,
            status: "approved".into(),
            reason: None,
            tool_name: "Bash".into(),
            permission: "execute".into(),
            resource: "/tmp/x".into(),
            created_at: "2026-01-01T00:00:00Z".into(),
            resolved_at: Some("2026-01-01T00:00:01Z".into()),
        };
        assert_eq!(event.session_id(), Some("parent-1"));
        assert!(event.is_durable_change());
        let value = serde_json::to_value(event).unwrap();
        assert_eq!(value["type"], "child_approval_changed");
        assert_eq!(value["status"], "approved");
        assert_eq!(value["child_attempt"], 3);

        let mut legacy = value;
        legacy.as_object_mut().unwrap().remove("child_attempt");
        let restored: AgentEvent = serde_json::from_value(legacy).unwrap();
        assert!(matches!(
            restored,
            AgentEvent::ChildApprovalChanged {
                child_attempt: 0,
                ..
            }
        ));
    }

    #[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,
            title_generated: true,
            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();
        assert!(matches!(
            decoded,
            AgentEvent::SessionTitleUpdated {
                title_generated: true,
                ..
            }
        ));

        let legacy = serde_json::json!({
            "type": "session_title_updated",
            "session_id": "sess-legacy",
            "title": "Existing title",
            "title_version": 2,
            "source": "manual",
            "updated_at": "2025-01-01T00:00:00Z"
        });
        let decoded: AgentEvent = serde_json::from_value(legacy).unwrap();
        assert!(matches!(
            decoded,
            AgentEvent::SessionTitleUpdated {
                title_generated: true,
                ..
            }
        ));
    }

    #[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:?}"),
        }
    }

    #[test]
    fn workflow_catalog_events_are_durable_and_account_scoped() {
        for event in [
            AgentEvent::WorkflowChanged {
                workflow_id: "review".to_string(),
                revision: 2,
                scope: "global".to_string(),
            },
            AgentEvent::WorkflowInvalid {
                workflow_id: "review".to_string(),
                revision: 3,
                scope: "workspace:1234".to_string(),
            },
            AgentEvent::WorkflowRecovered {
                workflow_id: "review".to_string(),
                revision: 4,
                scope: "workspace:1234".to_string(),
            },
        ] {
            assert!(event.is_durable_change());
            assert_eq!(event.session_id(), None);
            let encoded = serde_json::to_string(&event).expect("serialize");
            let _: AgentEvent = serde_json::from_str(&encoded).expect("deserialize");
        }
    }

    #[test]
    fn raw_child_and_streaming_payloads_are_never_durable_account_changes() {
        let raw_child = AgentEvent::SubAgentEvent {
            parent_session_id: "parent".to_string(),
            child_session_id: "child".to_string(),
            event: Box::new(AgentEvent::ReasoningToken {
                content: "private reasoning".to_string(),
            }),
        };
        for event in [
            AgentEvent::Token {
                content: "visible but ephemeral".to_string(),
            },
            AgentEvent::ReasoningToken {
                content: "private reasoning".to_string(),
            },
            AgentEvent::ToolToken {
                tool_call_id: "call-1".to_string(),
                content: "private tool output".to_string(),
            },
            raw_child,
        ] {
            assert!(
                !event.is_durable_change(),
                "unexpected durable event: {event:?}"
            );
        }

        assert!(AgentEvent::SubAgentStarted {
            parent_session_id: "parent".to_string(),
            child_session_id: "child".to_string(),
            title: None,
        }
        .is_durable_change());
        assert!(AgentEvent::SubAgentCompleted {
            parent_session_id: "parent".to_string(),
            child_session_id: "child".to_string(),
            status: "completed".to_string(),
            error: None,
        }
        .is_durable_change());
    }

    #[test]
    fn clarification_is_shared_replayable_state_but_tokens_are_not() {
        let clarification = AgentEvent::NeedClarification {
            question: "Choose".to_string(),
            options: Some(vec!["A".to_string()]),
            tool_call_id: Some("call-1".to_string()),
            tool_name: Some("ConclusionWithOptions".to_string()),
            allow_custom: false,
            source: Some(PendingQuestionSource::PauseTool),
        };
        assert!(clarification.is_replayable_session_state());
        assert!(!AgentEvent::Token {
            content: "ephemeral".to_string(),
        }
        .is_replayable_session_state());
    }
}