recursive-agent 0.8.3

//! Event-sink abstraction for consuming agent lifecycle events.
//!
//! This module defines [`AgentEvent`] (a serialisable, non-exhaustive enum of
//! agent lifecycle events), the [`EventSink`] trait (the abstract consumer),
//! and four concrete implementations:
//!
//! * [`ChannelSink`] — delivers events into a `tokio::sync::mpsc` channel.
//! * [`BroadcastSink`] — delivers events into a `tokio::sync::broadcast` channel.
//! * [`NullSink`] — discards every event (no-op).
//! * [`CompositeSink`] — fans out to multiple inner sinks.

use serde::{Deserialize, Serialize};
use tokio::sync::broadcast;
use tokio::sync::mpsc;

// ---------------------------------------------------------------------------
// AgentEvent
// ---------------------------------------------------------------------------

/// A serialisable, non-exhaustive agent lifecycle event.
///
/// Uses `String` for the finish reason to avoid coupling to the
/// `FinishReason` type.  New variants can be added without a breaking change
/// (see `#[non_exhaustive]`).
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
#[non_exhaustive]
pub enum AgentEvent {
    /// Model generated text without tool calls.
    AssistantText { text: String, step: usize },
    /// Model requested to execute a tool.
    ToolCall {
        name: String,
        id: String,
        arguments: String,
        step: usize,
    },
    /// Time taken for the LLM request (excluding tool execution), in ms.
    Latency { step: usize, llm_ms: u64 },
    /// Result of executing a tool call.
    ToolResult {
        id: String,
        name: String,
        output: String,
        step: usize,
        /// `true` if the tool execution produced an error (output begins with
        /// `"ERROR: "` or was a permission denial). UI consumers should use
        /// this field instead of inspecting the `output` string prefix.
        #[serde(default)]
        is_error: bool,
    },
    /// Token usage statistics from the LLM provider.
    Usage {
        input_tokens: u32,
        output_tokens: u32,
        /// Tokens served from the provider's prompt cache (cache-hit pricing applies).
        cache_hit_tokens: u32,
        /// Tokens written to the provider's prompt cache (full input pricing applies).
        cache_miss_tokens: u32,
        step: usize,
    },
    /// Goal-328: locally-estimated per-component token breakdown of the
    /// prompt sent to the provider for `step`. Distinct from
    /// [`Self::Usage`] (the provider's reported truth) — see
    /// [`crate::llm::ContextBreakdown`] for the bucket definitions.
    ///
    /// Emitted once per step right after [`Self::Usage`] so consumers
    /// see the provider truth first, then the local estimate. Steps
    /// without an LLM call (denial-limit finishes, stuck-detection
    /// short-circuits) skip the event entirely.
    ContextBreakdown {
        breakdown: crate::llm::ContextBreakdown,
        step: usize,
    },
    /// Partial token from streaming response (if streaming enabled).
    PartialToken { text: String, step: usize },
    /// Partial reasoning / thinking delta from a streaming response.
    /// Emitted live (before the answer's `PartialToken` deltas) for
    /// providers that stream an explicit reasoning channel, so UI
    /// layers can render the chain-of-thought as it is produced. The
    /// authoritative, fully-joined reasoning string still arrives once
    /// per step via [`AgentEvent::Reasoning`], which finalises the
    /// streamed block.
    PartialReasoning { text: String, step: usize },
    /// Reasoning / thinking content from a model that exposes an
    /// explicit reasoning channel (DeepSeek R1, OpenAI o1, etc.).
    /// Carries the full reasoning text for the current step;
    /// providers that stream reasoning tokens (DeepSeek's
    /// `reasoning_content` SSE deltas) accumulate them and emit
    /// the final, fully-joined string in this event. UI layers
    /// render it as a `thinking…` block separate from the
    /// assistant message body. Emitted exactly once per step
    /// that produced reasoning content; steps without reasoning
    /// skip the event.
    Reasoning { text: String, step: usize },
    /// No-LLM proactive pruning of old tool results by count.
    /// Emitted with `pruned > 0` when the [`Microcompactor`] replaced
    /// the content of old tool-result messages with a placeholder.
    ///
    /// [`Microcompactor`]: crate::compact::micro::Microcompactor
    Microcompact { step: usize, pruned: usize },
    /// Transcript was compacted to fit size constraints.
    Compacted {
        removed: usize,
        kept: usize,
        summary_chars: usize,
        step: usize,
    },
    /// Proactive compaction was skipped — either the circuit breaker tripped
    /// (`CompactionSkipReason::CircuitBreaker`) or the compaction call itself
    /// failed (`CompactionSkipReason::Error`). Compaction is best-effort;
    /// a skipped compaction is never fatal to the turn.
    CompactionSkipped {
        step: usize,
        reason: CompactionSkipReason,
    },
    /// Agent run completed.
    TurnFinished {
        /// Human-readable reason for termination (e.g. "no_more_tool_calls").
        reason: String,
        /// Number of iterations executed.
        steps: usize,
    },
    /// Goal-202: Agent is requesting permission to enter plan mode.
    /// Emitted by `RequestPlanModeTool` before any exploration begins.
    /// The TUI / HTTP surface should prompt the user and call
    /// `AgentRuntime::approve_plan_mode_request` or `reject_plan_mode_request`.
    PlanModeRequested { reason: String },
    /// Goal-202: The user approved the plan-mode entry request.
    PlanModeApproved,
    /// Goal-202: The user rejected the plan-mode entry request.
    PlanModeRejected { reason: String },

    /// Agent has produced a plan and is waiting for confirmation.
    PlanProposed {
        plan_text: String,
        tool_calls: Vec<serde_json::Value>,
    },
    /// Plan was confirmed, execution will proceed.
    PlanConfirmed,
    /// Plan was rejected with a reason.
    PlanRejected { reason: String },

    /// A complete message was just appended to the agent transcript.
    ///
    /// Fired exactly once per committed message inside the agent runtime:
    /// once for the user message that starts a turn, once for the compaction
    /// summary if cross-turn compaction fires, and once per message in the
    /// kernel's output batch. Carries the full `Message` (role, content,
    /// tool_calls, tool_call_id, reasoning_content) so persistence consumers
    /// can write the canonical record without reassembling it from the finer
    /// `AssistantText` / `ToolCall` / `ToolResult` streaming events.
    ///
    /// `usage` — token usage for this message (non-None for assistant messages
    /// produced by an LLM call, g156).
    ///
    /// Not emitted for seeded transcript messages loaded from an existing
    /// session on resume (those are already on disk).
    MessageAppended {
        message: crate::message::Message,
        /// Token usage for this message (g156).
        usage: Option<crate::session::UsageMeta>,
    },

    /// Variant of [`MessageAppended`] specifically for `Role::Tool` messages
    /// that have an associated [`AuditMeta`] (Goal 153). The persistence sink
    /// handles this identically to `MessageAppended` but populates the
    /// `audit` field of [`crate::session::TranscriptEntry`].
    ///
    /// Emitting a separate variant (rather than `Option<AuditMeta>` on
    /// `MessageAppended`) keeps the common path zero-cost and avoids
    /// making audit an optional field on every event.
    MessageAppendedWithAudit {
        message: crate::message::Message,
        audit: crate::tools::AuditMeta,
    },

    /// Cross-turn compaction just fired; a compact_boundary marker should be
    /// written to the session JSONL (g157).
    ///
    /// `turn` — the turn index when compaction occurred.
    /// `compacted_count` — how many messages were removed.
    /// `summary_uuid` — UUID of the compaction summary message that replaced them.
    /// `cache_hit_tokens` — tokens served from the provider's prompt cache on
    ///   the just-completed turn (g336). Zero when no usage was reported.
    /// `cache_miss_tokens` — tokens that missed the prompt cache on the
    ///   just-completed turn (g336). Zero when no usage was reported.
    /// `is_recompaction_in_chain` — true when this is not the first compaction
    ///   in the session (g338). Signals a recompaction chain — the compaction
    ///   strategy may be failing for this session.
    /// `turns_since_previous_compact` — how many turns elapsed since the
    ///   previous compaction (g338). `0` when no previous compaction.
    /// `previous_compact_turn` — the turn index of the previous compaction,
    ///   or `None` on the first compaction (g338).
    CompactionBoundary {
        turn: u32,
        compacted_count: usize,
        summary_uuid: Option<String>,
        cache_hit_tokens: u32,
        cache_miss_tokens: u32,
        is_recompaction_in_chain: bool,
        turns_since_previous_compact: u32,
        previous_compact_turn: Option<u32>,
    },

    /// Goal-167: emitted when the agent updates its task checklist via
    /// `todo_write`. Carries the complete replacement list so consumers
    /// can render the current state without storing diffs.
    TodoUpdated {
        todos: Vec<crate::tools::todo::TodoItem>,
    },

    // ── Goal-168: /goal condition-based autonomous loop ──────────────
    /// A `/goal` was set for this session.
    GoalSet {
        /// The completion condition as written by the user.
        condition: String,
        /// Hard cap on autonomous turns.
        max_turns: u32,
    },
    /// The judge evaluated the condition and found it not yet met.
    /// The loop will continue with another turn.
    GoalContinuing {
        /// The judge's explanation for why the condition is not yet met.
        reason: String,
        /// Number of turns elapsed so far.
        turns: u32,
    },
    /// The judge evaluated the condition and confirmed it is met.
    GoalAchieved {
        /// The original condition string.
        condition: String,
        /// Total turns taken to reach the goal.
        turns: u32,
    },
    /// The active goal was cleared — either by `/goal clear`, turn-budget
    /// exhaustion, or an explicit `DELETE /sessions/:id/goal` call.
    GoalCleared,

    // ── Goal 210: hook progress events ────────────────────────────
    /// A hook started executing.
    HookStarted {
        hook_event: String,
        hook_name: String,
        status_message: Option<String>,
    },
    /// A hook produced incremental stdout output.
    HookProgress {
        hook_event: String,
        hook_name: String,
        last_line: String,
    },
    /// A hook finished executing.
    HookFinished {
        hook_event: String,
        hook_name: String,
        outcome: String,
        duration_ms: u64,
    },
    /// A hook produced a system message to show to the user.
    HookSystemMessage { text: String },

    /// Emitted when the LLM call fails with a retryable error (rate limit or timeout)
    /// and the agent will back off before retrying.
    LlmRetry {
        /// Which step (ReAct iteration) triggered the retry.
        step: usize,
        /// Which retry attempt this is (1 = first retry after initial failure).
        attempt: u32,
        /// How many milliseconds the agent will sleep before the next attempt.
        wait_ms: u64,
        /// Short human-readable description of the error ("rate_limited" or "timeout").
        reason: String,
    },
}

/// Why a proactive compaction was skipped.
///
/// See [`AgentEvent::CompactionSkipped`].
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CompactionSkipReason {
    /// The circuit breaker tripped after `MAX_CONSECUTIVE_COMPACT_FAILURES`
    /// consecutive failures. Further proactive compaction attempts are
    /// suppressed until a successful compaction resets the counter.
    CircuitBreaker,
    /// The compaction call itself failed (provider error, network issue, etc.).
    Error,
}

// ---------------------------------------------------------------------------
// EventSink trait
// ---------------------------------------------------------------------------

/// Abstract consumer of [`AgentEvent`]s.
///
/// Implementations are free to buffer, forward, filter, or discard events.
/// The trait is designed to be object-safe so that a single `Box<dyn EventSink>`
/// can hold any implementation.
#[async_trait::async_trait]
pub trait EventSink: Send + Sync + 'static {
    /// Deliver one event to this sink.
    ///
    /// Implementations should not panic. If the sink is full or closed, the
    /// event may be silently dropped (or an error logged).
    async fn emit(&self, event: AgentEvent);
}

// ---------------------------------------------------------------------------
// ChannelSink
// ---------------------------------------------------------------------------

/// An [`EventSink`] that sends events into a `tokio::sync::mpsc` channel.
///
/// If the channel is full (respects the bounded capacity) the event is
/// silently dropped.
pub struct ChannelSink {
    tx: mpsc::UnboundedSender<AgentEvent>,
}

impl ChannelSink {
    /// Create a new `ChannelSink` and return it together with the receiver
    /// half.
    pub fn new() -> (Self, mpsc::UnboundedReceiver<AgentEvent>) {
        let (tx, rx) = mpsc::unbounded_channel();
        (Self { tx }, rx)
    }
}

#[async_trait::async_trait]
impl EventSink for ChannelSink {
    async fn emit(&self, event: AgentEvent) {
        let _ = self.tx.send(event);
    }
}

// ---------------------------------------------------------------------------
// BroadcastSink
// ---------------------------------------------------------------------------

/// An [`EventSink`] that sends events into a `tokio::sync::broadcast` channel.
///
/// If all receivers are lagging behind the channel capacity, the oldest
/// events are dropped (standard broadcast behaviour).
pub struct BroadcastSink {
    tx: broadcast::Sender<AgentEvent>,
}

impl BroadcastSink {
    /// Create a new `BroadcastSink` with the given channel capacity.
    pub fn new(capacity: usize) -> (Self, broadcast::Receiver<AgentEvent>) {
        let (tx, rx) = broadcast::channel(capacity);
        (Self { tx }, rx)
    }
}

#[async_trait::async_trait]
impl EventSink for BroadcastSink {
    async fn emit(&self, event: AgentEvent) {
        let _ = self.tx.send(event);
    }
}

// ---------------------------------------------------------------------------
// NullSink
// ---------------------------------------------------------------------------

/// An [`EventSink`] that discards every event.
pub struct NullSink;

#[async_trait::async_trait]
impl EventSink for NullSink {
    async fn emit(&self, _event: AgentEvent) {
        // no-op
    }
}

// ---------------------------------------------------------------------------
// CompositeSink
// ---------------------------------------------------------------------------

/// An [`EventSink`] that fans out to multiple inner sinks.
///
/// Each inner sink receives every event. If any sink panics, the panic
/// propagates (i.e. there is no `catch_unwind` barrier).
pub struct CompositeSink {
    sinks: Vec<Box<dyn EventSink>>,
}

impl CompositeSink {
    /// Create a new `CompositeSink` from an iterator of sinks.
    pub fn new(sinks: impl IntoIterator<Item = Box<dyn EventSink>>) -> Self {
        Self {
            sinks: sinks.into_iter().collect(),
        }
    }
}

#[async_trait::async_trait]
impl EventSink for CompositeSink {
    async fn emit(&self, event: AgentEvent) {
        for sink in &self.sinks {
            sink.emit(event.clone()).await;
        }
    }
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::*;

    // -- ChannelSink -------------------------------------------------------

    #[tokio::test]
    async fn channel_sink_delivers_events() {
        let (sink, mut rx) = ChannelSink::new();
        let event = AgentEvent::AssistantText {
            text: "hello".into(),
            step: 0,
        };

        sink.emit(event.clone()).await;
        let received = rx.recv().await.unwrap();
        assert_eq!(received, event);
    }

    // -- BroadcastSink -----------------------------------------------------

    #[tokio::test]
    async fn broadcast_sink_delivers_to_multiple() {
        let (sink, mut rx1) = BroadcastSink::new(16);
        let mut rx2 = sink.tx.subscribe();

        let event = AgentEvent::PlanConfirmed;
        sink.emit(event.clone()).await;

        assert_eq!(rx1.recv().await.unwrap(), event);
        assert_eq!(rx2.recv().await.unwrap(), event);
    }

    // -- NullSink ----------------------------------------------------------

    #[tokio::test]
    async fn null_sink_does_not_panic() {
        let sink = NullSink;
        sink.emit(AgentEvent::PlanConfirmed).await;
        // If we reach here, the test passes.
    }

    // -- CompositeSink -----------------------------------------------------

    #[tokio::test]
    async fn composite_sink_fans_out() {
        let (sink1, mut rx1) = ChannelSink::new();
        let (sink2, mut rx2) = ChannelSink::new();

        let composite = CompositeSink::new(vec![
            Box::new(sink1) as Box<dyn EventSink>,
            Box::new(sink2) as Box<dyn EventSink>,
        ]);

        let event = AgentEvent::PlanConfirmed;
        composite.emit(event.clone()).await;

        assert_eq!(rx1.recv().await.unwrap(), event);
        assert_eq!(rx2.recv().await.unwrap(), event);
    }

    // -- Serialisation round-trip ------------------------------------------

    #[test]
    fn agent_event_serialization_round_trip() {
        let events = vec![
            AgentEvent::AssistantText {
                text: "hello".into(),
                step: 0,
            },
            AgentEvent::ToolCall {
                name: "foo".into(),
                id: "1".into(),
                arguments: "{}".into(),
                step: 1,
            },
            AgentEvent::Latency {
                step: 2,
                llm_ms: 100,
            },
            AgentEvent::ToolResult {
                id: "1".into(),
                name: "foo".into(),
                output: "ok".into(),
                step: 3,
                is_error: false,
            },
            AgentEvent::Usage {
                input_tokens: 10,
                output_tokens: 20,
                cache_hit_tokens: 5,
                cache_miss_tokens: 15,
                step: 4,
            },
            AgentEvent::PartialToken {
                text: "hel".into(),
                step: 5,
            },
            AgentEvent::Microcompact { step: 5, pruned: 3 },
            AgentEvent::Compacted {
                removed: 5,
                kept: 3,
                summary_chars: 200,
                step: 6,
            },
            AgentEvent::TurnFinished {
                reason: "done".into(),
                steps: 7,
            },
            AgentEvent::PlanProposed {
                plan_text: "plan".into(),
                tool_calls: vec![],
            },
            AgentEvent::PlanConfirmed,
            AgentEvent::PlanRejected {
                reason: "nope".into(),
            },
        ];

        for event in &events {
            let json = serde_json::to_string(event).unwrap();
            let deserialized: AgentEvent = serde_json::from_str(&json).unwrap();
            assert_eq!(*event, deserialized, "round-trip failed for {json}");
        }
    }

    // -- MessageAppended unit tests ----------------------------------------

    /// `MessageAppended` carries all three fields (content, tool_calls,
    /// reasoning_content) through a JSON round-trip intact.
    #[test]
    fn message_appended_round_trips_through_event() {
        use crate::llm::ToolCall as LlmToolCall;
        use crate::message::Message;

        let tc = LlmToolCall {
            id: "call_1".into(),
            name: "Read".into(),
            arguments: serde_json::json!({"path": "/tmp/foo"}),
        };
        let msg = Message {
            role: crate::message::Role::Assistant,
            content: "some text".into(),
            tool_calls: vec![tc.clone()],
            tool_call_id: None,
            reasoning_content: Some("my reasoning".into()),
            is_compaction_summary: false,
        };
        let event = AgentEvent::MessageAppended {
            message: msg.clone(),
            usage: None,
        };
        let json = serde_json::to_string(&event).unwrap();
        let deserialized: AgentEvent = serde_json::from_str(&json).unwrap();
        assert_eq!(event, deserialized);
        if let AgentEvent::MessageAppended { message: m, .. } = deserialized {
            assert_eq!(m.content, "some text");
            assert_eq!(m.reasoning_content.as_deref(), Some("my reasoning"));
            assert_eq!(m.tool_calls.len(), 1);
            assert_eq!(m.tool_calls[0].name, "Read");
        } else {
            panic!("deserialized to wrong variant");
        }
    }

    /// `CompositeSink` fans out `MessageAppended` to both inner sinks.
    #[tokio::test]
    async fn composite_sink_preserves_message_appended() {
        use crate::message::Message;

        let (sink1, mut rx1) = ChannelSink::new();
        let (sink2, mut rx2) = ChannelSink::new();
        let composite = CompositeSink::new(vec![
            Box::new(sink1) as Box<dyn EventSink>,
            Box::new(sink2) as Box<dyn EventSink>,
        ]);

        let msg = Message::user("hello");
        let event = AgentEvent::MessageAppended {
            message: msg.clone(),
            usage: None,
        };
        composite.emit(event.clone()).await;

        let got1 = rx1.recv().await.unwrap();
        let got2 = rx2.recv().await.unwrap();
        assert_eq!(got1, event);
        assert_eq!(got2, event);

        // Other variant also propagates.
        composite.emit(AgentEvent::PlanConfirmed).await;
        assert_eq!(rx1.recv().await.unwrap(), AgentEvent::PlanConfirmed);
        assert_eq!(rx2.recv().await.unwrap(), AgentEvent::PlanConfirmed);
    }

    #[test]
    fn tool_result_is_error_defaults_to_false_when_absent() {
        // kills `#[serde(default)]` removal mutation on is_error field
        let json = r#"{"type":"tool_result","id":"x","name":"y","output":"ok","step":0}"#;
        let event: AgentEvent = serde_json::from_str(json).expect("must deserialize");
        match event {
            AgentEvent::ToolResult { is_error, .. } => {
                assert!(!is_error, "is_error must default to false when absent");
            }
            other => panic!("unexpected variant: {other:?}"),
        }
    }

    #[test]
    fn tool_result_is_error_true_round_trips() {
        // kills `is_error: false` mutation in the serialization of ToolResult
        let event = AgentEvent::ToolResult {
            id: "c1".into(),
            name: "Bash".into(),
            output: "ERROR: oops".into(),
            step: 3,
            is_error: true,
        };
        let json = serde_json::to_string(&event).expect("serialize");
        assert!(
            json.contains("\"is_error\":true"),
            "is_error=true must be in JSON; got: {json}"
        );
        let restored: AgentEvent = serde_json::from_str(&json).expect("round-trip");
        assert_eq!(restored, event);
    }

    /// Regression test for Goal 278: MessageAppended no longer has parent_uuid field.
    #[test]
    fn additional_event_variants_round_trip() {
        // kills mutations to event type tags and field names for variants
        // not covered by agent_event_serialization_round_trip
        let events: Vec<AgentEvent> = vec![
            AgentEvent::PartialReasoning {
                text: "thinking...".into(),
                step: 0,
            },
            AgentEvent::Reasoning {
                text: "full reasoning".into(),
                step: 1,
            },
            AgentEvent::PlanModeRequested {
                reason: "complex task".into(),
            },
            AgentEvent::PlanModeApproved,
            AgentEvent::PlanModeRejected {
                reason: "too risky".into(),
            },
            AgentEvent::CompactionBoundary {
                turn: 5,
                compacted_count: 10,
                summary_uuid: Some("abc-123".into()),
                cache_hit_tokens: 42,
                cache_miss_tokens: 158,
                is_recompaction_in_chain: true,
                turns_since_previous_compact: 3,
                previous_compact_turn: Some(2),
            },
            AgentEvent::GoalSet {
                condition: "all tests pass".into(),
                max_turns: 20,
            },
            AgentEvent::GoalContinuing {
                reason: "not yet".into(),
                turns: 3,
            },
            AgentEvent::GoalAchieved {
                condition: "all tests pass".into(),
                turns: 5,
            },
            AgentEvent::GoalCleared,
            AgentEvent::HookStarted {
                hook_event: "pre_tool".into(),
                hook_name: "safety-check".into(),
                status_message: Some("checking...".into()),
            },
            AgentEvent::HookProgress {
                hook_event: "pre_tool".into(),
                hook_name: "safety-check".into(),
                last_line: "progress line".into(),
            },
            AgentEvent::HookFinished {
                hook_event: "pre_tool".into(),
                hook_name: "safety-check".into(),
                outcome: "continue".into(),
                duration_ms: 42,
            },
            AgentEvent::HookSystemMessage {
                text: "hook message".into(),
            },
            AgentEvent::Microcompact { step: 3, pruned: 2 },
            AgentEvent::LlmRetry {
                step: 2,
                attempt: 1,
                wait_ms: 5000,
                reason: "rate_limited".into(),
            },
        ];

        for event in &events {
            let json = serde_json::to_string(event)
                .unwrap_or_else(|e| panic!("serialization failed for {event:?}: {e}"));
            let back: AgentEvent = serde_json::from_str(&json)
                .unwrap_or_else(|e| panic!("deserialization failed for {json}: {e}"));
            assert_eq!(*event, back, "round-trip failed for event type");
        }
    }

    #[test]
    fn todo_updated_and_message_appended_with_audit_round_trip() {
        // kills removal mutations for TodoUpdated and MessageAppendedWithAudit variants
        use crate::tools::todo::{TodoItem, TodoStatus};

        let todos = vec![TodoItem {
            content: "do something".into(),
            status: TodoStatus::InProgress,
            active_form: Some("doing something".into()),
        }];
        let event = AgentEvent::TodoUpdated { todos };
        let json = serde_json::to_string(&event).unwrap();
        let back: AgentEvent = serde_json::from_str(&json).unwrap();
        assert_eq!(event, back, "TodoUpdated must round-trip");

        // MessageAppendedWithAudit
        let msg = crate::message::Message::user("hello");
        let audit = crate::tools::AuditMeta::synthetic_unknown_tool("Read");
        let event2 = AgentEvent::MessageAppendedWithAudit {
            message: msg,
            audit,
        };
        let json2 = serde_json::to_string(&event2).unwrap();
        let back2: AgentEvent = serde_json::from_str(&json2).unwrap();
        assert_eq!(event2, back2, "MessageAppendedWithAudit must round-trip");
    }

    #[tokio::test]
    async fn broadcast_sink_no_subscriber_does_not_panic() {
        // kills `let _ = self.sender.send(event)` → unconditional panic mutation
        // When the receiver is dropped, BroadcastSink::emit must silently succeed.
        let (sink, rx) = BroadcastSink::new(16);
        drop(rx); // nobody is listening
                  // Must not panic
        sink.emit(AgentEvent::PlanConfirmed).await;
    }

    #[tokio::test]
    async fn channel_sink_delivers_in_order() {
        // kills any reordering mutation in ChannelSink::emit
        let (sink, mut rx) = ChannelSink::new();
        sink.emit(AgentEvent::PlanConfirmed).await;
        sink.emit(AgentEvent::PlanModeApproved).await;
        assert_eq!(rx.recv().await.unwrap(), AgentEvent::PlanConfirmed);
        assert_eq!(rx.recv().await.unwrap(), AgentEvent::PlanModeApproved);
    }

    #[test]
    fn message_appended_no_longer_has_parent_uuid_field() {
        // Source-grep check that src/event.rs doesn't define
        // parent_uuid inside AgentEvent::MessageAppended.
        let src = include_str!("event.rs");
        let arm = src
            .split("MessageAppended {")
            .nth(1)
            .expect("MessageAppended arm must exist")
            .split("}")
            .next()
            .expect("MessageAppended arm must close");
        assert!(
            !arm.contains("parent_uuid"),
            "MessageAppended arm must not reference parent_uuid: {arm}"
        );
    }
}